1mod loader;
4mod types;
5
6pub use types::{SparseIndex, VectorIndex, VectorSearchResult};
7
8#[derive(Debug, Clone, Default)]
10pub struct SegmentMemoryStats {
11 pub segment_id: u128,
13 pub num_docs: u32,
15 pub term_dict_cache_bytes: usize,
17 pub store_cache_bytes: usize,
19 pub sparse_index_bytes: usize,
21 pub dense_index_bytes: usize,
23 pub bloom_filter_bytes: usize,
25}
26
27impl SegmentMemoryStats {
28 pub fn total_bytes(&self) -> usize {
30 self.term_dict_cache_bytes
31 + self.store_cache_bytes
32 + self.sparse_index_bytes
33 + self.dense_index_bytes
34 + self.bloom_filter_bytes
35 }
36}
37
38use crate::structures::BlockSparsePostingList;
39
40use std::sync::Arc;
41
42use rustc_hash::FxHashMap;
43
44use crate::directories::{AsyncFileRead, Directory, LazyFileHandle, LazyFileSlice};
45use crate::dsl::{Document, Field, Schema};
46use crate::structures::{
47 AsyncSSTableReader, BlockPostingList, CoarseCentroids, IVFPQIndex, IVFRaBitQIndex, PQCodebook,
48 RaBitQIndex, SSTableStats, TermInfo,
49};
50use crate::{DocId, Error, Result};
51
52use super::store::{AsyncStoreReader, RawStoreBlock};
53use super::types::{SegmentFiles, SegmentId, SegmentMeta};
54
55pub struct AsyncSegmentReader {
61 meta: SegmentMeta,
62 term_dict: Arc<AsyncSSTableReader<TermInfo>>,
64 postings_handle: LazyFileHandle,
66 store: Arc<AsyncStoreReader>,
68 schema: Arc<Schema>,
69 doc_id_offset: DocId,
71 vector_indexes: FxHashMap<u32, VectorIndex>,
73 coarse_centroids: Option<Arc<CoarseCentroids>>,
75 sparse_indexes: FxHashMap<u32, SparseIndex>,
77 positions_handle: Option<LazyFileHandle>,
79}
80
81impl AsyncSegmentReader {
82 pub async fn open<D: Directory>(
84 dir: &D,
85 segment_id: SegmentId,
86 schema: Arc<Schema>,
87 doc_id_offset: DocId,
88 cache_blocks: usize,
89 ) -> Result<Self> {
90 let files = SegmentFiles::new(segment_id.0);
91
92 let meta_slice = dir.open_read(&files.meta).await?;
94 let meta_bytes = meta_slice.read_bytes().await?;
95 let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
96 debug_assert_eq!(meta.id, segment_id.0);
97
98 let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
100 let term_dict = AsyncSSTableReader::open(term_dict_handle, cache_blocks).await?;
101
102 let postings_handle = dir.open_lazy(&files.postings).await?;
104
105 let store_handle = dir.open_lazy(&files.store).await?;
107 let store = AsyncStoreReader::open(store_handle, cache_blocks).await?;
108
109 let (vector_indexes, coarse_centroids) =
111 loader::load_vectors_file(dir, &files, &schema).await?;
112
113 let sparse_indexes = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
115
116 let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
118
119 let sparse_dims: usize = sparse_indexes.values().map(|s| s.num_dimensions()).sum();
121 let sparse_mem = sparse_dims * 24; log::debug!(
123 "[segment] loaded {:016x}: docs={}, sparse_dims={}, sparse_mem={:.2} KB, vectors={}",
124 segment_id.0,
125 meta.num_docs,
126 sparse_dims,
127 sparse_mem as f64 / 1024.0,
128 vector_indexes.len()
129 );
130
131 Ok(Self {
132 meta,
133 term_dict: Arc::new(term_dict),
134 postings_handle,
135 store: Arc::new(store),
136 schema,
137 doc_id_offset,
138 vector_indexes,
139 coarse_centroids,
140 sparse_indexes,
141 positions_handle,
142 })
143 }
144
145 pub fn meta(&self) -> &SegmentMeta {
146 &self.meta
147 }
148
149 pub fn num_docs(&self) -> u32 {
150 self.meta.num_docs
151 }
152
153 pub fn avg_field_len(&self, field: Field) -> f32 {
155 self.meta.avg_field_len(field)
156 }
157
158 pub fn doc_id_offset(&self) -> DocId {
159 self.doc_id_offset
160 }
161
162 pub fn set_doc_id_offset(&mut self, offset: DocId) {
164 self.doc_id_offset = offset;
165 }
166
167 pub fn schema(&self) -> &Schema {
168 &self.schema
169 }
170
171 pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
173 &self.sparse_indexes
174 }
175
176 pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
178 &self.vector_indexes
179 }
180
181 pub fn term_dict_stats(&self) -> SSTableStats {
183 self.term_dict.stats()
184 }
185
186 pub fn memory_stats(&self) -> SegmentMemoryStats {
188 let term_dict_stats = self.term_dict.stats();
189
190 let term_dict_cache_bytes = self.term_dict.cached_blocks() * 4096;
192
193 let store_cache_bytes = self.store.cached_blocks() * 4096;
195
196 let sparse_index_bytes: usize = self
199 .sparse_indexes
200 .values()
201 .map(|s| s.num_dimensions() * 24)
202 .sum();
203
204 let dense_index_bytes: usize = self
207 .vector_indexes
208 .values()
209 .map(|v| v.estimated_memory_bytes())
210 .sum();
211
212 SegmentMemoryStats {
213 segment_id: self.meta.id,
214 num_docs: self.meta.num_docs,
215 term_dict_cache_bytes,
216 store_cache_bytes,
217 sparse_index_bytes,
218 dense_index_bytes,
219 bloom_filter_bytes: term_dict_stats.bloom_filter_size,
220 }
221 }
222
223 pub async fn get_postings(
228 &self,
229 field: Field,
230 term: &[u8],
231 ) -> Result<Option<BlockPostingList>> {
232 log::debug!(
233 "SegmentReader::get_postings field={} term_len={}",
234 field.0,
235 term.len()
236 );
237
238 let mut key = Vec::with_capacity(4 + term.len());
240 key.extend_from_slice(&field.0.to_le_bytes());
241 key.extend_from_slice(term);
242
243 let term_info = match self.term_dict.get(&key).await? {
245 Some(info) => {
246 log::debug!("SegmentReader::get_postings found term_info");
247 info
248 }
249 None => {
250 log::debug!("SegmentReader::get_postings term not found");
251 return Ok(None);
252 }
253 };
254
255 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
257 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
259 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs.into_iter()) {
260 posting_list.push(doc_id, tf);
261 }
262 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
263 return Ok(Some(block_list));
264 }
265
266 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
268 Error::Corruption("TermInfo has neither inline nor external data".to_string())
269 })?;
270
271 let start = posting_offset;
272 let end = start + posting_len as u64;
273
274 if end > self.postings_handle.len() {
275 return Err(Error::Corruption(
276 "Posting offset out of bounds".to_string(),
277 ));
278 }
279
280 let posting_bytes = self.postings_handle.read_bytes_range(start..end).await?;
281 let block_list = BlockPostingList::deserialize(&mut posting_bytes.as_slice())?;
282
283 Ok(Some(block_list))
284 }
285
286 pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
288 self.store
289 .get(local_doc_id, &self.schema)
290 .await
291 .map_err(Error::from)
292 }
293
294 pub async fn prefetch_terms(
296 &self,
297 field: Field,
298 start_term: &[u8],
299 end_term: &[u8],
300 ) -> Result<()> {
301 let mut start_key = Vec::with_capacity(4 + start_term.len());
302 start_key.extend_from_slice(&field.0.to_le_bytes());
303 start_key.extend_from_slice(start_term);
304
305 let mut end_key = Vec::with_capacity(4 + end_term.len());
306 end_key.extend_from_slice(&field.0.to_le_bytes());
307 end_key.extend_from_slice(end_term);
308
309 self.term_dict.prefetch_range(&start_key, &end_key).await?;
310 Ok(())
311 }
312
313 pub fn store_has_dict(&self) -> bool {
315 self.store.has_dict()
316 }
317
318 pub fn store(&self) -> &super::store::AsyncStoreReader {
320 &self.store
321 }
322
323 pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
325 self.store.raw_blocks()
326 }
327
328 pub fn store_data_slice(&self) -> &LazyFileSlice {
330 self.store.data_slice()
331 }
332
333 pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
335 self.term_dict.all_entries().await.map_err(Error::from)
336 }
337
338 pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
343 let entries = self.term_dict.all_entries().await?;
344 let mut result = Vec::with_capacity(entries.len());
345
346 for (key, term_info) in entries {
347 if key.len() > 4 {
349 let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
350 let term_bytes = &key[4..];
351 if let Ok(term_str) = std::str::from_utf8(term_bytes) {
352 result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
353 }
354 }
355 }
356
357 Ok(result)
358 }
359
360 pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
362 self.term_dict.iter()
363 }
364
365 pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
369 self.term_dict
370 .prefetch_all_data_bulk()
371 .await
372 .map_err(crate::Error::from)
373 }
374
375 pub async fn read_postings(&self, offset: u64, len: u32) -> Result<Vec<u8>> {
377 let start = offset;
378 let end = start + len as u64;
379 let bytes = self.postings_handle.read_bytes_range(start..end).await?;
380 Ok(bytes.to_vec())
381 }
382
383 pub async fn read_position_bytes(&self, offset: u64, len: u32) -> Result<Option<Vec<u8>>> {
385 let handle = match &self.positions_handle {
386 Some(h) => h,
387 None => return Ok(None),
388 };
389 let start = offset;
390 let end = start + len as u64;
391 let bytes = handle.read_bytes_range(start..end).await?;
392 Ok(Some(bytes.to_vec()))
393 }
394
395 pub fn has_positions_file(&self) -> bool {
397 self.positions_handle.is_some()
398 }
399
400 pub fn search_dense_vector(
407 &self,
408 field: Field,
409 query: &[f32],
410 k: usize,
411 rerank_factor: usize,
412 combiner: crate::query::MultiValueCombiner,
413 ) -> Result<Vec<VectorSearchResult>> {
414 let index = self
415 .vector_indexes
416 .get(&field.0)
417 .ok_or_else(|| Error::Schema(format!("No dense vector index for field {}", field.0)))?;
418
419 let mrl_dim = self
421 .schema
422 .get_field_entry(field)
423 .and_then(|e| e.dense_vector_config.as_ref())
424 .and_then(|c| c.mrl_dim);
425
426 let query_vec: Vec<f32>;
428 let effective_query = if let Some(trim_dim) = mrl_dim {
429 if trim_dim < query.len() {
430 query_vec = query[..trim_dim].to_vec();
431 query_vec.as_slice()
432 } else {
433 query
434 }
435 } else {
436 query
437 };
438
439 let results: Vec<(u32, u16, f32)> = match index {
441 VectorIndex::Flat(flat_data) => {
442 use crate::structures::simd::squared_euclidean_distance;
444
445 let mut candidates: Vec<(u32, u16, f32)> = flat_data
446 .vectors
447 .iter()
448 .zip(flat_data.doc_ids.iter())
449 .map(|(vec, &(doc_id, ordinal))| {
450 let dist = squared_euclidean_distance(effective_query, vec);
451 (doc_id, ordinal, dist)
452 })
453 .collect();
454 candidates
455 .sort_by(|a, b| a.2.partial_cmp(&b.2).unwrap_or(std::cmp::Ordering::Equal));
456 candidates.truncate(k * rerank_factor.max(1));
459 candidates
460 }
461 VectorIndex::RaBitQ(rabitq) => rabitq.search(effective_query, k, rerank_factor),
462 VectorIndex::IVF { index, codebook } => {
463 let centroids = self.coarse_centroids.as_ref().ok_or_else(|| {
464 Error::Schema("IVF index requires coarse centroids".to_string())
465 })?;
466 let nprobe = rerank_factor.max(32); index.search(centroids, codebook, effective_query, k, Some(nprobe))
468 }
469 VectorIndex::ScaNN { index, codebook } => {
470 let centroids = self.coarse_centroids.as_ref().ok_or_else(|| {
471 Error::Schema("ScaNN index requires coarse centroids".to_string())
472 })?;
473 let nprobe = rerank_factor.max(32);
474 index.search(centroids, codebook, effective_query, k, Some(nprobe))
475 }
476 };
477
478 let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
482 rustc_hash::FxHashMap::default();
483 for (doc_id, ordinal, dist) in results {
484 let score = 1.0 / (1.0 + dist); let ordinals = doc_ordinals.entry(doc_id as DocId).or_default();
486 ordinals.push((ordinal as u32, score));
487 }
488
489 let mut final_results: Vec<VectorSearchResult> = doc_ordinals
491 .into_iter()
492 .map(|(doc_id, ordinals)| {
493 let combined_score = combiner.combine(&ordinals);
494 VectorSearchResult::new(doc_id, combined_score, ordinals)
495 })
496 .collect();
497
498 final_results.sort_by(|a, b| {
500 b.score
501 .partial_cmp(&a.score)
502 .unwrap_or(std::cmp::Ordering::Equal)
503 });
504 final_results.truncate(k);
505
506 Ok(final_results)
507 }
508
509 pub fn has_dense_vector_index(&self, field: Field) -> bool {
511 self.vector_indexes.contains_key(&field.0)
512 }
513
514 pub fn get_dense_vector_index(&self, field: Field) -> Option<Arc<RaBitQIndex>> {
516 match self.vector_indexes.get(&field.0) {
517 Some(VectorIndex::RaBitQ(idx)) => Some(idx.clone()),
518 _ => None,
519 }
520 }
521
522 pub fn get_ivf_vector_index(&self, field: Field) -> Option<Arc<IVFRaBitQIndex>> {
524 match self.vector_indexes.get(&field.0) {
525 Some(VectorIndex::IVF { index, .. }) => Some(index.clone()),
526 _ => None,
527 }
528 }
529
530 pub fn get_scann_vector_index(
532 &self,
533 field: Field,
534 ) -> Option<(Arc<IVFPQIndex>, Arc<PQCodebook>)> {
535 match self.vector_indexes.get(&field.0) {
536 Some(VectorIndex::ScaNN { index, codebook }) => Some((index.clone(), codebook.clone())),
537 _ => None,
538 }
539 }
540
541 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
543 self.vector_indexes.get(&field.0)
544 }
545
546 pub async fn search_sparse_vector(
556 &self,
557 field: Field,
558 vector: &[(u32, f32)],
559 limit: usize,
560 combiner: crate::query::MultiValueCombiner,
561 heap_factor: f32,
562 ) -> Result<Vec<VectorSearchResult>> {
563 use crate::query::{BlockMaxScoreExecutor, BmpExecutor, SparseTermScorer};
564
565 let query_tokens = vector.len();
566
567 let sparse_index = match self.sparse_indexes.get(&field.0) {
569 Some(idx) => idx,
570 None => {
571 log::debug!(
572 "Sparse vector search: no index for field {}, returning empty",
573 field.0
574 );
575 return Ok(Vec::new());
576 }
577 };
578
579 let index_dimensions = sparse_index.num_dimensions();
580
581 let mut matched_tokens = Vec::new();
585 let mut missing_tokens = Vec::new();
586 let mut posting_lists: Vec<(u32, f32, Arc<BlockSparsePostingList>)> =
587 Vec::with_capacity(vector.len());
588
589 for &(dim_id, query_weight) in vector {
590 if !sparse_index.has_dimension(dim_id) {
592 missing_tokens.push(dim_id);
593 continue;
594 }
595
596 match sparse_index.get_posting(dim_id).await? {
598 Some(pl) => {
599 matched_tokens.push(dim_id);
600 posting_lists.push((dim_id, query_weight, pl));
601 }
602 None => {
603 missing_tokens.push(dim_id);
604 }
605 }
606 }
607
608 let scorers: Vec<SparseTermScorer> = posting_lists
610 .iter()
611 .map(|(_, query_weight, pl)| SparseTermScorer::from_arc(pl, *query_weight))
612 .collect();
613
614 log::debug!(
615 "Sparse vector search: query_tokens={}, matched={}, missing={}, index_dimensions={}",
616 query_tokens,
617 matched_tokens.len(),
618 missing_tokens.len(),
619 index_dimensions
620 );
621
622 if log::log_enabled!(log::Level::Debug) {
624 let query_details: Vec<_> = vector
625 .iter()
626 .take(30)
627 .map(|(id, w)| format!("{}:{:.3}", id, w))
628 .collect();
629 log::debug!("Query tokens (id:weight): [{}]", query_details.join(", "));
630 }
631
632 if !matched_tokens.is_empty() {
633 log::debug!(
634 "Matched token IDs: {:?}",
635 matched_tokens.iter().take(20).collect::<Vec<_>>()
636 );
637 }
638
639 if !missing_tokens.is_empty() {
640 log::debug!(
641 "Missing token IDs (not in index): {:?}",
642 missing_tokens.iter().take(20).collect::<Vec<_>>()
643 );
644 }
645
646 if scorers.is_empty() {
647 log::debug!("Sparse vector search: no matching tokens, returning empty");
648 return Ok(Vec::new());
649 }
650
651 let num_terms = scorers.len();
655 let over_fetch = limit * 2; let raw_results = if num_terms > 12 {
657 let pl_refs: Vec<_> = posting_lists
659 .iter()
660 .map(|(_, _, pl)| Arc::clone(pl))
661 .collect();
662 let weights: Vec<_> = posting_lists.iter().map(|(_, qw, _)| *qw).collect();
663 drop(scorers); BmpExecutor::new(pl_refs, weights, over_fetch, heap_factor).execute()
665 } else {
666 BlockMaxScoreExecutor::with_heap_factor(scorers, over_fetch, heap_factor).execute()
667 };
668
669 log::trace!(
670 "Sparse WAND returned {} raw results for segment (doc_id_offset={})",
671 raw_results.len(),
672 self.doc_id_offset
673 );
674 if log::log_enabled!(log::Level::Trace) && !raw_results.is_empty() {
675 for r in raw_results.iter().take(5) {
676 log::trace!(
677 " Raw result: doc_id={} (global={}), score={:.4}, ordinal={}",
678 r.doc_id,
679 r.doc_id + self.doc_id_offset,
680 r.score,
681 r.ordinal
682 );
683 }
684 }
685
686 let mut doc_ordinals: rustc_hash::FxHashMap<u32, Vec<(u32, f32)>> =
689 rustc_hash::FxHashMap::default();
690 for r in raw_results {
691 let ordinals = doc_ordinals.entry(r.doc_id).or_default();
692 ordinals.push((r.ordinal as u32, r.score));
693 }
694
695 let mut results: Vec<VectorSearchResult> = doc_ordinals
698 .into_iter()
699 .map(|(doc_id, ordinals)| {
700 let combined_score = combiner.combine(&ordinals);
701 VectorSearchResult::new(doc_id, combined_score, ordinals)
702 })
703 .collect();
704
705 results.sort_by(|a, b| {
707 b.score
708 .partial_cmp(&a.score)
709 .unwrap_or(std::cmp::Ordering::Equal)
710 });
711 results.truncate(limit);
712
713 Ok(results)
714 }
715
716 pub async fn get_positions(
721 &self,
722 field: Field,
723 term: &[u8],
724 ) -> Result<Option<crate::structures::PositionPostingList>> {
725 use std::io::Cursor;
726
727 let handle = match &self.positions_handle {
729 Some(h) => h,
730 None => return Ok(None),
731 };
732
733 let mut key = Vec::with_capacity(4 + term.len());
735 key.extend_from_slice(&field.0.to_le_bytes());
736 key.extend_from_slice(term);
737
738 let term_info = match self.term_dict.get(&key).await? {
740 Some(info) => info,
741 None => return Ok(None),
742 };
743
744 let (offset, length) = match term_info.position_info() {
746 Some((o, l)) => (o, l),
747 None => return Ok(None),
748 };
749
750 let slice = handle.slice(offset..offset + length as u64);
752 let data = slice.read_bytes().await?;
753
754 let mut cursor = Cursor::new(data.as_slice());
756 let pos_list = crate::structures::PositionPostingList::deserialize(&mut cursor)?;
757
758 Ok(Some(pos_list))
759 }
760
761 pub fn has_positions(&self, field: Field) -> bool {
763 if let Some(entry) = self.schema.get_field_entry(field) {
765 entry.positions.is_some()
766 } else {
767 false
768 }
769 }
770}
771
772pub type SegmentReader = AsyncSegmentReader;