Skip to main content

hermes_core/segment/reader/
mod.rs

1//! Async segment reader with lazy loading
2
3pub(crate) mod bmp;
4pub(crate) mod loader;
5mod types;
6
7pub use bmp::BmpIndex;
8#[cfg(feature = "native")]
9pub(crate) use types::DimRawData;
10pub use types::{SparseIndex, VectorIndex, VectorSearchResult};
11
12/// Bound vocabulary and posting expansion before a prefix query starts loading
13/// posting payloads. These are per-segment limits; callers should use exact-term
14/// or a more selective prefix when they are exceeded.
15const MAX_PREFIX_TERMS: usize = 1_024;
16const MAX_PREFIX_POSTINGS: u64 = 5_000_000;
17/// Hard guard for explicitly requested dense rerank documents and for the
18/// exact ordinal expansion of those documents. ANN heaps no longer deepen
19/// toward this value automatically for multi-valued fields.
20const MAX_DENSE_CANDIDATES_PER_SEGMENT: usize = 20_000;
21/// Preferred vector count; wide vectors reduce it to stay under the byte cap.
22const DENSE_SCORE_BATCH: usize = 4_096;
23const BINARY_SCORE_BATCH: usize = 8_192;
24const MAX_VECTOR_SCORE_BATCH_BYTES: usize = 8 * 1024 * 1024;
25
26/// Memory statistics for a single segment
27#[derive(Debug, Clone, Default)]
28pub struct SegmentMemoryStats {
29    /// Segment ID
30    pub segment_id: u128,
31    /// Number of documents in segment
32    pub num_docs: u32,
33    /// Term dictionary block cache bytes
34    pub term_dict_cache_bytes: usize,
35    /// Document store block cache bytes
36    pub store_cache_bytes: usize,
37    /// Sparse vector index bytes (in-memory posting lists)
38    pub sparse_index_bytes: usize,
39    /// Dense vector index bytes (cluster assignments, quantized codes)
40    pub dense_index_bytes: usize,
41    /// Bloom filter bytes
42    pub bloom_filter_bytes: usize,
43    /// Hot metadata bytes actually pinned (mlock/heap-copy) at open
44    pub pinned_metadata_bytes: u64,
45    /// Hot metadata bytes eligible for pinning (gap vs pinned = budget
46    /// exhausted or mlock failures — operator-visible)
47    pub pin_intended_bytes: u64,
48}
49
50impl SegmentMemoryStats {
51    /// Total estimated memory for this segment
52    pub fn total_bytes(&self) -> usize {
53        self.term_dict_cache_bytes
54            + self.store_cache_bytes
55            + self.sparse_index_bytes
56            + self.dense_index_bytes
57            + self.bloom_filter_bytes
58    }
59}
60
61use std::cmp::Ordering;
62use std::collections::BinaryHeap;
63use std::sync::Arc;
64
65use rustc_hash::FxHashMap;
66
67use super::vector_data::LazyFlatVectorData;
68use crate::directories::{Directory, FileHandle};
69use crate::dsl::{DenseVectorQuantization, Document, Field, Schema};
70use crate::query::{MAX_DENSE_NPROBE, MAX_DENSE_RERANK_FACTOR};
71use crate::structures::{
72    AsyncSSTableReader, BlockPostingList, CoarseCentroids, IVFPQIndex, SSTableStats, TermInfo,
73};
74use crate::{DocId, Error, Result};
75
76use super::store::{AsyncStoreReader, RawStoreBlock};
77use super::types::{SegmentFiles, SegmentId, SegmentMeta};
78
79/// Combine per-ordinal (doc_id, ordinal, score) triples into VectorSearchResults,
80/// applying the multi-value combiner, sorting by score desc, and truncating to `limit`.
81///
82/// Fast path: when all ordinals are 0 (single-valued field), skips the HashMap
83/// grouping entirely and just sorts + truncates the raw results.
84pub(crate) fn combine_ordinal_results(
85    raw: impl IntoIterator<Item = (u32, u16, f32)>,
86    combiner: crate::query::MultiValueCombiner,
87    limit: usize,
88) -> Vec<VectorSearchResult> {
89    let collected: Vec<(u32, u16, f32)> = raw.into_iter().collect();
90
91    let num_raw = collected.len();
92    if log::log_enabled!(log::Level::Debug) {
93        let mut ids: Vec<u32> = collected.iter().map(|(d, _, _)| *d).collect();
94        ids.sort_unstable();
95        ids.dedup();
96        log::debug!(
97            "combine_ordinal_results: {} raw entries, {} unique docs, combiner={:?}, limit={}",
98            num_raw,
99            ids.len(),
100            combiner,
101            limit
102        );
103    }
104
105    // Fast path: all ordinals are 0 → no grouping needed, skip HashMap
106    let all_single = collected.iter().all(|&(_, ord, _)| ord == 0);
107    if all_single {
108        let mut results: Vec<VectorSearchResult> = collected
109            .into_iter()
110            .map(|(doc_id, _, score)| VectorSearchResult::new(doc_id, score, vec![(0, score)]))
111            .collect();
112        results.sort_unstable_by(|a, b| {
113            b.score
114                .total_cmp(&a.score)
115                .then_with(|| a.doc_id.cmp(&b.doc_id))
116        });
117        results.truncate(limit);
118        return results;
119    }
120
121    // Slow path: multi-valued field — group by doc_id, apply combiner
122    let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
123        rustc_hash::FxHashMap::default();
124    for (doc_id, ordinal, score) in collected {
125        doc_ordinals
126            .entry(doc_id as DocId)
127            .or_default()
128            .push((ordinal as u32, score));
129    }
130    let mut results: Vec<VectorSearchResult> = doc_ordinals
131        .into_iter()
132        .map(|(doc_id, ordinals)| {
133            let combined_score = combiner.combine(&ordinals);
134            VectorSearchResult::new(doc_id, combined_score, ordinals)
135        })
136        .collect();
137    results.sort_unstable_by(|a, b| {
138        b.score
139            .total_cmp(&a.score)
140            .then_with(|| a.doc_id.cmp(&b.doc_id))
141    });
142    results.truncate(limit);
143    results
144}
145
146/// Heap entry used by exact flat-vector search after all values belonging to
147/// one document have been combined. Keeping the heap at document granularity
148/// prevents several strong values from one document from crowding other
149/// documents out of the raw vector top-k.
150struct HeapVectorResult(VectorSearchResult);
151
152impl PartialEq for HeapVectorResult {
153    fn eq(&self, other: &Self) -> bool {
154        self.0.score.to_bits() == other.0.score.to_bits() && self.0.doc_id == other.0.doc_id
155    }
156}
157
158impl Eq for HeapVectorResult {}
159
160impl Ord for HeapVectorResult {
161    fn cmp(&self, other: &Self) -> Ordering {
162        // BinaryHeap top is the worst retained document: lower score, then
163        // larger doc ID for deterministic equal-score eviction.
164        other
165            .0
166            .score
167            .total_cmp(&self.0.score)
168            .then_with(|| self.0.doc_id.cmp(&other.0.doc_id))
169    }
170}
171
172impl PartialOrd for HeapVectorResult {
173    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
174        Some(self.cmp(other))
175    }
176}
177
178/// Incrementally combine a flat vector stream sorted by `(doc_id, ordinal)`
179/// and retain only the best `limit` documents. Scratch is O(values in the
180/// current document + retained output), independent of the segment size.
181struct FlatDocumentCollector {
182    heap: BinaryHeap<HeapVectorResult>,
183    limit: usize,
184    combiner: crate::query::MultiValueCombiner,
185    current_doc: Option<DocId>,
186    current_ordinals: Vec<(u32, f32)>,
187}
188
189impl FlatDocumentCollector {
190    fn new(limit: usize, combiner: crate::query::MultiValueCombiner) -> Self {
191        Self {
192            heap: BinaryHeap::with_capacity(limit.min(8 * 1024)),
193            limit,
194            combiner,
195            current_doc: None,
196            current_ordinals: Vec::new(),
197        }
198    }
199
200    fn push(&mut self, doc_id: DocId, ordinal: u16, score: f32) {
201        if self.current_doc.is_some_and(|current| current != doc_id) {
202            self.finish_current();
203        }
204        self.current_doc = Some(doc_id);
205        self.current_ordinals.push((ordinal as u32, score));
206    }
207
208    fn finish_current(&mut self) {
209        let Some(doc_id) = self.current_doc.take() else {
210            return;
211        };
212        let score = self.combiner.combine(&self.current_ordinals);
213        let should_retain = self.heap.len() < self.limit
214            || self.heap.peek().is_some_and(|worst| {
215                HeapVectorResult(VectorSearchResult::new(doc_id, score, Vec::new()))
216                    .cmp(worst)
217                    .is_lt()
218            });
219
220        if !should_retain {
221            // The overwhelmingly common path once the heap is full. Reuse
222            // the ordinal scratch instead of allocating a fresh Vec for
223            // every rejected document in a flat scan.
224            self.current_ordinals.clear();
225            return;
226        }
227
228        let ordinals = std::mem::take(&mut self.current_ordinals);
229        let entry = HeapVectorResult(VectorSearchResult::new(doc_id, score, ordinals));
230        if self.heap.len() < self.limit {
231            self.heap.push(entry);
232        } else if let Some(mut worst) = self.heap.peek_mut() {
233            // Recycle the evicted result's allocation as the next document's
234            // scratch. PeekMut restores heap order when it is dropped.
235            let mut evicted = std::mem::replace(&mut worst.0, entry.0);
236            evicted.ordinals.clear();
237            self.current_ordinals = evicted.ordinals;
238        }
239    }
240
241    fn into_results(mut self) -> Vec<VectorSearchResult> {
242        self.finish_current();
243        let mut results: Vec<_> = self.heap.into_iter().map(|entry| entry.0).collect();
244        results.sort_unstable_by(|a, b| {
245            b.score
246                .total_cmp(&a.score)
247                .then_with(|| a.doc_id.cmp(&b.doc_id))
248        });
249        results
250    }
251}
252
253/// Collect a stream already grouped by document (the layout produced by flat
254/// storage expansion) without rebuilding a hash table for every candidate.
255fn combine_grouped_ordinal_results(
256    raw: impl IntoIterator<Item = RawVectorCandidate>,
257    combiner: crate::query::MultiValueCombiner,
258    limit: usize,
259) -> Vec<VectorSearchResult> {
260    let mut collector = FlatDocumentCollector::new(limit, combiner);
261    for (doc_id, ordinal, score) in raw {
262        collector.push(doc_id, ordinal, score);
263    }
264    collector.into_results()
265}
266
267#[derive(Clone, Copy)]
268struct DenseSearchParams {
269    dim: usize,
270    nprobe: usize,
271    unit_norm: bool,
272}
273
274/// Compute the ANN candidate count without relying on saturating float casts.
275fn checked_dense_fetch_k(k: usize, rerank_factor: f32) -> Result<usize> {
276    if !rerank_factor.is_finite() || !(1.0..=MAX_DENSE_RERANK_FACTOR).contains(&rerank_factor) {
277        return Err(Error::Query(format!(
278            "dense rerank_factor must be finite and in [1, {MAX_DENSE_RERANK_FACTOR}], got {rerank_factor}"
279        )));
280    }
281
282    let fetch = (k as f64) * (rerank_factor as f64);
283    if !fetch.is_finite()
284        || fetch > usize::MAX as f64
285        || fetch > MAX_DENSE_CANDIDATES_PER_SEGMENT as f64
286    {
287        return Err(Error::Query(format!(
288            "dense candidate count exceeds the per-segment maximum of \
289             {MAX_DENSE_CANDIDATES_PER_SEGMENT}: k={k}, rerank_factor={rerank_factor}"
290        )));
291    }
292    Ok(fetch.ceil() as usize)
293}
294
295#[inline]
296fn bounded_vector_score_batch(vector_byte_size: usize, preferred: usize) -> usize {
297    preferred.min((MAX_VECTOR_SCORE_BATCH_BYTES / vector_byte_size.max(1)).max(1))
298}
299
300fn checked_file_range(
301    offset: u64,
302    length: u64,
303    file_length: u64,
304    description: &str,
305) -> Result<std::ops::Range<u64>> {
306    let end = offset
307        .checked_add(length)
308        .ok_or_else(|| Error::Corruption(format!("{description} byte range overflows u64")))?;
309    if end > file_length {
310        return Err(Error::Corruption(format!(
311            "{description} byte range {offset}..{end} exceeds file length {file_length}"
312        )));
313    }
314    Ok(offset..end)
315}
316
317type RawVectorCandidate = (u32, u16, f32);
318type ResolvedVectorCandidate = (usize, usize); // (result index, flat-vector index)
319
320/// ANN operates on individual vectors, but document combiners require every
321/// stored value for a candidate document. Expand the deduplicated ANN document
322/// union before exact reranking, with a hard per-segment vector budget.
323fn expand_ann_candidate_documents(
324    ann_results: &[RawVectorCandidate],
325    flat: &LazyFlatVectorData,
326) -> Result<(Vec<RawVectorCandidate>, Vec<ResolvedVectorCandidate>)> {
327    let mut candidate_docs: Vec<DocId> = ann_results.iter().map(|candidate| candidate.0).collect();
328    candidate_docs.sort_unstable();
329    candidate_docs.dedup();
330
331    let mut expanded = Vec::new();
332    let mut resolved = Vec::new();
333    for doc_id in candidate_docs {
334        let (start, count) = flat.flat_indexes_for_doc_range(doc_id);
335        if count == 0 {
336            return Err(Error::Corruption(format!(
337                "ANN candidate document {doc_id} is missing from flat vector storage"
338            )));
339        }
340        let next_len = expanded
341            .len()
342            .checked_add(count)
343            .ok_or_else(|| Error::Query("ANN candidate vector expansion overflow".to_string()))?;
344        if next_len > MAX_DENSE_CANDIDATES_PER_SEGMENT {
345            return Err(Error::Query(format!(
346                "ANN candidate documents expand to more than \
347                 {MAX_DENSE_CANDIDATES_PER_SEGMENT} vectors in one segment"
348            )));
349        }
350        expanded.reserve(count);
351        resolved.reserve(count);
352        let end = start
353            .checked_add(count)
354            .ok_or_else(|| Error::Corruption("flat vector range overflow".to_string()))?;
355        for flat_index in start..end {
356            let (stored_doc_id, ordinal) = flat.get_doc_id(flat_index);
357            if stored_doc_id != doc_id {
358                return Err(Error::Corruption(format!(
359                    "flat vector doc map is not contiguous for document {doc_id}"
360                )));
361            }
362            let result_index = expanded.len();
363            expanded.push((doc_id, ordinal, 0.0));
364            resolved.push((result_index, flat_index));
365        }
366    }
367    Ok((expanded, resolved))
368}
369
370/// Reuse the probe's scores for (doc, ordinal) pairs it already returned.
371///
372/// Binary IVF clusters store the same packed codes as flat storage and score
373/// them with the same exact SIMD Hamming kernel, so re-reading those vectors
374/// from flat storage recomputes an identical score at the cost of one random
375/// read (and potential page fault on cold mmap) per candidate. Only the
376/// candidate documents' remaining ordinals still need flat reads. The
377/// per-query map is justified by removing one flat read per reused entry.
378fn fill_probe_scores_and_prune_resolved(
379    ann_results: &[RawVectorCandidate],
380    expanded: &mut [RawVectorCandidate],
381    resolved: Vec<ResolvedVectorCandidate>,
382) -> Vec<ResolvedVectorCandidate> {
383    let mut probe_scores: FxHashMap<(DocId, u16), f32> =
384        FxHashMap::with_capacity_and_hasher(ann_results.len(), Default::default());
385    for &(doc_id, ordinal, score) in ann_results {
386        probe_scores.insert((doc_id, ordinal), score);
387    }
388    resolved
389        .into_iter()
390        .filter(|&(result_index, _)| {
391            let (doc_id, ordinal, _) = expanded[result_index];
392            match probe_scores.get(&(doc_id, ordinal)) {
393                Some(&score) => {
394                    expanded[result_index].2 = score;
395                    false
396                }
397                None => true,
398            }
399        })
400        .collect()
401}
402
403async fn exact_score_binary_candidate_documents(
404    ann_results: &[RawVectorCandidate],
405    flat: &LazyFlatVectorData,
406    query: &[u8],
407    dim_bits: usize,
408) -> Result<Vec<RawVectorCandidate>> {
409    let (mut expanded, resolved) = expand_ann_candidate_documents(ann_results, flat)?;
410    let resolved = fill_probe_scores_and_prune_resolved(ann_results, &mut expanded, resolved);
411    let vector_byte_size = flat.vector_byte_size();
412    let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
413    let raw_capacity = batch_len
414        .checked_mul(vector_byte_size)
415        .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
416    let mut raw = vec![0u8; raw_capacity];
417    let mut scores = vec![0.0f32; batch_len];
418
419    for chunk in resolved.chunks(batch_len) {
420        #[cfg(feature = "native")]
421        flat.prefetch_vectors(chunk.iter().map(|&(_, flat_index)| flat_index));
422        let raw_len = chunk
423            .len()
424            .checked_mul(vector_byte_size)
425            .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
426        let raw = &mut raw[..raw_len];
427        for (buffer_index, &(_, flat_index)) in chunk.iter().enumerate() {
428            flat.read_vector_raw_into(
429                flat_index,
430                &mut raw[buffer_index * vector_byte_size..(buffer_index + 1) * vector_byte_size],
431            )
432            .await
433            .map_err(Error::Io)?;
434        }
435        crate::structures::simd::batch_hamming_scores(
436            query,
437            raw,
438            vector_byte_size,
439            dim_bits,
440            &mut scores[..chunk.len()],
441        );
442        for (buffer_index, &(result_index, _)) in chunk.iter().enumerate() {
443            expanded[result_index].2 = scores[buffer_index];
444        }
445    }
446    Ok(expanded)
447}
448
449#[cfg(feature = "sync")]
450fn exact_score_binary_candidate_documents_sync(
451    ann_results: &[RawVectorCandidate],
452    flat: &LazyFlatVectorData,
453    query: &[u8],
454    dim_bits: usize,
455) -> Result<Vec<RawVectorCandidate>> {
456    let (mut expanded, resolved) = expand_ann_candidate_documents(ann_results, flat)?;
457    let resolved = fill_probe_scores_and_prune_resolved(ann_results, &mut expanded, resolved);
458    let vector_byte_size = flat.vector_byte_size();
459    let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
460    let raw_capacity = batch_len
461        .checked_mul(vector_byte_size)
462        .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
463    let mut raw = vec![0u8; raw_capacity];
464    let mut scores = vec![0.0f32; batch_len];
465
466    for chunk in resolved.chunks(batch_len) {
467        let raw_len = chunk
468            .len()
469            .checked_mul(vector_byte_size)
470            .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
471        let raw = &mut raw[..raw_len];
472        for (buffer_index, &(_, flat_index)) in chunk.iter().enumerate() {
473            flat.read_vector_raw_into_sync(
474                flat_index,
475                &mut raw[buffer_index * vector_byte_size..(buffer_index + 1) * vector_byte_size],
476            )
477            .map_err(Error::Io)?;
478        }
479        crate::structures::simd::batch_hamming_scores(
480            query,
481            raw,
482            vector_byte_size,
483            dim_bits,
484            &mut scores[..chunk.len()],
485        );
486        for (buffer_index, &(result_index, _)) in chunk.iter().enumerate() {
487            expanded[result_index].2 = scores[buffer_index];
488        }
489    }
490    Ok(expanded)
491}
492
493fn validate_coarse_centroids(centroids: &CoarseCentroids, dim: usize) -> Result<()> {
494    let expected = (centroids.num_clusters as usize)
495        .checked_mul(dim)
496        .ok_or_else(|| Error::Corruption("coarse centroid size overflow".into()))?;
497    if centroids.num_clusters == 0
498        || centroids.dim != dim
499        || centroids.centroids.len() != expected
500        || centroids.centroids.iter().any(|value| !value.is_finite())
501    {
502        return Err(Error::Corruption(format!(
503            "invalid coarse centroids: clusters={}, dim={}, values={} (expected dim={dim}, values={expected})",
504            centroids.num_clusters,
505            centroids.dim,
506            centroids.centroids.len()
507        )));
508    }
509    Ok(())
510}
511
512fn float_query_plan(
513    centroids: &CoarseCentroids,
514    codebook: &crate::structures::PQCodebook,
515    query: &[f32],
516    nprobe: usize,
517    routing: crate::dsl::IvfRoutingMode,
518    cache: Option<&std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>>,
519) -> Result<std::sync::Arc<crate::structures::IvfPqQueryPlan>> {
520    let effective_nprobe = nprobe.clamp(1, centroids.num_clusters as usize);
521    let request_fingerprint = crate::structures::vector::ivf::routing::float_probe_fingerprint(
522        query,
523        effective_nprobe,
524        routing,
525    );
526    if let Some(cache) = cache {
527        let mut cached = cache
528            .lock()
529            .map_err(|_| Error::Internal("dense IVF probe cache is poisoned".into()))?;
530        if let Some(plan) = cached.as_ref()
531            && plan.quantizer_version == centroids.version
532            && plan.codebook_version == codebook.version
533            && plan.request_fingerprint == request_fingerprint
534            && plan.cluster_ids.len() == effective_nprobe
535        {
536            return Ok(std::sync::Arc::clone(plan));
537        }
538        let plan = std::sync::Arc::new(crate::structures::IvfPqQueryPlan::build(
539            centroids,
540            codebook,
541            query,
542            effective_nprobe,
543            routing,
544        ));
545        *cached = Some(std::sync::Arc::clone(&plan));
546        return Ok(plan);
547    }
548    Ok(std::sync::Arc::new(
549        crate::structures::IvfPqQueryPlan::build(
550            centroids,
551            codebook,
552            query,
553            effective_nprobe,
554            routing,
555        ),
556    ))
557}
558
559fn binary_probe_clusters(
560    quantizer: &crate::structures::BinaryCoarseQuantizer,
561    query: &[u8],
562    nprobe: usize,
563    routing: crate::dsl::IvfRoutingMode,
564    cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
565) -> Result<std::sync::Arc<[u32]>> {
566    let effective_nprobe = nprobe.clamp(1, quantizer.num_clusters as usize);
567    let request_fingerprint = crate::structures::vector::ivf::routing::binary_probe_fingerprint(
568        query,
569        effective_nprobe,
570        routing,
571    );
572    if let Some(cache) = cache {
573        let mut cached = cache
574            .lock()
575            .map_err(|_| Error::Internal("binary IVF probe cache is poisoned".into()))?;
576        if let Some(plan) = cached.as_ref()
577            && plan.quantizer_version == quantizer.version
578            && plan.request_fingerprint == request_fingerprint
579            && plan.cluster_ids.len() == effective_nprobe
580        {
581            return Ok(std::sync::Arc::clone(&plan.cluster_ids));
582        }
583        let plan = quantizer.probe(query, effective_nprobe, routing);
584        let clusters = std::sync::Arc::clone(&plan.cluster_ids);
585        *cached = Some(plan);
586        return Ok(clusters);
587    }
588    Ok(quantizer
589        .probe(query, effective_nprobe, routing)
590        .cluster_ids)
591}
592
593/// Async segment reader with lazy loading
594///
595/// - Term dictionary: only index loaded, blocks loaded on-demand
596/// - Postings: loaded on-demand per term via HTTP range requests
597/// - Document store: only index loaded, blocks loaded on-demand via HTTP range requests
598pub struct SegmentReader {
599    meta: SegmentMeta,
600    /// Term dictionary with lazy block loading
601    term_dict: Arc<AsyncSSTableReader<TermInfo>>,
602    /// Postings file handle - fetches ranges on demand
603    postings_handle: FileHandle,
604    /// Document store with lazy block loading
605    store: Arc<AsyncStoreReader>,
606    schema: Arc<Schema>,
607    /// Per-segment ANN payloads.
608    vector_indexes: FxHashMap<u32, VectorIndex>,
609    /// Lazy flat vectors per field — for reranking and merge (doc_ids in memory, vectors via mmap)
610    flat_vectors: FxHashMap<u32, LazyFlatVectorData>,
611    /// One immutable generation of all index-global ANN artifacts.
612    trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
613    /// Sparse vector indexes per field (MaxScore format)
614    sparse_indexes: FxHashMap<u32, SparseIndex>,
615    /// BMP sparse vector indexes per field (BMP format)
616    bmp_indexes: FxHashMap<u32, BmpIndex>,
617    /// Position file handle for phrase queries (lazy loading)
618    positions_handle: Option<FileHandle>,
619    /// Fast-field columnar readers per field_id
620    fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldReader>,
621    /// Per-segment MaxScore threshold (f32 stored as AtomicU32 bits).
622    /// Allows per-field MaxScore groups within a single query to share thresholds:
623    /// field A's result seeds field B's pruning on the same segment.
624    /// Hot-metadata pin accounting (see `segment::pin`)
625    #[cfg(feature = "native")]
626    pin_report: crate::segment::pin::PinReport,
627}
628
629impl SegmentReader {
630    /// Open a segment with lazy loading
631    pub async fn open<D: Directory>(
632        dir: &D,
633        segment_id: SegmentId,
634        schema: Arc<Schema>,
635        cache_blocks: usize,
636    ) -> Result<Self> {
637        Self::open_with_cache_blocks(dir, segment_id, schema, cache_blocks, cache_blocks).await
638    }
639
640    /// Open a segment with independent term-dictionary and document-store caches.
641    ///
642    /// [`Self::open`] keeps the historical single-capacity API for standalone
643    /// callers. Native indexes use this method so `IndexConfig::store_cache_blocks`
644    /// is not silently replaced by the (usually much larger) term cache capacity.
645    pub async fn open_with_cache_blocks<D: Directory>(
646        dir: &D,
647        segment_id: SegmentId,
648        schema: Arc<Schema>,
649        term_cache_blocks: usize,
650        store_cache_blocks: usize,
651    ) -> Result<Self> {
652        let files = SegmentFiles::new(segment_id.0);
653
654        // Read metadata (small, always loaded)
655        let meta_slice = dir.open_read(&files.meta).await?;
656        let meta_bytes = meta_slice.read_bytes().await?;
657        let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
658        debug_assert_eq!(meta.id, segment_id.0);
659
660        // Open term dictionary with lazy loading (fetches ranges on demand)
661        let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
662        let term_dict = AsyncSSTableReader::open(term_dict_handle, term_cache_blocks).await?;
663
664        // Get postings file handle (lazy - fetches ranges on demand)
665        let postings_handle = dir.open_lazy(&files.postings).await?;
666
667        // Open store with lazy loading
668        let store_handle = dir.open_lazy(&files.store).await?;
669        let store = AsyncStoreReader::open(store_handle, store_cache_blocks).await?;
670
671        // Load dense vector indexes from unified .vectors file
672        let vectors_data = loader::load_vectors_file(dir, &files, &schema, meta.num_docs).await?;
673        let vector_indexes = vectors_data.indexes;
674        let flat_vectors = vectors_data.flat_vectors;
675
676        // Fields served by an ANN index only touch flat vectors for scattered
677        // rerank reads — disable readahead for them once at open. Flat-only
678        // fields keep default advice: brute-force scans them sequentially.
679        // Advice is sticky on the mapping, so per-query re-advising is wasted.
680        #[cfg(feature = "native")]
681        for (field_id, lazy_flat) in &flat_vectors {
682            if vector_indexes.contains_key(field_id) {
683                lazy_flat.advise_random_access();
684            }
685        }
686
687        // Load sparse vector indexes from .sparse file (MaxScore + BMP)
688        let sparse_data = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
689        let sparse_indexes = sparse_data.maxscore_indexes;
690        let bmp_indexes = sparse_data.bmp_indexes;
691
692        // Open positions file handle (if exists) - offsets are now in TermInfo
693        let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
694
695        // Load fast-field columns from .fast file
696        let fast_fields = loader::load_fast_fields_file(dir, &files, &schema).await?;
697
698        // Log segment loading stats
699        {
700            let mut parts = vec![format!(
701                "[segment] loaded {:016x}: docs={}",
702                segment_id.0, meta.num_docs
703            )];
704            if !vector_indexes.is_empty() || !flat_vectors.is_empty() {
705                parts.push(format!(
706                    "dense: {} ann + {} flat fields",
707                    vector_indexes.len(),
708                    flat_vectors.len()
709                ));
710            }
711            for (field_id, idx) in &sparse_indexes {
712                parts.push(format!(
713                    "sparse field {}: {} dims, ~{:.1} KB",
714                    field_id,
715                    idx.num_dimensions(),
716                    idx.num_dimensions() as f64 * 24.0 / 1024.0
717                ));
718            }
719            for (field_id, idx) in &bmp_indexes {
720                parts.push(format!(
721                    "bmp field {}: {} dims, {} blocks",
722                    field_id,
723                    idx.dims(),
724                    idx.num_blocks
725                ));
726            }
727            if !fast_fields.is_empty() {
728                parts.push(format!("fast: {} fields", fast_fields.len()));
729            }
730            log::debug!("{}", parts.join(", "));
731        }
732
733        #[allow(unused_mut)]
734        let mut reader = Self {
735            meta,
736            term_dict: Arc::new(term_dict),
737            postings_handle,
738            store: Arc::new(store),
739            schema,
740            vector_indexes,
741            flat_vectors,
742            trained_vectors: Arc::new(crate::segment::TrainedVectorStructures::default()),
743            sparse_indexes,
744            bmp_indexes,
745            positions_handle,
746            fast_fields,
747            #[cfg(feature = "native")]
748            pin_report: Default::default(),
749        };
750
751        // Pin hot metadata per the process-wide policy (no-op when disabled)
752        #[cfg(feature = "native")]
753        reader.apply_pin_policy(&crate::segment::pin::pin_policy().to_owned());
754
755        Ok(reader)
756    }
757
758    /// Pin per-query-mandatory metadata sections in priority order until the
759    /// budget is exhausted (see `segment::pin` and docs/hot-metadata-pinning.md).
760    ///
761    /// Priority: BMP block-offset tables → sparse skip sections → doc-id maps
762    /// → BMP superblock grids. Bulk data (4-bit grid, block data, raw vectors)
763    /// is never pinned. Fail-loud: budget exhaustion and mlock failures are
764    /// logged and visible via `SegmentMemoryStats::{pin_intended_bytes,
765    /// pinned_metadata_bytes}`.
766    #[cfg(feature = "native")]
767    pub(crate) fn apply_pin_policy(&mut self, policy: &crate::segment::pin::PinPolicy) {
768        use crate::segment::pin::PinReport;
769
770        if !policy.is_enabled() {
771            return;
772        }
773        let mut remaining = policy.budget_bytes;
774        let mut report = PinReport::default();
775
776        // Priority 1: BMP block-offset tables
777        for bmp in self.bmp_indexes.values_mut() {
778            bmp.pin_block_starts(policy.mode, &mut remaining, &mut report);
779        }
780        // Priority 2: sparse skip sections
781        for sparse in self.sparse_indexes.values_mut() {
782            sparse.pin_skip_section(policy.mode, &mut remaining, &mut report);
783        }
784        // Priority 3: doc-id maps
785        for flat in self.flat_vectors.values_mut() {
786            flat.pin_doc_ids(policy.mode, &mut remaining, &mut report);
787        }
788        for bmp in self.bmp_indexes.values_mut() {
789            bmp.pin_doc_maps(policy.mode, &mut remaining, &mut report);
790        }
791        // Priority 4: BMP superblock grids
792        for bmp in self.bmp_indexes.values_mut() {
793            bmp.pin_sb_grid(policy.mode, &mut remaining, &mut report);
794        }
795
796        if report.skipped_budget_bytes > 0 || report.failed_bytes > 0 {
797            log::warn!(
798                "[pin] segment {:016x}: pinned {}/{} bytes (budget skipped {}, mlock failed {}) —                  raise HERMES_PIN_METADATA_BUDGET_MB or RLIMIT_MEMLOCK for full coverage",
799                self.meta.id,
800                report.pinned_bytes,
801                report.intended_bytes,
802                report.skipped_budget_bytes,
803                report.failed_bytes,
804            );
805        } else if report.pinned_bytes > 0 {
806            log::info!(
807                "[pin] segment {:016x}: pinned {} bytes of hot metadata ({:?})",
808                self.meta.id,
809                report.pinned_bytes,
810                policy.mode,
811            );
812        }
813        self.pin_report = report;
814    }
815
816    // NOTE: cross-group MaxScore threshold seeding is query-execution-local
817    // (a Cell in the boolean planner) — it must never live on the shared
818    // SegmentReader, where concurrent queries would leak thresholds into
819    // each other and wrongly prune results.
820
821    pub fn meta(&self) -> &SegmentMeta {
822        &self.meta
823    }
824
825    pub fn num_docs(&self) -> u32 {
826        self.meta.num_docs
827    }
828
829    /// Get average field length for BM25F scoring
830    pub fn avg_field_len(&self, field: Field) -> f32 {
831        self.meta.avg_field_len(field)
832    }
833
834    pub fn schema(&self) -> &Schema {
835        &self.schema
836    }
837
838    /// Get sparse indexes for all fields
839    pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
840        &self.sparse_indexes
841    }
842
843    /// Get sparse index for a specific field (MaxScore format)
844    pub fn sparse_index(&self, field: Field) -> Option<&SparseIndex> {
845        self.sparse_indexes.get(&field.0)
846    }
847
848    /// Get BMP index for a specific field
849    pub fn bmp_index(&self, field: Field) -> Option<&BmpIndex> {
850        self.bmp_indexes.get(&field.0)
851    }
852
853    /// Get all BMP indexes
854    pub fn bmp_indexes(&self) -> &FxHashMap<u32, BmpIndex> {
855        &self.bmp_indexes
856    }
857
858    /// Get vector indexes for all fields
859    pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
860        &self.vector_indexes
861    }
862
863    /// Get lazy flat vectors for all fields (for reranking and merge)
864    pub fn flat_vectors(&self) -> &FxHashMap<u32, LazyFlatVectorData> {
865        &self.flat_vectors
866    }
867
868    /// Get a fast-field reader for a specific field.
869    pub fn fast_field(
870        &self,
871        field_id: u32,
872    ) -> Option<&crate::structures::fast_field::FastFieldReader> {
873        self.fast_fields.get(&field_id)
874    }
875
876    /// Get all fast-field readers.
877    pub fn fast_fields(&self) -> &FxHashMap<u32, crate::structures::fast_field::FastFieldReader> {
878        &self.fast_fields
879    }
880
881    /// Get term dictionary stats for debugging
882    pub fn term_dict_stats(&self) -> SSTableStats {
883        self.term_dict.stats()
884    }
885
886    /// Estimate memory usage of this segment reader
887    pub fn memory_stats(&self) -> SegmentMemoryStats {
888        let term_dict_stats = self.term_dict.stats();
889
890        // Report actual decompressed heap retention. Both caches use variable
891        // boundary blocks, so multiplying a block count by a guessed size can
892        // materially under-report resident memory.
893        let term_dict_cache_bytes = self.term_dict.cached_bytes();
894        let store_cache_bytes = self.store.cached_bytes();
895
896        // Sparse index: SoA dim table + OwnedBytes skip section + BMP grids
897        let sparse_index_bytes: usize = self
898            .sparse_indexes
899            .values()
900            .map(|s| s.estimated_memory_bytes())
901            .sum::<usize>()
902            + self
903                .bmp_indexes
904                .values()
905                .map(|b| b.estimated_memory_bytes())
906                .sum::<usize>();
907
908        // Dense index: vectors are memory-mapped, but we track index structures
909        // IVF indexes keep their segment-level cluster payloads in memory.
910        let dense_index_bytes: usize = self
911            .vector_indexes
912            .values()
913            .map(|v| v.estimated_memory_bytes())
914            .sum();
915
916        #[cfg(feature = "native")]
917        let (pinned_metadata_bytes, pin_intended_bytes) =
918            (self.pin_report.pinned_bytes, self.pin_report.intended_bytes);
919        #[cfg(not(feature = "native"))]
920        let (pinned_metadata_bytes, pin_intended_bytes) = (0u64, 0u64);
921
922        SegmentMemoryStats {
923            segment_id: self.meta.id,
924            num_docs: self.meta.num_docs,
925            term_dict_cache_bytes,
926            store_cache_bytes,
927            sparse_index_bytes,
928            dense_index_bytes,
929            bloom_filter_bytes: term_dict_stats.bloom_filter_size,
930            pinned_metadata_bytes,
931            pin_intended_bytes,
932        }
933    }
934
935    /// Get posting list for a term (async - loads on demand)
936    ///
937    /// For small posting lists (1-3 docs), the data is inlined in the term dictionary
938    /// and no additional I/O is needed. For larger lists, reads from .post file.
939    pub async fn get_postings(
940        &self,
941        field: Field,
942        term: &[u8],
943    ) -> Result<Option<BlockPostingList>> {
944        log::debug!(
945            "SegmentReader::get_postings field={} term_len={}",
946            field.0,
947            term.len()
948        );
949
950        // Build key: field_id + term
951        let mut key = Vec::with_capacity(4 + term.len());
952        key.extend_from_slice(&field.0.to_le_bytes());
953        key.extend_from_slice(term);
954
955        // Look up in term dictionary
956        let term_info = match self.term_dict.get(&key).await? {
957            Some(info) => {
958                log::debug!("SegmentReader::get_postings found term_info");
959                info
960            }
961            None => {
962                log::debug!("SegmentReader::get_postings term not found");
963                return Ok(None);
964            }
965        };
966
967        // Check if posting list is inlined
968        if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
969            // Build BlockPostingList from inline data (no I/O needed!)
970            let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
971            for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
972                posting_list.push(doc_id, tf);
973            }
974            let block_list = BlockPostingList::from_posting_list(&posting_list)?;
975            return Ok(Some(block_list));
976        }
977
978        // External posting list - read from postings file handle (lazy - HTTP range request)
979        let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
980            Error::Corruption("TermInfo has neither inline nor external data".to_string())
981        })?;
982
983        let range = checked_file_range(
984            posting_offset,
985            posting_len,
986            self.postings_handle.len(),
987            "posting",
988        )?;
989        let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
990        let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
991
992        Ok(Some(block_list))
993    }
994
995    /// Get all posting lists for terms that start with `prefix` in the given field.
996    pub async fn get_prefix_postings(
997        &self,
998        field: Field,
999        prefix: &[u8],
1000    ) -> Result<Vec<BlockPostingList>> {
1001        if prefix.is_empty() {
1002            return Err(Error::Query("prefix must not be empty".into()));
1003        }
1004        // Build composite key prefix: field_id ++ prefix
1005        let mut key_prefix = Vec::with_capacity(4 + prefix.len());
1006        key_prefix.extend_from_slice(&field.0.to_le_bytes());
1007        key_prefix.extend_from_slice(prefix);
1008
1009        let (entries, truncated) = self
1010            .term_dict
1011            .prefix_scan_limited(&key_prefix, MAX_PREFIX_TERMS)
1012            .await?;
1013        if truncated {
1014            return Err(Error::Query(format!(
1015                "prefix expands to more than {MAX_PREFIX_TERMS} terms"
1016            )));
1017        }
1018        let posting_count: u64 = entries
1019            .iter()
1020            .map(|(_, term_info)| term_info.doc_freq() as u64)
1021            .sum();
1022        if posting_count > MAX_PREFIX_POSTINGS {
1023            return Err(Error::Query(format!(
1024                "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
1025            )));
1026        }
1027        let mut results = Vec::with_capacity(entries.len());
1028
1029        for (_key, term_info) in entries {
1030            if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1031                let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1032                for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1033                    posting_list.push(doc_id, tf);
1034                }
1035                results.push(BlockPostingList::from_posting_list(&posting_list)?);
1036            } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
1037                let range = checked_file_range(
1038                    posting_offset,
1039                    posting_len,
1040                    self.postings_handle.len(),
1041                    "prefix posting",
1042                )?;
1043                let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
1044                results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
1045            }
1046        }
1047
1048        Ok(results)
1049    }
1050
1051    /// Get document by local doc_id (async - loads on demand).
1052    ///
1053    /// Dense vector fields are hydrated from LazyFlatVectorData (not stored in .store).
1054    /// Uses binary search on sorted doc_ids for O(log N) lookup.
1055    pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
1056        self.doc_with_fields(local_doc_id, None).await
1057    }
1058
1059    /// Get document by local doc_id, hydrating only the specified fields.
1060    ///
1061    /// If `fields` is `None`, all fields (including dense vectors) are hydrated.
1062    /// If `fields` is `Some(set)`, only dense vector fields in the set are hydrated,
1063    /// skipping expensive mmap reads + dequantization for unrequested vector fields.
1064    pub async fn doc_with_fields(
1065        &self,
1066        local_doc_id: DocId,
1067        fields: Option<&rustc_hash::FxHashSet<u32>>,
1068    ) -> Result<Option<Document>> {
1069        let mut doc = match fields {
1070            Some(set) => {
1071                let field_ids: Vec<u32> = set.iter().copied().collect();
1072                match self
1073                    .store
1074                    .get_fields(local_doc_id, &self.schema, &field_ids)
1075                    .await
1076                {
1077                    Ok(Some(d)) => d,
1078                    Ok(None) => return Ok(None),
1079                    Err(e) => return Err(Error::from(e)),
1080                }
1081            }
1082            None => match self.store.get(local_doc_id, &self.schema).await {
1083                Ok(Some(d)) => d,
1084                Ok(None) => return Ok(None),
1085                Err(e) => return Err(Error::from(e)),
1086            },
1087        };
1088
1089        // Hydrate dense vector fields from flat vector data
1090        for (&field_id, lazy_flat) in &self.flat_vectors {
1091            // Skip vector fields not in the requested set
1092            if let Some(set) = fields
1093                && !set.contains(&field_id)
1094            {
1095                continue;
1096            }
1097
1098            let is_binary = lazy_flat.quantization == DenseVectorQuantization::Binary;
1099            let (start, entries) = lazy_flat.flat_indexes_for_doc(local_doc_id);
1100            for (j, &(_doc_id, _ordinal)) in entries.iter().enumerate() {
1101                let flat_idx = start + j;
1102                if is_binary {
1103                    let vbs = lazy_flat.vector_byte_size();
1104                    let mut raw = vec![0u8; vbs];
1105                    match lazy_flat.read_vector_raw_into(flat_idx, &mut raw).await {
1106                        Ok(()) => {
1107                            doc.add_binary_dense_vector(Field(field_id), raw);
1108                        }
1109                        Err(e) => {
1110                            log::warn!("Failed to hydrate binary vector field {}: {}", field_id, e);
1111                        }
1112                    }
1113                } else {
1114                    match lazy_flat.get_vector(flat_idx).await {
1115                        Ok(vec) => {
1116                            doc.add_dense_vector(Field(field_id), vec);
1117                        }
1118                        Err(e) => {
1119                            log::warn!("Failed to hydrate vector field {}: {}", field_id, e);
1120                        }
1121                    }
1122                }
1123            }
1124        }
1125
1126        Ok(Some(doc))
1127    }
1128
1129    /// Prefetch term dictionary blocks for a key range
1130    pub async fn prefetch_terms(
1131        &self,
1132        field: Field,
1133        start_term: &[u8],
1134        end_term: &[u8],
1135    ) -> Result<()> {
1136        let mut start_key = Vec::with_capacity(4 + start_term.len());
1137        start_key.extend_from_slice(&field.0.to_le_bytes());
1138        start_key.extend_from_slice(start_term);
1139
1140        let mut end_key = Vec::with_capacity(4 + end_term.len());
1141        end_key.extend_from_slice(&field.0.to_le_bytes());
1142        end_key.extend_from_slice(end_term);
1143
1144        self.term_dict.prefetch_range(&start_key, &end_key).await?;
1145        Ok(())
1146    }
1147
1148    /// Check if store uses dictionary compression (incompatible with raw merging)
1149    pub fn store_has_dict(&self) -> bool {
1150        self.store.has_dict()
1151    }
1152
1153    /// Get store reference for merge operations
1154    pub fn store(&self) -> &super::store::AsyncStoreReader {
1155        &self.store
1156    }
1157
1158    /// Get raw store blocks for optimized merging
1159    pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
1160        self.store.raw_blocks()
1161    }
1162
1163    /// Get store data slice for raw block access
1164    pub fn store_data_slice(&self) -> &FileHandle {
1165        self.store.data_slice()
1166    }
1167
1168    /// Get all terms from this segment (for merge)
1169    pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
1170        self.term_dict.all_entries().await.map_err(Error::from)
1171    }
1172
1173    /// Get all terms with parsed field and term string (for statistics aggregation)
1174    ///
1175    /// Returns (field, term_string, doc_freq) for each term in the dictionary.
1176    /// Skips terms that aren't valid UTF-8.
1177    pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
1178        let entries = self.term_dict.all_entries().await?;
1179        let mut result = Vec::with_capacity(entries.len());
1180
1181        for (key, term_info) in entries {
1182            // Key format: field_id (4 bytes little-endian) + term bytes
1183            if key.len() > 4 {
1184                let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
1185                let term_bytes = &key[4..];
1186                if let Ok(term_str) = std::str::from_utf8(term_bytes) {
1187                    result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
1188                }
1189            }
1190        }
1191
1192        Ok(result)
1193    }
1194
1195    /// Get streaming iterator over term dictionary (for memory-efficient merge)
1196    pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
1197        self.term_dict.iter()
1198    }
1199
1200    /// Prefetch all term dictionary blocks in a single bulk I/O call.
1201    ///
1202    /// Call before merge iteration to eliminate per-block cache misses.
1203    pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
1204        self.term_dict
1205            .prefetch_all_data_bulk()
1206            .await
1207            .map_err(crate::Error::from)
1208    }
1209
1210    /// Read raw posting bytes at offset
1211    pub async fn read_postings(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
1212        let range = checked_file_range(offset, len, self.postings_handle.len(), "posting")?;
1213        let bytes = self.postings_handle.read_bytes_range(range).await?;
1214        Ok(bytes.to_vec())
1215    }
1216
1217    /// Read raw position bytes at offset (for merge)
1218    pub async fn read_position_bytes(&self, offset: u64, len: u64) -> Result<Option<Vec<u8>>> {
1219        let handle = match &self.positions_handle {
1220            Some(h) => h,
1221            None => return Ok(None),
1222        };
1223        let range = checked_file_range(offset, len, handle.len(), "position")?;
1224        let bytes = handle.read_bytes_range(range).await?;
1225        Ok(Some(bytes.to_vec()))
1226    }
1227
1228    /// Check if this segment has a positions file
1229    pub fn has_positions_file(&self) -> bool {
1230        self.positions_handle.is_some()
1231    }
1232
1233    /// Validate all caller-controlled dense-search inputs before touching ANN
1234    /// structures or entering SIMD code. This is deliberately repeated at the
1235    /// segment boundary so non-server users receive the same safety guarantees.
1236    fn validate_dense_search_request(
1237        &self,
1238        field: Field,
1239        query: &[f32],
1240        nprobe: usize,
1241        rerank_factor: f32,
1242        combiner: crate::query::MultiValueCombiner,
1243    ) -> Result<DenseSearchParams> {
1244        let entry = self
1245            .schema
1246            .get_field_entry(field)
1247            .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1248        if entry.field_type != crate::dsl::FieldType::DenseVector {
1249            return Err(Error::InvalidFieldType {
1250                expected: "dense_vector".to_string(),
1251                got: format!("{:?}", entry.field_type),
1252            });
1253        }
1254        let config = entry.dense_vector_config.as_ref().ok_or_else(|| {
1255            Error::Schema(format!(
1256                "dense vector field '{}' has no dense vector configuration",
1257                entry.name
1258            ))
1259        })?;
1260
1261        if query.is_empty() {
1262            return Err(Error::Query(format!(
1263                "dense query vector for field '{}' must not be empty",
1264                entry.name
1265            )));
1266        }
1267        if query.len() != config.dim {
1268            return Err(Error::Query(format!(
1269                "dense query vector dimension {} does not match field '{}' dimension {}",
1270                query.len(),
1271                entry.name,
1272                config.dim
1273            )));
1274        }
1275        if let Some((index, value)) = query
1276            .iter()
1277            .enumerate()
1278            .find(|(_, value)| !value.is_finite())
1279        {
1280            return Err(Error::Query(format!(
1281                "dense query vector for field '{}' contains non-finite value {value} at index {index}",
1282                entry.name
1283            )));
1284        }
1285
1286        // A zero query override means "use the schema". Legacy schemas may
1287        // contain zero for flat fields, so retain 32 as a final ANN fallback.
1288        let nprobe = match (nprobe, config.nprobe) {
1289            (0, 0) => 32,
1290            (0, schema_nprobe) => schema_nprobe,
1291            (query_nprobe, _) => query_nprobe,
1292        };
1293        if nprobe > MAX_DENSE_NPROBE {
1294            return Err(Error::Query(format!(
1295                "dense nprobe must be at most {MAX_DENSE_NPROBE}, got {nprobe}"
1296            )));
1297        }
1298
1299        // Validate the factor here even for empty segments. Otherwise malformed
1300        // requests would succeed or fail depending on segment contents.
1301        checked_dense_fetch_k(0, rerank_factor)?;
1302        combiner.validate().map_err(Error::Query)?;
1303
1304        Ok(DenseSearchParams {
1305            dim: config.dim,
1306            nprobe,
1307            unit_norm: config.unit_norm,
1308        })
1309    }
1310
1311    fn validate_binary_search_request(&self, field: Field, query: &[u8]) -> Result<usize> {
1312        let entry = self
1313            .schema
1314            .get_field_entry(field)
1315            .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1316        if entry.field_type != crate::dsl::FieldType::BinaryDenseVector {
1317            return Err(Error::InvalidFieldType {
1318                expected: "binary_dense_vector".to_string(),
1319                got: format!("{:?}", entry.field_type),
1320            });
1321        }
1322        let config = entry.binary_dense_vector_config.as_ref().ok_or_else(|| {
1323            Error::Schema(format!(
1324                "binary dense vector field '{}' has no configuration",
1325                entry.name
1326            ))
1327        })?;
1328        if config.dim == 0 || !config.dim.is_multiple_of(8) {
1329            return Err(Error::Schema(format!(
1330                "binary dense vector field '{}' has invalid dimension {}",
1331                entry.name, config.dim
1332            )));
1333        }
1334        if query.len() != config.byte_len() {
1335            return Err(Error::Query(format!(
1336                "binary query byte length {} does not match field '{}' byte length {}",
1337                query.len(),
1338                entry.name,
1339                config.byte_len()
1340            )));
1341        }
1342        Ok(config.dim)
1343    }
1344
1345    /// Batch cosine scoring on raw quantized bytes.
1346    ///
1347    /// Dispatches to the appropriate SIMD scorer based on quantization type.
1348    /// Vectors file uses data-first layout (offset 0) with 8-byte padding between
1349    /// fields, so mmap slices are always properly aligned for f32/f16/u8 access.
1350    fn score_quantized_batch(
1351        query: &[f32],
1352        raw: &[u8],
1353        quant: crate::dsl::DenseVectorQuantization,
1354        dim: usize,
1355        scores: &mut [f32],
1356        unit_norm: bool,
1357    ) -> Result<()> {
1358        use crate::dsl::DenseVectorQuantization;
1359        use crate::structures::simd;
1360
1361        if query.len() != dim {
1362            return Err(Error::Query(format!(
1363                "dense SIMD query dimension {} does not match vector dimension {dim}",
1364                query.len()
1365            )));
1366        }
1367        let element_size = match quant {
1368            DenseVectorQuantization::F32 => std::mem::size_of::<f32>(),
1369            DenseVectorQuantization::F16 => std::mem::size_of::<u16>(),
1370            DenseVectorQuantization::UInt8 => 1,
1371            DenseVectorQuantization::Binary => {
1372                return Err(Error::InvalidFieldType {
1373                    expected: "non-binary dense vector".to_string(),
1374                    got: "binary dense vector".to_string(),
1375                });
1376            }
1377        };
1378        let required_bytes = scores
1379            .len()
1380            .checked_mul(dim)
1381            .and_then(|elements| elements.checked_mul(element_size))
1382            .ok_or_else(|| Error::Corruption("dense vector batch byte length overflow".into()))?;
1383        if raw.len() < required_bytes {
1384            return Err(Error::Corruption(format!(
1385                "dense vector batch is truncated: need {required_bytes} bytes, got {}",
1386                raw.len()
1387            )));
1388        }
1389        if quant == DenseVectorQuantization::F16
1390            && required_bytes > 0
1391            && !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<u16>())
1392        {
1393            return Err(Error::Corruption(
1394                "f16 vector data is not 2-byte aligned".to_string(),
1395            ));
1396        }
1397
1398        match (quant, unit_norm) {
1399            (DenseVectorQuantization::F32, false) => {
1400                let num_floats = scores.len() * dim;
1401                if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1402                    return Err(Error::Corruption(
1403                        "f32 vector data is not 4-byte aligned".to_string(),
1404                    ));
1405                }
1406                let vectors: &[f32] =
1407                    unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1408                simd::batch_cosine_scores(query, vectors, dim, scores);
1409            }
1410            (DenseVectorQuantization::F32, true) => {
1411                let num_floats = scores.len() * dim;
1412                if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1413                    return Err(Error::Corruption(
1414                        "f32 vector data is not 4-byte aligned".to_string(),
1415                    ));
1416                }
1417                let vectors: &[f32] =
1418                    unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1419                simd::batch_dot_scores(query, vectors, dim, scores);
1420            }
1421            (DenseVectorQuantization::F16, false) => {
1422                simd::batch_cosine_scores_f16(query, raw, dim, scores);
1423            }
1424            (DenseVectorQuantization::F16, true) => {
1425                simd::batch_dot_scores_f16(query, raw, dim, scores);
1426            }
1427            (DenseVectorQuantization::UInt8, false) => {
1428                simd::batch_cosine_scores_u8(query, raw, dim, scores);
1429            }
1430            (DenseVectorQuantization::UInt8, true) => {
1431                simd::batch_dot_scores_u8(query, raw, dim, scores);
1432            }
1433            (DenseVectorQuantization::Binary, _) => unreachable!("validated above"),
1434        }
1435        Ok(())
1436    }
1437
1438    /// Search dense vectors through the production IVF-PQ index.
1439    ///
1440    /// Returns VectorSearchResult with ordinal tracking for multi-value fields.
1441    /// Doc IDs are segment-local.
1442    /// For multi-valued documents, scores are combined using the specified combiner.
1443    pub async fn search_dense_vector(
1444        &self,
1445        field: Field,
1446        query: &[f32],
1447        k: usize,
1448        nprobe: usize,
1449        rerank_factor: f32,
1450        combiner: crate::query::MultiValueCombiner,
1451    ) -> Result<Vec<VectorSearchResult>> {
1452        self.search_dense_vector_impl(field, query, k, nprobe, rerank_factor, combiner, None)
1453            .await
1454    }
1455
1456    #[allow(clippy::too_many_arguments)]
1457    pub(crate) async fn search_dense_vector_with_probe_cache(
1458        &self,
1459        field: Field,
1460        query: &[f32],
1461        k: usize,
1462        nprobe: usize,
1463        rerank_factor: f32,
1464        combiner: crate::query::MultiValueCombiner,
1465        probe_cache: &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
1466    ) -> Result<Vec<VectorSearchResult>> {
1467        self.search_dense_vector_impl(
1468            field,
1469            query,
1470            k,
1471            nprobe,
1472            rerank_factor,
1473            combiner,
1474            Some(probe_cache),
1475        )
1476        .await
1477    }
1478
1479    #[allow(clippy::too_many_arguments)]
1480    async fn search_dense_vector_impl(
1481        &self,
1482        field: Field,
1483        query: &[f32],
1484        k: usize,
1485        nprobe: usize,
1486        rerank_factor: f32,
1487        combiner: crate::query::MultiValueCombiner,
1488        probe_cache: Option<
1489            &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
1490        >,
1491    ) -> Result<Vec<VectorSearchResult>> {
1492        let params =
1493            self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
1494        let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
1495        if k == 0 {
1496            return Ok(Vec::new());
1497        }
1498
1499        let ann_index = self.vector_indexes.get(&field.0);
1500        let lazy_flat = self.flat_vectors.get(&field.0);
1501        // No vectors at all for this field
1502        if ann_index.is_none() && lazy_flat.is_none() {
1503            return Ok(Vec::new());
1504        }
1505
1506        if ann_index.is_some() && lazy_flat.is_none() {
1507            return Err(Error::Corruption(format!(
1508                "dense ANN field {} is missing flat vector storage",
1509                field.0
1510            )));
1511        }
1512
1513        if let Some(flat) = lazy_flat
1514            && flat.dim != params.dim
1515        {
1516            return Err(Error::Corruption(format!(
1517                "dense vector field {} has schema dimension {} but flat storage dimension {}",
1518                field.0, params.dim, flat.dim
1519            )));
1520        }
1521
1522        // Results are (doc_id, ordinal, score) where score = similarity (higher = better)
1523        let t0 = std::time::Instant::now();
1524        let mut flat_results = None;
1525        let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
1526            // ANN search through the segment's IVF-PQ payload.
1527            match index {
1528                VectorIndex::IvfPq(lazy) => {
1529                    let index = lazy.get().ok_or_else(|| {
1530                        Error::Schema("IVF-PQ index deserialization failed".to_string())
1531                    })?;
1532                    let codebook =
1533                        self.trained_vectors
1534                            .codebooks
1535                            .get(&field.0)
1536                            .ok_or_else(|| {
1537                                Error::Schema(format!(
1538                                    "IVF-PQ index requires a global codebook for field {}",
1539                                    field.0
1540                                ))
1541                            })?;
1542                    let centroids =
1543                        self.trained_vectors
1544                            .centroids
1545                            .get(&field.0)
1546                            .ok_or_else(|| {
1547                                Error::Schema(format!(
1548                                    "IVF-PQ index requires coarse centroids for field {}",
1549                                    field.0
1550                                ))
1551                            })?;
1552                    validate_coarse_centroids(centroids, params.dim)?;
1553                    if index.config.dim != params.dim
1554                        || codebook.config.dim != params.dim
1555                        || index.config.code_size != codebook.config.num_subspaces
1556                        || index.centroids_version != centroids.version
1557                        || index.codebook_version != codebook.version
1558                        || index
1559                            .clusters
1560                            .iter()
1561                            .any(|(cluster_id, _)| *cluster_id >= centroids.num_clusters)
1562                    {
1563                        return Err(Error::Corruption(format!(
1564                            "IVF-PQ index/codebook/centroid metadata does not match schema dimension {}",
1565                            params.dim
1566                        )));
1567                    }
1568                    let routing = self
1569                        .schema
1570                        .get_field_entry(field)
1571                        .and_then(|entry| entry.dense_vector_config.as_ref())
1572                        .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
1573                            config.ivf_routing
1574                        });
1575                    let query_plan = float_query_plan(
1576                        centroids,
1577                        codebook,
1578                        query,
1579                        params.nprobe,
1580                        routing,
1581                        probe_cache,
1582                    )?;
1583                    let flat = lazy_flat.expect("ANN/flat pairing validated above");
1584                    index
1585                        .search_distinct_documents(
1586                            fetch_k.min(flat.num_docs_with_vectors()),
1587                            &query_plan,
1588                        )
1589                        .into_iter()
1590                        .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
1591                        .collect()
1592                }
1593                VectorIndex::BinaryIvf(_) => {
1594                    // Binary IVF serves Hamming queries only (BinaryDenseVectorQuery)
1595                    Vec::new()
1596                }
1597            }
1598        } else if let Some(lazy_flat) = lazy_flat {
1599            // Batched brute-force from lazy flat vectors (native-precision SIMD scoring).
1600            // Combine every value of a document before document-level top-k;
1601            // vector-level top-k loses documents on multi-valued fields.
1602            log::debug!(
1603                "[search_dense] field {}: brute-force on {} vectors (dim={}, quant={:?})",
1604                field.0,
1605                lazy_flat.num_vectors,
1606                lazy_flat.dim,
1607                lazy_flat.quantization
1608            );
1609            let dim = lazy_flat.dim;
1610            let n = lazy_flat.num_vectors;
1611            let quant = lazy_flat.quantization;
1612            let batch_len =
1613                bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
1614            let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
1615            let mut scores = vec![0f32; batch_len];
1616
1617            for batch_start in (0..n).step_by(batch_len) {
1618                let batch_count = batch_len.min(n - batch_start);
1619                let batch_bytes = lazy_flat
1620                    .read_vectors_batch(batch_start, batch_count)
1621                    .await
1622                    .map_err(crate::Error::Io)?;
1623                let raw = batch_bytes.as_slice();
1624
1625                Self::score_quantized_batch(
1626                    query,
1627                    raw,
1628                    quant,
1629                    dim,
1630                    &mut scores[..batch_count],
1631                    params.unit_norm,
1632                )?;
1633
1634                for (i, &score) in scores.iter().enumerate().take(batch_count) {
1635                    let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1636                    collector.push(doc_id, ordinal, score);
1637                }
1638            }
1639
1640            flat_results = Some(collector.into_results());
1641            Vec::new()
1642        } else {
1643            return Ok(Vec::new());
1644        };
1645        let l1_elapsed = t0.elapsed();
1646        {
1647            let kind = match ann_index {
1648                Some(VectorIndex::IvfPq(_)) => "ivf_pq",
1649                Some(VectorIndex::BinaryIvf(_)) => "binary_ivf",
1650                None => "flat",
1651            };
1652            crate::observe::dense_l1(
1653                self.schema.index_label(),
1654                self.schema.get_field_name(field).unwrap_or("?"),
1655                kind,
1656                l1_elapsed.as_secs_f64(),
1657                flat_results.as_ref().map_or(results.len(), Vec::len),
1658            );
1659        }
1660        log::debug!(
1661            "[search_dense] field {}: L1 returned {} candidates in {:.1}ms",
1662            field.0,
1663            flat_results.as_ref().map_or(results.len(), Vec::len),
1664            l1_elapsed.as_secs_f64() * 1000.0
1665        );
1666
1667        if let Some(results) = flat_results {
1668            return Ok(results);
1669        }
1670
1671        // Rerank ANN candidates using raw vectors from lazy flat (binary search lookup)
1672        // Uses native-precision SIMD scoring on quantized bytes — no dequantization overhead.
1673        if ann_index.is_some()
1674            && !results.is_empty()
1675            && let Some(lazy_flat) = lazy_flat
1676        {
1677            let t_rerank = std::time::Instant::now();
1678            let dim = lazy_flat.dim;
1679            let quant = lazy_flat.quantization;
1680            let vbs = lazy_flat.vector_byte_size();
1681
1682            // Deduplicate ANN hits by document, then exact-score every value
1683            // of each candidate document so the configured combiner sees a
1684            // complete value set.
1685            let (expanded, mut resolved) = expand_ann_candidate_documents(&results, lazy_flat)?;
1686            results = expanded;
1687
1688            let t_resolve = t_rerank.elapsed();
1689            if !resolved.is_empty() {
1690                // Sort by flat_idx for sequential mmap access (better page locality)
1691                resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
1692
1693                // Read and score bounded chunks. A legal high-recall request
1694                // can resolve hundreds of thousands of candidates; allocating
1695                // all raw vectors at once used to require gigabytes per segment.
1696                let batch_len = bounded_vector_score_batch(vbs, DENSE_SCORE_BATCH);
1697                let max_batch = batch_len.min(resolved.len());
1698                let max_raw_len = max_batch
1699                    .checked_mul(vbs)
1700                    .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
1701                let mut raw_buf = vec![0u8; max_raw_len];
1702                let mut scores = vec![0f32; max_batch];
1703                let mut read_elapsed = std::time::Duration::ZERO;
1704                let mut score_elapsed = std::time::Duration::ZERO;
1705
1706                for chunk in resolved.chunks(batch_len) {
1707                    // Advise only the bounded chunk about to be consumed. A
1708                    // whole-candidate MADV_WILLNEED can evict the rest of the
1709                    // working set long before those pages are read.
1710                    #[cfg(feature = "native")]
1711                    lazy_flat.prefetch_vectors(chunk.iter().map(|&(_, flat_idx)| flat_idx));
1712                    let raw_len = chunk
1713                        .len()
1714                        .checked_mul(vbs)
1715                        .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
1716                    let raw = &mut raw_buf[..raw_len];
1717
1718                    let t_read = std::time::Instant::now();
1719                    for (buf_idx, &(_, flat_idx)) in chunk.iter().enumerate() {
1720                        lazy_flat
1721                            .read_vector_raw_into(
1722                                flat_idx,
1723                                &mut raw[buf_idx * vbs..(buf_idx + 1) * vbs],
1724                            )
1725                            .await
1726                            .map_err(crate::Error::Io)?;
1727                    }
1728                    read_elapsed += t_read.elapsed();
1729
1730                    let t_score = std::time::Instant::now();
1731                    Self::score_quantized_batch(
1732                        query,
1733                        raw,
1734                        quant,
1735                        dim,
1736                        &mut scores[..chunk.len()],
1737                        params.unit_norm,
1738                    )?;
1739                    score_elapsed += t_score.elapsed();
1740
1741                    for (buf_idx, &(ri, _)) in chunk.iter().enumerate() {
1742                        results[ri].2 = scores[buf_idx];
1743                    }
1744                }
1745
1746                crate::observe::dense_rerank(
1747                    self.schema.index_label(),
1748                    self.schema.get_field_name(field).unwrap_or("?"),
1749                    t_rerank.elapsed().as_secs_f64(),
1750                    t_resolve.as_secs_f64(),
1751                    read_elapsed.as_secs_f64(),
1752                    resolved.len(),
1753                );
1754                log::debug!(
1755                    "[search_dense] field {}: rerank {} vectors (dim={}, quant={:?}, {}B/vec): resolve={:.1}ms read={:.1}ms score={:.1}ms",
1756                    field.0,
1757                    resolved.len(),
1758                    dim,
1759                    quant,
1760                    vbs,
1761                    t_resolve.as_secs_f64() * 1000.0,
1762                    read_elapsed.as_secs_f64() * 1000.0,
1763                    score_elapsed.as_secs_f64() * 1000.0,
1764                );
1765            }
1766
1767            log::debug!(
1768                "[search_dense] field {}: rerank total={:.1}ms",
1769                field.0,
1770                t_rerank.elapsed().as_secs_f64() * 1000.0
1771            );
1772        }
1773
1774        Ok(combine_grouped_ordinal_results(results, combiner, k))
1775    }
1776
1777    /// Search binary dense vectors using IVF when available, otherwise
1778    /// brute-force Hamming distance.
1779    ///
1780    /// Returns VectorSearchResult with ordinal tracking.
1781    async fn search_binary_dense_vector_impl(
1782        &self,
1783        field: Field,
1784        query: &[u8],
1785        k: usize,
1786        combiner: crate::query::MultiValueCombiner,
1787        probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
1788    ) -> Result<Vec<VectorSearchResult>> {
1789        let schema_dim = self.validate_binary_search_request(field, query)?;
1790        combiner.validate().map_err(Error::Query)?;
1791        if k == 0 {
1792            return Ok(Vec::new());
1793        }
1794        let t0 = crate::observe::Timer::start();
1795        if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
1796            let ivf = lazy.get().ok_or_else(|| {
1797                Error::Corruption(format!(
1798                    "global binary IVF field {} could not be decoded",
1799                    field.0
1800                ))
1801            })?;
1802            let config = self
1803                .schema
1804                .get_field_entry(field)
1805                .and_then(|entry| entry.binary_dense_vector_config.as_ref())
1806                .ok_or_else(|| {
1807                    Error::Schema(format!(
1808                        "binary IVF field {} has no schema configuration",
1809                        field.0
1810                    ))
1811                })?;
1812            let quantizer = self
1813                .trained_vectors
1814                .binary_quantizers
1815                .get(&field.0)
1816                .ok_or_else(|| {
1817                    Error::Schema(format!(
1818                        "global binary IVF field {} has no loaded quantizer",
1819                        field.0
1820                    ))
1821                })?;
1822            if ivf.dim_bits != schema_dim
1823                || quantizer.dim_bits != schema_dim
1824                || ivf.quantizer_version != quantizer.version
1825                || ivf.num_clusters != quantizer.num_clusters
1826            {
1827                return Err(Error::Corruption(format!(
1828                    "global binary IVF field {} does not match its quantizer/schema generation",
1829                    field.0
1830                )));
1831            }
1832            let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
1833                Error::Corruption(format!(
1834                    "global binary IVF field {} is missing flat vector storage",
1835                    field.0
1836                ))
1837            })?;
1838            let clusters = binary_probe_clusters(
1839                quantizer,
1840                query,
1841                config.nprobe,
1842                config.ivf_routing,
1843                probe_cache,
1844            )?;
1845            let single_valued = flat.num_vectors == flat.num_docs_with_vectors();
1846            let candidate_docs = k.min(flat.num_docs_with_vectors());
1847            let ann_results = if single_valued {
1848                ivf.search_in_clusters(query, candidate_docs, &clusters)
1849            } else {
1850                ivf.search_distinct_documents_in_clusters(query, candidate_docs, &clusters)
1851            };
1852            let results =
1853                exact_score_binary_candidate_documents(&ann_results, flat, query, schema_dim)
1854                    .await?;
1855            crate::observe::dense_l1(
1856                self.schema.index_label(),
1857                self.schema.get_field_name(field).unwrap_or("?"),
1858                "global_binary_ivf",
1859                t0.secs(),
1860                results.len(),
1861            );
1862            return Ok(combine_grouped_ordinal_results(results, combiner, k));
1863        }
1864        let lazy_flat = match self.flat_vectors.get(&field.0) {
1865            Some(f) => f,
1866            None => return Ok(Vec::new()),
1867        };
1868
1869        let dim_bits = lazy_flat.dim;
1870        let byte_len = lazy_flat.vector_byte_size();
1871        let n = lazy_flat.num_vectors;
1872
1873        if dim_bits != schema_dim {
1874            return Err(Error::Corruption(format!(
1875                "binary vector field {} has schema dimension {} but flat storage dimension {}",
1876                field.0, schema_dim, dim_bits
1877            )));
1878        }
1879
1880        if byte_len != query.len() {
1881            return Err(Error::Schema(format!(
1882                "Binary query vector byte length {} != field byte length {}",
1883                query.len(),
1884                byte_len
1885            )));
1886        }
1887
1888        let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
1889        let mut collector = FlatDocumentCollector::new(k, combiner);
1890        let mut scores = vec![0f32; batch_len];
1891
1892        for batch_start in (0..n).step_by(batch_len) {
1893            let batch_count = batch_len.min(n - batch_start);
1894            let batch_bytes = lazy_flat
1895                .read_vectors_batch(batch_start, batch_count)
1896                .await
1897                .map_err(crate::Error::Io)?;
1898            let raw = batch_bytes.as_slice();
1899
1900            crate::structures::simd::batch_hamming_scores(
1901                query,
1902                raw,
1903                byte_len,
1904                dim_bits,
1905                &mut scores[..batch_count],
1906            );
1907
1908            for (i, &score) in scores.iter().enumerate().take(batch_count) {
1909                let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1910                collector.push(doc_id, ordinal, score);
1911            }
1912        }
1913
1914        let results = collector.into_results();
1915
1916        crate::observe::dense_l1(
1917            self.schema.index_label(),
1918            self.schema.get_field_name(field).unwrap_or("?"),
1919            "binary_flat",
1920            t0.secs(),
1921            results.len(),
1922        );
1923        Ok(results)
1924    }
1925
1926    pub async fn search_binary_dense_vector(
1927        &self,
1928        field: Field,
1929        query: &[u8],
1930        k: usize,
1931        combiner: crate::query::MultiValueCombiner,
1932    ) -> Result<Vec<VectorSearchResult>> {
1933        self.search_binary_dense_vector_impl(field, query, k, combiner, None)
1934            .await
1935    }
1936
1937    pub(crate) async fn search_binary_dense_vector_with_probe_cache(
1938        &self,
1939        field: Field,
1940        query: &[u8],
1941        k: usize,
1942        combiner: crate::query::MultiValueCombiner,
1943        probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
1944    ) -> Result<Vec<VectorSearchResult>> {
1945        self.search_binary_dense_vector_impl(field, query, k, combiner, Some(probe_cache))
1946            .await
1947    }
1948
1949    /// Get coarse centroids for a field.
1950    pub fn coarse_centroids(&self, field_id: u32) -> Option<&Arc<CoarseCentroids>> {
1951        self.trained_vectors.centroids.get(&field_id)
1952    }
1953
1954    pub fn set_trained_vectors(
1955        &mut self,
1956        trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
1957    ) {
1958        self.trained_vectors = trained_vectors;
1959    }
1960
1961    /// Materialize ANN payloads before publishing a newly loaded segment to
1962    /// query threads. Multi-gigabyte bincode decoding on the first request can
1963    /// exceed the RPC deadline; searcher reload keeps serving its old snapshot
1964    /// while this one-time warmup runs. Unchanged readers are reused, so they
1965    /// are not decoded again on every metadata refresh.
1966    #[cfg(feature = "native")]
1967    pub(crate) fn warm_ann_indexes(&self) -> Result<()> {
1968        let started = std::time::Instant::now();
1969        for (&field_id, index) in &self.vector_indexes {
1970            let decoded = match index {
1971                VectorIndex::IvfPq(lazy) => lazy.get().is_some(),
1972                VectorIndex::BinaryIvf(lazy) => lazy.get().is_some(),
1973            };
1974            if !decoded {
1975                return Err(Error::Corruption(format!(
1976                    "ANN payload for field {field_id} could not be decoded"
1977                )));
1978            }
1979        }
1980        if !self.vector_indexes.is_empty() {
1981            log::info!(
1982                "[vector_warmup] segment {:016x}: decoded {} ANN payloads in {:.1}s",
1983                self.meta.id,
1984                self.vector_indexes.len(),
1985                started.elapsed().as_secs_f64(),
1986            );
1987        }
1988        Ok(())
1989    }
1990
1991    /// Get the float IVF-PQ vector index for a field (if available).
1992    pub fn get_ivf_pq_vector_index(&self, field: Field) -> Option<Arc<IVFPQIndex>> {
1993        match self.vector_indexes.get(&field.0) {
1994            Some(VectorIndex::IvfPq(lazy)) => lazy.get().map(Arc::clone),
1995            _ => None,
1996        }
1997    }
1998
1999    /// Get the vector index type for a field
2000    pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
2001        self.vector_indexes.get(&field.0)
2002    }
2003
2004    /// Get positions for a term (for phrase queries)
2005    ///
2006    /// Position offsets are now embedded in TermInfo, so we first look up
2007    /// the term to get its TermInfo, then use position_info() to get the offset.
2008    pub async fn get_positions(
2009        &self,
2010        field: Field,
2011        term: &[u8],
2012    ) -> Result<Option<crate::structures::PositionPostingList>> {
2013        // Get positions handle
2014        let handle = match &self.positions_handle {
2015            Some(h) => h,
2016            None => return Ok(None),
2017        };
2018
2019        // Build key: field_id + term
2020        let mut key = Vec::with_capacity(4 + term.len());
2021        key.extend_from_slice(&field.0.to_le_bytes());
2022        key.extend_from_slice(term);
2023
2024        // Look up term in dictionary to get TermInfo with position offset
2025        let term_info = match self.term_dict.get(&key).await? {
2026            Some(info) => info,
2027            None => return Ok(None),
2028        };
2029
2030        // Get position offset from TermInfo
2031        let (offset, length) = match term_info.position_info() {
2032            Some((o, l)) => (o, l),
2033            None => return Ok(None),
2034        };
2035
2036        // Read the position data only after validating untrusted offsets from
2037        // the term dictionary. Direct `offset + length` can wrap in release
2038        // builds and alias an unrelated range.
2039        let range = checked_file_range(offset, length, handle.len(), "position list")?;
2040        let slice = handle.slice(range);
2041        let data = slice.read_bytes().await?;
2042
2043        // Deserialize
2044        let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2045
2046        Ok(Some(pos_list))
2047    }
2048
2049    /// Check if positions are available for a field
2050    pub fn has_positions(&self, field: Field) -> bool {
2051        // Check schema for position mode on this field
2052        if let Some(entry) = self.schema.get_field_entry(field) {
2053            entry.positions.is_some()
2054        } else {
2055            false
2056        }
2057    }
2058}
2059
2060// ── Synchronous search methods (mmap/RAM only) ─────────────────────────────
2061#[cfg(feature = "sync")]
2062impl SegmentReader {
2063    /// Synchronous posting list lookup — requires Inline (mmap/RAM) file handles.
2064    pub fn get_postings_sync(&self, field: Field, term: &[u8]) -> Result<Option<BlockPostingList>> {
2065        // Build key: field_id + term
2066        let mut key = Vec::with_capacity(4 + term.len());
2067        key.extend_from_slice(&field.0.to_le_bytes());
2068        key.extend_from_slice(term);
2069
2070        // Look up in term dictionary (sync)
2071        let term_info = match self.term_dict.get_sync(&key)? {
2072            Some(info) => info,
2073            None => return Ok(None),
2074        };
2075
2076        // Check if posting list is inlined
2077        if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2078            let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2079            for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2080                posting_list.push(doc_id, tf);
2081            }
2082            let block_list = BlockPostingList::from_posting_list(&posting_list)?;
2083            return Ok(Some(block_list));
2084        }
2085
2086        // External posting list — sync range read
2087        let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
2088            Error::Corruption("TermInfo has neither inline nor external data".to_string())
2089        })?;
2090
2091        let range = checked_file_range(
2092            posting_offset,
2093            posting_len,
2094            self.postings_handle.len(),
2095            "posting",
2096        )?;
2097        let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2098        let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
2099
2100        Ok(Some(block_list))
2101    }
2102
2103    /// Synchronous prefix posting list lookup — requires Inline (mmap/RAM) file handles.
2104    pub fn get_prefix_postings_sync(
2105        &self,
2106        field: Field,
2107        prefix: &[u8],
2108    ) -> Result<Vec<BlockPostingList>> {
2109        if prefix.is_empty() {
2110            return Err(Error::Query("prefix must not be empty".into()));
2111        }
2112        let mut key_prefix = Vec::with_capacity(4 + prefix.len());
2113        key_prefix.extend_from_slice(&field.0.to_le_bytes());
2114        key_prefix.extend_from_slice(prefix);
2115
2116        let (entries, truncated) = self
2117            .term_dict
2118            .prefix_scan_limited_sync(&key_prefix, MAX_PREFIX_TERMS)?;
2119        if truncated {
2120            return Err(Error::Query(format!(
2121                "prefix expands to more than {MAX_PREFIX_TERMS} terms"
2122            )));
2123        }
2124        let posting_count: u64 = entries
2125            .iter()
2126            .map(|(_, term_info)| term_info.doc_freq() as u64)
2127            .sum();
2128        if posting_count > MAX_PREFIX_POSTINGS {
2129            return Err(Error::Query(format!(
2130                "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
2131            )));
2132        }
2133        let mut results = Vec::with_capacity(entries.len());
2134
2135        for (_key, term_info) in entries {
2136            if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2137                let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2138                for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2139                    posting_list.push(doc_id, tf);
2140                }
2141                results.push(BlockPostingList::from_posting_list(&posting_list)?);
2142            } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
2143                let range = checked_file_range(
2144                    posting_offset,
2145                    posting_len,
2146                    self.postings_handle.len(),
2147                    "prefix posting",
2148                )?;
2149                let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2150                results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
2151            }
2152        }
2153
2154        Ok(results)
2155    }
2156
2157    /// Synchronous position list lookup — requires Inline (mmap/RAM) file handles.
2158    pub fn get_positions_sync(
2159        &self,
2160        field: Field,
2161        term: &[u8],
2162    ) -> Result<Option<crate::structures::PositionPostingList>> {
2163        let handle = match &self.positions_handle {
2164            Some(h) => h,
2165            None => return Ok(None),
2166        };
2167
2168        // Build key: field_id + term
2169        let mut key = Vec::with_capacity(4 + term.len());
2170        key.extend_from_slice(&field.0.to_le_bytes());
2171        key.extend_from_slice(term);
2172
2173        // Look up term in dictionary (sync)
2174        let term_info = match self.term_dict.get_sync(&key)? {
2175            Some(info) => info,
2176            None => return Ok(None),
2177        };
2178
2179        let (offset, length) = match term_info.position_info() {
2180            Some((o, l)) => (o, l),
2181            None => return Ok(None),
2182        };
2183
2184        let range = checked_file_range(offset, length, handle.len(), "position list")?;
2185        let slice = handle.slice(range);
2186        let data = slice.read_bytes_sync()?;
2187
2188        let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2189        Ok(Some(pos_list))
2190    }
2191
2192    /// Synchronous dense vector search — ANN indexes are already sync,
2193    /// brute-force uses sync mmap reads.
2194    pub fn search_dense_vector_sync(
2195        &self,
2196        field: Field,
2197        query: &[f32],
2198        k: usize,
2199        nprobe: usize,
2200        rerank_factor: f32,
2201        combiner: crate::query::MultiValueCombiner,
2202    ) -> Result<Vec<VectorSearchResult>> {
2203        self.search_dense_vector_sync_impl(field, query, k, nprobe, rerank_factor, combiner, None)
2204    }
2205
2206    #[cfg(feature = "sync")]
2207    #[allow(clippy::too_many_arguments)]
2208    pub(crate) fn search_dense_vector_sync_with_probe_cache(
2209        &self,
2210        field: Field,
2211        query: &[f32],
2212        k: usize,
2213        nprobe: usize,
2214        rerank_factor: f32,
2215        combiner: crate::query::MultiValueCombiner,
2216        probe_cache: &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
2217    ) -> Result<Vec<VectorSearchResult>> {
2218        self.search_dense_vector_sync_impl(
2219            field,
2220            query,
2221            k,
2222            nprobe,
2223            rerank_factor,
2224            combiner,
2225            Some(probe_cache),
2226        )
2227    }
2228
2229    #[cfg(feature = "sync")]
2230    #[allow(clippy::too_many_arguments)]
2231    fn search_dense_vector_sync_impl(
2232        &self,
2233        field: Field,
2234        query: &[f32],
2235        k: usize,
2236        nprobe: usize,
2237        rerank_factor: f32,
2238        combiner: crate::query::MultiValueCombiner,
2239        probe_cache: Option<
2240            &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
2241        >,
2242    ) -> Result<Vec<VectorSearchResult>> {
2243        let params =
2244            self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
2245        let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
2246        if k == 0 {
2247            return Ok(Vec::new());
2248        }
2249
2250        let ann_index = self.vector_indexes.get(&field.0);
2251        let lazy_flat = self.flat_vectors.get(&field.0);
2252        if ann_index.is_none() && lazy_flat.is_none() {
2253            return Ok(Vec::new());
2254        }
2255
2256        if ann_index.is_some() && lazy_flat.is_none() {
2257            return Err(Error::Corruption(format!(
2258                "dense ANN field {} is missing flat vector storage",
2259                field.0
2260            )));
2261        }
2262
2263        if let Some(flat) = lazy_flat
2264            && flat.dim != params.dim
2265        {
2266            return Err(Error::Corruption(format!(
2267                "dense vector field {} has schema dimension {} but flat storage dimension {}",
2268                field.0, params.dim, flat.dim
2269            )));
2270        }
2271
2272        let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
2273            // ANN search (already sync)
2274            match index {
2275                VectorIndex::IvfPq(lazy) => {
2276                    let index = lazy.get().ok_or_else(|| {
2277                        Error::Schema("IVF-PQ index deserialization failed".to_string())
2278                    })?;
2279                    let codebook =
2280                        self.trained_vectors
2281                            .codebooks
2282                            .get(&field.0)
2283                            .ok_or_else(|| {
2284                                Error::Schema(format!(
2285                                    "IVF-PQ index requires a global codebook for field {}",
2286                                    field.0
2287                                ))
2288                            })?;
2289                    let centroids =
2290                        self.trained_vectors
2291                            .centroids
2292                            .get(&field.0)
2293                            .ok_or_else(|| {
2294                                Error::Schema(format!(
2295                                    "IVF-PQ index requires coarse centroids for field {}",
2296                                    field.0
2297                                ))
2298                            })?;
2299                    validate_coarse_centroids(centroids, params.dim)?;
2300                    if index.config.dim != params.dim
2301                        || codebook.config.dim != params.dim
2302                        || index.config.code_size != codebook.config.num_subspaces
2303                        || index.centroids_version != centroids.version
2304                        || index.codebook_version != codebook.version
2305                        || index
2306                            .clusters
2307                            .iter()
2308                            .any(|(cluster_id, _)| *cluster_id >= centroids.num_clusters)
2309                    {
2310                        return Err(Error::Corruption(format!(
2311                            "IVF-PQ index/codebook/centroid metadata does not match schema dimension {}",
2312                            params.dim
2313                        )));
2314                    }
2315                    let routing = self
2316                        .schema
2317                        .get_field_entry(field)
2318                        .and_then(|entry| entry.dense_vector_config.as_ref())
2319                        .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
2320                            config.ivf_routing
2321                        });
2322                    let query_plan = float_query_plan(
2323                        centroids,
2324                        codebook,
2325                        query,
2326                        params.nprobe,
2327                        routing,
2328                        probe_cache,
2329                    )?;
2330                    let flat = lazy_flat.expect("ANN/flat pairing validated above");
2331                    index
2332                        .search_distinct_documents(
2333                            fetch_k.min(flat.num_docs_with_vectors()),
2334                            &query_plan,
2335                        )
2336                        .into_iter()
2337                        .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
2338                        .collect()
2339                }
2340                VectorIndex::BinaryIvf(_) => {
2341                    // Binary IVF serves Hamming queries only (BinaryDenseVectorQuery)
2342                    Vec::new()
2343                }
2344            }
2345        } else if let Some(lazy_flat) = lazy_flat {
2346            // Batched brute-force (sync mmap reads)
2347            let dim = lazy_flat.dim;
2348            let n = lazy_flat.num_vectors;
2349            let quant = lazy_flat.quantization;
2350            let batch_len =
2351                bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
2352            let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
2353            let mut scores = vec![0f32; batch_len];
2354
2355            for batch_start in (0..n).step_by(batch_len) {
2356                let batch_count = batch_len.min(n - batch_start);
2357                let batch_bytes = lazy_flat
2358                    .read_vectors_batch_sync(batch_start, batch_count)
2359                    .map_err(crate::Error::Io)?;
2360                let raw = batch_bytes.as_slice();
2361
2362                Self::score_quantized_batch(
2363                    query,
2364                    raw,
2365                    quant,
2366                    dim,
2367                    &mut scores[..batch_count],
2368                    params.unit_norm,
2369                )?;
2370
2371                for (i, &score) in scores.iter().enumerate().take(batch_count) {
2372                    let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2373                    collector.push(doc_id, ordinal, score);
2374                }
2375            }
2376
2377            return Ok(collector.into_results());
2378        } else {
2379            return Ok(Vec::new());
2380        };
2381
2382        // Rerank ANN candidates using raw vectors (sync)
2383        if ann_index.is_some()
2384            && !results.is_empty()
2385            && let Some(lazy_flat) = lazy_flat
2386        {
2387            let dim = lazy_flat.dim;
2388            let quant = lazy_flat.quantization;
2389            let vbs = lazy_flat.vector_byte_size();
2390
2391            let (expanded, mut resolved) = expand_ann_candidate_documents(&results, lazy_flat)?;
2392            results = expanded;
2393
2394            if !resolved.is_empty() {
2395                resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
2396                let batch_len = bounded_vector_score_batch(vbs, DENSE_SCORE_BATCH);
2397                let max_batch = batch_len.min(resolved.len());
2398                let max_raw_len = max_batch
2399                    .checked_mul(vbs)
2400                    .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
2401                let mut raw_buf = vec![0u8; max_raw_len];
2402                let mut scores = vec![0f32; max_batch];
2403
2404                for chunk in resolved.chunks(batch_len) {
2405                    let raw_len = chunk
2406                        .len()
2407                        .checked_mul(vbs)
2408                        .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
2409                    let raw = &mut raw_buf[..raw_len];
2410                    for (buf_idx, &(_, flat_idx)) in chunk.iter().enumerate() {
2411                        lazy_flat
2412                            .read_vector_raw_into_sync(
2413                                flat_idx,
2414                                &mut raw[buf_idx * vbs..(buf_idx + 1) * vbs],
2415                            )
2416                            .map_err(crate::Error::Io)?;
2417                    }
2418
2419                    Self::score_quantized_batch(
2420                        query,
2421                        raw,
2422                        quant,
2423                        dim,
2424                        &mut scores[..chunk.len()],
2425                        params.unit_norm,
2426                    )?;
2427
2428                    for (buf_idx, &(ri, _)) in chunk.iter().enumerate() {
2429                        results[ri].2 = scores[buf_idx];
2430                    }
2431                }
2432            }
2433        }
2434
2435        Ok(combine_grouped_ordinal_results(results, combiner, k))
2436    }
2437
2438    /// Synchronous binary dense vector search (mmap/RAM only).
2439    ///
2440    /// Mirrors [`Self::search_binary_dense_vector`] for the rayon-parallel
2441    /// sync scorer path used by multi-threaded runtimes.
2442    #[cfg(feature = "sync")]
2443    fn search_binary_dense_vector_sync_impl(
2444        &self,
2445        field: Field,
2446        query: &[u8],
2447        k: usize,
2448        combiner: crate::query::MultiValueCombiner,
2449        probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
2450    ) -> Result<Vec<VectorSearchResult>> {
2451        let schema_dim = self.validate_binary_search_request(field, query)?;
2452        combiner.validate().map_err(Error::Query)?;
2453        if k == 0 {
2454            return Ok(Vec::new());
2455        }
2456        let t0 = crate::observe::Timer::start();
2457        if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
2458            let ivf = lazy.get().ok_or_else(|| {
2459                Error::Corruption(format!(
2460                    "global binary IVF field {} could not be decoded",
2461                    field.0
2462                ))
2463            })?;
2464            let config = self
2465                .schema
2466                .get_field_entry(field)
2467                .and_then(|entry| entry.binary_dense_vector_config.as_ref())
2468                .ok_or_else(|| {
2469                    Error::Schema(format!(
2470                        "binary IVF field {} has no schema configuration",
2471                        field.0
2472                    ))
2473                })?;
2474            let quantizer = self
2475                .trained_vectors
2476                .binary_quantizers
2477                .get(&field.0)
2478                .ok_or_else(|| {
2479                    Error::Schema(format!(
2480                        "global binary IVF field {} has no loaded quantizer",
2481                        field.0
2482                    ))
2483                })?;
2484            if ivf.dim_bits != schema_dim
2485                || quantizer.dim_bits != schema_dim
2486                || ivf.quantizer_version != quantizer.version
2487                || ivf.num_clusters != quantizer.num_clusters
2488            {
2489                return Err(Error::Corruption(format!(
2490                    "global binary IVF field {} does not match its quantizer/schema generation",
2491                    field.0
2492                )));
2493            }
2494            let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
2495                Error::Corruption(format!(
2496                    "global binary IVF field {} is missing flat vector storage",
2497                    field.0
2498                ))
2499            })?;
2500            let clusters = binary_probe_clusters(
2501                quantizer,
2502                query,
2503                config.nprobe,
2504                config.ivf_routing,
2505                probe_cache,
2506            )?;
2507            let candidate_docs = k.min(flat.num_docs_with_vectors());
2508            let ann_results = if flat.num_vectors == flat.num_docs_with_vectors() {
2509                ivf.search_in_clusters(query, candidate_docs, &clusters)
2510            } else {
2511                ivf.search_distinct_documents_in_clusters(query, candidate_docs, &clusters)
2512            };
2513            let results =
2514                exact_score_binary_candidate_documents_sync(&ann_results, flat, query, schema_dim)?;
2515            crate::observe::dense_l1(
2516                self.schema.index_label(),
2517                self.schema.get_field_name(field).unwrap_or("?"),
2518                "global_binary_ivf",
2519                t0.secs(),
2520                results.len(),
2521            );
2522            return Ok(combine_grouped_ordinal_results(results, combiner, k));
2523        }
2524        let lazy_flat = match self.flat_vectors.get(&field.0) {
2525            Some(f) => f,
2526            None => return Ok(Vec::new()),
2527        };
2528
2529        let dim_bits = lazy_flat.dim;
2530        let byte_len = lazy_flat.vector_byte_size();
2531        let n = lazy_flat.num_vectors;
2532
2533        if dim_bits != schema_dim {
2534            return Err(Error::Corruption(format!(
2535                "binary vector field {} has schema dimension {} but flat storage dimension {}",
2536                field.0, schema_dim, dim_bits
2537            )));
2538        }
2539
2540        if byte_len != query.len() {
2541            return Err(Error::Schema(format!(
2542                "Binary query vector byte length {} != field byte length {}",
2543                query.len(),
2544                byte_len
2545            )));
2546        }
2547
2548        let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
2549        let mut collector = FlatDocumentCollector::new(k, combiner);
2550        let mut scores = vec![0f32; batch_len];
2551
2552        for batch_start in (0..n).step_by(batch_len) {
2553            let batch_count = batch_len.min(n - batch_start);
2554            let batch_bytes = lazy_flat
2555                .read_vectors_batch_sync(batch_start, batch_count)
2556                .map_err(crate::Error::Io)?;
2557            let raw = batch_bytes.as_slice();
2558
2559            crate::structures::simd::batch_hamming_scores(
2560                query,
2561                raw,
2562                byte_len,
2563                dim_bits,
2564                &mut scores[..batch_count],
2565            );
2566
2567            for (i, &score) in scores.iter().enumerate().take(batch_count) {
2568                let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2569                collector.push(doc_id, ordinal, score);
2570            }
2571        }
2572
2573        let results = collector.into_results();
2574
2575        crate::observe::dense_l1(
2576            self.schema.index_label(),
2577            self.schema.get_field_name(field).unwrap_or("?"),
2578            "binary_flat",
2579            t0.secs(),
2580            results.len(),
2581        );
2582        Ok(results)
2583    }
2584
2585    #[cfg(feature = "sync")]
2586    pub fn search_binary_dense_vector_sync(
2587        &self,
2588        field: Field,
2589        query: &[u8],
2590        k: usize,
2591        combiner: crate::query::MultiValueCombiner,
2592    ) -> Result<Vec<VectorSearchResult>> {
2593        self.search_binary_dense_vector_sync_impl(field, query, k, combiner, None)
2594    }
2595
2596    #[cfg(feature = "sync")]
2597    pub(crate) fn search_binary_dense_vector_sync_with_probe_cache(
2598        &self,
2599        field: Field,
2600        query: &[u8],
2601        k: usize,
2602        combiner: crate::query::MultiValueCombiner,
2603        probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
2604    ) -> Result<Vec<VectorSearchResult>> {
2605        self.search_binary_dense_vector_sync_impl(field, query, k, combiner, Some(probe_cache))
2606    }
2607}
2608
2609#[cfg(test)]
2610mod dense_search_safety_tests {
2611    use super::*;
2612
2613    #[test]
2614    fn dense_fetch_count_rejects_non_finite_and_unbounded_factors() {
2615        for factor in [
2616            f32::NAN,
2617            f32::INFINITY,
2618            f32::NEG_INFINITY,
2619            0.0,
2620            0.5,
2621            MAX_DENSE_RERANK_FACTOR + 1.0,
2622        ] {
2623            assert!(
2624                checked_dense_fetch_k(10, factor).is_err(),
2625                "factor={factor}"
2626            );
2627        }
2628    }
2629
2630    #[test]
2631    fn flat_document_collector_does_not_let_one_multivalue_doc_crowd_out_others() {
2632        let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2633        collector.push(1, 0, 1.0);
2634        collector.push(1, 1, 0.9);
2635        collector.push(2, 0, 0.8);
2636
2637        let results = collector.into_results();
2638        assert_eq!(
2639            results
2640                .iter()
2641                .map(|result| result.doc_id)
2642                .collect::<Vec<_>>(),
2643            vec![1, 2]
2644        );
2645        assert_eq!(results[0].ordinals.len(), 2);
2646    }
2647
2648    #[test]
2649    fn flat_document_collector_evicts_by_score_then_doc_id() {
2650        let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2651        collector.push(1, 0, 0.5);
2652        collector.push(3, 0, 0.8);
2653        collector.push(2, 0, 0.9);
2654        let results = collector.into_results();
2655        assert_eq!(
2656            results
2657                .iter()
2658                .map(|result| result.doc_id)
2659                .collect::<Vec<_>>(),
2660            vec![2, 3]
2661        );
2662
2663        let mut tied = FlatDocumentCollector::new(1, crate::query::MultiValueCombiner::Max);
2664        tied.push(2, 0, 1.0);
2665        tied.push(1, 0, 1.0);
2666        let results = tied.into_results();
2667        assert_eq!(results[0].doc_id, 1);
2668    }
2669
2670    #[test]
2671    fn dense_fetch_count_rounds_up_and_detects_overflow() {
2672        assert_eq!(checked_dense_fetch_k(3, 1.5).unwrap(), 5);
2673        assert_eq!(checked_dense_fetch_k(5_000, 3.0).unwrap(), 15_000);
2674        assert!(checked_dense_fetch_k(50_000, 3.0).is_err());
2675        assert!(checked_dense_fetch_k(usize::MAX, 2.0).is_err());
2676    }
2677
2678    #[test]
2679    fn file_ranges_reject_overflow_and_truncation() {
2680        assert_eq!(checked_file_range(4, 3, 7, "test").unwrap(), 4..7);
2681        assert!(checked_file_range(u64::MAX, 1, u64::MAX, "test").is_err());
2682        assert!(checked_file_range(5, 3, 7, "test").is_err());
2683    }
2684
2685    /// Probe-returned (doc, ordinal) pairs keep their probe scores and are
2686    /// pruned from the flat-read list; only the candidate documents'
2687    /// remaining ordinals still need reads.
2688    #[test]
2689    fn probe_scored_candidates_skip_flat_rereads() {
2690        let ann_results: Vec<RawVectorCandidate> = vec![(7, 0, 0.75), (9, 2, 0.5)];
2691        // Expansion of docs 7 and 9 to all their ordinals (scores unfilled).
2692        let mut expanded: Vec<RawVectorCandidate> =
2693            vec![(7, 0, 0.0), (7, 1, 0.0), (9, 0, 0.0), (9, 2, 0.0)];
2694        let resolved: Vec<ResolvedVectorCandidate> = vec![(0, 10), (1, 11), (2, 20), (3, 22)];
2695
2696        let remaining = fill_probe_scores_and_prune_resolved(&ann_results, &mut expanded, resolved);
2697
2698        // Probe-scored pairs (7,0) and (9,2) are filled and pruned.
2699        assert_eq!(expanded[0].2, 0.75);
2700        assert_eq!(expanded[3].2, 0.5);
2701        // Unseen ordinals (7,1) and (9,0) still need flat reads.
2702        assert_eq!(remaining, vec![(1, 11), (2, 20)]);
2703        assert_eq!(expanded[1].2, 0.0);
2704        assert_eq!(expanded[2].2, 0.0);
2705    }
2706}