Skip to main content

hermes_core/segment/builder/
graph_bisection.rs

1//! Recursive Graph Bisection (BP) for BMP document ordering.
2//!
3//! Based on Dhulipala et al. (KDD 2016) and Mackenzie et al. — the same
4//! algorithm used in Lucene and PISA for document reordering.
5//!
6//! Directly optimizes log-gap cost: docs sharing dimensions end up in the
7//! same BMP blocks, producing tight upper bounds and effective pruning.
8//!
9//! Memory: forward index ~200 bytes/doc (temporary) + degree arrays ~840 KB.
10
11#[cfg(feature = "native")]
12use rayon::prelude::*;
13use rustc_hash::FxHashMap;
14
15// ── Forward index (CSR) ──────────────────────────────────────────────────
16
17/// Forward index in CSR format: doc `d`'s terms are `terms[offsets[d]..offsets[d+1]]`.
18///
19/// Term IDs are remapped to compact range `0..num_terms` for flat-array degree tracking.
20pub(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    /// Total postings in the forward index.
44    pub fn total_postings(&self) -> u32 {
45        self.offsets.last().copied().unwrap_or(0)
46    }
47}
48
49/// Build virtual→real and real→virtual vid maps from a BMP doc map.
50///
51/// A virtual slot is real iff its doc-map entry is not the `u32::MAX` padding
52/// sentinel. Realness must come from the doc map itself: block-copy merged
53/// segments carry each source's tail padding as *interior* padding, so
54/// `vid < num_real_docs` does NOT identify real docs there.
55///
56/// Returns `(virtual_to_real, real_to_virtual)` where `virtual_to_real[vid]`
57/// is the dense real index or `u32::MAX` for padding.
58pub(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
84/// One (source, block) unit of forward-index construction. Because
85/// [`build_vid_maps`] assigns real ids in ascending vid order, a block's real
86/// docs form the contiguous per-source range
87/// `real_start..real_start + real_len` — blocks can be processed in parallel
88/// with disjoint output slices.
89struct BlockJob {
90    src: u32,
91    block_id: u32,
92    /// Per-source real index of the block's first real doc.
93    real_start: u32,
94    /// Number of real (non-padding) docs in the block.
95    real_len: u32,
96}
97
98/// Enumerate jobs in (source, block) order — cumulative `real_len` tiles the
99/// global real-id space `0..total_docs` exactly.
100fn build_block_jobs(
101    bmps: &[&crate::segment::reader::bmp::BmpIndex],
102    vid_maps: &[(Vec<u32>, Vec<u32>)],
103) -> Vec<BlockJob> {
104    let total_blocks: usize = bmps.iter().map(|b| b.num_blocks as usize).sum();
105    let mut jobs = Vec::with_capacity(total_blocks);
106    for (src, (bmp, (v2r, _))) in bmps.iter().zip(vid_maps).enumerate() {
107        let block_size = bmp.bmp_block_size as usize;
108        let mut real_cursor = 0u32;
109        for block_id in 0..bmp.num_blocks as usize {
110            let vid_start = block_id * block_size;
111            let vid_end = ((block_id + 1) * block_size).min(v2r.len());
112            let real_len = v2r[vid_start..vid_end]
113                .iter()
114                .filter(|&&r| r != u32::MAX)
115                .count() as u32;
116            jobs.push(BlockJob {
117                src: src as u32,
118                block_id: block_id as u32,
119                real_start: real_cursor,
120                real_len,
121            });
122            real_cursor += real_len;
123        }
124    }
125    jobs
126}
127
128/// Merge the smaller df map into the larger (rayon reduce combiner).
129#[cfg(feature = "native")]
130fn merge_df_maps(
131    mut a: FxHashMap<u32, usize>,
132    mut b: FxHashMap<u32, usize>,
133) -> FxHashMap<u32, usize> {
134    if a.len() < b.len() {
135        std::mem::swap(&mut a, &mut b);
136    }
137    for (k, v) in b {
138        *a.entry(k).or_insert(0) += v;
139    }
140    a
141}
142
143/// Build forward index from BmpIndex sources (single or multi-source).
144///
145/// Documents are identified by dense *real* indices assigned sequentially
146/// across sources: source 0 gets 0..n0, source 1 gets n0..n0+n1, etc., where
147/// each n is the source's real (non-padding) doc count derived from its doc
148/// map via [`build_vid_maps`]. Returns `(forward_index, per_source_real_doc_counts)`.
149///
150/// Filters dims with doc_freq outside `[min_doc_freq, max_doc_freq]`.
151/// If the estimated forward index memory exceeds `memory_budget_bytes`, the
152/// highest-frequency dims are dropped to stay within budget. This prevents OOM
153/// for huge segments at the cost of slightly reduced reorder quality.
154///
155/// Remaps term IDs to compact range for flat-array degree tracking.
156pub(crate) fn build_forward_index_from_bmps(
157    bmps: &[&crate::segment::reader::bmp::BmpIndex],
158    min_doc_freq: usize,
159    max_doc_freq: usize,
160    memory_budget_bytes: usize,
161) -> (ForwardIndex, Vec<usize>) {
162    // Per-source virtual→real maps (padding-aware realness from the doc map).
163    let vid_maps: Vec<(Vec<u32>, Vec<u32>)> = bmps.iter().map(|b| build_vid_maps(b)).collect();
164    let source_doc_counts: Vec<usize> = vid_maps.iter().map(|(_, r2v)| r2v.len()).collect();
165    let total_docs: usize = source_doc_counts.iter().sum();
166
167    if total_docs == 0 {
168        return (
169            ForwardIndex {
170                terms: Vec::new(),
171                offsets: Vec::new(),
172                num_terms: 0,
173            },
174            source_doc_counts,
175        );
176    }
177
178    // Job list: one entry per (source, block). Real ids are assigned in
179    // ascending vid order (see build_vid_maps), so each block owns a
180    // contiguous real-id range — every phase below can process blocks in
181    // parallel, writing disjoint slices.
182    let jobs = build_block_jobs(bmps, &vid_maps);
183
184    // Phase 1: count doc freq per dimension across all sources + assign compact IDs
185    let count_block_df = |job: &BlockJob, acc: &mut FxHashMap<u32, usize>| {
186        let bmp = bmps[job.src as usize];
187        let (v2r, _) = &vid_maps[job.src as usize];
188        let block_size = bmp.bmp_block_size as usize;
189        for (dim_id, postings) in bmp.iter_block_terms(job.block_id) {
190            let mut n = 0usize;
191            for p in postings {
192                let vid = job.block_id as usize * block_size + p.local_slot as usize;
193                if v2r[vid] != u32::MAX && p.impact > 0 {
194                    n += 1;
195                }
196            }
197            if n > 0 {
198                *acc.entry(dim_id).or_insert(0) += n;
199            }
200        }
201    };
202    #[cfg(feature = "native")]
203    let dim_df: FxHashMap<u32, usize> = jobs
204        .par_iter()
205        .fold(FxHashMap::default, |mut acc, job| {
206            count_block_df(job, &mut acc);
207            acc
208        })
209        .reduce(FxHashMap::default, merge_df_maps);
210    #[cfg(not(feature = "native"))]
211    let dim_df: FxHashMap<u32, usize> = {
212        let mut acc = FxHashMap::default();
213        for job in &jobs {
214            count_block_df(job, &mut acc);
215        }
216        acc
217    };
218
219    // Filter dims by [min_doc_freq, max_doc_freq] range
220    let mut eligible: Vec<(u32, usize)> = dim_df
221        .iter()
222        .filter(|&(_, df)| *df >= min_doc_freq && *df <= max_doc_freq)
223        .map(|(&dim_id, &df)| (dim_id, df))
224        .collect();
225    drop(dim_df);
226
227    // Memory budget: estimate forward index + bisection scratch.
228    // Peak ≈ 4*total_postings (terms array) + 28*total_docs (offsets, counts,
229    //   docs, gains, indices, new_left, new_right) + 8*num_terms (degree arrays).
230    let total_postings_est: usize = eligible.iter().map(|(_, df)| *df).sum();
231    let estimated_bytes = total_postings_est * 4 + total_docs * 28 + eligible.len() * 8;
232
233    if estimated_bytes > memory_budget_bytes && !eligible.is_empty() {
234        // Sort by df ascending — keep discriminative low-df dims first,
235        // drop highest-df dims which contribute the most postings.
236        eligible.sort_by_key(|&(_, df)| df);
237
238        let target_postings =
239            memory_budget_bytes.saturating_sub(total_docs * 28 + eligible.len() * 8) / 4;
240        let mut cum = 0usize;
241        let mut keep_count = 0;
242        for &(_, df) in &eligible {
243            if cum + df > target_postings {
244                break;
245            }
246            cum += df;
247            keep_count += 1;
248        }
249
250        let dropped = eligible.len() - keep_count;
251        eligible.truncate(keep_count);
252
253        log::warn!(
254            "[reorder] memory budget {:.0} MB: estimated {:.0} MB, dropped {} highest-df dims, keeping {} ({} postings)",
255            memory_budget_bytes as f64 / (1024.0 * 1024.0),
256            estimated_bytes as f64 / (1024.0 * 1024.0),
257            dropped,
258            keep_count,
259            cum,
260        );
261    }
262
263    let mut term_remap: FxHashMap<u32, u32> = FxHashMap::default();
264    for &(dim_id, _) in &eligible {
265        let compact_id = term_remap.len() as u32;
266        term_remap.insert(dim_id, compact_id);
267    }
268    let num_active_terms = term_remap.len();
269    drop(eligible);
270
271    // Phase 2: count terms per doc (filtered) — per-block disjoint slices
272    let mut counts = vec![0u32; total_docs];
273    let fill_block_counts = |job: &BlockJob, out: &mut [u32]| {
274        let bmp = bmps[job.src as usize];
275        let (v2r, _) = &vid_maps[job.src as usize];
276        let block_size = bmp.bmp_block_size as usize;
277        for (dim_id, postings) in bmp.iter_block_terms(job.block_id) {
278            if !term_remap.contains_key(&dim_id) {
279                continue;
280            }
281            for p in postings {
282                let vid = job.block_id as usize * block_size + p.local_slot as usize;
283                let real = v2r[vid];
284                if real != u32::MAX && p.impact > 0 {
285                    out[(real - job.real_start) as usize] += 1;
286                }
287            }
288        }
289    };
290    {
291        let mut slices: Vec<(&BlockJob, &mut [u32])> = Vec::with_capacity(jobs.len());
292        let mut rest: &mut [u32] = &mut counts;
293        for job in &jobs {
294            let (head, tail) = rest.split_at_mut(job.real_len as usize);
295            slices.push((job, head));
296            rest = tail;
297        }
298        #[cfg(feature = "native")]
299        slices
300            .into_par_iter()
301            .for_each(|(job, out)| fill_block_counts(job, out));
302        #[cfg(not(feature = "native"))]
303        for (job, out) in slices {
304            fill_block_counts(job, out);
305        }
306    }
307
308    // Phase 3: build CSR offsets
309    let mut offsets = Vec::with_capacity(total_docs + 1);
310    offsets.push(0u32);
311    for &c in &counts {
312        offsets.push(offsets.last().unwrap() + c);
313    }
314    let total = *offsets.last().unwrap() as usize;
315    drop(counts);
316
317    // Phase 4: fill terms (compact IDs) — each block writes the contiguous
318    // terms range covering its real docs; per-doc write cursors are local.
319    let mut terms = vec![0u32; total];
320    let fill_block_terms = |job: &BlockJob, global_real_start: usize, out: &mut [u32]| {
321        let bmp = bmps[job.src as usize];
322        let (v2r, _) = &vid_maps[job.src as usize];
323        let block_size = bmp.bmp_block_size as usize;
324        // local_slot is u8, so a block never holds more than 256 real docs
325        assert!(job.real_len as usize <= 256, "BMP block exceeds 256 docs");
326        let mut cursor = [0u32; 256];
327        let base = offsets[global_real_start] as usize;
328        for (dim_id, postings) in bmp.iter_block_terms(job.block_id) {
329            let Some(&compact) = term_remap.get(&dim_id) else {
330                continue;
331            };
332            for p in postings {
333                let vid = job.block_id as usize * block_size + p.local_slot as usize;
334                let real = v2r[vid];
335                if real != u32::MAX && p.impact > 0 {
336                    let local = (real - job.real_start) as usize;
337                    let pos =
338                        offsets[global_real_start + local] as usize - base + cursor[local] as usize;
339                    out[pos] = compact;
340                    cursor[local] += 1;
341                }
342            }
343        }
344    };
345    {
346        let mut slices: Vec<(&BlockJob, usize, &mut [u32])> = Vec::with_capacity(jobs.len());
347        let mut rest: &mut [u32] = &mut terms;
348        let mut global_real = 0usize;
349        for job in &jobs {
350            let len =
351                (offsets[global_real + job.real_len as usize] - offsets[global_real]) as usize;
352            let (head, tail) = rest.split_at_mut(len);
353            slices.push((job, global_real, head));
354            rest = tail;
355            global_real += job.real_len as usize;
356        }
357        #[cfg(feature = "native")]
358        slices
359            .into_par_iter()
360            .for_each(|(job, g, out)| fill_block_terms(job, g, out));
361        #[cfg(not(feature = "native"))]
362        for (job, g, out) in slices {
363            fill_block_terms(job, g, out);
364        }
365    }
366
367    (
368        ForwardIndex {
369            terms,
370            offsets,
371            num_terms: num_active_terms,
372        },
373        source_doc_counts,
374    )
375}
376
377/// Build a forward index over BLOCKS (one entity per block, its terms = the
378/// block's header dim list). Used by block-level reorder: BP over blocks is
379/// ~block_size× cheaper than over records and only needs to decide superblock
380/// assignment. Blocks are numbered globally across sources in source order.
381///
382/// Dims appearing in fewer than 2 blocks carry no clustering signal and are
383/// dropped; the memory budget applies as in the record-level builder.
384pub(crate) fn build_forward_index_from_blocks(
385    bmps: &[&crate::segment::reader::bmp::BmpIndex],
386    memory_budget_bytes: usize,
387) -> ForwardIndex {
388    let total_blocks: usize = bmps.iter().map(|b| b.num_blocks as usize).sum();
389    if total_blocks == 0 {
390        return ForwardIndex {
391            terms: Vec::new(),
392            offsets: Vec::new(),
393            num_terms: 0,
394        };
395    }
396
397    // (source, block) pairs in global block order — the parallel unit.
398    let blocks: Vec<(u32, u32)> = bmps
399        .iter()
400        .enumerate()
401        .flat_map(|(src, bmp)| (0..bmp.num_blocks).map(move |b| (src as u32, b)))
402        .collect();
403
404    // Phase 1: dim → number of blocks containing it (block-level df)
405    let count_block_bf = |&(src, block_id): &(u32, u32), acc: &mut FxHashMap<u32, usize>| {
406        for (dim_id, _) in bmps[src as usize].iter_block_terms(block_id) {
407            *acc.entry(dim_id).or_insert(0) += 1;
408        }
409    };
410    #[cfg(feature = "native")]
411    let dim_bf: FxHashMap<u32, usize> = blocks
412        .par_iter()
413        .fold(FxHashMap::default, |mut acc, b| {
414            count_block_bf(b, &mut acc);
415            acc
416        })
417        .reduce(FxHashMap::default, merge_df_maps);
418    #[cfg(not(feature = "native"))]
419    let dim_bf: FxHashMap<u32, usize> = {
420        let mut acc = FxHashMap::default();
421        for b in &blocks {
422            count_block_bf(b, &mut acc);
423        }
424        acc
425    };
426
427    let max_bf = (total_blocks as f64 * 0.9) as usize;
428    let mut eligible: Vec<(u32, usize)> = dim_bf
429        .iter()
430        .filter(|&(_, bf)| *bf >= 2 && *bf <= max_bf.max(2))
431        .map(|(&dim_id, &bf)| (dim_id, bf))
432        .collect();
433    drop(dim_bf);
434
435    let total_postings_est: usize = eligible.iter().map(|(_, bf)| *bf).sum();
436    let estimated_bytes = total_postings_est * 4 + total_blocks * 28 + eligible.len() * 8;
437    if estimated_bytes > memory_budget_bytes && !eligible.is_empty() {
438        eligible.sort_by_key(|&(_, bf)| bf);
439        let target = memory_budget_bytes.saturating_sub(total_blocks * 28 + eligible.len() * 8) / 4;
440        let mut cum = 0usize;
441        let mut keep = 0;
442        for &(_, bf) in &eligible {
443            if cum + bf > target {
444                break;
445            }
446            cum += bf;
447            keep += 1;
448        }
449        log::warn!(
450            "[reorder] block-level fwd index over budget — dropped {} highest-bf dims",
451            eligible.len() - keep,
452        );
453        eligible.truncate(keep);
454    }
455
456    let mut term_remap: FxHashMap<u32, u32> = FxHashMap::default();
457    for &(dim_id, _) in &eligible {
458        let compact = term_remap.len() as u32;
459        term_remap.insert(dim_id, compact);
460    }
461    let num_terms = term_remap.len();
462    drop(eligible);
463
464    // Phase 2+3: counts and CSR fill — one entity per block, so each block
465    // maps to a single count cell and a contiguous terms range.
466    let count_remapped = |&(src, block_id): &(u32, u32)| -> u32 {
467        bmps[src as usize]
468            .iter_block_terms(block_id)
469            .filter(|(dim_id, _)| term_remap.contains_key(dim_id))
470            .count() as u32
471    };
472    #[cfg(feature = "native")]
473    let counts: Vec<u32> = blocks.par_iter().map(count_remapped).collect();
474    #[cfg(not(feature = "native"))]
475    let counts: Vec<u32> = blocks.iter().map(count_remapped).collect();
476
477    let mut offsets = Vec::with_capacity(total_blocks + 1);
478    offsets.push(0u32);
479    for &c in &counts {
480        offsets.push(offsets.last().unwrap() + c);
481    }
482    let total = *offsets.last().unwrap() as usize;
483    drop(counts);
484
485    let mut terms = vec![0u32; total];
486    let fill_block = |&(src, block_id): &(u32, u32), out: &mut [u32]| {
487        let mut n = 0usize;
488        for (dim_id, _) in bmps[src as usize].iter_block_terms(block_id) {
489            if let Some(&compact) = term_remap.get(&dim_id) {
490                out[n] = compact;
491                n += 1;
492            }
493        }
494    };
495    {
496        let mut slices: Vec<(&(u32, u32), &mut [u32])> = Vec::with_capacity(blocks.len());
497        let mut rest: &mut [u32] = &mut terms;
498        for (gb, b) in blocks.iter().enumerate() {
499            let len = (offsets[gb + 1] - offsets[gb]) as usize;
500            let (head, tail) = rest.split_at_mut(len);
501            slices.push((b, head));
502            rest = tail;
503        }
504        #[cfg(feature = "native")]
505        slices
506            .into_par_iter()
507            .for_each(|(b, out)| fill_block(b, out));
508        #[cfg(not(feature = "native"))]
509        for (b, out) in slices {
510            fill_block(b, out);
511        }
512    }
513
514    ForwardIndex {
515        terms,
516        offsets,
517        num_terms,
518    }
519}
520
521// ── Recursive Graph Bisection ────────────────────────────────────────────
522
523/// CPU/depth budget for a BP pass. BP is an anytime algorithm: stopping at
524/// any depth or deadline still yields a valid permutation, and because the
525/// output layout becomes the next pass's input order, repeated budgeted
526/// passes warm-start and deepen (top levels converge in ~0 swaps, the budget
527/// flows to deeper levels).
528#[derive(Clone, Copy, Debug, Default)]
529pub struct BpBudget {
530    /// Stop recursion at partitions of at most this many docs instead of
531    /// descending to block granularity. `None` = full depth. Capping at
532    /// superblock granularity (superblock_size × block_size docs) keeps most
533    /// of the superblock-pruning win at ~⅓ less depth.
534    pub min_partition_docs: Option<usize>,
535    /// Wall-clock cap for the whole BP computation. The pass ends cleanly at
536    /// the deadline with whatever depth it reached (`converged = false`).
537    /// Ignored on wasm (no monotonic clock).
538    pub time_budget: Option<std::time::Duration>,
539}
540
541impl BpBudget {
542    /// Unbudgeted: full depth, no deadline.
543    pub fn full() -> Self {
544        Self::default()
545    }
546}
547
548/// Recursive graph bisection. Returns `(perm, converged)` where
549/// `perm[new_pos] = old_index` and `converged` is false iff the wall-clock
550/// budget ended the pass before it finished (a depth cap alone is a chosen
551/// target, not an interruption — it reports converged).
552///
553/// `min_partition_size` should be the BMP block_size (64).
554/// `max_iters` controls convergence (20 is standard).
555///
556/// Term IDs in the forward index must be compact (0..num_terms) so we can
557/// use flat arrays for O(1) degree lookups instead of hash maps.
558pub(crate) fn graph_bisection(
559    fwd: &ForwardIndex,
560    min_partition_size: usize,
561    max_iters: usize,
562    budget: BpBudget,
563) -> (Vec<u32>, bool) {
564    let n = fwd.num_docs();
565    if n == 0 {
566        return (Vec::new(), true);
567    }
568
569    let effective_min_partition = budget
570        .min_partition_docs
571        .unwrap_or(0)
572        .max(min_partition_size);
573
574    let mut docs: Vec<u32> = (0..n as u32).collect();
575    let depth = if effective_min_partition > 0 {
576        ((n as f64) / (effective_min_partition as f64))
577            .log2()
578            .ceil() as usize
579    } else {
580        0
581    };
582    let log_table = build_log_table(4096);
583
584    log::debug!(
585        "BP graph_bisection: n={}, min_partition={}, max_iters={}, depth=~{}, time_budget={:?}",
586        n,
587        effective_min_partition,
588        max_iters,
589        depth,
590        budget.time_budget,
591    );
592
593    #[cfg(feature = "native")]
594    let deadline = budget.time_budget.map(|d| std::time::Instant::now() + d);
595    #[cfg(not(feature = "native"))]
596    let deadline: Option<()> = None;
597    #[cfg(not(feature = "native"))]
598    let _ = deadline;
599
600    let exhausted = std::sync::atomic::AtomicBool::new(false);
601    #[cfg(feature = "native")]
602    bisect(
603        &mut docs,
604        fwd,
605        effective_min_partition,
606        max_iters,
607        &log_table,
608        deadline,
609        &exhausted,
610    );
611    #[cfg(not(feature = "native"))]
612    bisect(
613        &mut docs,
614        fwd,
615        effective_min_partition,
616        max_iters,
617        &log_table,
618        None,
619        &exhausted,
620    );
621
622    let converged = !exhausted.load(std::sync::atomic::Ordering::Relaxed);
623    if !converged {
624        log::info!(
625            "BP graph_bisection: wall-clock budget {:?} exhausted at n={} — emitting partial (still valid) permutation",
626            budget.time_budget,
627            n,
628        );
629    }
630    (docs, converged)
631}
632
633/// Recursive bisection of a document slice.
634///
635/// Uses flat `Vec<u32>` degree arrays indexed by compact term_id for cache-friendly
636/// O(1) lookups (vs FxHashMap which has poor cache locality at scale).
637///
638/// Gain computation is parallelized via rayon for large partitions (n > 4096).
639/// Adaptive iteration count reduces work at top levels where coarse splits
640/// converge faster and dominate total runtime.
641fn bisect(
642    docs: &mut [u32],
643    fwd: &ForwardIndex,
644    min_partition_size: usize,
645    max_iters: usize,
646    log_table: &[f32],
647    #[cfg(feature = "native")] deadline: Option<std::time::Instant>,
648    #[cfg(not(feature = "native"))] deadline: Option<()>,
649    exhausted: &std::sync::atomic::AtomicBool,
650) {
651    let n = docs.len();
652    if n <= min_partition_size {
653        return;
654    }
655    // Anytime cutoff: leave this subtree in its current (valid) order.
656    if exhausted.load(std::sync::atomic::Ordering::Relaxed) {
657        return;
658    }
659    #[cfg(feature = "native")]
660    if let Some(dl) = deadline
661        && std::time::Instant::now() >= dl
662    {
663        exhausted.store(true, std::sync::atomic::Ordering::Relaxed);
664        return;
665    }
666    #[cfg(not(feature = "native"))]
667    let _ = deadline;
668
669    let mid = n / 2;
670    let nt = fwd.num_terms;
671
672    // Adaptive iteration count: large partitions converge faster with
673    // coarse splits, so fewer refinement passes suffice. The fine-grained
674    // clustering is handled by deeper recursion levels with full iterations.
675    let effective_iters = if n > 100_000 {
676        max_iters.min(12)
677    } else {
678        max_iters
679    };
680
681    // Flat degree arrays: left_deg[term] and right_deg[term].
682    // Allocated per bisection level; freed on return before recursion uses the
683    // memory for sub-problems. Peak = O(num_terms × log2(n/min_partition_size)).
684    let mut left_deg = vec![0u32; nt];
685    let mut right_deg = vec![0u32; nt];
686
687    for (i, &doc) in docs.iter().enumerate() {
688        let target = if i < mid {
689            &mut left_deg
690        } else {
691            &mut right_deg
692        };
693        for &term in fwd.doc_terms(doc as usize) {
694            target[term as usize] += 1;
695        }
696    }
697
698    // Scratch buffers reused across iterations
699    let mut gains: Vec<f32> = vec![0.0; n];
700    let mut indices: Vec<usize> = (0..n).collect();
701    let mut new_left: Vec<u32> = Vec::with_capacity(mid);
702    let mut new_right: Vec<u32> = Vec::with_capacity(n - mid);
703
704    for iter in 0..effective_iters {
705        // Anytime cutoff between refinement passes: keep the current split.
706        #[cfg(feature = "native")]
707        if let Some(dl) = deadline
708            && std::time::Instant::now() >= dl
709        {
710            exhausted.store(true, std::sync::atomic::Ordering::Relaxed);
711            break;
712        }
713        // Compute gain for each document (approx_1 from Dhulipala et al.)
714        // Parallelized for large partitions where per-doc work dominates.
715        compute_gains(docs, fwd, mid, &left_deg, &right_deg, log_table, &mut gains);
716
717        // Partition: the `mid` LOWEST keys (strongest left affinity) go left
718        indices.clear();
719        indices.extend(0..n);
720        indices.select_nth_unstable_by(mid, |&a, &b| {
721            gains[a]
722                .partial_cmp(&gains[b])
723                .unwrap_or(std::cmp::Ordering::Equal)
724        });
725
726        // Apply partition, update degree arrays for swapped docs
727        new_left.clear();
728        new_right.clear();
729        let mut swap_count: usize = 0;
730
731        for (rank, &idx) in indices.iter().enumerate() {
732            let doc = docs[idx];
733            let was_left = idx < mid;
734            let now_left = rank < mid;
735
736            if now_left {
737                new_left.push(doc);
738            } else {
739                new_right.push(doc);
740            }
741
742            if was_left != now_left {
743                swap_count += 1;
744                for &term in fwd.doc_terms(doc as usize) {
745                    let t = term as usize;
746                    if was_left {
747                        left_deg[t] -= 1;
748                        right_deg[t] += 1;
749                    } else {
750                        right_deg[t] -= 1;
751                        left_deg[t] += 1;
752                    }
753                }
754            }
755        }
756
757        docs[..mid].copy_from_slice(&new_left);
758        docs[mid..].copy_from_slice(&new_right);
759
760        if swap_count == 0 {
761            break;
762        }
763
764        // Early termination: if < 0.5% of docs swapped, partition is stable
765        if iter > 2 && swap_count < n / 200 {
766            break;
767        }
768
769        // Cooling: break early if gains are negligible
770        if iter > 5 {
771            let max_gain = gains.iter().copied().fold(f32::NEG_INFINITY, f32::max);
772            if max_gain.abs() < 0.001 {
773                break;
774            }
775        }
776    }
777
778    // Drop scratch before recursion to free memory for sub-problems
779    drop(left_deg);
780    drop(right_deg);
781    drop(gains);
782    drop(indices);
783    drop(new_left);
784    drop(new_right);
785
786    let (left, right) = docs.split_at_mut(mid);
787    #[cfg(feature = "native")]
788    rayon::join(
789        || {
790            bisect(
791                left,
792                fwd,
793                min_partition_size,
794                max_iters,
795                log_table,
796                deadline,
797                exhausted,
798            )
799        },
800        || {
801            bisect(
802                right,
803                fwd,
804                min_partition_size,
805                max_iters,
806                log_table,
807                deadline,
808                exhausted,
809            )
810        },
811    );
812    #[cfg(not(feature = "native"))]
813    {
814        bisect(
815            left,
816            fwd,
817            min_partition_size,
818            max_iters,
819            log_table,
820            deadline,
821            exhausted,
822        );
823        bisect(
824            right,
825            fwd,
826            min_partition_size,
827            max_iters,
828            log_table,
829            deadline,
830            exhausted,
831        );
832    }
833}
834
835/// Compute gains for all documents, parallelized via rayon for large partitions.
836///
837/// Each doc's gain is independent: iterate its terms, accumulate the log-gap
838/// cost delta of moving it to the other side. Read-only access to degree arrays
839/// makes this embarrassingly parallel.
840#[inline(never)]
841fn compute_gains(
842    docs: &[u32],
843    fwd: &ForwardIndex,
844    mid: usize,
845    left_deg: &[u32],
846    right_deg: &[u32],
847    log_table: &[f32],
848    gains: &mut [f32],
849) {
850    // Single coherent key: HIGH = belongs in the RIGHT half.
851    // Left docs get +approx_one(from=left, to=right) — a misplaced left doc
852    // (terms concentrated right) scores high. Right docs get
853    // -approx_one(from=right, to=left) — a misplaced right doc scores low.
854    // This matches the reference two-sided formulation (compute_gains_left /
855    // compute_gains_right with negation); ranking both halves by raw
856    // "move gain" instead made both sides' misplaced docs rank identically,
857    // so the partition step could never exchange them.
858    let gain_for_doc = |i: usize| -> f32 {
859        let doc = docs[i] as usize;
860        let in_left = i < mid;
861        let mut g = 0.0f32;
862        for &term in fwd.doc_terms(doc) {
863            let t = term as usize;
864            let (from, to) = if in_left {
865                (left_deg[t], right_deg[t])
866            } else {
867                (right_deg[t], left_deg[t])
868            };
869            let move_gain = fast_log2_lookup(to as usize + 2, log_table)
870                - fast_log2_lookup(from as usize, log_table)
871                - std::f32::consts::LOG2_E / (1.0 + to as f32);
872            g += if in_left { move_gain } else { -move_gain };
873        }
874        g
875    };
876
877    #[cfg(feature = "native")]
878    {
879        if docs.len() > 4096 {
880            gains
881                .par_iter_mut()
882                .enumerate()
883                .for_each(|(i, gain)| *gain = gain_for_doc(i));
884        } else {
885            for (i, gain) in gains.iter_mut().enumerate().take(docs.len()) {
886                *gain = gain_for_doc(i);
887            }
888        }
889    }
890    #[cfg(not(feature = "native"))]
891    {
892        for (i, gain) in gains.iter_mut().enumerate().take(docs.len()) {
893            *gain = gain_for_doc(i);
894        }
895    }
896}
897
898// ── Helpers ──────────────────────────────────────────────────────────────
899
900/// Build precomputed log2 table for values 0..size.
901fn build_log_table(size: usize) -> Vec<f32> {
902    let mut table = vec![0.0f32; size];
903    // log2(0) is undefined; use a large negative value
904    table[0] = -10.0;
905    for (i, entry) in table.iter_mut().enumerate().skip(1) {
906        *entry = (i as f32).log2();
907    }
908    table
909}
910
911/// Fast log2 with precomputed table lookup.
912#[inline]
913fn fast_log2_lookup(val: usize, table: &[f32]) -> f32 {
914    if val < table.len() {
915        table[val]
916    } else {
917        (val as f32).log2()
918    }
919}
920
921#[cfg(test)]
922mod tests {
923    use super::*;
924
925    /// Build a simple forward index from (doc_id, terms) pairs.
926    fn make_fwd(docs: &[&[u32]], num_terms: usize) -> ForwardIndex {
927        let mut terms = Vec::new();
928        let mut offsets = vec![0u32];
929        for doc_terms in docs {
930            terms.extend_from_slice(doc_terms);
931            offsets.push(terms.len() as u32);
932        }
933        ForwardIndex {
934            terms,
935            offsets,
936            num_terms,
937        }
938    }
939
940    #[test]
941    fn test_bp_empty() {
942        let fwd = ForwardIndex {
943            terms: Vec::new(),
944            offsets: Vec::new(),
945            num_terms: 0,
946        };
947        let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
948        assert!(perm.is_empty());
949    }
950
951    #[test]
952    fn test_bp_small() {
953        // 4 docs, min_partition_size=4 → no bisection, identity
954        let fwd = make_fwd(&[&[0, 1], &[0, 2], &[1, 3], &[2, 3]], 4);
955        let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
956        assert_eq!(perm.len(), 4);
957        // All docs present
958        let mut sorted = perm.clone();
959        sorted.sort();
960        assert_eq!(sorted, vec![0, 1, 2, 3]);
961    }
962
963    #[test]
964    fn test_bp_clusters() {
965        // 8 docs in 2 clear clusters:
966        // Cluster A (docs 0-3): share terms 0, 1
967        // Cluster B (docs 4-7): share terms 2, 3
968        let fwd = make_fwd(
969            &[
970                &[0, 1],
971                &[0, 1],
972                &[0, 1],
973                &[0, 1],
974                &[2, 3],
975                &[2, 3],
976                &[2, 3],
977                &[2, 3],
978            ],
979            4,
980        );
981        let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
982        assert_eq!(perm.len(), 8);
983
984        // After bisection, docs from same cluster should be in same half
985        let left: Vec<u32> = perm[..4].to_vec();
986
987        // Either all of cluster A is in left and B in right, or vice versa
988        let a_in_left = left.iter().filter(|&&d| d < 4).count();
989        let b_in_left = left.iter().filter(|&&d| d >= 4).count();
990        assert!(
991            (a_in_left == 4 && b_in_left == 0) || (a_in_left == 0 && b_in_left == 4),
992            "Clusters should be separated: a_left={}, b_left={}",
993            a_in_left,
994            b_in_left,
995        );
996    }
997
998    #[test]
999    fn test_bp_permutation_valid() {
1000        // 16 docs with mixed terms: terms range from 0..4 and 10..18
1001        let docs: Vec<Vec<u32>> = (0..16).map(|i| vec![i / 4, 10 + i / 2]).collect();
1002        let doc_refs: Vec<&[u32]> = docs.iter().map(|v| v.as_slice()).collect();
1003        let fwd = make_fwd(&doc_refs, 18); // max term = 10 + 15/2 = 17, so need 18
1004        let (perm, _) = graph_bisection(&fwd, 4, 20, BpBudget::full());
1005
1006        assert_eq!(perm.len(), 16);
1007        // Must be a valid permutation
1008        let mut sorted = perm.clone();
1009        sorted.sort();
1010        let expected: Vec<u32> = (0..16).collect();
1011        assert_eq!(sorted, expected);
1012    }
1013
1014    /// Depth-capped BP: with min_partition_docs above the cluster size, only
1015    /// the top-level split happens — clusters still separate (coarse
1016    /// clustering), the permutation stays valid, and the pass converges
1017    /// (a depth cap is a chosen target, not an interruption).
1018    #[test]
1019    fn test_bp_depth_cap_separates_clusters_and_converges() {
1020        // Mostly-separated clusters with one misplaced doc per half — the
1021        // top-level swap pass must exchange docs 3 and 4.
1022        let fwd = make_fwd(
1023            &[
1024                &[0, 1],
1025                &[0, 1],
1026                &[0, 1],
1027                &[2, 3],
1028                &[0, 1],
1029                &[2, 3],
1030                &[2, 3],
1031                &[2, 3],
1032            ],
1033            4,
1034        );
1035        let budget = BpBudget {
1036            min_partition_docs: Some(4),
1037            time_budget: None,
1038        };
1039        let (perm, converged) = graph_bisection(&fwd, 2, 20, budget);
1040        assert!(converged, "depth cap must report converged");
1041        assert_eq!(perm.len(), 8);
1042        let mut sorted = perm.clone();
1043        sorted.sort();
1044        assert_eq!(
1045            sorted,
1046            (0..8).collect::<Vec<u32>>(),
1047            "must stay a valid permutation"
1048        );
1049        // Top-level split separates the clusters (docs {0,1,2,4} share terms
1050        // 0/1; docs {3,5,6,7} share terms 2/3)
1051        let cluster_a = [0u32, 1, 2, 4];
1052        let a_in_left = perm[..4].iter().filter(|d| cluster_a.contains(d)).count();
1053        assert!(
1054            a_in_left == 4 || a_in_left == 0,
1055            "clusters should separate at the top level: {:?}",
1056            perm
1057        );
1058    }
1059
1060    /// Zero wall-clock budget: the pass ends immediately, reports
1061    /// converged=false, and still emits a valid (identity) permutation.
1062    #[test]
1063    fn test_bp_zero_time_budget_emits_valid_partial_permutation() {
1064        let docs: Vec<Vec<u32>> = (0..64).map(|i| vec![i % 4]).collect();
1065        let doc_refs: Vec<&[u32]> = docs.iter().map(|v| v.as_slice()).collect();
1066        let fwd = make_fwd(&doc_refs, 4);
1067        let budget = BpBudget {
1068            min_partition_docs: None,
1069            time_budget: Some(std::time::Duration::ZERO),
1070        };
1071        let (perm, converged) = graph_bisection(&fwd, 4, 20, budget);
1072        assert!(!converged, "zero budget must report unconverged");
1073        assert_eq!(perm.len(), 64);
1074        let mut sorted = perm.clone();
1075        sorted.sort();
1076        assert_eq!(sorted, (0..64).collect::<Vec<u32>>());
1077    }
1078
1079    #[test]
1080    fn test_fast_log2() {
1081        let table = build_log_table(4096);
1082        assert!((table[1] - 0.0).abs() < 0.001);
1083        assert!((table[2] - 1.0).abs() < 0.001);
1084        assert!((table[4] - 2.0).abs() < 0.001);
1085        assert!((table[1024] - 10.0).abs() < 0.001);
1086        // Fallback for values beyond table
1087        let val = fast_log2_lookup(8192, &table);
1088        assert!((val - 13.0).abs() < 0.001);
1089    }
1090}