Skip to main content

hermes_core/segment/reader/
mod.rs

1//! Async segment reader with lazy loading
2
3mod loader;
4mod types;
5
6pub use types::{SparseIndex, VectorIndex, VectorSearchResult};
7
8/// Memory statistics for a single segment
9#[derive(Debug, Clone, Default)]
10pub struct SegmentMemoryStats {
11    /// Segment ID
12    pub segment_id: u128,
13    /// Number of documents in segment
14    pub num_docs: u32,
15    /// Term dictionary block cache bytes
16    pub term_dict_cache_bytes: usize,
17    /// Document store block cache bytes
18    pub store_cache_bytes: usize,
19    /// Sparse vector index bytes (in-memory posting lists)
20    pub sparse_index_bytes: usize,
21    /// Dense vector index bytes (cluster assignments, quantized codes)
22    pub dense_index_bytes: usize,
23    /// Bloom filter bytes
24    pub bloom_filter_bytes: usize,
25}
26
27impl SegmentMemoryStats {
28    /// Total estimated memory for this segment
29    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 super::vector_data::LazyFlatVectorData;
45use crate::directories::{AsyncFileRead, Directory, LazyFileHandle, LazyFileSlice};
46use crate::dsl::{Document, Field, Schema};
47use crate::structures::{
48    AsyncSSTableReader, BlockPostingList, CoarseCentroids, IVFPQIndex, IVFRaBitQIndex, PQCodebook,
49    RaBitQIndex, SSTableStats, TermInfo,
50};
51use crate::{DocId, Error, Result};
52
53use super::store::{AsyncStoreReader, RawStoreBlock};
54use super::types::{SegmentFiles, SegmentId, SegmentMeta};
55
56/// Async segment reader with lazy loading
57///
58/// - Term dictionary: only index loaded, blocks loaded on-demand
59/// - Postings: loaded on-demand per term via HTTP range requests
60/// - Document store: only index loaded, blocks loaded on-demand via HTTP range requests
61pub struct AsyncSegmentReader {
62    meta: SegmentMeta,
63    /// Term dictionary with lazy block loading
64    term_dict: Arc<AsyncSSTableReader<TermInfo>>,
65    /// Postings file handle - fetches ranges on demand
66    postings_handle: LazyFileHandle,
67    /// Document store with lazy block loading
68    store: Arc<AsyncStoreReader>,
69    schema: Arc<Schema>,
70    /// Base doc_id offset for this segment
71    doc_id_offset: DocId,
72    /// Dense vector indexes per field (RaBitQ or IVF-RaBitQ) — for search
73    vector_indexes: FxHashMap<u32, VectorIndex>,
74    /// Lazy flat vectors per field — for reranking and merge (doc_ids in memory, vectors via mmap)
75    flat_vectors: FxHashMap<u32, LazyFlatVectorData>,
76    /// Per-field coarse centroids for IVF/ScaNN search
77    coarse_centroids: FxHashMap<u32, Arc<CoarseCentroids>>,
78    /// Sparse vector indexes per field
79    sparse_indexes: FxHashMap<u32, SparseIndex>,
80    /// Position file handle for phrase queries (lazy loading)
81    positions_handle: Option<LazyFileHandle>,
82}
83
84impl AsyncSegmentReader {
85    /// Open a segment with lazy loading
86    pub async fn open<D: Directory>(
87        dir: &D,
88        segment_id: SegmentId,
89        schema: Arc<Schema>,
90        doc_id_offset: DocId,
91        cache_blocks: usize,
92    ) -> Result<Self> {
93        let files = SegmentFiles::new(segment_id.0);
94
95        // Read metadata (small, always loaded)
96        let meta_slice = dir.open_read(&files.meta).await?;
97        let meta_bytes = meta_slice.read_bytes().await?;
98        let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
99        debug_assert_eq!(meta.id, segment_id.0);
100
101        // Open term dictionary with lazy loading (fetches ranges on demand)
102        let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
103        let term_dict = AsyncSSTableReader::open(term_dict_handle, cache_blocks).await?;
104
105        // Get postings file handle (lazy - fetches ranges on demand)
106        let postings_handle = dir.open_lazy(&files.postings).await?;
107
108        // Open store with lazy loading
109        let store_handle = dir.open_lazy(&files.store).await?;
110        let store = AsyncStoreReader::open(store_handle, cache_blocks).await?;
111
112        // Load dense vector indexes from unified .vectors file
113        let vectors_data = loader::load_vectors_file(dir, &files, &schema).await?;
114        let vector_indexes = vectors_data.indexes;
115        let flat_vectors = vectors_data.flat_vectors;
116
117        // Load sparse vector indexes from .sparse file
118        let sparse_indexes = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
119
120        // Open positions file handle (if exists) - offsets are now in TermInfo
121        let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
122
123        // Log segment loading stats (compact format: ~24 bytes per active dim in hashmap)
124        let sparse_dims: usize = sparse_indexes.values().map(|s| s.num_dimensions()).sum();
125        let sparse_mem = sparse_dims * 24; // HashMap entry overhead
126        log::debug!(
127            "[segment] loaded {:016x}: docs={}, sparse_dims={}, sparse_mem={:.2} KB, dense_flat={}, dense_ann={}",
128            segment_id.0,
129            meta.num_docs,
130            sparse_dims,
131            sparse_mem as f64 / 1024.0,
132            flat_vectors.len(),
133            vector_indexes.len()
134        );
135
136        Ok(Self {
137            meta,
138            term_dict: Arc::new(term_dict),
139            postings_handle,
140            store: Arc::new(store),
141            schema,
142            doc_id_offset,
143            vector_indexes,
144            flat_vectors,
145            coarse_centroids: FxHashMap::default(),
146            sparse_indexes,
147            positions_handle,
148        })
149    }
150
151    pub fn meta(&self) -> &SegmentMeta {
152        &self.meta
153    }
154
155    pub fn num_docs(&self) -> u32 {
156        self.meta.num_docs
157    }
158
159    /// Get average field length for BM25F scoring
160    pub fn avg_field_len(&self, field: Field) -> f32 {
161        self.meta.avg_field_len(field)
162    }
163
164    pub fn doc_id_offset(&self) -> DocId {
165        self.doc_id_offset
166    }
167
168    /// Set the doc_id_offset (used for parallel segment loading)
169    pub fn set_doc_id_offset(&mut self, offset: DocId) {
170        self.doc_id_offset = offset;
171    }
172
173    pub fn schema(&self) -> &Schema {
174        &self.schema
175    }
176
177    /// Get sparse indexes for all fields
178    pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
179        &self.sparse_indexes
180    }
181
182    /// Get vector indexes for all fields
183    pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
184        &self.vector_indexes
185    }
186
187    /// Get lazy flat vectors for all fields (for reranking and merge)
188    pub fn flat_vectors(&self) -> &FxHashMap<u32, LazyFlatVectorData> {
189        &self.flat_vectors
190    }
191
192    /// Get term dictionary stats for debugging
193    pub fn term_dict_stats(&self) -> SSTableStats {
194        self.term_dict.stats()
195    }
196
197    /// Estimate memory usage of this segment reader
198    pub fn memory_stats(&self) -> SegmentMemoryStats {
199        let term_dict_stats = self.term_dict.stats();
200
201        // Term dict cache: num_blocks * avg_block_size (estimate 4KB per cached block)
202        let term_dict_cache_bytes = self.term_dict.cached_blocks() * 4096;
203
204        // Store cache: similar estimate
205        let store_cache_bytes = self.store.cached_blocks() * 4096;
206
207        // Sparse index: each dimension has a posting list in memory
208        // Estimate: ~24 bytes per active dimension (HashMap entry overhead)
209        let sparse_index_bytes: usize = self
210            .sparse_indexes
211            .values()
212            .map(|s| s.num_dimensions() * 24)
213            .sum();
214
215        // Dense index: vectors are memory-mapped, but we track index structures
216        // RaBitQ/IVF indexes have cluster assignments in memory
217        let dense_index_bytes: usize = self
218            .vector_indexes
219            .values()
220            .map(|v| v.estimated_memory_bytes())
221            .sum();
222
223        SegmentMemoryStats {
224            segment_id: self.meta.id,
225            num_docs: self.meta.num_docs,
226            term_dict_cache_bytes,
227            store_cache_bytes,
228            sparse_index_bytes,
229            dense_index_bytes,
230            bloom_filter_bytes: term_dict_stats.bloom_filter_size,
231        }
232    }
233
234    /// Get posting list for a term (async - loads on demand)
235    ///
236    /// For small posting lists (1-3 docs), the data is inlined in the term dictionary
237    /// and no additional I/O is needed. For larger lists, reads from .post file.
238    pub async fn get_postings(
239        &self,
240        field: Field,
241        term: &[u8],
242    ) -> Result<Option<BlockPostingList>> {
243        log::debug!(
244            "SegmentReader::get_postings field={} term_len={}",
245            field.0,
246            term.len()
247        );
248
249        // Build key: field_id + term
250        let mut key = Vec::with_capacity(4 + term.len());
251        key.extend_from_slice(&field.0.to_le_bytes());
252        key.extend_from_slice(term);
253
254        // Look up in term dictionary
255        let term_info = match self.term_dict.get(&key).await? {
256            Some(info) => {
257                log::debug!("SegmentReader::get_postings found term_info");
258                info
259            }
260            None => {
261                log::debug!("SegmentReader::get_postings term not found");
262                return Ok(None);
263            }
264        };
265
266        // Check if posting list is inlined
267        if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
268            // Build BlockPostingList from inline data (no I/O needed!)
269            let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
270            for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs.into_iter()) {
271                posting_list.push(doc_id, tf);
272            }
273            let block_list = BlockPostingList::from_posting_list(&posting_list)?;
274            return Ok(Some(block_list));
275        }
276
277        // External posting list - read from postings file handle (lazy - HTTP range request)
278        let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
279            Error::Corruption("TermInfo has neither inline nor external data".to_string())
280        })?;
281
282        let start = posting_offset;
283        let end = start + posting_len as u64;
284
285        if end > self.postings_handle.len() {
286            return Err(Error::Corruption(
287                "Posting offset out of bounds".to_string(),
288            ));
289        }
290
291        let posting_bytes = self.postings_handle.read_bytes_range(start..end).await?;
292        let block_list = BlockPostingList::deserialize(&mut posting_bytes.as_slice())?;
293
294        Ok(Some(block_list))
295    }
296
297    /// Get document by local doc_id (async - loads on demand).
298    ///
299    /// Dense vector fields are hydrated from LazyFlatVectorData (not stored in .store).
300    /// Uses binary search on sorted doc_ids for O(log N) lookup.
301    pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
302        let mut doc = match self.store.get(local_doc_id, &self.schema).await {
303            Ok(Some(d)) => d,
304            Ok(None) => return Ok(None),
305            Err(e) => return Err(Error::from(e)),
306        };
307
308        // Hydrate dense vector fields from flat vector data
309        for (&field_id, lazy_flat) in &self.flat_vectors {
310            let (start, entries) = lazy_flat.flat_indexes_for_doc(local_doc_id);
311            for (j, &(_doc_id, _ordinal)) in entries.iter().enumerate() {
312                let flat_idx = start + j;
313                match lazy_flat.get_vector(flat_idx).await {
314                    Ok(vec) => {
315                        doc.add_dense_vector(Field(field_id), vec);
316                    }
317                    Err(e) => {
318                        log::warn!("Failed to hydrate vector field {}: {}", field_id, e);
319                    }
320                }
321            }
322        }
323
324        Ok(Some(doc))
325    }
326
327    /// Prefetch term dictionary blocks for a key range
328    pub async fn prefetch_terms(
329        &self,
330        field: Field,
331        start_term: &[u8],
332        end_term: &[u8],
333    ) -> Result<()> {
334        let mut start_key = Vec::with_capacity(4 + start_term.len());
335        start_key.extend_from_slice(&field.0.to_le_bytes());
336        start_key.extend_from_slice(start_term);
337
338        let mut end_key = Vec::with_capacity(4 + end_term.len());
339        end_key.extend_from_slice(&field.0.to_le_bytes());
340        end_key.extend_from_slice(end_term);
341
342        self.term_dict.prefetch_range(&start_key, &end_key).await?;
343        Ok(())
344    }
345
346    /// Check if store uses dictionary compression (incompatible with raw merging)
347    pub fn store_has_dict(&self) -> bool {
348        self.store.has_dict()
349    }
350
351    /// Get store reference for merge operations
352    pub fn store(&self) -> &super::store::AsyncStoreReader {
353        &self.store
354    }
355
356    /// Get raw store blocks for optimized merging
357    pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
358        self.store.raw_blocks()
359    }
360
361    /// Get store data slice for raw block access
362    pub fn store_data_slice(&self) -> &LazyFileSlice {
363        self.store.data_slice()
364    }
365
366    /// Get all terms from this segment (for merge)
367    pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
368        self.term_dict.all_entries().await.map_err(Error::from)
369    }
370
371    /// Get all terms with parsed field and term string (for statistics aggregation)
372    ///
373    /// Returns (field, term_string, doc_freq) for each term in the dictionary.
374    /// Skips terms that aren't valid UTF-8.
375    pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
376        let entries = self.term_dict.all_entries().await?;
377        let mut result = Vec::with_capacity(entries.len());
378
379        for (key, term_info) in entries {
380            // Key format: field_id (4 bytes little-endian) + term bytes
381            if key.len() > 4 {
382                let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
383                let term_bytes = &key[4..];
384                if let Ok(term_str) = std::str::from_utf8(term_bytes) {
385                    result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
386                }
387            }
388        }
389
390        Ok(result)
391    }
392
393    /// Get streaming iterator over term dictionary (for memory-efficient merge)
394    pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
395        self.term_dict.iter()
396    }
397
398    /// Prefetch all term dictionary blocks in a single bulk I/O call.
399    ///
400    /// Call before merge iteration to eliminate per-block cache misses.
401    pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
402        self.term_dict
403            .prefetch_all_data_bulk()
404            .await
405            .map_err(crate::Error::from)
406    }
407
408    /// Read raw posting bytes at offset
409    pub async fn read_postings(&self, offset: u64, len: u32) -> Result<Vec<u8>> {
410        let start = offset;
411        let end = start + len as u64;
412        let bytes = self.postings_handle.read_bytes_range(start..end).await?;
413        Ok(bytes.to_vec())
414    }
415
416    /// Read raw position bytes at offset (for merge)
417    pub async fn read_position_bytes(&self, offset: u64, len: u32) -> Result<Option<Vec<u8>>> {
418        let handle = match &self.positions_handle {
419            Some(h) => h,
420            None => return Ok(None),
421        };
422        let start = offset;
423        let end = start + len as u64;
424        let bytes = handle.read_bytes_range(start..end).await?;
425        Ok(Some(bytes.to_vec()))
426    }
427
428    /// Check if this segment has a positions file
429    pub fn has_positions_file(&self) -> bool {
430        self.positions_handle.is_some()
431    }
432
433    /// Batch cosine scoring on raw quantized bytes.
434    ///
435    /// Dispatches to the appropriate SIMD scorer based on quantization type.
436    /// Vectors file uses data-first layout (offset 0) with 8-byte padding between
437    /// fields, so mmap slices are always properly aligned for f32/f16/u8 access.
438    fn score_quantized_batch(
439        query: &[f32],
440        raw: &[u8],
441        quant: crate::dsl::DenseVectorQuantization,
442        dim: usize,
443        scores: &mut [f32],
444    ) {
445        match quant {
446            crate::dsl::DenseVectorQuantization::F32 => {
447                let num_floats = scores.len() * dim;
448                debug_assert!(
449                    (raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()),
450                    "f32 vector data not 4-byte aligned — vectors file may use legacy format"
451                );
452                let vectors: &[f32] =
453                    unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
454                crate::structures::simd::batch_cosine_scores(query, vectors, dim, scores);
455            }
456            crate::dsl::DenseVectorQuantization::F16 => {
457                crate::structures::simd::batch_cosine_scores_f16(query, raw, dim, scores);
458            }
459            crate::dsl::DenseVectorQuantization::UInt8 => {
460                crate::structures::simd::batch_cosine_scores_u8(query, raw, dim, scores);
461            }
462        }
463    }
464
465    /// Search dense vectors using RaBitQ
466    ///
467    /// Returns VectorSearchResult with ordinal tracking for multi-value fields.
468    /// The doc_ids are adjusted by doc_id_offset for this segment.
469    /// For multi-valued documents, scores are combined using the specified combiner.
470    pub async fn search_dense_vector(
471        &self,
472        field: Field,
473        query: &[f32],
474        k: usize,
475        nprobe: usize,
476        rerank_factor: usize,
477        combiner: crate::query::MultiValueCombiner,
478    ) -> Result<Vec<VectorSearchResult>> {
479        let ann_index = self.vector_indexes.get(&field.0);
480        let lazy_flat = self.flat_vectors.get(&field.0);
481
482        // No vectors at all for this field
483        if ann_index.is_none() && lazy_flat.is_none() {
484            return Ok(Vec::new());
485        }
486
487        /// Batch size for brute-force scoring (4096 vectors × 768 dims × 4 bytes ≈ 12MB)
488        const BRUTE_FORCE_BATCH: usize = 4096;
489
490        // Results are (doc_id, ordinal, score) where score = similarity (higher = better)
491        let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
492            // ANN search (RaBitQ, IVF, ScaNN)
493            match index {
494                VectorIndex::RaBitQ(rabitq) => {
495                    let fetch_k = k * rerank_factor.max(1);
496                    rabitq
497                        .search(query, fetch_k, rerank_factor)
498                        .into_iter()
499                        .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
500                        .collect()
501                }
502                VectorIndex::IVF(lazy) => {
503                    let (index, codebook) = lazy.get().ok_or_else(|| {
504                        Error::Schema("IVF index deserialization failed".to_string())
505                    })?;
506                    let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
507                        Error::Schema(format!(
508                            "IVF index requires coarse centroids for field {}",
509                            field.0
510                        ))
511                    })?;
512                    let effective_nprobe = if nprobe > 0 { nprobe } else { 32 };
513                    let fetch_k = k * rerank_factor.max(1);
514                    index
515                        .search(centroids, codebook, query, fetch_k, Some(effective_nprobe))
516                        .into_iter()
517                        .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
518                        .collect()
519                }
520                VectorIndex::ScaNN(lazy) => {
521                    let (index, codebook) = lazy.get().ok_or_else(|| {
522                        Error::Schema("ScaNN index deserialization failed".to_string())
523                    })?;
524                    let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
525                        Error::Schema(format!(
526                            "ScaNN index requires coarse centroids for field {}",
527                            field.0
528                        ))
529                    })?;
530                    let effective_nprobe = if nprobe > 0 { nprobe } else { 32 };
531                    let fetch_k = k * rerank_factor.max(1);
532                    index
533                        .search(centroids, codebook, query, fetch_k, Some(effective_nprobe))
534                        .into_iter()
535                        .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
536                        .collect()
537                }
538            }
539        } else if let Some(lazy_flat) = lazy_flat {
540            // Batched brute-force from lazy flat vectors (native-precision SIMD scoring)
541            // Uses a top-k heap to avoid collecting and sorting all N candidates.
542            let dim = lazy_flat.dim;
543            let n = lazy_flat.num_vectors;
544            let quant = lazy_flat.quantization;
545            let fetch_k = k * rerank_factor.max(1);
546            let mut collector = crate::query::ScoreCollector::new(fetch_k);
547
548            for batch_start in (0..n).step_by(BRUTE_FORCE_BATCH) {
549                let batch_count = BRUTE_FORCE_BATCH.min(n - batch_start);
550                let batch_bytes = lazy_flat
551                    .read_vectors_batch(batch_start, batch_count)
552                    .await
553                    .map_err(crate::Error::Io)?;
554                let raw = batch_bytes.as_slice();
555
556                let mut scores = vec![0f32; batch_count];
557                Self::score_quantized_batch(query, raw, quant, dim, &mut scores);
558
559                for (i, &score) in scores.iter().enumerate().take(batch_count) {
560                    let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
561                    collector.insert_with_ordinal(doc_id, score, ordinal);
562                }
563            }
564
565            collector
566                .into_sorted_results()
567                .into_iter()
568                .map(|(doc_id, score, ordinal)| (doc_id, ordinal, score))
569                .collect()
570        } else {
571            return Ok(Vec::new());
572        };
573
574        // Rerank ANN candidates using raw vectors from lazy flat (binary search lookup)
575        // Uses native-precision SIMD scoring on quantized bytes — no dequantization overhead.
576        if ann_index.is_some()
577            && !results.is_empty()
578            && let Some(lazy_flat) = lazy_flat
579        {
580            let dim = lazy_flat.dim;
581            let quant = lazy_flat.quantization;
582            let vbs = lazy_flat.vector_byte_size();
583
584            // Resolve flat indexes for each candidate via binary search
585            let mut resolved: Vec<(usize, usize)> = Vec::new(); // (result_idx, flat_idx)
586            for (ri, c) in results.iter().enumerate() {
587                let (start, entries) = lazy_flat.flat_indexes_for_doc(c.0);
588                for (j, &(_, ord)) in entries.iter().enumerate() {
589                    if ord == c.1 {
590                        resolved.push((ri, start + j));
591                        break;
592                    }
593                }
594            }
595
596            if !resolved.is_empty() {
597                // Sort by flat_idx for sequential mmap access (better page locality)
598                resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
599
600                // Batch-read raw quantized bytes into contiguous buffer
601                let mut raw_buf = vec![0u8; resolved.len() * vbs];
602                for (buf_idx, &(_, flat_idx)) in resolved.iter().enumerate() {
603                    let _ = lazy_flat
604                        .read_vector_raw_into(
605                            flat_idx,
606                            &mut raw_buf[buf_idx * vbs..(buf_idx + 1) * vbs],
607                        )
608                        .await;
609                }
610
611                // Native-precision batch SIMD cosine scoring
612                let mut scores = vec![0f32; resolved.len()];
613                Self::score_quantized_batch(query, &raw_buf, quant, dim, &mut scores);
614
615                // Write scores back to results
616                for (buf_idx, &(ri, _)) in resolved.iter().enumerate() {
617                    results[ri].2 = scores[buf_idx];
618                }
619            }
620
621            results.sort_by(|a, b| b.2.partial_cmp(&a.2).unwrap_or(std::cmp::Ordering::Equal));
622            results.truncate(k * rerank_factor.max(1));
623        }
624
625        // Track ordinals with individual scores for each doc_id
626        // Note: doc_id_offset is NOT applied here - the collector applies it uniformly
627        let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
628            rustc_hash::FxHashMap::default();
629        for (doc_id, ordinal, score) in results {
630            let ordinals = doc_ordinals.entry(doc_id as DocId).or_default();
631            ordinals.push((ordinal as u32, score));
632        }
633
634        // Combine scores and build results with ordinal tracking
635        let mut final_results: Vec<VectorSearchResult> = doc_ordinals
636            .into_iter()
637            .map(|(doc_id, ordinals)| {
638                let combined_score = combiner.combine(&ordinals);
639                VectorSearchResult::new(doc_id, combined_score, ordinals)
640            })
641            .collect();
642
643        // Sort by score descending and take top k
644        final_results.sort_by(|a, b| {
645            b.score
646                .partial_cmp(&a.score)
647                .unwrap_or(std::cmp::Ordering::Equal)
648        });
649        final_results.truncate(k);
650
651        Ok(final_results)
652    }
653
654    /// Check if this segment has dense vectors for the given field
655    pub fn has_dense_vector_index(&self, field: Field) -> bool {
656        self.vector_indexes.contains_key(&field.0) || self.flat_vectors.contains_key(&field.0)
657    }
658
659    /// Get the dense vector index for a field (if available)
660    pub fn get_dense_vector_index(&self, field: Field) -> Option<Arc<RaBitQIndex>> {
661        match self.vector_indexes.get(&field.0) {
662            Some(VectorIndex::RaBitQ(idx)) => Some(idx.clone()),
663            _ => None,
664        }
665    }
666
667    /// Get the IVF vector index for a field (if available)
668    pub fn get_ivf_vector_index(
669        &self,
670        field: Field,
671    ) -> Option<(Arc<IVFRaBitQIndex>, Arc<crate::structures::RaBitQCodebook>)> {
672        match self.vector_indexes.get(&field.0) {
673            Some(VectorIndex::IVF(lazy)) => lazy.get().map(|(i, c)| (i.clone(), c.clone())),
674            _ => None,
675        }
676    }
677
678    /// Get coarse centroids for a field
679    pub fn coarse_centroids(&self, field_id: u32) -> Option<&Arc<CoarseCentroids>> {
680        self.coarse_centroids.get(&field_id)
681    }
682
683    /// Set per-field coarse centroids from index-level trained structures
684    pub fn set_coarse_centroids(&mut self, centroids: FxHashMap<u32, Arc<CoarseCentroids>>) {
685        self.coarse_centroids = centroids;
686    }
687
688    /// Get the ScaNN vector index for a field (if available)
689    pub fn get_scann_vector_index(
690        &self,
691        field: Field,
692    ) -> Option<(Arc<IVFPQIndex>, Arc<PQCodebook>)> {
693        match self.vector_indexes.get(&field.0) {
694            Some(VectorIndex::ScaNN(lazy)) => lazy.get().map(|(i, c)| (i.clone(), c.clone())),
695            _ => None,
696        }
697    }
698
699    /// Get the vector index type for a field
700    pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
701        self.vector_indexes.get(&field.0)
702    }
703
704    /// Search for similar sparse vectors using dedicated sparse posting lists
705    ///
706    /// Uses shared `WandExecutor` with `SparseTermScorer` for efficient top-k retrieval.
707    /// Optimizations (via WandExecutor):
708    /// 1. **MaxScore pruning**: Dimensions sorted by max contribution
709    /// 2. **Block-Max WAND**: Skips blocks where max contribution < threshold
710    /// 3. **Top-K heap**: Efficient score collection
711    ///
712    /// Returns VectorSearchResult with ordinal tracking for multi-value fields.
713    pub async fn search_sparse_vector(
714        &self,
715        field: Field,
716        vector: &[(u32, f32)],
717        limit: usize,
718        combiner: crate::query::MultiValueCombiner,
719        heap_factor: f32,
720    ) -> Result<Vec<VectorSearchResult>> {
721        use crate::query::{BlockMaxScoreExecutor, BmpExecutor, SparseTermScorer};
722
723        let query_tokens = vector.len();
724
725        // Get sparse index for this field
726        let sparse_index = match self.sparse_indexes.get(&field.0) {
727            Some(idx) => idx,
728            None => {
729                log::debug!(
730                    "Sparse vector search: no index for field {}, returning empty",
731                    field.0
732                );
733                return Ok(Vec::new());
734            }
735        };
736
737        let index_dimensions = sparse_index.num_dimensions();
738
739        // Filter query terms to only those present in the index
740        let mut matched_terms: Vec<(u32, f32)> = Vec::with_capacity(vector.len());
741        let mut missing_tokens = Vec::new();
742
743        for &(dim_id, query_weight) in vector {
744            if sparse_index.has_dimension(dim_id) {
745                matched_terms.push((dim_id, query_weight));
746            } else {
747                missing_tokens.push(dim_id);
748            }
749        }
750
751        log::debug!(
752            "Sparse vector search: query_tokens={}, matched={}, missing={}, index_dimensions={}",
753            query_tokens,
754            matched_terms.len(),
755            missing_tokens.len(),
756            index_dimensions
757        );
758
759        if log::log_enabled!(log::Level::Debug) {
760            let query_details: Vec<_> = vector
761                .iter()
762                .take(30)
763                .map(|(id, w)| format!("{}:{:.3}", id, w))
764                .collect();
765            log::debug!("Query tokens (id:weight): [{}]", query_details.join(", "));
766        }
767
768        if !missing_tokens.is_empty() {
769            log::debug!(
770                "Missing token IDs (not in index): {:?}",
771                missing_tokens.iter().take(20).collect::<Vec<_>>()
772            );
773        }
774
775        if matched_terms.is_empty() {
776            log::debug!("Sparse vector search: no matching tokens, returning empty");
777            return Ok(Vec::new());
778        }
779
780        // Select executor based on number of query terms:
781        // - 12+ terms: BMP (block-at-a-time, lazy block loading, best for SPLADE)
782        // - 1-11 terms: BlockMaxScoreExecutor (unified MaxScore + block-max + conjunction)
783        let num_terms = matched_terms.len();
784        let over_fetch = limit * 2; // Over-fetch for multi-value combining
785        let raw_results = if num_terms > 12 {
786            // BMP: lazy block loading — only skip entries in memory, blocks loaded on-demand
787            BmpExecutor::new(sparse_index, matched_terms, over_fetch, heap_factor)
788                .execute()
789                .await?
790        } else {
791            // Load posting lists only for the few terms (1-11) used by BlockMaxScore
792            let mut posting_lists: Vec<(u32, f32, Arc<BlockSparsePostingList>)> =
793                Vec::with_capacity(num_terms);
794            for &(dim_id, query_weight) in &matched_terms {
795                if let Some(pl) = sparse_index.get_posting(dim_id).await? {
796                    posting_lists.push((dim_id, query_weight, pl));
797                }
798            }
799            let scorers: Vec<SparseTermScorer> = posting_lists
800                .iter()
801                .map(|(_, query_weight, pl)| SparseTermScorer::from_arc(pl, *query_weight))
802                .collect();
803            if scorers.is_empty() {
804                return Ok(Vec::new());
805            }
806            BlockMaxScoreExecutor::with_heap_factor(scorers, over_fetch, heap_factor).execute()
807        };
808
809        log::trace!(
810            "Sparse WAND returned {} raw results for segment (doc_id_offset={})",
811            raw_results.len(),
812            self.doc_id_offset
813        );
814        if log::log_enabled!(log::Level::Trace) && !raw_results.is_empty() {
815            for r in raw_results.iter().take(5) {
816                log::trace!(
817                    "  Raw result: doc_id={} (global={}), score={:.4}, ordinal={}",
818                    r.doc_id,
819                    r.doc_id + self.doc_id_offset,
820                    r.score,
821                    r.ordinal
822                );
823            }
824        }
825
826        // Track ordinals with individual scores for each doc_id
827        // Now using real ordinals from the posting lists
828        let mut doc_ordinals: rustc_hash::FxHashMap<u32, Vec<(u32, f32)>> =
829            rustc_hash::FxHashMap::default();
830        for r in raw_results {
831            let ordinals = doc_ordinals.entry(r.doc_id).or_default();
832            ordinals.push((r.ordinal as u32, r.score));
833        }
834
835        // Combine scores and build results with ordinal tracking
836        // Note: doc_id_offset is NOT applied here - the collector applies it uniformly
837        let mut results: Vec<VectorSearchResult> = doc_ordinals
838            .into_iter()
839            .map(|(doc_id, ordinals)| {
840                let combined_score = combiner.combine(&ordinals);
841                VectorSearchResult::new(doc_id, combined_score, ordinals)
842            })
843            .collect();
844
845        // Sort by score descending and take top limit
846        results.sort_by(|a, b| {
847            b.score
848                .partial_cmp(&a.score)
849                .unwrap_or(std::cmp::Ordering::Equal)
850        });
851        results.truncate(limit);
852
853        Ok(results)
854    }
855
856    /// Get positions for a term (for phrase queries)
857    ///
858    /// Position offsets are now embedded in TermInfo, so we first look up
859    /// the term to get its TermInfo, then use position_info() to get the offset.
860    pub async fn get_positions(
861        &self,
862        field: Field,
863        term: &[u8],
864    ) -> Result<Option<crate::structures::PositionPostingList>> {
865        use std::io::Cursor;
866
867        // Get positions handle
868        let handle = match &self.positions_handle {
869            Some(h) => h,
870            None => return Ok(None),
871        };
872
873        // Build key: field_id + term
874        let mut key = Vec::with_capacity(4 + term.len());
875        key.extend_from_slice(&field.0.to_le_bytes());
876        key.extend_from_slice(term);
877
878        // Look up term in dictionary to get TermInfo with position offset
879        let term_info = match self.term_dict.get(&key).await? {
880            Some(info) => info,
881            None => return Ok(None),
882        };
883
884        // Get position offset from TermInfo
885        let (offset, length) = match term_info.position_info() {
886            Some((o, l)) => (o, l),
887            None => return Ok(None),
888        };
889
890        // Read the position data
891        let slice = handle.slice(offset..offset + length as u64);
892        let data = slice.read_bytes().await?;
893
894        // Deserialize
895        let mut cursor = Cursor::new(data.as_slice());
896        let pos_list = crate::structures::PositionPostingList::deserialize(&mut cursor)?;
897
898        Ok(Some(pos_list))
899    }
900
901    /// Check if positions are available for a field
902    pub fn has_positions(&self, field: Field) -> bool {
903        // Check schema for position mode on this field
904        if let Some(entry) = self.schema.get_field_entry(field) {
905            entry.positions.is_some()
906        } else {
907            false
908        }
909    }
910}
911
912/// Alias for AsyncSegmentReader
913pub type SegmentReader = AsyncSegmentReader;