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 self.positions
132 .iter()
133 .map(|(field_id, scored_positions)| {
134 let mut ordinals = Vec::with_capacity(scored_positions.len());
139 ordinals.extend(scored_positions.iter().map(|sp| {
140 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 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
240#[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#[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
318pub struct TopKCollector {
320 heap: TopKHeap,
321 k: usize,
322 total_seen: u32,
324}
325
326const 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 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 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 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 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#[derive(Default)]
515pub struct CountCollector {
516 count: u64,
517}
518
519impl CountCollector {
520 pub fn new() -> Self {
521 Self { count: 0 }
522 }
523
524 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
542pub 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
554pub 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
566fn 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
591impl<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
633impl<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
687pub async fn collect_segment<C: Collector>(
705 reader: &SegmentReader,
706 query: &dyn Query,
707 collector: &mut C,
708) -> Result<()> {
709 collect_segment_with_limit(reader, query, collector, usize::MAX / 2).await
711}
712
713pub 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
730pub 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
752fn 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#[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#[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#[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#[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#[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#[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#[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
904pub 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
929pub 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
948pub(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); assert_eq!(results[1].doc_id, 1); assert_eq!(results[2].doc_id, 2); }
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 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 assert_eq!(count.count(), 5);
1220
1221 let results = top_k.into_sorted_results();
1223 assert_eq!(results.len(), 2);
1224 assert_eq!(results[0].doc_id, 3); assert_eq!(results[1].doc_id, 1); }
1227}