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_raw_blocks(&self) -> Vec<RawStoreBlock> {
320 self.store.raw_blocks()
321 }
322
323 pub fn store_data_slice(&self) -> &LazyFileSlice {
325 self.store.data_slice()
326 }
327
328 pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
330 self.term_dict.all_entries().await.map_err(Error::from)
331 }
332
333 pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
338 let entries = self.term_dict.all_entries().await?;
339 let mut result = Vec::with_capacity(entries.len());
340
341 for (key, term_info) in entries {
342 if key.len() > 4 {
344 let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
345 let term_bytes = &key[4..];
346 if let Ok(term_str) = std::str::from_utf8(term_bytes) {
347 result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
348 }
349 }
350 }
351
352 Ok(result)
353 }
354
355 pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
357 self.term_dict.iter()
358 }
359
360 pub async fn read_postings(&self, offset: u64, len: u32) -> Result<Vec<u8>> {
362 let start = offset;
363 let end = start + len as u64;
364 let bytes = self.postings_handle.read_bytes_range(start..end).await?;
365 Ok(bytes.to_vec())
366 }
367
368 pub fn search_dense_vector(
375 &self,
376 field: Field,
377 query: &[f32],
378 k: usize,
379 rerank_factor: usize,
380 combiner: crate::query::MultiValueCombiner,
381 ) -> Result<Vec<VectorSearchResult>> {
382 let index = self
383 .vector_indexes
384 .get(&field.0)
385 .ok_or_else(|| Error::Schema(format!("No dense vector index for field {}", field.0)))?;
386
387 let mrl_dim = self
389 .schema
390 .get_field_entry(field)
391 .and_then(|e| e.dense_vector_config.as_ref())
392 .and_then(|c| c.mrl_dim);
393
394 let query_vec: Vec<f32>;
396 let effective_query = if let Some(trim_dim) = mrl_dim {
397 if trim_dim < query.len() {
398 query_vec = query[..trim_dim].to_vec();
399 query_vec.as_slice()
400 } else {
401 query
402 }
403 } else {
404 query
405 };
406
407 let results: Vec<(u32, u16, f32)> = match index {
409 VectorIndex::Flat(flat_data) => {
410 use crate::structures::simd::squared_euclidean_distance;
412
413 let mut candidates: Vec<(u32, u16, f32)> = flat_data
414 .vectors
415 .iter()
416 .zip(flat_data.doc_ids.iter())
417 .map(|(vec, &(doc_id, ordinal))| {
418 let dist = squared_euclidean_distance(effective_query, vec);
419 (doc_id, ordinal, dist)
420 })
421 .collect();
422 candidates
423 .sort_by(|a, b| a.2.partial_cmp(&b.2).unwrap_or(std::cmp::Ordering::Equal));
424 candidates.truncate(k);
425 candidates
426 }
427 VectorIndex::RaBitQ(rabitq) => rabitq.search(effective_query, k, rerank_factor),
428 VectorIndex::IVF { index, codebook } => {
429 let centroids = self.coarse_centroids.as_ref().ok_or_else(|| {
430 Error::Schema("IVF index requires coarse centroids".to_string())
431 })?;
432 let nprobe = rerank_factor.max(32); index
434 .search(centroids, codebook, effective_query, k, Some(nprobe))
435 .into_iter()
436 .map(|(doc_id, dist)| (doc_id, 0u16, dist)) .collect()
438 }
439 VectorIndex::ScaNN { index, codebook } => {
440 let centroids = self.coarse_centroids.as_ref().ok_or_else(|| {
441 Error::Schema("ScaNN index requires coarse centroids".to_string())
442 })?;
443 let nprobe = rerank_factor.max(32);
444 index
445 .search(centroids, codebook, effective_query, k, Some(nprobe))
446 .into_iter()
447 .map(|(doc_id, dist)| (doc_id, 0u16, dist)) .collect()
449 }
450 };
451
452 let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
456 rustc_hash::FxHashMap::default();
457 for (doc_id, ordinal, dist) in results {
458 let doc_id = doc_id as DocId + self.doc_id_offset;
459 let score = 1.0 / (1.0 + dist); let ordinals = doc_ordinals.entry(doc_id).or_default();
461 ordinals.push((ordinal as u32, score));
462 }
463
464 let mut final_results: Vec<VectorSearchResult> = doc_ordinals
466 .into_iter()
467 .map(|(doc_id, ordinals)| {
468 let combined_score = combiner.combine(&ordinals);
469 VectorSearchResult::new(doc_id, combined_score, ordinals)
470 })
471 .collect();
472
473 final_results.sort_by(|a, b| {
475 b.score
476 .partial_cmp(&a.score)
477 .unwrap_or(std::cmp::Ordering::Equal)
478 });
479 final_results.truncate(k);
480
481 Ok(final_results)
482 }
483
484 pub fn has_dense_vector_index(&self, field: Field) -> bool {
486 self.vector_indexes.contains_key(&field.0)
487 }
488
489 pub fn get_dense_vector_index(&self, field: Field) -> Option<Arc<RaBitQIndex>> {
491 match self.vector_indexes.get(&field.0) {
492 Some(VectorIndex::RaBitQ(idx)) => Some(idx.clone()),
493 _ => None,
494 }
495 }
496
497 pub fn get_ivf_vector_index(&self, field: Field) -> Option<Arc<IVFRaBitQIndex>> {
499 match self.vector_indexes.get(&field.0) {
500 Some(VectorIndex::IVF { index, .. }) => Some(index.clone()),
501 _ => None,
502 }
503 }
504
505 pub fn get_scann_vector_index(
507 &self,
508 field: Field,
509 ) -> Option<(Arc<IVFPQIndex>, Arc<PQCodebook>)> {
510 match self.vector_indexes.get(&field.0) {
511 Some(VectorIndex::ScaNN { index, codebook }) => Some((index.clone(), codebook.clone())),
512 _ => None,
513 }
514 }
515
516 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
518 self.vector_indexes.get(&field.0)
519 }
520
521 pub async fn search_sparse_vector(
531 &self,
532 field: Field,
533 vector: &[(u32, f32)],
534 limit: usize,
535 combiner: crate::query::MultiValueCombiner,
536 heap_factor: f32,
537 ) -> Result<Vec<VectorSearchResult>> {
538 use crate::query::{BmpExecutor, MaxScoreExecutor, SparseTermScorer, WandExecutor};
539
540 let query_tokens = vector.len();
541
542 let sparse_index = match self.sparse_indexes.get(&field.0) {
544 Some(idx) => idx,
545 None => {
546 log::debug!(
547 "Sparse vector search: no index for field {}, returning empty",
548 field.0
549 );
550 return Ok(Vec::new());
551 }
552 };
553
554 let index_dimensions = sparse_index.num_dimensions();
555
556 let mut matched_tokens = Vec::new();
560 let mut missing_tokens = Vec::new();
561 let mut posting_lists: Vec<(u32, f32, Arc<BlockSparsePostingList>)> =
562 Vec::with_capacity(vector.len());
563
564 for &(dim_id, query_weight) in vector {
565 if !sparse_index.has_dimension(dim_id) {
567 missing_tokens.push(dim_id);
568 continue;
569 }
570
571 match sparse_index.get_posting(dim_id).await? {
573 Some(pl) => {
574 matched_tokens.push(dim_id);
575 posting_lists.push((dim_id, query_weight, pl));
576 }
577 None => {
578 missing_tokens.push(dim_id);
579 }
580 }
581 }
582
583 let scorers: Vec<SparseTermScorer> = posting_lists
585 .iter()
586 .map(|(_, query_weight, pl)| SparseTermScorer::from_arc(pl, *query_weight))
587 .collect();
588
589 log::debug!(
590 "Sparse vector search: query_tokens={}, matched={}, missing={}, index_dimensions={}",
591 query_tokens,
592 matched_tokens.len(),
593 missing_tokens.len(),
594 index_dimensions
595 );
596
597 if log::log_enabled!(log::Level::Debug) {
599 let query_details: Vec<_> = vector
600 .iter()
601 .take(30)
602 .map(|(id, w)| format!("{}:{:.3}", id, w))
603 .collect();
604 log::debug!("Query tokens (id:weight): [{}]", query_details.join(", "));
605 }
606
607 if !matched_tokens.is_empty() {
608 log::debug!(
609 "Matched token IDs: {:?}",
610 matched_tokens.iter().take(20).collect::<Vec<_>>()
611 );
612 }
613
614 if !missing_tokens.is_empty() {
615 log::debug!(
616 "Missing token IDs (not in index): {:?}",
617 missing_tokens.iter().take(20).collect::<Vec<_>>()
618 );
619 }
620
621 if scorers.is_empty() {
622 log::debug!("Sparse vector search: no matching tokens, returning empty");
623 return Ok(Vec::new());
624 }
625
626 let num_terms = scorers.len();
631 let over_fetch = limit * 2; let raw_results = if num_terms > 12 {
633 let pl_refs: Vec<_> = posting_lists
635 .iter()
636 .map(|(_, _, pl)| Arc::clone(pl))
637 .collect();
638 let weights: Vec<_> = posting_lists.iter().map(|(_, qw, _)| *qw).collect();
639 drop(scorers); BmpExecutor::new(pl_refs, weights, over_fetch, heap_factor).execute()
641 } else if num_terms > 6 {
642 MaxScoreExecutor::with_heap_factor(scorers, over_fetch, heap_factor).execute()
643 } else {
644 WandExecutor::with_heap_factor(scorers, over_fetch, heap_factor).execute()
645 };
646
647 log::trace!(
648 "Sparse WAND returned {} raw results for segment (doc_id_offset={})",
649 raw_results.len(),
650 self.doc_id_offset
651 );
652 if log::log_enabled!(log::Level::Trace) && !raw_results.is_empty() {
653 for r in raw_results.iter().take(5) {
654 log::trace!(
655 " Raw result: doc_id={} (global={}), score={:.4}, ordinal={}",
656 r.doc_id,
657 r.doc_id + self.doc_id_offset,
658 r.score,
659 r.ordinal
660 );
661 }
662 }
663
664 let mut doc_ordinals: rustc_hash::FxHashMap<u32, Vec<(u32, f32)>> =
667 rustc_hash::FxHashMap::default();
668 for r in raw_results {
669 let ordinals = doc_ordinals.entry(r.doc_id).or_default();
670 ordinals.push((r.ordinal as u32, r.score));
671 }
672
673 let mut results: Vec<VectorSearchResult> = doc_ordinals
676 .into_iter()
677 .map(|(doc_id, ordinals)| {
678 let global_doc_id = doc_id + self.doc_id_offset;
679 let combined_score = combiner.combine(&ordinals);
680 VectorSearchResult::new(global_doc_id, combined_score, ordinals)
681 })
682 .collect();
683
684 results.sort_by(|a, b| {
686 b.score
687 .partial_cmp(&a.score)
688 .unwrap_or(std::cmp::Ordering::Equal)
689 });
690 results.truncate(limit);
691
692 Ok(results)
693 }
694
695 pub async fn get_positions(
700 &self,
701 field: Field,
702 term: &[u8],
703 ) -> Result<Option<crate::structures::PositionPostingList>> {
704 use std::io::Cursor;
705
706 let handle = match &self.positions_handle {
708 Some(h) => h,
709 None => return Ok(None),
710 };
711
712 let mut key = Vec::with_capacity(4 + term.len());
714 key.extend_from_slice(&field.0.to_le_bytes());
715 key.extend_from_slice(term);
716
717 let term_info = match self.term_dict.get(&key).await? {
719 Some(info) => info,
720 None => return Ok(None),
721 };
722
723 let (offset, length) = match term_info.position_info() {
725 Some((o, l)) => (o, l),
726 None => return Ok(None),
727 };
728
729 let slice = handle.slice(offset..offset + length as u64);
731 let data = slice.read_bytes().await?;
732
733 let mut cursor = Cursor::new(data.as_slice());
735 let pos_list = crate::structures::PositionPostingList::deserialize(&mut cursor)?;
736
737 Ok(Some(pos_list))
738 }
739
740 pub fn has_positions(&self, field: Field) -> bool {
742 if let Some(entry) = self.schema.get_field_entry(field) {
744 entry.positions.is_some()
745 } else {
746 false
747 }
748 }
749}
750
751pub type SegmentReader = AsyncSegmentReader;