Skip to main content

hermes_core/query/
collector.rs

1//! Search result collection and response types
2
3use std::cmp::Ordering;
4use std::collections::BinaryHeap;
5
6use crate::segment::SegmentReader;
7use crate::structures::TERMINATED;
8use crate::{DocId, Result, Score};
9
10use super::Query;
11
12/// Unique document address: segment_id + local doc_id within segment.
13/// Stores segment_id as u128 internally (16 bytes) but serializes as hex string
14/// for backward compatibility with JSON/gRPC clients.
15#[derive(Debug, Clone, PartialEq, Eq, Hash)]
16pub struct DocAddress {
17    /// Segment ID as u128 (avoids heap allocation vs String)
18    segment_id_raw: u128,
19    /// Document ID within the segment
20    pub doc_id: DocId,
21}
22
23impl DocAddress {
24    pub fn new(segment_id: u128, doc_id: DocId) -> Self {
25        Self {
26            segment_id_raw: segment_id,
27            doc_id,
28        }
29    }
30
31    /// Get segment_id as hex string (for display/API)
32    pub fn segment_id(&self) -> String {
33        format!("{:032x}", self.segment_id_raw)
34    }
35
36    /// Get segment_id as u128 (zero-cost)
37    pub fn segment_id_u128(&self) -> Option<u128> {
38        Some(self.segment_id_raw)
39    }
40}
41
42impl serde::Serialize for DocAddress {
43    fn serialize<S: serde::Serializer>(
44        &self,
45        serializer: S,
46    ) -> std::result::Result<S::Ok, S::Error> {
47        use serde::ser::SerializeStruct;
48        let mut s = serializer.serialize_struct("DocAddress", 2)?;
49        s.serialize_field("segment_id", &format!("{:032x}", self.segment_id_raw))?;
50        s.serialize_field("doc_id", &self.doc_id)?;
51        s.end()
52    }
53}
54
55impl<'de> serde::Deserialize<'de> for DocAddress {
56    fn deserialize<D: serde::Deserializer<'de>>(
57        deserializer: D,
58    ) -> std::result::Result<Self, D::Error> {
59        #[derive(serde::Deserialize)]
60        struct Helper {
61            segment_id: String,
62            doc_id: DocId,
63        }
64        let h = Helper::deserialize(deserializer)?;
65        let raw = u128::from_str_radix(&h.segment_id, 16).map_err(serde::de::Error::custom)?;
66        Ok(DocAddress {
67            segment_id_raw: raw,
68            doc_id: h.doc_id,
69        })
70    }
71}
72
73/// A scored position/ordinal within a field
74/// For text fields: position is the token position
75/// For vector fields: position is the ordinal (which vector in multi-value)
76#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
77pub struct ScoredPosition {
78    /// Position (text) or ordinal (vector)
79    pub position: u32,
80    /// Individual score contribution from this position/ordinal
81    pub score: f32,
82}
83
84impl ScoredPosition {
85    pub fn new(position: u32, score: f32) -> Self {
86        Self { position, score }
87    }
88}
89
90/// Search result with doc_id and score (internal use)
91#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
92pub struct SearchResult {
93    pub doc_id: DocId,
94    pub score: Score,
95    /// Segment ID (set by searcher after collection)
96    #[serde(default, skip_serializing_if = "is_zero_u128")]
97    pub segment_id: u128,
98    /// Matched positions per field: (field_id, scored_positions)
99    /// Each position includes its individual score contribution
100    #[serde(default, skip_serializing_if = "Vec::is_empty")]
101    pub positions: Vec<(u32, Vec<ScoredPosition>)>,
102}
103
104fn is_zero_u128(v: &u128) -> bool {
105    *v == 0
106}
107
108/// Canonical result order used by search, reranking, fusion, and pagination.
109pub(crate) fn compare_search_results_desc(a: &SearchResult, b: &SearchResult) -> Ordering {
110    b.score
111        .total_cmp(&a.score)
112        .then_with(|| a.segment_id.cmp(&b.segment_id))
113        .then_with(|| a.doc_id.cmp(&b.doc_id))
114}
115
116/// Matched field info with ordinals (for multi-valued fields)
117#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
118pub struct MatchedField {
119    /// Field ID
120    pub field_id: u32,
121    /// Matched element ordinals (for multi-valued fields with position tracking)
122    /// Empty if position tracking is not enabled for this field
123    pub ordinals: Vec<u32>,
124}
125
126impl SearchResult {
127    /// Extract unique ordinals from positions for each field
128    /// For text fields: ordinal = position >> 20 (from encoded position)
129    /// For vector fields: position IS the ordinal directly
130    pub fn extract_ordinals(&self) -> Vec<MatchedField> {
131        self.positions
132            .iter()
133            .map(|(field_id, scored_positions)| {
134                // Position lists are typically short. Collecting into one
135                // compact buffer and deduplicating in place avoids both the
136                // hash-table allocation and the second allocation needed to
137                // turn that table back into a sorted response vector.
138                let mut ordinals = Vec::with_capacity(scored_positions.len());
139                ordinals.extend(scored_positions.iter().map(|sp| {
140                    // For text fields with encoded positions, extract ordinal.
141                    // For vector fields, position IS the ordinal.
142                    if sp.position > 0xFFFFF {
143                        sp.position >> 20
144                    } else {
145                        sp.position
146                    }
147                }));
148                ordinals.sort_unstable();
149                ordinals.dedup();
150                MatchedField {
151                    field_id: *field_id,
152                    ordinals,
153                }
154            })
155            .collect()
156    }
157
158    /// Get all scored positions for a specific field
159    pub fn field_positions(&self, field_id: u32) -> Option<&[ScoredPosition]> {
160        self.positions
161            .iter()
162            .find(|(fid, _)| *fid == field_id)
163            .map(|(_, positions)| positions.as_slice())
164    }
165}
166
167/// Search hit with unique document address and score
168#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
169pub struct SearchHit {
170    /// Unique document address (segment_id + local doc_id)
171    pub address: DocAddress,
172    pub score: Score,
173    /// Matched fields with element ordinals (populated when position tracking is enabled)
174    #[serde(default, skip_serializing_if = "Vec::is_empty")]
175    pub matched_fields: Vec<MatchedField>,
176}
177
178/// Search response with hits (IDs only, no documents)
179#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
180pub struct SearchResponse {
181    pub hits: Vec<SearchHit>,
182    pub total_hits: u32,
183}
184
185impl PartialEq for SearchResult {
186    fn eq(&self, other: &Self) -> bool {
187        self.score.to_bits() == other.score.to_bits()
188            && self.segment_id == other.segment_id
189            && self.doc_id == other.doc_id
190    }
191}
192
193impl Eq for SearchResult {}
194
195impl PartialOrd for SearchResult {
196    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
197        Some(self.cmp(other))
198    }
199}
200
201impl Ord for SearchResult {
202    fn cmp(&self, other: &Self) -> Ordering {
203        other
204            .score
205            .total_cmp(&self.score)
206            .then_with(|| self.segment_id.cmp(&other.segment_id))
207            .then_with(|| self.doc_id.cmp(&other.doc_id))
208    }
209}
210
211/// Trait for search result collectors
212///
213/// Implement this trait to create custom collectors that can be
214/// combined and passed to query execution.
215pub trait Collector {
216    /// Called for each matching document
217    /// positions: Vec of (field_id, scored_positions)
218    fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]);
219
220    /// Whether this score can enter the collector's retained result set.
221    ///
222    /// The scorer still calls `collect` when this returns false so counters and
223    /// other side effects remain exact; it only skips materializing positions.
224    fn would_collect(&self, _doc_id: DocId, _score: Score) -> bool {
225        true
226    }
227
228    /// Collect already-owned positions. Position-aware collectors can override
229    /// this to move the nested vectors instead of cloning them.
230    fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
231        self.collect(doc_id, score, &positions);
232    }
233
234    /// Whether this collector needs position information
235    fn needs_positions(&self) -> bool {
236        false
237    }
238}
239
240/// Compact score-only heap entry.
241///
242/// A segment-local collector does not know its segment ID yet and ordinary
243/// searches do not retain positions. Keeping only these two words while the
244/// scorer runs makes the common heap 8 bytes per hit instead of storing a
245/// full `SearchResult` (including an empty `Vec` and a zero `u128`).
246#[derive(Debug, Clone, Copy)]
247struct ScoreOnlyResult {
248    doc_id: DocId,
249    score: Score,
250}
251
252impl PartialEq for ScoreOnlyResult {
253    fn eq(&self, other: &Self) -> bool {
254        self.score.to_bits() == other.score.to_bits() && self.doc_id == other.doc_id
255    }
256}
257
258impl Eq for ScoreOnlyResult {}
259
260impl PartialOrd for ScoreOnlyResult {
261    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
262        Some(self.cmp(other))
263    }
264}
265
266impl Ord for ScoreOnlyResult {
267    fn cmp(&self, other: &Self) -> Ordering {
268        other
269            .score
270            .total_cmp(&self.score)
271            .then_with(|| self.doc_id.cmp(&other.doc_id))
272    }
273}
274
275/// Position-aware heap entry. The segment ID is stamped after collection, so
276/// omitting it here also keeps this variant smaller than `SearchResult`.
277#[derive(Debug, Clone)]
278struct PositionedResult {
279    doc_id: DocId,
280    score: Score,
281    positions: super::MatchedPositions,
282}
283
284impl PartialEq for PositionedResult {
285    fn eq(&self, other: &Self) -> bool {
286        self.score.to_bits() == other.score.to_bits() && self.doc_id == other.doc_id
287    }
288}
289
290impl Eq for PositionedResult {}
291
292impl PartialOrd for PositionedResult {
293    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
294        Some(self.cmp(other))
295    }
296}
297
298impl Ord for PositionedResult {
299    fn cmp(&self, other: &Self) -> Ordering {
300        other
301            .score
302            .total_cmp(&self.score)
303            .then_with(|| self.doc_id.cmp(&other.doc_id))
304    }
305}
306
307enum TopKHeap {
308    Scores(BinaryHeap<ScoreOnlyResult>),
309    Positions(BinaryHeap<PositionedResult>),
310}
311
312#[inline(always)]
313fn ranks_ahead(doc_id: DocId, score: Score, worst_doc_id: DocId, worst_score: Score) -> bool {
314    let order = score.total_cmp(&worst_score);
315    order.is_gt() || (order.is_eq() && doc_id < worst_doc_id)
316}
317
318/// Collector for top-k results
319pub struct TopKCollector {
320    heap: TopKHeap,
321    k: usize,
322    /// Total documents seen by this collector
323    total_seen: u32,
324}
325
326// Avoid trusting a caller-controlled `k` as an up-front allocation size. The
327// heap still grows to the number of results actually retained, but malformed
328// or overly broad requests cannot reserve gigabytes once per segment before
329// any document has been scored.
330const MAX_INITIAL_TOP_K_CAPACITY: usize = 8 * 1024;
331
332impl TopKCollector {
333    pub fn new(k: usize) -> Self {
334        Self {
335            heap: TopKHeap::Scores(BinaryHeap::with_capacity(k.min(MAX_INITIAL_TOP_K_CAPACITY))),
336            k,
337            total_seen: 0,
338        }
339    }
340
341    /// Create a collector that also collects positions
342    pub fn with_positions(k: usize) -> Self {
343        Self {
344            heap: TopKHeap::Positions(BinaryHeap::with_capacity(k.min(MAX_INITIAL_TOP_K_CAPACITY))),
345            k,
346            total_seen: 0,
347        }
348    }
349
350    /// Get the total number of documents seen (scored) by this collector
351    pub fn total_seen(&self) -> u32 {
352        self.total_seen
353    }
354
355    pub fn into_sorted_results(self) -> Vec<SearchResult> {
356        match self.heap {
357            TopKHeap::Scores(heap) => {
358                let mut compact = heap.into_vec();
359                compact.sort_unstable_by(|a, b| {
360                    b.score
361                        .total_cmp(&a.score)
362                        .then_with(|| a.doc_id.cmp(&b.doc_id))
363                });
364                compact
365                    .into_iter()
366                    .map(|result| SearchResult {
367                        doc_id: result.doc_id,
368                        score: result.score,
369                        segment_id: 0,
370                        positions: Vec::new(),
371                    })
372                    .collect()
373            }
374            TopKHeap::Positions(heap) => {
375                let mut positioned = heap.into_vec();
376                positioned.sort_unstable_by(|a, b| {
377                    b.score
378                        .total_cmp(&a.score)
379                        .then_with(|| a.doc_id.cmp(&b.doc_id))
380                });
381                positioned
382                    .into_iter()
383                    .map(|result| SearchResult {
384                        doc_id: result.doc_id,
385                        score: result.score,
386                        segment_id: 0,
387                        positions: result.positions,
388                    })
389                    .collect()
390            }
391        }
392    }
393
394    /// Consume collector and return (sorted_results, total_seen)
395    pub fn into_results_with_count(self) -> (Vec<SearchResult>, u32) {
396        let total = self.total_seen;
397        (self.into_sorted_results(), total)
398    }
399}
400
401impl Collector for TopKCollector {
402    #[inline]
403    fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]) {
404        self.total_seen = self.total_seen.saturating_add(1);
405        if self.k == 0 {
406            return;
407        }
408
409        match &mut self.heap {
410            TopKHeap::Scores(heap) => {
411                let result = ScoreOnlyResult { doc_id, score };
412                if heap.len() < self.k {
413                    heap.push(result);
414                } else if heap
415                    .peek()
416                    .is_some_and(|worst| ranks_ahead(doc_id, score, worst.doc_id, worst.score))
417                {
418                    *heap.peek_mut().expect("full top-k heap") = result;
419                }
420            }
421            TopKHeap::Positions(heap) => {
422                if heap.len() >= self.k
423                    && !heap
424                        .peek()
425                        .is_some_and(|worst| ranks_ahead(doc_id, score, worst.doc_id, worst.score))
426                {
427                    return;
428                }
429                let result = PositionedResult {
430                    doc_id,
431                    score,
432                    // Only clone positions after the hit is known to be
433                    // competitive. Replacing the root drops its old positions.
434                    positions: positions.to_vec(),
435                };
436                if heap.len() < self.k {
437                    heap.push(result);
438                } else {
439                    *heap.peek_mut().expect("full top-k heap") = result;
440                }
441            }
442        }
443    }
444
445    #[inline]
446    fn would_collect(&self, doc_id: DocId, score: Score) -> bool {
447        if self.k == 0 {
448            return false;
449        }
450        match &self.heap {
451            TopKHeap::Scores(heap) => {
452                heap.len() < self.k
453                    || heap
454                        .peek()
455                        .is_some_and(|min| ranks_ahead(doc_id, score, min.doc_id, min.score))
456            }
457            TopKHeap::Positions(heap) => {
458                heap.len() < self.k
459                    || heap
460                        .peek()
461                        .is_some_and(|min| ranks_ahead(doc_id, score, min.doc_id, min.score))
462            }
463        }
464    }
465
466    #[inline]
467    fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
468        self.total_seen = self.total_seen.saturating_add(1);
469        if self.k == 0 {
470            return;
471        }
472
473        match &mut self.heap {
474            TopKHeap::Scores(heap) => {
475                let result = ScoreOnlyResult { doc_id, score };
476                if heap.len() < self.k {
477                    heap.push(result);
478                } else if heap
479                    .peek()
480                    .is_some_and(|worst| ranks_ahead(doc_id, score, worst.doc_id, worst.score))
481                {
482                    *heap.peek_mut().expect("full top-k heap") = result;
483                }
484            }
485            TopKHeap::Positions(heap) => {
486                if heap.len() >= self.k
487                    && !heap
488                        .peek()
489                        .is_some_and(|worst| ranks_ahead(doc_id, score, worst.doc_id, worst.score))
490                {
491                    return;
492                }
493                let result = PositionedResult {
494                    doc_id,
495                    score,
496                    positions,
497                };
498                if heap.len() < self.k {
499                    heap.push(result);
500                } else {
501                    *heap.peek_mut().expect("full top-k heap") = result;
502                }
503            }
504        }
505    }
506
507    #[inline]
508    fn needs_positions(&self) -> bool {
509        matches!(&self.heap, TopKHeap::Positions(_))
510    }
511}
512
513/// Collector that counts all matching documents
514#[derive(Default)]
515pub struct CountCollector {
516    count: u64,
517}
518
519impl CountCollector {
520    pub fn new() -> Self {
521        Self { count: 0 }
522    }
523
524    /// Get the total count
525    pub fn count(&self) -> u64 {
526        self.count
527    }
528}
529
530impl Collector for CountCollector {
531    #[inline]
532    fn collect(
533        &mut self,
534        _doc_id: DocId,
535        _score: Score,
536        _positions: &[(u32, Vec<ScoredPosition>)],
537    ) {
538        self.count += 1;
539    }
540}
541
542/// Execute a search query on a single segment and return (results, total_seen) (async)
543pub async fn search_segment_with_count(
544    reader: &SegmentReader,
545    query: &dyn Query,
546    limit: usize,
547) -> Result<(Vec<SearchResult>, u32)> {
548    let segment_limit = limit.min(reader.num_docs() as usize);
549    let mut collector = TopKCollector::new(segment_limit);
550    collect_segment_with_limit(reader, query, &mut collector, segment_limit).await?;
551    Ok(collector.into_results_with_count())
552}
553
554/// Execute a search query on a single segment with positions and return (results, total_seen)
555pub async fn search_segment_with_positions_and_count(
556    reader: &SegmentReader,
557    query: &dyn Query,
558    limit: usize,
559) -> Result<(Vec<SearchResult>, u32)> {
560    let segment_limit = limit.min(reader.num_docs() as usize);
561    let mut collector = TopKCollector::with_positions(segment_limit);
562    collect_segment_with_limit(reader, query, &mut collector, segment_limit).await?;
563    Ok(collector.into_results_with_count())
564}
565
566/// Return positions for the next collector that can retain them. All but the
567/// final consumer receive a clone; the final consumer takes the original
568/// allocation. Tuple collectors use this to avoid a deep clone when only one
569/// child actually needs positions (the common top-k + count case).
570fn positions_for_next_collector(
571    positions: &mut Option<super::MatchedPositions>,
572    remaining_consumers: &mut usize,
573) -> super::MatchedPositions {
574    assert!(
575        *remaining_consumers > 0,
576        "position consumer count underflow"
577    );
578    *remaining_consumers -= 1;
579    if *remaining_consumers == 0 {
580        positions
581            .take()
582            .expect("owned positions must remain for the final collector")
583    } else {
584        positions
585            .as_ref()
586            .cloned()
587            .expect("owned positions must remain while collectors are pending")
588    }
589}
590
591// Implement Collector for tuple of 2 collectors
592impl<A: Collector, B: Collector> Collector for (&mut A, &mut B) {
593    fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]) {
594        self.0.collect(doc_id, score, positions);
595        self.1.collect(doc_id, score, positions);
596    }
597    fn needs_positions(&self) -> bool {
598        self.0.needs_positions() || self.1.needs_positions()
599    }
600    fn would_collect(&self, doc_id: DocId, score: Score) -> bool {
601        (self.0.needs_positions() && self.0.would_collect(doc_id, score))
602            || (self.1.needs_positions() && self.1.would_collect(doc_id, score))
603    }
604    fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
605        let wants = [
606            self.0.needs_positions() && self.0.would_collect(doc_id, score),
607            self.1.needs_positions() && self.1.would_collect(doc_id, score),
608        ];
609        let mut remaining = wants.iter().filter(|&&want| want).count();
610        let mut positions = Some(positions);
611
612        if wants[0] {
613            self.0.collect_owned(
614                doc_id,
615                score,
616                positions_for_next_collector(&mut positions, &mut remaining),
617            );
618        } else {
619            self.0.collect(doc_id, score, &[]);
620        }
621        if wants[1] {
622            self.1.collect_owned(
623                doc_id,
624                score,
625                positions_for_next_collector(&mut positions, &mut remaining),
626            );
627        } else {
628            self.1.collect(doc_id, score, &[]);
629        }
630    }
631}
632
633// Implement Collector for tuple of 3 collectors
634impl<A: Collector, B: Collector, C: Collector> Collector for (&mut A, &mut B, &mut C) {
635    fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]) {
636        self.0.collect(doc_id, score, positions);
637        self.1.collect(doc_id, score, positions);
638        self.2.collect(doc_id, score, positions);
639    }
640    fn needs_positions(&self) -> bool {
641        self.0.needs_positions() || self.1.needs_positions() || self.2.needs_positions()
642    }
643    fn would_collect(&self, doc_id: DocId, score: Score) -> bool {
644        (self.0.needs_positions() && self.0.would_collect(doc_id, score))
645            || (self.1.needs_positions() && self.1.would_collect(doc_id, score))
646            || (self.2.needs_positions() && self.2.would_collect(doc_id, score))
647    }
648    fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
649        let wants = [
650            self.0.needs_positions() && self.0.would_collect(doc_id, score),
651            self.1.needs_positions() && self.1.would_collect(doc_id, score),
652            self.2.needs_positions() && self.2.would_collect(doc_id, score),
653        ];
654        let mut remaining = wants.iter().filter(|&&want| want).count();
655        let mut positions = Some(positions);
656
657        if wants[0] {
658            self.0.collect_owned(
659                doc_id,
660                score,
661                positions_for_next_collector(&mut positions, &mut remaining),
662            );
663        } else {
664            self.0.collect(doc_id, score, &[]);
665        }
666        if wants[1] {
667            self.1.collect_owned(
668                doc_id,
669                score,
670                positions_for_next_collector(&mut positions, &mut remaining),
671            );
672        } else {
673            self.1.collect(doc_id, score, &[]);
674        }
675        if wants[2] {
676            self.2.collect_owned(
677                doc_id,
678                score,
679                positions_for_next_collector(&mut positions, &mut remaining),
680            );
681        } else {
682            self.2.collect(doc_id, score, &[]);
683        }
684    }
685}
686
687/// Execute a query with one or more collectors (async)
688///
689/// Uses a large limit for the scorer to disable MaxScore pruning.
690/// For queries that benefit from MaxScore pruning (e.g., sparse vector search),
691/// use `collect_segment_with_limit` instead.
692///
693/// # Examples
694/// ```ignore
695/// // Single collector
696/// let mut top_k = TopKCollector::new(10);
697/// collect_segment(reader, query, &mut top_k).await?;
698///
699/// // Multiple collectors (tuple)
700/// let mut top_k = TopKCollector::new(10);
701/// let mut count = CountCollector::new();
702/// collect_segment(reader, query, &mut (&mut top_k, &mut count)).await?;
703/// ```
704pub async fn collect_segment<C: Collector>(
705    reader: &SegmentReader,
706    query: &dyn Query,
707    collector: &mut C,
708) -> Result<()> {
709    // Use large limit to disable MaxScore skipping for exhaustive collection
710    collect_segment_with_limit(reader, query, collector, usize::MAX / 2).await
711}
712
713/// Execute a query with one or more collectors and a specific limit (async)
714///
715/// The limit is passed to the scorer to enable MaxScore pruning for queries
716/// that support it (e.g., sparse vector search). This significantly improves
717/// performance when only the top-k results are needed.
718///
719/// Doc IDs in the collector are segment-local. The searcher stamps each result
720/// with its segment_id, making (segment_id, doc_id) the unique document key.
721pub async fn collect_segment_with_limit<C: Collector>(
722    reader: &SegmentReader,
723    query: &dyn Query,
724    collector: &mut C,
725    limit: usize,
726) -> Result<()> {
727    collect_segment_with_limit_seeded(reader, query, collector, limit, 0.0).await
728}
729
730/// Async `collect_segment_with_limit` with a cross-segment threshold seed.
731///
732/// `initial_threshold` is passed to the scorer so exact MaxScore/BMP paths can
733/// start pruning from a nonzero floor carried over from earlier segments.
734pub async fn collect_segment_with_limit_seeded<C: Collector>(
735    reader: &SegmentReader,
736    query: &dyn Query,
737    collector: &mut C,
738    limit: usize,
739    initial_threshold: f32,
740) -> Result<()> {
741    let options = super::ScorerOptions {
742        collect_positions: collector.needs_positions(),
743        initial_threshold,
744        shared_threshold: None,
745        lsp_plan: None,
746    };
747    let mut scorer = query.scorer_with_options(reader, limit, options).await?;
748    drive_scorer(scorer.as_mut(), collector);
749    Ok(())
750}
751
752/// Drive a scorer through a collector (shared by async and sync paths).
753fn drive_scorer<C: Collector>(scorer: &mut dyn super::Scorer, collector: &mut C) {
754    let needs_positions = collector.needs_positions();
755    let mut doc = scorer.doc();
756    while doc != TERMINATED {
757        let score = scorer.score();
758        if needs_positions && collector.would_collect(doc, score) {
759            let positions = scorer.matched_positions().unwrap_or_default();
760            collector.collect_owned(doc, score, positions);
761        } else {
762            collector.collect(doc, score, &[]);
763        }
764        doc = scorer.advance();
765    }
766}
767
768// ── Synchronous collector functions (mmap/RAM only) ─────────────────────────
769
770/// Synchronous segment search — returns (results, total_seen).
771#[cfg(feature = "sync")]
772pub fn search_segment_with_count_sync(
773    reader: &SegmentReader,
774    query: &dyn Query,
775    limit: usize,
776) -> Result<(Vec<SearchResult>, u32)> {
777    let segment_limit = limit.min(reader.num_docs() as usize);
778    let mut collector = TopKCollector::new(segment_limit);
779    collect_segment_with_limit_sync(reader, query, &mut collector, segment_limit)?;
780    Ok(collector.into_results_with_count())
781}
782
783/// Synchronous segment search with positions — returns (results, total_seen).
784#[cfg(feature = "sync")]
785pub fn search_segment_with_positions_and_count_sync(
786    reader: &SegmentReader,
787    query: &dyn Query,
788    limit: usize,
789) -> Result<(Vec<SearchResult>, u32)> {
790    let segment_limit = limit.min(reader.num_docs() as usize);
791    let mut collector = TopKCollector::with_positions(segment_limit);
792    collect_segment_with_limit_sync(reader, query, &mut collector, segment_limit)?;
793    Ok(collector.into_results_with_count())
794}
795
796/// Synchronous collect with limit — uses `scorer_sync`.
797#[cfg(feature = "sync")]
798pub fn collect_segment_with_limit_sync<C: Collector>(
799    reader: &SegmentReader,
800    query: &dyn Query,
801    collector: &mut C,
802    limit: usize,
803) -> Result<()> {
804    collect_segment_with_limit_seeded_sync(reader, query, collector, limit, 0.0)
805}
806
807/// Synchronous `collect_segment_with_limit_sync` with a cross-segment threshold
808/// seed (see `collect_segment_with_limit_seeded`).
809#[cfg(feature = "sync")]
810pub fn collect_segment_with_limit_seeded_sync<C: Collector>(
811    reader: &SegmentReader,
812    query: &dyn Query,
813    collector: &mut C,
814    limit: usize,
815    initial_threshold: f32,
816) -> Result<()> {
817    let options = super::ScorerOptions {
818        collect_positions: collector.needs_positions(),
819        initial_threshold,
820        shared_threshold: None,
821        lsp_plan: None,
822    };
823    let mut scorer = query.scorer_sync_with_options(reader, limit, options)?;
824    drive_scorer(scorer.as_mut(), collector);
825    Ok(())
826}
827
828/// Per-segment search seeded with a cross-segment top-k floor (sync).
829///
830/// Behaves like `search_segment_with_count_sync` / its positions variant, but
831/// threads `initial_threshold` into the scorer so exact MaxScore/BMP paths
832/// prune from the running global k-th score. Used by the multi-segment
833/// searcher to propagate the threshold across segments.
834#[cfg(feature = "sync")]
835pub fn search_segment_seeded_sync(
836    reader: &SegmentReader,
837    query: &dyn Query,
838    limit: usize,
839    collect_positions: bool,
840    initial_threshold: f32,
841) -> Result<(Vec<SearchResult>, u32)> {
842    let segment_limit = limit.min(reader.num_docs() as usize);
843    let mut collector = if collect_positions {
844        TopKCollector::with_positions(segment_limit)
845    } else {
846        TopKCollector::new(segment_limit)
847    };
848    collect_segment_with_limit_seeded_sync(
849        reader,
850        query,
851        &mut collector,
852        segment_limit,
853        initial_threshold,
854    )?;
855    Ok(collector.into_results_with_count())
856}
857
858/// Per-segment search with a live cross-segment top-k floor (sync).
859#[cfg(feature = "sync")]
860pub fn search_segment_shared_sync(
861    reader: &SegmentReader,
862    query: &dyn Query,
863    limit: usize,
864    collect_positions: bool,
865    shared_threshold: super::SharedThreshold,
866) -> Result<(Vec<SearchResult>, u32)> {
867    search_segment_shared_sync_planned(
868        reader,
869        query,
870        limit,
871        collect_positions,
872        shared_threshold,
873        None,
874    )
875}
876
877/// Per-segment search with a live threshold and a query-global LSP/0 plan.
878#[cfg(feature = "sync")]
879pub(crate) fn search_segment_shared_sync_planned(
880    reader: &SegmentReader,
881    query: &dyn Query,
882    limit: usize,
883    collect_positions: bool,
884    shared_threshold: super::SharedThreshold,
885    lsp_plan: Option<std::sync::Arc<super::bmp::LspSegmentPlan>>,
886) -> Result<(Vec<SearchResult>, u32)> {
887    let segment_limit = limit.min(reader.num_docs() as usize);
888    let mut collector = if collect_positions {
889        TopKCollector::with_positions(segment_limit)
890    } else {
891        TopKCollector::new(segment_limit)
892    };
893    let options = super::ScorerOptions {
894        collect_positions,
895        initial_threshold: shared_threshold.get(),
896        shared_threshold: Some(shared_threshold),
897        lsp_plan,
898    };
899    let mut scorer = query.scorer_sync_with_options(reader, segment_limit, options)?;
900    drive_scorer(scorer.as_mut(), &mut collector);
901    Ok(collector.into_results_with_count())
902}
903
904/// Per-segment search seeded with a cross-segment top-k floor (async).
905pub async fn search_segment_seeded(
906    reader: &SegmentReader,
907    query: &dyn Query,
908    limit: usize,
909    collect_positions: bool,
910    initial_threshold: f32,
911) -> Result<(Vec<SearchResult>, u32)> {
912    let segment_limit = limit.min(reader.num_docs() as usize);
913    let mut collector = if collect_positions {
914        TopKCollector::with_positions(segment_limit)
915    } else {
916        TopKCollector::new(segment_limit)
917    };
918    collect_segment_with_limit_seeded(
919        reader,
920        query,
921        &mut collector,
922        segment_limit,
923        initial_threshold,
924    )
925    .await?;
926    Ok(collector.into_results_with_count())
927}
928
929/// Per-segment search with a live cross-segment top-k floor (async).
930pub async fn search_segment_shared(
931    reader: &SegmentReader,
932    query: &dyn Query,
933    limit: usize,
934    collect_positions: bool,
935    shared_threshold: super::SharedThreshold,
936) -> Result<(Vec<SearchResult>, u32)> {
937    search_segment_shared_planned(
938        reader,
939        query,
940        limit,
941        collect_positions,
942        shared_threshold,
943        None,
944    )
945    .await
946}
947
948/// Async per-segment search with a query-global LSP/0 plan.
949pub(crate) async fn search_segment_shared_planned(
950    reader: &SegmentReader,
951    query: &dyn Query,
952    limit: usize,
953    collect_positions: bool,
954    shared_threshold: super::SharedThreshold,
955    lsp_plan: Option<std::sync::Arc<super::bmp::LspSegmentPlan>>,
956) -> Result<(Vec<SearchResult>, u32)> {
957    let segment_limit = limit.min(reader.num_docs() as usize);
958    let mut collector = if collect_positions {
959        TopKCollector::with_positions(segment_limit)
960    } else {
961        TopKCollector::new(segment_limit)
962    };
963    let options = super::ScorerOptions {
964        collect_positions,
965        initial_threshold: shared_threshold.get(),
966        shared_threshold: Some(shared_threshold),
967        lsp_plan,
968    };
969    let mut scorer = query
970        .scorer_with_options(reader, segment_limit, options)
971        .await?;
972    drive_scorer(scorer.as_mut(), &mut collector);
973    Ok(collector.into_results_with_count())
974}
975
976#[cfg(test)]
977mod tests {
978    use super::*;
979    use std::sync::Arc;
980    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
981
982    #[derive(Default)]
983    struct OwnedPositionCollector {
984        owned_calls: usize,
985        borrowed_calls: usize,
986        positions: super::super::MatchedPositions,
987    }
988
989    impl Collector for OwnedPositionCollector {
990        fn collect(
991            &mut self,
992            _doc_id: DocId,
993            _score: Score,
994            positions: &[(u32, Vec<ScoredPosition>)],
995        ) {
996            self.borrowed_calls += 1;
997            self.positions = positions.to_vec();
998        }
999
1000        fn collect_owned(
1001            &mut self,
1002            _doc_id: DocId,
1003            _score: Score,
1004            positions: super::super::MatchedPositions,
1005        ) {
1006            self.owned_calls += 1;
1007            self.positions = positions;
1008        }
1009
1010        fn needs_positions(&self) -> bool {
1011            true
1012        }
1013    }
1014
1015    struct PositionCountingScorer {
1016        index: usize,
1017        position_calls: Arc<AtomicUsize>,
1018    }
1019
1020    impl super::super::DocSet for PositionCountingScorer {
1021        fn doc(&self) -> DocId {
1022            if self.index < 3 {
1023                self.index as DocId
1024            } else {
1025                TERMINATED
1026            }
1027        }
1028
1029        fn advance(&mut self) -> DocId {
1030            self.index += 1;
1031            self.doc()
1032        }
1033
1034        fn seek(&mut self, target: DocId) -> DocId {
1035            self.index = target.min(3) as usize;
1036            self.doc()
1037        }
1038
1039        fn size_hint(&self) -> u32 {
1040            3u32.saturating_sub(self.index as u32)
1041        }
1042    }
1043
1044    impl super::super::Scorer for PositionCountingScorer {
1045        fn score(&self) -> Score {
1046            [10.0, 1.0, 2.0][self.index]
1047        }
1048
1049        fn matched_positions(&self) -> Option<super::super::MatchedPositions> {
1050            self.position_calls.fetch_add(1, AtomicOrdering::Relaxed);
1051            Some(vec![(7, vec![ScoredPosition::new(self.index as u32, 1.0)])])
1052        }
1053    }
1054
1055    #[test]
1056    fn test_top_k_collector() {
1057        let mut collector = TopKCollector::new(3);
1058
1059        collector.collect(0, 1.0, &[]);
1060        collector.collect(1, 3.0, &[]);
1061        collector.collect(2, 2.0, &[]);
1062        collector.collect(3, 4.0, &[]);
1063        collector.collect(4, 0.5, &[]);
1064
1065        let results = collector.into_sorted_results();
1066
1067        assert_eq!(results.len(), 3);
1068        assert_eq!(results[0].doc_id, 3); // score 4.0
1069        assert_eq!(results[1].doc_id, 1); // score 3.0
1070        assert_eq!(results[2].doc_id, 2); // score 2.0
1071    }
1072
1073    #[test]
1074    fn top_k_zero_retains_no_results() {
1075        let mut collector = TopKCollector::new(0);
1076        collector.collect(1, 1.0, &[]);
1077
1078        assert!(collector.into_sorted_results().is_empty());
1079    }
1080
1081    #[test]
1082    fn huge_top_k_does_not_trigger_a_huge_initial_allocation() {
1083        let collector = TopKCollector::new(usize::MAX);
1084
1085        let TopKHeap::Scores(heap) = collector.heap else {
1086            panic!("score-only constructor selected the position heap");
1087        };
1088        assert!(heap.capacity() <= MAX_INITIAL_TOP_K_CAPACITY);
1089    }
1090
1091    #[test]
1092    fn score_only_heap_entry_stays_compact() {
1093        assert_eq!(std::mem::size_of::<ScoreOnlyResult>(), 8);
1094        assert!(std::mem::size_of::<SearchResult>() >= 4 * std::mem::size_of::<ScoreOnlyResult>());
1095    }
1096
1097    #[test]
1098    fn top_k_replacement_preserves_score_and_doc_ties() {
1099        let mut collector = TopKCollector::new(3);
1100        for (doc_id, score) in [(9, 2.0), (8, 2.0), (7, 2.0), (6, 2.0), (1, 1.0)] {
1101            collector.collect(doc_id, score, &[]);
1102        }
1103
1104        let results = collector.into_sorted_results();
1105        assert_eq!(
1106            results
1107                .iter()
1108                .map(|result| (result.doc_id, result.score))
1109                .collect::<Vec<_>>(),
1110            vec![(6, 2.0), (7, 2.0), (8, 2.0)]
1111        );
1112    }
1113
1114    #[test]
1115    fn extract_ordinals_sorts_and_deduplicates_without_hashing() {
1116        let result = SearchResult {
1117            doc_id: 1,
1118            score: 1.0,
1119            segment_id: 0,
1120            positions: vec![
1121                (
1122                    3,
1123                    vec![
1124                        ScoredPosition::new(5 << 20, 1.0),
1125                        ScoredPosition::new(2 << 20, 1.0),
1126                        ScoredPosition::new(5 << 20, 2.0),
1127                    ],
1128                ),
1129                (
1130                    7,
1131                    vec![
1132                        ScoredPosition::new(4, 1.0),
1133                        ScoredPosition::new(1, 1.0),
1134                        ScoredPosition::new(4, 2.0),
1135                    ],
1136                ),
1137            ],
1138        };
1139
1140        let fields = result.extract_ordinals();
1141        assert_eq!(fields[0].ordinals, vec![2, 5]);
1142        assert_eq!(fields[1].ordinals, vec![1, 4]);
1143    }
1144
1145    #[test]
1146    fn positions_are_only_materialized_for_competitive_hits() {
1147        let calls = Arc::new(AtomicUsize::new(0));
1148        let mut scorer = PositionCountingScorer {
1149            index: 0,
1150            position_calls: Arc::clone(&calls),
1151        };
1152        let mut collector = TopKCollector::with_positions(1);
1153
1154        drive_scorer(&mut scorer, &mut collector);
1155
1156        assert_eq!(calls.load(AtomicOrdering::Relaxed), 1);
1157        assert_eq!(collector.total_seen(), 3);
1158        let results = collector.into_sorted_results();
1159        assert_eq!(results.len(), 1);
1160        assert_eq!(results[0].doc_id, 0);
1161        assert_eq!(results[0].positions[0].0, 7);
1162    }
1163
1164    #[test]
1165    fn tuple_moves_owned_positions_to_single_position_collector() {
1166        let mut positions = OwnedPositionCollector::default();
1167        let mut count = CountCollector::new();
1168        let input = vec![(7, vec![ScoredPosition::new(3, 1.0)])];
1169        let input_ptr = input[0].1.as_ptr();
1170
1171        (&mut positions, &mut count).collect_owned(11, 2.0, input);
1172
1173        assert_eq!(positions.owned_calls, 1);
1174        assert_eq!(positions.borrowed_calls, 0);
1175        assert_eq!(positions.positions[0].1.as_ptr(), input_ptr);
1176        assert_eq!(count.count(), 1);
1177    }
1178
1179    #[test]
1180    fn tuple_clones_for_all_but_final_position_collector() {
1181        let mut first = OwnedPositionCollector::default();
1182        let mut second = OwnedPositionCollector::default();
1183        let mut count = CountCollector::new();
1184        let input = vec![(7, vec![ScoredPosition::new(3, 1.0)])];
1185        let input_ptr = input[0].1.as_ptr();
1186
1187        (&mut first, &mut count, &mut second).collect_owned(11, 2.0, input);
1188
1189        assert_eq!((first.owned_calls, first.borrowed_calls), (1, 0));
1190        assert_eq!((second.owned_calls, second.borrowed_calls), (1, 0));
1191        assert_ne!(first.positions[0].1.as_ptr(), input_ptr);
1192        assert_eq!(second.positions[0].1.as_ptr(), input_ptr);
1193        assert_eq!(count.count(), 1);
1194    }
1195
1196    #[test]
1197    fn test_count_collector() {
1198        let mut collector = CountCollector::new();
1199
1200        collector.collect(0, 1.0, &[]);
1201        collector.collect(1, 2.0, &[]);
1202        collector.collect(2, 3.0, &[]);
1203
1204        assert_eq!(collector.count(), 3);
1205    }
1206
1207    #[test]
1208    fn test_multi_collector() {
1209        let mut top_k = TopKCollector::new(2);
1210        let mut count = CountCollector::new();
1211
1212        // Simulate what collect_segment_multi does
1213        for (doc_id, score) in [(0, 1.0), (1, 3.0), (2, 2.0), (3, 4.0), (4, 0.5)] {
1214            top_k.collect(doc_id, score, &[]);
1215            count.collect(doc_id, score, &[]);
1216        }
1217
1218        // Count should have all 5 documents
1219        assert_eq!(count.count(), 5);
1220
1221        // TopK should only have top 2 results
1222        let results = top_k.into_sorted_results();
1223        assert_eq!(results.len(), 2);
1224        assert_eq!(results[0].doc_id, 3); // score 4.0
1225        assert_eq!(results[1].doc_id, 1); // score 3.0
1226    }
1227}