1use 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#[derive(Debug, Clone, PartialEq, Eq, Hash)]
16pub struct DocAddress {
17 segment_id_raw: u128,
19 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 pub fn segment_id(&self) -> String {
33 format!("{:032x}", self.segment_id_raw)
34 }
35
36 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#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
77pub struct ScoredPosition {
78 pub position: u32,
80 pub score: f32,
82}
83
84impl ScoredPosition {
85 pub fn new(position: u32, score: f32) -> Self {
86 Self { position, score }
87 }
88}
89
90#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
92pub struct SearchResult {
93 pub doc_id: DocId,
94 pub score: Score,
95 #[serde(default, skip_serializing_if = "is_zero_u128")]
97 pub segment_id: u128,
98 #[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
108pub(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#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
118pub struct MatchedField {
119 pub field_id: u32,
121 pub ordinals: Vec<u32>,
124}
125
126impl SearchResult {
127 pub fn extract_ordinals(&self) -> Vec<MatchedField> {
131 use rustc_hash::FxHashSet;
132
133 self.positions
134 .iter()
135 .map(|(field_id, scored_positions)| {
136 let mut ordinals: FxHashSet<u32> = FxHashSet::default();
137 for sp in scored_positions {
138 let ordinal = if sp.position > 0xFFFFF {
142 sp.position >> 20
143 } else {
144 sp.position
145 };
146 ordinals.insert(ordinal);
147 }
148 let mut ordinals: Vec<u32> = ordinals.into_iter().collect();
149 ordinals.sort_unstable();
150 MatchedField {
151 field_id: *field_id,
152 ordinals,
153 }
154 })
155 .collect()
156 }
157
158 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#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
169pub struct SearchHit {
170 pub address: DocAddress,
172 pub score: Score,
173 #[serde(default, skip_serializing_if = "Vec::is_empty")]
175 pub matched_fields: Vec<MatchedField>,
176}
177
178#[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
211pub trait Collector {
216 fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]);
219
220 fn would_collect(&self, _doc_id: DocId, _score: Score) -> bool {
225 true
226 }
227
228 fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
231 self.collect(doc_id, score, &positions);
232 }
233
234 fn needs_positions(&self) -> bool {
236 false
237 }
238}
239
240pub struct TopKCollector {
242 heap: BinaryHeap<SearchResult>,
243 k: usize,
244 collect_positions: bool,
245 total_seen: u32,
247}
248
249const MAX_INITIAL_TOP_K_CAPACITY: usize = 8 * 1024;
254
255impl TopKCollector {
256 pub fn new(k: usize) -> Self {
257 Self {
258 heap: BinaryHeap::with_capacity(k.min(MAX_INITIAL_TOP_K_CAPACITY)),
259 k,
260 collect_positions: false,
261 total_seen: 0,
262 }
263 }
264
265 pub fn with_positions(k: usize) -> Self {
267 Self {
268 heap: BinaryHeap::with_capacity(k.min(MAX_INITIAL_TOP_K_CAPACITY)),
269 k,
270 collect_positions: true,
271 total_seen: 0,
272 }
273 }
274
275 pub fn total_seen(&self) -> u32 {
277 self.total_seen
278 }
279
280 pub fn into_sorted_results(self) -> Vec<SearchResult> {
281 let mut results: Vec<_> = self.heap.into_vec();
282 results.sort_unstable_by(compare_search_results_desc);
283 results
284 }
285
286 pub fn into_results_with_count(self) -> (Vec<SearchResult>, u32) {
288 let total = self.total_seen;
289 (self.into_sorted_results(), total)
290 }
291}
292
293impl Collector for TopKCollector {
294 fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]) {
295 self.total_seen = self.total_seen.saturating_add(1);
296
297 if !self.would_collect(doc_id, score) {
301 return;
302 }
303
304 let positions = if self.collect_positions {
305 positions.to_vec()
306 } else {
307 Vec::new()
308 };
309
310 if self.heap.len() >= self.k {
311 self.heap.pop();
312 }
313 self.heap.push(SearchResult {
314 doc_id,
315 score,
316 segment_id: 0,
317 positions,
318 });
319 }
320
321 fn would_collect(&self, doc_id: DocId, score: Score) -> bool {
322 self.k > 0
323 && (self.heap.len() < self.k
324 || self.heap.peek().is_some_and(|min| {
325 score.total_cmp(&min.score).is_gt()
326 || (score.total_cmp(&min.score).is_eq() && doc_id < min.doc_id)
327 }))
328 }
329
330 fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
331 self.total_seen = self.total_seen.saturating_add(1);
332
333 if !self.would_collect(doc_id, score) {
334 return;
335 }
336
337 if self.heap.len() >= self.k {
338 self.heap.pop();
339 }
340 self.heap.push(SearchResult {
341 doc_id,
342 score,
343 segment_id: 0,
344 positions: if self.collect_positions {
345 positions
346 } else {
347 Vec::new()
348 },
349 });
350 }
351
352 fn needs_positions(&self) -> bool {
353 self.collect_positions
354 }
355}
356
357#[derive(Default)]
359pub struct CountCollector {
360 count: u64,
361}
362
363impl CountCollector {
364 pub fn new() -> Self {
365 Self { count: 0 }
366 }
367
368 pub fn count(&self) -> u64 {
370 self.count
371 }
372}
373
374impl Collector for CountCollector {
375 #[inline]
376 fn collect(
377 &mut self,
378 _doc_id: DocId,
379 _score: Score,
380 _positions: &[(u32, Vec<ScoredPosition>)],
381 ) {
382 self.count += 1;
383 }
384}
385
386pub async fn search_segment_with_count(
388 reader: &SegmentReader,
389 query: &dyn Query,
390 limit: usize,
391) -> Result<(Vec<SearchResult>, u32)> {
392 let segment_limit = limit.min(reader.num_docs() as usize);
393 let mut collector = TopKCollector::new(segment_limit);
394 collect_segment_with_limit(reader, query, &mut collector, segment_limit).await?;
395 Ok(collector.into_results_with_count())
396}
397
398pub async fn search_segment_with_positions_and_count(
400 reader: &SegmentReader,
401 query: &dyn Query,
402 limit: usize,
403) -> Result<(Vec<SearchResult>, u32)> {
404 let segment_limit = limit.min(reader.num_docs() as usize);
405 let mut collector = TopKCollector::with_positions(segment_limit);
406 collect_segment_with_limit(reader, query, &mut collector, segment_limit).await?;
407 Ok(collector.into_results_with_count())
408}
409
410fn positions_for_next_collector(
415 positions: &mut Option<super::MatchedPositions>,
416 remaining_consumers: &mut usize,
417) -> super::MatchedPositions {
418 assert!(
419 *remaining_consumers > 0,
420 "position consumer count underflow"
421 );
422 *remaining_consumers -= 1;
423 if *remaining_consumers == 0 {
424 positions
425 .take()
426 .expect("owned positions must remain for the final collector")
427 } else {
428 positions
429 .as_ref()
430 .cloned()
431 .expect("owned positions must remain while collectors are pending")
432 }
433}
434
435impl<A: Collector, B: Collector> Collector for (&mut A, &mut B) {
437 fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]) {
438 self.0.collect(doc_id, score, positions);
439 self.1.collect(doc_id, score, positions);
440 }
441 fn needs_positions(&self) -> bool {
442 self.0.needs_positions() || self.1.needs_positions()
443 }
444 fn would_collect(&self, doc_id: DocId, score: Score) -> bool {
445 (self.0.needs_positions() && self.0.would_collect(doc_id, score))
446 || (self.1.needs_positions() && self.1.would_collect(doc_id, score))
447 }
448 fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
449 let wants = [
450 self.0.needs_positions() && self.0.would_collect(doc_id, score),
451 self.1.needs_positions() && self.1.would_collect(doc_id, score),
452 ];
453 let mut remaining = wants.iter().filter(|&&want| want).count();
454 let mut positions = Some(positions);
455
456 if wants[0] {
457 self.0.collect_owned(
458 doc_id,
459 score,
460 positions_for_next_collector(&mut positions, &mut remaining),
461 );
462 } else {
463 self.0.collect(doc_id, score, &[]);
464 }
465 if wants[1] {
466 self.1.collect_owned(
467 doc_id,
468 score,
469 positions_for_next_collector(&mut positions, &mut remaining),
470 );
471 } else {
472 self.1.collect(doc_id, score, &[]);
473 }
474 }
475}
476
477impl<A: Collector, B: Collector, C: Collector> Collector for (&mut A, &mut B, &mut C) {
479 fn collect(&mut self, doc_id: DocId, score: Score, positions: &[(u32, Vec<ScoredPosition>)]) {
480 self.0.collect(doc_id, score, positions);
481 self.1.collect(doc_id, score, positions);
482 self.2.collect(doc_id, score, positions);
483 }
484 fn needs_positions(&self) -> bool {
485 self.0.needs_positions() || self.1.needs_positions() || self.2.needs_positions()
486 }
487 fn would_collect(&self, doc_id: DocId, score: Score) -> bool {
488 (self.0.needs_positions() && self.0.would_collect(doc_id, score))
489 || (self.1.needs_positions() && self.1.would_collect(doc_id, score))
490 || (self.2.needs_positions() && self.2.would_collect(doc_id, score))
491 }
492 fn collect_owned(&mut self, doc_id: DocId, score: Score, positions: super::MatchedPositions) {
493 let wants = [
494 self.0.needs_positions() && self.0.would_collect(doc_id, score),
495 self.1.needs_positions() && self.1.would_collect(doc_id, score),
496 self.2.needs_positions() && self.2.would_collect(doc_id, score),
497 ];
498 let mut remaining = wants.iter().filter(|&&want| want).count();
499 let mut positions = Some(positions);
500
501 if wants[0] {
502 self.0.collect_owned(
503 doc_id,
504 score,
505 positions_for_next_collector(&mut positions, &mut remaining),
506 );
507 } else {
508 self.0.collect(doc_id, score, &[]);
509 }
510 if wants[1] {
511 self.1.collect_owned(
512 doc_id,
513 score,
514 positions_for_next_collector(&mut positions, &mut remaining),
515 );
516 } else {
517 self.1.collect(doc_id, score, &[]);
518 }
519 if wants[2] {
520 self.2.collect_owned(
521 doc_id,
522 score,
523 positions_for_next_collector(&mut positions, &mut remaining),
524 );
525 } else {
526 self.2.collect(doc_id, score, &[]);
527 }
528 }
529}
530
531pub async fn collect_segment<C: Collector>(
549 reader: &SegmentReader,
550 query: &dyn Query,
551 collector: &mut C,
552) -> Result<()> {
553 collect_segment_with_limit(reader, query, collector, usize::MAX / 2).await
555}
556
557pub async fn collect_segment_with_limit<C: Collector>(
566 reader: &SegmentReader,
567 query: &dyn Query,
568 collector: &mut C,
569 limit: usize,
570) -> Result<()> {
571 let options = super::ScorerOptions {
572 collect_positions: collector.needs_positions(),
573 };
574 let mut scorer = query.scorer_with_options(reader, limit, options).await?;
575 drive_scorer(scorer.as_mut(), collector);
576 Ok(())
577}
578
579fn drive_scorer<C: Collector>(scorer: &mut dyn super::Scorer, collector: &mut C) {
581 let needs_positions = collector.needs_positions();
582 let mut doc = scorer.doc();
583 while doc != TERMINATED {
584 let score = scorer.score();
585 if needs_positions && collector.would_collect(doc, score) {
586 let positions = scorer.matched_positions().unwrap_or_default();
587 collector.collect_owned(doc, score, positions);
588 } else {
589 collector.collect(doc, score, &[]);
590 }
591 doc = scorer.advance();
592 }
593}
594
595#[cfg(feature = "sync")]
599pub fn search_segment_with_count_sync(
600 reader: &SegmentReader,
601 query: &dyn Query,
602 limit: usize,
603) -> Result<(Vec<SearchResult>, u32)> {
604 let segment_limit = limit.min(reader.num_docs() as usize);
605 let mut collector = TopKCollector::new(segment_limit);
606 collect_segment_with_limit_sync(reader, query, &mut collector, segment_limit)?;
607 Ok(collector.into_results_with_count())
608}
609
610#[cfg(feature = "sync")]
612pub fn search_segment_with_positions_and_count_sync(
613 reader: &SegmentReader,
614 query: &dyn Query,
615 limit: usize,
616) -> Result<(Vec<SearchResult>, u32)> {
617 let segment_limit = limit.min(reader.num_docs() as usize);
618 let mut collector = TopKCollector::with_positions(segment_limit);
619 collect_segment_with_limit_sync(reader, query, &mut collector, segment_limit)?;
620 Ok(collector.into_results_with_count())
621}
622
623#[cfg(feature = "sync")]
625pub fn collect_segment_with_limit_sync<C: Collector>(
626 reader: &SegmentReader,
627 query: &dyn Query,
628 collector: &mut C,
629 limit: usize,
630) -> Result<()> {
631 let options = super::ScorerOptions {
632 collect_positions: collector.needs_positions(),
633 };
634 let mut scorer = query.scorer_sync_with_options(reader, limit, options)?;
635 drive_scorer(scorer.as_mut(), collector);
636 Ok(())
637}
638
639#[cfg(test)]
640mod tests {
641 use super::*;
642 use std::sync::Arc;
643 use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
644
645 #[derive(Default)]
646 struct OwnedPositionCollector {
647 owned_calls: usize,
648 borrowed_calls: usize,
649 positions: super::super::MatchedPositions,
650 }
651
652 impl Collector for OwnedPositionCollector {
653 fn collect(
654 &mut self,
655 _doc_id: DocId,
656 _score: Score,
657 positions: &[(u32, Vec<ScoredPosition>)],
658 ) {
659 self.borrowed_calls += 1;
660 self.positions = positions.to_vec();
661 }
662
663 fn collect_owned(
664 &mut self,
665 _doc_id: DocId,
666 _score: Score,
667 positions: super::super::MatchedPositions,
668 ) {
669 self.owned_calls += 1;
670 self.positions = positions;
671 }
672
673 fn needs_positions(&self) -> bool {
674 true
675 }
676 }
677
678 struct PositionCountingScorer {
679 index: usize,
680 position_calls: Arc<AtomicUsize>,
681 }
682
683 impl super::super::DocSet for PositionCountingScorer {
684 fn doc(&self) -> DocId {
685 if self.index < 3 {
686 self.index as DocId
687 } else {
688 TERMINATED
689 }
690 }
691
692 fn advance(&mut self) -> DocId {
693 self.index += 1;
694 self.doc()
695 }
696
697 fn seek(&mut self, target: DocId) -> DocId {
698 self.index = target.min(3) as usize;
699 self.doc()
700 }
701
702 fn size_hint(&self) -> u32 {
703 3u32.saturating_sub(self.index as u32)
704 }
705 }
706
707 impl super::super::Scorer for PositionCountingScorer {
708 fn score(&self) -> Score {
709 [10.0, 1.0, 2.0][self.index]
710 }
711
712 fn matched_positions(&self) -> Option<super::super::MatchedPositions> {
713 self.position_calls.fetch_add(1, AtomicOrdering::Relaxed);
714 Some(vec![(7, vec![ScoredPosition::new(self.index as u32, 1.0)])])
715 }
716 }
717
718 #[test]
719 fn test_top_k_collector() {
720 let mut collector = TopKCollector::new(3);
721
722 collector.collect(0, 1.0, &[]);
723 collector.collect(1, 3.0, &[]);
724 collector.collect(2, 2.0, &[]);
725 collector.collect(3, 4.0, &[]);
726 collector.collect(4, 0.5, &[]);
727
728 let results = collector.into_sorted_results();
729
730 assert_eq!(results.len(), 3);
731 assert_eq!(results[0].doc_id, 3); assert_eq!(results[1].doc_id, 1); assert_eq!(results[2].doc_id, 2); }
735
736 #[test]
737 fn top_k_zero_retains_no_results() {
738 let mut collector = TopKCollector::new(0);
739 collector.collect(1, 1.0, &[]);
740
741 assert!(collector.into_sorted_results().is_empty());
742 }
743
744 #[test]
745 fn huge_top_k_does_not_trigger_a_huge_initial_allocation() {
746 let collector = TopKCollector::new(usize::MAX);
747
748 assert!(collector.heap.capacity() <= MAX_INITIAL_TOP_K_CAPACITY);
749 }
750
751 #[test]
752 fn positions_are_only_materialized_for_competitive_hits() {
753 let calls = Arc::new(AtomicUsize::new(0));
754 let mut scorer = PositionCountingScorer {
755 index: 0,
756 position_calls: Arc::clone(&calls),
757 };
758 let mut collector = TopKCollector::with_positions(1);
759
760 drive_scorer(&mut scorer, &mut collector);
761
762 assert_eq!(calls.load(AtomicOrdering::Relaxed), 1);
763 assert_eq!(collector.total_seen(), 3);
764 let results = collector.into_sorted_results();
765 assert_eq!(results.len(), 1);
766 assert_eq!(results[0].doc_id, 0);
767 assert_eq!(results[0].positions[0].0, 7);
768 }
769
770 #[test]
771 fn tuple_moves_owned_positions_to_single_position_collector() {
772 let mut positions = OwnedPositionCollector::default();
773 let mut count = CountCollector::new();
774 let input = vec![(7, vec![ScoredPosition::new(3, 1.0)])];
775 let input_ptr = input[0].1.as_ptr();
776
777 (&mut positions, &mut count).collect_owned(11, 2.0, input);
778
779 assert_eq!(positions.owned_calls, 1);
780 assert_eq!(positions.borrowed_calls, 0);
781 assert_eq!(positions.positions[0].1.as_ptr(), input_ptr);
782 assert_eq!(count.count(), 1);
783 }
784
785 #[test]
786 fn tuple_clones_for_all_but_final_position_collector() {
787 let mut first = OwnedPositionCollector::default();
788 let mut second = OwnedPositionCollector::default();
789 let mut count = CountCollector::new();
790 let input = vec![(7, vec![ScoredPosition::new(3, 1.0)])];
791 let input_ptr = input[0].1.as_ptr();
792
793 (&mut first, &mut count, &mut second).collect_owned(11, 2.0, input);
794
795 assert_eq!((first.owned_calls, first.borrowed_calls), (1, 0));
796 assert_eq!((second.owned_calls, second.borrowed_calls), (1, 0));
797 assert_ne!(first.positions[0].1.as_ptr(), input_ptr);
798 assert_eq!(second.positions[0].1.as_ptr(), input_ptr);
799 assert_eq!(count.count(), 1);
800 }
801
802 #[test]
803 fn test_count_collector() {
804 let mut collector = CountCollector::new();
805
806 collector.collect(0, 1.0, &[]);
807 collector.collect(1, 2.0, &[]);
808 collector.collect(2, 3.0, &[]);
809
810 assert_eq!(collector.count(), 3);
811 }
812
813 #[test]
814 fn test_multi_collector() {
815 let mut top_k = TopKCollector::new(2);
816 let mut count = CountCollector::new();
817
818 for (doc_id, score) in [(0, 1.0), (1, 3.0), (2, 2.0), (3, 4.0), (4, 0.5)] {
820 top_k.collect(doc_id, score, &[]);
821 count.collect(doc_id, score, &[]);
822 }
823
824 assert_eq!(count.count(), 5);
826
827 let results = top_k.into_sorted_results();
829 assert_eq!(results.len(), 2);
830 assert_eq!(results[0].doc_id, 3); assert_eq!(results[1].doc_id, 1); }
833}