Skip to main content

zeppelin_embed/fts/
postings.rs

1//! The persisted posting format: bit-packed FOR blocks with positions.
2//!
3//! # Geometry, and why the block size is a header field
4//!
5//! Blocks hold
6//! [`DEFAULT_POSTINGS_PER_BLOCK`](crate::fts::postings::DEFAULT_POSTINGS_PER_BLOCK)
7//! postings. The research
8//! disagrees with itself usefully here: ~40 postings per block prunes best
9//! on GOV2 (3.6 ms against 4.2 ms at 128 — `research/02a:283`), but 128 is
10//! what makes SIMD decode clean in Lucene and tantivy. Both numbers are
11//! x86, on a 25M-document collection; neither transfers to a machine whose
12//! cache line is 128 bytes (`tasks/evidence/19-M0-platform-premises.md:14`).
13//!
14//! 64 is the judgement call: 64 sixteen-bit deltas is exactly one 128-byte
15//! line, and the block metadata row below is exactly 32 bytes, so four rows
16//! also fill one line and the array indexes by shift rather than by
17//! multiply. It is a cache-line argument, not a measurement, and it is
18//! **not** a size-saving choice.
19//!
20//! The block size is written into the region header, so re-tuning it later
21//! (task 27-B4 sweeps {32, 40, 64, 128}) changes a constant, never the
22//! format. Do not "fix" a golden by editing the constant.
23//!
24//! # Streams
25//!
26//! Structure-of-arrays, all little-endian, all hand-written:
27//!
28//! - **docid deltas**, frame-of-reference bit-packed per block;
29//! - **term frequencies**, bit-packed per block;
30//! - **positions**, delta-packed per posting, `tf` of them per posting —
31//!   the term frequency *is* the position count, so no separate count
32//!   stream exists.
33//!
34//! Positions are in the format from day one because retrofitting them is a
35//! format break (`research/03:484`). Task 15 reads them; task 13 only
36//! stores them.
37//!
38//! # Block maxima
39//!
40//! Each block carries a `u8` quantized ceiling on the BM25 contribution of
41//! any posting inside it. Task 14 reads it; task 13 writes it. Quantization
42//! **rounds up**, so a stored maximum is never below a true score — a bound
43//! that is too low would let pruning drop a document that belonged in the
44//! top-k, and that is a wrong answer, not a slow one.
45//!
46//! `u8` is enough: 8-bit uniform impact quantization is statistically
47//! indistinguishable from exact weights (Lin & Trotman, IRJ 2017). It is
48//! kept for metadata density per cache line, not to save bytes, and the
49//! same slot is deliberately shaped so a future caller-supplied impact
50//! could reuse it.
51
52use super::bm25::{DocLen, Tf};
53
54/// Postings per block. Persisted; see the module docs before changing it.
55pub const DEFAULT_POSTINGS_PER_BLOCK: u16 = 64;
56
57/// Bytes in one block-metadata row. Four rows fill a 128-byte cache line.
58pub const BLOCK_META_LEN: usize = 32;
59
60/// Magic prefixing an encoded posting list.
61const POSTINGS_MAGIC: [u8; 4] = *b"ZPST";
62
63/// The original encoded-postings format version.
64///
65/// Its six reserved metadata bytes are zero. It remains readable, and the
66/// frozen golden is a v1 stream, but the query path refuses it: a v1 block
67/// carries no impact pair, and a silent exhaustive fallback would make a
68/// stale index look correct and slow forever.
69pub const POSTINGS_VERSION_V1: u16 = 1;
70
71/// The current encoded-postings format version.
72///
73/// Version 2 spends the six reserved metadata bytes on the impact pair
74/// `(max_tf: u32, min_len: u16)`. See [`BlockImpact`] for why that is a
75/// soundness fix and not merely a wider bound.
76pub const POSTINGS_VERSION: u16 = 2;
77
78/// A rejected or malformed posting stream.
79#[derive(Clone, Copy, Debug, Eq, PartialEq)]
80pub enum PostingsError {
81    /// The byte stream ended inside a structure.
82    Truncated {
83        /// Bytes the decoder needed.
84        needed: usize,
85        /// Bytes actually available.
86        available: usize,
87    },
88    /// The leading magic did not match.
89    BadMagic,
90    /// The declared version is not readable.
91    UnsupportedVersion {
92        /// Version found in the stream.
93        found: u16,
94    },
95    /// A declared bit width exceeded 32.
96    BitWidthTooLarge {
97        /// The rejected width.
98        bits: u8,
99    },
100    /// The block size in the header was zero.
101    ZeroBlockSize,
102    /// Document identifiers were not strictly ascending.
103    DocidsNotAscending {
104        /// The offending identifier.
105        docid: u32,
106    },
107    /// Positions within one posting were not strictly ascending.
108    PositionsNotAscending {
109        /// The offending position.
110        position: u32,
111    },
112    /// A term frequency was zero, which cannot occur in a posting list.
113    ZeroTermFrequency,
114    /// The stream declared more postings than its blocks can hold.
115    InconsistentPostingCount {
116        /// Postings declared in the header.
117        declared: u32,
118        /// Postings the blocks actually describe.
119        described: u32,
120    },
121}
122
123impl std::fmt::Display for PostingsError {
124    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125        match self {
126            Self::Truncated { needed, available } => write!(
127                formatter,
128                "posting stream truncated: needed {needed} bytes, had {available}"
129            ),
130            Self::BadMagic => formatter.write_str("posting stream magic did not match"),
131            Self::UnsupportedVersion { found } => {
132                write!(formatter, "posting stream version {found} is not readable")
133            }
134            Self::BitWidthTooLarge { bits } => {
135                write!(formatter, "declared bit width {bits} exceeds 32")
136            }
137            Self::ZeroBlockSize => formatter.write_str("posting block size was zero"),
138            Self::DocidsNotAscending { docid } => {
139                write!(formatter, "document ids are not ascending at {docid}")
140            }
141            Self::PositionsNotAscending { position } => {
142                write!(formatter, "positions are not ascending at {position}")
143            }
144            Self::ZeroTermFrequency => formatter.write_str("a posting had a zero term frequency"),
145            Self::InconsistentPostingCount {
146                declared,
147                described,
148            } => write!(
149                formatter,
150                "header declares {declared} postings, blocks describe {described}"
151            ),
152        }
153    }
154}
155
156impl std::error::Error for PostingsError {}
157
158/// One document's entry in a posting list.
159#[derive(Clone, Debug, Eq, PartialEq)]
160pub struct Posting {
161    /// Segment-local dense row id.
162    pub docid: u32,
163    /// Occurrences of the term in the document. Always at least one.
164    pub tf: u32,
165    /// Token positions, strictly ascending. Length equals `tf`.
166    pub positions: Vec<u32>,
167}
168
169pub(crate) struct OwnedPosting<'m, C> {
170    pub(crate) posting: Posting,
171    pub(crate) _positions_charge: C,
172    pub(crate) marker: std::marker::PhantomData<&'m ()>,
173}
174
175/// An in-memory posting list for one term.
176#[derive(Clone, Debug, Default, Eq, PartialEq)]
177pub struct PostingList {
178    postings: Vec<Posting>,
179}
180
181impl PostingList {
182    /// Creates an empty list.
183    #[must_use]
184    pub const fn new() -> Self {
185        Self {
186            postings: Vec::new(),
187        }
188    }
189
190    /// Returns the postings, ascending by document id.
191    #[must_use]
192    pub fn postings(&self) -> &[Posting] {
193        &self.postings
194    }
195
196    pub(crate) fn resident_bytes(&self) -> usize {
197        self.postings
198            .capacity()
199            .saturating_mul(std::mem::size_of::<Posting>())
200            .saturating_add(
201                self.postings
202                    .iter()
203                    .map(|posting| posting.positions.capacity().saturating_mul(4))
204                    .fold(0_usize, usize::saturating_add),
205            )
206    }
207
208    /// Returns the number of documents containing the term.
209    #[must_use]
210    pub fn document_frequency(&self) -> u32 {
211        u32::try_from(self.postings.len()).unwrap_or(u32::MAX)
212    }
213
214    /// Returns true when no document contains the term.
215    #[must_use]
216    pub fn is_empty(&self) -> bool {
217        self.postings.is_empty()
218    }
219
220    /// Appends one posting.
221    ///
222    /// # Errors
223    ///
224    /// Returns [`PostingsError`] when document ids are not strictly
225    /// ascending, when positions are not strictly ascending, or when the
226    /// term frequency is zero. These are contract violations by the caller,
227    /// so they fail loudly rather than being repaired.
228    pub fn push(&mut self, posting: Posting) -> Result<(), PostingsError> {
229        if posting.tf == 0 || posting.positions.is_empty() {
230            return Err(PostingsError::ZeroTermFrequency);
231        }
232        if let Some(last) = self.postings.last()
233            && posting.docid <= last.docid
234        {
235            return Err(PostingsError::DocidsNotAscending {
236                docid: posting.docid,
237            });
238        }
239        Self::validate_positions(&posting)?;
240        self.postings.push(posting);
241        Ok(())
242    }
243
244    /// Inserts one posting at its document-id position, anywhere in the list.
245    ///
246    /// # Errors
247    ///
248    /// As [`Self::push`]; a document id already in the list is
249    /// [`PostingsError::DocidsNotAscending`], since it would break strict
250    /// ascent.
251    pub(crate) fn insert(&mut self, posting: Posting) -> Result<(), PostingsError> {
252        if posting.tf == 0 || posting.positions.is_empty() {
253            return Err(PostingsError::ZeroTermFrequency);
254        }
255        let index = self
256            .postings
257            .partition_point(|existing| existing.docid < posting.docid);
258        if self
259            .postings
260            .get(index)
261            .is_some_and(|existing| existing.docid == posting.docid)
262        {
263            return Err(PostingsError::DocidsNotAscending {
264                docid: posting.docid,
265            });
266        }
267        Self::validate_positions(&posting)?;
268        self.postings.insert(index, posting);
269        Ok(())
270    }
271
272    /// Removes the posting for `docid`, returning whether one was present.
273    pub(crate) fn remove(&mut self, docid: u32) -> bool {
274        match self
275            .postings
276            .binary_search_by_key(&docid, |posting| posting.docid)
277        {
278            Ok(index) => {
279                self.postings.remove(index);
280                true
281            }
282            Err(_) => false,
283        }
284    }
285
286    fn validate_positions(posting: &Posting) -> Result<(), PostingsError> {
287        let mut previous: Option<u32> = None;
288        for position in &posting.positions {
289            if let Some(earlier) = previous
290                && *position <= earlier
291            {
292                return Err(PostingsError::PositionsNotAscending {
293                    position: *position,
294                });
295            }
296            previous = Some(*position);
297        }
298        Ok(())
299    }
300}
301
302/// Bits needed to represent the largest value in `values`.
303fn required_bits_controlled<E>(
304    values: &[u32],
305    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
306) -> Result<u8, E> {
307    let mut maximum = 0;
308    for value in values {
309        work.step()?;
310        #[cfg(test)]
311        ENCODE_BITS_PROBES.with(|value| value.set(value.get() + 1));
312        maximum = maximum.max(*value);
313    }
314    Ok(if maximum == 0 {
315        0
316    } else {
317        // 32 - leading_zeros is in 1..=32 for a non-zero value.
318        u8::try_from(32 - maximum.leading_zeros()).unwrap_or(32)
319    })
320}
321
322/// Appends `values` to `output`, packed at `bits` each.
323fn pack_bits_controlled<E>(
324    values: &[u32],
325    bits: u8,
326    output: &mut Vec<u8>,
327    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
328) -> Result<(), E> {
329    if bits == 0 {
330        return Ok(());
331    }
332    let width = u32::from(bits);
333    let mut accumulator: u64 = 0;
334    let mut filled: u32 = 0;
335    for value in values {
336        work.step()?;
337        #[cfg(test)]
338        ENCODE_PACK_PROBES.with(|value| value.set(value.get() + 1));
339        accumulator |= u64::from(*value) << filled;
340        filled += width;
341        while filled >= 8 {
342            output.push((accumulator & 0xFF) as u8);
343            accumulator >>= 8;
344            filled -= 8;
345        }
346    }
347    if filled > 0 {
348        output.push((accumulator & 0xFF) as u8);
349    }
350    Ok(())
351}
352
353#[cfg(test)]
354fn pack_bits(values: &[u32], bits: u8, output: &mut Vec<u8>) {
355    let mut work = super::control::WorkCheck::new(|| Ok::<(), std::convert::Infallible>(()));
356    match pack_bits_controlled(values, bits, output, &mut work) {
357        Ok(()) => (),
358        Err(never) => match never {},
359    }
360}
361
362/// Bytes a packed run of `count` values at `bits` each occupies.
363const fn packed_len(count: usize, bits: u8) -> usize {
364    if bits == 0 {
365        return 0;
366    }
367    (count * bits as usize).div_ceil(8)
368}
369
370/// Unpacks `count` values at `bits` each from the front of `input`.
371///
372/// Returns the number of bytes consumed.
373///
374/// # Errors
375///
376/// Returns [`PostingsError::BitWidthTooLarge`] for a width above 32 and
377/// [`PostingsError::Truncated`] when `input` is too short. It never panics
378/// and never reads past `input`.
379pub fn unpack_bits(
380    input: &[u8],
381    bits: u8,
382    count: usize,
383    output: &mut [u32],
384) -> Result<usize, PostingsError> {
385    if bits > 32 {
386        return Err(PostingsError::BitWidthTooLarge { bits });
387    }
388    if bits == 0 {
389        for slot in output.iter_mut().take(count) {
390            *slot = 0;
391        }
392        return Ok(0);
393    }
394    let needed = packed_len(count, bits);
395    if input.len() < needed {
396        return Err(PostingsError::Truncated {
397            needed,
398            available: input.len(),
399        });
400    }
401    let width = u32::from(bits);
402    let mask = if bits == 32 {
403        u32::MAX
404    } else {
405        (1_u32 << width) - 1
406    };
407    let mut accumulator: u64 = 0;
408    let mut filled: u32 = 0;
409    let mut cursor = 0_usize;
410    for slot in output.iter_mut().take(count) {
411        while filled < width {
412            let byte = input.get(cursor).copied().unwrap_or(0);
413            cursor += 1;
414            accumulator |= u64::from(byte) << filled;
415            filled += 8;
416        }
417        *slot = (accumulator as u32) & mask;
418        accumulator >>= width;
419        filled -= width;
420    }
421    Ok(needed)
422}
423
424/// One block's metadata row, exactly [`BLOCK_META_LEN`] bytes on disk.
425#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
426pub struct BlockMeta {
427    /// Largest document id in the block; the skip key.
428    pub last_docid: u32,
429    /// Byte offset of the block's packed docid deltas.
430    pub docids_offset: u32,
431    /// Byte offset of the block's packed term frequencies.
432    pub tfs_offset: u32,
433    /// Byte offset of the block's packed positions.
434    pub positions_offset: u32,
435    /// Packed positions in the block.
436    pub positions_count: u32,
437    /// Postings in the block; the final block may be partial.
438    pub count: u16,
439    /// Bit width of the docid deltas.
440    pub docid_bits: u8,
441    /// Bit width of the term frequencies.
442    pub tf_bits: u8,
443    /// Bit width of the position deltas.
444    pub position_bits: u8,
445    /// Reserved for a caller-supplied impact. Written zero by this engine.
446    ///
447    /// Version 1 stored `quantize(score / ceiling)` here and computed the
448    /// pruning bound from it. That was unsound: the quotient bakes in
449    /// seal-time `avgdl` and seal-time `(k1, b)`, neither of which cancels,
450    /// so the stored bound fell below a true score once statistics drifted.
451    /// The slot is kept — its shape suits a caller-supplied learned impact —
452    /// but nothing in the engine reads it for a bound.
453    pub block_max: u8,
454    /// Largest term frequency in the block. Zero in a version 1 stream.
455    pub max_tf: u32,
456    /// Smallest document length in the block, saturating DOWN to `u16::MAX`.
457    ///
458    /// Down, because a shorter document scores higher: saturating upward
459    /// would put the stored bound below a true score.
460    pub min_len: u16,
461}
462
463impl BlockMeta {
464    fn write(&self, output: &mut Vec<u8>) {
465        output.extend_from_slice(&self.last_docid.to_le_bytes());
466        output.extend_from_slice(&self.docids_offset.to_le_bytes());
467        output.extend_from_slice(&self.tfs_offset.to_le_bytes());
468        output.extend_from_slice(&self.positions_offset.to_le_bytes());
469        output.extend_from_slice(&self.positions_count.to_le_bytes());
470        output.extend_from_slice(&self.count.to_le_bytes());
471        output.push(self.docid_bits);
472        output.push(self.tf_bits);
473        output.push(self.position_bits);
474        output.push(self.block_max);
475        // 4 + 4 + 4 + 4 + 4 + 2 + 4 written above is 26 bytes; the impact
476        // pair spends the remaining six exactly. A version 1 encode leaves
477        // both fields zero, which reproduces the original zero padding byte
478        // for byte, so the v1 golden is unmoved.
479        output.extend_from_slice(&self.max_tf.to_le_bytes());
480        output.extend_from_slice(&self.min_len.to_le_bytes());
481    }
482
483    /// Reads one metadata row from the front of `input`.
484    ///
485    /// # Errors
486    ///
487    /// Returns [`PostingsError::Truncated`] when fewer than
488    /// [`BLOCK_META_LEN`] bytes are available.
489    pub fn read(input: &[u8]) -> Result<Self, PostingsError> {
490        let row = input
491            .get(..BLOCK_META_LEN)
492            .ok_or(PostingsError::Truncated {
493                needed: BLOCK_META_LEN,
494                available: input.len(),
495            })?;
496        Ok(Self {
497            last_docid: read_u32(row, 0)?,
498            docids_offset: read_u32(row, 4)?,
499            tfs_offset: read_u32(row, 8)?,
500            positions_offset: read_u32(row, 12)?,
501            positions_count: read_u32(row, 16)?,
502            count: read_u16(row, 20)?,
503            docid_bits: read_u8(row, 22)?,
504            tf_bits: read_u8(row, 23)?,
505            position_bits: read_u8(row, 24)?,
506            block_max: read_u8(row, 25)?,
507            max_tf: read_u32(row, 26)?,
508            min_len: read_u16(row, 30)?,
509        })
510    }
511}
512
513fn read_u8(input: &[u8], at: usize) -> Result<u8, PostingsError> {
514    input.get(at).copied().ok_or(PostingsError::Truncated {
515        needed: at + 1,
516        available: input.len(),
517    })
518}
519
520fn read_u16(input: &[u8], at: usize) -> Result<u16, PostingsError> {
521    let bytes = input.get(at..at + 2).ok_or(PostingsError::Truncated {
522        needed: at + 2,
523        available: input.len(),
524    })?;
525    let mut buffer = [0_u8; 2];
526    buffer.copy_from_slice(bytes);
527    Ok(u16::from_le_bytes(buffer))
528}
529
530fn read_u32(input: &[u8], at: usize) -> Result<u32, PostingsError> {
531    let bytes = input.get(at..at + 4).ok_or(PostingsError::Truncated {
532        needed: at + 4,
533        available: input.len(),
534    })?;
535    let mut buffer = [0_u8; 4];
536    buffer.copy_from_slice(bytes);
537    Ok(u32::from_le_bytes(buffer))
538}
539
540/// An encoded posting list: header, block metadata rows, then the streams.
541#[derive(Clone, Debug, Default, Eq, PartialEq)]
542pub struct EncodedPostings {
543    bytes: Vec<u8>,
544}
545
546impl EncodedPostings {
547    /// Returns the encoded bytes.
548    #[must_use]
549    pub fn as_bytes(&self) -> &[u8] {
550        &self.bytes
551    }
552
553    /// Wraps already-encoded bytes.
554    #[must_use]
555    pub const fn from_bytes(bytes: Vec<u8>) -> Self {
556        Self { bytes }
557    }
558}
559
560/// Header length: magic, version, block size, posting and block counts.
561pub const HEADER_LEN: usize = 16;
562
563/// Encodes one posting list in the original version 1 layout.
564///
565/// `block_maxima` supplies the per-block reserved `u8` slot; pass an empty
566/// slice to store zeros. The impact-pair bytes stay zero, which is what
567/// keeps the frozen v1 golden byte-identical.
568///
569/// New writers use [`encode_v2`]. This entry point exists so the v1 golden
570/// stays exercised and so a v1 stream can still be produced deliberately.
571///
572/// # Errors
573///
574/// Returns [`PostingsError::ZeroBlockSize`] when the geometry is degenerate.
575pub fn encode(
576    list: &PostingList,
577    postings_per_block: u16,
578    block_maxima: &[u8],
579) -> Result<EncodedPostings, PostingsError> {
580    encode_versioned(
581        list.postings(),
582        postings_per_block,
583        block_maxima,
584        &[],
585        POSTINGS_VERSION_V1,
586        &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
587    )
588}
589
590/// Encodes one posting list in the version 2 layout, carrying impact pairs.
591///
592/// `impacts` supplies one [`BlockImpact`] per block, normally from
593/// [`block_impacts`]. A missing entry stores a zero pair, which a reader
594/// rejects rather than treating as an unbounded block.
595///
596/// # Errors
597///
598/// Returns [`PostingsError::ZeroBlockSize`] when the geometry is degenerate.
599pub fn encode_v2(
600    list: &PostingList,
601    postings_per_block: u16,
602    block_maxima: &[u8],
603    impacts: &[BlockImpact],
604) -> Result<EncodedPostings, PostingsError> {
605    encode_versioned(
606        list.postings(),
607        postings_per_block,
608        block_maxima,
609        impacts,
610        POSTINGS_VERSION,
611        &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
612    )
613}
614
615pub(crate) fn encode_v2_controlled<E: From<PostingsError>>(
616    list: &PostingList,
617    postings_per_block: u16,
618    block_maxima: &[u8],
619    impacts: &[BlockImpact],
620    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
621) -> Result<EncodedPostings, E> {
622    encode_versioned(
623        list.postings(),
624        postings_per_block,
625        block_maxima,
626        impacts,
627        POSTINGS_VERSION,
628        work,
629    )
630}
631
632pub(crate) struct GraphEncoded<'m, C> {
633    pub(crate) bytes: super::control::GuardedVec<'m, u8, C>,
634    pub(crate) summary: GraphListSummary,
635}
636
637#[derive(Clone, Copy, Eq, PartialEq)]
638pub(crate) struct GraphListSummary {
639    pub(crate) postings_per_block: u16,
640    pub(crate) block_count: u32,
641    pub(crate) doc_freq: u32,
642    pub(crate) docid_bytes: u32,
643    pub(crate) overall_max_tf: u32,
644    pub(crate) overall_min_len: u16,
645}
646
647pub(crate) fn encode_graph_policy<'m, P>(
648    postings: &[OwnedPosting<'m, P::Charge>],
649    lengths: &[u32],
650    policy: &mut P,
651) -> Result<GraphEncoded<'m, P::Charge>, P::Error>
652where
653    P: super::control::BuildPolicy<'m>,
654    P::Error: From<PostingsError>,
655{
656    use super::control::GuardedVec;
657    let per_block = usize::from(DEFAULT_POSTINGS_PER_BLOCK);
658    let block_count = postings.len().div_ceil(per_block);
659    let mut docids = GuardedVec::with_capacity(policy, 0)?;
660    let mut tfs = GuardedVec::with_capacity(policy, 0)?;
661    let mut positions = GuardedVec::with_capacity(policy, 0)?;
662    let mut metadata = GuardedVec::with_capacity(policy, block_count)?;
663    let mut previous_docid = 0_u32;
664    let mut overall_max_tf = 0_u32;
665    let mut overall_min_len = u16::MAX;
666
667    for (block, chunk) in postings.chunks(per_block).enumerate() {
668        policy.lexical_block()?;
669        let position_count = chunk
670            .iter()
671            .try_fold(0_usize, |total, owned| {
672                total
673                    .checked_add(owned.posting.positions.len())
674                    .ok_or(PostingsError::Truncated {
675                        needed: usize::MAX,
676                        available: total,
677                    })
678            })
679            .map_err(P::Error::from)?;
680        let mut deltas = GuardedVec::with_capacity(policy, chunk.len())?;
681        let mut frequencies = GuardedVec::with_capacity(policy, chunk.len())?;
682        let mut position_deltas = GuardedVec::with_capacity(policy, position_count)?;
683        let mut base = previous_docid;
684        let mut max_tf = 0_u32;
685        let mut min_len = usize::MAX;
686        for (offset, owned) in chunk.iter().enumerate() {
687            let posting = &owned.posting;
688            policy.lexical_posting()?;
689            if posting.tf == 0 || usize::try_from(posting.tf).ok() != Some(posting.positions.len())
690            {
691                return Err(P::Error::from(PostingsError::ZeroTermFrequency));
692            }
693            let delta = if block == 0 && offset == 0 {
694                posting.docid
695            } else {
696                let Some(delta) = posting.docid.checked_sub(base) else {
697                    return Err(P::Error::from(PostingsError::DocidsNotAscending {
698                        docid: posting.docid,
699                    }));
700                };
701                if delta == 0 {
702                    return Err(P::Error::from(PostingsError::DocidsNotAscending {
703                        docid: posting.docid,
704                    }));
705                }
706                delta
707            };
708            deltas.push(delta, policy)?;
709            frequencies.push(posting.tf, policy)?;
710            let mut previous = 0_u32;
711            for (slot, position) in posting.positions.iter().enumerate() {
712                policy.step(1)?;
713                let gap = if slot == 0 {
714                    *position
715                } else {
716                    let Some(gap) = position.checked_sub(previous) else {
717                        return Err(P::Error::from(PostingsError::PositionsNotAscending {
718                            position: *position,
719                        }));
720                    };
721                    if gap == 0 {
722                        return Err(P::Error::from(PostingsError::PositionsNotAscending {
723                            position: *position,
724                        }));
725                    }
726                    gap
727                };
728                position_deltas.push(gap, policy)?;
729                #[cfg(test)]
730                super::control::test_stage_probe(super::control::TestStage::PositionsEncode);
731                previous = *position;
732            }
733            base = posting.docid;
734            max_tf = max_tf.max(posting.tf);
735            let length = usize::try_from(posting.docid)
736                .ok()
737                .and_then(|row| lengths.get(row).copied())
738                .and_then(|value| usize::try_from(value).ok())
739                .unwrap_or(1);
740            min_len = min_len.min(length);
741        }
742        previous_docid = base;
743        let docid_bits = required_bits_policy(deltas.as_slice(), policy)?;
744        let tf_bits = required_bits_policy(frequencies.as_slice(), policy)?;
745        let position_bits = required_bits_policy(position_deltas.as_slice(), policy)?;
746        let meta = BlockMeta {
747            last_docid: base,
748            docids_offset: u32::try_from(docids.len()).unwrap_or(u32::MAX),
749            tfs_offset: u32::try_from(tfs.len()).unwrap_or(u32::MAX),
750            positions_offset: u32::try_from(positions.len()).unwrap_or(u32::MAX),
751            positions_count: u32::try_from(position_deltas.len()).unwrap_or(u32::MAX),
752            count: u16::try_from(chunk.len()).unwrap_or(u16::MAX),
753            docid_bits,
754            tf_bits,
755            position_bits,
756            block_max: 0,
757            max_tf,
758            min_len: u16::try_from(min_len).unwrap_or(u16::MAX),
759        };
760        pack_bits_policy(deltas.as_slice(), docid_bits, &mut docids, policy)?;
761        pack_bits_policy(frequencies.as_slice(), tf_bits, &mut tfs, policy)?;
762        pack_bits_policy(
763            position_deltas.as_slice(),
764            position_bits,
765            &mut positions,
766            policy,
767        )?;
768        metadata.push(meta, policy)?;
769        overall_max_tf = overall_max_tf.max(meta.max_tf);
770        overall_min_len = overall_min_len.min(meta.min_len);
771    }
772
773    let capacity = HEADER_LEN
774        .saturating_add(metadata.len().saturating_mul(BLOCK_META_LEN))
775        .saturating_add(docids.len())
776        .saturating_add(tfs.len())
777        .saturating_add(positions.len());
778    let mut bytes = GuardedVec::with_capacity(policy, capacity)?;
779    bytes.extend_from_slice(&POSTINGS_MAGIC, policy)?;
780    bytes.extend_from_slice(&POSTINGS_VERSION.to_le_bytes(), policy)?;
781    bytes.extend_from_slice(&DEFAULT_POSTINGS_PER_BLOCK.to_le_bytes(), policy)?;
782    bytes.extend_from_slice(
783        &u32::try_from(postings.len())
784            .unwrap_or(u32::MAX)
785            .to_le_bytes(),
786        policy,
787    )?;
788    bytes.extend_from_slice(
789        &u32::try_from(metadata.len())
790            .unwrap_or(u32::MAX)
791            .to_le_bytes(),
792        policy,
793    )?;
794    for meta in metadata.as_slice() {
795        write_meta_policy(meta, &mut bytes, policy)?;
796    }
797    bytes.extend_from_slice(docids.as_slice(), policy)?;
798    bytes.extend_from_slice(tfs.as_slice(), policy)?;
799    bytes.extend_from_slice(positions.as_slice(), policy)?;
800    Ok(GraphEncoded {
801        bytes,
802        summary: GraphListSummary {
803            postings_per_block: DEFAULT_POSTINGS_PER_BLOCK,
804            block_count: u32::try_from(metadata.len()).unwrap_or(u32::MAX),
805            doc_freq: u32::try_from(postings.len()).unwrap_or(u32::MAX),
806            docid_bytes: u32::try_from(docids.len()).unwrap_or(u32::MAX),
807            overall_max_tf,
808            overall_min_len,
809        },
810    })
811}
812
813fn required_bits_policy<'m, P: super::control::BuildPolicy<'m>>(
814    values: &[u32],
815    policy: &mut P,
816) -> Result<u8, P::Error> {
817    let mut maximum = 0_u32;
818    for value in values {
819        policy.step(1)?;
820        maximum = maximum.max(*value);
821    }
822    Ok(if maximum == 0 {
823        0
824    } else {
825        u8::try_from(32 - maximum.leading_zeros()).unwrap_or(32)
826    })
827}
828
829fn pack_bits_policy<'m, P: super::control::BuildPolicy<'m>>(
830    values: &[u32],
831    bits: u8,
832    output: &mut super::control::GuardedVec<'m, u8, P::Charge>,
833    policy: &mut P,
834) -> Result<(), P::Error> {
835    if bits == 0 {
836        return Ok(());
837    }
838    let mut accumulator = 0_u64;
839    let mut filled = 0_u32;
840    for value in values {
841        policy.step(1)?;
842        accumulator |= u64::from(*value) << filled;
843        filled = filled.saturating_add(u32::from(bits));
844        while filled >= 8 {
845            output.push((accumulator & 0xff) as u8, policy)?;
846            accumulator >>= 8;
847            filled -= 8;
848        }
849    }
850    if filled > 0 {
851        output.push((accumulator & 0xff) as u8, policy)?;
852    }
853    Ok(())
854}
855
856fn write_meta_policy<'m, P: super::control::BuildPolicy<'m>>(
857    meta: &BlockMeta,
858    output: &mut super::control::GuardedVec<'m, u8, P::Charge>,
859    policy: &mut P,
860) -> Result<(), P::Error> {
861    for bytes in [
862        meta.last_docid.to_le_bytes(),
863        meta.docids_offset.to_le_bytes(),
864        meta.tfs_offset.to_le_bytes(),
865        meta.positions_offset.to_le_bytes(),
866        meta.positions_count.to_le_bytes(),
867    ] {
868        output.extend_from_slice(&bytes, policy)?;
869    }
870    output.extend_from_slice(&meta.count.to_le_bytes(), policy)?;
871    for byte in [
872        meta.docid_bits,
873        meta.tf_bits,
874        meta.position_bits,
875        meta.block_max,
876    ] {
877        output.push(byte, policy)?;
878    }
879    output.extend_from_slice(&meta.max_tf.to_le_bytes(), policy)?;
880    output.extend_from_slice(&meta.min_len.to_le_bytes(), policy)
881}
882
883pub(crate) fn validate_graph_policy<'m, P>(
884    bytes: &[u8],
885    row_count: u32,
886    policy: &mut P,
887) -> Result<GraphListSummary, P::Error>
888where
889    P: super::control::BuildPolicy<'m>,
890    P::Error: From<PostingsError>,
891{
892    let header = bytes
893        .get(..HEADER_LEN)
894        .ok_or(PostingsError::Truncated {
895            needed: HEADER_LEN,
896            available: bytes.len(),
897        })
898        .map_err(P::Error::from)?;
899    if header.get(..4) != Some(&POSTINGS_MAGIC[..]) {
900        return Err(P::Error::from(PostingsError::BadMagic));
901    }
902    let version = read_u16(header, 4).map_err(P::Error::from)?;
903    if version != POSTINGS_VERSION {
904        return Err(P::Error::from(PostingsError::UnsupportedVersion {
905            found: version,
906        }));
907    }
908    let per_block = read_u16(header, 6).map_err(P::Error::from)?;
909    if per_block == 0 {
910        return Err(P::Error::from(PostingsError::ZeroBlockSize));
911    }
912    let posting_count = read_u32(header, 8).map_err(P::Error::from)?;
913    let block_count = read_u32(header, 12).map_err(P::Error::from)?;
914    let blocks = usize::try_from(block_count).unwrap_or(usize::MAX);
915    let meta_len = blocks
916        .checked_mul(BLOCK_META_LEN)
917        .ok_or(PostingsError::Truncated {
918            needed: usize::MAX,
919            available: bytes.len(),
920        })
921        .map_err(P::Error::from)?;
922    let meta_end = HEADER_LEN
923        .checked_add(meta_len)
924        .ok_or(PostingsError::Truncated {
925            needed: usize::MAX,
926            available: bytes.len(),
927        })
928        .map_err(P::Error::from)?;
929    let metadata = bytes
930        .get(HEADER_LEN..meta_end)
931        .ok_or(PostingsError::Truncated {
932            needed: meta_end,
933            available: bytes.len(),
934        })
935        .map_err(P::Error::from)?;
936    let mut described = 0_u32;
937    let mut docid_bytes = 0_usize;
938    let mut tf_bytes = 0_usize;
939    let mut position_bytes = 0_usize;
940    let mut overall_max_tf = 0_u32;
941    let mut overall_min_len = u16::MAX;
942    for index in 0..blocks {
943        policy.lexical_block()?;
944        let start = index.saturating_mul(BLOCK_META_LEN);
945        let meta = BlockMeta::read(metadata.get(start..).unwrap_or(&[])).map_err(P::Error::from)?;
946        for bits in [meta.docid_bits, meta.tf_bits, meta.position_bits] {
947            if bits > 32 {
948                return Err(P::Error::from(PostingsError::BitWidthTooLarge { bits }));
949            }
950        }
951        if meta.count == 0 || meta.count > per_block {
952            return Err(P::Error::from(PostingsError::InconsistentPostingCount {
953                declared: posting_count,
954                described: described.saturating_add(u32::from(meta.count)),
955            }));
956        }
957        described = described.saturating_add(u32::from(meta.count));
958        let count = usize::from(meta.count);
959        docid_bytes = docid_bytes.max(
960            usize::try_from(meta.docids_offset)
961                .unwrap_or(usize::MAX)
962                .saturating_add(packed_len(count, meta.docid_bits)),
963        );
964        tf_bytes = tf_bytes.max(
965            usize::try_from(meta.tfs_offset)
966                .unwrap_or(usize::MAX)
967                .saturating_add(packed_len(count, meta.tf_bits)),
968        );
969        position_bytes = position_bytes.max(
970            usize::try_from(meta.positions_offset)
971                .unwrap_or(usize::MAX)
972                .saturating_add(packed_len(
973                    usize::try_from(meta.positions_count).unwrap_or(usize::MAX),
974                    meta.position_bits,
975                )),
976        );
977        overall_max_tf = overall_max_tf.max(meta.max_tf);
978        overall_min_len = overall_min_len.min(meta.min_len);
979    }
980    if described != posting_count {
981        return Err(P::Error::from(PostingsError::InconsistentPostingCount {
982            declared: posting_count,
983            described,
984        }));
985    }
986    let streams = bytes
987        .get(meta_end..)
988        .ok_or(PostingsError::Truncated {
989            needed: meta_end,
990            available: bytes.len(),
991        })
992        .map_err(P::Error::from)?;
993    let tf_start = docid_bytes;
994    let position_start = tf_start
995        .checked_add(tf_bytes)
996        .ok_or(PostingsError::Truncated {
997            needed: usize::MAX,
998            available: streams.len(),
999        })
1000        .map_err(P::Error::from)?;
1001    let end = position_start
1002        .checked_add(position_bytes)
1003        .ok_or(PostingsError::Truncated {
1004            needed: usize::MAX,
1005            available: streams.len(),
1006        })
1007        .map_err(P::Error::from)?;
1008    if streams.len() < end {
1009        return Err(P::Error::from(PostingsError::Truncated {
1010            needed: end,
1011            available: streams.len(),
1012        }));
1013    }
1014    let docids = streams.get(..tf_start).unwrap_or(&[]);
1015    let tfs = streams.get(tf_start..position_start).unwrap_or(&[]);
1016    let positions = streams.get(position_start..end).unwrap_or(&[]);
1017    let mut previous_docid = 0_u32;
1018    for index in 0..blocks {
1019        policy.lexical_block()?;
1020        let start = index.saturating_mul(BLOCK_META_LEN);
1021        let meta = BlockMeta::read(metadata.get(start..).unwrap_or(&[])).map_err(P::Error::from)?;
1022        let docid_stream = docids
1023            .get(usize::try_from(meta.docids_offset).unwrap_or(usize::MAX)..)
1024            .ok_or(PostingsError::Truncated {
1025                needed: usize::MAX,
1026                available: docids.len(),
1027            })
1028            .map_err(P::Error::from)?;
1029        let tf_stream = tfs
1030            .get(usize::try_from(meta.tfs_offset).unwrap_or(usize::MAX)..)
1031            .ok_or(PostingsError::Truncated {
1032                needed: usize::MAX,
1033                available: tfs.len(),
1034            })
1035            .map_err(P::Error::from)?;
1036        let position_stream = positions
1037            .get(usize::try_from(meta.positions_offset).unwrap_or(usize::MAX)..)
1038            .ok_or(PostingsError::Truncated {
1039                needed: usize::MAX,
1040                available: positions.len(),
1041            })
1042            .map_err(P::Error::from)?;
1043        let mut position_cursor = 0_usize;
1044        for slot in 0..usize::from(meta.count) {
1045            policy.lexical_posting()?;
1046            let delta =
1047                packed_value_at(docid_stream, meta.docid_bits, slot).map_err(P::Error::from)?;
1048            let docid = if index == 0 && slot == 0 {
1049                delta
1050            } else {
1051                if delta == 0 {
1052                    return Err(P::Error::from(PostingsError::DocidsNotAscending {
1053                        docid: previous_docid,
1054                    }));
1055                }
1056                previous_docid
1057                    .checked_add(delta)
1058                    .ok_or(PostingsError::DocidsNotAscending {
1059                        docid: previous_docid,
1060                    })
1061                    .map_err(P::Error::from)?
1062            };
1063            if docid >= row_count {
1064                return Err(P::Error::from(PostingsError::DocidsNotAscending { docid }));
1065            }
1066            previous_docid = docid;
1067            let tf = packed_value_at(tf_stream, meta.tf_bits, slot).map_err(P::Error::from)?;
1068            if tf == 0 {
1069                return Err(P::Error::from(PostingsError::ZeroTermFrequency));
1070            }
1071            let count = usize::try_from(tf).unwrap_or(usize::MAX);
1072            let mut position = 0_u32;
1073            for offset in 0..count {
1074                policy.step(1)?;
1075                let gap = packed_value_at(
1076                    position_stream,
1077                    meta.position_bits,
1078                    position_cursor.saturating_add(offset),
1079                )
1080                .map_err(P::Error::from)?;
1081                if offset > 0 && gap == 0 {
1082                    return Err(P::Error::from(PostingsError::PositionsNotAscending {
1083                        position,
1084                    }));
1085                }
1086                position = position
1087                    .checked_add(gap)
1088                    .ok_or(PostingsError::PositionsNotAscending { position })
1089                    .map_err(P::Error::from)?;
1090                #[cfg(test)]
1091                super::control::test_stage_probe(super::control::TestStage::PositionsDecode);
1092            }
1093            position_cursor = position_cursor
1094                .checked_add(count)
1095                .ok_or(PostingsError::Truncated {
1096                    needed: usize::MAX,
1097                    available: usize::try_from(meta.positions_count).unwrap_or(usize::MAX),
1098                })
1099                .map_err(P::Error::from)?;
1100        }
1101        if position_cursor != usize::try_from(meta.positions_count).unwrap_or(usize::MAX)
1102            || previous_docid != meta.last_docid
1103        {
1104            return Err(P::Error::from(PostingsError::InconsistentPostingCount {
1105                declared: posting_count,
1106                described,
1107            }));
1108        }
1109    }
1110    Ok(GraphListSummary {
1111        postings_per_block: per_block,
1112        block_count,
1113        doc_freq: posting_count,
1114        docid_bytes: u32::try_from(docid_bytes).unwrap_or(u32::MAX),
1115        overall_max_tf,
1116        overall_min_len,
1117    })
1118}
1119
1120/// The one encoder. Version selects only what the reserved bytes carry.
1121fn encode_versioned<E: From<PostingsError>>(
1122    postings: &[Posting],
1123    postings_per_block: u16,
1124    block_maxima: &[u8],
1125    impacts: &[BlockImpact],
1126    version: u16,
1127    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
1128) -> Result<EncodedPostings, E> {
1129    work.check_now()?;
1130    if postings_per_block == 0 {
1131        return Err(PostingsError::ZeroBlockSize.into());
1132    }
1133    let per_block = usize::from(postings_per_block);
1134    let block_count = postings.len().div_ceil(per_block);
1135
1136    let mut docid_stream: Vec<u8> = Vec::new();
1137    let mut tf_stream: Vec<u8> = Vec::new();
1138    let mut position_stream: Vec<u8> = Vec::new();
1139    let mut metadata: Vec<BlockMeta> = Vec::with_capacity(block_count);
1140
1141    let mut previous_last_docid = 0_u32;
1142    for (index, chunk) in postings.chunks(per_block).enumerate() {
1143        work.step()?;
1144        let mut deltas: Vec<u32> = Vec::with_capacity(chunk.len());
1145        let mut frequencies: Vec<u32> = Vec::with_capacity(chunk.len());
1146        let mut position_deltas: Vec<u32> = Vec::new();
1147
1148        let mut base = previous_last_docid;
1149        for (offset, posting) in chunk.iter().enumerate() {
1150            work.step()?;
1151            // The very first posting of the list is stored absolutely; every
1152            // other delta is a gap from the previous document id.
1153            let delta = if index == 0 && offset == 0 {
1154                posting.docid
1155            } else {
1156                posting.docid.saturating_sub(base)
1157            };
1158            deltas.push(delta);
1159            frequencies.push(posting.tf);
1160            let mut previous_position = 0_u32;
1161            for (slot, position) in posting.positions.iter().enumerate() {
1162                work.step()?;
1163                #[cfg(test)]
1164                ENCODE_POSITION_PROBES.with(|value| value.set(value.get() + 1));
1165                let gap = if slot == 0 {
1166                    *position
1167                } else {
1168                    position.saturating_sub(previous_position)
1169                };
1170                position_deltas.push(gap);
1171                previous_position = *position;
1172            }
1173            base = posting.docid;
1174        }
1175        previous_last_docid = base;
1176
1177        let docid_bits = required_bits_controlled(&deltas, work)?;
1178        let tf_bits = required_bits_controlled(&frequencies, work)?;
1179        let position_bits = required_bits_controlled(&position_deltas, work)?;
1180
1181        let meta = BlockMeta {
1182            last_docid: base,
1183            docids_offset: u32::try_from(docid_stream.len()).unwrap_or(u32::MAX),
1184            tfs_offset: u32::try_from(tf_stream.len()).unwrap_or(u32::MAX),
1185            positions_offset: u32::try_from(position_stream.len()).unwrap_or(u32::MAX),
1186            positions_count: u32::try_from(position_deltas.len()).unwrap_or(u32::MAX),
1187            count: u16::try_from(chunk.len()).unwrap_or(u16::MAX),
1188            docid_bits,
1189            tf_bits,
1190            position_bits,
1191            block_max: block_maxima.get(index).copied().unwrap_or(0),
1192            max_tf: impacts.get(index).map_or(0, |impact| impact.max_tf),
1193            min_len: impacts.get(index).map_or(0, |impact| impact.min_len),
1194        };
1195
1196        pack_bits_controlled(&deltas, docid_bits, &mut docid_stream, work)?;
1197        pack_bits_controlled(&frequencies, tf_bits, &mut tf_stream, work)?;
1198        pack_bits_controlled(&position_deltas, position_bits, &mut position_stream, work)?;
1199        metadata.push(meta);
1200    }
1201
1202    let mut bytes = Vec::with_capacity(
1203        HEADER_LEN
1204            + metadata.len() * BLOCK_META_LEN
1205            + docid_stream.len()
1206            + tf_stream.len()
1207            + position_stream.len(),
1208    );
1209    bytes.extend_from_slice(&POSTINGS_MAGIC);
1210    bytes.extend_from_slice(&version.to_le_bytes());
1211    bytes.extend_from_slice(&postings_per_block.to_le_bytes());
1212    bytes.extend_from_slice(
1213        &u32::try_from(postings.len())
1214            .unwrap_or(u32::MAX)
1215            .to_le_bytes(),
1216    );
1217    bytes.extend_from_slice(
1218        &u32::try_from(metadata.len())
1219            .unwrap_or(u32::MAX)
1220            .to_le_bytes(),
1221    );
1222    for meta in &metadata {
1223        work.step()?;
1224        meta.write(&mut bytes);
1225    }
1226    super::control::extend_bytes(&mut bytes, &docid_stream, work)?;
1227    super::control::extend_bytes(&mut bytes, &tf_stream, work)?;
1228    super::control::extend_bytes(&mut bytes, &position_stream, work)?;
1229
1230    work.check_now()?;
1231    Ok(EncodedPostings { bytes })
1232}
1233
1234/// A validated view over an encoded posting list.
1235#[derive(Clone, Debug)]
1236pub struct PostingsReader<'bytes> {
1237    version: u16,
1238    blocks: Vec<BlockMeta>,
1239    docids: &'bytes [u8],
1240    tfs: &'bytes [u8],
1241    positions: &'bytes [u8],
1242    posting_count: u32,
1243    postings_per_block: u16,
1244}
1245
1246#[cfg(test)]
1247std::thread_local! {
1248    static ENCODE_POSITION_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1249    static ENCODE_BITS_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1250    static ENCODE_PACK_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1251    static ENCODE_IMPACT_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1252    pub(super) static POSITION_OPEN_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1253    pub(super) static POSITION_MEMBER_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1254    pub(crate) static POSITION_DECODE_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1255}
1256
1257impl<'bytes> PostingsReader<'bytes> {
1258    /// Locates only one row's position deltas. Other rows' positions are not
1259    /// decoded; their frequencies establish the offset within this block.
1260    ///
1261    /// # Errors
1262    /// Returns a typed posting error for malformed frequencies or geometry.
1263    pub fn positions(&self, row: u32) -> Result<Option<RowPositions<'bytes>>, PostingsError> {
1264        self.positions_controlled(
1265            row,
1266            &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
1267        )
1268    }
1269
1270    pub(crate) fn positions_controlled<E: From<PostingsError>>(
1271        &self,
1272        row: u32,
1273        work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
1274    ) -> Result<Option<RowPositions<'bytes>>, E> {
1275        work.check_now()?;
1276        let block = self.blocks.partition_point(|meta| meta.last_docid < row);
1277        let Some(meta) = self.blocks.get(block) else {
1278            return Ok(None);
1279        };
1280        let docids =
1281            self.docids
1282                .get(meta.docids_offset as usize..)
1283                .ok_or(PostingsError::Truncated {
1284                    needed: meta.docids_offset as usize,
1285                    available: self.docids.len(),
1286                })?;
1287        let tfs = self
1288            .tfs
1289            .get(meta.tfs_offset as usize..)
1290            .ok_or(PostingsError::Truncated {
1291                needed: meta.tfs_offset as usize,
1292                available: self.tfs.len(),
1293            })?;
1294        let mut docid = block
1295            .checked_sub(1)
1296            .and_then(|previous| self.blocks.get(previous))
1297            .map_or(0, |meta| meta.last_docid);
1298        let mut first = 0_usize;
1299        for slot in 0..usize::from(meta.count) {
1300            work.step()?;
1301            #[cfg(test)]
1302            POSITION_MEMBER_PROBES.with(|value| value.set(value.get() + 1));
1303            let delta = packed_value_at(docids, meta.docid_bits, slot)?;
1304            docid = docid
1305                .checked_add(delta)
1306                .ok_or(PostingsError::DocidsNotAscending { docid })?;
1307            if docid > row {
1308                return Ok(None);
1309            }
1310            let count = packed_value_at(tfs, meta.tf_bits, slot)? as usize;
1311            if count == 0 {
1312                return Err(PostingsError::ZeroTermFrequency.into());
1313            }
1314            let end = first.checked_add(count).ok_or(PostingsError::Truncated {
1315                needed: usize::MAX,
1316                available: meta.positions_count as usize,
1317            })?;
1318            if end > meta.positions_count as usize {
1319                return Err(PostingsError::Truncated {
1320                    needed: end,
1321                    available: meta.positions_count as usize,
1322                }
1323                .into());
1324            }
1325            if docid == row {
1326                let start = meta.positions_offset as usize;
1327                let end = start
1328                    .checked_add(packed_len(
1329                        meta.positions_count as usize,
1330                        meta.position_bits,
1331                    ))
1332                    .ok_or(PostingsError::Truncated {
1333                        needed: usize::MAX,
1334                        available: self.positions.len(),
1335                    })?;
1336                let bytes = self
1337                    .positions
1338                    .get(start..end)
1339                    .ok_or(PostingsError::Truncated {
1340                        needed: end,
1341                        available: self.positions.len(),
1342                    })?;
1343                return Ok(Some(RowPositions {
1344                    bytes,
1345                    bits: meta.position_bits,
1346                    first,
1347                    count,
1348                }));
1349            }
1350            first = end;
1351        }
1352        Ok(None)
1353    }
1354
1355    /// Validates and opens an encoded posting list.
1356    ///
1357    /// Every structural claim the bytes make is checked here, so the
1358    /// iteration methods below cannot fail on a reader that exists. Corrupt
1359    /// bytes produce a typed error and never a panic.
1360    ///
1361    /// # Errors
1362    ///
1363    /// Returns [`PostingsError`] for bad magic, an unreadable version, a
1364    /// zero block size, an oversized bit width, an inconsistent posting
1365    /// count, or truncation anywhere.
1366    pub fn open(bytes: &'bytes [u8]) -> Result<Self, PostingsError> {
1367        Self::open_controlled(
1368            bytes,
1369            &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
1370        )
1371    }
1372
1373    pub(crate) fn open_controlled<E: From<PostingsError>>(
1374        bytes: &'bytes [u8],
1375        work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
1376    ) -> Result<Self, E> {
1377        work.check_now()?;
1378        let header = bytes.get(..HEADER_LEN).ok_or(PostingsError::Truncated {
1379            needed: HEADER_LEN,
1380            available: bytes.len(),
1381        })?;
1382        if header.get(..4) != Some(&POSTINGS_MAGIC[..]) {
1383            return Err(PostingsError::BadMagic.into());
1384        }
1385        let version = read_u16(header, 4)?;
1386        if version != POSTINGS_VERSION_V1 && version != POSTINGS_VERSION {
1387            return Err(PostingsError::UnsupportedVersion { found: version }.into());
1388        }
1389        let postings_per_block = read_u16(header, 6)?;
1390        if postings_per_block == 0 {
1391            return Err(PostingsError::ZeroBlockSize.into());
1392        }
1393        let posting_count = read_u32(header, 8)?;
1394        let block_count = read_u32(header, 12)?;
1395
1396        let block_count_usize = usize::try_from(block_count).unwrap_or(usize::MAX);
1397        let metadata_len =
1398            block_count_usize
1399                .checked_mul(BLOCK_META_LEN)
1400                .ok_or(PostingsError::Truncated {
1401                    needed: usize::MAX,
1402                    available: bytes.len(),
1403                })?;
1404        let metadata_end =
1405            HEADER_LEN
1406                .checked_add(metadata_len)
1407                .ok_or(PostingsError::Truncated {
1408                    needed: usize::MAX,
1409                    available: bytes.len(),
1410                })?;
1411        let metadata_bytes =
1412            bytes
1413                .get(HEADER_LEN..metadata_end)
1414                .ok_or(PostingsError::Truncated {
1415                    needed: metadata_end,
1416                    available: bytes.len(),
1417                })?;
1418
1419        let mut blocks = Vec::with_capacity(block_count_usize);
1420        let mut described = 0_u32;
1421        let mut docid_bytes = 0_usize;
1422        let mut tf_bytes = 0_usize;
1423        let mut position_bytes = 0_usize;
1424        for index in 0..block_count_usize {
1425            work.step()?;
1426            #[cfg(test)]
1427            POSITION_OPEN_PROBES.with(|value| value.set(value.get() + 1));
1428            let start = index.saturating_mul(BLOCK_META_LEN);
1429            let row = metadata_bytes.get(start..start + BLOCK_META_LEN).ok_or(
1430                PostingsError::Truncated {
1431                    needed: start + BLOCK_META_LEN,
1432                    available: metadata_bytes.len(),
1433                },
1434            )?;
1435            let meta = BlockMeta::read(row)?;
1436            for bits in [meta.docid_bits, meta.tf_bits, meta.position_bits] {
1437                if bits > 32 {
1438                    return Err(PostingsError::BitWidthTooLarge { bits }.into());
1439                }
1440            }
1441            if meta.count == 0 || meta.count > postings_per_block {
1442                return Err(PostingsError::InconsistentPostingCount {
1443                    declared: posting_count,
1444                    described: described.saturating_add(u32::from(meta.count)),
1445                }
1446                .into());
1447            }
1448            described = described.saturating_add(u32::from(meta.count));
1449            let count = usize::from(meta.count);
1450            docid_bytes = docid_bytes.max(
1451                usize::try_from(meta.docids_offset).unwrap_or(usize::MAX)
1452                    + packed_len(count, meta.docid_bits),
1453            );
1454            tf_bytes = tf_bytes.max(
1455                usize::try_from(meta.tfs_offset).unwrap_or(usize::MAX)
1456                    + packed_len(count, meta.tf_bits),
1457            );
1458            position_bytes = position_bytes.max(
1459                usize::try_from(meta.positions_offset).unwrap_or(usize::MAX)
1460                    + packed_len(
1461                        usize::try_from(meta.positions_count).unwrap_or(usize::MAX),
1462                        meta.position_bits,
1463                    ),
1464            );
1465            blocks.push(meta);
1466        }
1467        if described != posting_count {
1468            return Err(PostingsError::InconsistentPostingCount {
1469                declared: posting_count,
1470                described,
1471            }
1472            .into());
1473        }
1474
1475        let streams = bytes.get(metadata_end..).ok_or(PostingsError::Truncated {
1476            needed: metadata_end,
1477            available: bytes.len(),
1478        })?;
1479        let tf_start = docid_bytes;
1480        let position_start = tf_start
1481            .checked_add(tf_bytes)
1482            .ok_or(PostingsError::Truncated {
1483                needed: usize::MAX,
1484                available: streams.len(),
1485            })?;
1486        let position_end =
1487            position_start
1488                .checked_add(position_bytes)
1489                .ok_or(PostingsError::Truncated {
1490                    needed: usize::MAX,
1491                    available: streams.len(),
1492                })?;
1493        if streams.len() < position_end {
1494            return Err(PostingsError::Truncated {
1495                needed: position_end,
1496                available: streams.len(),
1497            }
1498            .into());
1499        }
1500        let docids = streams.get(..tf_start).ok_or(PostingsError::Truncated {
1501            needed: tf_start,
1502            available: streams.len(),
1503        })?;
1504        let tfs = streams
1505            .get(tf_start..position_start)
1506            .ok_or(PostingsError::Truncated {
1507                needed: position_start,
1508                available: streams.len(),
1509            })?;
1510        let positions =
1511            streams
1512                .get(position_start..position_end)
1513                .ok_or(PostingsError::Truncated {
1514                    needed: position_end,
1515                    available: streams.len(),
1516                })?;
1517
1518        work.check_now()?;
1519        Ok(Self {
1520            version,
1521            blocks,
1522            docids,
1523            tfs,
1524            positions,
1525            posting_count,
1526            postings_per_block,
1527        })
1528    }
1529
1530    /// Returns the block metadata rows.
1531    #[must_use]
1532    pub fn blocks(&self) -> &[BlockMeta] {
1533        &self.blocks
1534    }
1535
1536    /// Returns the format version the stream declared.
1537    ///
1538    /// A caller that needs impact pairs checks this rather than inferring
1539    /// their presence from a zero pair, so a v1 stream is refused loudly
1540    /// instead of silently bounding nothing.
1541    #[must_use]
1542    pub const fn version(&self) -> u16 {
1543        self.version
1544    }
1545
1546    /// Returns the postings-per-block geometry recorded in the header.
1547    #[must_use]
1548    pub const fn postings_per_block(&self) -> u16 {
1549        self.postings_per_block
1550    }
1551
1552    /// Returns the total posting count.
1553    #[must_use]
1554    pub const fn document_frequency(&self) -> u32 {
1555        self.posting_count
1556    }
1557
1558    /// Decodes one block into document ids, frequencies, and positions.
1559    ///
1560    /// # Errors
1561    ///
1562    /// Returns [`PostingsError::Truncated`] when a stream is short. Bounds
1563    /// were validated at [`PostingsReader::open`], so a well-formed reader
1564    /// decodes every block successfully.
1565    pub fn decode_block(&self, index: usize) -> Result<Vec<Posting>, PostingsError> {
1566        let Some(meta) = self.blocks.get(index) else {
1567            return Ok(Vec::new());
1568        };
1569        let count = usize::from(meta.count);
1570        let mut deltas = vec![0_u32; count];
1571        let docid_start = usize::try_from(meta.docids_offset).unwrap_or(usize::MAX);
1572        let docid_slice = self
1573            .docids
1574            .get(docid_start..)
1575            .ok_or(PostingsError::Truncated {
1576                needed: docid_start,
1577                available: self.docids.len(),
1578            })?;
1579        unpack_bits(docid_slice, meta.docid_bits, count, &mut deltas)?;
1580
1581        let mut frequencies = vec![0_u32; count];
1582        let tf_start = usize::try_from(meta.tfs_offset).unwrap_or(usize::MAX);
1583        let tf_slice = self.tfs.get(tf_start..).ok_or(PostingsError::Truncated {
1584            needed: tf_start,
1585            available: self.tfs.len(),
1586        })?;
1587        unpack_bits(tf_slice, meta.tf_bits, count, &mut frequencies)?;
1588
1589        let position_count = usize::try_from(meta.positions_count).unwrap_or(usize::MAX);
1590        let mut position_deltas = vec![0_u32; position_count];
1591        let position_start = usize::try_from(meta.positions_offset).unwrap_or(usize::MAX);
1592        let position_slice =
1593            self.positions
1594                .get(position_start..)
1595                .ok_or(PostingsError::Truncated {
1596                    needed: position_start,
1597                    available: self.positions.len(),
1598                })?;
1599        unpack_bits(
1600            position_slice,
1601            meta.position_bits,
1602            position_count,
1603            &mut position_deltas,
1604        )?;
1605
1606        // Reconstruct absolute ids by prefix sum over the block.
1607        let mut base = if index == 0 {
1608            0
1609        } else {
1610            self.blocks
1611                .get(index.saturating_sub(1))
1612                .map_or(0, |previous| previous.last_docid)
1613        };
1614        let mut postings = Vec::with_capacity(count);
1615        let mut position_cursor = 0_usize;
1616        for slot in 0..count {
1617            let delta = deltas.get(slot).copied().unwrap_or(0);
1618            let docid = if index == 0 && slot == 0 {
1619                delta
1620            } else {
1621                base.saturating_add(delta)
1622            };
1623            base = docid;
1624            let tf = frequencies.get(slot).copied().unwrap_or(0);
1625            if tf == 0 {
1626                return Err(PostingsError::ZeroTermFrequency);
1627            }
1628            let tf_usize = usize::try_from(tf).unwrap_or(usize::MAX);
1629            let mut positions = Vec::with_capacity(tf_usize.min(position_count));
1630            let mut running = 0_u32;
1631            for step in 0..tf_usize {
1632                let Some(gap) = position_deltas.get(position_cursor + step).copied() else {
1633                    return Err(PostingsError::Truncated {
1634                        needed: position_cursor + step + 1,
1635                        available: position_deltas.len(),
1636                    });
1637                };
1638                running = if step == 0 {
1639                    gap
1640                } else {
1641                    running.saturating_add(gap)
1642                };
1643                positions.push(running);
1644            }
1645            position_cursor = position_cursor.saturating_add(tf_usize);
1646            postings.push(Posting {
1647                docid,
1648                tf,
1649                positions,
1650            });
1651        }
1652        Ok(postings)
1653    }
1654
1655    /// Decodes the whole list.
1656    ///
1657    /// # Errors
1658    ///
1659    /// Propagates any [`PostingsError`] from block decoding.
1660    pub fn decode_all(&self) -> Result<PostingList, PostingsError> {
1661        let mut list = PostingList::new();
1662        for index in 0..self.blocks.len() {
1663            for posting in self.decode_block(index)? {
1664                list.push(posting)?;
1665            }
1666        }
1667        Ok(list)
1668    }
1669}
1670
1671/// Borrowed packed positions for one requested posting. Locating this view
1672/// does not decode position values or allocate a position array.
1673#[derive(Clone, Copy, Debug)]
1674pub struct RowPositions<'bytes> {
1675    bytes: &'bytes [u8],
1676    bits: u8,
1677    first: usize,
1678    count: usize,
1679}
1680
1681impl RowPositions<'_> {
1682    /// Number of positions of the requested term in this row.
1683    #[must_use]
1684    pub const fn len(&self) -> usize {
1685        self.count
1686    }
1687
1688    /// Whether this view contains no positions.
1689    #[must_use]
1690    pub const fn is_empty(&self) -> bool {
1691        self.count == 0
1692    }
1693
1694    /// Decodes only the requested row, validating strict position ordering.
1695    ///
1696    /// # Errors
1697    /// Returns a typed posting error for truncation, invalid widths or order.
1698    pub fn decode(&self) -> Result<Vec<u32>, PostingsError> {
1699        self.decode_controlled(&mut super::control::WorkCheck::new(|| {
1700            Ok::<(), PostingsError>(())
1701        }))
1702    }
1703
1704    pub(crate) fn decode_controlled<E: From<PostingsError>>(
1705        &self,
1706        work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
1707    ) -> Result<Vec<u32>, E> {
1708        work.check_now()?;
1709        let mut positions = Vec::with_capacity(self.count);
1710        let mut position = 0_u32;
1711        let result = (|| {
1712            for offset in 0..self.count {
1713                work.step()?;
1714                #[cfg(test)]
1715                POSITION_DECODE_PROBES.with(|value| value.set(value.get() + 1));
1716                let delta = packed_value_at(self.bytes, self.bits, self.first + offset)?;
1717                if offset > 0 && delta == 0 {
1718                    return Err(PostingsError::PositionsNotAscending { position }.into());
1719                }
1720                position = position
1721                    .checked_add(delta)
1722                    .ok_or(PostingsError::PositionsNotAscending { position })?;
1723                positions.push(position);
1724            }
1725            work.check_now()
1726        })();
1727        // Record completed decode work even when the next cooperative check
1728        // aborts. Partial scratch is never returned to the caller.
1729        #[cfg(any(test, feature = "test-seams"))]
1730        if !positions.is_empty() {
1731            let first_bit = self.first * usize::from(self.bits);
1732            let last_bit = (self.first + positions.len()) * usize::from(self.bits);
1733            super::preparation_observer::phrase_positions(
1734                positions.len(),
1735                last_bit.div_ceil(8) - first_bit / 8,
1736            );
1737        }
1738        result?;
1739        Ok(positions)
1740    }
1741}
1742
1743/// A bit-addressed read can start mid-byte, including 32-bit values spanning
1744/// five bytes. It never expands preceding position values to find this one.
1745fn packed_value_at(bytes: &[u8], bits: u8, slot: usize) -> Result<u32, PostingsError> {
1746    if bits > 32 {
1747        return Err(PostingsError::BitWidthTooLarge { bits });
1748    }
1749    if bits == 0 {
1750        return Ok(0);
1751    }
1752    let first = slot
1753        .checked_mul(usize::from(bits))
1754        .ok_or(PostingsError::Truncated {
1755            needed: usize::MAX,
1756            available: bytes.len(),
1757        })?;
1758    let end = first
1759        .checked_add(usize::from(bits))
1760        .ok_or(PostingsError::Truncated {
1761            needed: usize::MAX,
1762            available: bytes.len(),
1763        })?
1764        .div_ceil(8);
1765    let selected = bytes.get(first / 8..end).ok_or(PostingsError::Truncated {
1766        needed: end,
1767        available: bytes.len(),
1768    })?;
1769    let word = selected
1770        .iter()
1771        .enumerate()
1772        .fold(0_u64, |word, (offset, byte)| {
1773            word | (u64::from(*byte) << (offset * 8))
1774        });
1775    let mask = (1_u64 << bits) - 1;
1776    Ok(((word >> (first % 8)) & mask) as u32)
1777}
1778
1779#[cfg(test)]
1780#[allow(clippy::expect_used)]
1781mod astra_18_tests {
1782    use super::*;
1783    use crate::fts::control::WorkCheck;
1784    use std::cell::Cell;
1785
1786    #[derive(Debug, PartialEq)]
1787    enum TestError {
1788        Cancelled,
1789        Postings(PostingsError),
1790    }
1791
1792    impl From<PostingsError> for TestError {
1793        fn from(error: PostingsError) -> Self {
1794            Self::Postings(error)
1795        }
1796    }
1797
1798    fn one_position_per_row() -> PostingList {
1799        let mut list = PostingList::new();
1800        for row in 0..4_096 {
1801            list.push(Posting {
1802                docid: row,
1803                tf: 1,
1804                positions: vec![row * 2],
1805            })
1806            .expect("literal posting");
1807        }
1808        list
1809    }
1810
1811    #[test]
1812    fn astra_18_position_metadata_cancels_inside_block_walk() {
1813        let list = one_position_per_row();
1814        let encoded = encode(&list, 1, &[]).expect("one row per block");
1815        POSITION_OPEN_PROBES.with(|value| value.set(0));
1816        let mut work = WorkCheck::new(|| {
1817            if POSITION_OPEN_PROBES.with(Cell::get) >= 64 {
1818                Err(TestError::Cancelled)
1819            } else {
1820                Ok(())
1821            }
1822        });
1823        let result = PostingsReader::open_controlled(encoded.as_bytes(), &mut work);
1824        let probes = POSITION_OPEN_PROBES.with(Cell::get);
1825        println!("metadata_probes={probes}");
1826        assert!(matches!(result, Err(TestError::Cancelled)));
1827        assert!(
1828            (64..=128).contains(&probes),
1829            "metadata walk must stop inside the list"
1830        );
1831        let mut clean = WorkCheck::new(|| Ok::<(), TestError>(()));
1832        let reader = PostingsReader::open_controlled(encoded.as_bytes(), &mut clean)
1833            .expect("fresh control opens complete metadata");
1834        assert_eq!(reader.blocks().len(), 4_096);
1835        assert_eq!(reader.document_frequency(), 4_096);
1836        assert_eq!(reader.decode_all().expect("complete literal output"), list);
1837        assert!(matches!(
1838            PostingsReader::open_controlled(&[], &mut clean),
1839            Err(TestError::Postings(PostingsError::Truncated { .. }))
1840        ));
1841    }
1842
1843    #[test]
1844    fn astra_18_position_membership_cancels_inside_one_block() {
1845        let list = one_position_per_row();
1846        let encoded = encode(&list, 4_096, &[]).expect("one large legal block");
1847        let reader = PostingsReader::open(encoded.as_bytes()).expect("reader");
1848        POSITION_MEMBER_PROBES.with(|value| value.set(0));
1849        let mut work = WorkCheck::new(|| {
1850            if POSITION_MEMBER_PROBES.with(Cell::get) >= 64 {
1851                Err(TestError::Cancelled)
1852            } else {
1853                Ok(())
1854            }
1855        });
1856        let result = reader.positions_controlled(4_095, &mut work);
1857        let probes = POSITION_MEMBER_PROBES.with(Cell::get);
1858        println!("membership_probes={probes}");
1859        assert!(matches!(result, Err(TestError::Cancelled)));
1860        assert!(
1861            (64..=128).contains(&probes),
1862            "membership must stop inside the block"
1863        );
1864        let mut clean = WorkCheck::new(|| Ok::<(), TestError>(()));
1865        assert_eq!(
1866            reader
1867                .positions_controlled(4_095, &mut clean)
1868                .expect("fresh control")
1869                .expect("last row")
1870                .decode()
1871                .expect("position"),
1872            vec![8_190]
1873        );
1874        assert!(
1875            reader
1876                .positions_controlled(4_096, &mut clean)
1877                .expect("absent row")
1878                .is_none()
1879        );
1880    }
1881
1882    #[test]
1883    fn astra_18_position_decode_cancels_inside_one_row() {
1884        let literal = (0..4_096).collect::<Vec<_>>();
1885        let mut list = PostingList::new();
1886        list.push(Posting {
1887            docid: 0,
1888            tf: 4_096,
1889            positions: literal.clone(),
1890        })
1891        .expect("one long row");
1892        let encoded = encode(&list, 64, &[]).expect("encoded positions");
1893        let reader = PostingsReader::open(encoded.as_bytes()).expect("reader");
1894        let view = reader.positions(0).expect("membership").expect("present");
1895        POSITION_DECODE_PROBES.with(|value| value.set(0));
1896        crate::fts::preparation_observer::begin();
1897        let mut work = WorkCheck::new(|| {
1898            if POSITION_DECODE_PROBES.with(Cell::get) >= 64 {
1899                Err(TestError::Cancelled)
1900            } else {
1901                Ok(())
1902            }
1903        });
1904        let result = view.decode_controlled(&mut work);
1905        let probes = POSITION_DECODE_PROBES.with(Cell::get);
1906        let observed = crate::fts::preparation_observer::phrase_position_work();
1907        crate::fts::preparation_observer::take();
1908        println!("decode_probes={probes}, position_work={observed:?}");
1909        assert_eq!(result, Err(TestError::Cancelled));
1910        assert!(
1911            (64..=128).contains(&probes),
1912            "decode must stop inside the row"
1913        );
1914        assert_eq!(
1915            observed,
1916            (probes, probes.div_ceil(8)),
1917            "retain actual partial decode work"
1918        );
1919        let mut clean = WorkCheck::new(|| Ok::<(), TestError>(()));
1920        assert_eq!(
1921            view.decode_controlled(&mut clean).expect("fresh control"),
1922            literal
1923        );
1924    }
1925}
1926
1927#[cfg(test)]
1928#[allow(clippy::expect_used, clippy::indexing_slicing)]
1929mod astra_11_tests {
1930    use super::*;
1931
1932    #[test]
1933    fn astra_11_position_reader_only_decodes_requested_row() {
1934        for per_block in [1, 2, 64] {
1935            let mut list = PostingList::new();
1936            let expected = (0..130)
1937                .map(|row| {
1938                    if row % 3 == 0 {
1939                        vec![0, 1, 31, u32::MAX]
1940                    } else {
1941                        vec![row * 3]
1942                    }
1943                })
1944                .collect::<Vec<_>>();
1945            for (row, positions) in expected.iter().enumerate() {
1946                list.push(Posting {
1947                    docid: row as u32 * 2,
1948                    tf: positions.len() as u32,
1949                    positions: positions.clone(),
1950                })
1951                .expect("posting");
1952            }
1953            let encoded = encode(&list, per_block, &[]).expect("encoded positions");
1954            let reader = PostingsReader::open(encoded.as_bytes()).expect("persisted reader");
1955            for row in [0, 1, 63, 64, 65, 129] {
1956                crate::fts::preparation_observer::begin();
1957                let view = reader
1958                    .positions(row as u32 * 2)
1959                    .expect("locate")
1960                    .expect("present");
1961                assert_eq!(
1962                    crate::fts::preparation_observer::phrase_position_work(),
1963                    (0, 0)
1964                );
1965                assert_eq!(view.decode().expect("selective decode"), expected[row]);
1966                assert_eq!(
1967                    crate::fts::preparation_observer::phrase_position_work().0,
1968                    expected[row].len()
1969                );
1970                crate::fts::preparation_observer::take();
1971                assert!(
1972                    reader
1973                        .positions(row as u32 * 2 + 1)
1974                        .expect("absent")
1975                        .is_none()
1976                );
1977            }
1978        }
1979    }
1980
1981    #[test]
1982    fn astra_11_corrupt_position_payload_is_typed() {
1983        let mut list = PostingList::new();
1984        for (docid, positions) in [(0, vec![1, 3]), (1, vec![2, 5])] {
1985            list.push(Posting {
1986                docid,
1987                tf: 2,
1988                positions,
1989            })
1990            .expect("posting");
1991        }
1992        let encoded = encode(&list, 64, &[]).expect("encoded");
1993        let clean = PostingsReader::open(encoded.as_bytes()).expect("open");
1994        assert_eq!(
1995            clean
1996                .positions(0)
1997                .expect("locate")
1998                .expect("present")
1999                .decode()
2000                .expect("clean"),
2001            vec![1, 3]
2002        );
2003        let positions_start = HEADER_LEN + BLOCK_META_LEN + clean.docids.len() + clean.tfs.len();
2004        assert_eq!(clean.blocks()[0].position_bits, 2);
2005        let mut corrupt = encoded.as_bytes().to_vec();
2006        // Delta two becomes zero in the actual packed payload, while all
2007        // structural lengths and row/frequency data remain intact.
2008        corrupt[positions_start] &= !(3 << 2);
2009        let reader = PostingsReader::open(&corrupt).expect("geometry unchanged");
2010        assert_eq!(
2011            reader
2012                .positions(0)
2013                .expect("locate corrupt row")
2014                .expect("present")
2015                .decode(),
2016            Err(PostingsError::PositionsNotAscending { position: 1 })
2017        );
2018        assert_eq!(
2019            reader
2020                .positions(1)
2021                .expect("next row")
2022                .expect("present")
2023                .decode()
2024                .expect("unaffected row"),
2025            vec![2, 5]
2026        );
2027    }
2028}
2029
2030/// One block's impact pair: the two extremes that bound every posting in it.
2031///
2032/// # Why a pair and not a stored score
2033///
2034/// A stored score is a number computed under one set of statistics. BM25's
2035/// `score / ceiling` is `tf / (tf + k1 * (1 - b + b * len / avgdl))`, so
2036/// `idf` cancels but `avgdl` and `(k1, b)` do not: a bound sealed when
2037/// documents were short falls below a true score once `avgdl` rises, and a
2038/// bound sealed under `beir()` falls below one read under `anserini()`.
2039/// Measured shortfalls were 15.21% and 19.56%.
2040///
2041/// A pair carries no statistics at all. `term_score` is monotone increasing
2042/// in `tf` and monotone decreasing in `len`, so evaluating it at the block's
2043/// largest `tf` and shortest document dominates every posting in the block
2044/// under ANY statistics and ANY `(k1, b)`. The two extremes need not come
2045/// from the same document; pairing them is looser than the true maximum and
2046/// is therefore still an upper bound.
2047#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
2048pub struct BlockImpact {
2049    /// Largest term frequency in the block.
2050    pub max_tf: u32,
2051    /// Smallest document length in the block, saturating DOWN.
2052    pub min_len: u16,
2053}
2054
2055impl BlockImpact {
2056    /// Reads the pair out of a persisted metadata row.
2057    #[must_use]
2058    pub const fn from_meta(meta: &BlockMeta) -> Self {
2059        Self {
2060            max_tf: meta.max_tf,
2061            min_len: meta.min_len,
2062        }
2063    }
2064
2065    /// Returns true when the pair bounds nothing.
2066    ///
2067    /// A real block has at least one posting, whose term frequency is at
2068    /// least one, so a zero `max_tf` can only mean the pair was never
2069    /// written — a version 1 stream, or a v2 stream missing an entry.
2070    #[must_use]
2071    pub const fn is_absent(&self) -> bool {
2072        self.max_tf == 0
2073    }
2074
2075    /// Evaluates the bound under the caller's live scoring constants.
2076    ///
2077    /// This is the whole remedy: the bound is computed HERE, at query time,
2078    /// from statistics that are current and parameters the caller chose.
2079    #[must_use]
2080    pub fn bound(&self, scorer: &super::bm25::TermScorer) -> f64 {
2081        scorer.score(Tf(self.max_tf), DocLen(u32::from(self.min_len)))
2082    }
2083}
2084
2085/// Computes the per-block impact pairs for one term's postings.
2086///
2087/// `lengths` is the dense per-row length array; a row beyond it contributes
2088/// the shortest possible length, which keeps the bound above rather than
2089/// below the truth.
2090#[must_use]
2091pub fn block_impacts(
2092    list: &PostingList,
2093    postings_per_block: u16,
2094    lengths: &[u32],
2095) -> Vec<BlockImpact> {
2096    let mut work = super::control::WorkCheck::new(|| Ok::<(), std::convert::Infallible>(()));
2097    match block_impacts_controlled(list, postings_per_block, lengths, &mut work) {
2098        Ok(impacts) => impacts,
2099        Err(never) => match never {},
2100    }
2101}
2102
2103pub(crate) fn block_impacts_controlled<E>(
2104    list: &PostingList,
2105    postings_per_block: u16,
2106    lengths: &[u32],
2107    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
2108) -> Result<Vec<BlockImpact>, E> {
2109    block_impacts_slice_controlled(list.postings(), postings_per_block, lengths, work)
2110}
2111
2112pub(crate) fn block_impacts_slice_controlled<E>(
2113    postings: &[Posting],
2114    postings_per_block: u16,
2115    lengths: &[u32],
2116    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
2117) -> Result<Vec<BlockImpact>, E> {
2118    work.check_now()?;
2119    if postings_per_block == 0 {
2120        return Ok(Vec::new());
2121    }
2122    let chunks = postings.chunks(usize::from(postings_per_block));
2123    let mut impacts = Vec::with_capacity(chunks.len());
2124    for chunk in chunks {
2125        let mut max_tf = 0_u32;
2126        let mut min_len = u32::MAX;
2127        for posting in chunk {
2128            work.step()?;
2129            #[cfg(test)]
2130            ENCODE_IMPACT_PROBES.with(|value| value.set(value.get() + 1));
2131            max_tf = max_tf.max(posting.tf);
2132            let length = usize::try_from(posting.docid)
2133                .ok()
2134                .and_then(|slot| lengths.get(slot).copied())
2135                .unwrap_or(1);
2136            min_len = min_len.min(length);
2137        }
2138        impacts.push(BlockImpact {
2139            max_tf,
2140            // Saturate DOWN: a shorter document scores higher, keeping
2141            // this bound at or above the true score.
2142            min_len: u16::try_from(min_len).unwrap_or(u16::MAX),
2143        });
2144    }
2145    work.check_now()?;
2146    Ok(impacts)
2147}
2148
2149#[cfg(test)]
2150#[allow(
2151    clippy::expect_used,
2152    clippy::indexing_slicing,
2153    clippy::panic,
2154    clippy::unwrap_used
2155)]
2156mod tests {
2157    use super::*;
2158
2159    #[test]
2160    fn astra_18_postings_encoding_cancels_inside_positions() {
2161        let positions = (0..4_096).collect::<Vec<_>>();
2162        let list = list_from(&[(0, &positions)]);
2163        let impacts = block_impacts(&list, 64, &[4_096]);
2164        let reference = encode_v2(&list, 64, &[], &impacts).expect("reference");
2165        let mut observed = Vec::new();
2166        for (phase, probes) in [
2167            ("position gaps", &ENCODE_POSITION_PROBES),
2168            ("bit widths", &ENCODE_BITS_PROBES),
2169            ("bit packing", &ENCODE_PACK_PROBES),
2170        ] {
2171            probes.with(|value| value.set(0));
2172            let mut work = super::super::control::WorkCheck::new(|| {
2173                if probes.with(std::cell::Cell::get) >= 64 {
2174                    Err(PostingsError::ZeroBlockSize)
2175                } else {
2176                    Ok(())
2177                }
2178            });
2179            let result = encode_v2_controlled(&list, 64, &[], &impacts, &mut work);
2180            let count = probes.with(std::cell::Cell::get);
2181            println!("{phase}: canceled={}, probes={count}", result.is_err());
2182            observed.push((
2183                phase,
2184                matches!(result, Err(PostingsError::ZeroBlockSize)),
2185                count,
2186            ));
2187        }
2188        let mut work = super::super::control::WorkCheck::new(|| Ok::<(), PostingsError>(()));
2189        let clean = encode_v2_controlled(&list, 64, &[], &impacts, &mut work).expect("clean");
2190        assert_eq!(clean.as_bytes(), reference.as_bytes());
2191        assert_eq!(
2192            PostingsReader::open(clean.as_bytes())
2193                .expect("reader")
2194                .decode_all()
2195                .expect("decoded"),
2196            list
2197        );
2198        for (phase, canceled, count) in observed {
2199            assert!(canceled, "{phase}");
2200            assert!((64..=128).contains(&count), "{phase}: {count}");
2201        }
2202    }
2203
2204    #[test]
2205    fn astra_18_block_impacts_cancel_inside_rows() {
2206        let mut list = PostingList::new();
2207        for docid in 0..4_096 {
2208            list.push(Posting {
2209                docid,
2210                tf: 1,
2211                positions: vec![0],
2212            })
2213            .expect("posting");
2214        }
2215        let lengths = vec![7; 4_096];
2216        ENCODE_IMPACT_PROBES.with(|value| value.set(0));
2217        let mut work = super::super::control::WorkCheck::new(|| {
2218            if ENCODE_IMPACT_PROBES.with(std::cell::Cell::get) >= 64 {
2219                Err("cancelled")
2220            } else {
2221                Ok(())
2222            }
2223        });
2224        let result = block_impacts_controlled(&list, 64, &lengths, &mut work);
2225        let count = ENCODE_IMPACT_PROBES.with(std::cell::Cell::get);
2226        println!("impact rows={count}");
2227        assert_eq!(result, Err("cancelled"));
2228        assert!((64..=128).contains(&count));
2229        let mut work = super::super::control::WorkCheck::new(|| Ok::<(), ()>(()));
2230        assert_eq!(
2231            block_impacts_controlled(&list, 64, &lengths, &mut work),
2232            Ok(vec![
2233                BlockImpact {
2234                    max_tf: 1,
2235                    min_len: 7
2236                };
2237                64
2238            ])
2239        );
2240    }
2241
2242    fn list_from(entries: &[(u32, &[u32])]) -> PostingList {
2243        let mut list = PostingList::new();
2244        for (docid, positions) in entries {
2245            list.push(Posting {
2246                docid: *docid,
2247                tf: u32::try_from(positions.len()).expect("small"),
2248                positions: positions.to_vec(),
2249            })
2250            .expect("ascending fixture");
2251        }
2252        list
2253    }
2254
2255    #[test]
2256    fn a_single_block_round_trips() {
2257        let list = list_from(&[(0, &[0, 5]), (3, &[2]), (9, &[1, 4, 7])]);
2258        let encoded = encode(&list, DEFAULT_POSTINGS_PER_BLOCK, &[]).expect("encodes");
2259        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
2260        assert_eq!(reader.document_frequency(), 3);
2261        assert_eq!(reader.postings_per_block(), DEFAULT_POSTINGS_PER_BLOCK);
2262        assert_eq!(reader.decode_all().expect("decodes"), list);
2263    }
2264
2265    #[test]
2266    fn many_blocks_round_trip_across_the_geometry_boundary() {
2267        for count in [1_usize, 63, 64, 65, 128, 200] {
2268            let mut list = PostingList::new();
2269            for index in 0..count {
2270                let docid = u32::try_from(index * 3).expect("small");
2271                list.push(Posting {
2272                    docid,
2273                    tf: 2,
2274                    positions: vec![index as u32, index as u32 + 7],
2275                })
2276                .expect("ascending");
2277            }
2278            let encoded = encode(&list, DEFAULT_POSTINGS_PER_BLOCK, &[]).expect("encodes");
2279            let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
2280            assert_eq!(
2281                reader.blocks().len(),
2282                count.div_ceil(usize::from(DEFAULT_POSTINGS_PER_BLOCK)),
2283                "block count for {count} postings"
2284            );
2285            assert_eq!(reader.decode_all().expect("decodes"), list, "count {count}");
2286        }
2287    }
2288
2289    #[test]
2290    fn the_block_metadata_row_is_exactly_one_quarter_of_a_cache_line() {
2291        let mut bytes = Vec::new();
2292        BlockMeta::default().write(&mut bytes);
2293        assert_eq!(bytes.len(), BLOCK_META_LEN);
2294        assert_eq!(BLOCK_META_LEN * 4, 128);
2295    }
2296
2297    #[test]
2298    fn an_empty_list_encodes_and_decodes() {
2299        let list = PostingList::new();
2300        let encoded = encode(&list, DEFAULT_POSTINGS_PER_BLOCK, &[]).expect("encodes");
2301        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
2302        assert_eq!(reader.document_frequency(), 0);
2303        assert!(reader.decode_all().expect("decodes").is_empty());
2304    }
2305
2306    #[test]
2307    fn bit_widths_at_every_boundary_round_trip() {
2308        for bits in 0..=32_u8 {
2309            let value = if bits == 0 {
2310                0
2311            } else if bits == 32 {
2312                u32::MAX
2313            } else {
2314                (1_u32 << bits) - 1
2315            };
2316            let values = vec![value; 64];
2317            let mut packed = Vec::new();
2318            pack_bits(&values, bits, &mut packed);
2319            let mut output = vec![0_u32; 64];
2320            let consumed = unpack_bits(&packed, bits, 64, &mut output).expect("unpacks");
2321            assert_eq!(consumed, packed_len(64, bits), "width {bits}");
2322            assert_eq!(output, values, "width {bits}");
2323        }
2324    }
2325
2326    #[test]
2327    fn an_oversized_bit_width_is_a_typed_error() {
2328        let mut output = [0_u32; 4];
2329        assert_eq!(
2330            unpack_bits(&[0; 32], 33, 4, &mut output),
2331            Err(PostingsError::BitWidthTooLarge { bits: 33 })
2332        );
2333    }
2334
2335    #[test]
2336    fn a_truncated_packed_run_is_a_typed_error() {
2337        let mut output = [0_u32; 64];
2338        let error = unpack_bits(&[0; 2], 8, 64, &mut output).expect_err("must refuse");
2339        assert!(matches!(error, PostingsError::Truncated { .. }));
2340    }
2341
2342    #[test]
2343    fn out_of_order_docids_are_refused_rather_than_sorted() {
2344        let mut list = PostingList::new();
2345        list.push(Posting {
2346            docid: 5,
2347            tf: 1,
2348            positions: vec![0],
2349        })
2350        .expect("first");
2351        assert_eq!(
2352            list.push(Posting {
2353                docid: 5,
2354                tf: 1,
2355                positions: vec![0],
2356            }),
2357            Err(PostingsError::DocidsNotAscending { docid: 5 })
2358        );
2359    }
2360
2361    #[test]
2362    fn out_of_order_positions_are_refused() {
2363        let mut list = PostingList::new();
2364        assert_eq!(
2365            list.push(Posting {
2366                docid: 0,
2367                tf: 2,
2368                positions: vec![4, 4],
2369            }),
2370            Err(PostingsError::PositionsNotAscending { position: 4 })
2371        );
2372    }
2373
2374    #[test]
2375    fn a_zero_frequency_posting_is_refused() {
2376        let mut list = PostingList::new();
2377        assert_eq!(
2378            list.push(Posting {
2379                docid: 0,
2380                tf: 0,
2381                positions: vec![],
2382            }),
2383            Err(PostingsError::ZeroTermFrequency)
2384        );
2385    }
2386
2387    #[test]
2388    fn the_impact_pair_survives_the_persisted_round_trip() {
2389        // The pair is what makes a bound sound under drift, so it has to
2390        // come back out of the bytes exactly as it went in.
2391        let list = list_from(&[(0, &[0, 1, 2]), (5, &[0]), (9, &[0, 3])]);
2392        let lengths = vec![40_u32, 0, 0, 0, 0, 7, 0, 0, 0, 900];
2393        let impacts = block_impacts(&list, 2, &lengths);
2394        assert_eq!(impacts.len(), 2);
2395        assert_eq!(
2396            impacts[0],
2397            BlockImpact {
2398                max_tf: 3,
2399                min_len: 7
2400            }
2401        );
2402        assert_eq!(
2403            impacts[1],
2404            BlockImpact {
2405                max_tf: 2,
2406                min_len: 900
2407            }
2408        );
2409
2410        let encoded = encode_v2(&list, 2, &[], &impacts).expect("encodes");
2411        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
2412        assert_eq!(reader.version(), POSTINGS_VERSION);
2413        let read: Vec<BlockImpact> = reader.blocks().iter().map(BlockImpact::from_meta).collect();
2414        assert_eq!(read, impacts);
2415        assert_eq!(reader.decode_all().expect("decodes"), list);
2416    }
2417
2418    #[test]
2419    fn a_version_one_stream_reads_back_with_no_impact_pair() {
2420        // v1 is still readable, and its blocks report an ABSENT pair rather
2421        // than a zero one that a caller might mistake for a real bound.
2422        let list = list_from(&[(0, &[0]), (4, &[1])]);
2423        let encoded = encode(&list, 64, &[]).expect("encodes");
2424        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
2425        assert_eq!(reader.version(), POSTINGS_VERSION_V1);
2426        for meta in reader.blocks() {
2427            assert!(BlockImpact::from_meta(meta).is_absent());
2428        }
2429    }
2430
2431    #[test]
2432    fn an_empty_impact_slice_stores_an_absent_pair() {
2433        let list = list_from(&[(0, &[0])]);
2434        let encoded = encode_v2(&list, 64, &[], &[]).expect("encodes");
2435        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
2436        assert_eq!(reader.version(), POSTINGS_VERSION);
2437        assert!(BlockImpact::from_meta(&reader.blocks()[0]).is_absent());
2438        assert!(block_impacts(&list, 0, &[]).is_empty());
2439    }
2440
2441    #[test]
2442    fn bad_magic_and_version_are_typed_errors() {
2443        let list = list_from(&[(0, &[0])]);
2444        let encoded = encode(&list, 64, &[]).expect("encodes");
2445        let mut bytes = encoded.as_bytes().to_vec();
2446        bytes[0] = b'X';
2447        assert_eq!(
2448            PostingsReader::open(&bytes).err(),
2449            Some(PostingsError::BadMagic)
2450        );
2451
2452        let mut bytes = encoded.as_bytes().to_vec();
2453        bytes[4] = 9;
2454        assert_eq!(
2455            PostingsReader::open(&bytes).err(),
2456            Some(PostingsError::UnsupportedVersion { found: 9 })
2457        );
2458    }
2459
2460    #[test]
2461    fn a_zero_block_size_header_is_refused() {
2462        let list = list_from(&[(0, &[0])]);
2463        let encoded = encode(&list, 64, &[]).expect("encodes");
2464        let mut bytes = encoded.as_bytes().to_vec();
2465        bytes[6] = 0;
2466        bytes[7] = 0;
2467        assert_eq!(
2468            PostingsReader::open(&bytes).err(),
2469            Some(PostingsError::ZeroBlockSize)
2470        );
2471        assert_eq!(
2472            encode(&list, 0, &[]).err(),
2473            Some(PostingsError::ZeroBlockSize)
2474        );
2475    }
2476
2477    #[test]
2478    fn every_truncation_of_a_valid_stream_is_a_typed_error_not_a_panic() {
2479        let list = list_from(&[(0, &[0, 3]), (7, &[1]), (100, &[2, 9, 40])]);
2480        let encoded = encode(&list, 2, &[]).expect("encodes");
2481        let bytes = encoded.as_bytes();
2482        for cut in 0..bytes.len() {
2483            let truncated = &bytes[..cut];
2484            // A prefix that still parses must decode without panic.
2485            if let Ok(reader) = PostingsReader::open(truncated) {
2486                for index in 0..reader.blocks().len() {
2487                    let _ = reader.decode_block(index);
2488                }
2489            }
2490        }
2491    }
2492}