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 score = 1.0 / (1.0 + dist); let ordinals = doc_ordinals.entry(doc_id as DocId).or_default();
460 ordinals.push((ordinal as u32, score));
461 }
462
463 let mut final_results: Vec<VectorSearchResult> = doc_ordinals
465 .into_iter()
466 .map(|(doc_id, ordinals)| {
467 let combined_score = combiner.combine(&ordinals);
468 VectorSearchResult::new(doc_id, combined_score, ordinals)
469 })
470 .collect();
471
472 final_results.sort_by(|a, b| {
474 b.score
475 .partial_cmp(&a.score)
476 .unwrap_or(std::cmp::Ordering::Equal)
477 });
478 final_results.truncate(k);
479
480 Ok(final_results)
481 }
482
483 pub fn has_dense_vector_index(&self, field: Field) -> bool {
485 self.vector_indexes.contains_key(&field.0)
486 }
487
488 pub fn get_dense_vector_index(&self, field: Field) -> Option<Arc<RaBitQIndex>> {
490 match self.vector_indexes.get(&field.0) {
491 Some(VectorIndex::RaBitQ(idx)) => Some(idx.clone()),
492 _ => None,
493 }
494 }
495
496 pub fn get_ivf_vector_index(&self, field: Field) -> Option<Arc<IVFRaBitQIndex>> {
498 match self.vector_indexes.get(&field.0) {
499 Some(VectorIndex::IVF { index, .. }) => Some(index.clone()),
500 _ => None,
501 }
502 }
503
504 pub fn get_scann_vector_index(
506 &self,
507 field: Field,
508 ) -> Option<(Arc<IVFPQIndex>, Arc<PQCodebook>)> {
509 match self.vector_indexes.get(&field.0) {
510 Some(VectorIndex::ScaNN { index, codebook }) => Some((index.clone(), codebook.clone())),
511 _ => None,
512 }
513 }
514
515 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
517 self.vector_indexes.get(&field.0)
518 }
519
520 pub async fn search_sparse_vector(
530 &self,
531 field: Field,
532 vector: &[(u32, f32)],
533 limit: usize,
534 combiner: crate::query::MultiValueCombiner,
535 heap_factor: f32,
536 ) -> Result<Vec<VectorSearchResult>> {
537 use crate::query::{BlockMaxScoreExecutor, BmpExecutor, SparseTermScorer};
538
539 let query_tokens = vector.len();
540
541 let sparse_index = match self.sparse_indexes.get(&field.0) {
543 Some(idx) => idx,
544 None => {
545 log::debug!(
546 "Sparse vector search: no index for field {}, returning empty",
547 field.0
548 );
549 return Ok(Vec::new());
550 }
551 };
552
553 let index_dimensions = sparse_index.num_dimensions();
554
555 let mut matched_tokens = Vec::new();
559 let mut missing_tokens = Vec::new();
560 let mut posting_lists: Vec<(u32, f32, Arc<BlockSparsePostingList>)> =
561 Vec::with_capacity(vector.len());
562
563 for &(dim_id, query_weight) in vector {
564 if !sparse_index.has_dimension(dim_id) {
566 missing_tokens.push(dim_id);
567 continue;
568 }
569
570 match sparse_index.get_posting(dim_id).await? {
572 Some(pl) => {
573 matched_tokens.push(dim_id);
574 posting_lists.push((dim_id, query_weight, pl));
575 }
576 None => {
577 missing_tokens.push(dim_id);
578 }
579 }
580 }
581
582 let scorers: Vec<SparseTermScorer> = posting_lists
584 .iter()
585 .map(|(_, query_weight, pl)| SparseTermScorer::from_arc(pl, *query_weight))
586 .collect();
587
588 log::debug!(
589 "Sparse vector search: query_tokens={}, matched={}, missing={}, index_dimensions={}",
590 query_tokens,
591 matched_tokens.len(),
592 missing_tokens.len(),
593 index_dimensions
594 );
595
596 if log::log_enabled!(log::Level::Debug) {
598 let query_details: Vec<_> = vector
599 .iter()
600 .take(30)
601 .map(|(id, w)| format!("{}:{:.3}", id, w))
602 .collect();
603 log::debug!("Query tokens (id:weight): [{}]", query_details.join(", "));
604 }
605
606 if !matched_tokens.is_empty() {
607 log::debug!(
608 "Matched token IDs: {:?}",
609 matched_tokens.iter().take(20).collect::<Vec<_>>()
610 );
611 }
612
613 if !missing_tokens.is_empty() {
614 log::debug!(
615 "Missing token IDs (not in index): {:?}",
616 missing_tokens.iter().take(20).collect::<Vec<_>>()
617 );
618 }
619
620 if scorers.is_empty() {
621 log::debug!("Sparse vector search: no matching tokens, returning empty");
622 return Ok(Vec::new());
623 }
624
625 let num_terms = scorers.len();
629 let over_fetch = limit * 2; let raw_results = if num_terms > 12 {
631 let pl_refs: Vec<_> = posting_lists
633 .iter()
634 .map(|(_, _, pl)| Arc::clone(pl))
635 .collect();
636 let weights: Vec<_> = posting_lists.iter().map(|(_, qw, _)| *qw).collect();
637 drop(scorers); BmpExecutor::new(pl_refs, weights, over_fetch, heap_factor).execute()
639 } else {
640 BlockMaxScoreExecutor::with_heap_factor(scorers, over_fetch, heap_factor).execute()
641 };
642
643 log::trace!(
644 "Sparse WAND returned {} raw results for segment (doc_id_offset={})",
645 raw_results.len(),
646 self.doc_id_offset
647 );
648 if log::log_enabled!(log::Level::Trace) && !raw_results.is_empty() {
649 for r in raw_results.iter().take(5) {
650 log::trace!(
651 " Raw result: doc_id={} (global={}), score={:.4}, ordinal={}",
652 r.doc_id,
653 r.doc_id + self.doc_id_offset,
654 r.score,
655 r.ordinal
656 );
657 }
658 }
659
660 let mut doc_ordinals: rustc_hash::FxHashMap<u32, Vec<(u32, f32)>> =
663 rustc_hash::FxHashMap::default();
664 for r in raw_results {
665 let ordinals = doc_ordinals.entry(r.doc_id).or_default();
666 ordinals.push((r.ordinal as u32, r.score));
667 }
668
669 let mut results: Vec<VectorSearchResult> = doc_ordinals
672 .into_iter()
673 .map(|(doc_id, ordinals)| {
674 let combined_score = combiner.combine(&ordinals);
675 VectorSearchResult::new(doc_id, combined_score, ordinals)
676 })
677 .collect();
678
679 results.sort_by(|a, b| {
681 b.score
682 .partial_cmp(&a.score)
683 .unwrap_or(std::cmp::Ordering::Equal)
684 });
685 results.truncate(limit);
686
687 Ok(results)
688 }
689
690 pub async fn get_positions(
695 &self,
696 field: Field,
697 term: &[u8],
698 ) -> Result<Option<crate::structures::PositionPostingList>> {
699 use std::io::Cursor;
700
701 let handle = match &self.positions_handle {
703 Some(h) => h,
704 None => return Ok(None),
705 };
706
707 let mut key = Vec::with_capacity(4 + term.len());
709 key.extend_from_slice(&field.0.to_le_bytes());
710 key.extend_from_slice(term);
711
712 let term_info = match self.term_dict.get(&key).await? {
714 Some(info) => info,
715 None => return Ok(None),
716 };
717
718 let (offset, length) = match term_info.position_info() {
720 Some((o, l)) => (o, l),
721 None => return Ok(None),
722 };
723
724 let slice = handle.slice(offset..offset + length as u64);
726 let data = slice.read_bytes().await?;
727
728 let mut cursor = Cursor::new(data.as_slice());
730 let pos_list = crate::structures::PositionPostingList::deserialize(&mut cursor)?;
731
732 Ok(Some(pos_list))
733 }
734
735 pub fn has_positions(&self, field: Field) -> bool {
737 if let Some(entry) = self.schema.get_field_entry(field) {
739 entry.positions.is_some()
740 } else {
741 false
742 }
743 }
744}
745
746pub type SegmentReader = AsyncSegmentReader;