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