1use std::cmp::{Ordering, Reverse};
4use std::collections::{BinaryHeap, HashMap, HashSet};
5
6use ndarray::Array1;
7use ndarray::{Array2, ArrayView2};
8use rayon::prelude::*;
9use serde::{Deserialize, Serialize};
10
11use crate::codec::CentroidStore;
12use crate::error::Result;
13use crate::maxsim;
14
15type ProbePartial = (
17 Vec<BinaryHeap<(Reverse<OrdF32>, usize)>>,
18 HashMap<usize, f32>,
19);
20
21const DECOMPRESS_CHUNK_SIZE: usize = 128;
25
26#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct SearchParameters {
29 pub batch_size: usize,
31 pub n_full_scores: usize,
33 pub top_k: usize,
35 pub n_ivf_probe: usize,
37 #[serde(default = "default_centroid_batch_size")]
41 pub centroid_batch_size: usize,
42 #[serde(default = "default_centroid_score_threshold")]
47 pub centroid_score_threshold: Option<f32>,
48}
49
50fn default_centroid_batch_size() -> usize {
51 100_000
52}
53
54fn default_centroid_score_threshold() -> Option<f32> {
55 Some(0.4)
56}
57
58impl Default for SearchParameters {
59 fn default() -> Self {
60 Self {
61 batch_size: 2000,
62 n_full_scores: 4096,
63 top_k: 10,
64 n_ivf_probe: 8,
65 centroid_batch_size: default_centroid_batch_size(),
66 centroid_score_threshold: default_centroid_score_threshold(),
67 }
68 }
69}
70
71#[derive(Debug, Clone, Serialize, Deserialize)]
73pub struct QueryResult {
74 pub query_id: usize,
76 pub passage_ids: Vec<i64>,
78 pub scores: Vec<f32>,
80}
81
82fn colbert_score(query: &ArrayView2<f32>, doc: &ArrayView2<f32>) -> f32 {
89 maxsim::maxsim_score(query, doc)
90}
91
92#[derive(Clone, Copy, PartialEq)]
94struct OrdF32(f32);
95
96impl Eq for OrdF32 {}
97
98impl PartialOrd for OrdF32 {
99 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
100 Some(self.cmp(other))
101 }
102}
103
104impl Ord for OrdF32 {
105 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
106 cmp_score_ascending(self.0, other.0)
107 }
108}
109
110fn cmp_score_ascending(a: f32, b: f32) -> Ordering {
111 match (a.is_finite(), b.is_finite()) {
112 (true, true) => a.total_cmp(&b),
113 (true, false) => Ordering::Greater,
114 (false, true) => Ordering::Less,
115 (false, false) => Ordering::Equal,
116 }
117}
118
119fn cmp_score_descending(a: f32, b: f32) -> Ordering {
120 cmp_score_ascending(b, a)
121}
122
123fn is_score_better(candidate: f32, current: f32) -> bool {
124 cmp_score_ascending(candidate, current).is_gt()
125}
126
127fn max_score(a: f32, b: f32) -> f32 {
128 if is_score_better(b, a) {
129 b
130 } else {
131 a
132 }
133}
134
135fn ivf_probe_batched(
141 query: &Array2<f32>,
142 centroids: &CentroidStore,
143 n_probe: usize,
144 batch_size: usize,
145 centroid_score_threshold: Option<f32>,
146) -> Vec<usize> {
147 let num_centroids = centroids.nrows();
148 let num_tokens = query.nrows();
149
150 let batch_ranges: Vec<(usize, usize)> = (0..num_centroids)
152 .step_by(batch_size)
153 .map(|start| (start, (start + batch_size).min(num_centroids)))
154 .collect();
155
156 let local_results: Vec<ProbePartial> = batch_ranges
163 .par_iter()
164 .map(|&(batch_start, batch_end)| {
165 let mut heaps: Vec<BinaryHeap<(Reverse<OrdF32>, usize)>> = (0..num_tokens)
166 .map(|_| BinaryHeap::with_capacity(n_probe + 1))
167 .collect();
168 let mut max_scores: HashMap<usize, f32> = HashMap::new();
169
170 let batch_centroids = centroids.slice_rows(batch_start, batch_end);
172
173 let batch_scores = query.dot(&batch_centroids.t());
175
176 for (q_idx, heap) in heaps.iter_mut().enumerate() {
178 for (local_c, &score) in batch_scores.row(q_idx).iter().enumerate() {
179 let global_c = batch_start + local_c;
180 let entry = (Reverse(OrdF32(score)), global_c);
181
182 if heap.len() < n_probe {
183 heap.push(entry);
184 max_scores
185 .entry(global_c)
186 .and_modify(|s| *s = max_score(*s, score))
187 .or_insert(score);
188 } else if let Some(&(Reverse(OrdF32(min_score)), _)) = heap.peek() {
189 if is_score_better(score, min_score) {
190 heap.pop();
191 heap.push(entry);
192 max_scores
193 .entry(global_c)
194 .and_modify(|s| *s = max_score(*s, score))
195 .or_insert(score);
196 }
197 }
198 }
199 }
200
201 (heaps, max_scores)
202 })
203 .collect();
204
205 let mut final_heaps: Vec<BinaryHeap<(Reverse<OrdF32>, usize)>> = (0..num_tokens)
208 .map(|_| BinaryHeap::with_capacity(n_probe + 1))
209 .collect();
210 let mut final_max_scores: HashMap<usize, f32> = HashMap::new();
211
212 for (local_heaps, local_max_scores) in local_results {
213 for (q_idx, local_heap) in local_heaps.into_iter().enumerate() {
214 for entry in local_heap {
215 let (Reverse(OrdF32(score)), _) = entry;
216 if final_heaps[q_idx].len() < n_probe {
217 final_heaps[q_idx].push(entry);
218 } else if let Some(&(Reverse(OrdF32(min_score)), _)) = final_heaps[q_idx].peek() {
219 if is_score_better(score, min_score) {
220 final_heaps[q_idx].pop();
221 final_heaps[q_idx].push(entry);
222 }
223 }
224 }
225 }
226 for (c, score) in local_max_scores {
227 final_max_scores
228 .entry(c)
229 .and_modify(|s| *s = s.max(score))
230 .or_insert(score);
231 }
232 }
233
234 let mut selected: HashSet<usize> = HashSet::new();
236 for heap in final_heaps {
237 for (_, c) in heap {
238 selected.insert(c);
239 }
240 }
241
242 if let Some(threshold) = centroid_score_threshold {
244 selected.retain(|c| {
245 final_max_scores
246 .get(c)
247 .copied()
248 .unwrap_or(f32::NEG_INFINITY)
249 >= threshold
250 });
251 }
252
253 selected.into_iter().collect()
254}
255
256fn build_sparse_centroid_scores(
260 query: &Array2<f32>,
261 centroids: &CentroidStore,
262 centroid_ids: &HashSet<usize>,
263) -> HashMap<usize, Array1<f32>> {
264 centroid_ids
265 .iter()
266 .map(|&c| {
267 let centroid = centroids.row(c);
268 let scores: Array1<f32> = query.dot(¢roid);
269 (c, scores)
270 })
271 .collect()
272}
273
274fn approximate_score_sparse(
276 sparse_scores: &HashMap<usize, Array1<f32>>,
277 doc_codes: &[usize],
278 num_query_tokens: usize,
279) -> f32 {
280 let mut score = 0.0;
281
282 for q_idx in 0..num_query_tokens {
284 let mut max_score = f32::NEG_INFINITY;
285
286 for &code in doc_codes.iter() {
288 if let Some(centroid_scores) = sparse_scores.get(&code) {
289 let centroid_score = centroid_scores[q_idx];
290 if centroid_score > max_score {
291 max_score = centroid_score;
292 }
293 }
294 }
295
296 if max_score > f32::NEG_INFINITY {
297 score += max_score;
298 }
299 }
300
301 score
302}
303
304fn approximate_score_mmap(query_centroid_scores: &Array2<f32>, doc_codes: &[i64]) -> f32 {
306 let mut score = 0.0;
307
308 for q_idx in 0..query_centroid_scores.nrows() {
309 let mut max_score = f32::NEG_INFINITY;
310
311 for &code in doc_codes.iter() {
312 let centroid_score = query_centroid_scores[[q_idx, code as usize]];
313 if centroid_score > max_score {
314 max_score = centroid_score;
315 }
316 }
317
318 if max_score > f32::NEG_INFINITY {
319 score += max_score;
320 }
321 }
322
323 score
324}
325
326pub fn search_one_mmap(
328 index: &crate::index::MmapIndex,
329 query: &Array2<f32>,
330 params: &SearchParameters,
331 subset: Option<&[i64]>,
332) -> Result<QueryResult> {
333 let num_centroids = index.codec.num_centroids();
334 let num_query_tokens = query.nrows();
335
336 let use_batched = params.centroid_batch_size > 0 && num_centroids > params.centroid_batch_size;
338
339 if use_batched {
340 return search_one_mmap_batched(index, query, params, subset);
342 }
343
344 let query_centroid_scores = query.dot(&index.codec.centroids_view().t());
346
347 let eligible_centroids: Option<HashSet<usize>> = subset.map(|subset_docs| {
351 let mut centroids = HashSet::new();
352 for &doc_id in subset_docs {
353 let doc_idx = doc_id as usize;
354 if doc_idx < index.doc_lengths.len() {
355 let start = index.doc_offsets[doc_idx];
356 let end = index.doc_offsets[doc_idx + 1];
357 let codes = index.mmap_codes.slice(start, end);
358 for &c in codes.iter() {
359 centroids.insert(c as usize);
360 }
361 }
362 }
363 centroids
364 });
365
366 let effective_n_ivf_probe = match (&eligible_centroids, subset) {
371 (Some(eligible), Some(subset_docs)) if !eligible.is_empty() => {
372 let num_docs = index.doc_lengths.len();
373 let subset_len = subset_docs.len();
374 let scaled = if subset_len > 0 {
375 (params.n_ivf_probe as u64 * num_docs as u64 / subset_len as u64) as usize
376 } else {
377 params.n_ivf_probe
378 };
379 scaled.max(params.n_ivf_probe).min(eligible.len())
380 }
381 _ => params.n_ivf_probe,
382 };
383
384 let cells_to_probe: Vec<usize> = {
389 let mut selected_centroids = HashSet::new();
390
391 for q_idx in 0..num_query_tokens {
392 let mut centroid_scores: Vec<(usize, f32)> = match &eligible_centroids {
393 Some(eligible) => eligible
394 .iter()
395 .map(|&c| (c, query_centroid_scores[[q_idx, c]]))
396 .collect(),
397 None => (0..num_centroids)
398 .map(|c| (c, query_centroid_scores[[q_idx, c]]))
399 .collect(),
400 };
401
402 let n_probe = effective_n_ivf_probe.min(centroid_scores.len());
406 if centroid_scores.len() > n_probe {
407 centroid_scores
408 .select_nth_unstable_by(n_probe - 1, |a, b| cmp_score_descending(a.1, b.1));
409 }
410
411 for (c, _) in centroid_scores.iter().take(n_probe) {
412 selected_centroids.insert(*c);
413 }
414 }
415
416 if let Some(threshold) = params.centroid_score_threshold {
418 selected_centroids.retain(|&c| {
419 let max_score: f32 = (0..num_query_tokens)
420 .map(|q_idx| query_centroid_scores[[q_idx, c]])
421 .max_by(|a, b| cmp_score_ascending(*a, *b))
422 .unwrap_or(f32::NEG_INFINITY);
423 max_score >= threshold
424 });
425 }
426
427 selected_centroids.into_iter().collect()
428 };
429
430 let mut candidates = index.get_candidates(&cells_to_probe);
432
433 if let Some(subset_docs) = subset {
435 let subset_set: HashSet<i64> = subset_docs.iter().copied().collect();
436 candidates.retain(|&c| subset_set.contains(&c));
437 }
438
439 if candidates.is_empty() {
440 return Ok(QueryResult {
441 query_id: 0,
442 passage_ids: vec![],
443 scores: vec![],
444 });
445 }
446
447 let mut approx_scores: Vec<(i64, f32)> = candidates
449 .par_iter()
450 .map(|&doc_id| {
451 let start = index.doc_offsets[doc_id as usize];
452 let end = index.doc_offsets[doc_id as usize + 1];
453 let codes = index.mmap_codes.slice(start, end);
454 let score = approximate_score_mmap(&query_centroid_scores, &codes);
455 (doc_id, score)
456 })
457 .collect();
458
459 approx_scores.sort_by(|a, b| cmp_score_descending(a.1, b.1));
461 let top_candidates: Vec<i64> = approx_scores
462 .iter()
463 .take(params.n_full_scores)
464 .map(|(id, _)| *id)
465 .collect();
466
467 let n_decompress = (params.n_full_scores / 4).max(params.top_k);
469 let to_decompress: Vec<i64> = top_candidates.into_iter().take(n_decompress).collect();
470
471 if to_decompress.is_empty() {
472 return Ok(QueryResult {
473 query_id: 0,
474 passage_ids: vec![],
475 scores: vec![],
476 });
477 }
478
479 let mut exact_scores: Vec<(i64, f32)> = to_decompress
482 .par_chunks(DECOMPRESS_CHUNK_SIZE)
483 .flat_map(|chunk| {
484 chunk
485 .iter()
486 .filter_map(|&doc_id| {
487 let doc_embeddings = index.get_document_embeddings(doc_id as usize).ok()?;
488 let score = colbert_score(&query.view(), &doc_embeddings.view());
489 Some((doc_id, score))
490 })
491 .collect::<Vec<_>>()
492 })
493 .collect();
494
495 exact_scores.sort_by(|a, b| cmp_score_descending(a.1, b.1));
497
498 let result_count = params.top_k.min(exact_scores.len());
500 let passage_ids: Vec<i64> = exact_scores
501 .iter()
502 .take(result_count)
503 .map(|(id, _)| *id)
504 .collect();
505 let scores: Vec<f32> = exact_scores
506 .iter()
507 .take(result_count)
508 .map(|(_, s)| *s)
509 .collect();
510
511 Ok(QueryResult {
512 query_id: 0,
513 passage_ids,
514 scores,
515 })
516}
517
518fn search_one_mmap_batched(
522 index: &crate::index::MmapIndex,
523 query: &Array2<f32>,
524 params: &SearchParameters,
525 subset: Option<&[i64]>,
526) -> Result<QueryResult> {
527 let num_query_tokens = query.nrows();
528
529 let cells_to_probe = ivf_probe_batched(
531 query,
532 &index.codec.centroids,
533 params.n_ivf_probe,
534 params.centroid_batch_size,
535 params.centroid_score_threshold,
536 );
537
538 let mut candidates = index.get_candidates(&cells_to_probe);
540
541 if let Some(subset_docs) = subset {
543 let subset_set: HashSet<i64> = subset_docs.iter().copied().collect();
544 candidates.retain(|&c| subset_set.contains(&c));
545 }
546
547 if candidates.is_empty() {
548 return Ok(QueryResult {
549 query_id: 0,
550 passage_ids: vec![],
551 scores: vec![],
552 });
553 }
554
555 let mut unique_centroids: HashSet<usize> = HashSet::new();
557 for &doc_id in &candidates {
558 let start = index.doc_offsets[doc_id as usize];
559 let end = index.doc_offsets[doc_id as usize + 1];
560 let codes = index.mmap_codes.slice(start, end);
561 for &code in codes.iter() {
562 unique_centroids.insert(code as usize);
563 }
564 }
565
566 let sparse_scores =
568 build_sparse_centroid_scores(query, &index.codec.centroids, &unique_centroids);
569
570 let mut approx_scores: Vec<(i64, f32)> = candidates
572 .par_iter()
573 .map(|&doc_id| {
574 let start = index.doc_offsets[doc_id as usize];
575 let end = index.doc_offsets[doc_id as usize + 1];
576 let codes = index.mmap_codes.slice(start, end);
577 let doc_codes: Vec<usize> = codes.iter().map(|&c| c as usize).collect();
578 let score = approximate_score_sparse(&sparse_scores, &doc_codes, num_query_tokens);
579 (doc_id, score)
580 })
581 .collect();
582
583 approx_scores.sort_by(|a, b| cmp_score_descending(a.1, b.1));
585 let top_candidates: Vec<i64> = approx_scores
586 .iter()
587 .take(params.n_full_scores)
588 .map(|(id, _)| *id)
589 .collect();
590
591 let n_decompress = (params.n_full_scores / 4).max(params.top_k);
593 let to_decompress: Vec<i64> = top_candidates.into_iter().take(n_decompress).collect();
594
595 if to_decompress.is_empty() {
596 return Ok(QueryResult {
597 query_id: 0,
598 passage_ids: vec![],
599 scores: vec![],
600 });
601 }
602
603 let mut exact_scores: Vec<(i64, f32)> = to_decompress
606 .par_chunks(DECOMPRESS_CHUNK_SIZE)
607 .flat_map(|chunk| {
608 chunk
609 .iter()
610 .filter_map(|&doc_id| {
611 let doc_embeddings = index.get_document_embeddings(doc_id as usize).ok()?;
612 let score = colbert_score(&query.view(), &doc_embeddings.view());
613 Some((doc_id, score))
614 })
615 .collect::<Vec<_>>()
616 })
617 .collect();
618
619 exact_scores.sort_by(|a, b| cmp_score_descending(a.1, b.1));
621
622 let result_count = params.top_k.min(exact_scores.len());
624 let passage_ids: Vec<i64> = exact_scores
625 .iter()
626 .take(result_count)
627 .map(|(id, _)| *id)
628 .collect();
629 let scores: Vec<f32> = exact_scores
630 .iter()
631 .take(result_count)
632 .map(|(_, s)| *s)
633 .collect();
634
635 Ok(QueryResult {
636 query_id: 0,
637 passage_ids,
638 scores,
639 })
640}
641
642pub fn search_many_mmap(
644 index: &crate::index::MmapIndex,
645 queries: &[Array2<f32>],
646 params: &SearchParameters,
647 parallel: bool,
648 subset: Option<&[i64]>,
649) -> Result<Vec<QueryResult>> {
650 if parallel {
651 let results: Vec<QueryResult> = queries
652 .par_iter()
653 .enumerate()
654 .map(|(i, query)| {
655 let mut result =
656 search_one_mmap(index, query, params, subset).unwrap_or_else(|_| QueryResult {
657 query_id: i,
658 passage_ids: vec![],
659 scores: vec![],
660 });
661 result.query_id = i;
662 result
663 })
664 .collect();
665 Ok(results)
666 } else {
667 let mut results = Vec::with_capacity(queries.len());
668 for (i, query) in queries.iter().enumerate() {
669 let mut result = search_one_mmap(index, query, params, subset)?;
670 result.query_id = i;
671 results.push(result);
672 }
673 Ok(results)
674 }
675}
676
677pub type SearchResult = QueryResult;
679
680#[cfg(test)]
681mod tests {
682 use super::*;
683
684 #[test]
685 fn test_colbert_score() {
686 let query =
688 Array2::from_shape_vec((2, 4), vec![1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0]).unwrap();
689
690 let doc = Array2::from_shape_vec(
692 (3, 4),
693 vec![
694 0.5, 0.5, 0.0, 0.0, 0.8, 0.2, 0.0, 0.0, 0.0, 0.9, 0.1, 0.0, ],
698 )
699 .unwrap();
700
701 let score = colbert_score(&query.view(), &doc.view());
702 assert!((score - 1.7).abs() < 1e-5);
705 }
706
707 #[test]
708 fn test_search_params_default() {
709 let params = SearchParameters::default();
710 assert_eq!(params.batch_size, 2000);
711 assert_eq!(params.n_full_scores, 4096);
712 assert_eq!(params.top_k, 10);
713 assert_eq!(params.n_ivf_probe, 8);
714 assert_eq!(params.centroid_score_threshold, Some(0.4));
715 }
716
717 #[test]
718 fn test_cmp_score_descending_places_non_finite_scores_last() {
719 let mut scores = [1.0f32, f32::INFINITY, 0.5, f32::NAN];
720 scores.sort_by(|a, b| cmp_score_descending(*a, *b));
721
722 assert_eq!(scores[0], 1.0);
723 assert_eq!(scores[1], 0.5);
724 assert!(!scores[2].is_finite());
725 assert!(!scores[3].is_finite());
726 }
727
728 #[test]
729 fn test_score_replacement_treats_finite_values_as_better_than_non_finite() {
730 assert!(is_score_better(1.0, f32::NAN));
731 assert!(is_score_better(1.0, f32::INFINITY));
732 assert!(!is_score_better(f32::NAN, 1.0));
733 assert!(!is_score_better(f32::INFINITY, 1.0));
734 }
735
736 #[test]
737 fn test_max_score_keeps_finite_value_over_non_finite_value() {
738 assert_eq!(max_score(f32::NAN, 1.0), 1.0);
739 assert_eq!(max_score(1.0, f32::NAN), 1.0);
740 assert_eq!(max_score(f32::INFINITY, 1.0), 1.0);
741 assert_eq!(max_score(1.0, f32::INFINITY), 1.0);
742 }
743}