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 read_postings(&self, offset: u64, len: u32) -> Result<Vec<u8>> {
367 let start = offset;
368 let end = start + len as u64;
369 let bytes = self.postings_handle.read_bytes_range(start..end).await?;
370 Ok(bytes.to_vec())
371 }
372
373 pub async fn read_position_bytes(&self, offset: u64, len: u32) -> Result<Option<Vec<u8>>> {
375 let handle = match &self.positions_handle {
376 Some(h) => h,
377 None => return Ok(None),
378 };
379 let start = offset;
380 let end = start + len as u64;
381 let bytes = handle.read_bytes_range(start..end).await?;
382 Ok(Some(bytes.to_vec()))
383 }
384
385 pub fn has_positions_file(&self) -> bool {
387 self.positions_handle.is_some()
388 }
389
390 pub fn search_dense_vector(
397 &self,
398 field: Field,
399 query: &[f32],
400 k: usize,
401 rerank_factor: usize,
402 combiner: crate::query::MultiValueCombiner,
403 ) -> Result<Vec<VectorSearchResult>> {
404 let index = self
405 .vector_indexes
406 .get(&field.0)
407 .ok_or_else(|| Error::Schema(format!("No dense vector index for field {}", field.0)))?;
408
409 let mrl_dim = self
411 .schema
412 .get_field_entry(field)
413 .and_then(|e| e.dense_vector_config.as_ref())
414 .and_then(|c| c.mrl_dim);
415
416 let query_vec: Vec<f32>;
418 let effective_query = if let Some(trim_dim) = mrl_dim {
419 if trim_dim < query.len() {
420 query_vec = query[..trim_dim].to_vec();
421 query_vec.as_slice()
422 } else {
423 query
424 }
425 } else {
426 query
427 };
428
429 let results: Vec<(u32, u16, f32)> = match index {
431 VectorIndex::Flat(flat_data) => {
432 use crate::structures::simd::squared_euclidean_distance;
434
435 let mut candidates: Vec<(u32, u16, f32)> = flat_data
436 .vectors
437 .iter()
438 .zip(flat_data.doc_ids.iter())
439 .map(|(vec, &(doc_id, ordinal))| {
440 let dist = squared_euclidean_distance(effective_query, vec);
441 (doc_id, ordinal, dist)
442 })
443 .collect();
444 candidates
445 .sort_by(|a, b| a.2.partial_cmp(&b.2).unwrap_or(std::cmp::Ordering::Equal));
446 candidates.truncate(k);
447 candidates
448 }
449 VectorIndex::RaBitQ(rabitq) => rabitq.search(effective_query, k, rerank_factor),
450 VectorIndex::IVF { index, codebook } => {
451 let centroids = self.coarse_centroids.as_ref().ok_or_else(|| {
452 Error::Schema("IVF index requires coarse centroids".to_string())
453 })?;
454 let nprobe = rerank_factor.max(32); index
456 .search(centroids, codebook, effective_query, k, Some(nprobe))
457 .into_iter()
458 .map(|(doc_id, dist)| (doc_id, 0u16, dist)) .collect()
460 }
461 VectorIndex::ScaNN { index, codebook } => {
462 let centroids = self.coarse_centroids.as_ref().ok_or_else(|| {
463 Error::Schema("ScaNN index requires coarse centroids".to_string())
464 })?;
465 let nprobe = rerank_factor.max(32);
466 index
467 .search(centroids, codebook, effective_query, k, Some(nprobe))
468 .into_iter()
469 .map(|(doc_id, dist)| (doc_id, 0u16, dist)) .collect()
471 }
472 };
473
474 let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
478 rustc_hash::FxHashMap::default();
479 for (doc_id, ordinal, dist) in results {
480 let score = 1.0 / (1.0 + dist); let ordinals = doc_ordinals.entry(doc_id as DocId).or_default();
482 ordinals.push((ordinal as u32, score));
483 }
484
485 let mut final_results: Vec<VectorSearchResult> = doc_ordinals
487 .into_iter()
488 .map(|(doc_id, ordinals)| {
489 let combined_score = combiner.combine(&ordinals);
490 VectorSearchResult::new(doc_id, combined_score, ordinals)
491 })
492 .collect();
493
494 final_results.sort_by(|a, b| {
496 b.score
497 .partial_cmp(&a.score)
498 .unwrap_or(std::cmp::Ordering::Equal)
499 });
500 final_results.truncate(k);
501
502 Ok(final_results)
503 }
504
505 pub fn has_dense_vector_index(&self, field: Field) -> bool {
507 self.vector_indexes.contains_key(&field.0)
508 }
509
510 pub fn get_dense_vector_index(&self, field: Field) -> Option<Arc<RaBitQIndex>> {
512 match self.vector_indexes.get(&field.0) {
513 Some(VectorIndex::RaBitQ(idx)) => Some(idx.clone()),
514 _ => None,
515 }
516 }
517
518 pub fn get_ivf_vector_index(&self, field: Field) -> Option<Arc<IVFRaBitQIndex>> {
520 match self.vector_indexes.get(&field.0) {
521 Some(VectorIndex::IVF { index, .. }) => Some(index.clone()),
522 _ => None,
523 }
524 }
525
526 pub fn get_scann_vector_index(
528 &self,
529 field: Field,
530 ) -> Option<(Arc<IVFPQIndex>, Arc<PQCodebook>)> {
531 match self.vector_indexes.get(&field.0) {
532 Some(VectorIndex::ScaNN { index, codebook }) => Some((index.clone(), codebook.clone())),
533 _ => None,
534 }
535 }
536
537 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
539 self.vector_indexes.get(&field.0)
540 }
541
542 pub async fn search_sparse_vector(
552 &self,
553 field: Field,
554 vector: &[(u32, f32)],
555 limit: usize,
556 combiner: crate::query::MultiValueCombiner,
557 heap_factor: f32,
558 ) -> Result<Vec<VectorSearchResult>> {
559 use crate::query::{BlockMaxScoreExecutor, BmpExecutor, SparseTermScorer};
560
561 let query_tokens = vector.len();
562
563 let sparse_index = match self.sparse_indexes.get(&field.0) {
565 Some(idx) => idx,
566 None => {
567 log::debug!(
568 "Sparse vector search: no index for field {}, returning empty",
569 field.0
570 );
571 return Ok(Vec::new());
572 }
573 };
574
575 let index_dimensions = sparse_index.num_dimensions();
576
577 let mut matched_tokens = Vec::new();
581 let mut missing_tokens = Vec::new();
582 let mut posting_lists: Vec<(u32, f32, Arc<BlockSparsePostingList>)> =
583 Vec::with_capacity(vector.len());
584
585 for &(dim_id, query_weight) in vector {
586 if !sparse_index.has_dimension(dim_id) {
588 missing_tokens.push(dim_id);
589 continue;
590 }
591
592 match sparse_index.get_posting(dim_id).await? {
594 Some(pl) => {
595 matched_tokens.push(dim_id);
596 posting_lists.push((dim_id, query_weight, pl));
597 }
598 None => {
599 missing_tokens.push(dim_id);
600 }
601 }
602 }
603
604 let scorers: Vec<SparseTermScorer> = posting_lists
606 .iter()
607 .map(|(_, query_weight, pl)| SparseTermScorer::from_arc(pl, *query_weight))
608 .collect();
609
610 log::debug!(
611 "Sparse vector search: query_tokens={}, matched={}, missing={}, index_dimensions={}",
612 query_tokens,
613 matched_tokens.len(),
614 missing_tokens.len(),
615 index_dimensions
616 );
617
618 if log::log_enabled!(log::Level::Debug) {
620 let query_details: Vec<_> = vector
621 .iter()
622 .take(30)
623 .map(|(id, w)| format!("{}:{:.3}", id, w))
624 .collect();
625 log::debug!("Query tokens (id:weight): [{}]", query_details.join(", "));
626 }
627
628 if !matched_tokens.is_empty() {
629 log::debug!(
630 "Matched token IDs: {:?}",
631 matched_tokens.iter().take(20).collect::<Vec<_>>()
632 );
633 }
634
635 if !missing_tokens.is_empty() {
636 log::debug!(
637 "Missing token IDs (not in index): {:?}",
638 missing_tokens.iter().take(20).collect::<Vec<_>>()
639 );
640 }
641
642 if scorers.is_empty() {
643 log::debug!("Sparse vector search: no matching tokens, returning empty");
644 return Ok(Vec::new());
645 }
646
647 let num_terms = scorers.len();
651 let over_fetch = limit * 2; let raw_results = if num_terms > 12 {
653 let pl_refs: Vec<_> = posting_lists
655 .iter()
656 .map(|(_, _, pl)| Arc::clone(pl))
657 .collect();
658 let weights: Vec<_> = posting_lists.iter().map(|(_, qw, _)| *qw).collect();
659 drop(scorers); BmpExecutor::new(pl_refs, weights, over_fetch, heap_factor).execute()
661 } else {
662 BlockMaxScoreExecutor::with_heap_factor(scorers, over_fetch, heap_factor).execute()
663 };
664
665 log::trace!(
666 "Sparse WAND returned {} raw results for segment (doc_id_offset={})",
667 raw_results.len(),
668 self.doc_id_offset
669 );
670 if log::log_enabled!(log::Level::Trace) && !raw_results.is_empty() {
671 for r in raw_results.iter().take(5) {
672 log::trace!(
673 " Raw result: doc_id={} (global={}), score={:.4}, ordinal={}",
674 r.doc_id,
675 r.doc_id + self.doc_id_offset,
676 r.score,
677 r.ordinal
678 );
679 }
680 }
681
682 let mut doc_ordinals: rustc_hash::FxHashMap<u32, Vec<(u32, f32)>> =
685 rustc_hash::FxHashMap::default();
686 for r in raw_results {
687 let ordinals = doc_ordinals.entry(r.doc_id).or_default();
688 ordinals.push((r.ordinal as u32, r.score));
689 }
690
691 let mut results: Vec<VectorSearchResult> = doc_ordinals
694 .into_iter()
695 .map(|(doc_id, ordinals)| {
696 let combined_score = combiner.combine(&ordinals);
697 VectorSearchResult::new(doc_id, combined_score, ordinals)
698 })
699 .collect();
700
701 results.sort_by(|a, b| {
703 b.score
704 .partial_cmp(&a.score)
705 .unwrap_or(std::cmp::Ordering::Equal)
706 });
707 results.truncate(limit);
708
709 Ok(results)
710 }
711
712 pub async fn get_positions(
717 &self,
718 field: Field,
719 term: &[u8],
720 ) -> Result<Option<crate::structures::PositionPostingList>> {
721 use std::io::Cursor;
722
723 let handle = match &self.positions_handle {
725 Some(h) => h,
726 None => return Ok(None),
727 };
728
729 let mut key = Vec::with_capacity(4 + term.len());
731 key.extend_from_slice(&field.0.to_le_bytes());
732 key.extend_from_slice(term);
733
734 let term_info = match self.term_dict.get(&key).await? {
736 Some(info) => info,
737 None => return Ok(None),
738 };
739
740 let (offset, length) = match term_info.position_info() {
742 Some((o, l)) => (o, l),
743 None => return Ok(None),
744 };
745
746 let slice = handle.slice(offset..offset + length as u64);
748 let data = slice.read_bytes().await?;
749
750 let mut cursor = Cursor::new(data.as_slice());
752 let pos_list = crate::structures::PositionPostingList::deserialize(&mut cursor)?;
753
754 Ok(Some(pos_list))
755 }
756
757 pub fn has_positions(&self, field: Field) -> bool {
759 if let Some(entry) = self.schema.get_field_entry(field) {
761 entry.positions.is_some()
762 } else {
763 false
764 }
765 }
766}
767
768pub type SegmentReader = AsyncSegmentReader;