1mod rrf_scores;
22pub use rrf_scores::*;
23
24use rustc_hash::FxHashMap;
25
26use super::vector::MultiValueCombiner;
27use super::{ScoredPosition, SearchResult, compare_search_results_desc};
28
29pub const DEFAULT_RRF_K: f32 = 60.0;
31pub const MAX_FUSION_SUB_QUERIES: usize = 16;
33pub const MAX_FUSION_CANDIDATE_SLOTS: usize = 200_000;
35pub const MAX_FUSION_CHUNK_SLOTS: usize = 500_000;
37
38#[derive(Debug, Clone, Copy, PartialEq)]
40pub enum FusionMethod {
41 Rrf { k: f32 },
47 NormalizedWeightedSum,
59}
60
61impl Default for FusionMethod {
62 fn default() -> Self {
63 FusionMethod::Rrf { k: DEFAULT_RRF_K }
64 }
65}
66
67#[inline]
70pub(crate) fn rrf_contribution(k: f32, rank: usize) -> f32 {
71 1.0 / (k + rank as f32)
72}
73
74pub fn fuse_ranked_lists(
82 lists: Vec<(Vec<SearchResult>, f32)>,
83 method: FusionMethod,
84 limit: usize,
85) -> Vec<SearchResult> {
86 const MAX_INITIAL_FUSION_CAPACITY: usize = 200_000;
89 let capacity = lists
90 .iter()
91 .map(|(list, _)| list.len())
92 .fold(0usize, usize::saturating_add)
93 .min(MAX_INITIAL_FUSION_CAPACITY);
94 let mut fused: FxHashMap<(u128, u32), SearchResult> =
95 FxHashMap::with_capacity_and_hasher(capacity, Default::default());
96
97 for (list, weight) in lists {
98 let (min_score, inv_range) = match method {
100 FusionMethod::NormalizedWeightedSum if !list.is_empty() => {
101 let mut min = f32::INFINITY;
102 let mut max = f32::NEG_INFINITY;
103 for r in &list {
104 min = min.min(r.score);
105 max = max.max(r.score);
106 }
107 let range = max - min;
108 (min, if range > 0.0 { 1.0 / range } else { 0.0 })
109 }
110 _ => (0.0, 0.0),
111 };
112
113 for (idx, result) in list.into_iter().enumerate() {
114 let contribution = match method {
115 FusionMethod::Rrf { k } => weight * rrf_contribution(k, idx + 1),
116 FusionMethod::NormalizedWeightedSum => {
117 if inv_range > 0.0 {
119 weight * (result.score - min_score) * inv_range
120 } else {
121 weight
122 }
123 }
124 };
125 fused
126 .entry((result.segment_id, result.doc_id))
127 .and_modify(|r| r.score += contribution)
128 .or_insert_with(|| SearchResult {
129 score: contribution,
130 ..result
131 });
132 }
133 }
134
135 let mut results: Vec<SearchResult> = fused.into_values().collect();
136 if results.len() > limit {
137 results.select_nth_unstable_by(limit, compare_search_results_desc);
138 results.truncate(limit);
139 }
140 results.sort_unstable_by(compare_search_results_desc);
141 results
142}
143
144type ChunkKey = (u128, u32, u32);
145
146fn ranked_chunks(list: &[SearchResult], chunks: &mut Vec<(ChunkKey, f32)>) {
148 chunks.clear();
149 for result in list {
150 let mut had_positions = false;
151 for (_field_id, scored_positions) in &result.positions {
152 for sp in scored_positions {
153 had_positions = true;
154 chunks.push(((result.segment_id, result.doc_id, sp.position), sp.score));
155 }
156 }
157 if !had_positions {
158 chunks.push(((result.segment_id, result.doc_id, 0), result.score));
161 }
162 }
163
164 chunks.sort_unstable_by(|a, b| a.0.cmp(&b.0).then_with(|| b.1.total_cmp(&a.1)));
168 chunks.dedup_by_key(|entry| entry.0);
169
170 chunks.sort_unstable_by(|a, b| b.1.total_cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
173}
174
175pub fn fuse_ranked_lists_chunked(
197 lists: Vec<(Vec<SearchResult>, f32)>,
198 method: FusionMethod,
199 combiner: MultiValueCombiner,
200 limit: usize,
201) -> Vec<SearchResult> {
202 fuse_ranked_lists_chunked_impl(lists, method, combiner, limit)
203}
204
205fn fuse_ranked_lists_chunked_impl<L: AsRef<[SearchResult]>>(
206 lists: impl IntoIterator<Item = (L, f32)>,
207 method: FusionMethod,
208 combiner: MultiValueCombiner,
209 limit: usize,
210) -> Vec<SearchResult> {
211 let mut contributions: Vec<(ChunkKey, u16, f32)> = Vec::new();
217 let mut chunks: Vec<(ChunkKey, f32)> = Vec::new();
219
220 for (list_index, (list, weight)) in lists.into_iter().enumerate() {
221 ranked_chunks(list.as_ref(), &mut chunks);
222 if chunks.is_empty() {
223 continue;
224 }
225
226 let (min_score, inv_range) = match method {
228 FusionMethod::NormalizedWeightedSum => {
229 let max = chunks.first().map(|c| c.1).unwrap_or(0.0);
230 let min = chunks.last().map(|c| c.1).unwrap_or(0.0);
231 let range = max - min;
232 (min, if range > 0.0 { 1.0 / range } else { 0.0 })
233 }
234 _ => (0.0, 0.0),
235 };
236
237 contributions.reserve(chunks.len());
238 let list_index = list_index as u16;
239 for (rank, &(key, score)) in chunks.iter().enumerate() {
240 let contribution = match method {
241 FusionMethod::Rrf { k } => weight * rrf_contribution(k, rank + 1),
242 FusionMethod::NormalizedWeightedSum => {
243 if inv_range > 0.0 {
244 weight * (score - min_score) * inv_range
245 } else {
246 weight
247 }
248 }
249 };
250 contributions.push((key, list_index, contribution));
251 }
252 }
253
254 contributions.sort_unstable_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1)));
258
259 let mut results: Vec<SearchResult> = Vec::new();
260 let mut ordinals: Vec<(u32, f32)> = Vec::new();
261 let mut index = 0;
262 while index < contributions.len() {
263 let (segment_id, doc_id, _) = contributions[index].0;
264 ordinals.clear();
265 while index < contributions.len() {
266 let (seg, doc, ordinal) = contributions[index].0;
267 if seg != segment_id || doc != doc_id {
268 break;
269 }
270 let mut fused = 0.0f32;
271 while index < contributions.len()
272 && contributions[index].0 == (segment_id, doc_id, ordinal)
273 {
274 fused += contributions[index].2;
275 index += 1;
276 }
277 ordinals.push((ordinal, fused));
278 }
279 let score = combiner.combine(&ordinals);
280 let scored_positions: Vec<ScoredPosition> = ordinals
281 .iter()
282 .map(|&(ord, s)| ScoredPosition::new(ord, s))
283 .collect();
284 results.push(SearchResult {
285 doc_id,
286 score,
287 segment_id,
288 positions: vec![(0, scored_positions)],
289 });
290 }
291
292 if results.len() > limit {
293 results.select_nth_unstable_by(limit, compare_search_results_desc);
294 results.truncate(limit);
295 }
296 results.sort_unstable_by(compare_search_results_desc);
297 results
298}
299
300pub fn try_fuse_ranked_lists_chunked(
305 lists: Vec<(Vec<SearchResult>, f32)>,
306 method: FusionMethod,
307 combiner: MultiValueCombiner,
308 limit: usize,
309) -> Result<Vec<SearchResult>, String> {
310 let borrowed: Vec<_> = lists
311 .iter()
312 .map(|(list, weight)| (list.as_slice(), *weight))
313 .collect();
314 validate_fusion_lists(&borrowed, method, combiner)?;
315 Ok(fuse_ranked_lists_chunked(lists, method, combiner, limit))
316}
317
318pub fn try_fuse_ranked_lists_chunked_borrowed(
321 lists: &[(&[SearchResult], f32)],
322 method: FusionMethod,
323 combiner: MultiValueCombiner,
324 limit: usize,
325) -> Result<Vec<SearchResult>, String> {
326 validate_fusion_lists(lists, method, combiner)?;
327 Ok(fuse_ranked_lists_chunked_impl(
328 lists.iter().copied(),
329 method,
330 combiner,
331 limit,
332 ))
333}
334
335fn validate_fusion_lists(
336 lists: &[(&[SearchResult], f32)],
337 method: FusionMethod,
338 combiner: MultiValueCombiner,
339) -> Result<(), String> {
340 if lists.is_empty() {
341 return Err("fusion requires at least one ranked list".to_string());
342 }
343 if lists.len() > MAX_FUSION_SUB_QUERIES {
344 return Err(format!(
345 "fusion supports at most {MAX_FUSION_SUB_QUERIES} ranked lists"
346 ));
347 }
348 if let FusionMethod::Rrf { k } = method
349 && (!k.is_finite() || k < 0.0)
350 {
351 return Err(format!(
352 "fusion RRF k must be finite and non-negative, got {k}"
353 ));
354 }
355 combiner.validate()?;
356
357 let mut candidates = 0usize;
358 let mut chunks = 0usize;
359 for (list_index, &(list, weight)) in lists.iter().enumerate() {
360 if !weight.is_finite() || weight < 0.0 {
361 return Err(format!(
362 "fusion list weight at index {list_index} must be finite and non-negative, \
363 got {weight}"
364 ));
365 }
366 candidates = candidates
367 .checked_add(list.len())
368 .ok_or_else(|| "fusion candidate count overflow".to_string())?;
369 if candidates > MAX_FUSION_CANDIDATE_SLOTS {
370 return Err(format!(
371 "fusion contains more than {MAX_FUSION_CANDIDATE_SLOTS} candidate slots"
372 ));
373 }
374 for result in list {
375 let position_count = result
376 .positions
377 .iter()
378 .try_fold(0usize, |count, (_, positions)| {
379 count.checked_add(positions.len())
380 })
381 .ok_or_else(|| "fusion chunk count overflow".to_string())?;
382 chunks = chunks
384 .checked_add(position_count.max(1))
385 .ok_or_else(|| "fusion chunk count overflow".to_string())?;
386 if chunks > MAX_FUSION_CHUNK_SLOTS {
387 return Err(format!(
388 "fusion expands to more than {MAX_FUSION_CHUNK_SLOTS} ordinal chunks"
389 ));
390 }
391 }
392 }
393
394 Ok(())
395}
396
397#[cfg(test)]
398mod tests {
399 use super::*;
400
401 fn result(doc_id: u32, score: f32) -> SearchResult {
402 SearchResult {
403 doc_id,
404 score,
405 segment_id: 1,
406 positions: Vec::new(),
407 }
408 }
409
410 #[test]
411 fn test_rrf_union_includes_single_list_docs() {
412 let sparse = vec![result(1, 10.0), result(2, 5.0)];
414 let dense = vec![result(3, 0.9), result(1, 0.8)];
415
416 let fused = fuse_ranked_lists(
417 vec![(sparse, 1.0), (dense, 1.0)],
418 FusionMethod::Rrf { k: 60.0 },
419 10,
420 );
421
422 assert_eq!(fused.len(), 3);
423 assert_eq!(fused[0].doc_id, 1);
425 let expected = 1.0 / 61.0 + 1.0 / 62.0;
426 assert!((fused[0].score - expected).abs() < 1e-6);
427 let ids: Vec<u32> = fused.iter().map(|r| r.doc_id).collect();
429 assert!(ids.contains(&2) && ids.contains(&3));
430 }
431
432 #[test]
433 fn test_rrf_weights_scale_contribution() {
434 let a = vec![result(1, 1.0)];
435 let b = vec![result(2, 1.0)];
436
437 let fused = fuse_ranked_lists(vec![(a, 1.0), (b, 2.0)], FusionMethod::Rrf { k: 60.0 }, 10);
439 assert_eq!(fused[0].doc_id, 2);
440 assert!((fused[0].score - 2.0 / 61.0).abs() < 1e-6);
441 }
442
443 #[test]
444 fn test_normalized_weighted_sum() {
445 let sparse = vec![result(1, 20.0), result(2, 10.0), result(3, 0.0)];
447 let dense = vec![result(2, 0.99), result(1, 0.55), result(3, 0.11)];
448
449 let fused = fuse_ranked_lists(
450 vec![(sparse, 0.5), (dense, 0.5)],
451 FusionMethod::NormalizedWeightedSum,
452 10,
453 );
454
455 assert_eq!(fused.len(), 3);
456 assert_eq!(fused[0].doc_id, 1);
459 assert!((fused[0].score - 0.75).abs() < 1e-6);
460 assert!((fused[1].score - 0.75).abs() < 1e-6);
461 assert_eq!(fused[2].doc_id, 3);
462 assert!(fused[2].score.abs() < 1e-6);
463 }
464
465 #[test]
466 fn test_limit_truncation() {
467 let list: Vec<SearchResult> = (0..100).map(|i| result(i, 100.0 - i as f32)).collect();
468 let fused = fuse_ranked_lists(vec![(list, 1.0)], FusionMethod::default(), 5);
469 assert_eq!(fused.len(), 5);
470 assert_eq!(fused[0].doc_id, 0);
471 }
472
473 fn chunked(doc_id: u32, chunks: &[(u32, f32)]) -> SearchResult {
474 let positions = vec![(
475 0u32,
476 chunks
477 .iter()
478 .map(|&(ord, s)| ScoredPosition::new(ord, s))
479 .collect(),
480 )];
481 SearchResult {
482 doc_id,
483 score: chunks.iter().map(|&(_, s)| s).fold(0.0, f32::max),
485 segment_id: 1,
486 positions,
487 }
488 }
489
490 #[test]
491 fn one_branch_cannot_vote_twice_for_the_same_passage_across_fields() {
492 let mut duplicate = chunked(1, &[(0, 10.0)]);
493 duplicate
494 .positions
495 .push((1, vec![ScoredPosition::new(0, 9.0)]));
496 let unique = chunked(2, &[(0, 11.0)]);
497 let fused = fuse_ranked_lists_chunked(
498 vec![(vec![unique, duplicate], 1.0)],
499 FusionMethod::Rrf { k: 60.0 },
500 MultiValueCombiner::Max,
501 10,
502 );
503 assert_eq!(fused[0].doc_id, 2);
504 assert_eq!(fused[1].score, 1.0 / 62.0);
505 }
506
507 #[test]
512 fn test_chunked_fusion_junk_vertical_does_not_outvote() {
513 let sparse = vec![
515 chunked(1, &[(0, 10.0)]),
516 chunked(2, &[(0, 5.0)]),
517 chunked(3, &[(0, 4.0)]),
518 chunked(4, &[(0, 3.0)]),
519 chunked(9, &[(2, 2.0)]),
520 ];
521 let dense = vec![
524 chunked(7, &[(0, 0.31)]),
525 chunked(8, &[(1, 0.30)]),
526 chunked(6, &[(0, 0.29)]),
527 chunked(5, &[(3, 0.28)]),
528 chunked(9, &[(5, 0.27)]),
529 ];
530
531 let fused = fuse_ranked_lists_chunked(
532 vec![(sparse, 1.0), (dense, 1.0)],
533 FusionMethod::Rrf { k: 60.0 },
534 MultiValueCombiner::Max,
535 10,
536 );
537
538 assert_eq!(
539 fused[0].doc_id, 1,
540 "sparse rank-1 doc must win over doc 9 (present in both lists on different chunks)"
541 );
542 }
543
544 #[test]
547 fn test_chunked_fusion_same_chunk_corroboration_wins() {
548 let sparse = vec![chunked(1, &[(3, 9.0)]), chunked(2, &[(0, 8.0)])];
551 let dense = vec![chunked(1, &[(3, 0.9)]), chunked(2, &[(7, 0.8)])];
552
553 let fused = fuse_ranked_lists_chunked(
554 vec![(sparse, 1.0), (dense, 1.0)],
555 FusionMethod::Rrf { k: 60.0 },
556 MultiValueCombiner::Max,
557 10,
558 );
559
560 assert_eq!(fused[0].doc_id, 1);
561 let expected_doc1 = 2.0 / 61.0;
563 assert!((fused[0].score - expected_doc1).abs() < 1e-6);
564 assert!(fused[1].score < expected_doc1 / 1.9);
565
566 let (_, positions) = &fused[0].positions[0..1][0];
568 assert_eq!(positions.len(), 1);
569 assert_eq!(positions[0].position, 3, "fused chunk ordinal preserved");
570 }
571
572 #[test]
575 fn test_chunked_fusion_pseudo_chunk_for_docs_without_positions() {
576 let text = vec![result(1, 3.0), result(2, 2.0)]; let dense = vec![chunked(1, &[(0, 0.9)])];
578
579 let fused = fuse_ranked_lists_chunked(
580 vec![(text, 1.0), (dense, 1.0)],
581 FusionMethod::Rrf { k: 60.0 },
582 MultiValueCombiner::Max,
583 10,
584 );
585
586 assert_eq!(fused[0].doc_id, 1);
587 assert!((fused[0].score - 2.0 / 61.0).abs() < 1e-6);
588 assert_eq!(fused.len(), 2);
589 }
590
591 #[test]
592 fn test_validated_chunked_fusion_rejects_invalid_parameters() {
593 assert!(
594 try_fuse_ranked_lists_chunked(
595 vec![(vec![result(1, 1.0)], -1.0)],
596 FusionMethod::default(),
597 MultiValueCombiner::Max,
598 10,
599 )
600 .is_err()
601 );
602 assert!(
603 try_fuse_ranked_lists_chunked(
604 vec![(vec![result(1, 1.0)], 1.0)],
605 FusionMethod::Rrf { k: f32::NAN },
606 MultiValueCombiner::Max,
607 10,
608 )
609 .is_err()
610 );
611 }
612
613 #[test]
617 fn chunked_fusion_sort_grouping_matches_hash_map_reference() {
618 use rustc_hash::FxHashMap;
619
620 fn seg(mut result: SearchResult, segment_id: u128) -> SearchResult {
621 result.segment_id = segment_id;
622 result
623 }
624 let lists = vec![
625 (
626 vec![
627 chunked(1, &[(2, 9.0), (0, 8.5)]),
628 seg(chunked(1, &[(0, 7.0)]), 2),
629 chunked(5, &[(1, 6.0)]),
630 result(8, 5.0),
631 ],
632 1.0,
633 ),
634 (
635 vec![
636 chunked(5, &[(1, 0.9), (4, 0.8)]),
637 chunked(1, &[(0, 0.7)]),
638 seg(chunked(1, &[(0, 0.6)]), 2),
639 result(9, 0.5),
640 ],
641 0.7,
642 ),
643 (vec![chunked(1, &[(2, 3.0)]), chunked(8, &[(0, 2.0)])], 1.3),
644 ];
645
646 let mut reference: FxHashMap<(u128, u32, u32), f32> = FxHashMap::default();
648 for (list, weight) in &lists {
649 let mut chunks: Vec<((u128, u32, u32), f32)> = Vec::new();
650 for r in list {
651 let mut had = false;
652 for (_, positions) in &r.positions {
653 for p in positions {
654 had = true;
655 chunks.push(((r.segment_id, r.doc_id, p.position), p.score));
656 }
657 }
658 if !had {
659 chunks.push(((r.segment_id, r.doc_id, 0), r.score));
660 }
661 }
662 chunks.sort_unstable_by(|a, b| b.1.total_cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
663 for (rank, &(key, _)) in chunks.iter().enumerate() {
664 *reference.entry(key).or_insert(0.0) += weight * rrf_contribution(60.0, rank + 1);
665 }
666 }
667
668 let fused = fuse_ranked_lists_chunked(
669 lists,
670 FusionMethod::Rrf { k: 60.0 },
671 MultiValueCombiner::Max,
672 100,
673 );
674 let mut seen = 0;
675 for result in &fused {
676 let (_, positions) = &result.positions[0];
677 let ordinals: Vec<u32> = positions.iter().map(|p| p.position).collect();
678 let mut sorted = ordinals.clone();
679 sorted.sort_unstable();
680 assert_eq!(
681 ordinals, sorted,
682 "ordinals ascending for doc {}",
683 result.doc_id
684 );
685 let mut best = f32::NEG_INFINITY;
686 for p in positions {
687 let expected = reference[&(result.segment_id, result.doc_id, p.position)];
688 assert_eq!(
689 p.score.to_bits(),
690 expected.to_bits(),
691 "chunk ({}, {}, {}) fused score",
692 result.segment_id,
693 result.doc_id,
694 p.position
695 );
696 best = best.max(expected);
697 seen += 1;
698 }
699 assert_eq!(result.score.to_bits(), best.to_bits());
700 }
701 assert_eq!(seen, reference.len(), "every fused chunk is reported once");
702 assert_eq!(fused.len(), 5, "(1,seg1) (1,seg2) 5 8 9");
703 for pair in fused.windows(2) {
704 assert!(compare_search_results_desc(&pair[0], &pair[1]).is_le());
705 }
706 }
707
708 #[test]
709 fn test_duplicate_across_segments_not_merged() {
710 let mut a = result(1, 1.0);
712 a.segment_id = 1;
713 let mut b = result(1, 1.0);
714 b.segment_id = 2;
715
716 let fused = fuse_ranked_lists(
717 vec![(vec![a], 1.0), (vec![b], 1.0)],
718 FusionMethod::default(),
719 10,
720 );
721 assert_eq!(fused.len(), 2);
722 }
723}