1pub(crate) mod bmp;
4pub(crate) mod loader;
5mod types;
6
7pub use bmp::BmpIndex;
8#[cfg(feature = "diagnostics")]
9pub use types::DimRawData;
10pub use types::{SparseIndex, VectorIndex, VectorSearchResult};
11
12#[derive(Debug, Clone, Default)]
14pub struct SegmentMemoryStats {
15 pub segment_id: u128,
17 pub num_docs: u32,
19 pub term_dict_cache_bytes: usize,
21 pub store_cache_bytes: usize,
23 pub sparse_index_bytes: usize,
25 pub dense_index_bytes: usize,
27 pub bloom_filter_bytes: usize,
29 pub pinned_metadata_bytes: u64,
31 pub pin_intended_bytes: u64,
34}
35
36impl SegmentMemoryStats {
37 pub fn total_bytes(&self) -> usize {
39 self.term_dict_cache_bytes
40 + self.store_cache_bytes
41 + self.sparse_index_bytes
42 + self.dense_index_bytes
43 + self.bloom_filter_bytes
44 }
45}
46
47use std::sync::Arc;
48
49use rustc_hash::FxHashMap;
50
51use super::vector_data::LazyFlatVectorData;
52use crate::directories::{Directory, FileHandle};
53use crate::dsl::{DenseVectorQuantization, Document, Field, Schema};
54use crate::structures::{
55 AsyncSSTableReader, BlockPostingList, CoarseCentroids, IVFPQIndex, IVFRaBitQIndex, PQCodebook,
56 RaBitQIndex, SSTableStats, TermInfo,
57};
58use crate::{DocId, Error, Result};
59
60use super::store::{AsyncStoreReader, RawStoreBlock};
61use super::types::{SegmentFiles, SegmentId, SegmentMeta};
62
63pub(crate) fn combine_ordinal_results(
69 raw: impl IntoIterator<Item = (u32, u16, f32)>,
70 combiner: crate::query::MultiValueCombiner,
71 limit: usize,
72) -> Vec<VectorSearchResult> {
73 let collected: Vec<(u32, u16, f32)> = raw.into_iter().collect();
74
75 let num_raw = collected.len();
76 if log::log_enabled!(log::Level::Debug) {
77 let mut ids: Vec<u32> = collected.iter().map(|(d, _, _)| *d).collect();
78 ids.sort_unstable();
79 ids.dedup();
80 log::debug!(
81 "combine_ordinal_results: {} raw entries, {} unique docs, combiner={:?}, limit={}",
82 num_raw,
83 ids.len(),
84 combiner,
85 limit
86 );
87 }
88
89 let all_single = collected.iter().all(|&(_, ord, _)| ord == 0);
91 if all_single {
92 let mut results: Vec<VectorSearchResult> = collected
93 .into_iter()
94 .map(|(doc_id, _, score)| VectorSearchResult::new(doc_id, score, vec![(0, score)]))
95 .collect();
96 results.sort_unstable_by(|a, b| {
97 b.score
98 .partial_cmp(&a.score)
99 .unwrap_or(std::cmp::Ordering::Equal)
100 });
101 results.truncate(limit);
102 return results;
103 }
104
105 let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
107 rustc_hash::FxHashMap::default();
108 for (doc_id, ordinal, score) in collected {
109 doc_ordinals
110 .entry(doc_id as DocId)
111 .or_default()
112 .push((ordinal as u32, score));
113 }
114 let mut results: Vec<VectorSearchResult> = doc_ordinals
115 .into_iter()
116 .map(|(doc_id, ordinals)| {
117 let combined_score = combiner.combine(&ordinals);
118 VectorSearchResult::new(doc_id, combined_score, ordinals)
119 })
120 .collect();
121 results.sort_unstable_by(|a, b| {
122 b.score
123 .partial_cmp(&a.score)
124 .unwrap_or(std::cmp::Ordering::Equal)
125 });
126 results.truncate(limit);
127 results
128}
129
130pub struct SegmentReader {
136 meta: SegmentMeta,
137 term_dict: Arc<AsyncSSTableReader<TermInfo>>,
139 postings_handle: FileHandle,
141 store: Arc<AsyncStoreReader>,
143 schema: Arc<Schema>,
144 vector_indexes: FxHashMap<u32, VectorIndex>,
146 flat_vectors: FxHashMap<u32, LazyFlatVectorData>,
148 coarse_centroids: FxHashMap<u32, Arc<CoarseCentroids>>,
150 sparse_indexes: FxHashMap<u32, SparseIndex>,
152 bmp_indexes: FxHashMap<u32, BmpIndex>,
154 positions_handle: Option<FileHandle>,
156 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldReader>,
158 #[cfg(feature = "native")]
163 pin_report: crate::segment::pin::PinReport,
164}
165
166impl SegmentReader {
167 pub async fn open<D: Directory>(
169 dir: &D,
170 segment_id: SegmentId,
171 schema: Arc<Schema>,
172 cache_blocks: usize,
173 ) -> Result<Self> {
174 let files = SegmentFiles::new(segment_id.0);
175
176 let meta_slice = dir.open_read(&files.meta).await?;
178 let meta_bytes = meta_slice.read_bytes().await?;
179 let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
180 debug_assert_eq!(meta.id, segment_id.0);
181
182 let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
184 let term_dict = AsyncSSTableReader::open(term_dict_handle, cache_blocks).await?;
185
186 let postings_handle = dir.open_lazy(&files.postings).await?;
188
189 let store_handle = dir.open_lazy(&files.store).await?;
191 let store = AsyncStoreReader::open(store_handle, cache_blocks).await?;
192
193 let vectors_data = loader::load_vectors_file(dir, &files, &schema).await?;
195 let vector_indexes = vectors_data.indexes;
196 let flat_vectors = vectors_data.flat_vectors;
197
198 #[cfg(feature = "native")]
203 for (field_id, lazy_flat) in &flat_vectors {
204 if vector_indexes.contains_key(field_id) {
205 lazy_flat.advise_random_access();
206 }
207 }
208
209 let sparse_data = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
211 let sparse_indexes = sparse_data.maxscore_indexes;
212 let bmp_indexes = sparse_data.bmp_indexes;
213
214 let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
216
217 let fast_fields = loader::load_fast_fields_file(dir, &files, &schema).await?;
219
220 {
222 let mut parts = vec![format!(
223 "[segment] loaded {:016x}: docs={}",
224 segment_id.0, meta.num_docs
225 )];
226 if !vector_indexes.is_empty() || !flat_vectors.is_empty() {
227 parts.push(format!(
228 "dense: {} ann + {} flat fields",
229 vector_indexes.len(),
230 flat_vectors.len()
231 ));
232 }
233 for (field_id, idx) in &sparse_indexes {
234 parts.push(format!(
235 "sparse field {}: {} dims, ~{:.1} KB",
236 field_id,
237 idx.num_dimensions(),
238 idx.num_dimensions() as f64 * 24.0 / 1024.0
239 ));
240 }
241 for (field_id, idx) in &bmp_indexes {
242 parts.push(format!(
243 "bmp field {}: {} dims, {} blocks",
244 field_id,
245 idx.dims(),
246 idx.num_blocks
247 ));
248 }
249 if !fast_fields.is_empty() {
250 parts.push(format!("fast: {} fields", fast_fields.len()));
251 }
252 log::debug!("{}", parts.join(", "));
253 }
254
255 #[allow(unused_mut)]
256 let mut reader = Self {
257 meta,
258 term_dict: Arc::new(term_dict),
259 postings_handle,
260 store: Arc::new(store),
261 schema,
262 vector_indexes,
263 flat_vectors,
264 coarse_centroids: FxHashMap::default(),
265 sparse_indexes,
266 bmp_indexes,
267 positions_handle,
268 fast_fields,
269 #[cfg(feature = "native")]
270 pin_report: Default::default(),
271 };
272
273 #[cfg(feature = "native")]
275 reader.apply_pin_policy(&crate::segment::pin::pin_policy().to_owned());
276
277 Ok(reader)
278 }
279
280 #[cfg(feature = "native")]
289 pub(crate) fn apply_pin_policy(&mut self, policy: &crate::segment::pin::PinPolicy) {
290 use crate::segment::pin::PinReport;
291
292 if !policy.is_enabled() {
293 return;
294 }
295 let mut remaining = policy.budget_bytes;
296 let mut report = PinReport::default();
297
298 for bmp in self.bmp_indexes.values_mut() {
300 bmp.pin_block_starts(policy.mode, &mut remaining, &mut report);
301 }
302 for sparse in self.sparse_indexes.values_mut() {
304 sparse.pin_skip_section(policy.mode, &mut remaining, &mut report);
305 }
306 for flat in self.flat_vectors.values_mut() {
308 flat.pin_doc_ids(policy.mode, &mut remaining, &mut report);
309 }
310 for bmp in self.bmp_indexes.values_mut() {
311 bmp.pin_doc_maps(policy.mode, &mut remaining, &mut report);
312 }
313 for bmp in self.bmp_indexes.values_mut() {
315 bmp.pin_sb_grid(policy.mode, &mut remaining, &mut report);
316 }
317
318 if report.skipped_budget_bytes > 0 || report.failed_bytes > 0 {
319 log::warn!(
320 "[pin] segment {:016x}: pinned {}/{} bytes (budget skipped {}, mlock failed {}) — raise HERMES_PIN_METADATA_BUDGET_MB or RLIMIT_MEMLOCK for full coverage",
321 self.meta.id,
322 report.pinned_bytes,
323 report.intended_bytes,
324 report.skipped_budget_bytes,
325 report.failed_bytes,
326 );
327 } else if report.pinned_bytes > 0 {
328 log::info!(
329 "[pin] segment {:016x}: pinned {} bytes of hot metadata ({:?})",
330 self.meta.id,
331 report.pinned_bytes,
332 policy.mode,
333 );
334 }
335 self.pin_report = report;
336 }
337
338 pub fn meta(&self) -> &SegmentMeta {
344 &self.meta
345 }
346
347 pub fn num_docs(&self) -> u32 {
348 self.meta.num_docs
349 }
350
351 pub fn avg_field_len(&self, field: Field) -> f32 {
353 self.meta.avg_field_len(field)
354 }
355
356 pub fn schema(&self) -> &Schema {
357 &self.schema
358 }
359
360 pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
362 &self.sparse_indexes
363 }
364
365 pub fn sparse_index(&self, field: Field) -> Option<&SparseIndex> {
367 self.sparse_indexes.get(&field.0)
368 }
369
370 pub fn bmp_index(&self, field: Field) -> Option<&BmpIndex> {
372 self.bmp_indexes.get(&field.0)
373 }
374
375 pub fn bmp_indexes(&self) -> &FxHashMap<u32, BmpIndex> {
377 &self.bmp_indexes
378 }
379
380 pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
382 &self.vector_indexes
383 }
384
385 pub fn flat_vectors(&self) -> &FxHashMap<u32, LazyFlatVectorData> {
387 &self.flat_vectors
388 }
389
390 pub fn fast_field(
392 &self,
393 field_id: u32,
394 ) -> Option<&crate::structures::fast_field::FastFieldReader> {
395 self.fast_fields.get(&field_id)
396 }
397
398 pub fn fast_fields(&self) -> &FxHashMap<u32, crate::structures::fast_field::FastFieldReader> {
400 &self.fast_fields
401 }
402
403 pub fn term_dict_stats(&self) -> SSTableStats {
405 self.term_dict.stats()
406 }
407
408 pub fn memory_stats(&self) -> SegmentMemoryStats {
410 let term_dict_stats = self.term_dict.stats();
411
412 let term_dict_cache_bytes = self.term_dict.cached_blocks() * 4096;
414
415 let store_cache_bytes = self.store.cached_blocks() * 4096;
417
418 let sparse_index_bytes: usize = self
420 .sparse_indexes
421 .values()
422 .map(|s| s.estimated_memory_bytes())
423 .sum::<usize>()
424 + self
425 .bmp_indexes
426 .values()
427 .map(|b| b.estimated_memory_bytes())
428 .sum::<usize>();
429
430 let dense_index_bytes: usize = self
433 .vector_indexes
434 .values()
435 .map(|v| v.estimated_memory_bytes())
436 .sum();
437
438 #[cfg(feature = "native")]
439 let (pinned_metadata_bytes, pin_intended_bytes) =
440 (self.pin_report.pinned_bytes, self.pin_report.intended_bytes);
441 #[cfg(not(feature = "native"))]
442 let (pinned_metadata_bytes, pin_intended_bytes) = (0u64, 0u64);
443
444 SegmentMemoryStats {
445 segment_id: self.meta.id,
446 num_docs: self.meta.num_docs,
447 term_dict_cache_bytes,
448 store_cache_bytes,
449 sparse_index_bytes,
450 dense_index_bytes,
451 bloom_filter_bytes: term_dict_stats.bloom_filter_size,
452 pinned_metadata_bytes,
453 pin_intended_bytes,
454 }
455 }
456
457 pub async fn get_postings(
462 &self,
463 field: Field,
464 term: &[u8],
465 ) -> Result<Option<BlockPostingList>> {
466 log::debug!(
467 "SegmentReader::get_postings field={} term_len={}",
468 field.0,
469 term.len()
470 );
471
472 let mut key = Vec::with_capacity(4 + term.len());
474 key.extend_from_slice(&field.0.to_le_bytes());
475 key.extend_from_slice(term);
476
477 let term_info = match self.term_dict.get(&key).await? {
479 Some(info) => {
480 log::debug!("SegmentReader::get_postings found term_info");
481 info
482 }
483 None => {
484 log::debug!("SegmentReader::get_postings term not found");
485 return Ok(None);
486 }
487 };
488
489 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
491 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
493 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
494 posting_list.push(doc_id, tf);
495 }
496 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
497 return Ok(Some(block_list));
498 }
499
500 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
502 Error::Corruption("TermInfo has neither inline nor external data".to_string())
503 })?;
504
505 let start = posting_offset;
506 let end = start + posting_len;
507
508 if end > self.postings_handle.len() {
509 return Err(Error::Corruption(
510 "Posting offset out of bounds".to_string(),
511 ));
512 }
513
514 let posting_bytes = self.postings_handle.read_bytes_range(start..end).await?;
515 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
516
517 Ok(Some(block_list))
518 }
519
520 pub async fn get_prefix_postings(
522 &self,
523 field: Field,
524 prefix: &[u8],
525 ) -> Result<Vec<BlockPostingList>> {
526 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
528 key_prefix.extend_from_slice(&field.0.to_le_bytes());
529 key_prefix.extend_from_slice(prefix);
530
531 let entries = self.term_dict.prefix_scan(&key_prefix).await?;
532 let mut results = Vec::with_capacity(entries.len());
533
534 for (_key, term_info) in entries {
535 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
536 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
537 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
538 posting_list.push(doc_id, tf);
539 }
540 results.push(BlockPostingList::from_posting_list(&posting_list)?);
541 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
542 let start = posting_offset;
543 let end = start + posting_len;
544 if end > self.postings_handle.len() {
545 continue;
546 }
547 let posting_bytes = self.postings_handle.read_bytes_range(start..end).await?;
548 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
549 }
550 }
551
552 Ok(results)
553 }
554
555 pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
560 self.doc_with_fields(local_doc_id, None).await
561 }
562
563 pub async fn doc_with_fields(
569 &self,
570 local_doc_id: DocId,
571 fields: Option<&rustc_hash::FxHashSet<u32>>,
572 ) -> Result<Option<Document>> {
573 let mut doc = match fields {
574 Some(set) => {
575 let field_ids: Vec<u32> = set.iter().copied().collect();
576 match self
577 .store
578 .get_fields(local_doc_id, &self.schema, &field_ids)
579 .await
580 {
581 Ok(Some(d)) => d,
582 Ok(None) => return Ok(None),
583 Err(e) => return Err(Error::from(e)),
584 }
585 }
586 None => match self.store.get(local_doc_id, &self.schema).await {
587 Ok(Some(d)) => d,
588 Ok(None) => return Ok(None),
589 Err(e) => return Err(Error::from(e)),
590 },
591 };
592
593 for (&field_id, lazy_flat) in &self.flat_vectors {
595 if let Some(set) = fields
597 && !set.contains(&field_id)
598 {
599 continue;
600 }
601
602 let is_binary = lazy_flat.quantization == DenseVectorQuantization::Binary;
603 let (start, entries) = lazy_flat.flat_indexes_for_doc(local_doc_id);
604 for (j, &(_doc_id, _ordinal)) in entries.iter().enumerate() {
605 let flat_idx = start + j;
606 if is_binary {
607 let vbs = lazy_flat.vector_byte_size();
608 let mut raw = vec![0u8; vbs];
609 match lazy_flat.read_vector_raw_into(flat_idx, &mut raw).await {
610 Ok(()) => {
611 doc.add_binary_dense_vector(Field(field_id), raw);
612 }
613 Err(e) => {
614 log::warn!("Failed to hydrate binary vector field {}: {}", field_id, e);
615 }
616 }
617 } else {
618 match lazy_flat.get_vector(flat_idx).await {
619 Ok(vec) => {
620 doc.add_dense_vector(Field(field_id), vec);
621 }
622 Err(e) => {
623 log::warn!("Failed to hydrate vector field {}: {}", field_id, e);
624 }
625 }
626 }
627 }
628 }
629
630 Ok(Some(doc))
631 }
632
633 pub async fn prefetch_terms(
635 &self,
636 field: Field,
637 start_term: &[u8],
638 end_term: &[u8],
639 ) -> Result<()> {
640 let mut start_key = Vec::with_capacity(4 + start_term.len());
641 start_key.extend_from_slice(&field.0.to_le_bytes());
642 start_key.extend_from_slice(start_term);
643
644 let mut end_key = Vec::with_capacity(4 + end_term.len());
645 end_key.extend_from_slice(&field.0.to_le_bytes());
646 end_key.extend_from_slice(end_term);
647
648 self.term_dict.prefetch_range(&start_key, &end_key).await?;
649 Ok(())
650 }
651
652 pub fn store_has_dict(&self) -> bool {
654 self.store.has_dict()
655 }
656
657 pub fn store(&self) -> &super::store::AsyncStoreReader {
659 &self.store
660 }
661
662 pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
664 self.store.raw_blocks()
665 }
666
667 pub fn store_data_slice(&self) -> &FileHandle {
669 self.store.data_slice()
670 }
671
672 pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
674 self.term_dict.all_entries().await.map_err(Error::from)
675 }
676
677 pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
682 let entries = self.term_dict.all_entries().await?;
683 let mut result = Vec::with_capacity(entries.len());
684
685 for (key, term_info) in entries {
686 if key.len() > 4 {
688 let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
689 let term_bytes = &key[4..];
690 if let Ok(term_str) = std::str::from_utf8(term_bytes) {
691 result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
692 }
693 }
694 }
695
696 Ok(result)
697 }
698
699 pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
701 self.term_dict.iter()
702 }
703
704 pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
708 self.term_dict
709 .prefetch_all_data_bulk()
710 .await
711 .map_err(crate::Error::from)
712 }
713
714 pub async fn read_postings(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
716 let start = offset;
717 let end = start + len;
718 let bytes = self.postings_handle.read_bytes_range(start..end).await?;
719 Ok(bytes.to_vec())
720 }
721
722 pub async fn read_position_bytes(&self, offset: u64, len: u64) -> Result<Option<Vec<u8>>> {
724 let handle = match &self.positions_handle {
725 Some(h) => h,
726 None => return Ok(None),
727 };
728 let start = offset;
729 let end = start + len;
730 let bytes = handle.read_bytes_range(start..end).await?;
731 Ok(Some(bytes.to_vec()))
732 }
733
734 pub fn has_positions_file(&self) -> bool {
736 self.positions_handle.is_some()
737 }
738
739 fn score_quantized_batch(
745 query: &[f32],
746 raw: &[u8],
747 quant: crate::dsl::DenseVectorQuantization,
748 dim: usize,
749 scores: &mut [f32],
750 unit_norm: bool,
751 ) {
752 use crate::dsl::DenseVectorQuantization;
753 use crate::structures::simd;
754 match (quant, unit_norm) {
755 (DenseVectorQuantization::F32, false) => {
756 let num_floats = scores.len() * dim;
757 debug_assert!(
758 (raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()),
759 "f32 vector data not 4-byte aligned — vectors file may use legacy format"
760 );
761 let vectors: &[f32] =
762 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
763 simd::batch_cosine_scores(query, vectors, dim, scores);
764 }
765 (DenseVectorQuantization::F32, true) => {
766 let num_floats = scores.len() * dim;
767 debug_assert!(
768 (raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()),
769 "f32 vector data not 4-byte aligned"
770 );
771 let vectors: &[f32] =
772 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
773 simd::batch_dot_scores(query, vectors, dim, scores);
774 }
775 (DenseVectorQuantization::F16, false) => {
776 simd::batch_cosine_scores_f16(query, raw, dim, scores);
777 }
778 (DenseVectorQuantization::F16, true) => {
779 simd::batch_dot_scores_f16(query, raw, dim, scores);
780 }
781 (DenseVectorQuantization::UInt8, false) => {
782 simd::batch_cosine_scores_u8(query, raw, dim, scores);
783 }
784 (DenseVectorQuantization::UInt8, true) => {
785 simd::batch_dot_scores_u8(query, raw, dim, scores);
786 }
787 (DenseVectorQuantization::Binary, _) => {
788 unreachable!("Binary quantization should not reach score_quantized_batch");
790 }
791 }
792 }
793
794 pub async fn search_dense_vector(
800 &self,
801 field: Field,
802 query: &[f32],
803 k: usize,
804 nprobe: usize,
805 rerank_factor: f32,
806 combiner: crate::query::MultiValueCombiner,
807 ) -> Result<Vec<VectorSearchResult>> {
808 let ann_index = self.vector_indexes.get(&field.0);
809 let lazy_flat = self.flat_vectors.get(&field.0);
810
811 if ann_index.is_none() && lazy_flat.is_none() {
813 return Ok(Vec::new());
814 }
815
816 let unit_norm = self
818 .schema
819 .get_field_entry(field)
820 .and_then(|e| e.dense_vector_config.as_ref())
821 .is_some_and(|c| c.unit_norm);
822
823 const BRUTE_FORCE_BATCH: usize = 4096;
825
826 let fetch_k = (k as f32 * rerank_factor.max(1.0)).ceil() as usize;
827
828 let t0 = std::time::Instant::now();
830 let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
831 match index {
833 VectorIndex::RaBitQ(lazy) => {
834 let rabitq = lazy.get().ok_or_else(|| {
835 Error::Schema("RaBitQ index deserialization failed".to_string())
836 })?;
837 rabitq
838 .search(query, fetch_k)
839 .into_iter()
840 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
841 .collect()
842 }
843 VectorIndex::IVF(lazy) => {
844 let (index, codebook) = lazy.get().ok_or_else(|| {
845 Error::Schema("IVF index deserialization failed".to_string())
846 })?;
847 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
848 Error::Schema(format!(
849 "IVF index requires coarse centroids for field {}",
850 field.0
851 ))
852 })?;
853 let effective_nprobe = if nprobe > 0 { nprobe } else { 32 };
854 index
855 .search(centroids, codebook, query, fetch_k, Some(effective_nprobe))
856 .into_iter()
857 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
858 .collect()
859 }
860 VectorIndex::ScaNN(lazy) => {
861 let (index, codebook) = lazy.get().ok_or_else(|| {
862 Error::Schema("ScaNN index deserialization failed".to_string())
863 })?;
864 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
865 Error::Schema(format!(
866 "ScaNN index requires coarse centroids for field {}",
867 field.0
868 ))
869 })?;
870 let effective_nprobe = if nprobe > 0 { nprobe } else { 32 };
871 index
872 .search(centroids, codebook, query, fetch_k, Some(effective_nprobe))
873 .into_iter()
874 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
875 .collect()
876 }
877 VectorIndex::BinaryIvf(_) => {
878 Vec::new()
880 }
881 }
882 } else if let Some(lazy_flat) = lazy_flat {
883 log::debug!(
886 "[search_dense] field {}: brute-force on {} vectors (dim={}, quant={:?})",
887 field.0,
888 lazy_flat.num_vectors,
889 lazy_flat.dim,
890 lazy_flat.quantization
891 );
892 let dim = lazy_flat.dim;
893 let n = lazy_flat.num_vectors;
894 let quant = lazy_flat.quantization;
895 let mut collector = crate::query::ScoreCollector::new(fetch_k);
896 let mut scores = vec![0f32; BRUTE_FORCE_BATCH];
897
898 for batch_start in (0..n).step_by(BRUTE_FORCE_BATCH) {
899 let batch_count = BRUTE_FORCE_BATCH.min(n - batch_start);
900 let batch_bytes = lazy_flat
901 .read_vectors_batch(batch_start, batch_count)
902 .await
903 .map_err(crate::Error::Io)?;
904 let raw = batch_bytes.as_slice();
905
906 Self::score_quantized_batch(
907 query,
908 raw,
909 quant,
910 dim,
911 &mut scores[..batch_count],
912 unit_norm,
913 );
914
915 for (i, &score) in scores.iter().enumerate().take(batch_count) {
916 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
917 collector.insert_with_ordinal(doc_id, score, ordinal);
918 }
919 }
920
921 collector
922 .into_sorted_results()
923 .into_iter()
924 .map(|(doc_id, score, ordinal)| (doc_id, ordinal, score))
925 .collect()
926 } else {
927 return Ok(Vec::new());
928 };
929 let l1_elapsed = t0.elapsed();
930 {
931 let kind = match ann_index {
932 Some(VectorIndex::RaBitQ(_)) => "rabitq",
933 Some(VectorIndex::IVF(_)) => "ivf_rabitq",
934 Some(VectorIndex::ScaNN(_)) => "scann",
935 Some(VectorIndex::BinaryIvf(_)) => "binary_ivf",
936 None => "flat",
937 };
938 crate::observe::dense_l1(
939 self.schema.index_label(),
940 self.schema.get_field_name(field).unwrap_or("?"),
941 kind,
942 l1_elapsed.as_secs_f64(),
943 results.len(),
944 );
945 }
946 log::debug!(
947 "[search_dense] field {}: L1 returned {} candidates in {:.1}ms",
948 field.0,
949 results.len(),
950 l1_elapsed.as_secs_f64() * 1000.0
951 );
952
953 if ann_index.is_some()
956 && !results.is_empty()
957 && let Some(lazy_flat) = lazy_flat
958 {
959 let t_rerank = std::time::Instant::now();
960 let dim = lazy_flat.dim;
961 let quant = lazy_flat.quantization;
962 let vbs = lazy_flat.vector_byte_size();
963
964 let mut resolved: Vec<(usize, usize)> = Vec::new(); for (ri, c) in results.iter().enumerate() {
967 let (start, entries) = lazy_flat.flat_indexes_for_doc(c.0);
968 for (j, &(_, ord)) in entries.iter().enumerate() {
969 if ord == c.1 {
970 resolved.push((ri, start + j));
971 break;
972 }
973 }
974 }
975
976 let t_resolve = t_rerank.elapsed();
977 if !resolved.is_empty() {
978 resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
980
981 #[cfg(feature = "native")]
986 lazy_flat.prefetch_vectors(resolved.iter().map(|&(_, flat_idx)| flat_idx));
987
988 let t_read = std::time::Instant::now();
990 let mut raw_buf = vec![0u8; resolved.len() * vbs];
991 for (buf_idx, &(_, flat_idx)) in resolved.iter().enumerate() {
992 let _ = lazy_flat
993 .read_vector_raw_into(
994 flat_idx,
995 &mut raw_buf[buf_idx * vbs..(buf_idx + 1) * vbs],
996 )
997 .await;
998 }
999
1000 let read_elapsed = t_read.elapsed();
1001
1002 let t_score = std::time::Instant::now();
1004 let mut scores = vec![0f32; resolved.len()];
1005 Self::score_quantized_batch(query, &raw_buf, quant, dim, &mut scores, unit_norm);
1006 let score_elapsed = t_score.elapsed();
1007
1008 for (buf_idx, &(ri, _)) in resolved.iter().enumerate() {
1010 results[ri].2 = scores[buf_idx];
1011 }
1012
1013 crate::observe::dense_rerank(
1014 self.schema.index_label(),
1015 self.schema.get_field_name(field).unwrap_or("?"),
1016 t_rerank.elapsed().as_secs_f64(),
1017 t_resolve.as_secs_f64(),
1018 read_elapsed.as_secs_f64(),
1019 resolved.len(),
1020 );
1021 log::debug!(
1022 "[search_dense] field {}: rerank {} vectors (dim={}, quant={:?}, {}B/vec): resolve={:.1}ms read={:.1}ms score={:.1}ms",
1023 field.0,
1024 resolved.len(),
1025 dim,
1026 quant,
1027 vbs,
1028 t_resolve.as_secs_f64() * 1000.0,
1029 read_elapsed.as_secs_f64() * 1000.0,
1030 score_elapsed.as_secs_f64() * 1000.0,
1031 );
1032 }
1033
1034 if results.len() > fetch_k {
1035 results.select_nth_unstable_by(fetch_k, |a, b| b.2.total_cmp(&a.2));
1036 results.truncate(fetch_k);
1037 }
1038 results.sort_unstable_by(|a, b| b.2.total_cmp(&a.2));
1039 log::debug!(
1040 "[search_dense] field {}: rerank total={:.1}ms",
1041 field.0,
1042 t_rerank.elapsed().as_secs_f64() * 1000.0
1043 );
1044 }
1045
1046 Ok(combine_ordinal_results(results, combiner, k))
1047 }
1048
1049 fn binary_ivf_nprobe(&self, field: Field) -> Option<usize> {
1051 self.schema
1052 .get_field_entry(field)
1053 .and_then(|e| e.binary_dense_vector_config.as_ref())
1054 .map(|c| c.nprobe)
1055 .filter(|&n| n > 0)
1056 }
1057
1058 pub async fn search_binary_dense_vector(
1062 &self,
1063 field: Field,
1064 query: &[u8],
1065 k: usize,
1066 combiner: crate::query::MultiValueCombiner,
1067 ) -> Result<Vec<VectorSearchResult>> {
1068 let t0 = crate::observe::Timer::start();
1069 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0)
1072 && let Some(ivf) = lazy.get()
1073 {
1074 let nprobe = self.binary_ivf_nprobe(field);
1075 let results = ivf.search(query, k, nprobe);
1076 crate::observe::dense_l1(
1077 self.schema.index_label(),
1078 self.schema.get_field_name(field).unwrap_or("?"),
1079 "binary_ivf",
1080 t0.secs(),
1081 results.len(),
1082 );
1083 return Ok(combine_ordinal_results(results, combiner, k));
1084 }
1085
1086 let lazy_flat = match self.flat_vectors.get(&field.0) {
1087 Some(f) => f,
1088 None => return Ok(Vec::new()),
1089 };
1090
1091 const BRUTE_FORCE_BATCH: usize = 8192; let dim_bits = lazy_flat.dim;
1094 let byte_len = lazy_flat.vector_byte_size();
1095 let n = lazy_flat.num_vectors;
1096
1097 if byte_len != query.len() {
1098 return Err(Error::Schema(format!(
1099 "Binary query vector byte length {} != field byte length {}",
1100 query.len(),
1101 byte_len
1102 )));
1103 }
1104
1105 let mut collector = crate::query::ScoreCollector::new(k);
1106 let mut scores = vec![0f32; BRUTE_FORCE_BATCH];
1107
1108 for batch_start in (0..n).step_by(BRUTE_FORCE_BATCH) {
1109 let batch_count = BRUTE_FORCE_BATCH.min(n - batch_start);
1110 let batch_bytes = lazy_flat
1111 .read_vectors_batch(batch_start, batch_count)
1112 .await
1113 .map_err(crate::Error::Io)?;
1114 let raw = batch_bytes.as_slice();
1115
1116 crate::structures::simd::batch_hamming_scores(
1117 query,
1118 raw,
1119 byte_len,
1120 dim_bits,
1121 &mut scores[..batch_count],
1122 );
1123
1124 let threshold = collector.threshold();
1125 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1126 if score > threshold {
1127 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1128 collector.insert_with_ordinal(doc_id, score, ordinal);
1129 }
1130 }
1131 }
1132
1133 let results: Vec<(u32, u16, f32)> = collector
1134 .into_sorted_results()
1135 .into_iter()
1136 .map(|(doc_id, score, ordinal)| (doc_id, ordinal, score))
1137 .collect();
1138
1139 crate::observe::dense_l1(
1140 self.schema.index_label(),
1141 self.schema.get_field_name(field).unwrap_or("?"),
1142 "binary_flat",
1143 t0.secs(),
1144 results.len(),
1145 );
1146 Ok(combine_ordinal_results(results, combiner, k))
1147 }
1148
1149 pub fn has_dense_vector_index(&self, field: Field) -> bool {
1151 self.vector_indexes.contains_key(&field.0) || self.flat_vectors.contains_key(&field.0)
1152 }
1153
1154 pub fn get_dense_vector_index(&self, field: Field) -> Option<Arc<RaBitQIndex>> {
1156 match self.vector_indexes.get(&field.0) {
1157 Some(VectorIndex::RaBitQ(lazy)) => lazy.get().cloned(),
1158 _ => None,
1159 }
1160 }
1161
1162 pub fn get_ivf_vector_index(
1164 &self,
1165 field: Field,
1166 ) -> Option<(Arc<IVFRaBitQIndex>, Arc<crate::structures::RaBitQCodebook>)> {
1167 match self.vector_indexes.get(&field.0) {
1168 Some(VectorIndex::IVF(lazy)) => lazy.get().map(|(i, c)| (i.clone(), c.clone())),
1169 _ => None,
1170 }
1171 }
1172
1173 pub fn coarse_centroids(&self, field_id: u32) -> Option<&Arc<CoarseCentroids>> {
1175 self.coarse_centroids.get(&field_id)
1176 }
1177
1178 pub fn set_coarse_centroids(&mut self, centroids: FxHashMap<u32, Arc<CoarseCentroids>>) {
1180 self.coarse_centroids = centroids;
1181 }
1182
1183 pub fn get_scann_vector_index(
1185 &self,
1186 field: Field,
1187 ) -> Option<(Arc<IVFPQIndex>, Arc<PQCodebook>)> {
1188 match self.vector_indexes.get(&field.0) {
1189 Some(VectorIndex::ScaNN(lazy)) => lazy.get().map(|(i, c)| (i.clone(), c.clone())),
1190 _ => None,
1191 }
1192 }
1193
1194 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
1196 self.vector_indexes.get(&field.0)
1197 }
1198
1199 pub async fn get_positions(
1204 &self,
1205 field: Field,
1206 term: &[u8],
1207 ) -> Result<Option<crate::structures::PositionPostingList>> {
1208 let handle = match &self.positions_handle {
1210 Some(h) => h,
1211 None => return Ok(None),
1212 };
1213
1214 let mut key = Vec::with_capacity(4 + term.len());
1216 key.extend_from_slice(&field.0.to_le_bytes());
1217 key.extend_from_slice(term);
1218
1219 let term_info = match self.term_dict.get(&key).await? {
1221 Some(info) => info,
1222 None => return Ok(None),
1223 };
1224
1225 let (offset, length) = match term_info.position_info() {
1227 Some((o, l)) => (o, l),
1228 None => return Ok(None),
1229 };
1230
1231 let slice = handle.slice(offset..offset + length);
1233 let data = slice.read_bytes().await?;
1234
1235 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
1237
1238 Ok(Some(pos_list))
1239 }
1240
1241 pub fn has_positions(&self, field: Field) -> bool {
1243 if let Some(entry) = self.schema.get_field_entry(field) {
1245 entry.positions.is_some()
1246 } else {
1247 false
1248 }
1249 }
1250}
1251
1252#[cfg(feature = "sync")]
1254impl SegmentReader {
1255 pub fn get_postings_sync(&self, field: Field, term: &[u8]) -> Result<Option<BlockPostingList>> {
1257 let mut key = Vec::with_capacity(4 + term.len());
1259 key.extend_from_slice(&field.0.to_le_bytes());
1260 key.extend_from_slice(term);
1261
1262 let term_info = match self.term_dict.get_sync(&key)? {
1264 Some(info) => info,
1265 None => return Ok(None),
1266 };
1267
1268 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1270 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1271 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1272 posting_list.push(doc_id, tf);
1273 }
1274 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
1275 return Ok(Some(block_list));
1276 }
1277
1278 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
1280 Error::Corruption("TermInfo has neither inline nor external data".to_string())
1281 })?;
1282
1283 let start = posting_offset;
1284 let end = start + posting_len;
1285
1286 if end > self.postings_handle.len() {
1287 return Err(Error::Corruption(
1288 "Posting offset out of bounds".to_string(),
1289 ));
1290 }
1291
1292 let posting_bytes = self.postings_handle.read_bytes_range_sync(start..end)?;
1293 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
1294
1295 Ok(Some(block_list))
1296 }
1297
1298 pub fn get_prefix_postings_sync(
1300 &self,
1301 field: Field,
1302 prefix: &[u8],
1303 ) -> Result<Vec<BlockPostingList>> {
1304 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
1305 key_prefix.extend_from_slice(&field.0.to_le_bytes());
1306 key_prefix.extend_from_slice(prefix);
1307
1308 let entries = self.term_dict.prefix_scan_sync(&key_prefix)?;
1309 let mut results = Vec::with_capacity(entries.len());
1310
1311 for (_key, term_info) in entries {
1312 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1313 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1314 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1315 posting_list.push(doc_id, tf);
1316 }
1317 results.push(BlockPostingList::from_posting_list(&posting_list)?);
1318 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
1319 let start = posting_offset;
1320 let end = start + posting_len;
1321 if end > self.postings_handle.len() {
1322 continue;
1323 }
1324 let posting_bytes = self.postings_handle.read_bytes_range_sync(start..end)?;
1325 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
1326 }
1327 }
1328
1329 Ok(results)
1330 }
1331
1332 pub fn get_positions_sync(
1334 &self,
1335 field: Field,
1336 term: &[u8],
1337 ) -> Result<Option<crate::structures::PositionPostingList>> {
1338 let handle = match &self.positions_handle {
1339 Some(h) => h,
1340 None => return Ok(None),
1341 };
1342
1343 let mut key = Vec::with_capacity(4 + term.len());
1345 key.extend_from_slice(&field.0.to_le_bytes());
1346 key.extend_from_slice(term);
1347
1348 let term_info = match self.term_dict.get_sync(&key)? {
1350 Some(info) => info,
1351 None => return Ok(None),
1352 };
1353
1354 let (offset, length) = match term_info.position_info() {
1355 Some((o, l)) => (o, l),
1356 None => return Ok(None),
1357 };
1358
1359 let slice = handle.slice(offset..offset + length);
1360 let data = slice.read_bytes_sync()?;
1361
1362 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
1363 Ok(Some(pos_list))
1364 }
1365
1366 pub fn search_dense_vector_sync(
1369 &self,
1370 field: Field,
1371 query: &[f32],
1372 k: usize,
1373 nprobe: usize,
1374 rerank_factor: f32,
1375 combiner: crate::query::MultiValueCombiner,
1376 ) -> Result<Vec<VectorSearchResult>> {
1377 let ann_index = self.vector_indexes.get(&field.0);
1378 let lazy_flat = self.flat_vectors.get(&field.0);
1379
1380 if ann_index.is_none() && lazy_flat.is_none() {
1381 return Ok(Vec::new());
1382 }
1383
1384 let unit_norm = self
1385 .schema
1386 .get_field_entry(field)
1387 .and_then(|e| e.dense_vector_config.as_ref())
1388 .is_some_and(|c| c.unit_norm);
1389
1390 const BRUTE_FORCE_BATCH: usize = 4096;
1391 let fetch_k = (k as f32 * rerank_factor.max(1.0)).ceil() as usize;
1392
1393 let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
1394 match index {
1396 VectorIndex::RaBitQ(lazy) => {
1397 let rabitq = lazy.get().ok_or_else(|| {
1398 Error::Schema("RaBitQ index deserialization failed".to_string())
1399 })?;
1400 rabitq
1401 .search(query, fetch_k)
1402 .into_iter()
1403 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
1404 .collect()
1405 }
1406 VectorIndex::IVF(lazy) => {
1407 let (index, codebook) = lazy.get().ok_or_else(|| {
1408 Error::Schema("IVF index deserialization failed".to_string())
1409 })?;
1410 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
1411 Error::Schema(format!(
1412 "IVF index requires coarse centroids for field {}",
1413 field.0
1414 ))
1415 })?;
1416 let effective_nprobe = if nprobe > 0 { nprobe } else { 32 };
1417 index
1418 .search(centroids, codebook, query, fetch_k, Some(effective_nprobe))
1419 .into_iter()
1420 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
1421 .collect()
1422 }
1423 VectorIndex::ScaNN(lazy) => {
1424 let (index, codebook) = lazy.get().ok_or_else(|| {
1425 Error::Schema("ScaNN index deserialization failed".to_string())
1426 })?;
1427 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
1428 Error::Schema(format!(
1429 "ScaNN index requires coarse centroids for field {}",
1430 field.0
1431 ))
1432 })?;
1433 let effective_nprobe = if nprobe > 0 { nprobe } else { 32 };
1434 index
1435 .search(centroids, codebook, query, fetch_k, Some(effective_nprobe))
1436 .into_iter()
1437 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
1438 .collect()
1439 }
1440 VectorIndex::BinaryIvf(_) => {
1441 Vec::new()
1443 }
1444 }
1445 } else if let Some(lazy_flat) = lazy_flat {
1446 let dim = lazy_flat.dim;
1448 let n = lazy_flat.num_vectors;
1449 let quant = lazy_flat.quantization;
1450 let mut collector = crate::query::ScoreCollector::new(fetch_k);
1451 let mut scores = vec![0f32; BRUTE_FORCE_BATCH];
1452
1453 for batch_start in (0..n).step_by(BRUTE_FORCE_BATCH) {
1454 let batch_count = BRUTE_FORCE_BATCH.min(n - batch_start);
1455 let batch_bytes = lazy_flat
1456 .read_vectors_batch_sync(batch_start, batch_count)
1457 .map_err(crate::Error::Io)?;
1458 let raw = batch_bytes.as_slice();
1459
1460 Self::score_quantized_batch(
1461 query,
1462 raw,
1463 quant,
1464 dim,
1465 &mut scores[..batch_count],
1466 unit_norm,
1467 );
1468
1469 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1470 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1471 collector.insert_with_ordinal(doc_id, score, ordinal);
1472 }
1473 }
1474
1475 collector
1476 .into_sorted_results()
1477 .into_iter()
1478 .map(|(doc_id, score, ordinal)| (doc_id, ordinal, score))
1479 .collect()
1480 } else {
1481 return Ok(Vec::new());
1482 };
1483
1484 if ann_index.is_some()
1486 && !results.is_empty()
1487 && let Some(lazy_flat) = lazy_flat
1488 {
1489 let dim = lazy_flat.dim;
1490 let quant = lazy_flat.quantization;
1491 let vbs = lazy_flat.vector_byte_size();
1492
1493 let mut resolved: Vec<(usize, usize)> = Vec::new();
1494 for (ri, c) in results.iter().enumerate() {
1495 let (start, entries) = lazy_flat.flat_indexes_for_doc(c.0);
1496 for (j, &(_, ord)) in entries.iter().enumerate() {
1497 if ord == c.1 {
1498 resolved.push((ri, start + j));
1499 break;
1500 }
1501 }
1502 }
1503
1504 if !resolved.is_empty() {
1505 resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
1506 let mut raw_buf = vec![0u8; resolved.len() * vbs];
1507 for (buf_idx, &(_, flat_idx)) in resolved.iter().enumerate() {
1508 let _ = lazy_flat.read_vector_raw_into_sync(
1509 flat_idx,
1510 &mut raw_buf[buf_idx * vbs..(buf_idx + 1) * vbs],
1511 );
1512 }
1513
1514 let mut scores = vec![0f32; resolved.len()];
1515 Self::score_quantized_batch(query, &raw_buf, quant, dim, &mut scores, unit_norm);
1516
1517 for (buf_idx, &(ri, _)) in resolved.iter().enumerate() {
1518 results[ri].2 = scores[buf_idx];
1519 }
1520 }
1521
1522 if results.len() > fetch_k {
1523 results.select_nth_unstable_by(fetch_k, |a, b| b.2.total_cmp(&a.2));
1524 results.truncate(fetch_k);
1525 }
1526 results.sort_unstable_by(|a, b| b.2.total_cmp(&a.2));
1527 }
1528
1529 Ok(combine_ordinal_results(results, combiner, k))
1530 }
1531
1532 #[cfg(feature = "sync")]
1537 pub fn search_binary_dense_vector_sync(
1538 &self,
1539 field: Field,
1540 query: &[u8],
1541 k: usize,
1542 combiner: crate::query::MultiValueCombiner,
1543 ) -> Result<Vec<VectorSearchResult>> {
1544 let t0 = crate::observe::Timer::start();
1545 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0)
1548 && let Some(ivf) = lazy.get()
1549 {
1550 let nprobe = self.binary_ivf_nprobe(field);
1551 let results = ivf.search(query, k, nprobe);
1552 crate::observe::dense_l1(
1553 self.schema.index_label(),
1554 self.schema.get_field_name(field).unwrap_or("?"),
1555 "binary_ivf",
1556 t0.secs(),
1557 results.len(),
1558 );
1559 return Ok(combine_ordinal_results(results, combiner, k));
1560 }
1561
1562 let lazy_flat = match self.flat_vectors.get(&field.0) {
1563 Some(f) => f,
1564 None => return Ok(Vec::new()),
1565 };
1566
1567 const BRUTE_FORCE_BATCH: usize = 8192; let dim_bits = lazy_flat.dim;
1570 let byte_len = lazy_flat.vector_byte_size();
1571 let n = lazy_flat.num_vectors;
1572
1573 if byte_len != query.len() {
1574 return Err(Error::Schema(format!(
1575 "Binary query vector byte length {} != field byte length {}",
1576 query.len(),
1577 byte_len
1578 )));
1579 }
1580
1581 let mut collector = crate::query::ScoreCollector::new(k);
1582 let mut scores = vec![0f32; BRUTE_FORCE_BATCH];
1583
1584 for batch_start in (0..n).step_by(BRUTE_FORCE_BATCH) {
1585 let batch_count = BRUTE_FORCE_BATCH.min(n - batch_start);
1586 let batch_bytes = lazy_flat
1587 .read_vectors_batch_sync(batch_start, batch_count)
1588 .map_err(crate::Error::Io)?;
1589 let raw = batch_bytes.as_slice();
1590
1591 crate::structures::simd::batch_hamming_scores(
1592 query,
1593 raw,
1594 byte_len,
1595 dim_bits,
1596 &mut scores[..batch_count],
1597 );
1598
1599 let threshold = collector.threshold();
1600 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1601 if score > threshold {
1602 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1603 collector.insert_with_ordinal(doc_id, score, ordinal);
1604 }
1605 }
1606 }
1607
1608 let results: Vec<(u32, u16, f32)> = collector
1609 .into_sorted_results()
1610 .into_iter()
1611 .map(|(doc_id, score, ordinal)| (doc_id, ordinal, score))
1612 .collect();
1613
1614 crate::observe::dense_l1(
1615 self.schema.index_label(),
1616 self.schema.get_field_name(field).unwrap_or("?"),
1617 "binary_flat",
1618 t0.secs(),
1619 results.len(),
1620 );
1621 Ok(combine_ordinal_results(results, combiner, k))
1622 }
1623}