1#[cfg(feature = "native")]
12use rayon::prelude::*;
13use rustc_hash::FxHashMap;
14
15pub(crate) struct ForwardIndex {
21 terms: Vec<u32>,
22 offsets: Vec<u32>,
23 pub num_terms: usize,
24}
25
26impl ForwardIndex {
27 #[inline]
28 pub fn num_docs(&self) -> usize {
29 if self.offsets.is_empty() {
30 0
31 } else {
32 self.offsets.len() - 1
33 }
34 }
35
36 #[inline]
37 fn doc_terms(&self, doc: usize) -> &[u32] {
38 let start = self.offsets[doc] as usize;
39 let end = self.offsets[doc + 1] as usize;
40 &self.terms[start..end]
41 }
42
43 pub fn total_postings(&self) -> u32 {
45 self.offsets.last().copied().unwrap_or(0)
46 }
47}
48
49pub(crate) fn build_vid_maps(bmp: &crate::segment::reader::bmp::BmpIndex) -> (Vec<u32>, Vec<u32>) {
59 let ids = bmp.doc_map_ids_slice();
60 let num_virtual = bmp.num_virtual_docs as usize;
61 let mut virtual_to_real = vec![u32::MAX; num_virtual];
62 let mut real_to_virtual = Vec::with_capacity(bmp.num_real_docs() as usize);
63 for (vid, (slot, chunk)) in virtual_to_real
64 .iter_mut()
65 .zip(ids.as_chunks::<4>().0)
66 .enumerate()
67 {
68 let doc_id = u32::from_le_bytes(*chunk);
69 if doc_id != u32::MAX {
70 *slot = real_to_virtual.len() as u32;
71 real_to_virtual.push(vid as u32);
72 }
73 }
74 if real_to_virtual.len() != bmp.num_real_docs() as usize {
75 log::warn!(
76 "[reorder] BMP doc map has {} real slots but footer says num_real_docs={} — trusting the doc map",
77 real_to_virtual.len(),
78 bmp.num_real_docs(),
79 );
80 }
81 (virtual_to_real, real_to_virtual)
82}
83
84pub(crate) fn build_forward_index_from_bmps(
98 bmps: &[&crate::segment::reader::bmp::BmpIndex],
99 min_doc_freq: usize,
100 max_doc_freq: usize,
101 memory_budget_bytes: usize,
102) -> (ForwardIndex, Vec<usize>) {
103 let vid_maps: Vec<(Vec<u32>, Vec<u32>)> = bmps.iter().map(|b| build_vid_maps(b)).collect();
105 let source_doc_counts: Vec<usize> = vid_maps.iter().map(|(_, r2v)| r2v.len()).collect();
106 let total_docs: usize = source_doc_counts.iter().sum();
107
108 if total_docs == 0 {
109 return (
110 ForwardIndex {
111 terms: Vec::new(),
112 offsets: Vec::new(),
113 num_terms: 0,
114 },
115 source_doc_counts,
116 );
117 }
118
119 let mut dim_df: FxHashMap<u32, usize> = FxHashMap::default();
121 for (bmp, (v2r, _)) in bmps.iter().zip(&vid_maps) {
122 let num_blocks = bmp.num_blocks as usize;
123 let block_size = bmp.bmp_block_size as usize;
124 for block_id in 0..num_blocks {
125 for (dim_id, postings) in bmp.iter_block_terms(block_id as u32) {
126 for p in postings {
127 let vid = block_id * block_size + p.local_slot as usize;
128 if v2r[vid] != u32::MAX && p.impact > 0 {
129 *dim_df.entry(dim_id).or_insert(0) += 1;
130 }
131 }
132 }
133 }
134 }
135
136 let mut eligible: Vec<(u32, usize)> = dim_df
138 .iter()
139 .filter(|&(_, df)| *df >= min_doc_freq && *df <= max_doc_freq)
140 .map(|(&dim_id, &df)| (dim_id, df))
141 .collect();
142 drop(dim_df);
143
144 let total_postings_est: usize = eligible.iter().map(|(_, df)| *df).sum();
148 let estimated_bytes = total_postings_est * 4 + total_docs * 28 + eligible.len() * 8;
149
150 if estimated_bytes > memory_budget_bytes && !eligible.is_empty() {
151 eligible.sort_by_key(|&(_, df)| df);
154
155 let target_postings =
156 memory_budget_bytes.saturating_sub(total_docs * 28 + eligible.len() * 8) / 4;
157 let mut cum = 0usize;
158 let mut keep_count = 0;
159 for &(_, df) in &eligible {
160 if cum + df > target_postings {
161 break;
162 }
163 cum += df;
164 keep_count += 1;
165 }
166
167 let dropped = eligible.len() - keep_count;
168 eligible.truncate(keep_count);
169
170 log::warn!(
171 "[reorder] memory budget {:.0} MB: estimated {:.0} MB, dropped {} highest-df dims, keeping {} ({} postings)",
172 memory_budget_bytes as f64 / (1024.0 * 1024.0),
173 estimated_bytes as f64 / (1024.0 * 1024.0),
174 dropped,
175 keep_count,
176 cum,
177 );
178 }
179
180 let mut term_remap: FxHashMap<u32, u32> = FxHashMap::default();
181 for &(dim_id, _) in &eligible {
182 let compact_id = term_remap.len() as u32;
183 term_remap.insert(dim_id, compact_id);
184 }
185 let num_active_terms = term_remap.len();
186 drop(eligible);
187
188 let mut counts = vec![0u32; total_docs];
190 let mut real_offset = 0usize;
191
192 for (src_idx, bmp) in bmps.iter().enumerate() {
193 let (v2r, _) = &vid_maps[src_idx];
194 let num_blocks = bmp.num_blocks as usize;
195 let block_size = bmp.bmp_block_size as usize;
196 for block_id in 0..num_blocks {
197 for (dim_id, postings) in bmp.iter_block_terms(block_id as u32) {
198 if !term_remap.contains_key(&dim_id) {
199 continue;
200 }
201 for p in postings {
202 let vid = block_id * block_size + p.local_slot as usize;
203 let real = v2r[vid];
204 if real != u32::MAX && p.impact > 0 {
205 counts[real_offset + real as usize] += 1;
206 }
207 }
208 }
209 }
210 real_offset += source_doc_counts[src_idx];
211 }
212
213 let mut offsets = Vec::with_capacity(total_docs + 1);
215 offsets.push(0u32);
216 for &c in &counts {
217 offsets.push(offsets.last().unwrap() + c);
218 }
219 let total = *offsets.last().unwrap() as usize;
220
221 let mut terms = vec![0u32; total];
223 counts.fill(0);
224 real_offset = 0;
225
226 for (src_idx, bmp) in bmps.iter().enumerate() {
227 let (v2r, _) = &vid_maps[src_idx];
228 let num_blocks = bmp.num_blocks as usize;
229 let block_size = bmp.bmp_block_size as usize;
230 for block_id in 0..num_blocks {
231 for (dim_id, postings) in bmp.iter_block_terms(block_id as u32) {
232 let Some(&compact) = term_remap.get(&dim_id) else {
233 continue;
234 };
235 for p in postings {
236 let vid = block_id * block_size + p.local_slot as usize;
237 let real = v2r[vid];
238 if real != u32::MAX && p.impact > 0 {
239 let global_real = real_offset + real as usize;
240 let pos = offsets[global_real] as usize + counts[global_real] as usize;
241 terms[pos] = compact;
242 counts[global_real] += 1;
243 }
244 }
245 }
246 }
247 real_offset += source_doc_counts[src_idx];
248 }
249
250 (
251 ForwardIndex {
252 terms,
253 offsets,
254 num_terms: num_active_terms,
255 },
256 source_doc_counts,
257 )
258}
259
260#[derive(Clone, Copy, Debug, Default)]
268pub struct BpBudget {
269 pub min_partition_docs: Option<usize>,
274 pub time_budget: Option<std::time::Duration>,
278}
279
280impl BpBudget {
281 pub fn full() -> Self {
283 Self::default()
284 }
285}
286
287pub(crate) fn graph_bisection(
298 fwd: &ForwardIndex,
299 min_partition_size: usize,
300 max_iters: usize,
301 budget: BpBudget,
302) -> (Vec<u32>, bool) {
303 let n = fwd.num_docs();
304 if n == 0 {
305 return (Vec::new(), true);
306 }
307
308 let effective_min_partition = budget
309 .min_partition_docs
310 .unwrap_or(0)
311 .max(min_partition_size);
312
313 let mut docs: Vec<u32> = (0..n as u32).collect();
314 let depth = if effective_min_partition > 0 {
315 ((n as f64) / (effective_min_partition as f64))
316 .log2()
317 .ceil() as usize
318 } else {
319 0
320 };
321 let log_table = build_log_table(4096);
322
323 log::debug!(
324 "BP graph_bisection: n={}, min_partition={}, max_iters={}, depth=~{}, time_budget={:?}",
325 n,
326 effective_min_partition,
327 max_iters,
328 depth,
329 budget.time_budget,
330 );
331
332 #[cfg(feature = "native")]
333 let deadline = budget.time_budget.map(|d| std::time::Instant::now() + d);
334 #[cfg(not(feature = "native"))]
335 let deadline: Option<()> = None;
336 #[cfg(not(feature = "native"))]
337 let _ = deadline;
338
339 let exhausted = std::sync::atomic::AtomicBool::new(false);
340 #[cfg(feature = "native")]
341 bisect(
342 &mut docs,
343 fwd,
344 effective_min_partition,
345 max_iters,
346 &log_table,
347 deadline,
348 &exhausted,
349 );
350 #[cfg(not(feature = "native"))]
351 bisect(
352 &mut docs,
353 fwd,
354 effective_min_partition,
355 max_iters,
356 &log_table,
357 None,
358 &exhausted,
359 );
360
361 let converged = !exhausted.load(std::sync::atomic::Ordering::Relaxed);
362 if !converged {
363 log::info!(
364 "BP graph_bisection: wall-clock budget {:?} exhausted at n={} — emitting partial (still valid) permutation",
365 budget.time_budget,
366 n,
367 );
368 }
369 (docs, converged)
370}
371
372fn bisect(
381 docs: &mut [u32],
382 fwd: &ForwardIndex,
383 min_partition_size: usize,
384 max_iters: usize,
385 log_table: &[f32],
386 #[cfg(feature = "native")] deadline: Option<std::time::Instant>,
387 #[cfg(not(feature = "native"))] deadline: Option<()>,
388 exhausted: &std::sync::atomic::AtomicBool,
389) {
390 let n = docs.len();
391 if n <= min_partition_size {
392 return;
393 }
394 if exhausted.load(std::sync::atomic::Ordering::Relaxed) {
396 return;
397 }
398 #[cfg(feature = "native")]
399 if let Some(dl) = deadline
400 && std::time::Instant::now() >= dl
401 {
402 exhausted.store(true, std::sync::atomic::Ordering::Relaxed);
403 return;
404 }
405 #[cfg(not(feature = "native"))]
406 let _ = deadline;
407
408 let mid = n / 2;
409 let nt = fwd.num_terms;
410
411 let effective_iters = if n > 100_000 {
415 max_iters.min(12)
416 } else {
417 max_iters
418 };
419
420 let mut left_deg = vec![0u32; nt];
424 let mut right_deg = vec![0u32; nt];
425
426 for (i, &doc) in docs.iter().enumerate() {
427 let target = if i < mid {
428 &mut left_deg
429 } else {
430 &mut right_deg
431 };
432 for &term in fwd.doc_terms(doc as usize) {
433 target[term as usize] += 1;
434 }
435 }
436
437 let mut gains: Vec<f32> = vec![0.0; n];
439 let mut indices: Vec<usize> = (0..n).collect();
440 let mut new_left: Vec<u32> = Vec::with_capacity(mid);
441 let mut new_right: Vec<u32> = Vec::with_capacity(n - mid);
442
443 for iter in 0..effective_iters {
444 #[cfg(feature = "native")]
446 if let Some(dl) = deadline
447 && std::time::Instant::now() >= dl
448 {
449 exhausted.store(true, std::sync::atomic::Ordering::Relaxed);
450 break;
451 }
452 compute_gains(docs, fwd, mid, &left_deg, &right_deg, log_table, &mut gains);
455
456 indices.clear();
458 indices.extend(0..n);
459 indices.select_nth_unstable_by(mid, |&a, &b| {
460 gains[a]
461 .partial_cmp(&gains[b])
462 .unwrap_or(std::cmp::Ordering::Equal)
463 });
464
465 new_left.clear();
467 new_right.clear();
468 let mut swap_count: usize = 0;
469
470 for (rank, &idx) in indices.iter().enumerate() {
471 let doc = docs[idx];
472 let was_left = idx < mid;
473 let now_left = rank < mid;
474
475 if now_left {
476 new_left.push(doc);
477 } else {
478 new_right.push(doc);
479 }
480
481 if was_left != now_left {
482 swap_count += 1;
483 for &term in fwd.doc_terms(doc as usize) {
484 let t = term as usize;
485 if was_left {
486 left_deg[t] -= 1;
487 right_deg[t] += 1;
488 } else {
489 right_deg[t] -= 1;
490 left_deg[t] += 1;
491 }
492 }
493 }
494 }
495
496 docs[..mid].copy_from_slice(&new_left);
497 docs[mid..].copy_from_slice(&new_right);
498
499 if swap_count == 0 {
500 break;
501 }
502
503 if iter > 2 && swap_count < n / 200 {
505 break;
506 }
507
508 if iter > 5 {
510 let max_gain = gains.iter().copied().fold(f32::NEG_INFINITY, f32::max);
511 if max_gain.abs() < 0.001 {
512 break;
513 }
514 }
515 }
516
517 drop(left_deg);
519 drop(right_deg);
520 drop(gains);
521 drop(indices);
522 drop(new_left);
523 drop(new_right);
524
525 let (left, right) = docs.split_at_mut(mid);
526 #[cfg(feature = "native")]
527 rayon::join(
528 || {
529 bisect(
530 left,
531 fwd,
532 min_partition_size,
533 max_iters,
534 log_table,
535 deadline,
536 exhausted,
537 )
538 },
539 || {
540 bisect(
541 right,
542 fwd,
543 min_partition_size,
544 max_iters,
545 log_table,
546 deadline,
547 exhausted,
548 )
549 },
550 );
551 #[cfg(not(feature = "native"))]
552 {
553 bisect(
554 left,
555 fwd,
556 min_partition_size,
557 max_iters,
558 log_table,
559 deadline,
560 exhausted,
561 );
562 bisect(
563 right,
564 fwd,
565 min_partition_size,
566 max_iters,
567 log_table,
568 deadline,
569 exhausted,
570 );
571 }
572}
573
574#[inline(never)]
580fn compute_gains(
581 docs: &[u32],
582 fwd: &ForwardIndex,
583 mid: usize,
584 left_deg: &[u32],
585 right_deg: &[u32],
586 log_table: &[f32],
587 gains: &mut [f32],
588) {
589 let gain_for_doc = |i: usize| -> f32 {
598 let doc = docs[i] as usize;
599 let in_left = i < mid;
600 let mut g = 0.0f32;
601 for &term in fwd.doc_terms(doc) {
602 let t = term as usize;
603 let (from, to) = if in_left {
604 (left_deg[t], right_deg[t])
605 } else {
606 (right_deg[t], left_deg[t])
607 };
608 let move_gain = fast_log2_lookup(to as usize + 2, log_table)
609 - fast_log2_lookup(from as usize, log_table)
610 - std::f32::consts::LOG2_E / (1.0 + to as f32);
611 g += if in_left { move_gain } else { -move_gain };
612 }
613 g
614 };
615
616 #[cfg(feature = "native")]
617 {
618 if docs.len() > 4096 {
619 gains
620 .par_iter_mut()
621 .enumerate()
622 .for_each(|(i, gain)| *gain = gain_for_doc(i));
623 } else {
624 for (i, gain) in gains.iter_mut().enumerate().take(docs.len()) {
625 *gain = gain_for_doc(i);
626 }
627 }
628 }
629 #[cfg(not(feature = "native"))]
630 {
631 for (i, gain) in gains.iter_mut().enumerate().take(docs.len()) {
632 *gain = gain_for_doc(i);
633 }
634 }
635}
636
637fn build_log_table(size: usize) -> Vec<f32> {
641 let mut table = vec![0.0f32; size];
642 table[0] = -10.0;
644 for (i, entry) in table.iter_mut().enumerate().skip(1) {
645 *entry = (i as f32).log2();
646 }
647 table
648}
649
650#[inline]
652fn fast_log2_lookup(val: usize, table: &[f32]) -> f32 {
653 if val < table.len() {
654 table[val]
655 } else {
656 (val as f32).log2()
657 }
658}
659
660#[cfg(test)]
661mod tests {
662 use super::*;
663
664 fn make_fwd(docs: &[&[u32]], num_terms: usize) -> ForwardIndex {
666 let mut terms = Vec::new();
667 let mut offsets = vec![0u32];
668 for doc_terms in docs {
669 terms.extend_from_slice(doc_terms);
670 offsets.push(terms.len() as u32);
671 }
672 ForwardIndex {
673 terms,
674 offsets,
675 num_terms,
676 }
677 }
678
679 #[test]
680 fn test_bp_empty() {
681 let fwd = ForwardIndex {
682 terms: Vec::new(),
683 offsets: Vec::new(),
684 num_terms: 0,
685 };
686 let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
687 assert!(perm.is_empty());
688 }
689
690 #[test]
691 fn test_bp_small() {
692 let fwd = make_fwd(&[&[0, 1], &[0, 2], &[1, 3], &[2, 3]], 4);
694 let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
695 assert_eq!(perm.len(), 4);
696 let mut sorted = perm.clone();
698 sorted.sort();
699 assert_eq!(sorted, vec![0, 1, 2, 3]);
700 }
701
702 #[test]
703 fn test_bp_clusters() {
704 let fwd = make_fwd(
708 &[
709 &[0, 1],
710 &[0, 1],
711 &[0, 1],
712 &[0, 1],
713 &[2, 3],
714 &[2, 3],
715 &[2, 3],
716 &[2, 3],
717 ],
718 4,
719 );
720 let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
721 assert_eq!(perm.len(), 8);
722
723 let left: Vec<u32> = perm[..4].to_vec();
725
726 let a_in_left = left.iter().filter(|&&d| d < 4).count();
728 let b_in_left = left.iter().filter(|&&d| d >= 4).count();
729 assert!(
730 (a_in_left == 4 && b_in_left == 0) || (a_in_left == 0 && b_in_left == 4),
731 "Clusters should be separated: a_left={}, b_left={}",
732 a_in_left,
733 b_in_left,
734 );
735 }
736
737 #[test]
738 fn test_bp_permutation_valid() {
739 let docs: Vec<Vec<u32>> = (0..16).map(|i| vec![i / 4, 10 + i / 2]).collect();
741 let doc_refs: Vec<&[u32]> = docs.iter().map(|v| v.as_slice()).collect();
742 let fwd = make_fwd(&doc_refs, 18); let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
744
745 assert_eq!(perm.len(), 16);
746 let mut sorted = perm.clone();
748 sorted.sort();
749 let expected: Vec<u32> = (0..16).collect();
750 assert_eq!(sorted, expected);
751 }
752
753 #[test]
758 fn test_bp_depth_cap_separates_clusters_and_converges() {
759 let fwd = make_fwd(
762 &[
763 &[0, 1],
764 &[0, 1],
765 &[0, 1],
766 &[2, 3],
767 &[0, 1],
768 &[2, 3],
769 &[2, 3],
770 &[2, 3],
771 ],
772 4,
773 );
774 let budget = BpBudget {
775 min_partition_docs: Some(4),
776 time_budget: None,
777 };
778 let (perm, converged) = graph_bisection(&fwd, 2, 20, budget);
779 assert!(converged, "depth cap must report converged");
780 assert_eq!(perm.len(), 8);
781 let mut sorted = perm.clone();
782 sorted.sort();
783 assert_eq!(
784 sorted,
785 (0..8).collect::<Vec<u32>>(),
786 "must stay a valid permutation"
787 );
788 let cluster_a = [0u32, 1, 2, 4];
791 let a_in_left = perm[..4].iter().filter(|d| cluster_a.contains(d)).count();
792 assert!(
793 a_in_left == 4 || a_in_left == 0,
794 "clusters should separate at the top level: {:?}",
795 perm
796 );
797 }
798
799 #[test]
802 fn test_bp_zero_time_budget_emits_valid_partial_permutation() {
803 let docs: Vec<Vec<u32>> = (0..64).map(|i| vec![i % 4]).collect();
804 let doc_refs: Vec<&[u32]> = docs.iter().map(|v| v.as_slice()).collect();
805 let fwd = make_fwd(&doc_refs, 4);
806 let budget = BpBudget {
807 min_partition_docs: None,
808 time_budget: Some(std::time::Duration::ZERO),
809 };
810 let (perm, converged) = graph_bisection(&fwd, 4, 20, budget);
811 assert!(!converged, "zero budget must report unconverged");
812 assert_eq!(perm.len(), 64);
813 let mut sorted = perm.clone();
814 sorted.sort();
815 assert_eq!(sorted, (0..64).collect::<Vec<u32>>());
816 }
817
818 #[test]
819 fn test_fast_log2() {
820 let table = build_log_table(4096);
821 assert!((table[1] - 0.0).abs() < 0.001);
822 assert!((table[2] - 1.0).abs() < 0.001);
823 assert!((table[4] - 2.0).abs() < 0.001);
824 assert!((table[1024] - 10.0).abs() < 0.001);
825 let val = fast_log2_lookup(8192, &table);
827 assert!((val - 13.0).abs() < 0.001);
828 }
829}