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
169/// An in-memory posting list for one term.
170#[derive(Clone, Debug, Default, Eq, PartialEq)]
171pub struct PostingList {
172    postings: Vec<Posting>,
173}
174
175impl PostingList {
176    /// Creates an empty list.
177    #[must_use]
178    pub const fn new() -> Self {
179        Self {
180            postings: Vec::new(),
181        }
182    }
183
184    /// Returns the postings, ascending by document id.
185    #[must_use]
186    pub fn postings(&self) -> &[Posting] {
187        &self.postings
188    }
189
190    pub(crate) fn resident_bytes(&self) -> usize {
191        self.postings
192            .capacity()
193            .saturating_mul(std::mem::size_of::<Posting>())
194            .saturating_add(
195                self.postings
196                    .iter()
197                    .map(|posting| posting.positions.capacity().saturating_mul(4))
198                    .fold(0_usize, usize::saturating_add),
199            )
200    }
201
202    /// Returns the number of documents containing the term.
203    #[must_use]
204    pub fn document_frequency(&self) -> u32 {
205        u32::try_from(self.postings.len()).unwrap_or(u32::MAX)
206    }
207
208    /// Returns true when no document contains the term.
209    #[must_use]
210    pub fn is_empty(&self) -> bool {
211        self.postings.is_empty()
212    }
213
214    /// Appends one posting.
215    ///
216    /// # Errors
217    ///
218    /// Returns [`PostingsError`] when document ids are not strictly
219    /// ascending, when positions are not strictly ascending, or when the
220    /// term frequency is zero. These are contract violations by the caller,
221    /// so they fail loudly rather than being repaired.
222    pub fn push(&mut self, posting: Posting) -> Result<(), PostingsError> {
223        if posting.tf == 0 || posting.positions.is_empty() {
224            return Err(PostingsError::ZeroTermFrequency);
225        }
226        if let Some(last) = self.postings.last()
227            && posting.docid <= last.docid
228        {
229            return Err(PostingsError::DocidsNotAscending {
230                docid: posting.docid,
231            });
232        }
233        let mut previous: Option<u32> = None;
234        for position in &posting.positions {
235            if let Some(earlier) = previous
236                && *position <= earlier
237            {
238                return Err(PostingsError::PositionsNotAscending {
239                    position: *position,
240                });
241            }
242            previous = Some(*position);
243        }
244        self.postings.push(posting);
245        Ok(())
246    }
247}
248
249/// Bits needed to represent the largest value in `values`.
250fn required_bits_controlled<E>(
251    values: &[u32],
252    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
253) -> Result<u8, E> {
254    let mut maximum = 0;
255    for value in values {
256        work.step()?;
257        #[cfg(test)]
258        ENCODE_BITS_PROBES.with(|value| value.set(value.get() + 1));
259        maximum = maximum.max(*value);
260    }
261    Ok(if maximum == 0 {
262        0
263    } else {
264        // 32 - leading_zeros is in 1..=32 for a non-zero value.
265        u8::try_from(32 - maximum.leading_zeros()).unwrap_or(32)
266    })
267}
268
269/// Appends `values` to `output`, packed at `bits` each.
270fn pack_bits_controlled<E>(
271    values: &[u32],
272    bits: u8,
273    output: &mut Vec<u8>,
274    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
275) -> Result<(), E> {
276    if bits == 0 {
277        return Ok(());
278    }
279    let width = u32::from(bits);
280    let mut accumulator: u64 = 0;
281    let mut filled: u32 = 0;
282    for value in values {
283        work.step()?;
284        #[cfg(test)]
285        ENCODE_PACK_PROBES.with(|value| value.set(value.get() + 1));
286        accumulator |= u64::from(*value) << filled;
287        filled += width;
288        while filled >= 8 {
289            output.push((accumulator & 0xFF) as u8);
290            accumulator >>= 8;
291            filled -= 8;
292        }
293    }
294    if filled > 0 {
295        output.push((accumulator & 0xFF) as u8);
296    }
297    Ok(())
298}
299
300#[cfg(test)]
301fn pack_bits(values: &[u32], bits: u8, output: &mut Vec<u8>) {
302    let mut work = super::control::WorkCheck::new(|| Ok::<(), std::convert::Infallible>(()));
303    match pack_bits_controlled(values, bits, output, &mut work) {
304        Ok(()) => (),
305        Err(never) => match never {},
306    }
307}
308
309/// Bytes a packed run of `count` values at `bits` each occupies.
310const fn packed_len(count: usize, bits: u8) -> usize {
311    if bits == 0 {
312        return 0;
313    }
314    (count * bits as usize).div_ceil(8)
315}
316
317/// Unpacks `count` values at `bits` each from the front of `input`.
318///
319/// Returns the number of bytes consumed.
320///
321/// # Errors
322///
323/// Returns [`PostingsError::BitWidthTooLarge`] for a width above 32 and
324/// [`PostingsError::Truncated`] when `input` is too short. It never panics
325/// and never reads past `input`.
326pub fn unpack_bits(
327    input: &[u8],
328    bits: u8,
329    count: usize,
330    output: &mut [u32],
331) -> Result<usize, PostingsError> {
332    if bits > 32 {
333        return Err(PostingsError::BitWidthTooLarge { bits });
334    }
335    if bits == 0 {
336        for slot in output.iter_mut().take(count) {
337            *slot = 0;
338        }
339        return Ok(0);
340    }
341    let needed = packed_len(count, bits);
342    if input.len() < needed {
343        return Err(PostingsError::Truncated {
344            needed,
345            available: input.len(),
346        });
347    }
348    let width = u32::from(bits);
349    let mask = if bits == 32 {
350        u32::MAX
351    } else {
352        (1_u32 << width) - 1
353    };
354    let mut accumulator: u64 = 0;
355    let mut filled: u32 = 0;
356    let mut cursor = 0_usize;
357    for slot in output.iter_mut().take(count) {
358        while filled < width {
359            let byte = input.get(cursor).copied().unwrap_or(0);
360            cursor += 1;
361            accumulator |= u64::from(byte) << filled;
362            filled += 8;
363        }
364        *slot = (accumulator as u32) & mask;
365        accumulator >>= width;
366        filled -= width;
367    }
368    Ok(needed)
369}
370
371/// One block's metadata row, exactly [`BLOCK_META_LEN`] bytes on disk.
372#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
373pub struct BlockMeta {
374    /// Largest document id in the block; the skip key.
375    pub last_docid: u32,
376    /// Byte offset of the block's packed docid deltas.
377    pub docids_offset: u32,
378    /// Byte offset of the block's packed term frequencies.
379    pub tfs_offset: u32,
380    /// Byte offset of the block's packed positions.
381    pub positions_offset: u32,
382    /// Packed positions in the block.
383    pub positions_count: u32,
384    /// Postings in the block; the final block may be partial.
385    pub count: u16,
386    /// Bit width of the docid deltas.
387    pub docid_bits: u8,
388    /// Bit width of the term frequencies.
389    pub tf_bits: u8,
390    /// Bit width of the position deltas.
391    pub position_bits: u8,
392    /// Reserved for a caller-supplied impact. Written zero by this engine.
393    ///
394    /// Version 1 stored `quantize(score / ceiling)` here and computed the
395    /// pruning bound from it. That was unsound: the quotient bakes in
396    /// seal-time `avgdl` and seal-time `(k1, b)`, neither of which cancels,
397    /// so the stored bound fell below a true score once statistics drifted.
398    /// The slot is kept — its shape suits a caller-supplied learned impact —
399    /// but nothing in the engine reads it for a bound.
400    pub block_max: u8,
401    /// Largest term frequency in the block. Zero in a version 1 stream.
402    pub max_tf: u32,
403    /// Smallest document length in the block, saturating DOWN to `u16::MAX`.
404    ///
405    /// Down, because a shorter document scores higher: saturating upward
406    /// would put the stored bound below a true score.
407    pub min_len: u16,
408}
409
410impl BlockMeta {
411    fn write(&self, output: &mut Vec<u8>) {
412        output.extend_from_slice(&self.last_docid.to_le_bytes());
413        output.extend_from_slice(&self.docids_offset.to_le_bytes());
414        output.extend_from_slice(&self.tfs_offset.to_le_bytes());
415        output.extend_from_slice(&self.positions_offset.to_le_bytes());
416        output.extend_from_slice(&self.positions_count.to_le_bytes());
417        output.extend_from_slice(&self.count.to_le_bytes());
418        output.push(self.docid_bits);
419        output.push(self.tf_bits);
420        output.push(self.position_bits);
421        output.push(self.block_max);
422        // 4 + 4 + 4 + 4 + 4 + 2 + 4 written above is 26 bytes; the impact
423        // pair spends the remaining six exactly. A version 1 encode leaves
424        // both fields zero, which reproduces the original zero padding byte
425        // for byte, so the v1 golden is unmoved.
426        output.extend_from_slice(&self.max_tf.to_le_bytes());
427        output.extend_from_slice(&self.min_len.to_le_bytes());
428    }
429
430    /// Reads one metadata row from the front of `input`.
431    ///
432    /// # Errors
433    ///
434    /// Returns [`PostingsError::Truncated`] when fewer than
435    /// [`BLOCK_META_LEN`] bytes are available.
436    pub fn read(input: &[u8]) -> Result<Self, PostingsError> {
437        let row = input
438            .get(..BLOCK_META_LEN)
439            .ok_or(PostingsError::Truncated {
440                needed: BLOCK_META_LEN,
441                available: input.len(),
442            })?;
443        Ok(Self {
444            last_docid: read_u32(row, 0)?,
445            docids_offset: read_u32(row, 4)?,
446            tfs_offset: read_u32(row, 8)?,
447            positions_offset: read_u32(row, 12)?,
448            positions_count: read_u32(row, 16)?,
449            count: read_u16(row, 20)?,
450            docid_bits: read_u8(row, 22)?,
451            tf_bits: read_u8(row, 23)?,
452            position_bits: read_u8(row, 24)?,
453            block_max: read_u8(row, 25)?,
454            max_tf: read_u32(row, 26)?,
455            min_len: read_u16(row, 30)?,
456        })
457    }
458}
459
460fn read_u8(input: &[u8], at: usize) -> Result<u8, PostingsError> {
461    input.get(at).copied().ok_or(PostingsError::Truncated {
462        needed: at + 1,
463        available: input.len(),
464    })
465}
466
467fn read_u16(input: &[u8], at: usize) -> Result<u16, PostingsError> {
468    let bytes = input.get(at..at + 2).ok_or(PostingsError::Truncated {
469        needed: at + 2,
470        available: input.len(),
471    })?;
472    let mut buffer = [0_u8; 2];
473    buffer.copy_from_slice(bytes);
474    Ok(u16::from_le_bytes(buffer))
475}
476
477fn read_u32(input: &[u8], at: usize) -> Result<u32, PostingsError> {
478    let bytes = input.get(at..at + 4).ok_or(PostingsError::Truncated {
479        needed: at + 4,
480        available: input.len(),
481    })?;
482    let mut buffer = [0_u8; 4];
483    buffer.copy_from_slice(bytes);
484    Ok(u32::from_le_bytes(buffer))
485}
486
487/// An encoded posting list: header, block metadata rows, then the streams.
488#[derive(Clone, Debug, Default, Eq, PartialEq)]
489pub struct EncodedPostings {
490    bytes: Vec<u8>,
491}
492
493impl EncodedPostings {
494    /// Returns the encoded bytes.
495    #[must_use]
496    pub fn as_bytes(&self) -> &[u8] {
497        &self.bytes
498    }
499
500    /// Wraps already-encoded bytes.
501    #[must_use]
502    pub const fn from_bytes(bytes: Vec<u8>) -> Self {
503        Self { bytes }
504    }
505}
506
507/// Header length: magic, version, block size, posting and block counts.
508pub const HEADER_LEN: usize = 16;
509
510/// Encodes one posting list in the original version 1 layout.
511///
512/// `block_maxima` supplies the per-block reserved `u8` slot; pass an empty
513/// slice to store zeros. The impact-pair bytes stay zero, which is what
514/// keeps the frozen v1 golden byte-identical.
515///
516/// New writers use [`encode_v2`]. This entry point exists so the v1 golden
517/// stays exercised and so a v1 stream can still be produced deliberately.
518///
519/// # Errors
520///
521/// Returns [`PostingsError::ZeroBlockSize`] when the geometry is degenerate.
522pub fn encode(
523    list: &PostingList,
524    postings_per_block: u16,
525    block_maxima: &[u8],
526) -> Result<EncodedPostings, PostingsError> {
527    encode_versioned(
528        list,
529        postings_per_block,
530        block_maxima,
531        &[],
532        POSTINGS_VERSION_V1,
533        &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
534    )
535}
536
537/// Encodes one posting list in the version 2 layout, carrying impact pairs.
538///
539/// `impacts` supplies one [`BlockImpact`] per block, normally from
540/// [`block_impacts`]. A missing entry stores a zero pair, which a reader
541/// rejects rather than treating as an unbounded block.
542///
543/// # Errors
544///
545/// Returns [`PostingsError::ZeroBlockSize`] when the geometry is degenerate.
546pub fn encode_v2(
547    list: &PostingList,
548    postings_per_block: u16,
549    block_maxima: &[u8],
550    impacts: &[BlockImpact],
551) -> Result<EncodedPostings, PostingsError> {
552    encode_versioned(
553        list,
554        postings_per_block,
555        block_maxima,
556        impacts,
557        POSTINGS_VERSION,
558        &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
559    )
560}
561
562pub(crate) fn encode_v2_controlled<E: From<PostingsError>>(
563    list: &PostingList,
564    postings_per_block: u16,
565    block_maxima: &[u8],
566    impacts: &[BlockImpact],
567    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
568) -> Result<EncodedPostings, E> {
569    encode_versioned(
570        list,
571        postings_per_block,
572        block_maxima,
573        impacts,
574        POSTINGS_VERSION,
575        work,
576    )
577}
578
579/// The one encoder. Version selects only what the reserved bytes carry.
580fn encode_versioned<E: From<PostingsError>>(
581    list: &PostingList,
582    postings_per_block: u16,
583    block_maxima: &[u8],
584    impacts: &[BlockImpact],
585    version: u16,
586    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
587) -> Result<EncodedPostings, E> {
588    work.check_now()?;
589    if postings_per_block == 0 {
590        return Err(PostingsError::ZeroBlockSize.into());
591    }
592    let per_block = usize::from(postings_per_block);
593    let postings = list.postings();
594    let block_count = postings.len().div_ceil(per_block);
595
596    let mut docid_stream: Vec<u8> = Vec::new();
597    let mut tf_stream: Vec<u8> = Vec::new();
598    let mut position_stream: Vec<u8> = Vec::new();
599    let mut metadata: Vec<BlockMeta> = Vec::with_capacity(block_count);
600
601    let mut previous_last_docid = 0_u32;
602    for (index, chunk) in postings.chunks(per_block).enumerate() {
603        work.step()?;
604        let mut deltas: Vec<u32> = Vec::with_capacity(chunk.len());
605        let mut frequencies: Vec<u32> = Vec::with_capacity(chunk.len());
606        let mut position_deltas: Vec<u32> = Vec::new();
607
608        let mut base = previous_last_docid;
609        for (offset, posting) in chunk.iter().enumerate() {
610            work.step()?;
611            // The very first posting of the list is stored absolutely; every
612            // other delta is a gap from the previous document id.
613            let delta = if index == 0 && offset == 0 {
614                posting.docid
615            } else {
616                posting.docid.saturating_sub(base)
617            };
618            deltas.push(delta);
619            frequencies.push(posting.tf);
620            let mut previous_position = 0_u32;
621            for (slot, position) in posting.positions.iter().enumerate() {
622                work.step()?;
623                #[cfg(test)]
624                ENCODE_POSITION_PROBES.with(|value| value.set(value.get() + 1));
625                let gap = if slot == 0 {
626                    *position
627                } else {
628                    position.saturating_sub(previous_position)
629                };
630                position_deltas.push(gap);
631                previous_position = *position;
632            }
633            base = posting.docid;
634        }
635        previous_last_docid = base;
636
637        let docid_bits = required_bits_controlled(&deltas, work)?;
638        let tf_bits = required_bits_controlled(&frequencies, work)?;
639        let position_bits = required_bits_controlled(&position_deltas, work)?;
640
641        let meta = BlockMeta {
642            last_docid: base,
643            docids_offset: u32::try_from(docid_stream.len()).unwrap_or(u32::MAX),
644            tfs_offset: u32::try_from(tf_stream.len()).unwrap_or(u32::MAX),
645            positions_offset: u32::try_from(position_stream.len()).unwrap_or(u32::MAX),
646            positions_count: u32::try_from(position_deltas.len()).unwrap_or(u32::MAX),
647            count: u16::try_from(chunk.len()).unwrap_or(u16::MAX),
648            docid_bits,
649            tf_bits,
650            position_bits,
651            block_max: block_maxima.get(index).copied().unwrap_or(0),
652            max_tf: impacts.get(index).map_or(0, |impact| impact.max_tf),
653            min_len: impacts.get(index).map_or(0, |impact| impact.min_len),
654        };
655
656        pack_bits_controlled(&deltas, docid_bits, &mut docid_stream, work)?;
657        pack_bits_controlled(&frequencies, tf_bits, &mut tf_stream, work)?;
658        pack_bits_controlled(&position_deltas, position_bits, &mut position_stream, work)?;
659        metadata.push(meta);
660    }
661
662    let mut bytes = Vec::with_capacity(
663        HEADER_LEN
664            + metadata.len() * BLOCK_META_LEN
665            + docid_stream.len()
666            + tf_stream.len()
667            + position_stream.len(),
668    );
669    bytes.extend_from_slice(&POSTINGS_MAGIC);
670    bytes.extend_from_slice(&version.to_le_bytes());
671    bytes.extend_from_slice(&postings_per_block.to_le_bytes());
672    bytes.extend_from_slice(
673        &u32::try_from(postings.len())
674            .unwrap_or(u32::MAX)
675            .to_le_bytes(),
676    );
677    bytes.extend_from_slice(
678        &u32::try_from(metadata.len())
679            .unwrap_or(u32::MAX)
680            .to_le_bytes(),
681    );
682    for meta in &metadata {
683        work.step()?;
684        meta.write(&mut bytes);
685    }
686    super::control::extend_bytes(&mut bytes, &docid_stream, work)?;
687    super::control::extend_bytes(&mut bytes, &tf_stream, work)?;
688    super::control::extend_bytes(&mut bytes, &position_stream, work)?;
689
690    work.check_now()?;
691    Ok(EncodedPostings { bytes })
692}
693
694/// A validated view over an encoded posting list.
695#[derive(Clone, Debug)]
696pub struct PostingsReader<'bytes> {
697    version: u16,
698    blocks: Vec<BlockMeta>,
699    docids: &'bytes [u8],
700    tfs: &'bytes [u8],
701    positions: &'bytes [u8],
702    posting_count: u32,
703    postings_per_block: u16,
704}
705
706#[cfg(test)]
707std::thread_local! {
708    static ENCODE_POSITION_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
709    static ENCODE_BITS_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
710    static ENCODE_PACK_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
711    static ENCODE_IMPACT_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
712    pub(super) static POSITION_OPEN_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
713    pub(super) static POSITION_MEMBER_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
714    pub(crate) static POSITION_DECODE_PROBES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
715}
716
717impl<'bytes> PostingsReader<'bytes> {
718    /// Locates only one row's position deltas. Other rows' positions are not
719    /// decoded; their frequencies establish the offset within this block.
720    ///
721    /// # Errors
722    /// Returns a typed posting error for malformed frequencies or geometry.
723    pub fn positions(&self, row: u32) -> Result<Option<RowPositions<'bytes>>, PostingsError> {
724        self.positions_controlled(
725            row,
726            &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
727        )
728    }
729
730    pub(crate) fn positions_controlled<E: From<PostingsError>>(
731        &self,
732        row: u32,
733        work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
734    ) -> Result<Option<RowPositions<'bytes>>, E> {
735        work.check_now()?;
736        let block = self.blocks.partition_point(|meta| meta.last_docid < row);
737        let Some(meta) = self.blocks.get(block) else {
738            return Ok(None);
739        };
740        let docids =
741            self.docids
742                .get(meta.docids_offset as usize..)
743                .ok_or(PostingsError::Truncated {
744                    needed: meta.docids_offset as usize,
745                    available: self.docids.len(),
746                })?;
747        let tfs = self
748            .tfs
749            .get(meta.tfs_offset as usize..)
750            .ok_or(PostingsError::Truncated {
751                needed: meta.tfs_offset as usize,
752                available: self.tfs.len(),
753            })?;
754        let mut docid = block
755            .checked_sub(1)
756            .and_then(|previous| self.blocks.get(previous))
757            .map_or(0, |meta| meta.last_docid);
758        let mut first = 0_usize;
759        for slot in 0..usize::from(meta.count) {
760            work.step()?;
761            #[cfg(test)]
762            POSITION_MEMBER_PROBES.with(|value| value.set(value.get() + 1));
763            let delta = packed_value_at(docids, meta.docid_bits, slot)?;
764            docid = docid
765                .checked_add(delta)
766                .ok_or(PostingsError::DocidsNotAscending { docid })?;
767            if docid > row {
768                return Ok(None);
769            }
770            let count = packed_value_at(tfs, meta.tf_bits, slot)? as usize;
771            if count == 0 {
772                return Err(PostingsError::ZeroTermFrequency.into());
773            }
774            let end = first.checked_add(count).ok_or(PostingsError::Truncated {
775                needed: usize::MAX,
776                available: meta.positions_count as usize,
777            })?;
778            if end > meta.positions_count as usize {
779                return Err(PostingsError::Truncated {
780                    needed: end,
781                    available: meta.positions_count as usize,
782                }
783                .into());
784            }
785            if docid == row {
786                let start = meta.positions_offset as usize;
787                let end = start
788                    .checked_add(packed_len(
789                        meta.positions_count as usize,
790                        meta.position_bits,
791                    ))
792                    .ok_or(PostingsError::Truncated {
793                        needed: usize::MAX,
794                        available: self.positions.len(),
795                    })?;
796                let bytes = self
797                    .positions
798                    .get(start..end)
799                    .ok_or(PostingsError::Truncated {
800                        needed: end,
801                        available: self.positions.len(),
802                    })?;
803                return Ok(Some(RowPositions {
804                    bytes,
805                    bits: meta.position_bits,
806                    first,
807                    count,
808                }));
809            }
810            first = end;
811        }
812        Ok(None)
813    }
814
815    /// Validates and opens an encoded posting list.
816    ///
817    /// Every structural claim the bytes make is checked here, so the
818    /// iteration methods below cannot fail on a reader that exists. Corrupt
819    /// bytes produce a typed error and never a panic.
820    ///
821    /// # Errors
822    ///
823    /// Returns [`PostingsError`] for bad magic, an unreadable version, a
824    /// zero block size, an oversized bit width, an inconsistent posting
825    /// count, or truncation anywhere.
826    pub fn open(bytes: &'bytes [u8]) -> Result<Self, PostingsError> {
827        Self::open_controlled(
828            bytes,
829            &mut super::control::WorkCheck::new(|| Ok::<(), PostingsError>(())),
830        )
831    }
832
833    pub(crate) fn open_controlled<E: From<PostingsError>>(
834        bytes: &'bytes [u8],
835        work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
836    ) -> Result<Self, E> {
837        work.check_now()?;
838        let header = bytes.get(..HEADER_LEN).ok_or(PostingsError::Truncated {
839            needed: HEADER_LEN,
840            available: bytes.len(),
841        })?;
842        if header.get(..4) != Some(&POSTINGS_MAGIC[..]) {
843            return Err(PostingsError::BadMagic.into());
844        }
845        let version = read_u16(header, 4)?;
846        if version != POSTINGS_VERSION_V1 && version != POSTINGS_VERSION {
847            return Err(PostingsError::UnsupportedVersion { found: version }.into());
848        }
849        let postings_per_block = read_u16(header, 6)?;
850        if postings_per_block == 0 {
851            return Err(PostingsError::ZeroBlockSize.into());
852        }
853        let posting_count = read_u32(header, 8)?;
854        let block_count = read_u32(header, 12)?;
855
856        let block_count_usize = usize::try_from(block_count).unwrap_or(usize::MAX);
857        let metadata_len =
858            block_count_usize
859                .checked_mul(BLOCK_META_LEN)
860                .ok_or(PostingsError::Truncated {
861                    needed: usize::MAX,
862                    available: bytes.len(),
863                })?;
864        let metadata_end =
865            HEADER_LEN
866                .checked_add(metadata_len)
867                .ok_or(PostingsError::Truncated {
868                    needed: usize::MAX,
869                    available: bytes.len(),
870                })?;
871        let metadata_bytes =
872            bytes
873                .get(HEADER_LEN..metadata_end)
874                .ok_or(PostingsError::Truncated {
875                    needed: metadata_end,
876                    available: bytes.len(),
877                })?;
878
879        let mut blocks = Vec::with_capacity(block_count_usize);
880        let mut described = 0_u32;
881        let mut docid_bytes = 0_usize;
882        let mut tf_bytes = 0_usize;
883        let mut position_bytes = 0_usize;
884        for index in 0..block_count_usize {
885            work.step()?;
886            #[cfg(test)]
887            POSITION_OPEN_PROBES.with(|value| value.set(value.get() + 1));
888            let start = index.saturating_mul(BLOCK_META_LEN);
889            let row = metadata_bytes.get(start..start + BLOCK_META_LEN).ok_or(
890                PostingsError::Truncated {
891                    needed: start + BLOCK_META_LEN,
892                    available: metadata_bytes.len(),
893                },
894            )?;
895            let meta = BlockMeta::read(row)?;
896            for bits in [meta.docid_bits, meta.tf_bits, meta.position_bits] {
897                if bits > 32 {
898                    return Err(PostingsError::BitWidthTooLarge { bits }.into());
899                }
900            }
901            if meta.count == 0 || meta.count > postings_per_block {
902                return Err(PostingsError::InconsistentPostingCount {
903                    declared: posting_count,
904                    described: described.saturating_add(u32::from(meta.count)),
905                }
906                .into());
907            }
908            described = described.saturating_add(u32::from(meta.count));
909            let count = usize::from(meta.count);
910            docid_bytes = docid_bytes.max(
911                usize::try_from(meta.docids_offset).unwrap_or(usize::MAX)
912                    + packed_len(count, meta.docid_bits),
913            );
914            tf_bytes = tf_bytes.max(
915                usize::try_from(meta.tfs_offset).unwrap_or(usize::MAX)
916                    + packed_len(count, meta.tf_bits),
917            );
918            position_bytes = position_bytes.max(
919                usize::try_from(meta.positions_offset).unwrap_or(usize::MAX)
920                    + packed_len(
921                        usize::try_from(meta.positions_count).unwrap_or(usize::MAX),
922                        meta.position_bits,
923                    ),
924            );
925            blocks.push(meta);
926        }
927        if described != posting_count {
928            return Err(PostingsError::InconsistentPostingCount {
929                declared: posting_count,
930                described,
931            }
932            .into());
933        }
934
935        let streams = bytes.get(metadata_end..).ok_or(PostingsError::Truncated {
936            needed: metadata_end,
937            available: bytes.len(),
938        })?;
939        let tf_start = docid_bytes;
940        let position_start = tf_start
941            .checked_add(tf_bytes)
942            .ok_or(PostingsError::Truncated {
943                needed: usize::MAX,
944                available: streams.len(),
945            })?;
946        let position_end =
947            position_start
948                .checked_add(position_bytes)
949                .ok_or(PostingsError::Truncated {
950                    needed: usize::MAX,
951                    available: streams.len(),
952                })?;
953        if streams.len() < position_end {
954            return Err(PostingsError::Truncated {
955                needed: position_end,
956                available: streams.len(),
957            }
958            .into());
959        }
960        let docids = streams.get(..tf_start).ok_or(PostingsError::Truncated {
961            needed: tf_start,
962            available: streams.len(),
963        })?;
964        let tfs = streams
965            .get(tf_start..position_start)
966            .ok_or(PostingsError::Truncated {
967                needed: position_start,
968                available: streams.len(),
969            })?;
970        let positions =
971            streams
972                .get(position_start..position_end)
973                .ok_or(PostingsError::Truncated {
974                    needed: position_end,
975                    available: streams.len(),
976                })?;
977
978        work.check_now()?;
979        Ok(Self {
980            version,
981            blocks,
982            docids,
983            tfs,
984            positions,
985            posting_count,
986            postings_per_block,
987        })
988    }
989
990    /// Returns the block metadata rows.
991    #[must_use]
992    pub fn blocks(&self) -> &[BlockMeta] {
993        &self.blocks
994    }
995
996    /// Returns the format version the stream declared.
997    ///
998    /// A caller that needs impact pairs checks this rather than inferring
999    /// their presence from a zero pair, so a v1 stream is refused loudly
1000    /// instead of silently bounding nothing.
1001    #[must_use]
1002    pub const fn version(&self) -> u16 {
1003        self.version
1004    }
1005
1006    /// Returns the postings-per-block geometry recorded in the header.
1007    #[must_use]
1008    pub const fn postings_per_block(&self) -> u16 {
1009        self.postings_per_block
1010    }
1011
1012    /// Returns the total posting count.
1013    #[must_use]
1014    pub const fn document_frequency(&self) -> u32 {
1015        self.posting_count
1016    }
1017
1018    /// Decodes one block into document ids, frequencies, and positions.
1019    ///
1020    /// # Errors
1021    ///
1022    /// Returns [`PostingsError::Truncated`] when a stream is short. Bounds
1023    /// were validated at [`PostingsReader::open`], so a well-formed reader
1024    /// decodes every block successfully.
1025    pub fn decode_block(&self, index: usize) -> Result<Vec<Posting>, PostingsError> {
1026        let Some(meta) = self.blocks.get(index) else {
1027            return Ok(Vec::new());
1028        };
1029        let count = usize::from(meta.count);
1030        let mut deltas = vec![0_u32; count];
1031        let docid_start = usize::try_from(meta.docids_offset).unwrap_or(usize::MAX);
1032        let docid_slice = self
1033            .docids
1034            .get(docid_start..)
1035            .ok_or(PostingsError::Truncated {
1036                needed: docid_start,
1037                available: self.docids.len(),
1038            })?;
1039        unpack_bits(docid_slice, meta.docid_bits, count, &mut deltas)?;
1040
1041        let mut frequencies = vec![0_u32; count];
1042        let tf_start = usize::try_from(meta.tfs_offset).unwrap_or(usize::MAX);
1043        let tf_slice = self.tfs.get(tf_start..).ok_or(PostingsError::Truncated {
1044            needed: tf_start,
1045            available: self.tfs.len(),
1046        })?;
1047        unpack_bits(tf_slice, meta.tf_bits, count, &mut frequencies)?;
1048
1049        let position_count = usize::try_from(meta.positions_count).unwrap_or(usize::MAX);
1050        let mut position_deltas = vec![0_u32; position_count];
1051        let position_start = usize::try_from(meta.positions_offset).unwrap_or(usize::MAX);
1052        let position_slice =
1053            self.positions
1054                .get(position_start..)
1055                .ok_or(PostingsError::Truncated {
1056                    needed: position_start,
1057                    available: self.positions.len(),
1058                })?;
1059        unpack_bits(
1060            position_slice,
1061            meta.position_bits,
1062            position_count,
1063            &mut position_deltas,
1064        )?;
1065
1066        // Reconstruct absolute ids by prefix sum over the block.
1067        let mut base = if index == 0 {
1068            0
1069        } else {
1070            self.blocks
1071                .get(index.saturating_sub(1))
1072                .map_or(0, |previous| previous.last_docid)
1073        };
1074        let mut postings = Vec::with_capacity(count);
1075        let mut position_cursor = 0_usize;
1076        for slot in 0..count {
1077            let delta = deltas.get(slot).copied().unwrap_or(0);
1078            let docid = if index == 0 && slot == 0 {
1079                delta
1080            } else {
1081                base.saturating_add(delta)
1082            };
1083            base = docid;
1084            let tf = frequencies.get(slot).copied().unwrap_or(0);
1085            if tf == 0 {
1086                return Err(PostingsError::ZeroTermFrequency);
1087            }
1088            let tf_usize = usize::try_from(tf).unwrap_or(usize::MAX);
1089            let mut positions = Vec::with_capacity(tf_usize.min(position_count));
1090            let mut running = 0_u32;
1091            for step in 0..tf_usize {
1092                let Some(gap) = position_deltas.get(position_cursor + step).copied() else {
1093                    return Err(PostingsError::Truncated {
1094                        needed: position_cursor + step + 1,
1095                        available: position_deltas.len(),
1096                    });
1097                };
1098                running = if step == 0 {
1099                    gap
1100                } else {
1101                    running.saturating_add(gap)
1102                };
1103                positions.push(running);
1104            }
1105            position_cursor = position_cursor.saturating_add(tf_usize);
1106            postings.push(Posting {
1107                docid,
1108                tf,
1109                positions,
1110            });
1111        }
1112        Ok(postings)
1113    }
1114
1115    /// Decodes the whole list.
1116    ///
1117    /// # Errors
1118    ///
1119    /// Propagates any [`PostingsError`] from block decoding.
1120    pub fn decode_all(&self) -> Result<PostingList, PostingsError> {
1121        let mut list = PostingList::new();
1122        for index in 0..self.blocks.len() {
1123            for posting in self.decode_block(index)? {
1124                list.push(posting)?;
1125            }
1126        }
1127        Ok(list)
1128    }
1129}
1130
1131/// Borrowed packed positions for one requested posting. Locating this view
1132/// does not decode position values or allocate a position array.
1133#[derive(Clone, Copy, Debug)]
1134pub struct RowPositions<'bytes> {
1135    bytes: &'bytes [u8],
1136    bits: u8,
1137    first: usize,
1138    count: usize,
1139}
1140
1141impl RowPositions<'_> {
1142    /// Number of positions of the requested term in this row.
1143    #[must_use]
1144    pub const fn len(&self) -> usize {
1145        self.count
1146    }
1147
1148    /// Whether this view contains no positions.
1149    #[must_use]
1150    pub const fn is_empty(&self) -> bool {
1151        self.count == 0
1152    }
1153
1154    /// Decodes only the requested row, validating strict position ordering.
1155    ///
1156    /// # Errors
1157    /// Returns a typed posting error for truncation, invalid widths or order.
1158    pub fn decode(&self) -> Result<Vec<u32>, PostingsError> {
1159        self.decode_controlled(&mut super::control::WorkCheck::new(|| {
1160            Ok::<(), PostingsError>(())
1161        }))
1162    }
1163
1164    pub(crate) fn decode_controlled<E: From<PostingsError>>(
1165        &self,
1166        work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
1167    ) -> Result<Vec<u32>, E> {
1168        work.check_now()?;
1169        let mut positions = Vec::with_capacity(self.count);
1170        let mut position = 0_u32;
1171        let result = (|| {
1172            for offset in 0..self.count {
1173                work.step()?;
1174                #[cfg(test)]
1175                POSITION_DECODE_PROBES.with(|value| value.set(value.get() + 1));
1176                let delta = packed_value_at(self.bytes, self.bits, self.first + offset)?;
1177                if offset > 0 && delta == 0 {
1178                    return Err(PostingsError::PositionsNotAscending { position }.into());
1179                }
1180                position = position
1181                    .checked_add(delta)
1182                    .ok_or(PostingsError::PositionsNotAscending { position })?;
1183                positions.push(position);
1184            }
1185            work.check_now()
1186        })();
1187        // Record completed decode work even when the next cooperative check
1188        // aborts. Partial scratch is never returned to the caller.
1189        #[cfg(any(test, feature = "test-support"))]
1190        if !positions.is_empty() {
1191            let first_bit = self.first * usize::from(self.bits);
1192            let last_bit = (self.first + positions.len()) * usize::from(self.bits);
1193            super::preparation_observer::phrase_positions(
1194                positions.len(),
1195                last_bit.div_ceil(8) - first_bit / 8,
1196            );
1197        }
1198        result?;
1199        Ok(positions)
1200    }
1201}
1202
1203/// A bit-addressed read can start mid-byte, including 32-bit values spanning
1204/// five bytes. It never expands preceding position values to find this one.
1205fn packed_value_at(bytes: &[u8], bits: u8, slot: usize) -> Result<u32, PostingsError> {
1206    if bits > 32 {
1207        return Err(PostingsError::BitWidthTooLarge { bits });
1208    }
1209    if bits == 0 {
1210        return Ok(0);
1211    }
1212    let first = slot
1213        .checked_mul(usize::from(bits))
1214        .ok_or(PostingsError::Truncated {
1215            needed: usize::MAX,
1216            available: bytes.len(),
1217        })?;
1218    let end = first
1219        .checked_add(usize::from(bits))
1220        .ok_or(PostingsError::Truncated {
1221            needed: usize::MAX,
1222            available: bytes.len(),
1223        })?
1224        .div_ceil(8);
1225    let selected = bytes.get(first / 8..end).ok_or(PostingsError::Truncated {
1226        needed: end,
1227        available: bytes.len(),
1228    })?;
1229    let word = selected
1230        .iter()
1231        .enumerate()
1232        .fold(0_u64, |word, (offset, byte)| {
1233            word | (u64::from(*byte) << (offset * 8))
1234        });
1235    let mask = (1_u64 << bits) - 1;
1236    Ok(((word >> (first % 8)) & mask) as u32)
1237}
1238
1239#[cfg(test)]
1240#[allow(clippy::expect_used)]
1241mod astra_18_tests {
1242    use super::*;
1243    use crate::fts::control::WorkCheck;
1244    use std::cell::Cell;
1245
1246    #[derive(Debug, PartialEq)]
1247    enum TestError {
1248        Cancelled,
1249        Postings(PostingsError),
1250    }
1251
1252    impl From<PostingsError> for TestError {
1253        fn from(error: PostingsError) -> Self {
1254            Self::Postings(error)
1255        }
1256    }
1257
1258    fn one_position_per_row() -> PostingList {
1259        let mut list = PostingList::new();
1260        for row in 0..4_096 {
1261            list.push(Posting {
1262                docid: row,
1263                tf: 1,
1264                positions: vec![row * 2],
1265            })
1266            .expect("literal posting");
1267        }
1268        list
1269    }
1270
1271    #[test]
1272    fn astra_18_position_metadata_cancels_inside_block_walk() {
1273        let list = one_position_per_row();
1274        let encoded = encode(&list, 1, &[]).expect("one row per block");
1275        POSITION_OPEN_PROBES.with(|value| value.set(0));
1276        let mut work = WorkCheck::new(|| {
1277            if POSITION_OPEN_PROBES.with(Cell::get) >= 64 {
1278                Err(TestError::Cancelled)
1279            } else {
1280                Ok(())
1281            }
1282        });
1283        let result = PostingsReader::open_controlled(encoded.as_bytes(), &mut work);
1284        let probes = POSITION_OPEN_PROBES.with(Cell::get);
1285        println!("metadata_probes={probes}");
1286        assert!(matches!(result, Err(TestError::Cancelled)));
1287        assert!(
1288            (64..=128).contains(&probes),
1289            "metadata walk must stop inside the list"
1290        );
1291        let mut clean = WorkCheck::new(|| Ok::<(), TestError>(()));
1292        let reader = PostingsReader::open_controlled(encoded.as_bytes(), &mut clean)
1293            .expect("fresh control opens complete metadata");
1294        assert_eq!(reader.blocks().len(), 4_096);
1295        assert_eq!(reader.document_frequency(), 4_096);
1296        assert_eq!(reader.decode_all().expect("complete literal output"), list);
1297        assert!(matches!(
1298            PostingsReader::open_controlled(&[], &mut clean),
1299            Err(TestError::Postings(PostingsError::Truncated { .. }))
1300        ));
1301    }
1302
1303    #[test]
1304    fn astra_18_position_membership_cancels_inside_one_block() {
1305        let list = one_position_per_row();
1306        let encoded = encode(&list, 4_096, &[]).expect("one large legal block");
1307        let reader = PostingsReader::open(encoded.as_bytes()).expect("reader");
1308        POSITION_MEMBER_PROBES.with(|value| value.set(0));
1309        let mut work = WorkCheck::new(|| {
1310            if POSITION_MEMBER_PROBES.with(Cell::get) >= 64 {
1311                Err(TestError::Cancelled)
1312            } else {
1313                Ok(())
1314            }
1315        });
1316        let result = reader.positions_controlled(4_095, &mut work);
1317        let probes = POSITION_MEMBER_PROBES.with(Cell::get);
1318        println!("membership_probes={probes}");
1319        assert!(matches!(result, Err(TestError::Cancelled)));
1320        assert!(
1321            (64..=128).contains(&probes),
1322            "membership must stop inside the block"
1323        );
1324        let mut clean = WorkCheck::new(|| Ok::<(), TestError>(()));
1325        assert_eq!(
1326            reader
1327                .positions_controlled(4_095, &mut clean)
1328                .expect("fresh control")
1329                .expect("last row")
1330                .decode()
1331                .expect("position"),
1332            vec![8_190]
1333        );
1334        assert!(
1335            reader
1336                .positions_controlled(4_096, &mut clean)
1337                .expect("absent row")
1338                .is_none()
1339        );
1340    }
1341
1342    #[test]
1343    fn astra_18_position_decode_cancels_inside_one_row() {
1344        let literal = (0..4_096).collect::<Vec<_>>();
1345        let mut list = PostingList::new();
1346        list.push(Posting {
1347            docid: 0,
1348            tf: 4_096,
1349            positions: literal.clone(),
1350        })
1351        .expect("one long row");
1352        let encoded = encode(&list, 64, &[]).expect("encoded positions");
1353        let reader = PostingsReader::open(encoded.as_bytes()).expect("reader");
1354        let view = reader.positions(0).expect("membership").expect("present");
1355        POSITION_DECODE_PROBES.with(|value| value.set(0));
1356        crate::fts::preparation_observer::begin();
1357        let mut work = WorkCheck::new(|| {
1358            if POSITION_DECODE_PROBES.with(Cell::get) >= 64 {
1359                Err(TestError::Cancelled)
1360            } else {
1361                Ok(())
1362            }
1363        });
1364        let result = view.decode_controlled(&mut work);
1365        let probes = POSITION_DECODE_PROBES.with(Cell::get);
1366        let observed = crate::fts::preparation_observer::phrase_position_work();
1367        crate::fts::preparation_observer::take();
1368        println!("decode_probes={probes}, position_work={observed:?}");
1369        assert_eq!(result, Err(TestError::Cancelled));
1370        assert!(
1371            (64..=128).contains(&probes),
1372            "decode must stop inside the row"
1373        );
1374        assert_eq!(
1375            observed,
1376            (probes, probes.div_ceil(8)),
1377            "retain actual partial decode work"
1378        );
1379        let mut clean = WorkCheck::new(|| Ok::<(), TestError>(()));
1380        assert_eq!(
1381            view.decode_controlled(&mut clean).expect("fresh control"),
1382            literal
1383        );
1384    }
1385}
1386
1387#[cfg(test)]
1388#[allow(clippy::expect_used, clippy::indexing_slicing)]
1389mod astra_11_tests {
1390    use super::*;
1391
1392    #[test]
1393    fn astra_11_position_reader_only_decodes_requested_row() {
1394        for per_block in [1, 2, 64] {
1395            let mut list = PostingList::new();
1396            let expected = (0..130)
1397                .map(|row| {
1398                    if row % 3 == 0 {
1399                        vec![0, 1, 31, u32::MAX]
1400                    } else {
1401                        vec![row * 3]
1402                    }
1403                })
1404                .collect::<Vec<_>>();
1405            for (row, positions) in expected.iter().enumerate() {
1406                list.push(Posting {
1407                    docid: row as u32 * 2,
1408                    tf: positions.len() as u32,
1409                    positions: positions.clone(),
1410                })
1411                .expect("posting");
1412            }
1413            let encoded = encode(&list, per_block, &[]).expect("encoded positions");
1414            let reader = PostingsReader::open(encoded.as_bytes()).expect("persisted reader");
1415            for row in [0, 1, 63, 64, 65, 129] {
1416                crate::fts::preparation_observer::begin();
1417                let view = reader
1418                    .positions(row as u32 * 2)
1419                    .expect("locate")
1420                    .expect("present");
1421                assert_eq!(
1422                    crate::fts::preparation_observer::phrase_position_work(),
1423                    (0, 0)
1424                );
1425                assert_eq!(view.decode().expect("selective decode"), expected[row]);
1426                assert_eq!(
1427                    crate::fts::preparation_observer::phrase_position_work().0,
1428                    expected[row].len()
1429                );
1430                crate::fts::preparation_observer::take();
1431                assert!(
1432                    reader
1433                        .positions(row as u32 * 2 + 1)
1434                        .expect("absent")
1435                        .is_none()
1436                );
1437            }
1438        }
1439    }
1440
1441    #[test]
1442    fn astra_11_corrupt_position_payload_is_typed() {
1443        let mut list = PostingList::new();
1444        for (docid, positions) in [(0, vec![1, 3]), (1, vec![2, 5])] {
1445            list.push(Posting {
1446                docid,
1447                tf: 2,
1448                positions,
1449            })
1450            .expect("posting");
1451        }
1452        let encoded = encode(&list, 64, &[]).expect("encoded");
1453        let clean = PostingsReader::open(encoded.as_bytes()).expect("open");
1454        assert_eq!(
1455            clean
1456                .positions(0)
1457                .expect("locate")
1458                .expect("present")
1459                .decode()
1460                .expect("clean"),
1461            vec![1, 3]
1462        );
1463        let positions_start = HEADER_LEN + BLOCK_META_LEN + clean.docids.len() + clean.tfs.len();
1464        assert_eq!(clean.blocks()[0].position_bits, 2);
1465        let mut corrupt = encoded.as_bytes().to_vec();
1466        // Delta two becomes zero in the actual packed payload, while all
1467        // structural lengths and row/frequency data remain intact.
1468        corrupt[positions_start] &= !(3 << 2);
1469        let reader = PostingsReader::open(&corrupt).expect("geometry unchanged");
1470        assert_eq!(
1471            reader
1472                .positions(0)
1473                .expect("locate corrupt row")
1474                .expect("present")
1475                .decode(),
1476            Err(PostingsError::PositionsNotAscending { position: 1 })
1477        );
1478        assert_eq!(
1479            reader
1480                .positions(1)
1481                .expect("next row")
1482                .expect("present")
1483                .decode()
1484                .expect("unaffected row"),
1485            vec![2, 5]
1486        );
1487    }
1488}
1489
1490/// One block's impact pair: the two extremes that bound every posting in it.
1491///
1492/// # Why a pair and not a stored score
1493///
1494/// A stored score is a number computed under one set of statistics. BM25's
1495/// `score / ceiling` is `tf / (tf + k1 * (1 - b + b * len / avgdl))`, so
1496/// `idf` cancels but `avgdl` and `(k1, b)` do not: a bound sealed when
1497/// documents were short falls below a true score once `avgdl` rises, and a
1498/// bound sealed under `beir()` falls below one read under `anserini()`.
1499/// Measured shortfalls were 15.21% and 19.56%.
1500///
1501/// A pair carries no statistics at all. `term_score` is monotone increasing
1502/// in `tf` and monotone decreasing in `len`, so evaluating it at the block's
1503/// largest `tf` and shortest document dominates every posting in the block
1504/// under ANY statistics and ANY `(k1, b)`. The two extremes need not come
1505/// from the same document; pairing them is looser than the true maximum and
1506/// is therefore still an upper bound.
1507#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
1508pub struct BlockImpact {
1509    /// Largest term frequency in the block.
1510    pub max_tf: u32,
1511    /// Smallest document length in the block, saturating DOWN.
1512    pub min_len: u16,
1513}
1514
1515impl BlockImpact {
1516    /// Reads the pair out of a persisted metadata row.
1517    #[must_use]
1518    pub const fn from_meta(meta: &BlockMeta) -> Self {
1519        Self {
1520            max_tf: meta.max_tf,
1521            min_len: meta.min_len,
1522        }
1523    }
1524
1525    /// Returns true when the pair bounds nothing.
1526    ///
1527    /// A real block has at least one posting, whose term frequency is at
1528    /// least one, so a zero `max_tf` can only mean the pair was never
1529    /// written — a version 1 stream, or a v2 stream missing an entry.
1530    #[must_use]
1531    pub const fn is_absent(&self) -> bool {
1532        self.max_tf == 0
1533    }
1534
1535    /// Evaluates the bound under the caller's live scoring constants.
1536    ///
1537    /// This is the whole remedy: the bound is computed HERE, at query time,
1538    /// from statistics that are current and parameters the caller chose.
1539    #[must_use]
1540    pub fn bound(&self, scorer: &super::bm25::TermScorer) -> f64 {
1541        scorer.score(Tf(self.max_tf), DocLen(u32::from(self.min_len)))
1542    }
1543}
1544
1545/// Computes the per-block impact pairs for one term's postings.
1546///
1547/// `lengths` is the dense per-row length array; a row beyond it contributes
1548/// the shortest possible length, which keeps the bound above rather than
1549/// below the truth.
1550#[must_use]
1551pub fn block_impacts(
1552    list: &PostingList,
1553    postings_per_block: u16,
1554    lengths: &[u32],
1555) -> Vec<BlockImpact> {
1556    let mut work = super::control::WorkCheck::new(|| Ok::<(), std::convert::Infallible>(()));
1557    match block_impacts_controlled(list, postings_per_block, lengths, &mut work) {
1558        Ok(impacts) => impacts,
1559        Err(never) => match never {},
1560    }
1561}
1562
1563pub(crate) fn block_impacts_controlled<E>(
1564    list: &PostingList,
1565    postings_per_block: u16,
1566    lengths: &[u32],
1567    work: &mut super::control::WorkCheck<impl FnMut() -> Result<(), E>>,
1568) -> Result<Vec<BlockImpact>, E> {
1569    work.check_now()?;
1570    if postings_per_block == 0 {
1571        return Ok(Vec::new());
1572    }
1573    let chunks = list.postings().chunks(usize::from(postings_per_block));
1574    let mut impacts = Vec::with_capacity(chunks.len());
1575    for chunk in chunks {
1576        let mut max_tf = 0_u32;
1577        let mut min_len = u32::MAX;
1578        for posting in chunk {
1579            work.step()?;
1580            #[cfg(test)]
1581            ENCODE_IMPACT_PROBES.with(|value| value.set(value.get() + 1));
1582            max_tf = max_tf.max(posting.tf);
1583            let length = usize::try_from(posting.docid)
1584                .ok()
1585                .and_then(|slot| lengths.get(slot).copied())
1586                .unwrap_or(1);
1587            min_len = min_len.min(length);
1588        }
1589        impacts.push(BlockImpact {
1590            max_tf,
1591            // Saturate DOWN: a shorter document scores higher, keeping
1592            // this bound at or above the true score.
1593            min_len: u16::try_from(min_len).unwrap_or(u16::MAX),
1594        });
1595    }
1596    work.check_now()?;
1597    Ok(impacts)
1598}
1599
1600#[cfg(test)]
1601#[allow(
1602    clippy::expect_used,
1603    clippy::indexing_slicing,
1604    clippy::panic,
1605    clippy::unwrap_used
1606)]
1607mod tests {
1608    use super::*;
1609
1610    #[test]
1611    fn astra_18_postings_encoding_cancels_inside_positions() {
1612        let positions = (0..4_096).collect::<Vec<_>>();
1613        let list = list_from(&[(0, &positions)]);
1614        let impacts = block_impacts(&list, 64, &[4_096]);
1615        let reference = encode_v2(&list, 64, &[], &impacts).expect("reference");
1616        let mut observed = Vec::new();
1617        for (phase, probes) in [
1618            ("position gaps", &ENCODE_POSITION_PROBES),
1619            ("bit widths", &ENCODE_BITS_PROBES),
1620            ("bit packing", &ENCODE_PACK_PROBES),
1621        ] {
1622            probes.with(|value| value.set(0));
1623            let mut work = super::super::control::WorkCheck::new(|| {
1624                if probes.with(std::cell::Cell::get) >= 64 {
1625                    Err(PostingsError::ZeroBlockSize)
1626                } else {
1627                    Ok(())
1628                }
1629            });
1630            let result = encode_v2_controlled(&list, 64, &[], &impacts, &mut work);
1631            let count = probes.with(std::cell::Cell::get);
1632            println!("{phase}: canceled={}, probes={count}", result.is_err());
1633            observed.push((
1634                phase,
1635                matches!(result, Err(PostingsError::ZeroBlockSize)),
1636                count,
1637            ));
1638        }
1639        let mut work = super::super::control::WorkCheck::new(|| Ok::<(), PostingsError>(()));
1640        let clean = encode_v2_controlled(&list, 64, &[], &impacts, &mut work).expect("clean");
1641        assert_eq!(clean.as_bytes(), reference.as_bytes());
1642        assert_eq!(
1643            PostingsReader::open(clean.as_bytes())
1644                .expect("reader")
1645                .decode_all()
1646                .expect("decoded"),
1647            list
1648        );
1649        for (phase, canceled, count) in observed {
1650            assert!(canceled, "{phase}");
1651            assert!((64..=128).contains(&count), "{phase}: {count}");
1652        }
1653    }
1654
1655    #[test]
1656    fn astra_18_block_impacts_cancel_inside_rows() {
1657        let mut list = PostingList::new();
1658        for docid in 0..4_096 {
1659            list.push(Posting {
1660                docid,
1661                tf: 1,
1662                positions: vec![0],
1663            })
1664            .expect("posting");
1665        }
1666        let lengths = vec![7; 4_096];
1667        ENCODE_IMPACT_PROBES.with(|value| value.set(0));
1668        let mut work = super::super::control::WorkCheck::new(|| {
1669            if ENCODE_IMPACT_PROBES.with(std::cell::Cell::get) >= 64 {
1670                Err("cancelled")
1671            } else {
1672                Ok(())
1673            }
1674        });
1675        let result = block_impacts_controlled(&list, 64, &lengths, &mut work);
1676        let count = ENCODE_IMPACT_PROBES.with(std::cell::Cell::get);
1677        println!("impact rows={count}");
1678        assert_eq!(result, Err("cancelled"));
1679        assert!((64..=128).contains(&count));
1680        let mut work = super::super::control::WorkCheck::new(|| Ok::<(), ()>(()));
1681        assert_eq!(
1682            block_impacts_controlled(&list, 64, &lengths, &mut work),
1683            Ok(vec![
1684                BlockImpact {
1685                    max_tf: 1,
1686                    min_len: 7
1687                };
1688                64
1689            ])
1690        );
1691    }
1692
1693    fn list_from(entries: &[(u32, &[u32])]) -> PostingList {
1694        let mut list = PostingList::new();
1695        for (docid, positions) in entries {
1696            list.push(Posting {
1697                docid: *docid,
1698                tf: u32::try_from(positions.len()).expect("small"),
1699                positions: positions.to_vec(),
1700            })
1701            .expect("ascending fixture");
1702        }
1703        list
1704    }
1705
1706    #[test]
1707    fn a_single_block_round_trips() {
1708        let list = list_from(&[(0, &[0, 5]), (3, &[2]), (9, &[1, 4, 7])]);
1709        let encoded = encode(&list, DEFAULT_POSTINGS_PER_BLOCK, &[]).expect("encodes");
1710        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
1711        assert_eq!(reader.document_frequency(), 3);
1712        assert_eq!(reader.postings_per_block(), DEFAULT_POSTINGS_PER_BLOCK);
1713        assert_eq!(reader.decode_all().expect("decodes"), list);
1714    }
1715
1716    #[test]
1717    fn many_blocks_round_trip_across_the_geometry_boundary() {
1718        for count in [1_usize, 63, 64, 65, 128, 200] {
1719            let mut list = PostingList::new();
1720            for index in 0..count {
1721                let docid = u32::try_from(index * 3).expect("small");
1722                list.push(Posting {
1723                    docid,
1724                    tf: 2,
1725                    positions: vec![index as u32, index as u32 + 7],
1726                })
1727                .expect("ascending");
1728            }
1729            let encoded = encode(&list, DEFAULT_POSTINGS_PER_BLOCK, &[]).expect("encodes");
1730            let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
1731            assert_eq!(
1732                reader.blocks().len(),
1733                count.div_ceil(usize::from(DEFAULT_POSTINGS_PER_BLOCK)),
1734                "block count for {count} postings"
1735            );
1736            assert_eq!(reader.decode_all().expect("decodes"), list, "count {count}");
1737        }
1738    }
1739
1740    #[test]
1741    fn the_block_metadata_row_is_exactly_one_quarter_of_a_cache_line() {
1742        let mut bytes = Vec::new();
1743        BlockMeta::default().write(&mut bytes);
1744        assert_eq!(bytes.len(), BLOCK_META_LEN);
1745        assert_eq!(BLOCK_META_LEN * 4, 128);
1746    }
1747
1748    #[test]
1749    fn an_empty_list_encodes_and_decodes() {
1750        let list = PostingList::new();
1751        let encoded = encode(&list, DEFAULT_POSTINGS_PER_BLOCK, &[]).expect("encodes");
1752        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
1753        assert_eq!(reader.document_frequency(), 0);
1754        assert!(reader.decode_all().expect("decodes").is_empty());
1755    }
1756
1757    #[test]
1758    fn bit_widths_at_every_boundary_round_trip() {
1759        for bits in 0..=32_u8 {
1760            let value = if bits == 0 {
1761                0
1762            } else if bits == 32 {
1763                u32::MAX
1764            } else {
1765                (1_u32 << bits) - 1
1766            };
1767            let values = vec![value; 64];
1768            let mut packed = Vec::new();
1769            pack_bits(&values, bits, &mut packed);
1770            let mut output = vec![0_u32; 64];
1771            let consumed = unpack_bits(&packed, bits, 64, &mut output).expect("unpacks");
1772            assert_eq!(consumed, packed_len(64, bits), "width {bits}");
1773            assert_eq!(output, values, "width {bits}");
1774        }
1775    }
1776
1777    #[test]
1778    fn an_oversized_bit_width_is_a_typed_error() {
1779        let mut output = [0_u32; 4];
1780        assert_eq!(
1781            unpack_bits(&[0; 32], 33, 4, &mut output),
1782            Err(PostingsError::BitWidthTooLarge { bits: 33 })
1783        );
1784    }
1785
1786    #[test]
1787    fn a_truncated_packed_run_is_a_typed_error() {
1788        let mut output = [0_u32; 64];
1789        let error = unpack_bits(&[0; 2], 8, 64, &mut output).expect_err("must refuse");
1790        assert!(matches!(error, PostingsError::Truncated { .. }));
1791    }
1792
1793    #[test]
1794    fn out_of_order_docids_are_refused_rather_than_sorted() {
1795        let mut list = PostingList::new();
1796        list.push(Posting {
1797            docid: 5,
1798            tf: 1,
1799            positions: vec![0],
1800        })
1801        .expect("first");
1802        assert_eq!(
1803            list.push(Posting {
1804                docid: 5,
1805                tf: 1,
1806                positions: vec![0],
1807            }),
1808            Err(PostingsError::DocidsNotAscending { docid: 5 })
1809        );
1810    }
1811
1812    #[test]
1813    fn out_of_order_positions_are_refused() {
1814        let mut list = PostingList::new();
1815        assert_eq!(
1816            list.push(Posting {
1817                docid: 0,
1818                tf: 2,
1819                positions: vec![4, 4],
1820            }),
1821            Err(PostingsError::PositionsNotAscending { position: 4 })
1822        );
1823    }
1824
1825    #[test]
1826    fn a_zero_frequency_posting_is_refused() {
1827        let mut list = PostingList::new();
1828        assert_eq!(
1829            list.push(Posting {
1830                docid: 0,
1831                tf: 0,
1832                positions: vec![],
1833            }),
1834            Err(PostingsError::ZeroTermFrequency)
1835        );
1836    }
1837
1838    #[test]
1839    fn the_impact_pair_survives_the_persisted_round_trip() {
1840        // The pair is what makes a bound sound under drift, so it has to
1841        // come back out of the bytes exactly as it went in.
1842        let list = list_from(&[(0, &[0, 1, 2]), (5, &[0]), (9, &[0, 3])]);
1843        let lengths = vec![40_u32, 0, 0, 0, 0, 7, 0, 0, 0, 900];
1844        let impacts = block_impacts(&list, 2, &lengths);
1845        assert_eq!(impacts.len(), 2);
1846        assert_eq!(
1847            impacts[0],
1848            BlockImpact {
1849                max_tf: 3,
1850                min_len: 7
1851            }
1852        );
1853        assert_eq!(
1854            impacts[1],
1855            BlockImpact {
1856                max_tf: 2,
1857                min_len: 900
1858            }
1859        );
1860
1861        let encoded = encode_v2(&list, 2, &[], &impacts).expect("encodes");
1862        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
1863        assert_eq!(reader.version(), POSTINGS_VERSION);
1864        let read: Vec<BlockImpact> = reader.blocks().iter().map(BlockImpact::from_meta).collect();
1865        assert_eq!(read, impacts);
1866        assert_eq!(reader.decode_all().expect("decodes"), list);
1867    }
1868
1869    #[test]
1870    fn a_version_one_stream_reads_back_with_no_impact_pair() {
1871        // v1 is still readable, and its blocks report an ABSENT pair rather
1872        // than a zero one that a caller might mistake for a real bound.
1873        let list = list_from(&[(0, &[0]), (4, &[1])]);
1874        let encoded = encode(&list, 64, &[]).expect("encodes");
1875        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
1876        assert_eq!(reader.version(), POSTINGS_VERSION_V1);
1877        for meta in reader.blocks() {
1878            assert!(BlockImpact::from_meta(meta).is_absent());
1879        }
1880    }
1881
1882    #[test]
1883    fn an_empty_impact_slice_stores_an_absent_pair() {
1884        let list = list_from(&[(0, &[0])]);
1885        let encoded = encode_v2(&list, 64, &[], &[]).expect("encodes");
1886        let reader = PostingsReader::open(encoded.as_bytes()).expect("opens");
1887        assert_eq!(reader.version(), POSTINGS_VERSION);
1888        assert!(BlockImpact::from_meta(&reader.blocks()[0]).is_absent());
1889        assert!(block_impacts(&list, 0, &[]).is_empty());
1890    }
1891
1892    #[test]
1893    fn bad_magic_and_version_are_typed_errors() {
1894        let list = list_from(&[(0, &[0])]);
1895        let encoded = encode(&list, 64, &[]).expect("encodes");
1896        let mut bytes = encoded.as_bytes().to_vec();
1897        bytes[0] = b'X';
1898        assert_eq!(
1899            PostingsReader::open(&bytes).err(),
1900            Some(PostingsError::BadMagic)
1901        );
1902
1903        let mut bytes = encoded.as_bytes().to_vec();
1904        bytes[4] = 9;
1905        assert_eq!(
1906            PostingsReader::open(&bytes).err(),
1907            Some(PostingsError::UnsupportedVersion { found: 9 })
1908        );
1909    }
1910
1911    #[test]
1912    fn a_zero_block_size_header_is_refused() {
1913        let list = list_from(&[(0, &[0])]);
1914        let encoded = encode(&list, 64, &[]).expect("encodes");
1915        let mut bytes = encoded.as_bytes().to_vec();
1916        bytes[6] = 0;
1917        bytes[7] = 0;
1918        assert_eq!(
1919            PostingsReader::open(&bytes).err(),
1920            Some(PostingsError::ZeroBlockSize)
1921        );
1922        assert_eq!(
1923            encode(&list, 0, &[]).err(),
1924            Some(PostingsError::ZeroBlockSize)
1925        );
1926    }
1927
1928    #[test]
1929    fn every_truncation_of_a_valid_stream_is_a_typed_error_not_a_panic() {
1930        let list = list_from(&[(0, &[0, 3]), (7, &[1]), (100, &[2, 9, 40])]);
1931        let encoded = encode(&list, 2, &[]).expect("encodes");
1932        let bytes = encoded.as_bytes();
1933        for cut in 0..bytes.len() {
1934            let truncated = &bytes[..cut];
1935            // A prefix that still parses must decode without panic.
1936            if let Ok(reader) = PostingsReader::open(truncated) {
1937                for index in 0..reader.blocks().len() {
1938                    let _ = reader.decode_block(index);
1939                }
1940            }
1941        }
1942    }
1943}