1#[cfg(feature = "native")]
13use rayon::prelude::*;
14const TERM_DEGREE_VALUE_BYTES: usize = std::mem::size_of::<[u32; 2]>();
15const CANDIDATE_ENTRY_BYTES: usize = std::mem::size_of::<(usize, u32)>();
16const PARALLEL_BP_MIN_ENTITIES: usize = 1_048_576;
20const MIN_RELATIVE_OBJECTIVE_IMPROVEMENT: f64 = 1e-6;
21const MIN_OBJECTIVE_ITERATIONS: usize = 4;
22const OBJECTIVE_STALL_ITERATIONS: usize = 2;
23
24fn term_degree_bytes(num_terms: usize) -> usize {
25 let bitmap_words = num_terms.div_ceil(64);
26 num_terms
27 .saturating_mul(TERM_DEGREE_VALUE_BYTES)
28 .saturating_add(bitmap_words.saturating_mul(std::mem::size_of::<u64>()))
29 .saturating_add(bitmap_words.saturating_mul(std::mem::size_of::<u32>()))
33}
34
35#[cfg(feature = "native")]
40trait FrequencyParallelSafe: Sync {}
41#[cfg(feature = "native")]
42impl<T: Sync + ?Sized> FrequencyParallelSafe for T {}
43
44#[cfg(not(feature = "native"))]
45trait FrequencyParallelSafe {}
46#[cfg(not(feature = "native"))]
47impl<T: ?Sized> FrequencyParallelSafe for T {}
48
49fn count_frequencies_bounded<T: FrequencyParallelSafe>(
50 items: &[T],
51 num_terms: usize,
52 available_bytes: usize,
53 count_item: impl Fn(&T, &mut [u32]) + FrequencyParallelSafe,
54) -> Option<Vec<u32>> {
55 if num_terms == 0 {
56 return Some(Vec::new());
57 }
58 let table_bytes = num_terms
59 .checked_mul(std::mem::size_of::<u32>())?
60 .checked_add(std::mem::size_of::<Vec<u32>>())?;
61 let affordable_tables = available_bytes.checked_div(table_bytes)?;
62 if affordable_tables == 0 {
63 return None;
64 }
65
66 #[cfg(feature = "native")]
67 {
68 let lanes = affordable_tables
69 .min(rayon::current_num_threads().max(1))
70 .min(items.len().max(1));
71 if lanes > 1 {
72 let chunk_len = items.len().div_ceil(lanes);
73 return items
74 .par_chunks(chunk_len)
75 .map(|chunk| {
76 let mut counts = vec![0u32; num_terms];
77 for item in chunk {
78 count_item(item, &mut counts);
79 }
80 counts
81 })
82 .reduce_with(|mut left, right| {
83 for (total, count) in left.iter_mut().zip(right) {
84 *total = total.saturating_add(count);
85 }
86 left
87 });
88 }
89 }
90
91 let mut counts = vec![0u32; num_terms];
92 for item in items {
93 count_item(item, &mut counts);
94 }
95 Some(counts)
96}
97
98fn select_frequency_candidates(
101 frequencies: &[u32],
102 min_frequency: usize,
103 max_frequency: usize,
104 candidate_budget_bytes: usize,
105) -> (Vec<(u32, usize)>, bool) {
106 let eligible_count = frequencies
107 .iter()
108 .filter(|&&frequency| {
109 let frequency = frequency as usize;
110 frequency >= min_frequency && frequency <= max_frequency
111 })
112 .count();
113 let capacity = candidate_budget_bytes
114 .checked_div(CANDIDATE_ENTRY_BYTES)
115 .unwrap_or(0)
116 .min(eligible_count);
117 let mut candidates = std::collections::BinaryHeap::with_capacity(capacity);
118 for (term_id, &frequency) in frequencies.iter().enumerate() {
119 let frequency = frequency as usize;
120 if frequency < min_frequency || frequency > max_frequency {
121 continue;
122 }
123 let candidate = (frequency, term_id as u32);
124 if candidates.len() < capacity {
125 candidates.push(candidate);
126 } else if capacity > 0 && candidate < *candidates.peek().unwrap() {
127 candidates.pop();
128 candidates.push(candidate);
129 }
130 }
131 let selected = candidates
132 .into_vec()
133 .into_iter()
134 .map(|(frequency, term_id)| (term_id, frequency))
135 .collect();
136 (selected, capacity < eligible_count)
137}
138
139struct CandidateFit {
140 estimated_bytes: usize,
141 retained_postings: usize,
142 dropped: usize,
143}
144
145fn fit_candidates_to_budget(
149 candidates: &mut Vec<(u32, usize)>,
150 fixed_bytes: usize,
151 memory_budget_bytes: usize,
152) -> CandidateFit {
153 let total_postings = candidates.iter().fold(0usize, |total, (_, frequency)| {
154 total.saturating_add(*frequency)
155 });
156 let estimated_bytes = total_postings
157 .saturating_mul(std::mem::size_of::<u32>())
158 .saturating_add(fixed_bytes)
159 .saturating_add(candidates.len().saturating_mul(CANDIDATE_ENTRY_BYTES))
160 .saturating_add(term_degree_bytes(candidates.len()));
161 if estimated_bytes <= memory_budget_bytes || candidates.is_empty() {
162 return CandidateFit {
163 estimated_bytes,
164 retained_postings: total_postings,
165 dropped: 0,
166 };
167 }
168
169 candidates.sort_by_key(|&(_, frequency)| frequency);
170 let mut used_bytes = fixed_bytes;
171 let mut retained_postings = 0usize;
172 let mut keep = 0usize;
173 for &(_, frequency) in candidates.iter() {
174 let term_bytes = frequency
175 .saturating_mul(std::mem::size_of::<u32>())
176 .saturating_add(TERM_DEGREE_VALUE_BYTES + 1)
177 .saturating_add(CANDIDATE_ENTRY_BYTES);
178 if term_bytes > memory_budget_bytes.saturating_sub(used_bytes) {
179 break;
180 }
181 used_bytes = used_bytes.saturating_add(term_bytes);
182 retained_postings = retained_postings.saturating_add(frequency);
183 keep += 1;
184 }
185 let dropped = candidates.len() - keep;
186 candidates.truncate(keep);
187 candidates.shrink_to_fit();
190 CandidateFit {
191 estimated_bytes,
192 retained_postings,
193 dropped,
194 }
195}
196
197fn parallel_bisect_lanes(
198 memory_budget_bytes: usize,
199 non_degree_bytes: usize,
200 num_terms: usize,
201) -> usize {
202 let per_node = term_degree_bytes(num_terms).max(1);
203 let affordable_nodes = memory_budget_bytes
204 .saturating_sub(non_degree_bytes)
205 .checked_div(per_node)
206 .unwrap_or(0)
207 .max(1);
208 #[cfg(feature = "native")]
209 let worker_limit = rayon::current_num_threads().max(1);
210 #[cfg(not(feature = "native"))]
211 let worker_limit = 1usize;
212 affordable_nodes.min(worker_limit)
213}
214
215struct TermDegrees {
223 values: Vec<std::mem::MaybeUninit<[u32; 2]>>,
224 initialized: Vec<u64>,
225 touched_words: Vec<u32>,
226}
227
228impl TermDegrees {
229 fn new(num_terms: usize) -> Self {
230 let bitmap_words = num_terms.div_ceil(64);
231 let mut values = Vec::with_capacity(num_terms);
232 values.resize_with(num_terms, std::mem::MaybeUninit::uninit);
233 Self {
234 values,
235 initialized: vec![0; bitmap_words],
236 touched_words: Vec::with_capacity(bitmap_words),
237 }
238 }
239
240 fn reset(&mut self) {
243 for word in self.touched_words.drain(..) {
244 self.initialized[word as usize] = 0;
245 }
246 }
247
248 fn sort_touched_words(&mut self) {
249 self.touched_words.sort_unstable();
250 }
251
252 #[inline]
253 fn entry_mut(&mut self, term: usize) -> &mut [u32; 2] {
254 let word = term / 64;
255 let mask = 1u64 << (term % 64);
256 if self.initialized[word] & mask == 0 {
257 if self.initialized[word] == 0 {
258 self.touched_words.push(word as u32);
259 }
260 self.values[term].write([0, 0]);
261 self.initialized[word] |= mask;
262 }
263 unsafe { self.values[term].assume_init_mut() }
265 }
266
267 #[inline]
268 fn get(&self, term: usize) -> [u32; 2] {
269 let word = term / 64;
270 let mask = 1u64 << (term % 64);
271 if self.initialized[word] & mask == 0 {
272 return [0, 0];
273 }
274 unsafe { *self.values[term].assume_init_ref() }
277 }
278
279 fn merge_from(&mut self, other: &Self) {
280 for &word_idx in &other.touched_words {
281 let word_idx = word_idx as usize;
282 let mut pending = other.initialized[word_idx];
283 while pending != 0 {
284 let bit = pending.trailing_zeros() as usize;
285 let term = word_idx * 64 + bit;
286 let [left, right] = unsafe { *other.values[term].assume_init_ref() };
288 let entry = self.entry_mut(term);
289 entry[0] += left;
290 entry[1] += right;
291 pending &= pending - 1;
292 }
293 }
294 }
295
296 fn apply_moves_to(&self, degrees: &mut Self) {
302 for &word_idx in &self.touched_words {
303 let word_idx = word_idx as usize;
304 let mut pending = self.initialized[word_idx];
305 while pending != 0 {
306 let bit = pending.trailing_zeros() as usize;
307 let term = word_idx * 64 + bit;
308 let [right_to_left, left_to_right] =
310 unsafe { *self.values[term].assume_init_ref() };
311 debug_assert!(
312 degrees.initialized[word_idx] & (1u64 << bit) != 0,
313 "a moved term must already exist in the partition degrees"
314 );
315 let degree = degrees.entry_mut(term);
316 let new_left =
317 i64::from(degree[0]) + i64::from(right_to_left) - i64::from(left_to_right);
318 let new_right =
319 i64::from(degree[1]) + i64::from(left_to_right) - i64::from(right_to_left);
320 debug_assert!(new_left >= 0 && new_right >= 0);
321 debug_assert!(new_left <= i64::from(u32::MAX));
322 debug_assert!(new_right <= i64::from(u32::MAX));
323 degree[0] = new_left as u32;
324 degree[1] = new_right as u32;
325 pending &= pending - 1;
326 }
327 }
328 }
329
330 fn bisection_objective(&self, left_size: usize, right_size: usize, log_table: &[f32]) -> f64 {
337 let mut objective = 0.0f64;
338 let side_log = [
339 fast_log2_lookup(left_size, log_table) as f64,
340 fast_log2_lookup(right_size, log_table) as f64,
341 ];
342 for &word_idx in &self.touched_words {
343 let word_idx = word_idx as usize;
344 let mut pending = self.initialized[word_idx];
345 while pending != 0 {
346 let bit = pending.trailing_zeros() as usize;
347 let term = word_idx * 64 + bit;
348 let [left, right] = unsafe { *self.values[term].assume_init_ref() };
350 for (side, count) in [left, right].into_iter().enumerate() {
351 if count > 0 {
352 objective += count as f64
353 * (fast_log2_lookup(count as usize + 1, log_table) as f64
354 - side_log[side]);
355 }
356 }
357 pending &= pending - 1;
358 }
359 }
360 objective
361 }
362}
363
364pub(crate) struct ForwardIndex {
370 terms: Vec<u32>,
371 offsets: Vec<u64>,
376 pub num_terms: usize,
377 parallel_bisect_lanes: usize,
381 budget_limited: bool,
385}
386
387fn build_csr_offsets(counts: &[u32]) -> Vec<u64> {
390 let mut offsets = Vec::with_capacity(counts.len() + 1);
391 offsets.push(0u64);
392 for &c in counts {
393 offsets.push(offsets.last().unwrap() + c as u64);
394 }
395 offsets
396}
397
398impl ForwardIndex {
399 #[inline]
400 pub fn num_docs(&self) -> usize {
401 if self.offsets.is_empty() {
402 0
403 } else {
404 self.offsets.len() - 1
405 }
406 }
407
408 #[inline]
409 fn doc_terms(&self, doc: usize) -> &[u32] {
410 let start = self.offsets[doc] as usize;
411 let end = self.offsets[doc + 1] as usize;
412 &self.terms[start..end]
413 }
414
415 pub fn total_postings(&self) -> u64 {
417 self.offsets.last().copied().unwrap_or(0)
418 }
419
420 #[inline]
421 pub fn budget_limited(&self) -> bool {
422 self.budget_limited
423 }
424}
425
426pub(crate) fn build_vid_maps(
436 bmp: &crate::segment::reader::bmp::BmpIndex,
437) -> crate::Result<(Vec<u32>, Vec<u32>)> {
438 let ids = bmp.doc_map_ids_slice();
439 let num_virtual = bmp.num_virtual_docs as usize;
440 let expected_real = bmp.num_real_docs() as usize;
441 let mut virtual_to_real = vec![u32::MAX; num_virtual];
442 let mut real_to_virtual = Vec::with_capacity(expected_real);
443 debug_assert_eq!(ids.len(), virtual_to_real.len() * 4);
444 bmp.visit_real_slots_for_rewrite(|vid| {
445 virtual_to_real[vid] = real_to_virtual.len() as u32;
446 real_to_virtual.push(vid as u32);
447 })?;
448 Ok((virtual_to_real, real_to_virtual))
449}
450
451struct BlockJob {
457 src: u32,
458 block_id: u32,
459 real_start: u32,
461 real_len: u32,
463}
464
465fn build_block_jobs(
468 bmps: &[&crate::segment::reader::bmp::BmpIndex],
469 vid_maps: &[(Vec<u32>, Vec<u32>)],
470) -> Vec<BlockJob> {
471 let total_blocks: usize = bmps.iter().map(|b| b.num_blocks as usize).sum();
472 let mut jobs = Vec::with_capacity(total_blocks);
473 for (src, (bmp, (v2r, _))) in bmps.iter().zip(vid_maps).enumerate() {
474 let block_size = bmp.bmp_block_size as usize;
475 let mut real_cursor = 0u32;
476 for block_id in 0..bmp.num_blocks as usize {
477 let vid_start = block_id * block_size;
478 let vid_end = ((block_id + 1) * block_size).min(v2r.len());
479 let real_len = v2r[vid_start..vid_end]
480 .iter()
481 .filter(|&&r| r != u32::MAX)
482 .count() as u32;
483 jobs.push(BlockJob {
484 src: src as u32,
485 block_id: block_id as u32,
486 real_start: real_cursor,
487 real_len,
488 });
489 real_cursor += real_len;
490 }
491 }
492 jobs
493}
494
495#[cfg(test)]
509pub(crate) fn build_forward_index_from_bmps(
510 bmps: &[&crate::segment::reader::bmp::BmpIndex],
511 min_doc_freq: usize,
512 max_doc_freq: usize,
513 memory_budget_bytes: usize,
514) -> crate::Result<ForwardIndex> {
515 let vid_maps: Vec<(Vec<u32>, Vec<u32>)> = bmps
516 .iter()
517 .map(|bmp| build_vid_maps(bmp))
518 .collect::<crate::Result<_>>()?;
519 Ok(build_forward_index_from_bmps_with_maps(
520 bmps,
521 &vid_maps,
522 min_doc_freq,
523 max_doc_freq,
524 memory_budget_bytes,
525 ))
526}
527
528pub(crate) fn build_forward_index_from_bmps_with_maps(
532 bmps: &[&crate::segment::reader::bmp::BmpIndex],
533 vid_maps: &[(Vec<u32>, Vec<u32>)],
534 min_doc_freq: usize,
535 max_doc_freq: usize,
536 memory_budget_bytes: usize,
537) -> ForwardIndex {
538 debug_assert_eq!(bmps.len(), vid_maps.len());
539 let total_docs: usize = vid_maps.iter().map(|(_, r2v)| r2v.len()).sum();
540
541 if total_docs == 0 {
542 return ForwardIndex {
543 terms: Vec::new(),
544 offsets: Vec::new(),
545 num_terms: 0,
546 parallel_bisect_lanes: 1,
547 budget_limited: false,
548 };
549 }
550
551 let jobs = build_block_jobs(bmps, vid_maps);
556
557 let max_dims = bmps
561 .iter()
562 .map(|bmp| bmp.dims() as usize)
563 .max()
564 .unwrap_or(0);
565 let jobs_bytes = jobs
566 .len()
567 .saturating_mul(std::mem::size_of::<BlockJob>().saturating_add(40));
568 let frequency_bytes = max_dims.saturating_mul(std::mem::size_of::<u32>());
569 if frequency_bytes > memory_budget_bytes.saturating_sub(jobs_bytes) {
570 log::warn!(
571 "[reorder] memory budget {} cannot hold the {} dimension-frequency table; using identity order",
572 crate::format_bytes(memory_budget_bytes as u64),
573 crate::format_bytes(frequency_bytes as u64),
574 );
575 return ForwardIndex {
576 terms: Vec::new(),
577 offsets: Vec::new(),
578 num_terms: 0,
579 parallel_bisect_lanes: 1,
580 budget_limited: true,
581 };
582 }
583 let Some(dim_df) = count_frequencies_bounded(
584 &jobs,
585 max_dims,
586 memory_budget_bytes.saturating_sub(jobs_bytes),
587 |job, counts| {
588 let bmp = bmps[job.src as usize];
589 let (v2r, _) = &vid_maps[job.src as usize];
590 let block_size = bmp.bmp_block_size as usize;
591 for (dim_id, _, postings) in bmp.iter_block_terms(job.block_id) {
592 let mut frequency = 0u32;
593 for posting in postings {
594 let vid = job.block_id as usize * block_size + posting.local_slot as usize;
595 if v2r.get(vid).is_some_and(|&real| real != u32::MAX) && posting.impact > 0 {
596 frequency = frequency.saturating_add(1);
597 }
598 }
599 if frequency > 0
600 && let Some(total) = counts.get_mut(dim_id as usize)
601 {
602 *total = total.saturating_add(frequency);
603 }
604 }
605 },
606 ) else {
607 log::warn!(
608 "[reorder] memory budget {} cannot hold a bounded dimension-frequency table; using identity order",
609 crate::format_bytes(memory_budget_bytes as u64),
610 );
611 return ForwardIndex {
612 terms: Vec::new(),
613 offsets: Vec::new(),
614 num_terms: 0,
615 parallel_bisect_lanes: 1,
616 budget_limited: true,
617 };
618 };
619
620 let (mut eligible, mut budget_limited) = select_frequency_candidates(
624 &dim_df,
625 min_doc_freq,
626 max_doc_freq,
627 memory_budget_bytes
628 .saturating_sub(jobs_bytes)
629 .saturating_sub(frequency_bytes),
630 );
631 drop(dim_df);
632
633 let entity_scratch_bytes = total_docs.saturating_mul(32);
637 let remap_bytes = max_dims.saturating_mul(4);
638 let fixed_bytes = entity_scratch_bytes
639 .saturating_add(remap_bytes)
640 .saturating_add(jobs_bytes);
641 let fit = fit_candidates_to_budget(&mut eligible, fixed_bytes, memory_budget_bytes);
642 if fit.dropped > 0 {
643 budget_limited = true;
644 log::warn!(
645 "[reorder] memory budget {}: estimated {}, dropped {} highest-df dims, keeping {} ({} postings)",
646 crate::format_bytes(memory_budget_bytes as u64),
647 crate::format_bytes(fit.estimated_bytes as u64),
648 fit.dropped,
649 eligible.len(),
650 fit.retained_postings,
651 );
652 }
653
654 if eligible.is_empty() {
655 return ForwardIndex {
660 terms: Vec::new(),
661 offsets: Vec::new(),
662 num_terms: 0,
663 parallel_bisect_lanes: 1,
664 budget_limited,
665 };
666 }
667
668 let mut term_remap = vec![u32::MAX; max_dims];
669 for (compact_id, &(dim_id, _)) in eligible.iter().enumerate() {
670 term_remap[dim_id as usize] = compact_id as u32;
671 }
672 let num_active_terms = eligible.len();
673 let non_degree_bytes = entity_scratch_bytes.saturating_add(
678 fit.retained_postings
679 .saturating_mul(std::mem::size_of::<u32>()),
680 );
681 let parallel_bisect_lanes =
682 parallel_bisect_lanes(memory_budget_bytes, non_degree_bytes, num_active_terms);
683 drop(eligible);
684
685 let mut counts = vec![0u32; total_docs];
687 let fill_block_counts = |job: &BlockJob, out: &mut [u32]| {
688 let bmp = bmps[job.src as usize];
689 let (v2r, _) = &vid_maps[job.src as usize];
690 let block_size = bmp.bmp_block_size as usize;
691 for (dim_id, _, postings) in bmp.iter_block_terms(job.block_id) {
692 if term_remap.get(dim_id as usize).copied().unwrap_or(u32::MAX) == u32::MAX {
693 continue;
694 }
695 for p in postings {
696 let vid = job.block_id as usize * block_size + p.local_slot as usize;
697 let Some(&real) = v2r.get(vid) else {
698 continue;
699 };
700 if real != u32::MAX && p.impact > 0 {
701 out[(real - job.real_start) as usize] += 1;
702 }
703 }
704 }
705 };
706 {
707 let mut slices: Vec<(&BlockJob, &mut [u32])> = Vec::with_capacity(jobs.len());
708 let mut rest: &mut [u32] = &mut counts;
709 for job in &jobs {
710 let (head, tail) = rest.split_at_mut(job.real_len as usize);
711 slices.push((job, head));
712 rest = tail;
713 }
714 #[cfg(feature = "native")]
715 slices
716 .into_par_iter()
717 .for_each(|(job, out)| fill_block_counts(job, out));
718 #[cfg(not(feature = "native"))]
719 for (job, out) in slices {
720 fill_block_counts(job, out);
721 }
722 }
723
724 let offsets = build_csr_offsets(&counts);
726 let total = *offsets.last().unwrap() as usize;
727 drop(counts);
728
729 let mut terms = vec![0u32; total];
732 let fill_block_terms = |job: &BlockJob, global_real_start: usize, out: &mut [u32]| {
733 let bmp = bmps[job.src as usize];
734 let (v2r, _) = &vid_maps[job.src as usize];
735 let block_size = bmp.bmp_block_size as usize;
736 assert!(job.real_len as usize <= 256, "BMP block exceeds 256 docs");
738 let mut cursor = [0u32; 256];
739 let base = offsets[global_real_start] as usize;
740 for (dim_id, _, postings) in bmp.iter_block_terms(job.block_id) {
741 let compact = term_remap.get(dim_id as usize).copied().unwrap_or(u32::MAX);
742 if compact == u32::MAX {
743 continue;
744 }
745 for p in postings {
746 let vid = job.block_id as usize * block_size + p.local_slot as usize;
747 let Some(&real) = v2r.get(vid) else {
748 continue;
749 };
750 if real != u32::MAX && p.impact > 0 {
751 let local = (real - job.real_start) as usize;
752 let pos =
753 offsets[global_real_start + local] as usize - base + cursor[local] as usize;
754 out[pos] = compact;
755 cursor[local] += 1;
756 }
757 }
758 }
759 };
760 {
761 let mut slices: Vec<(&BlockJob, usize, &mut [u32])> = Vec::with_capacity(jobs.len());
762 let mut rest: &mut [u32] = &mut terms;
763 let mut global_real = 0usize;
764 for job in &jobs {
765 let len =
766 (offsets[global_real + job.real_len as usize] - offsets[global_real]) as usize;
767 let (head, tail) = rest.split_at_mut(len);
768 slices.push((job, global_real, head));
769 rest = tail;
770 global_real += job.real_len as usize;
771 }
772 #[cfg(feature = "native")]
773 slices
774 .into_par_iter()
775 .for_each(|(job, g, out)| fill_block_terms(job, g, out));
776 #[cfg(not(feature = "native"))]
777 for (job, g, out) in slices {
778 fill_block_terms(job, g, out);
779 }
780 }
781
782 ForwardIndex {
783 terms,
784 offsets,
785 num_terms: num_active_terms,
786 parallel_bisect_lanes,
787 budget_limited,
788 }
789}
790
791pub(crate) fn build_forward_index_from_blocks(
799 bmps: &[&crate::segment::reader::bmp::BmpIndex],
800 memory_budget_bytes: usize,
801) -> ForwardIndex {
802 let total_blocks: usize = bmps.iter().map(|b| b.num_blocks as usize).sum();
803 if total_blocks == 0 {
804 return ForwardIndex {
805 terms: Vec::new(),
806 offsets: Vec::new(),
807 num_terms: 0,
808 parallel_bisect_lanes: 1,
809 budget_limited: false,
810 };
811 }
812
813 let blocks: Vec<(u32, u32)> = bmps
815 .iter()
816 .enumerate()
817 .flat_map(|(src, bmp)| (0..bmp.num_blocks).map(move |b| (src as u32, b)))
818 .collect();
819
820 let max_dims = bmps
823 .iter()
824 .map(|bmp| bmp.dims() as usize)
825 .max()
826 .unwrap_or(0);
827 let blocks_bytes = blocks
828 .len()
829 .saturating_mul(std::mem::size_of::<(u32, u32)>().saturating_add(32));
830 let frequency_bytes = max_dims.saturating_mul(std::mem::size_of::<u32>());
831 if frequency_bytes > memory_budget_bytes.saturating_sub(blocks_bytes) {
832 log::warn!(
833 "[reorder] block-level frequency table exceeds memory budget; using identity order"
834 );
835 return ForwardIndex {
836 terms: Vec::new(),
837 offsets: Vec::new(),
838 num_terms: 0,
839 parallel_bisect_lanes: 1,
840 budget_limited: true,
841 };
842 }
843 let Some(dim_bf) = count_frequencies_bounded(
844 &blocks,
845 max_dims,
846 memory_budget_bytes.saturating_sub(blocks_bytes),
847 |&(src, block_id), counts| {
848 for (dim_id, _, _) in bmps[src as usize].iter_block_terms(block_id) {
849 if let Some(count) = counts.get_mut(dim_id as usize) {
850 *count = count.saturating_add(1);
851 }
852 }
853 },
854 ) else {
855 log::warn!(
856 "[reorder] block-level frequency table cannot fit its bounded allocation; using identity order"
857 );
858 return ForwardIndex {
859 terms: Vec::new(),
860 offsets: Vec::new(),
861 num_terms: 0,
862 parallel_bisect_lanes: 1,
863 budget_limited: true,
864 };
865 };
866
867 let max_bf = (total_blocks as f64 * 0.9) as usize;
868 let (mut eligible, mut budget_limited) = select_frequency_candidates(
869 &dim_bf,
870 2,
871 max_bf.max(2),
872 memory_budget_bytes
873 .saturating_sub(blocks_bytes)
874 .saturating_sub(frequency_bytes),
875 );
876 drop(dim_bf);
877
878 let entity_scratch_bytes = total_blocks.saturating_mul(32);
879 let remap_bytes = max_dims.saturating_mul(4);
880 let fixed_bytes = entity_scratch_bytes
881 .saturating_add(remap_bytes)
882 .saturating_add(blocks_bytes);
883 let fit = fit_candidates_to_budget(&mut eligible, fixed_bytes, memory_budget_bytes);
884 if fit.dropped > 0 {
885 budget_limited = true;
886 log::warn!(
887 "[reorder] block-level fwd index over budget — dropped {} highest-bf dims",
888 fit.dropped,
889 );
890 }
891
892 if eligible.is_empty() {
893 return ForwardIndex {
894 terms: Vec::new(),
895 offsets: Vec::new(),
896 num_terms: 0,
897 parallel_bisect_lanes: 1,
898 budget_limited,
899 };
900 }
901
902 let mut term_remap = vec![u32::MAX; max_dims];
903 for (compact, &(dim_id, _)) in eligible.iter().enumerate() {
904 term_remap[dim_id as usize] = compact as u32;
905 }
906 let num_terms = eligible.len();
907 let non_degree_bytes = entity_scratch_bytes.saturating_add(
908 fit.retained_postings
909 .saturating_mul(std::mem::size_of::<u32>()),
910 );
911 let parallel_bisect_lanes =
912 parallel_bisect_lanes(memory_budget_bytes, non_degree_bytes, num_terms);
913 drop(eligible);
914
915 let count_remapped = |&(src, block_id): &(u32, u32)| -> u32 {
918 bmps[src as usize]
919 .iter_block_terms(block_id)
920 .filter(|(dim_id, _, _)| {
921 term_remap
922 .get(*dim_id as usize)
923 .copied()
924 .unwrap_or(u32::MAX)
925 != u32::MAX
926 })
927 .count() as u32
928 };
929 #[cfg(feature = "native")]
930 let counts: Vec<u32> = blocks.par_iter().map(count_remapped).collect();
931 #[cfg(not(feature = "native"))]
932 let counts: Vec<u32> = blocks.iter().map(count_remapped).collect();
933
934 let offsets = build_csr_offsets(&counts);
935 let total = *offsets.last().unwrap() as usize;
936 drop(counts);
937
938 let mut terms = vec![0u32; total];
939 let fill_block = |&(src, block_id): &(u32, u32), out: &mut [u32]| {
940 let mut n = 0usize;
941 for (dim_id, _, _) in bmps[src as usize].iter_block_terms(block_id) {
942 let compact = term_remap.get(dim_id as usize).copied().unwrap_or(u32::MAX);
943 if compact != u32::MAX {
944 out[n] = compact;
945 n += 1;
946 }
947 }
948 };
949 {
950 let mut slices: Vec<(&(u32, u32), &mut [u32])> = Vec::with_capacity(blocks.len());
951 let mut rest: &mut [u32] = &mut terms;
952 for (gb, b) in blocks.iter().enumerate() {
953 let len = (offsets[gb + 1] - offsets[gb]) as usize;
954 let (head, tail) = rest.split_at_mut(len);
955 slices.push((b, head));
956 rest = tail;
957 }
958 #[cfg(feature = "native")]
959 slices
960 .into_par_iter()
961 .for_each(|(b, out)| fill_block(b, out));
962 #[cfg(not(feature = "native"))]
963 for (b, out) in slices {
964 fill_block(b, out);
965 }
966 }
967
968 ForwardIndex {
969 terms,
970 offsets,
971 num_terms,
972 parallel_bisect_lanes,
973 budget_limited,
974 }
975}
976
977#[derive(Clone, Copy, Debug, Default)]
985pub struct BpBudget {
986 pub min_partition_docs: Option<usize>,
991 pub time_budget: Option<std::time::Duration>,
995}
996
997impl BpBudget {
998 pub fn full() -> Self {
1000 Self::default()
1001 }
1002}
1003
1004fn build_term_degrees(
1011 docs: &[u32],
1012 mid: usize,
1013 fwd: &ForwardIndex,
1014 workspaces: &mut [TermDegrees],
1015) {
1016 debug_assert!(!workspaces.is_empty());
1017 let build_range = |degrees: &mut TermDegrees, start: usize, chunk: &[u32]| {
1018 degrees.reset();
1019 for (offset, &doc) in chunk.iter().enumerate() {
1020 let side = usize::from(start + offset >= mid);
1021 for &term in fwd.doc_terms(doc as usize) {
1022 degrees.entry_mut(term as usize)[side] += 1;
1023 }
1024 }
1025 };
1026
1027 #[cfg(feature = "native")]
1028 {
1029 let workers = workspaces
1030 .len()
1031 .min(docs.len().div_ceil(PARALLEL_BP_MIN_ENTITIES).max(1));
1032 if workers > 1 {
1033 let chunk_len = docs.len().div_ceil(workers);
1034 workspaces[..workers]
1035 .par_iter_mut()
1036 .enumerate()
1037 .for_each(|(worker, degrees)| {
1038 let start = worker * chunk_len;
1039 let end = (start + chunk_len).min(docs.len());
1040 build_range(degrees, start, &docs[start..end]);
1041 });
1042 let (degrees, partials) = workspaces[..workers]
1043 .split_first_mut()
1044 .expect("at least one BP degree workspace");
1045 for partial in partials {
1046 degrees.merge_from(partial);
1047 }
1048 return;
1049 }
1050 build_range(&mut workspaces[0], 0, docs);
1051 }
1052
1053 #[cfg(not(feature = "native"))]
1054 build_range(&mut workspaces[0], 0, docs);
1055}
1056
1057#[inline]
1063fn gain_order_key(gain: f32) -> u32 {
1064 let bits = gain.to_bits();
1065 if bits & 0x8000_0000 != 0 {
1066 !bits
1067 } else {
1068 bits ^ 0x8000_0000
1069 }
1070}
1071
1072fn select_gain_threshold(gains: &[f32], left_count: usize) -> (u32, usize) {
1076 debug_assert!(left_count > 0 && left_count <= gains.len());
1077 let mut rank_within_prefix = left_count - 1;
1078 let mut strictly_lower = 0usize;
1079 let mut prefix = 0u32;
1080 let mut prefix_mask = 0u32;
1081
1082 for shift in [24u32, 16, 8, 0] {
1083 let histogram = {
1084 #[cfg(feature = "native")]
1085 {
1086 if gains.len() >= PARALLEL_BP_MIN_ENTITIES {
1087 gains
1088 .par_iter()
1089 .fold(
1090 || Box::new([0usize; 256]),
1091 |mut counts, &gain| {
1092 let key = gain_order_key(gain);
1093 if key & prefix_mask == prefix {
1094 counts[((key >> shift) & 0xff) as usize] += 1;
1095 }
1096 counts
1097 },
1098 )
1099 .reduce(
1100 || Box::new([0usize; 256]),
1101 |mut left, right| {
1102 for (dst, &count) in left.iter_mut().zip(right.iter()) {
1103 *dst += count;
1104 }
1105 left
1106 },
1107 )
1108 } else {
1109 let mut counts = [0usize; 256];
1110 for &gain in gains {
1111 let key = gain_order_key(gain);
1112 if key & prefix_mask == prefix {
1113 counts[((key >> shift) & 0xff) as usize] += 1;
1114 }
1115 }
1116 Box::new(counts)
1117 }
1118 }
1119 #[cfg(not(feature = "native"))]
1120 {
1121 let mut counts = [0usize; 256];
1122 for &gain in gains {
1123 let key = gain_order_key(gain);
1124 if key & prefix_mask == prefix {
1125 counts[((key >> shift) & 0xff) as usize] += 1;
1126 }
1127 }
1128 counts
1129 }
1130 };
1131
1132 let mut before_bucket = 0usize;
1133 let mut selected_bucket = None;
1134 for (bucket, count) in histogram.iter().copied().enumerate() {
1135 if rank_within_prefix < before_bucket + count {
1136 selected_bucket = Some(bucket as u32);
1137 rank_within_prefix -= before_bucket;
1138 strictly_lower += before_bucket;
1139 break;
1140 }
1141 before_bucket += count;
1142 }
1143 let selected_bucket = selected_bucket.expect("BP radix selection lost the requested rank");
1144 prefix |= selected_bucket << shift;
1145 prefix_mask |= 0xffu32 << shift;
1146 }
1147
1148 (prefix, strictly_lower)
1149}
1150
1151enum PartitionDegreeUpdate {
1152 Moves,
1155 Ranked,
1159 Threshold {
1162 threshold_key: u32,
1163 ties_left: usize,
1164 },
1165}
1166
1167struct PartitionOutcome {
1168 swap_count: usize,
1169 degree_update: PartitionDegreeUpdate,
1170}
1171
1172#[derive(Clone, Copy)]
1173struct PartitionChunk {
1174 start: usize,
1175 end: usize,
1176 strictly_lower: usize,
1177 equal: usize,
1178 ties_left: usize,
1179}
1180
1181#[inline]
1182fn select_left(key: u32, threshold_key: u32, equal_seen: &mut usize, ties_left: usize) -> bool {
1183 if key < threshold_key {
1184 true
1185 } else if key == threshold_key {
1186 let selected = *equal_seen < ties_left;
1187 *equal_seen += 1;
1188 selected
1189 } else {
1190 false
1191 }
1192}
1193
1194fn partition_by_gain(
1203 docs: &[u32],
1204 gains: &[f32],
1205 mid: usize,
1206 fwd: &ForwardIndex,
1207 movement_workspaces: &mut [TermDegrees],
1208 output: &mut [u32],
1209 ranked_scratch: &mut [usize],
1210) -> PartitionOutcome {
1211 #[cfg(not(feature = "native"))]
1212 let _ = (fwd, &movement_workspaces);
1213
1214 #[cfg(feature = "native")]
1215 if !movement_workspaces.is_empty() && docs.len() >= PARALLEL_BP_MIN_ENTITIES {
1216 let (threshold_key, strictly_lower) = select_gain_threshold(gains, mid);
1217 let ties_left = mid - strictly_lower;
1218
1219 let chunk_count = movement_workspaces
1222 .len()
1223 .min(docs.len().div_ceil(PARALLEL_BP_MIN_ENTITIES).max(1));
1224 let chunk_len = docs.len().div_ceil(chunk_count);
1225 let mut chunks: Vec<PartitionChunk> = gains
1226 .par_chunks(chunk_len)
1227 .enumerate()
1228 .map(|(chunk_id, chunk)| {
1229 let mut lower = 0usize;
1230 let mut equal = 0usize;
1231 for &gain in chunk {
1232 match gain_order_key(gain).cmp(&threshold_key) {
1233 std::cmp::Ordering::Less => lower += 1,
1234 std::cmp::Ordering::Equal => equal += 1,
1235 std::cmp::Ordering::Greater => {}
1236 }
1237 }
1238 let start = chunk_id * chunk_len;
1239 PartitionChunk {
1240 start,
1241 end: start + chunk.len(),
1242 strictly_lower: lower,
1243 equal,
1244 ties_left: 0,
1245 }
1246 })
1247 .collect();
1248
1249 let mut remaining_ties = ties_left;
1250 for chunk in &mut chunks {
1251 chunk.ties_left = remaining_ties.min(chunk.equal);
1252 remaining_ties -= chunk.ties_left;
1253 }
1254 debug_assert_eq!(remaining_ties, 0);
1255
1256 let (mut left_rest, mut right_rest) = output.split_at_mut(mid);
1257 let mut jobs = Vec::with_capacity(chunks.len());
1258 for chunk in chunks {
1259 let left_len = chunk.strictly_lower + chunk.ties_left;
1260 let right_len = chunk.end - chunk.start - left_len;
1261 let (left_out, next_left) = left_rest.split_at_mut(left_len);
1262 let (right_out, next_right) = right_rest.split_at_mut(right_len);
1263 jobs.push((
1264 chunk.start,
1265 &docs[chunk.start..chunk.end],
1266 &gains[chunk.start..chunk.end],
1267 chunk.ties_left,
1268 left_out,
1269 right_out,
1270 ));
1271 left_rest = next_left;
1272 right_rest = next_right;
1273 }
1274 debug_assert!(left_rest.is_empty() && right_rest.is_empty());
1275
1276 let swap_count = movement_workspaces[..chunk_count]
1277 .par_iter_mut()
1278 .zip(jobs.into_par_iter())
1279 .map(
1280 |(moves, (start, docs, gains, ties_for_chunk, left_out, right_out))| {
1281 moves.reset();
1282 let mut equal_seen = 0usize;
1283 let mut left_cursor = 0usize;
1284 let mut right_cursor = 0usize;
1285 let mut swaps = 0usize;
1286
1287 for (offset, (&doc, &gain)) in docs.iter().zip(gains).enumerate() {
1288 let key = gain_order_key(gain);
1289 let now_left =
1290 select_left(key, threshold_key, &mut equal_seen, ties_for_chunk);
1291 if now_left {
1292 left_out[left_cursor] = doc;
1293 left_cursor += 1;
1294 } else {
1295 right_out[right_cursor] = doc;
1296 right_cursor += 1;
1297 }
1298
1299 let was_left = start + offset < mid;
1300 if was_left != now_left {
1301 swaps += 1;
1302 let direction = usize::from(was_left);
1304 for &term in fwd.doc_terms(doc as usize) {
1305 moves.entry_mut(term as usize)[direction] += 1;
1306 }
1307 }
1308 }
1309 debug_assert_eq!(left_cursor, left_out.len());
1310 debug_assert_eq!(right_cursor, right_out.len());
1311 swaps
1312 },
1313 )
1314 .sum();
1315
1316 let (moves, partials) = movement_workspaces[..chunk_count]
1317 .split_first_mut()
1318 .expect("parallel BP partition must use at least one movement workspace");
1319 for partial in partials {
1320 moves.merge_from(partial);
1321 }
1322
1323 return PartitionOutcome {
1324 swap_count,
1325 degree_update: PartitionDegreeUpdate::Moves,
1326 };
1327 }
1328
1329 if docs.len() < PARALLEL_BP_MIN_ENTITIES {
1330 debug_assert_eq!(ranked_scratch.len(), docs.len());
1331 for (index, rank) in ranked_scratch.iter_mut().enumerate() {
1332 *rank = index;
1333 }
1334 ranked_scratch.select_nth_unstable_by(mid, |&left, &right| {
1335 gains[left]
1336 .total_cmp(&gains[right])
1337 .then_with(|| left.cmp(&right))
1338 });
1339
1340 let mut swaps = 0usize;
1341 for (rank, &old_index) in ranked_scratch.iter().enumerate() {
1342 output[rank] = docs[old_index];
1343 swaps += usize::from((old_index < mid) != (rank < mid));
1344 }
1345 return PartitionOutcome {
1346 swap_count: swaps,
1347 degree_update: PartitionDegreeUpdate::Ranked,
1348 };
1349 }
1350
1351 let (threshold_key, strictly_lower) = select_gain_threshold(gains, mid);
1352 let ties_left = mid - strictly_lower;
1353 let mut equal_seen = 0usize;
1354 let mut left_cursor = 0usize;
1355 let mut right_cursor = mid;
1356 let mut swaps = 0usize;
1357 for (idx, (&doc, &gain)) in docs.iter().zip(gains).enumerate() {
1358 let key = gain_order_key(gain);
1359 let now_left = select_left(key, threshold_key, &mut equal_seen, ties_left);
1360 if now_left {
1361 output[left_cursor] = doc;
1362 left_cursor += 1;
1363 } else {
1364 output[right_cursor] = doc;
1365 right_cursor += 1;
1366 }
1367 swaps += usize::from((idx < mid) != now_left);
1368 }
1369 debug_assert_eq!(left_cursor, mid);
1370 debug_assert_eq!(right_cursor, docs.len());
1371
1372 PartitionOutcome {
1373 swap_count: swaps,
1374 degree_update: PartitionDegreeUpdate::Threshold {
1375 threshold_key,
1376 ties_left,
1377 },
1378 }
1379}
1380
1381fn update_degrees_for_threshold_partition(
1383 docs: &[u32],
1384 gains: &[f32],
1385 mid: usize,
1386 threshold_key: u32,
1387 ties_left: usize,
1388 fwd: &ForwardIndex,
1389 degrees: &mut TermDegrees,
1390) {
1391 let mut equal_seen = 0usize;
1392 for (idx, (&doc, &gain)) in docs.iter().zip(gains).enumerate() {
1393 let key = gain_order_key(gain);
1394 let now_left = select_left(key, threshold_key, &mut equal_seen, ties_left);
1395 let was_left = idx < mid;
1396 if was_left == now_left {
1397 continue;
1398 }
1399 let left_delta = if was_left { -1i64 } else { 1i64 };
1400 for &term in fwd.doc_terms(doc as usize) {
1401 let degree = degrees.entry_mut(term as usize);
1402 let new_left = degree[0] as i64 + left_delta;
1403 let new_right = degree[1] as i64 - left_delta;
1404 debug_assert!(new_left >= 0 && new_right >= 0);
1405 degree[0] = new_left as u32;
1406 degree[1] = new_right as u32;
1407 }
1408 }
1409}
1410
1411fn update_degrees_for_ranked_partition(
1414 docs: &[u32],
1415 ranked: &[usize],
1416 mid: usize,
1417 fwd: &ForwardIndex,
1418 degrees: &mut TermDegrees,
1419) {
1420 for (rank, &old_index) in ranked.iter().enumerate() {
1421 let was_left = old_index < mid;
1422 let now_left = rank < mid;
1423 if was_left == now_left {
1424 continue;
1425 }
1426 let left_delta = if was_left { -1i64 } else { 1i64 };
1427 for &term in fwd.doc_terms(docs[old_index] as usize) {
1428 let degree = degrees.entry_mut(term as usize);
1429 let new_left = degree[0] as i64 + left_delta;
1430 let new_right = degree[1] as i64 - left_delta;
1431 debug_assert!(new_left >= 0 && new_right >= 0);
1432 degree[0] = new_left as u32;
1433 degree[1] = new_right as u32;
1434 }
1435 }
1436}
1437
1438#[derive(Clone, Copy)]
1439pub(crate) struct BpProgressLabel<'a> {
1440 pub index: &'a str,
1441 pub field: &'a str,
1442 pub entity_kind: &'static str,
1443}
1444
1445#[cfg(test)]
1446impl BpProgressLabel<'static> {
1447 fn anonymous() -> Self {
1448 Self {
1449 index: "unknown",
1450 field: "unknown",
1451 entity_kind: "entities",
1452 }
1453 }
1454}
1455
1456#[cfg(feature = "native")]
1457struct BpProgress<'a> {
1458 label: BpProgressLabel<'a>,
1459 start: std::time::Instant,
1460 total_entities: usize,
1461 total_postings: u64,
1462 expected_depth: usize,
1463 next_log_ms: std::sync::atomic::AtomicU64,
1464 active_partitions: std::sync::atomic::AtomicU64,
1465 partitions_started: std::sync::atomic::AtomicU64,
1466 partitions_completed: std::sync::atomic::AtomicU64,
1467 iterations: std::sync::atomic::AtomicU64,
1468 entity_passes: std::sync::atomic::AtomicU64,
1469 swaps: std::sync::atomic::AtomicU64,
1470 deepest_level: std::sync::atomic::AtomicU64,
1471 objective_stops: std::sync::atomic::AtomicU64,
1472 last_objective_delta_bits: std::sync::atomic::AtomicU64,
1473 last_relative_delta_bits: std::sync::atomic::AtomicU64,
1474 active_metric_released: std::sync::atomic::AtomicBool,
1475}
1476
1477#[cfg(feature = "native")]
1478impl<'a> BpProgress<'a> {
1479 fn new(
1480 label: BpProgressLabel<'a>,
1481 total_entities: usize,
1482 total_postings: u64,
1483 expected_depth: usize,
1484 ) -> Self {
1485 log::info!(
1486 "[reorder][bp] started: index={} field={} entity_kind={} entities={} postings={} expected_depth={} objective_stall_threshold={:.1e}x{} min_objective_iterations={}",
1487 label.index,
1488 label.field,
1489 label.entity_kind,
1490 total_entities,
1491 total_postings,
1492 expected_depth,
1493 MIN_RELATIVE_OBJECTIVE_IMPROVEMENT,
1494 OBJECTIVE_STALL_ITERATIONS,
1495 MIN_OBJECTIVE_ITERATIONS,
1496 );
1497 crate::observe::reorder_bp_started(label.index, label.field, label.entity_kind);
1498 Self {
1499 label,
1500 start: std::time::Instant::now(),
1501 total_entities,
1502 total_postings,
1503 expected_depth,
1504 next_log_ms: std::sync::atomic::AtomicU64::new(30_000),
1505 active_partitions: std::sync::atomic::AtomicU64::new(0),
1506 partitions_started: std::sync::atomic::AtomicU64::new(0),
1507 partitions_completed: std::sync::atomic::AtomicU64::new(0),
1508 iterations: std::sync::atomic::AtomicU64::new(0),
1509 entity_passes: std::sync::atomic::AtomicU64::new(0),
1510 swaps: std::sync::atomic::AtomicU64::new(0),
1511 deepest_level: std::sync::atomic::AtomicU64::new(0),
1512 objective_stops: std::sync::atomic::AtomicU64::new(0),
1513 last_objective_delta_bits: std::sync::atomic::AtomicU64::new(0f64.to_bits()),
1514 last_relative_delta_bits: std::sync::atomic::AtomicU64::new(0f64.to_bits()),
1515 active_metric_released: std::sync::atomic::AtomicBool::new(false),
1516 }
1517 }
1518
1519 fn partition_started(&self, level: usize) {
1520 self.active_partitions
1521 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1522 self.partitions_started
1523 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1524 self.deepest_level
1525 .fetch_max(level as u64, std::sync::atomic::Ordering::Relaxed);
1526 }
1527
1528 fn partition_finished(&self) {
1529 self.partitions_completed
1530 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1531 self.active_partitions
1532 .fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
1533 }
1534
1535 fn iteration(&self, entities: usize, swaps: usize, objective_delta: f64, relative_delta: f64) {
1536 self.iterations
1537 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1538 self.entity_passes
1539 .fetch_add(entities as u64, std::sync::atomic::Ordering::Relaxed);
1540 self.swaps
1541 .fetch_add(swaps as u64, std::sync::atomic::Ordering::Relaxed);
1542 self.last_objective_delta_bits.store(
1543 objective_delta.to_bits(),
1544 std::sync::atomic::Ordering::Relaxed,
1545 );
1546 self.last_relative_delta_bits.store(
1547 relative_delta.to_bits(),
1548 std::sync::atomic::Ordering::Relaxed,
1549 );
1550 self.maybe_log();
1551 }
1552
1553 fn objective_stop(&self) {
1554 self.objective_stops
1555 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1556 }
1557
1558 fn maybe_log(&self) {
1559 let elapsed_ms = self.start.elapsed().as_millis().min(u64::MAX as u128) as u64;
1560 let next = self.next_log_ms.load(std::sync::atomic::Ordering::Relaxed);
1561 if elapsed_ms < next
1562 || self
1563 .next_log_ms
1564 .compare_exchange(
1565 next,
1566 elapsed_ms.saturating_add(30_000),
1567 std::sync::atomic::Ordering::Relaxed,
1568 std::sync::atomic::Ordering::Relaxed,
1569 )
1570 .is_err()
1571 {
1572 return;
1573 }
1574
1575 let active = self
1576 .active_partitions
1577 .load(std::sync::atomic::Ordering::Relaxed);
1578 let started = self
1579 .partitions_started
1580 .load(std::sync::atomic::Ordering::Relaxed);
1581 let completed = self
1582 .partitions_completed
1583 .load(std::sync::atomic::Ordering::Relaxed);
1584 let iterations = self.iterations.load(std::sync::atomic::Ordering::Relaxed);
1585 let entity_passes = self
1586 .entity_passes
1587 .load(std::sync::atomic::Ordering::Relaxed);
1588 let swaps = self.swaps.load(std::sync::atomic::Ordering::Relaxed);
1589 let deepest = self
1590 .deepest_level
1591 .load(std::sync::atomic::Ordering::Relaxed);
1592 let objective_delta = f64::from_bits(
1593 self.last_objective_delta_bits
1594 .load(std::sync::atomic::Ordering::Relaxed),
1595 );
1596 let relative_delta = f64::from_bits(
1597 self.last_relative_delta_bits
1598 .load(std::sync::atomic::Ordering::Relaxed),
1599 );
1600 log::info!(
1601 "[reorder][bp] progress: index={} field={} entity_kind={} elapsed={:.1}s depth={}/{} partitions={}/{} active={} iterations={} entity_passes={} swaps={} last_objective_delta={:.3} relative={:.3e}",
1602 self.label.index,
1603 self.label.field,
1604 self.label.entity_kind,
1605 self.start.elapsed().as_secs_f64(),
1606 deepest,
1607 self.expected_depth,
1608 completed,
1609 started,
1610 active,
1611 iterations,
1612 entity_passes,
1613 swaps,
1614 objective_delta,
1615 relative_delta,
1616 );
1617 }
1618
1619 fn finish(&self, converged: bool, memory_limited: bool, deadline_exhausted: bool) {
1620 let elapsed = self.start.elapsed().as_secs_f64();
1621 let partitions = self
1622 .partitions_completed
1623 .load(std::sync::atomic::Ordering::Relaxed);
1624 let iterations = self.iterations.load(std::sync::atomic::Ordering::Relaxed);
1625 let entity_passes = self
1626 .entity_passes
1627 .load(std::sync::atomic::Ordering::Relaxed);
1628 let swaps = self.swaps.load(std::sync::atomic::Ordering::Relaxed);
1629 let deepest = self
1630 .deepest_level
1631 .load(std::sync::atomic::Ordering::Relaxed);
1632 let objective_stops = self
1633 .objective_stops
1634 .load(std::sync::atomic::Ordering::Relaxed);
1635 let stop_reason = if memory_limited {
1636 "memory_budget"
1637 } else if deadline_exhausted {
1638 "time_budget"
1639 } else if objective_stops > 0 {
1640 "objective"
1641 } else {
1642 "complete"
1643 };
1644 log::info!(
1645 "[reorder][bp] completed: index={} field={} entity_kind={} entities={} postings={} elapsed={:.1}s depth={}/{} partitions={} iterations={} entity_passes={} swaps={} objective_stops={} converged={} stop_reason={}",
1646 self.label.index,
1647 self.label.field,
1648 self.label.entity_kind,
1649 self.total_entities,
1650 self.total_postings,
1651 elapsed,
1652 deepest,
1653 self.expected_depth,
1654 partitions,
1655 iterations,
1656 entity_passes,
1657 swaps,
1658 objective_stops,
1659 converged,
1660 stop_reason,
1661 );
1662 crate::observe::reorder_bp_pass(
1663 self.label.index,
1664 self.label.field,
1665 self.label.entity_kind,
1666 stop_reason,
1667 elapsed,
1668 self.total_entities,
1669 self.total_postings,
1670 partitions,
1671 iterations,
1672 entity_passes,
1673 swaps,
1674 converged,
1675 );
1676 self.release_active_metric();
1677 }
1678
1679 fn release_active_metric(&self) {
1680 if !self
1681 .active_metric_released
1682 .swap(true, std::sync::atomic::Ordering::AcqRel)
1683 {
1684 crate::observe::reorder_bp_finished(
1685 self.label.index,
1686 self.label.field,
1687 self.label.entity_kind,
1688 );
1689 }
1690 }
1691}
1692
1693#[cfg(feature = "native")]
1694impl Drop for BpProgress<'_> {
1695 fn drop(&mut self) {
1696 self.release_active_metric();
1697 }
1698}
1699
1700#[cfg(not(feature = "native"))]
1701struct BpProgress<'a>(std::marker::PhantomData<&'a ()>);
1702
1703#[cfg(not(feature = "native"))]
1704impl BpProgress<'_> {
1705 fn new(_: BpProgressLabel<'_>, _: usize, _: u64, _: usize) -> Self {
1706 Self(std::marker::PhantomData)
1707 }
1708 fn partition_started(&self, _: usize) {}
1709 fn partition_finished(&self) {}
1710 fn iteration(&self, _: usize, _: usize, _: f64, _: f64) {}
1711 fn objective_stop(&self) {}
1712 fn finish(&self, _: bool, _: bool, _: bool) {}
1713}
1714
1715#[cfg(test)]
1726pub(crate) fn graph_bisection(
1727 fwd: &ForwardIndex,
1728 min_partition_size: usize,
1729 max_iters: usize,
1730 budget: BpBudget,
1731) -> (Vec<u32>, bool) {
1732 graph_bisection_with_progress(
1733 fwd,
1734 min_partition_size,
1735 max_iters,
1736 budget,
1737 BpProgressLabel::anonymous(),
1738 )
1739}
1740
1741pub(crate) fn graph_bisection_with_progress(
1742 fwd: &ForwardIndex,
1743 min_partition_size: usize,
1744 max_iters: usize,
1745 budget: BpBudget,
1746 progress_label: BpProgressLabel<'_>,
1747) -> (Vec<u32>, bool) {
1748 let n = fwd.num_docs();
1749 if n == 0 {
1750 return (Vec::new(), !fwd.budget_limited);
1751 }
1752
1753 let effective_min_partition = budget
1754 .min_partition_docs
1755 .unwrap_or(0)
1756 .max(min_partition_size);
1757
1758 let mut docs: Vec<u32> = (0..n as u32).collect();
1759 let depth = if effective_min_partition > 0 {
1760 ((n as f64) / (effective_min_partition as f64))
1761 .log2()
1762 .ceil() as usize
1763 } else {
1764 0
1765 };
1766 let log_table = build_log_table(4096);
1767 let progress = BpProgress::new(progress_label, n, fwd.total_postings(), depth);
1768
1769 log::debug!(
1770 "BP graph_bisection: n={}, min_partition={}, max_iters={}, depth=~{}, time_budget={:?}",
1771 n,
1772 effective_min_partition,
1773 max_iters,
1774 depth,
1775 budget.time_budget,
1776 );
1777
1778 #[cfg(feature = "native")]
1779 let deadline = budget.time_budget.map(|duration| {
1780 let now = std::time::Instant::now();
1781 now.checked_add(duration).unwrap_or(now)
1782 });
1783 #[cfg(not(feature = "native"))]
1784 let deadline: Option<()> = None;
1785
1786 let exhausted = std::sync::atomic::AtomicBool::new(false);
1787 let context = BisectContext {
1788 fwd,
1789 min_partition_size: effective_min_partition,
1790 max_iters,
1791 log_table: &log_table,
1792 #[cfg(feature = "native")]
1793 deadline,
1794 #[cfg(not(feature = "native"))]
1795 deadline,
1796 exhausted: &exhausted,
1797 progress: &progress,
1798 };
1799 let immediately_exhausted = {
1800 #[cfg(feature = "native")]
1801 {
1802 deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline)
1803 }
1804 #[cfg(not(feature = "native"))]
1805 {
1806 false
1807 }
1808 };
1809 if immediately_exhausted {
1810 exhausted.store(true, std::sync::atomic::Ordering::Relaxed);
1811 } else if n > effective_min_partition {
1812 let mut gains = vec![0.0f32; n];
1816 let mut partitioned = vec![0u32; n];
1817 let mut ranked = vec![0usize; n];
1818 #[cfg(feature = "native")]
1819 let degree_lanes = fwd.parallel_bisect_lanes.max(1);
1820 #[cfg(not(feature = "native"))]
1821 let degree_lanes = 1usize;
1822 let mut degree_workspaces: Vec<TermDegrees> = (0..degree_lanes)
1823 .map(|_| TermDegrees::new(fwd.num_terms))
1824 .collect();
1825 bisect(
1826 &mut docs,
1827 &mut gains,
1828 &mut partitioned,
1829 &mut ranked,
1830 &mut degree_workspaces,
1831 0,
1832 &context,
1833 );
1834 }
1835
1836 let deadline_exhausted = exhausted.load(std::sync::atomic::Ordering::Relaxed);
1837 let converged = !fwd.budget_limited && !deadline_exhausted;
1838 progress.finish(converged, fwd.budget_limited, deadline_exhausted);
1839 if !converged {
1840 log::info!(
1841 "BP graph_bisection: budget incomplete at n={} (time={:?}, memory_limited={}) — emitting partial (still valid) permutation",
1842 n,
1843 budget.time_budget,
1844 fwd.budget_limited,
1845 );
1846 }
1847 (docs, converged)
1848}
1849
1850struct BisectContext<'a> {
1859 fwd: &'a ForwardIndex,
1860 min_partition_size: usize,
1861 max_iters: usize,
1862 log_table: &'a [f32],
1863 #[cfg(feature = "native")]
1864 deadline: Option<std::time::Instant>,
1865 #[cfg(not(feature = "native"))]
1866 deadline: Option<()>,
1867 exhausted: &'a std::sync::atomic::AtomicBool,
1868 progress: &'a BpProgress<'a>,
1869}
1870
1871#[allow(clippy::too_many_arguments)]
1872fn bisect(
1873 docs: &mut [u32],
1874 gains: &mut [f32],
1875 partitioned: &mut [u32],
1876 ranked_scratch: &mut [usize],
1877 degree_workspaces: &mut [TermDegrees],
1878 level: usize,
1879 context: &BisectContext<'_>,
1880) {
1881 let n = docs.len();
1882 debug_assert_eq!(gains.len(), n);
1883 debug_assert_eq!(partitioned.len(), n);
1884 debug_assert_eq!(ranked_scratch.len(), n);
1885 debug_assert!(!degree_workspaces.is_empty());
1886 if n <= context.min_partition_size {
1887 return;
1888 }
1889 if context.exhausted.load(std::sync::atomic::Ordering::Relaxed) {
1891 return;
1892 }
1893 #[cfg(feature = "native")]
1894 if let Some(dl) = context.deadline
1895 && std::time::Instant::now() >= dl
1896 {
1897 context
1898 .exhausted
1899 .store(true, std::sync::atomic::Ordering::Relaxed);
1900 return;
1901 }
1902 #[cfg(not(feature = "native"))]
1903 let _ = context.deadline;
1904
1905 context.progress.partition_started(level);
1906 let mid = n / 2;
1907
1908 let effective_iters = if n > 100_000 {
1912 context.max_iters.min(12)
1913 } else {
1914 context.max_iters
1915 };
1916
1917 build_term_degrees(docs, mid, context.fwd, degree_workspaces);
1922 let track_objective = n >= PARALLEL_BP_MIN_ENTITIES;
1927 if track_objective {
1928 degree_workspaces[0].sort_touched_words();
1932 }
1933 let mut previous_objective = if track_objective {
1934 degree_workspaces[0].bisection_objective(mid, n - mid, context.log_table)
1935 } else {
1936 0.0
1937 };
1938 let mut best_objective = previous_objective;
1939 let mut objective_stalls = 0usize;
1940
1941 for iter in 0..effective_iters {
1942 #[cfg(feature = "native")]
1944 if let Some(dl) = context.deadline
1945 && std::time::Instant::now() >= dl
1946 {
1947 context
1948 .exhausted
1949 .store(true, std::sync::atomic::Ordering::Relaxed);
1950 break;
1951 }
1952 compute_gains(
1955 docs,
1956 context.fwd,
1957 mid,
1958 °ree_workspaces[0],
1959 context.log_table,
1960 gains,
1961 );
1962
1963 let partition = partition_by_gain(
1968 docs,
1969 gains,
1970 mid,
1971 context.fwd,
1972 &mut degree_workspaces[1..],
1973 partitioned,
1974 ranked_scratch,
1975 );
1976
1977 if partition.swap_count == 0 {
1978 context.progress.iteration(n, 0, 0.0, 0.0);
1979 docs.copy_from_slice(partitioned);
1983 break;
1984 }
1985
1986 match &partition.degree_update {
1987 PartitionDegreeUpdate::Moves => {
1988 let (degrees, movement_workspaces) = degree_workspaces
1989 .split_first_mut()
1990 .expect("BP always has one degree workspace");
1991 movement_workspaces
1992 .first()
1993 .expect("parallel BP movement update requires a spare workspace")
1994 .apply_moves_to(degrees);
1995 }
1996 PartitionDegreeUpdate::Ranked => update_degrees_for_ranked_partition(
1997 docs,
1998 ranked_scratch,
1999 mid,
2000 context.fwd,
2001 &mut degree_workspaces[0],
2002 ),
2003 PartitionDegreeUpdate::Threshold {
2004 threshold_key,
2005 ties_left,
2006 } => update_degrees_for_threshold_partition(
2007 docs,
2008 gains,
2009 mid,
2010 *threshold_key,
2011 *ties_left,
2012 context.fwd,
2013 &mut degree_workspaces[0],
2014 ),
2015 }
2016
2017 let (new_objective, objective_improvement, relative_improvement) = if track_objective {
2018 let new_objective =
2019 degree_workspaces[0].bisection_objective(mid, n - mid, context.log_table);
2020 let objective_improvement = new_objective - previous_objective;
2021 let relative_improvement = objective_improvement / previous_objective.abs().max(1.0);
2022 (new_objective, objective_improvement, relative_improvement)
2023 } else {
2024 (0.0, 0.0, 0.0)
2025 };
2026 context.progress.iteration(
2027 n,
2028 partition.swap_count,
2029 objective_improvement,
2030 relative_improvement,
2031 );
2032
2033 docs.copy_from_slice(partitioned);
2040 if track_objective {
2041 previous_objective = new_objective;
2042 let relative_best_improvement =
2043 (new_objective - best_objective) / best_objective.abs().max(1.0);
2044 if relative_best_improvement >= MIN_RELATIVE_OBJECTIVE_IMPROVEMENT {
2045 best_objective = new_objective;
2046 objective_stalls = 0;
2047 } else if iter + 1 >= MIN_OBJECTIVE_ITERATIONS {
2048 objective_stalls += 1;
2049 }
2050 if objective_stalls >= OBJECTIVE_STALL_ITERATIONS {
2051 context.progress.objective_stop();
2052 break;
2053 }
2054 }
2055
2056 if iter > 2 && partition.swap_count < n / 200 {
2058 break;
2059 }
2060
2061 if !track_objective && iter > 5 {
2065 let max_abs_gain = gains
2066 .iter()
2067 .copied()
2068 .fold(0.0f32, |max_gain, gain| max_gain.max(gain.abs()));
2069 if max_abs_gain < 0.001 {
2070 break;
2071 }
2072 }
2073 }
2074
2075 context.progress.partition_finished();
2076
2077 let (left, right) = docs.split_at_mut(mid);
2078 let (left_gains, right_gains) = gains.split_at_mut(mid);
2079 let (left_partitioned, right_partitioned) = partitioned.split_at_mut(mid);
2080 let (left_ranked, right_ranked) = ranked_scratch.split_at_mut(mid);
2081 #[cfg(feature = "native")]
2082 if degree_workspaces.len() > 1 {
2083 let left_lanes = degree_workspaces.len() / 2;
2084 let (left_workspaces, right_workspaces) = degree_workspaces.split_at_mut(left_lanes);
2085 rayon::join(
2086 || {
2087 bisect(
2088 left,
2089 left_gains,
2090 left_partitioned,
2091 left_ranked,
2092 left_workspaces,
2093 level + 1,
2094 context,
2095 )
2096 },
2097 || {
2098 bisect(
2099 right,
2100 right_gains,
2101 right_partitioned,
2102 right_ranked,
2103 right_workspaces,
2104 level + 1,
2105 context,
2106 )
2107 },
2108 );
2109 } else {
2110 bisect(
2114 left,
2115 left_gains,
2116 left_partitioned,
2117 left_ranked,
2118 degree_workspaces,
2119 level + 1,
2120 context,
2121 );
2122 bisect(
2123 right,
2124 right_gains,
2125 right_partitioned,
2126 right_ranked,
2127 degree_workspaces,
2128 level + 1,
2129 context,
2130 );
2131 }
2132 #[cfg(not(feature = "native"))]
2133 {
2134 bisect(
2135 left,
2136 left_gains,
2137 left_partitioned,
2138 left_ranked,
2139 degree_workspaces,
2140 level + 1,
2141 context,
2142 );
2143 bisect(
2144 right,
2145 right_gains,
2146 right_partitioned,
2147 right_ranked,
2148 degree_workspaces,
2149 level + 1,
2150 context,
2151 );
2152 }
2153}
2154
2155#[inline(never)]
2161fn compute_gains(
2162 docs: &[u32],
2163 fwd: &ForwardIndex,
2164 mid: usize,
2165 degrees: &TermDegrees,
2166 log_table: &[f32],
2167 gains: &mut [f32],
2168) {
2169 let gain_for_doc = |i: usize| -> f32 {
2178 let doc = docs[i] as usize;
2179 let in_left = i < mid;
2180 let mut g = 0.0f32;
2181 for &term in fwd.doc_terms(doc) {
2182 let [left, right] = degrees.get(term as usize);
2183 let (from, to) = if in_left {
2184 (left, right)
2185 } else {
2186 (right, left)
2187 };
2188 let move_gain = fast_log2_lookup(to as usize + 2, log_table)
2189 - fast_log2_lookup(from as usize, log_table)
2190 - std::f32::consts::LOG2_E / (1.0 + to as f32);
2191 g += if in_left { move_gain } else { -move_gain };
2192 }
2193 g
2194 };
2195
2196 #[cfg(feature = "native")]
2197 {
2198 if docs.len() > 4096 {
2199 gains
2200 .par_iter_mut()
2201 .enumerate()
2202 .for_each(|(i, gain)| *gain = gain_for_doc(i));
2203 } else {
2204 for (i, gain) in gains.iter_mut().enumerate().take(docs.len()) {
2205 *gain = gain_for_doc(i);
2206 }
2207 }
2208 }
2209 #[cfg(not(feature = "native"))]
2210 {
2211 for (i, gain) in gains.iter_mut().enumerate().take(docs.len()) {
2212 *gain = gain_for_doc(i);
2213 }
2214 }
2215}
2216
2217fn build_log_table(size: usize) -> Vec<f32> {
2221 let mut table = vec![0.0f32; size];
2222 table[0] = -10.0;
2224 for (i, entry) in table.iter_mut().enumerate().skip(1) {
2225 *entry = (i as f32).log2();
2226 }
2227 table
2228}
2229
2230#[inline]
2232fn fast_log2_lookup(val: usize, table: &[f32]) -> f32 {
2233 if val < table.len() {
2234 table[val]
2235 } else {
2236 (val as f32).log2()
2237 }
2238}
2239
2240#[cfg(test)]
2241mod tests {
2242 use super::*;
2243
2244 #[test]
2245 fn bounded_frequency_count_and_candidate_selection_are_exact() {
2246 let items: Vec<u32> = (0..10_000).collect();
2247 let frequencies = count_frequencies_bounded(&items, 7, 16 * 1024, |item, counts| {
2248 let term = *item as usize % counts.len();
2249 counts[term] += 1;
2250 })
2251 .unwrap();
2252 assert_eq!(frequencies.iter().sum::<u32>(), items.len() as u32);
2253 for (term, &frequency) in frequencies.iter().enumerate() {
2254 let expected = (term..items.len()).step_by(frequencies.len()).count() as u32;
2255 assert_eq!(frequency, expected);
2256 }
2257
2258 let (selected, limited) =
2259 select_frequency_candidates(&[8, 3, 0, 5, 3], 1, 10, 2 * CANDIDATE_ENTRY_BYTES);
2260 assert!(limited);
2261 let mut selected = selected;
2262 selected.sort_unstable();
2263 assert_eq!(selected, vec![(1, 3), (4, 3)]);
2264 }
2265
2266 #[test]
2267 fn bounded_frequency_count_rejects_an_undersized_budget() {
2268 assert!(
2269 count_frequencies_bounded(&[0u32], 32, 32, |_, _| {}).is_none(),
2270 "one complete dense frequency table must fit before counting"
2271 );
2272 }
2273
2274 #[test]
2275 fn lazy_term_degrees_initialize_only_on_first_write() {
2276 let mut degrees = TermDegrees::new(130);
2277 let values_ptr = degrees.values.as_ptr();
2278 let bitmap_ptr = degrees.initialized.as_ptr();
2279 assert_eq!(degrees.get(65), [0, 0]);
2280 degrees.entry_mut(65)[0] += 3;
2281 degrees.entry_mut(65)[1] += 2;
2282 assert_eq!(degrees.get(65), [3, 2]);
2283 assert_eq!(degrees.get(64), [0, 0]);
2284 assert_eq!(
2285 degrees
2286 .initialized
2287 .iter()
2288 .map(|w| w.count_ones())
2289 .sum::<u32>(),
2290 1
2291 );
2292
2293 degrees.reset();
2294 assert_eq!(degrees.values.as_ptr(), values_ptr);
2295 assert_eq!(degrees.initialized.as_ptr(), bitmap_ptr);
2296 assert_eq!(degrees.get(65), [0, 0]);
2297 assert!(degrees.touched_words.is_empty());
2298 degrees.entry_mut(129)[1] = 7;
2299 assert_eq!(degrees.get(129), [0, 7]);
2300 assert_eq!(degrees.get(65), [0, 0]);
2301 }
2302
2303 #[test]
2304 fn sparse_objective_keeps_the_original_ascending_term_order() {
2305 let mut degrees = TermDegrees::new(130);
2306 *degrees.entry_mut(129) = [5, 2];
2307 *degrees.entry_mut(1) = [3, 7];
2308 *degrees.entry_mut(65) = [11, 13];
2309 assert_eq!(degrees.touched_words, vec![2, 0, 1]);
2310 degrees.sort_touched_words();
2311 assert_eq!(degrees.touched_words, vec![0, 1, 2]);
2312
2313 let log_table = build_log_table(4096);
2314 let actual = degrees.bisection_objective(31, 32, &log_table);
2315 let side_log = [
2316 fast_log2_lookup(31, &log_table) as f64,
2317 fast_log2_lookup(32, &log_table) as f64,
2318 ];
2319 let mut original_bitmap_scan = 0.0f64;
2320 for (word_idx, &initialized) in degrees.initialized.iter().enumerate() {
2321 let mut pending = initialized;
2322 while pending != 0 {
2323 let bit = pending.trailing_zeros() as usize;
2324 let term = word_idx * 64 + bit;
2325 let [left, right] = degrees.get(term);
2326 for (side, count) in [left, right].into_iter().enumerate() {
2327 if count > 0 {
2328 original_bitmap_scan += count as f64
2329 * (fast_log2_lookup(count as usize + 1, &log_table) as f64
2330 - side_log[side]);
2331 }
2332 }
2333 pending &= pending - 1;
2334 }
2335 }
2336 assert_eq!(actual.to_bits(), original_bitmap_scan.to_bits());
2337 }
2338
2339 #[test]
2340 fn gain_radix_key_matches_total_cmp() {
2341 let values = [
2342 f32::from_bits(0xffc0_0001),
2343 f32::NEG_INFINITY,
2344 -42.0,
2345 -0.0,
2346 0.0,
2347 42.0,
2348 f32::INFINITY,
2349 f32::from_bits(0x7fc0_0001),
2350 ];
2351 let mut by_cmp = values;
2352 by_cmp.sort_by(f32::total_cmp);
2353 let mut by_key = values;
2354 by_key.sort_by_key(|value| gain_order_key(*value));
2355 assert_eq!(
2356 by_cmp.map(f32::to_bits),
2357 by_key.map(f32::to_bits),
2358 "radix selection must preserve the former total_cmp order"
2359 );
2360 }
2361
2362 #[test]
2363 fn radix_threshold_matches_exact_rank_with_ties() {
2364 let gains = [3.0, -1.0, 7.0, -1.0, 0.0, -0.0, 3.0, 9.0, 3.0, 2.0, 2.0];
2365 let mut sorted: Vec<(u32, usize)> = gains
2366 .iter()
2367 .enumerate()
2368 .map(|(idx, &gain)| (gain_order_key(gain), idx))
2369 .collect();
2370 sorted.sort_unstable();
2371
2372 for left_count in 1..=gains.len() {
2373 let (threshold, lower) = select_gain_threshold(&gains, left_count);
2374 assert_eq!(threshold, sorted[left_count - 1].0);
2375 assert_eq!(lower, sorted.partition_point(|&(key, _)| key < threshold),);
2376 }
2377 }
2378
2379 #[cfg(feature = "native")]
2380 #[test]
2381 fn parallel_partition_matches_exact_selection_and_degree_rebuild() {
2382 const N: usize = PARALLEL_BP_MIN_ENTITIES + 1;
2383 const TERMS: usize = 101;
2384 let mut terms = Vec::with_capacity(N * 3);
2385 let mut offsets = Vec::with_capacity(N + 1);
2386 offsets.push(0);
2387 for doc in 0..N {
2388 terms.extend_from_slice(&[
2389 (doc % TERMS) as u32,
2390 ((doc / 7) % TERMS) as u32,
2391 ((doc * 13) % TERMS) as u32,
2392 ]);
2393 offsets.push(terms.len() as u64);
2394 }
2395 let fwd = ForwardIndex {
2396 terms,
2397 offsets,
2398 num_terms: TERMS,
2399 parallel_bisect_lanes: 4,
2400 budget_limited: false,
2401 };
2402 let docs: Vec<u32> = (0..N as u32)
2403 .map(|idx| ((idx as usize * 7_919) % N) as u32)
2404 .collect();
2405 let gains: Vec<f32> = docs
2406 .iter()
2407 .map(|&doc| ((doc as usize * 37) % 257) as f32 - 128.0)
2408 .collect();
2409 let mid = N / 2;
2410
2411 let mut ranked: Vec<usize> = (0..N).collect();
2412 ranked.sort_unstable_by(|&left, &right| {
2413 gains[left]
2414 .total_cmp(&gains[right])
2415 .then_with(|| left.cmp(&right))
2416 });
2417 let mut selected_left = vec![false; N];
2418 for &idx in &ranked[..mid] {
2419 selected_left[idx] = true;
2420 }
2421 let expected: Vec<u32> = docs
2422 .iter()
2423 .enumerate()
2424 .filter(|(idx, _)| selected_left[*idx])
2425 .chain(
2426 docs.iter()
2427 .enumerate()
2428 .filter(|(idx, _)| !selected_left[*idx]),
2429 )
2430 .map(|(_, &doc)| doc)
2431 .collect();
2432
2433 let mut output = vec![0; N];
2434 let mut ranked_scratch = Vec::new();
2435 let mut movement_workspaces: Vec<_> = (0..3).map(|_| TermDegrees::new(TERMS)).collect();
2436 let outcome = partition_by_gain(
2437 &docs,
2438 &gains,
2439 mid,
2440 &fwd,
2441 &mut movement_workspaces,
2442 &mut output,
2443 &mut ranked_scratch,
2444 );
2445 assert_eq!(output, expected);
2446 assert!(
2447 matches!(&outcome.degree_update, PartitionDegreeUpdate::Moves),
2448 "test must exercise parallel movement counts"
2449 );
2450
2451 let mut updated_workspaces: Vec<_> = (0..4).map(|_| TermDegrees::new(TERMS)).collect();
2452 build_term_degrees(&docs, mid, &fwd, &mut updated_workspaces);
2453 movement_workspaces[0].apply_moves_to(&mut updated_workspaces[0]);
2454
2455 let mut rebuilt_workspaces: Vec<_> = (0..4).map(|_| TermDegrees::new(TERMS)).collect();
2456 build_term_degrees(&output, mid, &fwd, &mut rebuilt_workspaces);
2457 for term in 0..TERMS {
2458 assert_eq!(
2459 updated_workspaces[0].get(term),
2460 rebuilt_workspaces[0].get(term),
2461 "parallel movement-count mismatch for term {term}"
2462 );
2463 }
2464 }
2465
2466 #[test]
2472 fn test_csr_offsets_do_not_wrap_past_u32() {
2473 let counts = [1_500_000_000u32; 3]; let offsets = build_csr_offsets(&counts);
2475 assert_eq!(
2476 offsets,
2477 vec![0, 1_500_000_000, 3_000_000_000, 4_500_000_000]
2478 );
2479 assert!(*offsets.last().unwrap() > u32::MAX as u64);
2480 }
2481
2482 fn make_fwd(docs: &[&[u32]], num_terms: usize) -> ForwardIndex {
2484 let mut terms = Vec::new();
2485 let mut offsets = vec![0u64];
2486 for doc_terms in docs {
2487 terms.extend_from_slice(doc_terms);
2488 offsets.push(terms.len() as u64);
2489 }
2490 ForwardIndex {
2491 terms,
2492 offsets,
2493 num_terms,
2494 parallel_bisect_lanes: 1,
2495 budget_limited: false,
2496 }
2497 }
2498
2499 #[test]
2500 fn test_bp_empty() {
2501 let fwd = ForwardIndex {
2502 terms: Vec::new(),
2503 offsets: Vec::new(),
2504 num_terms: 0,
2505 parallel_bisect_lanes: 1,
2506 budget_limited: false,
2507 };
2508 let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
2509 assert!(perm.is_empty());
2510 }
2511
2512 #[test]
2513 fn test_bp_small() {
2514 let fwd = make_fwd(&[&[0, 1], &[0, 2], &[1, 3], &[2, 3]], 4);
2516 let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
2517 assert_eq!(perm.len(), 4);
2518 let mut sorted = perm.clone();
2520 sorted.sort();
2521 assert_eq!(sorted, vec![0, 1, 2, 3]);
2522 }
2523
2524 #[test]
2525 fn test_bp_clusters() {
2526 let fwd = make_fwd(
2530 &[
2531 &[0, 1],
2532 &[0, 1],
2533 &[0, 1],
2534 &[0, 1],
2535 &[2, 3],
2536 &[2, 3],
2537 &[2, 3],
2538 &[2, 3],
2539 ],
2540 4,
2541 );
2542 let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
2543 assert_eq!(perm.len(), 8);
2544
2545 let left: Vec<u32> = perm[..4].to_vec();
2547
2548 let a_in_left = left.iter().filter(|&&d| d < 4).count();
2550 let b_in_left = left.iter().filter(|&&d| d >= 4).count();
2551 assert!(
2552 (a_in_left == 4 && b_in_left == 0) || (a_in_left == 0 && b_in_left == 4),
2553 "Clusters should be separated: a_left={}, b_left={}",
2554 a_in_left,
2555 b_in_left,
2556 );
2557 }
2558
2559 #[test]
2560 fn test_bp_permutation_valid() {
2561 let docs: Vec<Vec<u32>> = (0..16).map(|i| vec![i / 4, 10 + i / 2]).collect();
2563 let doc_refs: Vec<&[u32]> = docs.iter().map(|v| v.as_slice()).collect();
2564 let fwd = make_fwd(&doc_refs, 18); let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
2566
2567 assert_eq!(perm.len(), 16);
2568 let mut sorted = perm.clone();
2570 sorted.sort();
2571 let expected: Vec<u32> = (0..16).collect();
2572 assert_eq!(sorted, expected);
2573 }
2574
2575 #[test]
2580 fn test_bp_depth_cap_separates_clusters_and_converges() {
2581 let fwd = make_fwd(
2584 &[
2585 &[0, 1],
2586 &[0, 1],
2587 &[0, 1],
2588 &[2, 3],
2589 &[0, 1],
2590 &[2, 3],
2591 &[2, 3],
2592 &[2, 3],
2593 ],
2594 4,
2595 );
2596 let budget = BpBudget {
2597 min_partition_docs: Some(4),
2598 time_budget: None,
2599 };
2600 let (perm, converged) = graph_bisection(&fwd, 2, 20, budget);
2601 assert!(converged, "depth cap must report converged");
2602 assert_eq!(perm.len(), 8);
2603 let mut sorted = perm.clone();
2604 sorted.sort();
2605 assert_eq!(
2606 sorted,
2607 (0..8).collect::<Vec<u32>>(),
2608 "must stay a valid permutation"
2609 );
2610 let cluster_a = [0u32, 1, 2, 4];
2613 let a_in_left = perm[..4].iter().filter(|d| cluster_a.contains(d)).count();
2614 assert!(
2615 a_in_left == 4 || a_in_left == 0,
2616 "clusters should separate at the top level: {:?}",
2617 perm
2618 );
2619 }
2620
2621 #[test]
2624 fn test_bp_zero_time_budget_emits_valid_partial_permutation() {
2625 let docs: Vec<Vec<u32>> = (0..64).map(|i| vec![i % 4]).collect();
2626 let doc_refs: Vec<&[u32]> = docs.iter().map(|v| v.as_slice()).collect();
2627 let fwd = make_fwd(&doc_refs, 4);
2628 let budget = BpBudget {
2629 min_partition_docs: None,
2630 time_budget: Some(std::time::Duration::ZERO),
2631 };
2632 let (perm, converged) = graph_bisection(&fwd, 4, 20, budget);
2633 assert!(!converged, "zero budget must report unconverged");
2634 assert_eq!(perm.len(), 64);
2635 let mut sorted = perm.clone();
2636 sorted.sort();
2637 assert_eq!(sorted, (0..64).collect::<Vec<u32>>());
2638 }
2639
2640 #[test]
2641 fn test_memory_limited_graph_never_reports_converged() {
2642 let mut fwd = make_fwd(&[&[0], &[0], &[1], &[1]], 2);
2643 fwd.budget_limited = true;
2644
2645 let (perm, converged) = graph_bisection(&fwd, 2, 20, BpBudget::full());
2646
2647 assert!(!converged);
2648 let mut sorted = perm;
2649 sorted.sort_unstable();
2650 assert_eq!(sorted, vec![0, 1, 2, 3]);
2651 }
2652
2653 #[test]
2654 fn test_fast_log2() {
2655 let table = build_log_table(4096);
2656 assert!((table[1] - 0.0).abs() < 0.001);
2657 assert!((table[2] - 1.0).abs() < 0.001);
2658 assert!((table[4] - 2.0).abs() < 0.001);
2659 assert!((table[1024] - 10.0).abs() < 0.001);
2660 let val = fast_log2_lookup(8192, &table);
2662 assert!((val - 13.0).abs() < 0.001);
2663 }
2664}