Skip to main content

plugmem_core/index/
bm25.rs

1//! The lexical index: classic BM25 over delta-encoded postings
2//!
3//! Scoring is the standard formula with the Robertson idf:
4//!
5//! ```text
6//! idf(t)      = ln(1 + (N - df + 0.5) / (df + 0.5))
7//! tf_norm(d)  = tf · (k1 + 1) / (tf + k1 · (1 - b + b · len(d) / avg_len))
8//! score(d, q) = Σ_t idf(t) · tf_norm(d, t)
9//! ```
10//!
11//! A query decodes every query term's postings — O(Σ df), and there is no
12//! WAND-style pruning to avoid it. That is affordable because a decode is a
13//! few nanoseconds of varint over a contiguous chunk chain; what is not
14//! affordable is a *random* lookup per posting, and the scan is built to have
15//! none:
16//!
17//! - document lengths come from a flat array indexed by fact id, not from the
18//!   stored arena (see `Bm25Index::doc_len_dense`);
19//! - partial scores accumulate by merging sorted runs, because the postings
20//!   are already sorted by fact id, so no map is probed;
21//! - the caller's `live` predicate — a fact-record lookup on the engine side —
22//!   is asked only about documents in contention for the top `k`, since a
23//!   filter can remove entries from a ranking but never reorder it.
24//!
25//! The `decoded`, `scored` and `admitted` counters gate exactly this split in
26//! CI, and `bm25_probe_work_is_bounded` pins the last one at `k`.
27//!
28//! Deletions never touch the postings: tombstoned facts are filtered per
29//! candidate by the caller's `live` predicate and fall out physically
30//! when `maintain` rebuilds the index.
31
32use alloc::vec::Vec;
33
34#[cfg(feature = "counters")]
35use core::cell::Cell;
36
37use plugmem_arena::{Arena, ArenaCfg, ShardMode, Slot, key};
38
39use crate::error::Error;
40use crate::id::FactId;
41use crate::index::postings::PostingStore;
42
43/// Byte layout of [`DocLenSlot`]. Every offset is the previous field's offset
44/// plus its width, so a field cannot be moved by editing one number.
45mod doclen_at {
46    use core::mem::size_of;
47
48    pub(super) const FACT: usize = 0;
49    pub(super) const KEY_LEN: usize = FACT + size_of::<u32>();
50    pub(super) const LEN: usize = KEY_LEN;
51    pub(super) const DISTINCT: usize = LEN + size_of::<u16>();
52    pub(super) const SIG: usize = DISTINCT + size_of::<u16>();
53    pub(super) const SIZE: usize = SIG + size_of::<u64>();
54}
55
56/// Width of a [`DocLenSlot::sig`] signature in bits.
57const SIG_BITS: u32 = 64;
58/// Fibonacci hashing multiplier — the same constant the arena shards with, so
59/// term ids scatter over the signature's bits without a second hash family.
60const SIG_MULT: u64 = 0x9E37_79B9_7F4A_7C15;
61
62/// The signature bit a term claims. Distinct terms may collide; that is what
63/// makes [`DocLenSlot::sig`] an over-approximation and never an
64/// under-approximation of a document's term set.
65pub(crate) fn sig_bit(term: u32) -> u64 {
66    // The top `log2(SIG_BITS)` bits of the multiplied word, so the index is in
67    // range by construction.
68    let index = u64::from(term).wrapping_mul(SIG_MULT) >> (64 - SIG_BITS.trailing_zeros());
69    1u64 << index
70}
71
72/// Per-document record: `[fact 4 | len u16 | distinct u16 | sig u64]`,
73/// Uniform arena.
74///
75/// Beyond the length BM25 scores with, the slot carries a summary of the
76/// document's *term set*: how many distinct terms it has, and one bit per
77/// term hashed into a 64-bit word. That summary is what lets the write path
78/// bound the term-set overlap of two facts without re-reading and
79/// re-tokenizing their texts (see `Memory::find_similar`). It is written when
80/// the document is indexed, where the term set is already in hand, so it
81/// costs nothing to produce.
82#[derive(Clone, Copy, Debug, PartialEq, Eq)]
83#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
84pub struct DocLenSlot {
85    /// The document (fact) id — the key.
86    pub fact: FactId,
87    /// Token count of the document, saturated at `u16::MAX`.
88    pub len: u16,
89    /// Number of distinct terms, saturated at `u16::MAX`. Zero means
90    /// "unknown": a document indexed before the signature existed, read
91    /// through the legacy migration.
92    pub distinct: u16,
93    /// Union of `sig_bit` over the document's distinct terms. A term absent
94    /// from this word is definitely absent from the document; a term present
95    /// may still be absent (bits collide). Zero alongside `distinct == 0`
96    /// means "unknown".
97    pub sig: u64,
98}
99
100impl DocLenSlot {
101    /// Whether the term-set summary is present. A legacy document carries
102    /// none, and callers must fall back to reading its text.
103    pub fn has_signature(&self) -> bool {
104        self.distinct != 0
105    }
106
107    /// An upper bound on how many of `terms` this document also holds.
108    ///
109    /// Exact in the direction that matters: a term whose bit is clear cannot
110    /// be in the document, so the true intersection is never larger than the
111    /// count returned here. Callers use it to rule overlap *out*.
112    pub fn overlap_bound(&self, terms: &[u32]) -> usize {
113        terms
114            .iter()
115            .filter(|&&term| self.sig & sig_bit(term) != 0)
116            .count()
117    }
118}
119
120impl Slot for DocLenSlot {
121    const SIZE: usize = doclen_at::SIZE;
122    const KEY_LEN: usize = doclen_at::KEY_LEN;
123
124    fn write(&self, out: &mut [u8]) {
125        key::write_u32(&mut out[doclen_at::FACT..], self.fact.0);
126        out[doclen_at::LEN..doclen_at::DISTINCT].copy_from_slice(&self.len.to_be_bytes());
127        out[doclen_at::DISTINCT..doclen_at::SIG].copy_from_slice(&self.distinct.to_be_bytes());
128        out[doclen_at::SIG..doclen_at::SIZE].copy_from_slice(&self.sig.to_be_bytes());
129    }
130
131    fn read(bytes: &[u8]) -> Self {
132        Self {
133            fact: FactId(key::read_u32(&bytes[doclen_at::FACT..])),
134            len: u16::from_be_bytes(
135                bytes[doclen_at::LEN..doclen_at::DISTINCT]
136                    .try_into()
137                    .unwrap(),
138            ),
139            distinct: u16::from_be_bytes(
140                bytes[doclen_at::DISTINCT..doclen_at::SIG]
141                    .try_into()
142                    .unwrap(),
143            ),
144            sig: u64::from_be_bytes(bytes[doclen_at::SIG..doclen_at::SIZE].try_into().unwrap()),
145        }
146    }
147}
148
149/// Reusable query scratch: score accumulator and top-k selection buffer. One
150/// per concurrent reader; after warm-up a query allocates nothing (the
151/// zero-alloc recall invariant).
152#[derive(Debug, Default)]
153pub struct Bm25Scratch {
154    /// Partial scores as `(fact, score)` **sorted by fact id**.
155    ///
156    /// A map would be the obvious shape, and was the original one, but the
157    /// postings are already sorted by fact id: accumulating a term is then a
158    /// linear merge of two sorted runs instead of one hash probe per posting.
159    /// The difference is not the hashing — it is that a map big enough to hold
160    /// a frequent term's postings misses cache on essentially every probe,
161    /// while a merge walks three arrays forwards.
162    acc: Vec<(u32, f32)>,
163    /// Merge target, swapped with `acc` after each term past the first.
164    merge: Vec<(u32, f32)>,
165    /// Selection buffer for the top-k extraction.
166    top: Vec<(f32, u32)>,
167}
168
169impl Bm25Scratch {
170    /// Empty scratch buffers.
171    pub fn new() -> Self {
172        Self::default()
173    }
174}
175
176/// The BM25 index: postings with term frequencies plus per-document
177/// lengths and corpus statistics.
178#[derive(Debug)]
179pub struct Bm25Index<'a> {
180    postings: PostingStore<'a, true>,
181    doc_len: Arena<'a, DocLenSlot>,
182    total_docs: u64,
183    total_len: u64,
184    /// Posting entries decoded by queries (feature `counters`) — the
185    /// deterministic cost metric of the lexical source.
186    #[cfg(feature = "counters")]
187    decoded: Cell<u64>,
188    /// Documents whose BM25 contribution was actually evaluated — a document
189    /// length fetched and `tf_norm` computed (feature `counters`).
190    ///
191    /// Decoding a posting entry is a few nanoseconds of varint; *scoring* it
192    /// costs a document-length lookup, which is a random probe into an arena
193    /// that grows with the corpus. The two counters therefore measure
194    /// different things, and this is the one that dominates a large query.
195    #[cfg(feature = "counters")]
196    scored: Cell<u64>,
197    /// Calls to the caller's `live` predicate (feature `counters`).
198    ///
199    /// Every call is a fact-record lookup on the engine side — the second
200    /// random probe per candidate. A candidate that cannot reach the top `k`
201    /// should never cost one.
202    #[cfg(feature = "counters")]
203    admitted: Cell<u64>,
204    /// Some documents arrived without a term-set summary — this index came out
205    /// of a pre-signature image. Derived at load, never persisted: the next
206    /// compaction fills the summaries in and clears it.
207    unsummarized: bool,
208    /// Document length by fact id, [`DOC_LEN_ABSENT`] where the id names no
209    /// indexed document.
210    ///
211    /// Scoring needs the length of every document a query term's postings
212    /// name, which is one lookup per posting entry — the single hottest read
213    /// in the engine, and a random probe into an arena that deepens as the
214    /// corpus grows. Fact ids are dense and monotone, so a flat array answers
215    /// it by indexing, at four bytes per id.
216    ///
217    /// Runtime-only, like the arena's own page directory: rebuilt from
218    /// `doc_len` on load, never written to the snapshot. `doc_len` remains the
219    /// stored form and the only place the term-set summary lives.
220    doc_len_dense: Vec<u32>,
221    /// Exclusive upper bound of the fact ids `Bm25Index::doc_len_dense` may
222    /// be trusted for. `usize::MAX` while it has covered every document it was
223    /// offered; lowered to the first id it declined.
224    dense_limit: usize,
225}
226
227/// `Bm25Index::doc_len_dense` entry for a fact id with no indexed document.
228/// Lengths saturate at `u16::MAX`, so this cannot collide with a real one.
229const DOC_LEN_ABSENT: u32 = u32::MAX;
230
231impl<'a> Bm25Index<'a> {
232    /// Creates an empty index; `shards` per the engine config
233    /// (`shards_postings`), `max_bytes` bounds each underlying pool.
234    pub fn new(shards: usize, max_bytes: usize) -> Result<Self, Error> {
235        Ok(Self {
236            postings: PostingStore::new(shards, max_bytes)?,
237            doc_len: Arena::new(
238                ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(max_bytes),
239            )?,
240            total_docs: 0,
241            total_len: 0,
242            #[cfg(feature = "counters")]
243            decoded: Cell::new(0),
244            #[cfg(feature = "counters")]
245            scored: Cell::new(0),
246            #[cfg(feature = "counters")]
247            admitted: Cell::new(0),
248            unsummarized: false,
249            doc_len_dense: Vec::new(),
250            dense_limit: usize::MAX,
251        })
252    }
253
254    /// Indexes one document given its `(term, tf)` pairs (the caller
255    /// tokenizes and counts; pairs may arrive in any order, terms must be
256    /// unique). Documents must arrive in ascending fact-id order.
257    ///
258    /// # Errors
259    ///
260    /// [`Error::Arena`] when a pool hits its byte ceiling; the index may
261    /// then hold the document partially — the engine treats that as fatal
262    /// for the whole operation (journal replay rebuilds consistently).
263    pub fn index_doc(&mut self, fact: FactId, term_tfs: &[(u32, u8)]) -> Result<(), Error> {
264        let mut len = 0u32;
265        let mut sig = 0u64;
266        for &(term, tf) in term_tfs {
267            self.postings.push(term, fact, tf)?;
268            len += u32::from(tf);
269            sig |= sig_bit(term);
270        }
271        let doc = DocLenSlot {
272            fact,
273            len: u16::try_from(len).unwrap_or(u16::MAX),
274            // The caller's pairs are already unique per term, so their count
275            // is the distinct-term count.
276            distinct: u16::try_from(term_tfs.len()).unwrap_or(u16::MAX),
277            sig,
278        };
279        self.doc_len.insert(&doc)?;
280        self.note_dense(&doc);
281        self.total_docs += 1;
282        self.total_len += u64::from(len);
283        Ok(())
284    }
285
286    /// Records a document's length in the flat index, growing it to reach the
287    /// id. Fact ids are dense and monotone, so the growth is amortized.
288    ///
289    /// Growth stops where the id space stops being dense — see
290    /// [`Bm25Index::dense_capacity`]. Nothing is lost when it does:
291    /// [`Bm25Index::doc_len_of`] reads the stored arena for an id the flat
292    /// index does not cover.
293    fn note_dense(&mut self, doc: &DocLenSlot) {
294        // A `u32` id always fits a `usize`, on wasm32 as on a 64-bit host; the
295        // bound that matters is the capacity below.
296        let at = doc.fact.0 as usize;
297        if at >= self.dense_capacity() {
298            // Declined. The array can never speak for this id, and documents
299            // do not arrive in id order (compaction walks a hashed arena), so
300            // a later id may extend the array right over this one — hence the
301            // watermark rather than a bare skip.
302            self.dense_limit = self.dense_limit.min(at);
303            return;
304        }
305        if at >= self.doc_len_dense.len() {
306            self.doc_len_dense.resize(at + 1, DOC_LEN_ABSENT);
307        }
308        self.doc_len_dense[at] = u32::from(doc.len);
309    }
310
311    /// Highest fact id the flat length index will grow to cover.
312    ///
313    /// The index is worth its four bytes per id only while ids are dense,
314    /// which they are by construction — they are handed out in order, and only
315    /// purged facts leave holes. An id far past the document count means the
316    /// space is sparse or the image is lying, and a flat array would then cost
317    /// more memory than the documents themselves; a snapshot is untrusted
318    /// input, and nothing range-checks the ids inside the stored records.
319    /// `usize` being 32 bits on wasm32 makes the same claim sharper there.
320    ///
321    /// This bounds memory, never correctness: an id past the cap is answered
322    /// from the arena.
323    /// Saturating throughout, which is also what makes `at + 1` safe wherever
324    /// `at < dense_capacity()` holds: the capacity never exceeds `usize::MAX`,
325    /// so an id below it is below `usize::MAX` too.
326    fn dense_capacity(&self) -> usize {
327        const SLACK: usize = 8;
328        let docs = self.total_docs.min(usize::MAX as u64) as usize;
329        docs.saturating_add(1).saturating_mul(SLACK)
330    }
331
332    /// Rebuilds the flat length index from the stored records — the load path,
333    /// and the last step of [`Self::compact_live`].
334    ///
335    /// Deriving it in bulk rather than record by record is what makes it whole:
336    /// [`Self::note_dense`] judges an id against a capacity that grows with
337    /// `total_docs`, which is right while documents arrive in ascending id
338    /// order (the write path) and wrong when they arrive hashed, because the
339    /// first few are then measured against a capacity of eight. Here the
340    /// document count is already final, so every id is judged against the same
341    /// bound whatever order the arena yields it in.
342    ///
343    /// Two passes and no owned copy of the corpus: the first finds the
344    /// watermark and how far the array has to reach, the second fills it.
345    fn rebuild_dense(&mut self) {
346        let cap = self.dense_capacity();
347        let Self {
348            doc_len,
349            doc_len_dense,
350            dense_limit,
351            ..
352        } = self;
353        doc_len_dense.clear();
354        *dense_limit = usize::MAX;
355        let mut reach = 0usize;
356        for doc in doc_len.iter() {
357            let at = doc.fact.0 as usize;
358            if at >= cap {
359                // Declined, exactly as `note_dense` declines it: the array can
360                // never speak for this id, so the watermark drops to it.
361                *dense_limit = (*dense_limit).min(at);
362            } else {
363                // `at < cap` bounds `at + 1` — see `dense_capacity`.
364                reach = reach.max(at + 1);
365            }
366        }
367        doc_len_dense.resize(reach, DOC_LEN_ABSENT);
368        for doc in doc_len.iter() {
369            let at = doc.fact.0 as usize;
370            if at < cap {
371                doc_len_dense[at] = u32::from(doc.len);
372            }
373        }
374    }
375
376    /// The per-document record of `fact`, or `None` when the document is not
377    /// indexed. Carries the term-set summary the write path bounds overlap
378    /// with — see [`DocLenSlot`].
379    pub fn doc(&self, fact: FactId) -> Option<DocLenSlot> {
380        self.doc_len.get(&fact.0.to_be_bytes())
381    }
382
383    /// Whether this index holds documents with no term-set summary, which
384    /// compaction can fill in from the postings. True only after opening a
385    /// pre-signature image.
386    pub(crate) fn needs_resummarize(&self) -> bool {
387        self.unsummarized
388    }
389
390    /// Marks the index as holding unsummarized documents (the load path, after
391    /// a legacy migration).
392    pub(crate) fn mark_unsummarized(&mut self) {
393        self.unsummarized = true;
394    }
395
396    /// Builds a compacted BM25 index by filtering this index's existing
397    /// postings and document lengths through `live`.
398    ///
399    /// This is the ordinary-maintenance path: it preserves the exact term ids
400    /// and term frequencies already indexed, so compaction does not have to
401    /// read and tokenize every live document again. A tokenizer migration must
402    /// use the text reindex path instead.
403    pub(crate) fn compact_live(
404        &self,
405        shards: usize,
406        max_bytes: usize,
407        mut live: impl FnMut(FactId) -> bool,
408    ) -> Result<Bm25Index<'static>, Error> {
409        let mut out = Bm25Index::new(shards, max_bytes)?;
410        // Documents carried over from a pre-signature image, and the id range
411        // they span. Compaction is the cheapest place to fill their term-set
412        // summaries in: it already walks every posting, which is the transpose
413        // of what a signature needs, so no text is read and nothing is
414        // tokenized.
415        let mut legacy = 0usize;
416        let mut max_fact = 0u32;
417        for doc in self.doc_len.iter() {
418            if live(doc.fact) {
419                out.doc_len.insert(&doc)?;
420                out.total_docs += 1;
421                out.total_len += u64::from(doc.len);
422                if !doc.has_signature() {
423                    legacy += 1;
424                    max_fact = max_fact.max(doc.fact.0);
425                }
426            }
427        }
428        // `(sig, distinct)` per fact id, dense because the transpose visits
429        // facts in posting order rather than document order. Allocated only
430        // when there is something to fill, and dropped with this call.
431        //
432        // Sized like the flat length index and for the same reasons: `usize`
433        // is 32 bits on wasm32, and the ids come from an untrusted image, so
434        // the array is allocated only for an id space that is actually dense.
435        // Declining costs speed and nothing else — an unsummarized document
436        // keeps reading its text.
437        let highest = max_fact as usize;
438        let mut rebuilt = if legacy > 0 && highest < out.dense_capacity() {
439            alloc::vec![(0u64, 0u16); highest + 1]
440        } else {
441            Vec::new()
442        };
443        for slot in self.postings.slots() {
444            for (fact, tf) in self.postings.entries(slot.key) {
445                if !live(fact) {
446                    continue;
447                }
448                out.postings.push(slot.key, fact, tf)?;
449                if let Some(entry) = rebuilt.get_mut(fact.0 as usize) {
450                    entry.0 |= sig_bit(slot.key);
451                    entry.1 = entry.1.saturating_add(1);
452                }
453            }
454        }
455        if !rebuilt.is_empty() {
456            out.fill_missing_signatures(&rebuilt);
457        }
458        // Last, with the document count final: the flat length index is derived
459        // state, and deriving it the way the load path does is what keeps a
460        // compacted index and a reopened one the same index. Building it inside
461        // the loop above judged the earliest documents against a capacity of
462        // eight — and compaction walks a hashed arena, so whichever id came
463        // first pinned the watermark there for the whole corpus.
464        out.rebuild_dense();
465        Ok(out)
466    }
467
468    /// Writes the recomputed term-set summaries of [`Self::compact_live`]
469    /// into the documents that arrived without one. Documents that already
470    /// carry a signature keep it: it was written from the exact term set the
471    /// indexer saw, and the transpose can only reproduce it.
472    ///
473    /// The compacted index never inherits [`Self::unsummarized`]. A document
474    /// still unsummarized after this pass has no indexed terms at all, so no
475    /// later pass could summarize it either — carrying the flag forward would
476    /// make every future `maintain` recompact for work that cannot be done.
477    fn fill_missing_signatures(&mut self, rebuilt: &[(u64, u16)]) {
478        let stale: Vec<DocLenSlot> = self
479            .doc_len
480            .iter()
481            .filter(|doc| !doc.has_signature())
482            .collect();
483        for mut doc in stale {
484            let Some(&(sig, distinct)) = rebuilt.get(doc.fact.0 as usize) else {
485                continue;
486            };
487            if distinct == 0 {
488                continue; // a document with no indexed terms has nothing to summarize
489            }
490            doc.sig = sig;
491            doc.distinct = distinct;
492            let Some(payload) = self.doc_len.payload_mut(&doc.fact.0.to_be_bytes()) else {
493                continue;
494            };
495            let mut full = [0u8; DocLenSlot::SIZE];
496            doc.write(&mut full);
497            payload.copy_from_slice(&full[DocLenSlot::KEY_LEN..]);
498        }
499    }
500
501    /// Document frequency of a term.
502    pub fn df(&self, term: u32) -> u32 {
503        self.postings.count(term)
504    }
505
506    /// Number of indexed documents.
507    pub fn docs(&self) -> u64 {
508        self.total_docs
509    }
510
511    /// Robertson idf for a term with document frequency `df` in this
512    /// corpus (monotonically decreasing in `df`, always positive).
513    pub fn idf(&self, df: u32) -> f32 {
514        let n = self.total_docs as f32;
515        let df = df as f32;
516        libm::logf(1.0 + (n - df + 0.5) / (df + 0.5))
517    }
518
519    /// Scores `terms` against the corpus and writes the top `k` live
520    /// documents into `out` (descending score, ties by ascending id).
521    /// `live` filters candidates (tombstones, as-of, tag allow-sets) —
522    /// filtered documents cost their posting decode but never rank.
523    ///
524    /// Duplicate query terms are the caller's choice: each occurrence
525    /// accumulates again (a term repeated in the query weighs more).
526    ///
527    /// Cost is O(Σ df) in *decodes*, which is the honest price of a lexical
528    /// scan, but the expensive part of a candidate is not its decode — it is
529    /// the two random lookups that used to follow it, one for the document
530    /// length and one for the `live` predicate. Neither is paid per candidate
531    /// any more: lengths come from a flat array, and `live` is asked only
532    /// about documents that are actually in contention for the top `k`.
533    pub fn search(
534        &self,
535        (k1, b): (f32, f32),
536        terms: &[u32],
537        k: usize,
538        live: &mut dyn FnMut(FactId) -> bool,
539        scratch: &mut Bm25Scratch,
540        out: &mut Vec<(FactId, f32)>,
541    ) {
542        out.clear();
543        if self.total_docs == 0 || k == 0 {
544            return;
545        }
546        let Bm25Scratch { acc, merge, top } = scratch;
547        acc.clear();
548        let avg_len = self.total_len as f32 / self.total_docs as f32;
549        #[cfg(feature = "counters")]
550        let (mut decoded, mut scored) = (0u64, 0u64);
551
552        for &term in terms {
553            let df = self.postings.count(term);
554            if df == 0 {
555                continue;
556            }
557            let idf = self.idf(df);
558            // A term's postings are already ascending by fact id and `acc`
559            // holds the same order, so accumulating is a merge of two sorted
560            // runs. The first term has nothing to merge against and fills
561            // `acc` directly.
562            let mut ahead = 0usize;
563            merge.clear();
564            for (fact, tf) in self.postings.entries(term) {
565                #[cfg(feature = "counters")]
566                {
567                    decoded += 1;
568                }
569                // Everything in `acc` below this posting keeps its score.
570                while let Some(&entry) = acc.get(ahead)
571                    && entry.0 < fact.0
572                {
573                    merge.push(entry);
574                    ahead += 1;
575                }
576                let carried = match acc.get(ahead) {
577                    Some(&entry) if entry.0 == fact.0 => {
578                        ahead += 1;
579                        Some(entry.1)
580                    }
581                    _ => None,
582                };
583                // A posting naming a document with no length record scores
584                // nothing — and must not create a candidate either.
585                let Some(len) = self.doc_len_of(fact) else {
586                    if let Some(score) = carried {
587                        merge.push((fact.0, score));
588                    }
589                    continue;
590                };
591                #[cfg(feature = "counters")]
592                {
593                    scored += 1;
594                }
595                let tf = f32::from(tf);
596                let norm = tf * (k1 + 1.0) / (tf + k1 * (1.0 - b + b * f32::from(len) / avg_len));
597                // Terms are summed in query order, the order the accumulating
598                // map used, so the float result is bit-for-bit the same.
599                merge.push((fact.0, carried.unwrap_or(0.0) + idf * norm));
600            }
601            merge.extend_from_slice(&acc[ahead.min(acc.len())..]);
602            core::mem::swap(acc, merge);
603        }
604        #[cfg(feature = "counters")]
605        {
606            self.decoded.set(self.decoded.get() + decoded);
607            self.scored.set(self.scored.get() + scored);
608        }
609
610        // Top-k. Ranking is by score alone, so `live` cannot change the order
611        // — only remove entries from it. Asking it about every candidate is
612        // therefore wasted work: rank first, then walk the ranking and ask
613        // only until `k` survivors are found. The result is the same set in
614        // the same order an exhaustive filter produces.
615        let order = |a: &(f32, u32), b: &(f32, u32)| b.0.total_cmp(&a.0).then(a.1.cmp(&b.1));
616        // Collecting every candidate and partitioning the whole thing was the
617        // shape that made a corpus-wide term expensive twice over: an 8-byte
618        // copy per scored document, then a quickselect across all of them, for
619        // an answer of `k`. Only the best `k` are kept, behind a running limit
620        // that rejects the rest with one comparison; compacting at twice `k`
621        // amortizes the partition to O(candidates).
622        //
623        // The kept set is the same one the full partition produced: the
624        // ordering is total (score descending, id ascending), ids are unique,
625        // so no candidate ties with the limit and none that fails it can belong
626        // to the best `k`.
627        // `k` is the caller's — `recall` clamps it, but this is a public entry
628        // point and nothing here may assume that. Saturating keeps the compaction
629        // threshold above `k` for every `k`: past `usize::MAX / 2` a wrapping
630        // double would land *below* it, and the partition at `k - 1` would then
631        // run off the end of a buffer it had just filled. Saturated, the branch
632        // simply never fires and every candidate is kept, which is the answer.
633        let cap = k.saturating_mul(2).max(2);
634        top.clear();
635        // The reserve is bounded by the candidates as well: at most one entry per
636        // scored document is ever pushed, so sizing on `k` alone would ask the
637        // allocator for an answer the corpus cannot supply.
638        top.reserve(cap.min(acc.len()));
639        let mut limit: Option<(f32, u32)> = None;
640        for &(id, score) in acc.iter() {
641            let entry = (score, id);
642            if limit.is_some_and(|worst| !order(&entry, &worst).is_lt()) {
643                continue;
644            }
645            top.push(entry);
646            if top.len() == cap {
647                top.select_nth_unstable_by(k - 1, order);
648                top.truncate(k);
649                limit = Some(top[k - 1]);
650            }
651        }
652        #[cfg(feature = "counters")]
653        let mut admitted = 0u64;
654        let mut consume = |band: &[(f32, u32)], out: &mut Vec<(FactId, f32)>| {
655            for &(score, id) in band {
656                if out.len() == k {
657                    return;
658                }
659                #[cfg(feature = "counters")]
660                {
661                    admitted += 1;
662                }
663                if live(FactId(id)) {
664                    out.push((FactId(id), score));
665                }
666            }
667        };
668
669        // The usual case: partition the `k` highest scores to the front,
670        // order them, and take the survivors. Linear, and it asks `live`
671        // about `k` documents rather than every candidate.
672        let band = k.min(top.len());
673        if band > 0 {
674            if band < top.len() {
675                top.select_nth_unstable_by(band - 1, order);
676            }
677            top[..band].sort_unstable_by(order);
678            consume(&top[..band], out);
679        }
680        // The band was thinned by tombstones or a filter. Order what is left
681        // in one pass and continue down it — the same total cost the
682        // exhaustive filter used to pay on every query, now only on a query
683        // that needs it. The band is the best `band` candidates and was just
684        // consumed, so ordering `acc` and resuming past it walks exactly the
685        // documents the exhaustive path would have reached, in the same order.
686        if out.len() < k && band < acc.len() {
687            acc.sort_unstable_by(|a, b| b.1.total_cmp(&a.1).then(a.0.cmp(&b.0)));
688            top.clear();
689            top.extend(acc[band..].iter().map(|&(id, score)| (score, id)));
690            consume(top, out);
691        }
692        #[cfg(feature = "counters")]
693        self.admitted.set(self.admitted.get() + admitted);
694    }
695
696    /// Length of the document `fact` names, or `None` when it names none.
697    ///
698    /// The flat index is authoritative below [`Bm25Index::dense_limit`],
699    /// because every stored document with an id there was written into it. At
700    /// or above that mark it may have holes it never filled, so the answer
701    /// comes from the stored arena — slower, and correct.
702    fn doc_len_of(&self, fact: FactId) -> Option<u16> {
703        let at = fact.0 as usize;
704        if at < self.dense_limit
705            && let Some(&len) = self.doc_len_dense.get(at)
706        {
707            return (len != DOC_LEN_ABSENT).then_some(len as u16);
708        }
709        self.doc_len.get(&fact.0.to_be_bytes()).map(|doc| doc.len)
710    }
711
712    /// Bytes held by the underlying pools.
713    pub fn pool_bytes(&self) -> usize {
714        self.postings.pool_bytes() + self.doc_len.pool_bytes()
715    }
716
717    /// Total token count across the corpus (persisted in the engine
718    /// state).
719    pub fn total_len(&self) -> u64 {
720        self.total_len
721    }
722
723    /// The underlying posting store (the persistence composer dumps it).
724    pub(crate) fn postings(&self) -> &PostingStore<'a, true> {
725        &self.postings
726    }
727
728    /// The per-document length arena (the persistence composer dumps it).
729    pub(crate) fn doc_len_arena(&self) -> &Arena<'a, DocLenSlot> {
730        &self.doc_len
731    }
732
733    /// Assembles an index from already-validated parts (the load path).
734    pub(crate) fn from_parts(
735        postings: PostingStore<'a, true>,
736        doc_len: Arena<'a, DocLenSlot>,
737        total_docs: u64,
738        total_len: u64,
739    ) -> Self {
740        let mut index = Self {
741            postings,
742            doc_len,
743            total_docs,
744            total_len,
745            #[cfg(feature = "counters")]
746            decoded: Cell::new(0),
747            #[cfg(feature = "counters")]
748            scored: Cell::new(0),
749            #[cfg(feature = "counters")]
750            admitted: Cell::new(0),
751            unsummarized: false,
752            doc_len_dense: Vec::new(),
753            dense_limit: usize::MAX,
754        };
755        // One sequential pass over the stored records; the flat index is
756        // derived state and is never part of the image.
757        index.rebuild_dense();
758        index
759    }
760
761    /// Posting entries decoded so far (feature `counters`).
762    #[cfg(feature = "counters")]
763    pub fn decoded(&self) -> u64 {
764        self.decoded.get()
765    }
766
767    /// Documents scored so far — document-length fetches (feature
768    /// `counters`). See the [`Bm25Index::scored`] field docs for why this is
769    /// tracked apart from [`Bm25Index::decoded`].
770    #[cfg(feature = "counters")]
771    pub fn scored(&self) -> u64 {
772        self.scored.get()
773    }
774
775    /// `live` predicate calls so far (feature `counters`).
776    #[cfg(feature = "counters")]
777    pub fn admitted(&self) -> u64 {
778        self.admitted.get()
779    }
780
781    /// Resets the query work counters (feature `counters`).
782    #[cfg(feature = "counters")]
783    pub fn reset_query_counters(&self) {
784        self.decoded.set(0);
785        self.scored.set(0);
786        self.admitted.set(0);
787    }
788}
789
790#[cfg(test)]
791mod tests {
792    use super::*;
793
794    /// Builds `n` single-term documents with ascending ids, the order
795    /// [`Bm25Index::index_doc`] requires.
796    fn indexed(n: u32) -> Bm25Index<'static> {
797        let mut idx = Bm25Index::new(64, usize::MAX).unwrap();
798        for id in 0..n {
799            idx.index_doc(FactId(id), &[(id % 16, 1)]).unwrap();
800        }
801        idx
802    }
803
804    /// Compaction must hand the compacted index the same flat length cache
805    /// the load path would build from the same records.
806    #[test]
807    fn compaction_keeps_the_flat_length_index_dense() {
808        let idx = indexed(512);
809        let compacted = idx.compact_live(64, usize::MAX, |_| true).unwrap();
810        assert_eq!(compacted.total_docs, 512);
811        assert_eq!(
812            compacted.dense_limit,
813            usize::MAX,
814            "compaction declined ids the load path covers"
815        );
816        for id in 0..512u32 {
817            assert!(
818                (id as usize) < compacted.dense_limit
819                    && compacted.doc_len_dense.get(id as usize).copied() != Some(DOC_LEN_ABSENT),
820                "id {id} fell through to the arena"
821            );
822        }
823    }
824
825    /// Tombstoning most of a corpus leaves the survivors' ids sparse. The
826    /// array still covers them — the bound is generous by design — and the
827    /// lengths it reports are the survivors' own.
828    #[test]
829    fn compaction_covers_a_corpus_thinned_by_tombstones() {
830        let idx = indexed(512);
831        let compacted = idx
832            .compact_live(64, usize::MAX, |fact| fact.0.is_multiple_of(4))
833            .unwrap();
834        assert_eq!(compacted.total_docs, 128);
835        for id in (0..512u32).step_by(4) {
836            assert_eq!(compacted.doc_len_of(FactId(id)), Some(1));
837        }
838        assert_eq!(compacted.doc_len_of(FactId(1)), None);
839    }
840
841    /// The highest id a fact can carry, which on wasm32 is also `usize::MAX`
842    /// — the watermark's own "nothing was declined" sentinel, and the one
843    /// value where computing the array's reach as `at + 1` would overflow.
844    ///
845    /// Neither can happen, and for the same reason: the reach is only taken
846    /// for an id *below* the capacity, and the capacity never exceeds
847    /// `usize::MAX`, so `usize::MAX` is always declined instead. Declining it
848    /// leaves the watermark at `usize::MAX`, which is exactly right — it is an
849    /// exclusive bound, and every id strictly below it really is covered.
850    #[test]
851    fn the_highest_fact_id_is_declined_rather_than_overflowing_the_reach() {
852        let mut idx = Bm25Index::new(64, usize::MAX).unwrap();
853        idx.index_doc(FactId(0), &[(1, 1)]).unwrap();
854        idx.index_doc(FactId(u32::MAX), &[(1, 3)]).unwrap();
855        let compacted = idx.compact_live(64, usize::MAX, |_| true).unwrap();
856        assert_eq!(compacted.doc_len_of(FactId(0)), Some(1));
857        assert_eq!(compacted.doc_len_of(FactId(u32::MAX)), Some(3));
858    }
859
860    /// An id far past the document count is declined, and the watermark sends
861    /// every id at or above it to the arena — which answers correctly.
862    #[test]
863    fn a_sparse_id_is_declined_and_answered_by_the_arena() {
864        let mut idx = Bm25Index::new(64, usize::MAX).unwrap();
865        idx.index_doc(FactId(0), &[(1, 1)]).unwrap();
866        idx.index_doc(FactId(9_000), &[(1, 2)]).unwrap();
867        let compacted = idx.compact_live(64, usize::MAX, |_| true).unwrap();
868        assert_eq!(
869            compacted.dense_limit, 9_000,
870            "the sparse id should pin the watermark"
871        );
872        assert_eq!(compacted.doc_len_of(FactId(0)), Some(1));
873        assert_eq!(compacted.doc_len_of(FactId(9_000)), Some(2));
874    }
875}