mod groups;
mod impacts;
mod reader;
mod validation;
use groups::GroupWords;
use impacts::{ImpactBuilder, ImpactTable};
pub(crate) use reader::{DeferredPosting, PostingListReader};
#[cfg(feature = "native")]
mod compact;
#[cfg(feature = "native")]
pub(crate) use compact::{PostingBlockSource, PostingStreamWriter};
use byteorder::{LittleEndian, WriteBytesExt};
use std::io::{self, Write};
use super::bitpacking4x;
use super::horizontal_bp128::{pack_block_n as pack_bits, unpack_block_n as unpack_bits};
use super::opt_p4d::{find_optimal_bit_width, pack_with_exceptions};
use crate::DocId;
use crate::directories::OwnedBytes;
use crate::structures::simd;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PostingCodec {
#[default]
Rounded = 0,
Packed = 1,
Pfor = 2,
Simd4x = 3,
}
impl PostingCodec {
const HEADER_SHIFT: u32 = 6;
const WIDTH_MASK: u8 = 0x3F;
fn from_header_byte(doc_bits: u8) -> io::Result<(Self, u8)> {
let width = doc_bits & Self::WIDTH_MASK;
let codec = match doc_bits >> Self::HEADER_SHIFT {
0 => PostingCodec::Rounded,
1 => PostingCodec::Packed,
2 => PostingCodec::Pfor,
_ => PostingCodec::Simd4x,
};
if width > 32 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!("posting block doc-id width {width} exceeds 32 bits"),
));
}
Ok((codec, width))
}
pub(super) fn for_count(self, count: usize) -> Self {
if self == Self::Simd4x && count < BLOCK_SIZE {
Self::Rounded
} else {
self
}
}
fn header_byte(self, width: u8) -> u8 {
((self as u8) << Self::HEADER_SHIFT) | width
}
pub fn parse(s: &str) -> Option<Self> {
match s.to_ascii_lowercase().as_str() {
"rounded" | "default" => Some(PostingCodec::Rounded),
"packed" | "bp128" | "exact" => Some(PostingCodec::Packed),
"pfor" | "optp4d" | "patched" => Some(PostingCodec::Pfor),
"simd4x" => Some(PostingCodec::Simd4x),
_ => None,
}
}
}
impl std::fmt::Display for PostingCodec {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
PostingCodec::Rounded => "rounded",
PostingCodec::Packed => "packed",
PostingCodec::Pfor => "pfor",
PostingCodec::Simd4x => "simd4x",
})
}
}
#[inline]
fn packed_bytes(count: usize, width: u8) -> usize {
(count * width as usize).div_ceil(8)
}
fn pack_pfor(values: &[u32], out: &mut Vec<u8>) -> u8 {
let (width, _, _) = find_optimal_bit_width(values);
let (packed, exceptions) = pack_with_exceptions(values, width);
out.push(exceptions.len() as u8);
out.extend_from_slice(&packed);
for (pos, high) in exceptions {
out.push(pos);
out.extend_from_slice(&high.to_le_bytes());
}
width
}
fn pfor_payload_len(input: &[u8], count: usize, width: u8) -> io::Result<usize> {
let n_exceptions = *input
.first()
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "posting block truncated"))?
as usize;
Ok(1 + packed_bytes(count, width) + n_exceptions * 5)
}
fn unpack_pfor(input: &[u8], width: u8, out: &mut [u32], count: usize) -> io::Result<()> {
let n_exceptions = *input
.first()
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "posting block truncated"))?
as usize;
let table_at = 1 + packed_bytes(count, width);
let table_end = table_at + n_exceptions * 5;
if input.len() < table_end {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"posting block exception table truncated",
));
}
let out = &mut out[..count];
unpack_bits(&input[1..table_at], width, out, count);
if width < 32 {
for entry in input[table_at..table_end].chunks_exact(5) {
let pos = entry[0] as usize;
let high = u32::from_le_bytes([entry[1], entry[2], entry[3], entry[4]]);
if pos < count {
out[pos] |= high << width;
}
}
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Posting {
pub doc_id: DocId,
pub term_freq: u32,
}
#[derive(Debug, Clone, Default)]
pub struct PostingList {
postings: Vec<Posting>,
}
impl PostingList {
pub fn new() -> Self {
Self::default()
}
pub fn with_capacity(capacity: usize) -> Self {
Self {
postings: Vec::with_capacity(capacity),
}
}
pub fn push(&mut self, doc_id: DocId, term_freq: u32) {
debug_assert!(
self.postings.is_empty() || self.postings.last().unwrap().doc_id < doc_id,
"Postings must be added in sorted order"
);
self.postings.push(Posting { doc_id, term_freq });
}
pub fn add(&mut self, doc_id: DocId, term_freq: u32) {
if let Some(last) = self.postings.last_mut()
&& last.doc_id == doc_id
{
last.term_freq += term_freq;
return;
}
self.postings.push(Posting { doc_id, term_freq });
}
pub fn doc_count(&self) -> u32 {
self.postings.len() as u32
}
pub fn len(&self) -> usize {
self.postings.len()
}
pub fn is_empty(&self) -> bool {
self.postings.is_empty()
}
pub fn iter(&self) -> impl Iterator<Item = &Posting> {
self.postings.iter()
}
}
pub struct PostingListIterator<'a> {
postings: &'a [Posting],
position: usize,
}
impl<'a> PostingListIterator<'a> {
pub fn new(posting_list: &'a PostingList) -> Self {
Self {
postings: &posting_list.postings,
position: 0,
}
}
pub fn doc(&self) -> DocId {
if self.position < self.postings.len() {
self.postings[self.position].doc_id
} else {
TERMINATED
}
}
pub fn term_freq(&self) -> u32 {
if self.position < self.postings.len() {
self.postings[self.position].term_freq
} else {
0
}
}
pub fn advance(&mut self) -> DocId {
self.position += 1;
self.doc()
}
pub fn seek(&mut self, target: DocId) -> DocId {
crate::observe::search_work!(posting_seeks += 1);
let remaining = &self.postings[self.position..];
let offset = remaining.partition_point(|p| p.doc_id < target);
self.position += offset;
self.doc()
}
pub fn size_hint(&self) -> usize {
self.postings.len().saturating_sub(self.position)
}
}
pub const TERMINATED: DocId = DocId::MAX;
pub const BLOCK_SIZE: usize = 128;
const L1_INTERVAL: usize = 8;
const L0_SIZE: usize = 16;
const L1_SIZE: usize = 4;
const FOOTER_SIZE: usize = 24;
const FOOTER_V2_SIZE: usize = FOOTER_SIZE + 20;
const FOOTER_MAGIC: u32 = 0x324C_5042;
const FLAG_POS_CURSORS: u32 = 1;
const FLAG_LEN_BOUNDS: u32 = 2;
const FLAG_L1_BOUNDS: u32 = 4;
const FLAG_RATIO_BOUNDS: u32 = 8;
const FLAG_IMPACT_BOUNDS: u32 = 16;
const FLAG_GROUP_IMPACT_BOUNDS: u32 = 32;
const FLAG_COMPACT_HEADERS: u32 = 64;
const FLAG_SHORT_CURSORS: u32 = 128;
fn append_ratio_groups(ratios: &mut Vec<u8>, blocks: usize) {
for start in (0..blocks).step_by(L1_INTERVAL) {
let ratio = (start..(start + L1_INTERVAL).min(blocks))
.map(|i| read_ratio(ratios, i))
.fold(f32::INFINITY, f32::min);
ratios.extend_from_slice(&ratio.to_le_bytes());
}
}
#[inline]
fn read_ratio(bytes: &[u8], index: usize) -> f32 {
let at = index * 4;
f32::from_le_bytes(bytes[at..at + 4].try_into().unwrap())
}
fn validate_ratios(bytes: &[u8]) -> io::Result<()> {
if bytes.chunks_exact(4).any(|value| {
let ratio = f32::from_le_bytes(value.try_into().unwrap());
!ratio.is_finite() || ratio < 0.0
}) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"invalid posting ratio bound",
));
}
Ok(())
}
fn lower_length_ratio(length: u32, tf: u32) -> f32 {
if tf == 0 {
return 0.0;
}
((length as f64 / tf as f64) as f32).next_down().max(0.0)
}
fn group_bounds_from_l0(l0: &[u8], l0_count: usize) -> Vec<u32> {
let mut groups = Vec::with_capacity(l0_count.div_ceil(L1_INTERVAL));
let mut idx = 0;
while idx < l0_count {
let end = (idx + L1_INTERVAL).min(l0_count);
let mut max_tf = 0u32;
let mut min_len = u32::MAX;
for block in idx..end {
let (_, _, _, word) = read_l0(l0, block);
let (tf, len) = unpack_bounds(word, true);
max_tf = max_tf.max(tf);
min_len = min_len.min(len.unwrap_or(1));
}
groups.push(pack_bounds(max_tf, min_len));
idx = end;
}
groups
}
#[inline]
fn pack_bounds(max_tf: u32, min_len: u32) -> u32 {
max_tf.min(u16::MAX as u32) | (min_len.min(u16::MAX as u32) << 16)
}
#[inline]
fn unpack_bounds(word: u32, packed: bool) -> (u32, Option<u32>) {
if packed {
(word & 0xFFFF, Some(word >> 16))
} else {
(f32::from_bits(word) as u32, None)
}
}
const CURSOR_SIZE: usize = 8;
#[derive(Debug, Clone, Copy)]
struct Footer {
compact_headers: bool,
short_cursors: bool,
stream_len: usize,
l0_count: usize,
l1_count: usize,
doc_count: u32,
max_tf: u32,
total_positions: u64,
has_cursors: bool,
len_bounds: bool,
l1_bounds: bool,
ratio_bounds: bool,
impact_bounds: bool,
group_impact_bounds: bool,
min_len: u32,
}
impl Footer {
fn parse(raw: &[u8]) -> io::Result<Self> {
Self::parse_tail(raw, raw.len())
}
fn parse_tail(raw: &[u8], total_len: usize) -> io::Result<Self> {
if raw.len() < FOOTER_SIZE {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"posting data too short",
));
}
let extended = raw.len() >= FOOTER_V2_SIZE
&& u32::from_le_bytes(raw[raw.len() - 4..].try_into().unwrap()) == FOOTER_MAGIC;
let f = raw.len()
- if extended {
FOOTER_V2_SIZE
} else {
FOOTER_SIZE
};
let stream_len = usize::try_from(u64::from_le_bytes(raw[f..f + 8].try_into().unwrap()))
.map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidData,
"posting stream exceeds address space",
)
})?;
let l0_count = u32::from_le_bytes(raw[f + 8..f + 12].try_into().unwrap()) as usize;
let l1_count = u32::from_le_bytes(raw[f + 12..f + 16].try_into().unwrap()) as usize;
let doc_count = u32::from_le_bytes(raw[f + 16..f + 20].try_into().unwrap());
let max_tf = u32::from_le_bytes(raw[f + 20..f + 24].try_into().unwrap());
let (total_positions, flags, min_len) = if extended {
let total = u64::from_le_bytes(raw[f + 24..f + 32].try_into().unwrap());
let flags = u32::from_le_bytes(raw[f + 32..f + 36].try_into().unwrap());
let min_len = u32::from_le_bytes(raw[f + 36..f + 40].try_into().unwrap());
(total, flags, min_len)
} else {
(0, 0, 0)
};
if flags
& !(FLAG_POS_CURSORS
| FLAG_LEN_BOUNDS
| FLAG_L1_BOUNDS
| FLAG_RATIO_BOUNDS
| FLAG_IMPACT_BOUNDS
| FLAG_GROUP_IMPACT_BOUNDS
| FLAG_COMPACT_HEADERS
| FLAG_SHORT_CURSORS)
!= 0
{
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"unknown posting footer flags",
));
}
let footer = Self {
compact_headers: flags & FLAG_COMPACT_HEADERS != 0,
short_cursors: flags & FLAG_SHORT_CURSORS != 0,
stream_len,
l0_count,
l1_count,
doc_count,
max_tf,
total_positions,
has_cursors: flags & FLAG_POS_CURSORS != 0,
len_bounds: flags & FLAG_LEN_BOUNDS != 0,
l1_bounds: flags & FLAG_L1_BOUNDS != 0,
ratio_bounds: flags & FLAG_RATIO_BOUNDS != 0,
impact_bounds: flags & FLAG_IMPACT_BOUNDS != 0,
group_impact_bounds: flags & FLAG_GROUP_IMPACT_BOUNDS != 0,
min_len,
};
if footer.short_cursors && (!footer.has_cursors || footer.total_positions > u32::MAX as u64)
{
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"invalid short position cursors",
));
}
if footer.group_impact_bounds && (!footer.impact_bounds || !footer.l1_bounds) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"group impacts require L0 impacts and L1 bounds",
));
}
let end = l0_count
.checked_mul(L0_SIZE + if footer.compact_headers { 4 } else { 0 })
.and_then(|n| {
l1_count
.checked_mul(L1_SIZE + if footer.l1_bounds { 4 } else { 0 })
.and_then(|m| n.checked_add(m))
})
.and_then(|n| {
l0_count
.checked_mul(if footer.has_cursors {
footer.cursor_size()
} else {
0
})
.and_then(|m| n.checked_add(m))
})
.and_then(|n| {
if footer.ratio_bounds {
l0_count
.checked_add(l1_count)
.and_then(|m| m.checked_mul(4))
.and_then(|m| n.checked_add(m))
} else {
Some(n)
}
})
.and_then(|n| n.checked_add(stream_len));
let footer_offset = total_len.saturating_sub(raw.len() - f);
if end.is_none_or(|end| {
if footer.impact_bounds {
!footer.len_bounds
|| !footer.ratio_bounds
|| l0_count
.checked_add(if footer.group_impact_bounds {
l1_count
} else {
0
})
.and_then(|n| n.checked_add(1))
.and_then(|n| n.checked_mul(4))
.and_then(|n| n.checked_add(end))
.is_none_or(|minimum| minimum > footer_offset)
} else {
end != footer_offset
}
}) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"posting list sections do not match the footer offset",
));
}
Ok(footer)
}
fn l0_start(&self) -> usize {
self.stream_len
}
fn impact_record_count(&self) -> usize {
self.l0_count
+ if self.group_impact_bounds {
self.l1_count
} else {
0
}
}
fn l0_end(&self) -> usize {
self.l0_start() + self.l0_count * L0_SIZE
}
fn l1_start(&self) -> usize {
self.l0_end()
+ if self.compact_headers {
self.l0_count * 4
} else {
0
}
}
fn cursor_size(&self) -> usize {
if self.short_cursors { 4 } else { CURSOR_SIZE }
}
fn l1_end(&self) -> usize {
self.l1_start() + self.l1_count * L1_SIZE
}
fn l1_bounds_end(&self) -> usize {
self.l1_end() + if self.l1_bounds { self.l1_count * 4 } else { 0 }
}
fn ratios_end(&self) -> usize {
self.cursors_end()
+ if self.ratio_bounds {
(self.l0_count + self.l1_count) * 4
} else {
0
}
}
fn cursors_end(&self) -> usize {
self.l1_bounds_end()
+ if self.has_cursors {
self.l0_count * self.cursor_size()
} else {
0
}
}
}
#[inline]
fn read_l0(bytes: &[u8], idx: usize) -> (u32, u32, u32, u32) {
let b = &bytes[idx * L0_SIZE..][..L0_SIZE];
let first_doc = u32::from_le_bytes([b[0], b[1], b[2], b[3]]);
let last_doc = u32::from_le_bytes([b[4], b[5], b[6], b[7]]);
let offset = u32::from_le_bytes([b[8], b[9], b[10], b[11]]);
let bounds = u32::from_le_bytes([b[12], b[13], b[14], b[15]]);
(first_doc, last_doc, offset, bounds)
}
#[inline]
fn verify_block_docs(
docs: &[u32],
first: u32,
last: u32,
byte_gaps: Option<&[u8]>,
strict_gap_width: Option<u8>,
) -> bool {
let ordered = if strict_gap_width.is_some_and(|width| width <= 25) {
debug_assert!(docs.len() <= BLOCK_SIZE);
true
} else if let Some(gaps) = byte_gaps {
debug_assert_eq!(gaps.len() + 1, docs.len());
let mut nonzero = true;
for &gap in gaps {
nonzero &= gap != 0;
}
nonzero
} else {
let mut ordered = true;
for pair in docs.windows(2) {
ordered &= pair[0] < pair[1];
}
ordered
};
ordered
&& first <= last
&& last != TERMINATED
&& docs.first() == Some(&first)
&& docs.last() == Some(&last)
}
#[inline]
fn write_l0(buf: &mut Vec<u8>, first_doc: u32, last_doc: u32, offset: u32, bounds: u32) {
buf.extend_from_slice(&first_doc.to_le_bytes());
buf.extend_from_slice(&last_doc.to_le_bytes());
buf.extend_from_slice(&offset.to_le_bytes());
buf.extend_from_slice(&bounds.to_le_bytes());
}
#[inline]
fn block_len_from_l0(l0_bytes: &[u8], l0_count: usize, stream_len: usize, idx: usize) -> usize {
let (_, _, offset, _) = read_l0(l0_bytes, idx);
let end = if idx + 1 < l0_count {
read_l0(l0_bytes, idx + 1).2 as usize
} else {
stream_len
};
end.saturating_sub(offset as usize)
}
struct EncodedBlock {
doc_bits: u8,
tf_bits: u8,
}
fn encode_block_arrays(
codec: PostingCodec,
deltas: &[u32],
tfs: &[u32],
stream: &mut Vec<u8>,
) -> EncodedBlock {
let codec = codec.for_count(tfs.len());
match codec {
PostingCodec::Simd4x => EncodedBlock {
doc_bits: codec.header_byte(bitpacking4x::encode_gaps(deltas, stream)),
tf_bits: bitpacking4x::encode(tfs, stream),
},
PostingCodec::Rounded => {
let max_delta = deltas.iter().copied().max().unwrap_or(0);
let doc_bits = simd::round_bit_width(simd::bits_needed(max_delta));
let max_tf = tfs.iter().copied().max().unwrap_or(0);
let tf_bits = simd::round_bit_width(simd::bits_needed(max_tf));
if !deltas.is_empty() {
let rounded = simd::RoundedBitWidth::from_u8(doc_bits);
let start = stream.len();
stream.resize(start + deltas.len() * rounded.bytes_per_value(), 0);
simd::pack_rounded(deltas, rounded, &mut stream[start..]);
}
{
let rounded = simd::RoundedBitWidth::from_u8(tf_bits);
let start = stream.len();
stream.resize(start + tfs.len() * rounded.bytes_per_value(), 0);
simd::pack_rounded(tfs, rounded, &mut stream[start..]);
}
EncodedBlock {
doc_bits: codec.header_byte(doc_bits),
tf_bits,
}
}
PostingCodec::Packed => {
let max_delta = deltas.iter().copied().max().unwrap_or(0);
let doc_bits = simd::bits_needed(max_delta);
let max_tf = tfs.iter().copied().max().unwrap_or(0);
let tf_bits = simd::bits_needed(max_tf);
pack_bits(deltas, doc_bits, stream);
pack_bits(tfs, tf_bits, stream);
EncodedBlock {
doc_bits: codec.header_byte(doc_bits),
tf_bits,
}
}
PostingCodec::Pfor => {
let doc_bits = if deltas.is_empty() {
0
} else {
pack_pfor(deltas, stream)
};
let tf_bits = pack_pfor(tfs, stream);
EncodedBlock {
doc_bits: codec.header_byte(doc_bits),
tf_bits,
}
}
}
}
#[derive(Debug, Clone)]
pub struct BlockPostingList {
compact_headers: bool,
short_cursors: bool,
verify_content: bool,
content_error: Option<std::sync::Arc<reader::PostingIntegrity>>,
stream: OwnedBytes,
l0_bytes: OwnedBytes,
l0_count: usize,
l1_docs: GroupWords,
l1_bounds: GroupWords,
ratios: Option<OwnedBytes>,
impacts: Option<ImpactTable>,
doc_count: u32,
max_tf: u32,
pos_cursors: Option<OwnedBytes>,
total_positions: u64,
len_bounds: bool,
min_len: u32,
}
impl BlockPostingList {
#[inline]
fn read_l0_entry(&self, idx: usize) -> (u32, u32, u32, u32) {
read_l0(&self.l0_bytes, idx)
}
pub fn from_posting_list(list: &PostingList) -> io::Result<Self> {
Self::build(list, false, None, PostingCodec::Rounded, false, false)
}
pub fn from_posting_list_with_codec(
list: &PostingList,
codec: PostingCodec,
) -> io::Result<Self> {
Self::build(list, false, None, codec, false, false)
}
pub fn from_posting_list_with(
list: &PostingList,
with_positions: bool,
length_of: Option<&dyn Fn(DocId) -> u32>,
) -> io::Result<Self> {
Self::build(
list,
with_positions,
length_of,
PostingCodec::Rounded,
false,
false,
)
}
pub fn from_posting_list_with_options(
list: &PostingList,
with_positions: bool,
length_of: Option<&dyn Fn(DocId) -> u32>,
codec: PostingCodec,
) -> io::Result<Self> {
Self::build(list, with_positions, length_of, codec, false, false)
}
pub fn from_posting_list_with_ratio_bounds(
list: &PostingList,
with_positions: bool,
length_of: Option<&dyn Fn(DocId) -> u32>,
codec: PostingCodec,
) -> io::Result<Self> {
Self::build(list, with_positions, length_of, codec, true, false)
}
pub fn from_posting_list_with_impact_bounds(
list: &PostingList,
with_positions: bool,
length_of: Option<&dyn Fn(DocId) -> u32>,
codec: PostingCodec,
) -> io::Result<Self> {
if length_of.is_none() {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"impact bounds require scoring lengths",
));
}
Self::build(list, with_positions, length_of, codec, true, true)
}
fn build(
list: &PostingList,
with_positions: bool,
length_of: Option<&dyn Fn(DocId) -> u32>,
codec: PostingCodec,
ratio_bounds: bool,
impact_bounds: bool,
) -> io::Result<Self> {
let capped = |id: DocId| {
length_of
.map_or(0, |length_of| length_of(id))
.min(crate::segment::chunk_map::MAX_CHUNK_LENGTH)
};
let length_of: Option<&dyn Fn(DocId) -> u32> = if ratio_bounds {
length_of.map(|_| &capped as &dyn Fn(DocId) -> u32)
} else {
length_of
};
let mut ratios = (ratio_bounds && length_of.is_some()).then(Vec::new);
let mut impacts = (impact_bounds && length_of.is_some() && list.len() > BLOCK_SIZE)
.then(|| ImpactBuilder::with_groups(list.len().div_ceil(BLOCK_SIZE)))
.transpose()?;
let mut points = [(0u32, 0u32); BLOCK_SIZE];
let mut stream: Vec<u8> = Vec::new();
let mut l0_buf: Vec<u8> = Vec::new();
let mut l1_docs: Vec<u32> = Vec::new();
let mut cursors: Vec<u8> = Vec::new();
let mut positions_so_far = 0u64;
let mut l0_count = 0usize;
let mut max_tf = 0u32;
let mut list_min_len = u32::MAX;
let postings = &list.postings;
let mut i = 0;
let mut deltas = Vec::with_capacity(BLOCK_SIZE);
let mut tf_buf = Vec::with_capacity(BLOCK_SIZE);
while i < postings.len() {
if stream.len() > u32::MAX as usize {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"posting list stream exceeds u32::MAX bytes",
));
}
let block_start = stream.len() as u32;
let block_end = (i + BLOCK_SIZE).min(postings.len());
let block = &postings[i..block_end];
let count = block.len();
let block_max_tf = block.iter().map(|p| p.term_freq).max().unwrap_or(0);
max_tf = max_tf.max(block_max_tf);
let base_doc_id = block.first().unwrap().doc_id;
let last_doc_id = block.last().unwrap().doc_id;
deltas.clear();
let mut prev = base_doc_id;
for posting in block.iter().skip(1) {
deltas.push(posting.doc_id - prev);
prev = posting.doc_id;
}
tf_buf.clear();
tf_buf.extend(block.iter().map(|p| p.term_freq));
stream.write_u16::<LittleEndian>(count as u16)?;
stream.write_u32::<LittleEndian>(base_doc_id)?;
let header_at = stream.len();
stream.push(0);
stream.push(0);
let encoded = encode_block_arrays(codec, &deltas, &tf_buf, &mut stream);
stream[header_at] = encoded.doc_bits;
stream[header_at + 1] = encoded.tf_bits;
let block_min_len = length_of.map_or(1, |length_of| {
block
.iter()
.map(|p| length_of(p.doc_id).max(1))
.min()
.unwrap_or(1)
});
list_min_len = list_min_len.min(block_min_len);
if let Some(ratios) = &mut ratios {
let ratio = block
.iter()
.map(|p| lower_length_ratio(length_of.unwrap()(p.doc_id).max(1), p.term_freq))
.fold(f32::INFINITY, f32::min);
ratios.extend_from_slice(&ratio.to_le_bytes());
}
if let Some(impacts) = &mut impacts {
for (point, posting) in points.iter_mut().zip(block) {
let length = length_of.unwrap()(posting.doc_id);
*point = (
posting.term_freq,
if length == 0 {
posting.term_freq
} else {
length
},
);
}
impacts.append_points(&mut points[..count])?;
}
write_l0(
&mut l0_buf,
base_doc_id,
last_doc_id,
block_start,
pack_bounds(block_max_tf, block_min_len),
);
l0_count += 1;
if with_positions {
cursors.extend_from_slice(&positions_so_far.to_le_bytes());
positions_so_far += block.iter().map(|p| p.term_freq as u64).sum::<u64>();
}
if l0_count.is_multiple_of(L1_INTERVAL) {
l1_docs.push(last_doc_id);
}
i = block_end;
}
if !l0_count.is_multiple_of(L1_INTERVAL) && l0_count > 0 {
let (_, last_doc, _, _) = read_l0(&l0_buf, l0_count - 1);
l1_docs.push(last_doc);
}
let l1_bounds = group_bounds_from_l0(&l0_buf, l0_count);
if let Some(ratios) = &mut ratios {
append_ratio_groups(ratios, l0_count);
}
Ok(Self {
compact_headers: false,
short_cursors: false,
verify_content: true,
content_error: None,
stream: OwnedBytes::new(stream),
l0_bytes: OwnedBytes::new(l0_buf),
l0_count,
l1_docs: l1_docs.into(),
l1_bounds: l1_bounds.into(),
ratios: ratios.map(OwnedBytes::new),
impacts: if let Some(mut impacts) = impacts {
impacts.append_groups_with(l0_count, |_| Ok(None))?;
impacts.finish()
} else {
None
},
doc_count: postings.len() as u32,
max_tf,
pos_cursors: with_positions.then(|| OwnedBytes::new(cursors)),
total_positions: positions_so_far,
len_bounds: true,
min_len: if list_min_len == u32::MAX {
1
} else {
list_min_len
},
})
}
pub fn serialize<W: Write>(&self, writer: &mut W) -> io::Result<()> {
self.serialize_layout(writer, self.compact_headers)
}
#[cfg(all(feature = "native", test))]
pub(crate) fn has_compact_headers(&self) -> bool {
self.compact_headers
}
pub fn serialize_compact<W: Write>(&self, writer: &mut W) -> io::Result<()> {
let compact =
(0..self.num_blocks()).all(|i| self.block_codec(i) != Some(PostingCodec::Pfor));
self.serialize_layout(writer, compact)
}
fn serialize_layout<W: Write>(&self, writer: &mut W, compact: bool) -> io::Result<()> {
let source_header = if self.compact_headers { 0 } else { 8 };
let output_header = if compact { 0 } else { 8 };
let output_offset =
|offset: usize, block: usize| offset - block * source_header + block * output_header;
let stream_len = output_offset(self.stream.len(), self.num_blocks());
if compact == self.compact_headers {
writer.write_all(&self.stream)?;
} else {
for i in 0..self.num_blocks() {
if !compact {
writer.write_all(&self.block_header(i))?;
}
writer.write_all(self.block_payload(i))?;
}
}
for i in 0..self.num_blocks() {
let (first, last, offset, bounds) = self.read_l0_entry(i);
writer.write_u32::<LittleEndian>(first)?;
writer.write_u32::<LittleEndian>(last)?;
writer.write_u32::<LittleEndian>(
u32::try_from(output_offset(offset as usize, i))
.map_err(|_| io::Error::other("posting stream offset overflow"))?,
)?;
writer.write_u32::<LittleEndian>(bounds)?;
}
if compact {
for i in 0..self.num_blocks() {
let header = self.block_header(i);
writer.write_all(&header[..2])?;
writer.write_all(&header[6..])?;
}
}
writer.write_all(self.l1_docs.bytes())?;
writer.write_all(self.l1_bounds.bytes())?;
let short_cursors =
compact && self.pos_cursors.is_some() && self.total_positions <= u32::MAX as u64;
if self.pos_cursors.is_some() {
for i in 0..self.num_blocks() {
let cursor = self.pos_cursor(i).unwrap();
if short_cursors {
writer.write_u32::<LittleEndian>(cursor as u32)?;
} else {
writer.write_u64::<LittleEndian>(cursor)?;
}
}
}
if let Some(ratios) = &self.ratios {
writer.write_all(ratios)?;
}
if let Some(impacts) = &self.impacts {
writer.write_all(impacts.bytes())?;
}
Self::write_footer(
writer,
stream_len as u64,
self.l0_count,
self.l1_docs.len(),
self.doc_count,
self.max_tf,
self.total_positions,
self.pos_cursors.is_some(),
self.len_bounds.then_some(self.min_len),
!self.l1_bounds.is_empty(),
self.ratios.is_some(),
self.impacts.is_some(),
self.has_group_impact_bounds(),
if compact { FLAG_COMPACT_HEADERS } else { 0 }
| if short_cursors { FLAG_SHORT_CURSORS } else { 0 },
)
}
#[allow(clippy::too_many_arguments)]
fn write_footer<W: Write>(
writer: &mut W,
stream_len: u64,
l0_count: usize,
l1_count: usize,
doc_count: u32,
max_tf: u32,
total_positions: u64,
has_cursors: bool,
min_len: Option<u32>,
l1_bounds: bool,
ratio_bounds: bool,
impact_bounds: bool,
group_impact_bounds: bool,
layout_flags: u32,
) -> io::Result<()> {
writer.write_u64::<LittleEndian>(stream_len)?;
writer.write_u32::<LittleEndian>(l0_count as u32)?;
writer.write_u32::<LittleEndian>(l1_count as u32)?;
writer.write_u32::<LittleEndian>(doc_count)?;
writer.write_u32::<LittleEndian>(max_tf)?;
writer.write_u64::<LittleEndian>(total_positions)?;
let mut flags = layout_flags;
if has_cursors {
flags |= FLAG_POS_CURSORS;
}
if min_len.is_some() {
flags |= FLAG_LEN_BOUNDS;
}
if l1_bounds {
flags |= FLAG_L1_BOUNDS;
}
if ratio_bounds {
flags |= FLAG_RATIO_BOUNDS;
}
if impact_bounds {
flags |= FLAG_IMPACT_BOUNDS;
}
if group_impact_bounds {
flags |= FLAG_GROUP_IMPACT_BOUNDS;
}
writer.write_u32::<LittleEndian>(flags)?;
writer.write_u32::<LittleEndian>(min_len.unwrap_or(0))?;
writer.write_u32::<LittleEndian>(FOOTER_MAGIC)?;
Ok(())
}
pub fn deserialize(raw: &[u8]) -> io::Result<Self> {
Self::deserialize_zero_copy(OwnedBytes::new(raw.to_vec()))
}
pub fn deserialize_zero_copy(raw: OwnedBytes) -> io::Result<Self> {
let footer = Self::validate_bytes(&raw)?;
Ok(Self::from_layout(raw, footer))
}
fn validate_bytes(raw: &[u8]) -> io::Result<Footer> {
let footer = Footer::parse(raw)?;
validation::validate_list(raw, &footer)?;
if footer.ratio_bounds {
validate_ratios(&raw[footer.cursors_end()..footer.ratios_end()])?;
}
if footer.impact_bounds {
ImpactTable::validate(
&raw[footer.ratios_end()..raw.len() - FOOTER_V2_SIZE],
footer.impact_record_count(),
)?;
}
Ok(footer)
}
fn from_layout(raw: OwnedBytes, footer: Footer) -> Self {
let ratios = footer
.ratio_bounds
.then(|| raw.slice(footer.cursors_end()..footer.ratios_end()));
let l1_docs = GroupWords::borrowed(raw.slice(footer.l1_start()..footer.l1_end()));
let l1_bounds = GroupWords::borrowed(raw.slice(footer.l1_end()..footer.l1_bounds_end()));
let pos_cursors = footer
.has_cursors
.then(|| raw.slice(footer.l1_bounds_end()..footer.cursors_end()));
Self {
compact_headers: footer.compact_headers,
short_cursors: footer.short_cursors,
verify_content: true,
content_error: None,
stream: raw.slice(0..footer.stream_len),
l0_bytes: raw.slice(footer.l0_start()..footer.l1_start()),
l0_count: footer.l0_count,
l1_docs,
l1_bounds,
ratios,
impacts: footer.impact_bounds.then(|| {
ImpactTable::from_validated(
raw.slice(footer.ratios_end()..raw.len() - FOOTER_V2_SIZE),
footer.impact_record_count(),
)
}),
doc_count: footer.doc_count,
max_tf: footer.max_tf,
pos_cursors,
total_positions: footer.total_positions,
len_bounds: footer.len_bounds,
min_len: footer.min_len,
}
}
pub fn min_len(&self) -> Option<u32> {
self.len_bounds.then_some(self.min_len)
}
pub fn has_ratio_bounds(&self) -> bool {
self.ratios.is_some()
}
pub fn has_impact_bounds(&self) -> bool {
self.impacts.is_some()
}
pub fn block_impact_point_count(&self, block: usize) -> Option<usize> {
if block >= self.l0_count {
return None;
}
self.impacts
.as_ref()?
.record(block)
.map(|r| r.first().copied().unwrap_or(0) as usize)
}
pub fn has_group_impact_bounds(&self) -> bool {
self.impacts
.as_ref()
.is_some_and(|t| t.record_count() > self.l0_count)
}
#[cfg(test)]
pub fn group_impact_point_count(&self, block: usize) -> Option<usize> {
if block >= self.l0_count {
return None;
}
self.impacts
.as_ref()?
.record(self.l0_count + block / L1_INTERVAL)
.map(|r| r.first().copied().unwrap_or(0) as usize)
}
pub(crate) fn group_impact_minimum(
&self,
block: usize,
reciprocal: f64,
ratio: f64,
) -> Option<f64> {
if block >= self.l0_count {
return None;
}
self.impacts
.as_ref()?
.minimum(self.l0_count + block / L1_INTERVAL, reciprocal, ratio)
}
pub(crate) fn block_impact_minimum(
&self,
block: usize,
reciprocal: f64,
ratio: f64,
) -> Option<f64> {
if block >= self.l0_count {
return None;
}
self.impacts.as_ref()?.minimum(block, reciprocal, ratio)
}
pub(crate) fn block_length_ratio(&self, block: usize) -> f32 {
self.ratios
.as_ref()
.map_or(0.0, |ratios| read_ratio(ratios, block))
}
pub(crate) fn group_length_ratio(&self, block: usize) -> f32 {
self.ratios.as_ref().map_or(0.0, |ratios| {
read_ratio(ratios, self.l0_count + block / L1_INTERVAL)
})
}
#[inline]
pub fn block_bounds(&self, block_idx: usize) -> Option<(u32, Option<u32>)> {
if block_idx >= self.l0_count {
return None;
}
let (_, _, _, word) = self.read_l0_entry(block_idx);
let (max_tf, min_len) = unpack_bounds(word, self.len_bounds);
let max_tf = if self.len_bounds && max_tf == u16::MAX as u32 {
max_tf.max(self.max_tf)
} else {
max_tf
};
Some((max_tf, min_len))
}
#[inline]
pub fn group_bounds(&self, block_idx: usize) -> Option<(u32, u32)> {
if block_idx >= self.l0_count {
return None;
}
let word = self.l1_bounds.get(block_idx / L1_INTERVAL)?;
let (max_tf, min_len) = unpack_bounds(word, true);
let max_tf = if max_tf == u16::MAX as u32 {
max_tf.max(self.max_tf)
} else {
max_tf
};
Some((max_tf, min_len.unwrap_or(1)))
}
#[inline]
pub fn group_last_doc(&self, block_idx: usize) -> Option<DocId> {
self.l1_docs.get(block_idx / L1_INTERVAL)
}
#[inline]
pub fn is_group_start(&self, block_idx: usize) -> bool {
block_idx.is_multiple_of(L1_INTERVAL)
}
#[inline]
pub fn next_group_block(&self, block_idx: usize) -> usize {
((block_idx / L1_INTERVAL + 1) * L1_INTERVAL).min(self.l0_count)
}
pub fn has_cursors_bytes(raw: &[u8]) -> bool {
Footer::parse(raw).is_ok_and(|footer| footer.has_cursors)
}
pub fn has_position_cursors(&self) -> bool {
self.pos_cursors.is_some()
}
#[inline]
fn block_header(&self, block: usize) -> [u8; 8] {
let (first, _, offset, _) = self.read_l0_entry(block);
if self.compact_headers {
let headers = &self.l0_bytes[self.l0_count * L0_SIZE..];
let mut header = [0; 8];
header[..2].copy_from_slice(&headers[block * 4..block * 4 + 2]);
header[2..6].copy_from_slice(&first.to_le_bytes());
header[6..].copy_from_slice(&headers[block * 4 + 2..block * 4 + 4]);
header
} else {
self.stream[offset as usize..offset as usize + 8]
.try_into()
.unwrap()
}
}
#[inline]
fn block_payload(&self, block: usize) -> &[u8] {
let offset = self.read_l0_entry(block).2 as usize;
let header = if self.compact_headers { 0 } else { 8 };
&self.stream[offset + header..offset + self.block_len(block)]
}
pub fn total_positions(&self) -> u64 {
self.total_positions
}
#[inline]
pub fn pos_cursor(&self, block_idx: usize) -> Option<u64> {
let cursors = self.pos_cursors.as_ref()?;
let size = if self.short_cursors { 4 } else { CURSOR_SIZE };
let p = block_idx * size;
cursors.get(p..p + size).map(|b| {
if self.short_cursors {
u64::from(u32::from_le_bytes(b.try_into().unwrap()))
} else {
u64::from_le_bytes(b.try_into().unwrap())
}
})
}
pub fn doc_count(&self) -> u32 {
self.doc_count
}
pub fn max_tf(&self) -> u32 {
self.max_tf
}
pub fn num_blocks(&self) -> usize {
self.l0_count
}
pub fn block_max_tf(&self, block_idx: usize) -> Option<u32> {
self.block_bounds(block_idx).map(|(max_tf, _)| max_tf)
}
pub fn concatenate_blocks(sources: &[(BlockPostingList, u32)]) -> io::Result<Self> {
let mut previous_last = None;
let mut total_docs = 0u32;
let mut total_positions = 0u64;
for (source, offset) in sources {
validation::validate_remap(
&source.l0_bytes,
source.l0_count,
*offset,
&mut previous_last,
)?;
total_docs = total_docs.checked_add(source.doc_count).ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidData, "merged posting count overflow")
})?;
total_positions = total_positions
.checked_add(source.total_positions)
.ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidData,
"merged position cursor overflow",
)
})?;
}
let mut stream: Vec<u8> = Vec::new();
let mut l0_buf: Vec<u8> = Vec::new();
let mut l1_docs: Vec<u32> = Vec::new();
let mut l0_count = 0usize;
let mut ratios = sources
.iter()
.any(|(s, _)| s.has_ratio_bounds())
.then(Vec::new);
let mut impacts = if sources.iter().any(|(s, _)| s.has_impact_bounds()) {
let blocks = sources
.iter()
.try_fold(0usize, |n, (s, _)| n.checked_add(s.num_blocks()))
.ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidData, "impact block count overflow")
})?;
Some(ImpactBuilder::with_groups(blocks)?)
} else {
None
};
let mut max_tf = 0u32;
let all_cursors = sources.iter().all(|(s, _)| s.has_position_cursors());
if !all_cursors && sources.iter().any(|(s, _)| s.has_position_cursors()) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"cannot concatenate posting lists with and without position cursors",
));
}
let mut cursors: Vec<u8> = Vec::new();
let mut positions_before = 0u64;
let mut min_len = u32::MAX;
for (source, doc_offset) in sources {
max_tf = max_tf.max(source.max_tf);
min_len = min_len.min(source.min_len().unwrap_or(1));
for block_idx in 0..source.num_blocks() {
if let Some(impacts) = &mut impacts {
impacts.append(
source
.impacts
.as_ref()
.and_then(|t| t.record(block_idx))
.unwrap_or(&[]),
)?;
}
if let Some(ratios) = &mut ratios {
ratios.extend_from_slice(&source.block_length_ratio(block_idx).to_le_bytes());
}
if all_cursors {
let cursor = source.pos_cursor(block_idx).unwrap_or(0) + positions_before;
cursors.extend_from_slice(&cursor.to_le_bytes());
}
let (first_doc, last_doc, _, word) = source.read_l0_entry(block_idx);
let (block_max_tf, block_min_len) = unpack_bounds(word, source.len_bounds);
let bounds = pack_bounds(block_max_tf, block_min_len.unwrap_or(1));
let header = source.block_header(block_idx);
let count = u16::from_le_bytes(header[..2].try_into().unwrap());
if stream.len() > u32::MAX as usize {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"posting list stream exceeds u32::MAX bytes during concatenation",
));
}
let new_offset = stream.len() as u32;
stream.write_u16::<LittleEndian>(count)?;
stream.write_u32::<LittleEndian>(first_doc + doc_offset)?;
stream.extend_from_slice(&header[6..]);
stream.extend_from_slice(source.block_payload(block_idx));
let new_last = last_doc + doc_offset;
write_l0(
&mut l0_buf,
first_doc + doc_offset,
new_last,
new_offset,
bounds,
);
l0_count += 1;
if l0_count.is_multiple_of(L1_INTERVAL) {
l1_docs.push(new_last);
}
}
positions_before += source.total_positions;
}
if !l0_count.is_multiple_of(L1_INTERVAL) && l0_count > 0 {
let (_, last_doc, _, _) = read_l0(&l0_buf, l0_count - 1);
l1_docs.push(last_doc);
}
let l1_bounds = group_bounds_from_l0(&l0_buf, l0_count);
if let Some(ratios) = &mut ratios {
append_ratio_groups(ratios, l0_count);
}
Ok(Self {
compact_headers: false,
short_cursors: false,
verify_content: true,
content_error: None,
stream: OwnedBytes::new(stream),
l0_bytes: OwnedBytes::new(l0_buf),
l0_count,
l1_docs: l1_docs.into(),
l1_bounds: l1_bounds.into(),
ratios: ratios.map(OwnedBytes::new),
impacts: if let Some(mut impacts) = impacts {
let mut source = 0;
let mut base = 0;
impacts.append_groups_with(l0_count, |range| {
while base + sources[source].0.num_blocks() <= range.start {
base += sources[source].0.num_blocks();
source += 1;
}
let list = &sources[source].0;
let local = range.start - base;
let record = if local.is_multiple_of(L1_INTERVAL)
&& range.end - base == (local + L1_INTERVAL).min(list.num_blocks())
{
list.impacts
.as_ref()
.and_then(|t| t.record(list.l0_count + local / L1_INTERVAL))
} else {
None
};
Ok(record)
})?;
impacts.finish()
} else {
None
},
doc_count: total_docs,
max_tf,
pos_cursors: all_cursors.then(|| OwnedBytes::new(cursors)),
total_positions: if all_cursors { total_positions } else { 0 },
len_bounds: true,
min_len: if min_len == u32::MAX { 1 } else { min_len },
})
}
pub fn concatenate_streaming<W: Write>(
sources: &[(&[u8], u32)], writer: &mut W,
) -> crate::Result<(u32, usize)> {
let mut metas: Vec<Footer> = Vec::with_capacity(sources.len());
let mut total_docs = 0u32;
let mut merged_max_tf = 0u32;
let mut merged_min_len = u32::MAX;
let mut previous_last = None;
let mut total_positions = 0u64;
for (source_index, (raw, offset)) in sources.iter().enumerate() {
let invalid_source = |error: io::Error| {
crate::Error::Corruption(format!(
"posting list source {source_index} is invalid: {error}"
))
};
let footer = Self::validate_bytes(raw).map_err(invalid_source)?;
validation::validate_remap(
&raw[footer.l0_start()..footer.l0_end()],
footer.l0_count,
*offset,
&mut previous_last,
)
.map_err(invalid_source)?;
total_docs = total_docs
.checked_add(footer.doc_count)
.ok_or_else(|| crate::Error::Corruption("merged posting count overflow".into()))?;
total_positions = total_positions
.checked_add(footer.total_positions)
.ok_or_else(|| {
crate::Error::Corruption("merged position cursor overflow".into())
})?;
merged_max_tf = merged_max_tf.max(footer.max_tf);
merged_min_len = merged_min_len.min(if footer.len_bounds { footer.min_len } else { 1 });
metas.push(footer);
}
if sources.len() == 1 && sources[0].1 == 0 {
writer.write_all(sources[0].0)?;
return Ok((metas[0].doc_count, sources[0].0.len()));
}
let compact_output = !metas.is_empty() && metas.iter().all(|meta| meta.compact_headers);
let short_cursors = compact_output && total_positions <= u32::MAX as u64;
let mut out_headers = Vec::new();
let all_cursors = metas.iter().all(|m| m.has_cursors);
if !all_cursors && metas.iter().any(|m| m.has_cursors) {
return Err(crate::Error::Corruption(
"cannot concatenate posting lists with and without position cursors".into(),
));
}
let mut out_impacts = if metas.iter().any(|m| m.impact_bounds) {
let blocks = metas
.iter()
.try_fold(0usize, |n, m| n.checked_add(m.l0_count))
.ok_or_else(|| crate::Error::Corruption("impact block count overflow".into()))?;
Some(ImpactBuilder::with_groups(blocks)?)
} else {
None
};
let mut out_ratios = metas.iter().any(|m| m.ratio_bounds).then(Vec::new);
let mut out_l0: Vec<u8> = Vec::new();
let mut out_l1_docs: Vec<u32> = Vec::new();
let mut out_cursors: Vec<u8> = Vec::new();
let mut positions_before = 0u64;
let mut out_l0_count = 0usize;
let mut stream_written = 0u64;
let mut patch_buf = [0u8; 8];
for (src_idx, meta) in metas.iter().enumerate() {
let (raw, doc_offset) = &sources[src_idx];
let l0_base = meta.l0_start(); let src_stream = &raw[..meta.stream_len];
let cursors_base = meta.l1_bounds_end();
for i in 0..meta.l0_count {
if let Some(impacts) = &mut out_impacts {
let record = if meta.impact_bounds {
ImpactTable::record_from_validated(
&raw[meta.ratios_end()..raw.len() - FOOTER_V2_SIZE],
meta.impact_record_count(),
i,
)
} else {
&[]
};
impacts.append(record)?;
}
if let Some(ratios) = &mut out_ratios {
let ratio = if meta.ratio_bounds {
read_ratio(&raw[meta.cursors_end()..], i)
} else {
0.0
};
ratios.extend_from_slice(&ratio.to_le_bytes());
}
let (first_doc, last_doc, offset, word) = read_l0(&raw[l0_base..], i);
let (block_max_tf, block_min_len) = unpack_bounds(word, meta.len_bounds);
let bounds = pack_bounds(block_max_tf, block_min_len.unwrap_or(1));
if all_cursors {
let size = meta.cursor_size();
let p = cursors_base + i * size;
let cursor = if meta.short_cursors {
u64::from(u32::from_le_bytes(raw[p..p + size].try_into().unwrap()))
} else {
u64::from_le_bytes(raw[p..p + size].try_into().unwrap())
};
if short_cursors {
out_cursors
.extend_from_slice(&((cursor + positions_before) as u32).to_le_bytes());
} else {
out_cursors.extend_from_slice(&(cursor + positions_before).to_le_bytes());
}
}
let blk_size =
block_len_from_l0(&raw[l0_base..], meta.l0_count, meta.stream_len, i);
let block = &src_stream[offset as usize..offset as usize + blk_size];
let new_last = last_doc + doc_offset;
if stream_written > u32::MAX as u64 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"posting list stream exceeds u32::MAX bytes during streaming merge",
)
.into());
}
write_l0(
&mut out_l0,
first_doc + doc_offset,
new_last,
stream_written as u32,
bounds,
);
out_l0_count += 1;
if out_l0_count.is_multiple_of(L1_INTERVAL) {
out_l1_docs.push(new_last);
}
let payload = if meta.compact_headers {
let header = &raw[meta.l0_end() + i * 4..meta.l0_end() + i * 4 + 4];
patch_buf[..2].copy_from_slice(&header[..2]);
patch_buf[6..].copy_from_slice(&header[2..]);
block
} else {
patch_buf.copy_from_slice(&block[..8]);
&block[8..]
};
patch_buf[2..6].copy_from_slice(&(first_doc + doc_offset).to_le_bytes());
if compact_output {
out_headers.extend_from_slice(&patch_buf[..2]);
out_headers.extend_from_slice(&patch_buf[6..]);
} else {
writer.write_all(&patch_buf)?;
}
writer.write_all(payload)?;
stream_written += (if compact_output { 0 } else { 8 } + payload.len()) as u64;
}
positions_before += meta.total_positions;
}
if !out_l0_count.is_multiple_of(L1_INTERVAL) && out_l0_count > 0 {
let (_, last_doc, _, _) = read_l0(&out_l0, out_l0_count - 1);
out_l1_docs.push(last_doc);
}
let out_l1_bounds = group_bounds_from_l0(&out_l0, out_l0_count);
writer.write_all(&out_l0)?;
writer.write_all(&out_headers)?;
for &doc in &out_l1_docs {
writer.write_u32::<LittleEndian>(doc)?;
}
for &bounds in &out_l1_bounds {
writer.write_u32::<LittleEndian>(bounds)?;
}
writer.write_all(&out_cursors)?;
if let Some(ratios) = &mut out_ratios {
append_ratio_groups(ratios, out_l0_count);
writer.write_all(ratios)?;
}
let out_impacts = if let Some(mut impacts) = out_impacts {
let mut source = 0;
let mut base = 0;
impacts.append_groups_with(out_l0_count, |range| {
while base + metas[source].l0_count <= range.start {
base += metas[source].l0_count;
source += 1;
}
let meta = &metas[source];
let local = range.start - base;
let record = if meta.group_impact_bounds
&& local.is_multiple_of(L1_INTERVAL)
&& range.end - base == (local + L1_INTERVAL).min(meta.l0_count)
{
let raw = sources[source].0;
Some(ImpactTable::record_from_validated(
&raw[meta.ratios_end()..raw.len() - FOOTER_V2_SIZE],
meta.impact_record_count(),
meta.l0_count + local / L1_INTERVAL,
))
} else {
None
};
Ok(record)
})?;
impacts.finish()
} else {
None
};
if let Some(impacts) = &out_impacts {
writer.write_all(impacts.bytes())?;
}
Self::write_footer(
writer,
stream_written,
out_l0_count,
out_l1_docs.len(),
total_docs,
merged_max_tf,
if all_cursors { total_positions } else { 0 },
all_cursors,
Some(if merged_min_len == u32::MAX {
1
} else {
merged_min_len
}),
true,
out_ratios.is_some(),
out_impacts.is_some(),
out_impacts.is_some(),
if compact_output {
FLAG_COMPACT_HEADERS
} else {
0
} | if short_cursors && all_cursors {
FLAG_SHORT_CURSORS
} else {
0
},
)?;
let l1_bytes_len = out_l1_docs.len() * L1_SIZE + out_l1_bounds.len() * 4;
let total_bytes = stream_written as usize
+ out_l0.len()
+ out_headers.len()
+ l1_bytes_len
+ out_cursors.len()
+ out_ratios.as_ref().map_or(0, Vec::len)
+ out_impacts.as_ref().map_or(0, |t| t.bytes().len())
+ FOOTER_V2_SIZE;
Ok((total_docs, total_bytes))
}
pub fn decode_block_into(
&self,
block_idx: usize,
doc_ids: &mut Vec<u32>,
tfs: &mut Vec<u32>,
) -> bool {
if let Some((offset, tf_start, count)) = self.decode_block_doc_ids_only(block_idx, doc_ids)
{
self.decode_block_tfs_deferred(offset, tf_start, count, tfs);
true
} else {
false
}
}
pub fn decode_block_doc_ids_only(
&self,
block_idx: usize,
doc_ids: &mut Vec<u32>,
) -> Option<(usize, usize, usize)> {
match self.decode_block_doc_ids_checked(block_idx, doc_ids) {
Ok(state) => state,
Err(error) => {
log::error!(
"posting block {block_idx} of {} is corrupt; the cursor ends here: {error}",
self.l0_count
);
None
}
}
}
fn decode_block_doc_ids_checked(
&self,
block_idx: usize,
doc_ids: &mut Vec<u32>,
) -> io::Result<Option<(usize, usize, usize)>> {
if block_idx >= self.l0_count {
return Ok(None);
}
let decoded = (|| {
let (first, last, offset, _) = self.read_l0_entry(block_idx);
let header = self.block_header(block_idx);
let payload = self.block_payload(block_idx);
let state = self.decode_block_doc_ids_unchecked(
offset as usize,
block_idx,
header,
payload,
doc_ids,
)?;
if self.verify_content {
let byte_gaps = (header[6] == 8).then(|| &payload[..doc_ids.len() - 1]);
if header[6] >> PostingCodec::HEADER_SHIFT == PostingCodec::Simd4x as u8
&& doc_ids.len() == BLOCK_SIZE
&& !bitpacking4x::first_gap_is_zero(
payload,
header[6] & PostingCodec::WIDTH_MASK,
)
{
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"SIMD posting first gap must be zero",
));
}
let strict_gap_width = (header[6] >> PostingCodec::HEADER_SHIFT
== PostingCodec::Simd4x as u8)
.then_some(header[6] & PostingCodec::WIDTH_MASK);
if !verify_block_docs(doc_ids, first, last, byte_gaps, strict_gap_width) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!("decoded doc ids leave the directory range {first}..={last}"),
));
}
}
crate::observe::search_work!(
doc_blocks += 1,
doc_values += doc_ids.len(),
doc_payload_bytes += state.1 - if self.compact_headers { 0 } else { 8 }
);
Ok(Some(state))
})();
if decoded.is_err() {
doc_ids.clear();
if let Some(error) = &self.content_error {
error.record(block_idx);
}
}
decoded
}
fn decode_block_doc_ids_unchecked(
&self,
pos: usize,
block_idx: usize,
header: [u8; 8],
payload: &[u8],
doc_ids: &mut Vec<u32>,
) -> io::Result<(usize, usize, usize)> {
let invalid = |message: &str| io::Error::new(io::ErrorKind::InvalidData, message);
let count = u16::from_le_bytes(header[..2].try_into().unwrap()) as usize;
let first_doc = u32::from_le_bytes(header[2..6].try_into().unwrap());
let (codec, doc_width) = PostingCodec::from_header_byte(header[6])?;
let header_len = if self.compact_headers { 0 } else { 8 };
let state = if self.compact_headers { block_idx } else { pos };
if count == 0 || count > BLOCK_SIZE {
return Err(invalid("invalid posting block count"));
}
doc_ids.resize(count, 0);
doc_ids[0] = first_doc;
if codec == PostingCodec::Simd4x {
let values = if count == BLOCK_SIZE {
count
} else {
count - 1
};
let bytes = bitpacking4x::encoded_len(values, doc_width);
bitpacking4x::decode_docs(&payload[..bytes], doc_width, first_doc, doc_ids);
return Ok((state, header_len + bytes, count));
}
let deltas_bytes = if count > 1 {
match codec {
PostingCodec::Rounded => {
let rounded = simd::RoundedBitWidth::try_from_u8(doc_width)
.ok_or_else(|| invalid("invalid rounded posting width"))?;
let bytes = (count - 1) * rounded.bytes_per_value();
simd::unpack_rounded_raw_delta_decode(
&payload[..bytes],
rounded,
doc_ids,
first_doc,
count,
);
return Ok((state, header_len + bytes, count));
}
PostingCodec::Packed => {
let bytes = packed_bytes(count - 1, doc_width);
unpack_bits(&payload[..bytes], doc_width, &mut doc_ids[1..], count - 1);
bytes
}
PostingCodec::Pfor => {
let bytes = pfor_payload_len(payload, count - 1, doc_width)?;
unpack_pfor(&payload[..bytes], doc_width, &mut doc_ids[1..], count - 1)?;
bytes
}
PostingCodec::Simd4x => unreachable!("SIMD documents were decoded above"),
}
} else {
0
};
for i in 1..count {
doc_ids[i] = doc_ids[i].wrapping_add(doc_ids[i - 1]);
}
let tfs_start = header_len + deltas_bytes;
Ok((state, tfs_start, count))
}
pub fn decode_block_tfs_deferred(
&self,
block_offset: usize,
tf_start: usize,
count: usize,
tfs: &mut Vec<u32>,
) {
tfs.resize(count, 0);
self.decode_block_tfs_slice(block_offset, tf_start, tfs);
}
fn decode_block_tfs_slice(&self, block_offset: usize, tf_start: usize, tfs: &mut [u32]) {
let count = tfs.len();
let (header, block_data) = if self.compact_headers {
(
self.block_header(block_offset),
self.block_payload(block_offset),
)
} else {
(
self.stream[block_offset..block_offset + 8]
.try_into()
.unwrap(),
&self.stream[block_offset..],
)
};
let (codec, _) = PostingCodec::from_header_byte(header[6]).expect("admitted posting codec");
let tf_bits = header[7];
let payload = &block_data[tf_start..];
crate::observe::search_work!(
tf_blocks += 1,
tf_values += count,
tf_payload_bytes += match codec {
PostingCodec::Pfor =>
pfor_payload_len(payload, count, tf_bits).expect("admitted frequency payload"),
_ => packed_bytes(count, tf_bits),
}
);
match codec {
PostingCodec::Simd4x => {
bitpacking4x::decode(&payload[..packed_bytes(count, tf_bits)], tf_bits, tfs);
}
PostingCodec::Rounded => {
let rounded = simd::RoundedBitWidth::try_from_u8(tf_bits)
.expect("invalid rounded posting frequency width");
simd::unpack_rounded(
&payload[..count * rounded.bytes_per_value()],
rounded,
tfs,
count,
);
}
PostingCodec::Packed => {
unpack_bits(
&payload[..packed_bytes(count, tf_bits)],
tf_bits,
tfs,
count,
);
}
PostingCodec::Pfor => {
let len = pfor_payload_len(payload, count, tf_bits)
.expect("invalid patched posting frequency payload");
unpack_pfor(&payload[..len], tf_bits, tfs, count)
.expect("invalid patched posting frequency table");
}
}
}
#[inline]
fn block_len(&self, block_idx: usize) -> usize {
block_len_from_l0(&self.l0_bytes, self.l0_count, self.stream.len(), block_idx)
}
pub fn block_codec(&self, block_idx: usize) -> Option<PostingCodec> {
if block_idx >= self.l0_count {
return None;
}
PostingCodec::from_header_byte(self.block_header(block_idx)[6])
.ok()
.map(|(codec, _)| codec)
}
#[inline]
pub fn block_first_doc(&self, block_idx: usize) -> Option<DocId> {
if block_idx >= self.l0_count {
return None;
}
let (first_doc, _, _, _) = self.read_l0_entry(block_idx);
Some(first_doc)
}
#[inline]
pub fn block_last_doc(&self, block_idx: usize) -> Option<DocId> {
if block_idx >= self.l0_count {
return None;
}
let (_, last_doc, _, _) = self.read_l0_entry(block_idx);
Some(last_doc)
}
pub fn seek_block(&self, target: DocId, from_block: usize) -> Option<usize> {
if from_block >= self.l0_count {
return None;
}
if self
.block_last_doc(from_block)
.is_some_and(|last| last >= target)
{
return Some(from_block);
}
let from_l1 = from_block / L1_INTERVAL;
let groups = self.l1_docs.words().get(from_l1..)?;
let first = u32::from_le_bytes(*groups.first()?);
let offset = if first >= target {
0
} else {
let mut bound = 1usize;
while bound < groups.len() && u32::from_le_bytes(groups[bound]) < target {
bound = bound.saturating_mul(2);
}
let lo = bound / 2;
let hi = bound.saturating_add(1).min(groups.len());
lo + groups[lo..hi].partition_point(|&last| u32::from_le_bytes(last) < target)
};
let l1_idx = from_l1 + offset;
if l1_idx >= self.l1_docs.len() {
return None;
}
let start = (l1_idx * L1_INTERVAL).max(from_block + 1);
let end = ((l1_idx + 1) * L1_INTERVAL).min(self.l0_count);
let entries = self.l0_bytes.as_slice().as_chunks::<L0_SIZE>().0;
let within = entries[start..end].partition_point(|entry| {
u32::from_le_bytes([entry[4], entry[5], entry[6], entry[7]]) < target
});
let block_idx = start + within;
if block_idx < self.l0_count {
Some(block_idx)
} else {
None
}
}
pub fn iterator(&self) -> BlockPostingIterator<'_> {
BlockPostingIterator::new(self)
}
pub fn into_iterator(self) -> BlockPostingIterator<'static> {
BlockPostingIterator::owned(self)
}
pub(crate) fn into_candidate_iterator(
self,
first_target: DocId,
scratch: &mut PostingDecodeScratch,
) -> BlockPostingIterator<'static> {
let first_block = self.seek_block(first_target, 0);
let PostingDecodeScratch {
doc_ids,
term_freqs,
} = std::mem::take(scratch);
let mut block_tfs = term_freqs.unwrap_or_default();
block_tfs.take();
let mut iterator = BlockPostingIterator {
block_list: std::borrow::Cow::Owned(self),
current_block: first_block.unwrap_or(0),
block_doc_ids: doc_ids,
block_tfs,
tf_state: (0, 0, 0),
position_in_block: 0,
position_offsets: None,
position_offsets_ready: false,
block_position_cursor: 0,
exhausted: first_block.is_none(),
};
if let Some(block) = first_block {
iterator.load_block(block);
}
iterator
}
}
#[derive(Default)]
pub(crate) struct PostingDecodeScratch {
doc_ids: Vec<u32>,
term_freqs: Option<std::sync::OnceLock<[u32; BLOCK_SIZE]>>,
}
pub struct BlockPostingIterator<'a> {
block_list: std::borrow::Cow<'a, BlockPostingList>,
current_block: usize,
block_doc_ids: Vec<u32>,
block_tfs: std::sync::OnceLock<[u32; BLOCK_SIZE]>,
tf_state: (usize, usize, usize),
position_in_block: usize,
position_offsets: Option<Box<[u64; BLOCK_SIZE + 1]>>,
position_offsets_ready: bool,
block_position_cursor: u64,
exhausted: bool,
}
impl<'a> BlockPostingIterator<'a> {
pub(crate) fn recycle(self, scratch: &mut PostingDecodeScratch) {
*scratch = PostingDecodeScratch {
doc_ids: self.block_doc_ids,
term_freqs: Some(self.block_tfs),
};
}
fn new(block_list: &'a BlockPostingList) -> Self {
let exhausted = block_list.l0_count == 0;
let mut iter = Self {
block_list: std::borrow::Cow::Borrowed(block_list),
current_block: 0,
block_doc_ids: Vec::with_capacity(BLOCK_SIZE),
block_tfs: std::sync::OnceLock::new(),
tf_state: (0, 0, 0),
position_in_block: 0,
position_offsets: None,
position_offsets_ready: false,
block_position_cursor: 0,
exhausted,
};
if !iter.exhausted {
iter.load_block(0);
}
iter
}
fn owned(block_list: BlockPostingList) -> BlockPostingIterator<'static> {
let exhausted = block_list.l0_count == 0;
let mut iter = BlockPostingIterator {
block_list: std::borrow::Cow::Owned(block_list),
current_block: 0,
block_doc_ids: Vec::with_capacity(BLOCK_SIZE),
block_tfs: std::sync::OnceLock::new(),
tf_state: (0, 0, 0),
position_in_block: 0,
position_offsets: None,
position_offsets_ready: false,
block_position_cursor: 0,
exhausted,
};
if !iter.exhausted {
iter.load_block(0);
}
iter
}
fn load_block(&mut self, block_idx: usize) {
if block_idx >= self.block_list.l0_count {
self.exhausted = true;
return;
}
self.current_block = block_idx;
self.position_in_block = 0;
self.position_offsets_ready = false;
self.block_position_cursor = self.block_list.pos_cursor(block_idx).unwrap_or(0);
self.block_tfs.take();
match self
.block_list
.decode_block_doc_ids_checked(block_idx, &mut self.block_doc_ids)
{
Ok(Some(state)) => self.tf_state = state,
Ok(None) => unreachable!("block index was range-checked above"),
Err(error) => {
log::error!(
"posting block {block_idx} of {} is corrupt; the cursor ends here: {error}",
self.block_list.l0_count
);
self.block_doc_ids.clear();
self.tf_state = (0, 0, 0);
self.exhausted = true;
}
}
}
fn frequencies(&self) -> &[u32] {
let (offset, start, count) = self.tf_state;
if count == 0 {
return &[];
}
&self.block_tfs.get_or_init(|| {
let mut tfs = [0; BLOCK_SIZE];
self.block_list
.decode_block_tfs_slice(offset, start, &mut tfs[..count]);
tfs
})[..count]
}
#[inline]
pub fn position_cursor(&self) -> u64 {
if self.position_offsets_ready {
self.position_offsets.as_ref().unwrap()[self.position_in_block]
} else {
self.block_position_cursor
+ self.frequencies()[..self.position_in_block]
.iter()
.map(|&tf| u64::from(tf))
.sum::<u64>()
}
}
pub(crate) fn position_cursor_mut(&mut self) -> u64 {
if !self.position_offsets_ready {
self.initialize_position_offsets();
}
self.position_offsets.as_ref().unwrap()[self.position_in_block]
}
#[inline]
pub(crate) fn position_range(&mut self) -> (u64, u32) {
let cursor = self.position_cursor_mut();
(cursor, self.frequencies()[self.position_in_block])
}
fn initialize_position_offsets(&mut self) {
self.frequencies();
let offsets = self
.position_offsets
.get_or_insert_with(|| Box::new([0; BLOCK_SIZE + 1]));
let mut total = self.block_position_cursor;
offsets[0] = total;
if let Some(frequencies) = self.block_tfs.get() {
for (offset, &tf) in offsets[1..].iter_mut().zip(&frequencies[..self.tf_state.2]) {
total += u64::from(tf);
*offset = total;
}
}
self.position_offsets_ready = true;
}
pub fn doc(&self) -> DocId {
if self.exhausted {
TERMINATED
} else if self.position_in_block < self.block_doc_ids.len() {
self.block_doc_ids[self.position_in_block]
} else {
TERMINATED
}
}
pub fn term_freq(&self) -> u32 {
if self.exhausted || self.position_in_block >= self.block_doc_ids.len() {
0
} else {
self.frequencies()[self.position_in_block]
}
}
pub fn advance(&mut self) -> DocId {
if self.exhausted {
return TERMINATED;
}
self.position_in_block += 1;
if self.position_in_block >= self.block_doc_ids.len() {
self.load_block(self.current_block + 1);
}
self.doc()
}
#[inline]
pub fn seek(&mut self, target: DocId) -> DocId {
crate::observe::search_work!(posting_seeks += 1);
if self.exhausted {
return TERMINATED;
}
let current = self.block_doc_ids[self.position_in_block];
if target <= current {
return current;
}
if target > *self.block_doc_ids.last().unwrap() {
return self.seek_later_block(target);
}
let next = self.position_in_block + 1;
self.position_in_block = if self.block_doc_ids[next] >= target {
next
} else {
let remaining = &self.block_doc_ids[next + 1..];
let mut bound = 1;
while bound < remaining.len() && remaining[bound] < target {
bound *= 2;
}
let lo = bound / 2;
let hi = (bound + 1).min(remaining.len());
next + 1 + lo + remaining[lo..hi].partition_point(|&doc| doc < target)
};
self.block_doc_ids[self.position_in_block]
}
pub(crate) fn find_in_block(
&mut self,
mut find: impl FnMut(&[u32], &[u32]) -> Option<usize>,
) -> Option<DocId> {
if self.exhausted {
return Some(TERMINATED);
}
let start = self.position_in_block;
if let Some(offset) = find(&self.block_doc_ids[start..], &self.frequencies()[start..]) {
self.position_in_block += offset;
return Some(self.block_doc_ids[self.position_in_block]);
}
self.load_block(self.current_block + 1);
None
}
pub(crate) fn seek_physical(&mut self, target: DocId) -> DocId {
if !self.exhausted && target >= self.doc() {
return self.seek(target);
}
let Some(block) = self.block_list.seek_block(target, 0) else {
self.exhausted = true;
return TERMINATED;
};
self.exhausted = false;
if block != self.current_block || self.block_doc_ids.is_empty() {
self.load_block(block);
}
if self.exhausted {
return TERMINATED;
}
self.position_in_block = self.block_doc_ids.partition_point(|&doc| doc < target);
self.doc()
}
fn seek_later_block(&mut self, target: DocId) -> DocId {
let Some(block_idx) = self.block_list.seek_block(target, self.current_block + 1) else {
self.exhausted = true;
return TERMINATED;
};
self.load_block(block_idx);
if self.exhausted {
return TERMINATED;
}
self.position_in_block = self.block_doc_ids.partition_point(|&doc| doc < target);
self.block_doc_ids[self.position_in_block]
}
pub(crate) fn fill_doc_batch(&mut self, docs: &mut [DocId]) -> usize {
self.fill_batch::<false>(docs, &mut [])
}
pub(crate) fn fill_scored_doc_batch(&mut self, docs: &mut [DocId], tfs: &mut [u32]) -> usize {
assert!(tfs.len() >= docs.len());
self.fill_batch::<true>(docs, tfs)
}
fn fill_batch<const WITH_FREQUENCIES: bool>(
&mut self,
docs: &mut [DocId],
tfs: &mut [u32],
) -> usize {
assert!(docs.len() <= BLOCK_SIZE);
let mut count = 0;
while count < docs.len() && !self.exhausted {
let remaining = &self.block_doc_ids[self.position_in_block..];
let take = remaining.len().min(docs.len() - count);
docs[count..count + take].copy_from_slice(&remaining[..take]);
if WITH_FREQUENCIES {
tfs[count..count + take].copy_from_slice(
&self.frequencies()[self.position_in_block..self.position_in_block + take],
);
}
count += take;
self.position_in_block += take;
if self.position_in_block == self.block_doc_ids.len() {
self.load_block(self.current_block + 1);
}
}
count
}
pub(crate) fn retain_doc_batch(&mut self, docs: &mut [DocId]) -> usize {
self.retain_batch::<false>(docs, &mut [])
}
pub(crate) fn retain_scored_doc_batch(&mut self, docs: &mut [DocId], tfs: &mut [u32]) -> usize {
assert!(tfs.len() >= docs.len());
self.retain_batch::<true>(docs, tfs)
}
fn retain_batch<const WITH_FREQUENCIES: bool>(
&mut self,
docs: &mut [DocId],
tfs: &mut [u32],
) -> usize {
assert!(docs.len() <= BLOCK_SIZE);
let mut input = 0;
let mut kept = 0;
while input < docs.len() {
if self.seek(docs[input]) == TERMINATED {
break;
}
let last = *self.block_doc_ids.last().unwrap();
while input < docs.len() && docs[input] <= last {
let doc = docs[input];
self.position_in_block = simd::find_first_ge_block_from(
&self.block_doc_ids,
self.position_in_block,
doc,
);
if self.block_doc_ids[self.position_in_block] == doc {
docs[kept] = doc;
if WITH_FREQUENCIES {
tfs[kept] = self.frequencies()[self.position_in_block];
}
kept += 1;
}
input += 1;
}
}
kept
}
pub(crate) fn fill_doc_window(&mut self, base: DocId, bits: &mut [u64]) {
let span = u32::try_from(bits.len()).unwrap().checked_mul(64).unwrap();
let end = base.saturating_add(span);
bits.fill(0);
self.seek(base);
let list = self.block_list.as_ref();
let dense = list.doc_count() >= 16
&& list
.block_first_doc(0)
.zip(list.block_last_doc(list.num_blocks().saturating_sub(1)))
.is_some_and(|(first, last)| {
u64::from(last)
.checked_sub(u64::from(first))
.is_some_and(|span| span < u64::from(list.doc_count()) * 2)
});
if dense {
self.fill_doc_words::<true>(base, end, bits);
} else {
self.fill_doc_words::<false>(base, end, bits);
}
}
fn fill_doc_words<const GROUPED: bool>(&mut self, base: DocId, end: DocId, bits: &mut [u64]) {
self.visit_until::<false>(end, |docs, _| {
if !GROUPED {
for &doc in docs {
let offset = (doc - base) as usize;
bits[offset / 64] |= 1u64 << (offset % 64);
}
return true;
}
let mut index = 0;
while index < docs.len() {
let offset = (docs[index] - base) as usize;
let word_index = offset / 64;
let mut mask = 1u64 << (offset % 64);
index += 1;
while index < docs.len() {
let offset = (docs[index] - base) as usize;
if offset / 64 != word_index {
break;
}
mask |= 1u64 << (offset % 64);
index += 1;
}
bits[word_index] |= mask;
}
true
});
}
pub(crate) fn visit_postings_until(
&mut self,
end: DocId,
mut visit: impl FnMut(&[u32], &[u32]) -> bool,
) {
self.visit_until::<true>(end, |docs, tfs| visit(docs, tfs.unwrap()));
}
pub(crate) fn visit_doc_ids_until(
&mut self,
end: DocId,
mut visit: impl FnMut(&[u32]) -> bool,
) {
self.visit_until::<false>(end, |docs, _| visit(docs));
}
fn visit_until<const WITH_FREQUENCIES: bool>(
&mut self,
end: DocId,
mut visit: impl FnMut(&[u32], Option<&[u32]>) -> bool,
) {
while self.doc() < end {
let start = self.position_in_block;
let count = self.block_doc_ids[start..].partition_point(|&doc| doc < end);
let tfs = WITH_FREQUENCIES.then(|| &self.frequencies()[start..start + count]);
if !visit(&self.block_doc_ids[start..start + count], tfs) {
return;
}
self.position_in_block += count;
if self.position_in_block == self.block_doc_ids.len() {
self.load_block(self.current_block + 1);
}
}
}
pub fn skip_to_next_block(&mut self) -> DocId {
if self.exhausted {
return TERMINATED;
}
self.load_block(self.current_block + 1);
self.doc()
}
#[inline]
pub fn current_block_idx(&self) -> usize {
self.current_block
}
#[inline]
pub fn num_blocks(&self) -> usize {
self.block_list.l0_count
}
pub(crate) fn current_block_metadata(&self) -> Option<(&BlockPostingList, usize)> {
(!self.exhausted).then_some((&self.block_list, self.current_block))
}
#[inline]
pub fn current_block_max_tf(&self) -> u32 {
if self.exhausted || self.current_block >= self.block_list.l0_count {
0
} else {
self.block_list
.block_max_tf(self.current_block)
.unwrap_or(0)
}
}
}
pub(crate) struct PostingIntersection {
seek_driven: bool,
left_is_rare: bool,
blocks: (usize, usize),
pairs: [(u8, u8); BLOCK_SIZE],
next: usize,
count: usize,
ends: (usize, usize),
}
impl Default for PostingIntersection {
fn default() -> Self {
Self {
seek_driven: false,
left_is_rare: true,
blocks: (usize::MAX, usize::MAX),
pairs: [(0, 0); BLOCK_SIZE],
next: 0,
count: 0,
ends: (0, 0),
}
}
}
impl PostingIntersection {
pub(crate) fn with_costs(left: u32, right: u32) -> Self {
Self {
seek_driven: left.min(right).saturating_mul(4) < left.max(right),
left_is_rare: left <= right,
..Self::default()
}
}
pub(crate) fn reset(&mut self) {
self.blocks = (usize::MAX, usize::MAX);
}
pub(crate) fn intersect_block(
&mut self,
left: &mut BlockPostingIterator<'_>,
right: &mut BlockPostingIterator<'_>,
) -> Option<DocId> {
if left.exhausted || right.exhausted {
return Some(TERMINATED);
}
if left.doc() == right.doc() {
return Some(left.doc());
}
if self.seek_driven {
fn align(
lead: &mut BlockPostingIterator<'_>,
other: &mut BlockPostingIterator<'_>,
) -> Option<DocId> {
let candidate = lead.doc();
let next = other.seek(candidate);
if next == candidate {
return Some(candidate);
}
lead.seek(next);
None
}
return if self.left_is_rare {
align(left, right)
} else {
align(right, left)
};
}
if *left.block_doc_ids.last().unwrap() < right.doc() {
left.seek_later_block(right.doc());
return None;
}
if *right.block_doc_ids.last().unwrap() < left.doc() {
right.seek_later_block(left.doc());
return None;
}
let blocks = (left.current_block, right.current_block);
if self.blocks != blocks {
let mut a = left.position_in_block;
let mut b = right.position_in_block;
self.count = simd::intersect_posting_blocks(
&left.block_doc_ids,
&mut a,
&right.block_doc_ids,
&mut b,
&mut self.pairs,
);
self.blocks = blocks;
self.ends = (a, b);
self.next = 0;
}
while self.next < self.count {
let (a, b) = self.pairs[self.next];
self.next += 1;
let (a, b) = (usize::from(a), usize::from(b));
if a >= left.position_in_block && b >= right.position_in_block {
left.position_in_block = a;
right.position_in_block = b;
return Some(left.doc());
}
}
left.position_in_block = left.position_in_block.max(self.ends.0);
right.position_in_block = right.position_in_block.max(self.ends.1);
if left.position_in_block == left.block_doc_ids.len() {
left.load_block(left.current_block + 1);
}
if right.position_in_block == right.block_doc_ids.len() {
right.load_block(right.current_block + 1);
}
None
}
}
#[cfg(test)]
mod compact_layout_tests {
use super::*;
#[test]
fn batched_intersection_preserves_monotone_seeks_frequencies_and_position_offsets() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let make = |divisor| {
let mut postings = PostingList::new();
for doc in 0..3001 {
if doc % divisor != 1 {
postings.push(doc, 1 + doc % 7);
}
}
BlockPostingList::from_posting_list_with_options(&postings, true, None, codec)
.unwrap()
};
let a = make(3);
let b = make(5);
for (seek_driven, left_is_rare) in [(false, true), (true, true), (true, false)] {
let mut left = a.iterator();
let mut right = b.iterator();
let mut reference_left = a.iterator();
let mut reference_right = b.iterator();
let mut intersection = PostingIntersection {
seek_driven,
left_is_rare,
..Default::default()
};
let mut target = 0;
loop {
left.seek(target);
right.seek(target);
let doc = loop {
if let Some(doc) = intersection.intersect_block(&mut left, &mut right) {
break doc;
}
};
let expected = (target..3001)
.find(|doc| doc % 3 != 1 && doc % 5 != 1)
.unwrap_or(TERMINATED);
assert_eq!(doc, expected);
assert_eq!(
intersection.intersect_block(&mut left, &mut right),
Some(doc)
);
if doc == TERMINATED {
break;
}
reference_left.seek(doc);
reference_right.seek(doc);
assert_eq!(left.term_freq(), reference_left.term_freq());
assert_eq!(right.term_freq(), reference_right.term_freq());
assert_eq!(left.position_cursor(), reference_left.position_cursor());
assert_eq!(right.position_cursor(), reference_right.position_cursor());
target = doc + if doc % 7 == 0 { 19 } else { 1 };
}
for target in [0, 200, 129, 3, 2048, 3000, 2] {
intersection.reset();
left.seek_physical(target);
right.seek_physical(target);
let doc = loop {
if let Some(doc) = intersection.intersect_block(&mut left, &mut right) {
break doc;
}
};
let expected = (target..3001)
.find(|doc| doc % 3 != 1 && doc % 5 != 1)
.unwrap_or(TERMINATED);
assert_eq!(doc, expected);
reference_left.seek_physical(doc);
reference_right.seek_physical(doc);
assert_eq!(left.position_cursor(), reference_left.position_cursor());
assert_eq!(right.position_cursor(), reference_right.position_cursor());
}
}
}
}
#[test]
fn compact_postings_preserve_payload_scores_and_copy_merges() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Simd4x,
PostingCodec::Pfor,
] {
for count in [1, 2, 127, 128, 129, 1025] {
let mut postings = PostingList::new();
for doc in 0..count {
postings.push(doc * 3, doc % 13);
}
let list = BlockPostingList::from_posting_list_with_ratio_bounds(
&postings,
true,
Some(&|doc| doc % 100 + 1),
codec,
)
.unwrap();
let mut old = Vec::new();
list.serialize(&mut old).unwrap();
let mut bytes = Vec::new();
list.serialize_compact(&mut bytes).unwrap();
let compact = BlockPostingList::deserialize(&bytes).unwrap();
assert_eq!(compact.compact_headers, codec != PostingCodec::Pfor);
if codec != PostingCodec::Pfor {
assert_eq!(old.len() - bytes.len(), list.num_blocks() * 8);
}
let mut a = Vec::new();
let mut b = Vec::new();
for i in 0..list.num_blocks() {
assert_eq!(list.block_payload(i), compact.block_payload(i));
assert_eq!(list.pos_cursor(i), compact.pos_cursor(i));
assert!(compact.decode_block_into(i, &mut a, &mut b));
assert_eq!(
a,
postings.postings
[i * BLOCK_SIZE..(i * BLOCK_SIZE + BLOCK_SIZE).min(count as usize)]
.iter()
.map(|p| p.doc_id)
.collect::<Vec<_>>()
);
assert_eq!(b, a.iter().map(|doc| (doc / 3) % 13).collect::<Vec<_>>());
}
let mut repeated = Vec::new();
compact.serialize(&mut repeated).unwrap();
assert_eq!(bytes, repeated);
for second in [&old, &bytes] {
let mut merged = Vec::new();
let (docs, len) = BlockPostingList::concatenate_streaming(
&[(&bytes, 0), (second, count * 3)],
&mut merged,
)
.unwrap();
assert_eq!(len, merged.len());
assert_eq!(docs, count * 2);
let merged = BlockPostingList::deserialize(&merged).unwrap();
for block in 0..merged.num_blocks() {
assert_eq!(
merged.block_payload(block),
list.block_payload(block % list.num_blocks())
);
assert!(merged.decode_block_into(block, &mut a, &mut b));
}
}
}
}
}
#[test]
fn compact_posting_metadata_rejects_bad_descriptors_and_content_checks_remain_observable() {
let mut postings = PostingList::new();
for doc in 0..257 {
postings.push(doc * 2, 1);
}
let list = BlockPostingList::from_posting_list_with_options(
&postings,
true,
None,
PostingCodec::Rounded,
)
.unwrap();
let mut bytes = Vec::new();
list.serialize_compact(&mut bytes).unwrap();
let footer = Footer::parse(&bytes).unwrap();
for (at, value) in [
(footer.l0_end(), 0),
(footer.l0_end() + 1, 1),
(footer.l0_end() + 2, 33),
(footer.l0_end() + 3, 33),
(footer.l1_bounds_end(), 1),
] {
let mut corrupt = bytes.clone();
corrupt[at] = value;
assert!(
BlockPostingList::deserialize(&corrupt).is_err(),
"byte {at}"
);
}
let mut corrupt = bytes.clone();
corrupt[0] = 0;
let compact = BlockPostingList::deserialize(&corrupt).unwrap();
let mut docs = Vec::new();
assert!(compact.decode_block_doc_ids_checked(0, &mut docs).is_err());
assert!(docs.is_empty());
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn block_decoding_overwrites_stale_values_across_lengths_and_codecs() {
let mut docs = vec![u32::MAX; BLOCK_SIZE * 2];
let mut tfs = docs.clone();
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
for count in [128, 1, 127, 128, 17, 256, 257] {
for freq in [0, 1, 255, 65536] {
let mut postings = PostingList::new();
for i in 0..count {
postings.push(i * 3 + 7, freq);
}
let list = BlockPostingList::from_posting_list_with_options(
&postings, true, None, codec,
)
.unwrap();
for compact in [false, true] {
let mut bytes = Vec::new();
if compact {
list.serialize_compact(&mut bytes).unwrap();
} else {
list.serialize(&mut bytes).unwrap();
}
let list = BlockPostingList::deserialize(&bytes).unwrap();
for block in (0..list.num_blocks()).rev().chain(0..list.num_blocks()) {
docs.fill(u32::MAX);
tfs.fill(u32::MAX);
assert!(list.decode_block_into(block, &mut docs, &mut tfs));
let expected: Vec<_> = postings.postings[block * BLOCK_SIZE
..postings.postings.len().min((block + 1) * BLOCK_SIZE)]
.iter()
.map(|p| p.doc_id)
.collect();
assert_eq!(docs, expected);
assert_eq!(tfs, vec![freq; expected.len()]);
}
}
}
}
}
}
#[test]
fn compact_membership_batches_preserve_resume_frequencies_and_position_prefixes() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let mut postings = PostingList::new();
let mut expected = Vec::new();
let mut prefix = 0u64;
for i in 0..701u32 {
let tf = i % 19 + 1;
let doc = i * 13 + 5;
postings.push(doc, tf);
expected.push((doc, tf, prefix));
prefix += u64::from(tf);
}
let list =
BlockPostingList::from_posting_list_with_options(&postings, true, None, codec)
.unwrap();
let bytes = serialize_bpl(&list);
let mut cursor = list.iterator();
let mut docs = [0; BLOCK_SIZE];
let mut consumed = 0;
while cursor.doc() != TERMINATED {
let count = cursor.fill_doc_batch(&mut docs);
assert_eq!(
&docs[..count],
&expected[consumed..consumed + count]
.iter()
.map(|e| e.0)
.collect::<Vec<_>>()
);
consumed += count;
if consumed < expected.len() {
assert!(cursor.block_tfs.get().is_none(), "copy initialized TFs");
assert_eq!(cursor.doc(), expected[consumed].0);
assert_eq!(cursor.term_freq(), expected[consumed].1);
assert_eq!(cursor.position_cursor_mut(), expected[consumed].2);
}
}
assert_eq!(consumed, expected.len());
assert_eq!(cursor.fill_doc_batch(&mut docs), 0);
let mut cursor = list.iterator();
for start in (0..10_000u32).step_by(BLOCK_SIZE) {
let mut candidates: Vec<_> = (start..start + BLOCK_SIZE as u32).collect();
let wanted: Vec<_> = candidates
.iter()
.copied()
.filter(|doc| expected.iter().any(|e| e.0 == *doc))
.collect();
let kept = cursor.retain_doc_batch(&mut candidates);
assert_eq!(&candidates[..kept], wanted, "{codec:?} start={start}");
if cursor.doc() != TERMINATED {
let entry = expected.iter().find(|e| e.0 == cursor.doc()).unwrap();
assert_eq!(cursor.term_freq(), entry.1);
assert_eq!(cursor.position_cursor_mut(), entry.2);
}
}
assert_eq!(cursor.doc(), TERMINATED);
assert_eq!(serialize_bpl(&list), bytes);
}
}
#[test]
fn membership_words_preserve_unaligned_dense_sparse_windows_and_resume() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
for stride in [1, 2, 3, 67, 129] {
let docs: Vec<u32> = (0..10_000).step_by(stride).collect();
let mut postings = PostingList::new();
for &doc in &docs {
postings.push(doc, 1 + doc % 7);
}
let list =
BlockPostingList::from_posting_list_with_options(&postings, true, None, codec)
.unwrap();
for start in [0, 1, 17, 63, 64, 127, 511] {
let mut cursor = list.iterator();
for base in [start, start + 4096, start + 8192] {
let mut bits = [u64::MAX; 64];
cursor.fill_doc_window(base, &mut bits);
let mut expected = [0u64; 64];
for &doc in &docs {
if (base..base + 4096).contains(&doc) {
let offset = (doc - base) as usize;
expected[offset / 64] |= 1u64 << (offset % 64);
}
}
assert_eq!(bits, expected, "{codec:?}, stride={stride}, base={base}");
let next = docs
.iter()
.copied()
.find(|&doc| doc >= base + 4096)
.unwrap_or(TERMINATED);
assert_eq!(cursor.doc(), next);
if next != TERMINATED {
assert_eq!(cursor.term_freq(), 1 + next % 7);
}
}
}
}
}
}
#[test]
fn document_only_navigation_defers_frequencies_and_resumes_exact_position_reads() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let mut postings = PostingList::new();
let mut expected = Vec::new();
let mut prefix = 0u64;
for i in 0..701u32 {
let tf = if i % 37 == 0 { u32::MAX } else { i % 23 + 1 };
let doc = i * 13 + 5;
postings.push(doc, tf);
expected.push((doc, tf, prefix));
prefix += u64::from(tf);
}
let list =
BlockPostingList::from_posting_list_with_options(&postings, true, None, codec)
.unwrap();
let bytes = serialize_bpl(&list);
for owned in [false, true] {
let mut cursor = if owned {
list.clone().into_iterator()
} else {
list.iterator()
};
assert!(
cursor.block_tfs.get().is_none(),
"{codec:?}: ID-only open decoded frequencies"
);
for target in [20, 205, 1800, 3900] {
let entry = expected.iter().find(|e| e.0 >= target).unwrap();
assert_eq!(cursor.seek(target), entry.0);
assert!(
cursor.block_tfs.get().is_none(),
"ID-only seek decoded frequencies"
);
}
let entry = expected.iter().find(|e| e.0 == cursor.doc()).unwrap();
#[cfg(feature = "native")]
std::thread::scope(|scope| {
for _ in 0..4 {
let shared = &cursor;
scope.spawn(move || {
assert_eq!(shared.term_freq(), entry.1);
assert_eq!(shared.position_cursor(), entry.2);
});
}
});
assert_eq!(cursor.position_cursor(), entry.2);
assert_eq!(cursor.term_freq(), entry.1);
assert_eq!(cursor.position_cursor_mut(), entry.2);
let mut bits = [0u64; 64];
cursor.fill_doc_window(4096, &mut bits);
for &(doc, _, _) in &expected {
if (4096..8192).contains(&doc) {
assert_ne!(bits[(doc as usize - 4096) / 64] & (1 << (doc % 64)), 0);
}
}
assert_eq!(
bits.iter().map(|word| word.count_ones()).sum::<u32>(),
expected
.iter()
.filter(|e| (4096..8192).contains(&e.0))
.count() as u32
);
assert!(
cursor.block_tfs.get().is_none(),
"membership windows decoded frequencies"
);
let entry = expected.iter().find(|e| e.0 == cursor.doc()).unwrap();
assert_eq!(cursor.position_cursor_mut(), entry.2);
assert_eq!(cursor.term_freq(), entry.1);
let parked = cursor.doc();
cursor.visit_postings_until(TERMINATED, |docs, tfs| {
for (&doc, &tf) in docs.iter().zip(tfs) {
assert_eq!(tf, expected.iter().find(|e| e.0 == doc).unwrap().1);
}
false
});
assert_eq!(cursor.doc(), parked);
assert_eq!(cursor.position_cursor(), entry.2);
let mut seen = Vec::new();
cursor.visit_postings_until(TERMINATED, |docs, tfs| {
seen.extend(docs.iter().copied().zip(tfs.iter().copied()));
true
});
assert_eq!(
seen,
expected
.iter()
.filter(|e| e.0 >= parked)
.map(|e| (e.0, e.1))
.collect::<Vec<_>>()
);
assert_eq!(cursor.doc(), TERMINATED);
assert_eq!(cursor.term_freq(), 0);
assert_eq!(cursor.seek(0), TERMINATED);
}
assert_eq!(serialize_bpl(&list), bytes);
}
}
#[test]
fn document_navigation_defers_position_accounting_without_changing_cursors() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let mut postings = PostingList::new();
let mut expected = Vec::new();
let mut total = 0u64;
for doc in 0..701u32 {
expected.push(total);
let tf = doc % 23 + 1;
postings.push(doc * 7, tf);
total += u64::from(tf);
}
let list =
BlockPostingList::from_posting_list_with_options(&postings, true, None, codec)
.unwrap();
let mut cursor = list.iterator();
for target in [7, 70, 777, 896, 1400, 3500, 4900] {
assert_eq!(cursor.seek(target), target);
assert!(
cursor.position_offsets.is_none(),
"navigation must not prepare unused position offsets"
);
assert_eq!(cursor.position_cursor(), expected[target as usize / 7]);
assert_eq!(cursor.term_freq(), target / 7 % 23 + 1);
assert_eq!(cursor.seek(target - 1), target);
}
assert_eq!(cursor.advance(), TERMINATED);
}
}
#[test]
fn deferred_position_accounting_resumes_after_reads_windows_and_stopped_runs() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let docs: Vec<_> = (0..600u32).map(|doc| (doc * 11, u32::MAX - doc)).collect();
let prefixes = expected_cursors(&docs);
let mut input = PostingList::new();
for &(doc, tf) in &docs {
input.push(doc, tf);
}
let list = BlockPostingList::build(&input, true, None, codec, false, false).unwrap();
let before = serialize_bpl(&list);
let mut cursor = list.iterator();
for at in [13, 25, 127, 128, 199] {
cursor.seek(docs[at].0);
assert_eq!(cursor.position_range(), (prefixes[at], docs[at].1));
assert_eq!(cursor.position_range(), (prefixes[at], docs[at].1));
assert_eq!(cursor.position_cursor_mut(), prefixes[at]);
assert_eq!(cursor.position_cursor_mut(), prefixes[at]);
assert_eq!(cursor.position_cursor(), prefixes[at]);
cursor.seek(docs[at].0 - 1);
assert_eq!(cursor.position_cursor_mut(), prefixes[at]);
}
cursor.visit_postings_until(docs[220].0, |_, _| true);
assert_eq!(cursor.position_cursor_mut(), prefixes[220]);
cursor.visit_postings_until(docs[250].0, |_, _| false);
assert_eq!(cursor.position_cursor_mut(), prefixes[220]);
let mut bits = [0u64; 4];
cursor.fill_doc_window(docs[220].0, &mut bits);
let at = docs.partition_point(|&(doc, _)| doc < docs[220].0 + 256);
assert_eq!(cursor.position_cursor_mut(), prefixes[at]);
cursor.skip_to_next_block();
assert_eq!(cursor.position_cursor_mut(), prefixes[256]);
assert_eq!(serialize_bpl(&list), before);
}
}
#[test]
fn posting_runs_preserve_frequencies_positions_and_resumption_after_stop() {
let docs: Vec<_> = (0..1027u32).map(|i| (i * 7, i % 31 + 1)).collect();
let prefixes = expected_cursors(&docs);
let mut list = PostingList::new();
for &(doc, tf) in &docs {
list.push(doc, tf);
}
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let postings = BlockPostingList::build(&list, true, None, codec, false, false).unwrap();
let bytes = serialize_bpl(&postings);
let mut cursor = postings.iterator();
cursor.seek(docs[17].0);
let mut actual = Vec::new();
let mut calls = 0;
cursor.visit_postings_until(TERMINATED, |ids, tfs| {
calls += 1;
assert_eq!(ids.len(), tfs.len());
assert!(ids.len() <= BLOCK_SIZE);
if calls == 3 {
return false;
}
actual.extend(ids.iter().copied().zip(tfs.iter().copied()));
true
});
assert_eq!(actual, docs[17..256]);
assert_eq!(cursor.doc(), docs[256].0);
assert_eq!(cursor.position_cursor(), prefixes[256]);
cursor.visit_postings_until(docs[331].0, |ids, tfs| {
actual.extend(ids.iter().copied().zip(tfs.iter().copied()));
true
});
assert_eq!(actual, docs[17..331]);
assert_eq!(cursor.doc(), docs[331].0);
assert_eq!(cursor.term_freq(), docs[331].1);
assert_eq!(cursor.position_cursor(), prefixes[331]);
cursor.visit_postings_until(TERMINATED, |ids, tfs| {
actual.extend(ids.iter().copied().zip(tfs.iter().copied()));
true
});
assert_eq!(actual, docs[17..]);
cursor.visit_postings_until(TERMINATED, |_, _| panic!("already exhausted"));
assert_eq!(cursor.doc(), TERMINATED);
assert_eq!(serialize_bpl(&postings), bytes);
}
}
#[test]
fn document_windows_preserve_posting_frequencies_positions_and_bytes() {
for near_end in [false, true] {
let base = if near_end { TERMINATED - 30_000 } else { 0 };
let docs: Vec<_> = (0..5000u32).map(|i| (base + i * 5, i % 31 + 1)).collect();
let prefixes = expected_cursors(&docs);
let mut list = PostingList::new();
for &(doc, tf) in &docs {
list.push(doc, tf);
}
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let postings =
BlockPostingList::build(&list, true, None, codec, false, false).unwrap();
let bytes = serialize_bpl(&postings);
let mut cursor = postings.iterator();
let mut actual = Vec::new();
while cursor.doc() != TERMINATED {
let base = cursor.doc();
let mut bits = [u64::MAX; 64];
cursor.fill_doc_window(base, &mut bits);
for (index, word) in bits.into_iter().enumerate() {
for bit in 0..64 {
if word & (1 << bit) != 0 {
actual.push(base + index as u32 * 64 + bit);
}
}
}
let at = docs.partition_point(|&(doc, _)| doc < base.saturating_add(4096));
assert_eq!(cursor.doc(), docs.get(at).map_or(TERMINATED, |p| p.0));
if at < docs.len() {
assert_eq!(cursor.term_freq(), docs[at].1);
assert_eq!(cursor.position_cursor(), prefixes[at]);
}
}
assert_eq!(actual, docs.iter().map(|p| p.0).collect::<Vec<_>>());
assert_eq!(serialize_bpl(&postings), bytes);
}
}
}
#[test]
fn nearby_and_distant_seeks_preserve_postings_position_cursors_and_bytes() {
let docs: Vec<_> = (0..32_769u32)
.map(|i| (i * 5 + i % 3, i % 31 + 1))
.collect();
let prefixes = expected_cursors(&docs);
let mut list = PostingList::new();
for &(doc, tf) in &docs {
list.push(doc, tf);
}
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let postings = BlockPostingList::build(&list, true, None, codec, false, false).unwrap();
let bytes = serialize_bpl(&postings);
let lasts: Vec<_> = (0..postings.num_blocks())
.map(|i| postings.block_last_doc(i).unwrap())
.collect();
for from in [0, 1, 7, 8, 64, 200, 256, 257] {
for target in [0, 1, 100, 639, 640, 641, 700, 160_000, 163_840, TERMINATED] {
let expected = (from..lasts.len()).find(|&i| lasts[i] >= target);
assert_eq!(
postings.seek_block(target, from),
expected,
"{codec:?} from={from} target={target}"
);
}
}
let mut cursor = postings.iterator();
let mut at = 0;
for target in [
0,
1,
20,
19,
639,
640,
641,
700,
160_000,
161_000,
160_000,
docs.last().unwrap().0 - 1,
docs.last().unwrap().0,
TERMINATED,
0,
] {
while at < docs.len() && docs[at].0 < target {
at += 1;
}
let expected = docs.get(at).map_or(TERMINATED, |&(doc, _)| doc);
assert_eq!(cursor.seek(target), expected, "{codec:?} target={target}");
if at < docs.len() {
assert_eq!(cursor.term_freq(), docs[at].1);
assert_eq!(cursor.position_cursor(), prefixes[at]);
}
}
assert_eq!(serialize_bpl(&postings), bytes);
}
}
#[test]
fn candidate_posting_probes_reuse_buffers_and_preserve_seek_and_position_cursors() {
let mut list = PostingList::new();
for i in 0..700 {
list.push(i * 3, i % 7 + 1);
}
let postings = BlockPostingList::from_posting_list(&list).unwrap();
let mut scratch = PostingDecodeScratch::default();
let mut doc_buffer = None;
for first in [0, 385, 900, 1800, 2100, 100, 2097] {
let mut reference = postings.clone().into_iterator();
let mut selected = postings
.clone()
.into_candidate_iterator(first, &mut scratch);
assert_eq!(
selected.current_block_idx(),
postings.seek_block(first, 0).unwrap_or(0)
);
for target in (first..2200).step_by(17) {
assert_eq!(selected.seek(target), reference.seek(target));
assert_eq!(selected.term_freq(), reference.term_freq());
if selected.doc() != TERMINATED {
assert_eq!(selected.position_cursor(), reference.position_cursor());
}
}
selected.recycle(&mut scratch);
assert!(scratch.doc_ids.capacity() >= BLOCK_SIZE);
assert!(scratch.term_freqs.is_some());
let address = scratch.doc_ids.as_ptr() as usize;
if let Some(previous) = doc_buffer {
assert_eq!(
address, previous,
"document scratch allocation must be reused"
);
}
doc_buffer = Some(address);
}
}
#[test]
fn test_posting_list_basic() {
let mut list = PostingList::new();
list.push(1, 2);
list.push(5, 1);
list.push(10, 3);
assert_eq!(list.len(), 3);
let mut iter = PostingListIterator::new(&list);
assert_eq!(iter.doc(), 1);
assert_eq!(iter.term_freq(), 2);
assert_eq!(iter.advance(), 5);
assert_eq!(iter.term_freq(), 1);
assert_eq!(iter.advance(), 10);
assert_eq!(iter.term_freq(), 3);
assert_eq!(iter.advance(), TERMINATED);
}
#[test]
fn test_posting_list_seek() {
let mut list = PostingList::new();
for i in 0..100 {
list.push(i * 2, 1);
}
let mut iter = PostingListIterator::new(&list);
assert_eq!(iter.seek(50), 50);
assert_eq!(iter.seek(51), 52);
assert_eq!(iter.seek(200), TERMINATED);
}
#[test]
fn test_block_posting_list() {
let mut list = PostingList::new();
for i in 0..500 {
list.push(i * 2, (i % 10) + 1);
}
let block_list = BlockPostingList::from_posting_list(&list).unwrap();
assert_eq!(block_list.doc_count(), 500);
let mut iter = block_list.iterator();
assert_eq!(iter.doc(), 0);
assert_eq!(iter.term_freq(), 1);
assert_eq!(iter.seek(500), 500);
assert_eq!(iter.seek(998), 998);
assert_eq!(iter.seek(1000), TERMINATED);
}
#[test]
fn test_block_posting_list_serialization() {
let mut list = PostingList::new();
for i in 0..300 {
list.push(i * 3, i + 1);
}
let block_list = BlockPostingList::from_posting_list(&list).unwrap();
let mut buffer = Vec::new();
block_list.serialize(&mut buffer).unwrap();
let deserialized = BlockPostingList::deserialize(&buffer[..]).unwrap();
assert_eq!(deserialized.doc_count(), block_list.doc_count());
let mut iter1 = block_list.iterator();
let mut iter2 = deserialized.iterator();
while iter1.doc() != TERMINATED {
assert_eq!(iter1.doc(), iter2.doc());
assert_eq!(iter1.term_freq(), iter2.term_freq());
iter1.advance();
iter2.advance();
}
assert_eq!(iter2.doc(), TERMINATED);
}
fn collect_postings(bpl: &BlockPostingList) -> Vec<(u32, u32)> {
let mut result = Vec::new();
let mut it = bpl.iterator();
while it.doc() != TERMINATED {
result.push((it.doc(), it.term_freq()));
it.advance();
}
result
}
#[test]
fn deserialization_rejects_unsupported_width_instead_of_empty_results() {
let list = build_bpl(&[(0, 1), (1, 2)]);
let mut bytes = Vec::new();
list.serialize(&mut bytes).unwrap();
bytes[6] = 0xe1;
let error = BlockPostingList::deserialize(&bytes).unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::InvalidData);
assert!(error.to_string().contains("exceeds 32 bits"));
}
fn build_bpl(postings: &[(u32, u32)]) -> BlockPostingList {
let mut pl = PostingList::new();
for &(doc_id, tf) in postings {
pl.push(doc_id, tf);
}
BlockPostingList::from_posting_list(&pl).unwrap()
}
fn serialize_bpl(bpl: &BlockPostingList) -> Vec<u8> {
let mut buf = Vec::new();
bpl.serialize(&mut buf).unwrap();
buf
}
#[test]
fn test_concatenate_blocks_two_segments() {
let a: Vec<(u32, u32)> = (0..100).map(|i| (i * 2, i + 1)).collect();
let bpl_a = build_bpl(&a);
let b: Vec<(u32, u32)> = (0..100).map(|i| (i * 3, i + 2)).collect();
let bpl_b = build_bpl(&b);
let merged =
BlockPostingList::concatenate_blocks(&[(bpl_a.clone(), 0), (bpl_b.clone(), 200)])
.unwrap();
assert_eq!(merged.doc_count(), 200);
let postings = collect_postings(&merged);
assert_eq!(postings.len(), 200);
for (i, p) in postings.iter().enumerate().take(100) {
assert_eq!(*p, (i as u32 * 2, i as u32 + 1));
}
for i in 0..100 {
assert_eq!(postings[100 + i], (i as u32 * 3 + 200, i as u32 + 2));
}
}
#[test]
fn test_concatenate_streaming_matches_blocks() {
let seg_a: Vec<(u32, u32)> = (0..250).map(|i| (i * 2, (i % 7) + 1)).collect();
let seg_b: Vec<(u32, u32)> = (0..180).map(|i| (i * 5, (i % 3) + 1)).collect();
let seg_c: Vec<(u32, u32)> = (0..90).map(|i| (i * 10, (i % 11) + 1)).collect();
let bpl_a = build_bpl(&seg_a);
let bpl_b = build_bpl(&seg_b);
let bpl_c = build_bpl(&seg_c);
let offset_b = 1000u32;
let offset_c = 2000u32;
let ref_merged = BlockPostingList::concatenate_blocks(&[
(bpl_a.clone(), 0),
(bpl_b.clone(), offset_b),
(bpl_c.clone(), offset_c),
])
.unwrap();
let mut ref_buf = Vec::new();
ref_merged.serialize(&mut ref_buf).unwrap();
let bytes_a = serialize_bpl(&bpl_a);
let bytes_b = serialize_bpl(&bpl_b);
let bytes_c = serialize_bpl(&bpl_c);
let sources: Vec<(&[u8], u32)> =
vec![(&bytes_a, 0), (&bytes_b, offset_b), (&bytes_c, offset_c)];
let mut stream_buf = Vec::new();
let (doc_count, bytes_written) =
BlockPostingList::concatenate_streaming(&sources, &mut stream_buf).unwrap();
assert_eq!(doc_count, 520); assert_eq!(bytes_written, stream_buf.len());
let ref_postings = collect_postings(&BlockPostingList::deserialize(&ref_buf).unwrap());
let stream_postings =
collect_postings(&BlockPostingList::deserialize(&stream_buf).unwrap());
assert_eq!(ref_postings.len(), stream_postings.len());
for (i, (r, s)) in ref_postings.iter().zip(stream_postings.iter()).enumerate() {
assert_eq!(r, s, "mismatch at posting {}", i);
}
}
#[test]
fn test_concatenate_streaming_short_source_returns_corruption() {
let seg_a: Vec<(u32, u32)> = (0..250).map(|i| (i * 2, (i % 7) + 1)).collect();
let seg_c: Vec<(u32, u32)> = (0..90).map(|i| (i * 10, (i % 11) + 1)).collect();
let bytes_a = serialize_bpl(&build_bpl(&seg_a));
let bytes_c = serialize_bpl(&build_bpl(&seg_c));
let short = vec![0u8; FOOTER_SIZE - 1];
let sources: Vec<(&[u8], u32)> = vec![(&bytes_a, 0), (&short, 1000), (&bytes_c, 2000)];
let mut out = Vec::new();
let result = BlockPostingList::concatenate_streaming(&sources, &mut out);
assert!(
matches!(result, Err(crate::Error::Corruption(_))),
"short/corrupt source must be a Corruption error, not silently skipped: {:?}",
result.map(|r| r.0)
);
}
#[test]
fn test_multi_round_merge() {
let segments: Vec<Vec<(u32, u32)>> = (0..4)
.map(|seg| (0..200).map(|i| (i * 3, (i + seg * 7) % 10 + 1)).collect())
.collect();
let bpls: Vec<BlockPostingList> = segments.iter().map(|s| build_bpl(s)).collect();
let serialized: Vec<Vec<u8>> = bpls.iter().map(serialize_bpl).collect();
let mut merged_01 = Vec::new();
let sources_01: Vec<(&[u8], u32)> = vec![(&serialized[0], 0), (&serialized[1], 600)];
let (dc_01, _) =
BlockPostingList::concatenate_streaming(&sources_01, &mut merged_01).unwrap();
assert_eq!(dc_01, 400);
let mut merged_23 = Vec::new();
let sources_23: Vec<(&[u8], u32)> = vec![(&serialized[2], 0), (&serialized[3], 600)];
let (dc_23, _) =
BlockPostingList::concatenate_streaming(&sources_23, &mut merged_23).unwrap();
assert_eq!(dc_23, 400);
let mut final_merged = Vec::new();
let sources_final: Vec<(&[u8], u32)> = vec![(&merged_01, 0), (&merged_23, 1200)];
let (dc_final, _) =
BlockPostingList::concatenate_streaming(&sources_final, &mut final_merged).unwrap();
assert_eq!(dc_final, 800);
let final_bpl = BlockPostingList::deserialize(&final_merged).unwrap();
let postings = collect_postings(&final_bpl);
assert_eq!(postings.len(), 800);
assert_eq!(postings[0].0, 0); assert_eq!(postings[199].0, 597); assert_eq!(postings[200].0, 600); assert_eq!(postings[399].0, 1197); assert_eq!(postings[400].0, 1200); assert_eq!(postings[799].0, 2397);
for seg in 0u32..4 {
for i in 0u32..200 {
let idx = (seg * 200 + i) as usize;
assert_eq!(
postings[idx].1,
(i + seg * 7) % 10 + 1,
"seg{} tf[{}]",
seg,
i
);
}
}
let mut it = final_bpl.iterator();
assert_eq!(it.seek(600), 600);
assert_eq!(it.seek(1200), 1200);
assert_eq!(it.seek(2397), 2397);
assert_eq!(it.seek(2398), TERMINATED);
}
#[test]
fn test_large_scale_merge() {
let num_segments = 5;
let docs_per_segment = 2000;
let docs_gap = 3;
let segments: Vec<Vec<(u32, u32)>> = (0..num_segments)
.map(|seg| {
(0..docs_per_segment)
.map(|i| (i as u32 * docs_gap, (i as u32 + seg as u32) % 20 + 1))
.collect()
})
.collect();
let bpls: Vec<BlockPostingList> = segments.iter().map(|s| build_bpl(s)).collect();
for bpl in &bpls {
assert!(
bpl.num_blocks() >= 15,
"expected >=15 blocks, got {}",
bpl.num_blocks()
);
}
let serialized: Vec<Vec<u8>> = bpls.iter().map(serialize_bpl).collect();
let max_doc_per_seg = (docs_per_segment as u32 - 1) * docs_gap;
let offsets: Vec<u32> = (0..num_segments)
.map(|i| i as u32 * (max_doc_per_seg + 1))
.collect();
let sources: Vec<(&[u8], u32)> = serialized
.iter()
.zip(offsets.iter())
.map(|(b, o)| (b.as_slice(), *o))
.collect();
let mut merged = Vec::new();
let (doc_count, _) =
BlockPostingList::concatenate_streaming(&sources, &mut merged).unwrap();
assert_eq!(doc_count, (num_segments * docs_per_segment) as u32);
let merged_bpl = BlockPostingList::deserialize(&merged).unwrap();
let postings = collect_postings(&merged_bpl);
assert_eq!(postings.len(), num_segments * docs_per_segment);
for i in 1..postings.len() {
assert!(
postings[i].0 > postings[i - 1].0 || (i % docs_per_segment == 0), "doc_id not increasing at {}: {} vs {}",
i,
postings[i - 1].0,
postings[i].0,
);
}
let mut it = merged_bpl.iterator();
for (seg, &expected_first) in offsets.iter().enumerate() {
assert_eq!(
it.seek(expected_first),
expected_first,
"seek to segment {} start",
seg
);
}
}
#[test]
fn test_merge_edge_cases() {
let bpl_a = build_bpl(&[(0, 5)]);
let bpl_b = build_bpl(&[(0, 3)]);
let merged =
BlockPostingList::concatenate_blocks(&[(bpl_a.clone(), 0), (bpl_b.clone(), 1)])
.unwrap();
assert_eq!(merged.doc_count(), 2);
let p = collect_postings(&merged);
assert_eq!(p, vec![(0, 5), (1, 3)]);
let exact_block: Vec<(u32, u32)> = (0..BLOCK_SIZE as u32).map(|i| (i, i % 5 + 1)).collect();
let bpl_exact = build_bpl(&exact_block);
assert_eq!(bpl_exact.num_blocks(), 1);
let bytes = serialize_bpl(&bpl_exact);
let mut out = Vec::new();
let sources: Vec<(&[u8], u32)> = vec![(&bytes, 0), (&bytes, BLOCK_SIZE as u32)];
let (dc, _) = BlockPostingList::concatenate_streaming(&sources, &mut out).unwrap();
assert_eq!(dc, BLOCK_SIZE as u32 * 2);
let merged = BlockPostingList::deserialize(&out).unwrap();
let postings = collect_postings(&merged);
assert_eq!(postings.len(), BLOCK_SIZE * 2);
assert_eq!(postings[BLOCK_SIZE].0, BLOCK_SIZE as u32);
let over_block: Vec<(u32, u32)> = (0..BLOCK_SIZE as u32 + 1).map(|i| (i * 2, 1)).collect();
let bpl_over = build_bpl(&over_block);
assert_eq!(bpl_over.num_blocks(), 2);
}
#[test]
fn test_streaming_roundtrip_single_source() {
let docs: Vec<(u32, u32)> = (0..500).map(|i| (i * 7, i % 15 + 1)).collect();
let bpl = build_bpl(&docs);
let direct = serialize_bpl(&bpl);
let sources: Vec<(&[u8], u32)> = vec![(&direct, 0)];
let mut streamed = Vec::new();
BlockPostingList::concatenate_streaming(&sources, &mut streamed).unwrap();
let p1 = collect_postings(&BlockPostingList::deserialize(&direct).unwrap());
let p2 = collect_postings(&BlockPostingList::deserialize(&streamed).unwrap());
assert_eq!(p1, p2);
}
#[test]
fn test_max_tf_preserved_through_merge() {
let mut a = Vec::new();
for i in 0..200 {
a.push((i * 2, if i == 100 { 50 } else { 1 }));
}
let bpl_a = build_bpl(&a);
assert_eq!(bpl_a.max_tf(), 50);
let mut b = Vec::new();
for i in 0..200 {
b.push((i * 2, if i == 50 { 30 } else { 2 }));
}
let bpl_b = build_bpl(&b);
assert_eq!(bpl_b.max_tf(), 30);
let bytes_a = serialize_bpl(&bpl_a);
let bytes_b = serialize_bpl(&bpl_b);
let sources: Vec<(&[u8], u32)> = vec![(&bytes_a, 0), (&bytes_b, 1000)];
let mut out = Vec::new();
BlockPostingList::concatenate_streaming(&sources, &mut out).unwrap();
let merged = BlockPostingList::deserialize(&out).unwrap();
assert_eq!(merged.max_tf(), 50);
assert_eq!(merged.doc_count(), 400);
}
#[test]
fn test_l0_l1_counts() {
let bpl = build_bpl(&(0..50u32).map(|i| (i, 1)).collect::<Vec<_>>());
assert_eq!(bpl.num_blocks(), 1);
assert_eq!(bpl.l1_docs.len(), 1);
let n = BLOCK_SIZE * L1_INTERVAL;
let bpl = build_bpl(&(0..n as u32).map(|i| (i * 2, 1)).collect::<Vec<_>>());
assert_eq!(bpl.num_blocks(), L1_INTERVAL);
assert_eq!(bpl.l1_docs.len(), 1);
let n = BLOCK_SIZE * L1_INTERVAL + 1;
let bpl = build_bpl(&(0..n as u32).map(|i| (i * 2, 1)).collect::<Vec<_>>());
assert_eq!(bpl.num_blocks(), L1_INTERVAL + 1);
assert_eq!(bpl.l1_docs.len(), 2);
let n = BLOCK_SIZE * L1_INTERVAL * 3;
let bpl = build_bpl(&(0..n as u32).map(|i| (i, 1)).collect::<Vec<_>>());
assert_eq!(bpl.num_blocks(), L1_INTERVAL * 3);
assert_eq!(bpl.l1_docs.len(), 3);
}
#[test]
fn test_l1_last_doc_values() {
let n = BLOCK_SIZE * 20;
let docs: Vec<(u32, u32)> = (0..n as u32).map(|i| (i * 3, 1)).collect();
let bpl = build_bpl(&docs);
assert_eq!(bpl.num_blocks(), 20);
assert_eq!(bpl.l1_docs.len(), 3);
let expected_l1_0 = bpl.block_last_doc(7).unwrap();
assert_eq!(bpl.l1_docs.get(0).unwrap(), expected_l1_0);
let expected_l1_1 = bpl.block_last_doc(15).unwrap();
assert_eq!(bpl.l1_docs.get(1).unwrap(), expected_l1_1);
let expected_l1_2 = bpl.block_last_doc(19).unwrap();
assert_eq!(bpl.l1_docs.get(2).unwrap(), expected_l1_2);
}
#[test]
fn test_seek_block_basic() {
let n = BLOCK_SIZE * 20;
let docs: Vec<(u32, u32)> = (0..n as u32).map(|i| (i * 10, 1)).collect();
let bpl = build_bpl(&docs);
assert_eq!(bpl.seek_block(0, 0), Some(0));
for blk in 0..20 {
let first = bpl.block_first_doc(blk).unwrap();
assert_eq!(
bpl.seek_block(first, 0),
Some(blk),
"seek to block {} first_doc",
blk
);
}
for blk in 0..20 {
let last = bpl.block_last_doc(blk).unwrap();
assert_eq!(
bpl.seek_block(last, 0),
Some(blk),
"seek to block {} last_doc",
blk
);
}
let max_doc = bpl.block_last_doc(19).unwrap();
assert_eq!(bpl.seek_block(max_doc + 1, 0), None);
let mid_doc = bpl.block_first_doc(10).unwrap();
assert_eq!(bpl.seek_block(mid_doc, 10), Some(10));
assert_eq!(
bpl.seek_block(mid_doc, 11),
Some(11).or(bpl.seek_block(mid_doc, 11))
);
}
#[test]
fn test_seek_block_across_l1_boundaries() {
let n = BLOCK_SIZE * 24;
let docs: Vec<(u32, u32)> = (0..n as u32).map(|i| (i * 5, 1)).collect();
let bpl = build_bpl(&docs);
assert_eq!(bpl.l1_docs.len(), 3);
for group in 0..3 {
let blk = group * L1_INTERVAL;
let target = bpl.block_first_doc(blk).unwrap();
assert_eq!(
bpl.seek_block(target, 0),
Some(blk),
"seek to group {} block {}",
group,
blk
);
}
let target = bpl.block_first_doc(20).unwrap() + 1;
assert_eq!(bpl.seek_block(target, 0), Some(20));
}
#[test]
fn block_len_matches_l0_offsets() {
let bpl = build_bpl(&(0..1000).map(|i| (i * 3, 1 + i % 4)).collect::<Vec<_>>());
let mut total = 0usize;
for b in 0..bpl.num_blocks() {
let (_, _, offset, _) = bpl.read_l0_entry(b);
assert_eq!(offset as usize, total, "block {b} offset");
total += bpl.block_len(b);
}
assert_eq!(total, bpl.stream.len());
}
#[test]
fn every_codec_round_trips_and_seeks() {
let mut postings: Vec<(u32, u32)> = Vec::new();
let mut doc = 0u32;
for i in 0..5000u32 {
doc += if i % 97 == 0 { 100_000 } else { 1 + i % 7 };
let tf = if i % 131 == 0 { 5000 } else { 1 + i % 3 };
postings.push((doc, tf));
}
let mut list = PostingList::new();
for &(d, tf) in &postings {
list.push(d, tf);
}
let rounded = BlockPostingList::from_posting_list(&list).unwrap();
let mut sizes = Vec::new();
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let bpl = BlockPostingList::from_posting_list_with_codec(&list, codec).unwrap();
assert_eq!(collect_postings(&bpl), postings, "{codec}");
for b in 0..bpl.num_blocks() {
let count = (postings.len() - b * BLOCK_SIZE).min(BLOCK_SIZE);
let expected_codec = if codec == PostingCodec::Simd4x && count < BLOCK_SIZE {
PostingCodec::Rounded
} else {
codec
};
assert_eq!(bpl.block_codec(b), Some(expected_codec));
assert_eq!(bpl.block_max_tf(b), rounded.block_max_tf(b));
}
let bytes = serialize_bpl(&bpl);
let back = BlockPostingList::deserialize(&bytes).unwrap();
assert_eq!(collect_postings(&back), postings, "{codec} deserialize");
let back =
BlockPostingList::deserialize_zero_copy(OwnedBytes::new(bytes.clone())).unwrap();
assert_eq!(collect_postings(&back), postings, "{codec} zero-copy");
let mut a = rounded.iterator();
let mut b = back.iterator();
for target in (0..postings.last().unwrap().0 + 10).step_by(2_003) {
assert_eq!(a.seek(target), b.seek(target), "{codec} seek {target}");
assert_eq!(a.term_freq(), b.term_freq());
}
sizes.push((codec, bytes.len()));
}
let rounded_bytes = serialize_bpl(&rounded);
assert_eq!(
serialize_bpl(
&BlockPostingList::from_posting_list_with_codec(&list, PostingCodec::Rounded)
.unwrap()
),
rounded_bytes,
"Rounded must stay byte-identical"
);
assert!(matches!(rounded.stream[6], 0 | 8 | 16 | 32));
let size = |c: PostingCodec| sizes.iter().find(|(k, _)| *k == c).unwrap().1;
assert!(size(PostingCodec::Packed) < size(PostingCodec::Rounded));
assert!(size(PostingCodec::Pfor) < size(PostingCodec::Packed));
}
#[test]
fn mixed_codec_sources_concatenate() {
let a: Vec<(u32, u32)> = (0..300u32).map(|i| (i * 5, 1 + i % 9)).collect();
let b: Vec<(u32, u32)> = (0..300u32).map(|i| (i * 11 + 3, 2 + i % 5)).collect();
let list_a = {
let mut l = PostingList::new();
a.iter().for_each(|&(d, t)| l.push(d, t));
BlockPostingList::from_posting_list_with_codec(&l, PostingCodec::Pfor).unwrap()
};
let list_b = {
let mut l = PostingList::new();
b.iter().for_each(|&(d, t)| l.push(d, t));
BlockPostingList::from_posting_list_with_codec(&l, PostingCodec::Packed).unwrap()
};
let offset_b = a.last().unwrap().0 + 1;
let expected: Vec<(u32, u32)> = a
.iter()
.copied()
.chain(b.iter().map(|&(d, t)| (d + offset_b, t)))
.collect();
let merged = BlockPostingList::concatenate_blocks(&[
(list_a.clone(), 0),
(list_b.clone(), offset_b),
])
.unwrap();
assert_eq!(collect_postings(&merged), expected);
let bytes_a = serialize_bpl(&list_a);
let bytes_b = serialize_bpl(&list_b);
let mut out = Vec::new();
let (docs, written) = BlockPostingList::concatenate_streaming(
&[(bytes_a.as_slice(), 0), (bytes_b.as_slice(), offset_b)],
&mut out,
)
.unwrap();
assert_eq!(docs, 600);
assert_eq!(written, out.len());
let streamed = BlockPostingList::deserialize(&out).unwrap();
assert_eq!(collect_postings(&streamed), expected);
assert_eq!(streamed.block_codec(0), Some(PostingCodec::Pfor));
assert_eq!(
streamed.block_codec(streamed.num_blocks() - 1),
Some(PostingCodec::Packed)
);
}
#[test]
fn test_l0_entry_roundtrip() {
let docs: Vec<(u32, u32)> = (0..1000u32).map(|i| (i * 3, (i % 10) + 1)).collect();
let bpl = build_bpl(&docs);
let bytes = serialize_bpl(&bpl);
let bpl2 = BlockPostingList::deserialize(&bytes).unwrap();
assert_eq!(bpl.num_blocks(), bpl2.num_blocks());
for blk in 0..bpl.num_blocks() {
assert_eq!(
bpl.read_l0_entry(blk),
bpl2.read_l0_entry(blk),
"L0 entry mismatch at block {}",
blk
);
}
assert_eq!(bpl.l1_docs.bytes(), bpl2.l1_docs.bytes());
}
#[test]
fn posting_open_borrows_unaligned_group_directories_and_preserves_bytes_and_seeks() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let mut postings = PostingList::new();
for i in 0..3457u32 {
postings.push(i * 7 + 3, i % 13 + 1);
}
let built = BlockPostingList::from_posting_list_with_codec(&postings, codec).unwrap();
let encoded = serialize_bpl(&built);
for prefix in [1, 3, 7] {
let mut padded = vec![0; prefix];
padded.extend_from_slice(&encoded);
let owner = OwnedBytes::new(padded);
let raw = owner.slice(prefix..owner.len());
let footer = Footer::parse(&raw).unwrap();
let docs_start = raw[footer.l1_start()..].as_ptr();
let bounds_start = raw[footer.l1_end()..].as_ptr();
let opened = BlockPostingList::deserialize_zero_copy(raw).unwrap();
assert_eq!(
opened.l1_docs.bytes().as_ptr(),
docs_start,
"opening must borrow group document metadata"
);
assert_eq!(
opened.l1_bounds.bytes().as_ptr(),
bounds_start,
"opening must borrow group bound metadata"
);
assert_eq!(serialize_bpl(&opened), encoded);
for from in 0..=built.num_blocks() {
for target in (0..=3457 * 7 + 4).step_by(37) {
let expected = (from..built.num_blocks())
.find(|&block| built.block_last_doc(block).unwrap() >= target);
assert_eq!(opened.seek_block(target, from), expected);
}
}
assert_eq!(collect_postings(&opened), collect_postings(&built));
}
}
}
#[test]
fn test_zero_copy_deserialize_matches() {
let docs: Vec<(u32, u32)> = (0..2000u32).map(|i| (i * 2, (i % 5) + 1)).collect();
let bpl = build_bpl(&docs);
let bytes = serialize_bpl(&bpl);
let copied = BlockPostingList::deserialize(&bytes).unwrap();
let zero_copy =
BlockPostingList::deserialize_zero_copy(OwnedBytes::new(bytes.clone())).unwrap();
assert_eq!(copied.l0_count, zero_copy.l0_count);
assert_eq!(copied.l1_docs.bytes(), zero_copy.l1_docs.bytes());
assert_eq!(copied.doc_count, zero_copy.doc_count);
assert_eq!(copied.max_tf, zero_copy.max_tf);
let p1 = collect_postings(&copied);
let p2 = collect_postings(&zero_copy);
assert_eq!(p1, p2);
}
#[test]
fn test_l1_preserved_through_streaming_merge() {
let seg_a = build_bpl(&(0..1000u32).map(|i| (i * 2, 1)).collect::<Vec<_>>());
let seg_b = build_bpl(&(0..800u32).map(|i| (i * 3, 2)).collect::<Vec<_>>());
let seg_c = build_bpl(&(0..500u32).map(|i| (i * 5, 3)).collect::<Vec<_>>());
let bytes_a = serialize_bpl(&seg_a);
let bytes_b = serialize_bpl(&seg_b);
let bytes_c = serialize_bpl(&seg_c);
let sources: Vec<(&[u8], u32)> = vec![(&bytes_a, 0), (&bytes_b, 10000), (&bytes_c, 20000)];
let mut out = Vec::new();
BlockPostingList::concatenate_streaming(&sources, &mut out).unwrap();
let merged = BlockPostingList::deserialize(&out).unwrap();
let expected_l1_count = merged.num_blocks().div_ceil(L1_INTERVAL);
assert_eq!(merged.l1_docs.len(), expected_l1_count);
for (i, word) in merged.l1_docs.words().iter().enumerate() {
let l1_doc = u32::from_le_bytes(*word);
let last_block_in_group = ((i + 1) * L1_INTERVAL - 1).min(merged.num_blocks() - 1);
let expected = merged.block_last_doc(last_block_in_group).unwrap();
assert_eq!(l1_doc, expected, "L1[{}] mismatch", i);
}
for blk in 0..merged.num_blocks() {
let first = merged.block_first_doc(blk).unwrap();
assert_eq!(merged.seek_block(first, 0), Some(blk));
}
}
#[test]
fn test_seek_block_single_block() {
let bpl = build_bpl(&[(0, 1), (10, 2), (20, 3)]);
assert_eq!(bpl.num_blocks(), 1);
assert_eq!(bpl.l1_docs.len(), 1);
assert_eq!(bpl.seek_block(0, 0), Some(0));
assert_eq!(bpl.seek_block(10, 0), Some(0));
assert_eq!(bpl.seek_block(20, 0), Some(0));
assert_eq!(bpl.seek_block(21, 0), None);
}
#[test]
fn test_footer_size() {
let docs: Vec<(u32, u32)> = (0..500u32).map(|i| (i * 2, 1)).collect();
let bpl = build_bpl(&docs);
let bytes = serialize_bpl(&bpl);
let expected = bpl.stream.len()
+ bpl.l0_count * L0_SIZE
+ bpl.l1_docs.len() * (L1_SIZE + 4)
+ FOOTER_V2_SIZE;
assert_eq!(bytes.len(), expected);
}
fn build_bpl_with_positions(postings: &[(u32, u32)]) -> BlockPostingList {
let mut list = PostingList::new();
for &(doc, tf) in postings {
list.push(doc, tf);
}
BlockPostingList::from_posting_list_with(&list, true, None).unwrap()
}
fn expected_cursors(postings: &[(u32, u32)]) -> Vec<u64> {
let mut acc = 0u64;
postings
.iter()
.map(|&(_, tf)| {
let c = acc;
acc += tf as u64;
c
})
.collect()
}
fn iterator_cursors(bpl: &BlockPostingList) -> Vec<u64> {
let mut it = bpl.iterator();
let mut out = Vec::new();
while it.doc() != TERMINATED {
out.push(it.position_cursor());
it.advance();
}
out
}
#[test]
fn position_cursors_survive_serialization_and_seeks() {
let docs: Vec<(u32, u32)> = (0..700u32).map(|i| (i * 3, i % 5 + 1)).collect();
let bpl = build_bpl_with_positions(&docs);
assert!(bpl.has_position_cursors());
assert_eq!(
bpl.total_positions(),
docs.iter().map(|&(_, tf)| tf as u64).sum::<u64>()
);
assert_eq!(bpl.pos_cursor(0), Some(0));
assert_eq!(
bpl.pos_cursor(1),
Some(docs[..128].iter().map(|&(_, tf)| tf as u64).sum::<u64>())
);
assert_eq!(iterator_cursors(&bpl), expected_cursors(&docs));
let bytes = serialize_bpl(&bpl);
assert_eq!(
bytes.len(),
bpl.stream.len()
+ bpl.l0_count * (L0_SIZE + CURSOR_SIZE)
+ bpl.l1_docs.len() * (L1_SIZE + 4)
+ FOOTER_V2_SIZE
);
assert!(BlockPostingList::has_cursors_bytes(&bytes));
let decoded =
BlockPostingList::deserialize_zero_copy(OwnedBytes::new(bytes.clone())).unwrap();
assert_eq!(iterator_cursors(&decoded), expected_cursors(&docs));
assert_eq!(decoded.total_positions(), bpl.total_positions());
let mut it = decoded.iterator();
let expected = expected_cursors(&docs);
for (i, &(doc, _)) in docs.iter().enumerate().step_by(37) {
assert_eq!(it.seek(doc), doc);
assert_eq!(it.position_cursor(), expected[i], "cursor at doc {doc}");
}
let mut it = decoded.iterator();
assert_eq!(it.seek(docs[600].0 + 1), docs[601].0);
assert_eq!(it.position_cursor(), expected[601]);
let plain = build_bpl(&docs);
assert!(!plain.has_position_cursors());
assert_eq!(plain.pos_cursor(0), None);
assert_eq!(plain.total_positions(), 0);
}
#[test]
fn length_bounds_are_packed_per_block_and_survive_merges() {
let docs: Vec<(u32, u32)> = (0..300u32).map(|i| (i, i % 3 + 1)).collect();
let length_of = |doc: u32| 10 + (doc % 50) * 7;
let mut list = PostingList::new();
for &(doc, tf) in &docs {
list.push(doc, tf);
}
let bpl = BlockPostingList::from_posting_list_with(&list, true, Some(&length_of)).unwrap();
assert_eq!(bpl.min_len(), Some(10));
assert_eq!(bpl.block_bounds(0), Some((3, Some(10))));
assert_eq!(bpl.block_bounds(2), Some((3, Some(52))));
assert_eq!(bpl.block_max_tf(2), Some(3));
let bytes = serialize_bpl(&bpl);
let decoded = BlockPostingList::deserialize(&bytes).unwrap();
assert_eq!(decoded.min_len(), Some(10));
assert_eq!(decoded.block_bounds(2), Some((3, Some(52))));
assert_eq!(decoded.group_bounds(0), Some((3, 10)));
assert_eq!(decoded.group_bounds(2), Some((3, 10)));
assert_eq!(decoded.group_bounds(3), None);
assert_eq!(decoded.group_last_doc(1), Some(299));
assert_eq!(decoded.next_group_block(1), 3);
let plain = build_bpl(&docs);
assert_eq!(plain.min_len(), Some(1));
assert_eq!(plain.block_bounds(0), Some((3, Some(1))));
let mut out = Vec::new();
BlockPostingList::concatenate_streaming(&[(&bytes, 0), (&bytes, 1000)], &mut out).unwrap();
let merged = BlockPostingList::deserialize(&out).unwrap();
assert_eq!(merged.min_len(), Some(10));
assert_eq!(merged.block_bounds(2), Some((3, Some(52))));
assert_eq!(merged.block_bounds(3), Some((3, Some(10))));
assert_eq!(merged.block_max_tf(5), Some(3));
assert_eq!(merged.group_bounds(5), Some((3, 10)));
assert_eq!(merged.group_last_doc(5), Some(1299));
assert_eq!(merged.next_group_block(5), 6);
}
#[test]
fn legacy_footer_without_magic_still_deserializes() {
let docs: Vec<(u32, u32)> = (0..300u32).map(|i| (i * 2, 1 + i % 3)).collect();
let bpl = build_bpl(&docs);
let bytes = serialize_bpl(&bpl);
let footer = Footer::parse(&bytes).unwrap();
let mut legacy = bytes[..footer.l1_end()].to_vec();
for block in 0..bpl.num_blocks() {
let at = footer.l0_start() + block * L0_SIZE + 12;
legacy[at..at + 4]
.copy_from_slice(&(bpl.block_max_tf(block).unwrap() as f32).to_le_bytes());
}
legacy.extend_from_slice(
&bytes[bytes.len() - FOOTER_V2_SIZE..bytes.len() - (FOOTER_V2_SIZE - FOOTER_SIZE)],
);
assert!(!BlockPostingList::has_cursors_bytes(&legacy));
let decoded = BlockPostingList::deserialize(&legacy).unwrap();
assert_eq!(collect_postings(&decoded), docs);
assert_eq!(decoded.max_tf(), 3);
assert!(!decoded.has_position_cursors());
assert_eq!(decoded.min_len(), None);
assert_eq!(decoded.group_bounds(0), None);
let mut out = Vec::new();
let (count, written) =
BlockPostingList::concatenate_streaming(&[(&legacy, 0), (&legacy, 1000)], &mut out)
.unwrap();
assert_eq!(count, 600);
assert_eq!(written, out.len());
let merged = BlockPostingList::deserialize(&out).unwrap();
assert_eq!(merged.doc_count(), 600);
assert!(!merged.has_position_cursors());
}
#[test]
fn streaming_merge_rebases_position_cursors() {
let a: Vec<(u32, u32)> = (0..200u32).map(|i| (i, i % 4 + 1)).collect();
let b: Vec<(u32, u32)> = (0..150u32).map(|i| (i * 2, 2)).collect();
let bytes_a = serialize_bpl(&build_bpl_with_positions(&a));
let bytes_b = serialize_bpl(&build_bpl_with_positions(&b));
let mut out = Vec::new();
let (count, written) =
BlockPostingList::concatenate_streaming(&[(&bytes_a, 0), (&bytes_b, 1000)], &mut out)
.unwrap();
assert_eq!(count, 350);
assert_eq!(written, out.len());
let merged = BlockPostingList::deserialize(&out).unwrap();
assert!(merged.has_position_cursors());
let all: Vec<(u32, u32)> = a
.iter()
.copied()
.chain(b.iter().map(|&(d, tf)| (d + 1000, tf)))
.collect();
assert_eq!(collect_postings(&merged), all);
assert_eq!(iterator_cursors(&merged), expected_cursors(&all));
assert_eq!(
merged.total_positions(),
all.iter().map(|&(_, tf)| tf as u64).sum::<u64>()
);
let reference = BlockPostingList::concatenate_blocks(&[
(build_bpl_with_positions(&a), 0),
(build_bpl_with_positions(&b), 1000),
])
.unwrap();
assert_eq!(iterator_cursors(&reference), expected_cursors(&all));
let plain = serialize_bpl(&build_bpl(&b));
assert!(
BlockPostingList::concatenate_streaming(
&[(&bytes_a, 0), (&plain, 1000)],
&mut Vec::new()
)
.is_err()
);
}
#[test]
fn test_seek_block_from_block_skips_earlier() {
let n = BLOCK_SIZE * 16;
let docs: Vec<(u32, u32)> = (0..n as u32).map(|i| (i * 3, 1)).collect();
let bpl = build_bpl(&docs);
let target_in_5 = bpl.block_first_doc(5).unwrap() + 1;
let result = bpl.seek_block(target_in_5, 8);
assert!(result.is_some());
assert!(result.unwrap() >= 8);
}
#[test]
fn saturated_tf_bounds_remain_upper_bounds_without_changing_encoded_bytes() {
let docs: Vec<_> = (0..(BLOCK_SIZE * 18) as u32)
.map(|doc| {
(
doc,
if doc == (BLOCK_SIZE * 17) as u32 {
100_000
} else {
1
},
)
})
.collect();
let list = build_bpl(&docs);
let bytes = serialize_bpl(&list);
assert_eq!(list.max_tf(), 100_000);
assert!(list.block_bounds(17).unwrap().0 >= 100_000);
assert!(list.group_bounds(17).unwrap().0 >= 100_000);
assert_eq!(list.block_bounds(0).unwrap().0, 1);
let decoded = BlockPostingList::deserialize(&bytes).unwrap();
assert!(decoded.block_bounds(17).unwrap().0 >= 100_000);
assert!(decoded.group_bounds(17).unwrap().0 >= 100_000);
assert_eq!(serialize_bpl(&decoded), bytes);
assert_eq!(collect_postings(&decoded), docs);
}
#[test]
fn ratio_bounds_preserve_payload_bytes_and_copy_merge_with_legacy_blocks() {
for codec in [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
] {
let mut postings = PostingList::new();
for i in 0..1100 {
postings.push(i * 3, 1 + i % 97);
}
let length = |id| 10 + id % 997;
let plain = BlockPostingList::from_posting_list_with_options(
&postings,
true,
Some(&length),
codec,
)
.unwrap();
let tight = BlockPostingList::from_posting_list_with_ratio_bounds(
&postings,
true,
Some(&length),
codec,
)
.unwrap();
assert_eq!(plain.stream.as_slice(), tight.stream.as_slice());
assert_eq!(plain.l0_bytes.as_slice(), tight.l0_bytes.as_slice());
assert_eq!(
plain.pos_cursors.as_ref().map(OwnedBytes::as_slice),
tight.pos_cursors.as_ref().map(OwnedBytes::as_slice)
);
let mut raw = Vec::new();
tight.serialize(&mut raw).unwrap();
let decoded = BlockPostingList::deserialize(&raw).unwrap();
assert!(decoded.has_ratio_bounds());
for block in 0..tight.num_blocks() {
assert!(decoded.block_length_ratio(block) > 0.0);
for posting in &postings.postings
[block * BLOCK_SIZE..((block + 1) * BLOCK_SIZE).min(postings.len())]
{
assert!(
decoded.block_length_ratio(block) as f64
<= length(posting.doc_id) as f64 / posting.term_freq as f64
);
assert!(decoded.group_length_ratio(block) <= decoded.block_length_ratio(block));
}
}
let merged = BlockPostingList::concatenate_blocks(&[
(tight.clone(), 0),
(plain.clone(), 10_000),
])
.unwrap();
let mut plain_raw = Vec::new();
plain.serialize(&mut plain_raw).unwrap();
let mut streaming = Vec::new();
let (_, size) = BlockPostingList::concatenate_streaming(
&[(&raw, 0), (&plain_raw, 10_000)],
&mut streaming,
)
.unwrap();
let mut expected = Vec::new();
merged.serialize(&mut expected).unwrap();
assert_eq!(streaming, expected);
assert_eq!(streaming.len(), size);
for block in 0..tight.num_blocks() {
assert_eq!(
merged.block_length_ratio(block),
tight.block_length_ratio(block)
);
assert_eq!(merged.block_length_ratio(block + tight.num_blocks()), 0.0);
let mut before = Vec::new();
let mut after = Vec::new();
let mut tfs_before = Vec::new();
let mut tfs_after = Vec::new();
tight.decode_block_into(block, &mut before, &mut tfs_before);
merged.decode_block_into(block, &mut after, &mut tfs_after);
assert_eq!((before, tfs_before), (after, tfs_after));
}
let footer = Footer::parse(&raw).unwrap();
raw.drain(footer.cursors_end()..footer.ratios_end());
let flags = raw.len() - 12;
raw[flags] &= !(FLAG_RATIO_BOUNDS as u8);
let legacy_view = BlockPostingList::deserialize(&raw).unwrap();
assert!(!legacy_view.has_ratio_bounds());
assert_eq!(legacy_view.stream.as_slice(), tight.stream.as_slice());
}
}
#[test]
fn ratio_bounds_reject_truncation_nonfinite_negative_and_unknown_flags() {
let mut postings = PostingList::new();
postings.push(1, 3);
let list = BlockPostingList::from_posting_list_with_ratio_bounds(
&postings,
false,
Some(&|_| 7),
PostingCodec::Rounded,
)
.unwrap();
let mut raw = Vec::new();
list.serialize(&mut raw).unwrap();
let footer = Footer::parse(&raw).unwrap();
for bad in [f32::NAN, f32::INFINITY, -1.0] {
let mut corrupt = raw.clone();
corrupt[footer.cursors_end()..footer.cursors_end() + 4]
.copy_from_slice(&bad.to_le_bytes());
assert!(BlockPostingList::deserialize(&corrupt).is_err());
assert!(
BlockPostingList::concatenate_streaming(&[(&corrupt, 0)], &mut Vec::new()).is_err()
);
}
let mut short = raw.clone();
short.remove(footer.cursors_end());
assert!(BlockPostingList::deserialize(&short).is_err());
let at = raw.len() - 12;
raw[at] |= 128;
assert!(BlockPostingList::deserialize(&raw).is_err());
}
const ALL_CODECS: [PostingCodec; 4] = [
PostingCodec::Rounded,
PostingCodec::Packed,
PostingCodec::Pfor,
PostingCodec::Simd4x,
];
fn first_delta_byte(list: &BlockPostingList, block: usize) -> usize {
let (_, _, offset, _) = list.read_l0_entry(block);
let header = offset as usize;
header + 8 + usize::from(list.block_codec(block) == Some(PostingCodec::Pfor))
}
#[test]
fn content_corrupt_block_never_panics_or_truncates_silently() {
for codec in ALL_CODECS {
let mut postings = PostingList::new();
for doc in 0..128 {
postings.push(doc, doc % 5 + 1);
}
postings.push(200, 2);
postings.push(300, 3);
let list =
BlockPostingList::from_posting_list_with_options(&postings, true, None, codec)
.unwrap();
let bytes = serialize_bpl(&list);
let at = first_delta_byte(&list, 1);
assert_eq!(
bytes[at], 100,
"{codec}: delta 300-200 at the first payload byte"
);
let mut corrupt = bytes.clone();
corrupt[at] = 1; let admitted = BlockPostingList::deserialize(&corrupt)
.unwrap_or_else(|e| panic!("{codec}: structural admission must pass: {e}"));
let mut cursor = admitted.iterator();
assert_eq!(cursor.seek(250), TERMINATED, "{codec}");
assert_eq!(cursor.term_freq(), 0);
assert_eq!(cursor.advance(), TERMINATED);
let mut decoded = Vec::new();
assert!(
admitted
.decode_block_doc_ids_only(0, &mut decoded)
.is_some()
);
assert_eq!(decoded.len(), 128);
assert!(
admitted
.decode_block_doc_ids_only(1, &mut decoded)
.is_none(),
"{codec}: a content-corrupt block must be reported, not decoded"
);
assert!(decoded.is_empty(), "no ids escape a corrupt block");
assert!(!admitted.decode_block_into(1, &mut decoded, &mut Vec::new()));
assert_eq!(
collect_postings(&admitted),
collect_postings(&list)[..128],
"{codec}"
);
let mut window = admitted.iterator();
let mut bits = [0u64; 8];
window.fill_doc_window(0, &mut bits);
assert_eq!(bits[..2], [u64::MAX, u64::MAX]);
assert_eq!(window.doc(), TERMINATED);
assert_eq!(
collect_postings(&BlockPostingList::deserialize(&bytes).unwrap()).len(),
130
);
}
let mut postings = PostingList::new();
for doc in 0..128 {
postings.push(doc, 1);
}
for i in 0..128 {
postings.push(200 + i * 2, 1);
}
let list = BlockPostingList::from_posting_list_with_codec(&postings, PostingCodec::Simd4x)
.unwrap();
assert_eq!(list.block_codec(1), Some(PostingCodec::Simd4x));
let mut bytes = serialize_bpl(&list);
let at = first_delta_byte(&list, 1) + 1;
assert_ne!(bytes[at], 0);
bytes[at] = 0;
let admitted = BlockPostingList::deserialize(&bytes).unwrap();
let mut cursor = admitted.iterator();
assert_eq!(cursor.seek(450), TERMINATED);
assert!(
admitted
.decode_block_doc_ids_only(1, &mut Vec::new())
.is_none()
);
assert_eq!(collect_postings(&admitted).len(), 128);
}
fn simd_exact_tail_bytes(docs: &[(u32, u32)]) -> Vec<u8> {
assert!(!docs.is_empty() && docs.len() < BLOCK_SIZE);
let count = docs.len();
let (first, last) = (docs[0].0, docs[count - 1].0);
let deltas: Vec<u32> = docs.windows(2).map(|w| w[1].0 - w[0].0).collect();
let tfs: Vec<u32> = docs.iter().map(|d| d.1).collect();
let max_tf = tfs.iter().copied().max().unwrap();
let mut stream = Vec::new();
stream.write_u16::<LittleEndian>(count as u16).unwrap();
stream.write_u32::<LittleEndian>(first).unwrap();
let header_at = stream.len();
stream.extend_from_slice(&[0, 0]);
let doc_bits = bitpacking4x::encode_gaps(&deltas, &mut stream);
let tf_bits = bitpacking4x::encode(&tfs, &mut stream);
stream[header_at] = PostingCodec::Simd4x.header_byte(doc_bits);
stream[header_at + 1] = tf_bits;
let mut bytes = stream.clone();
write_l0(&mut bytes, first, last, 0, pack_bounds(max_tf, 1));
bytes.extend_from_slice(&last.to_le_bytes());
bytes.extend_from_slice(&pack_bounds(max_tf, 1).to_le_bytes());
BlockPostingList::write_footer(
&mut bytes,
stream.len() as u64,
1,
1,
count as u32,
max_tf,
0,
false,
Some(1),
true,
false,
false,
false,
0,
)
.unwrap();
bytes
}
#[test]
fn simd4x_exact_tail_blocks_decode_seek_and_copy_through_merge() {
for count in [1usize, 2, 3, 17, 127] {
let docs: Vec<(u32, u32)> = (0..count as u32)
.map(|i| (i * 3 + 7, i % 4 + 1 + (i == 5) as u32 * 900))
.collect();
let bytes = simd_exact_tail_bytes(&docs);
let list = BlockPostingList::deserialize(&bytes).unwrap();
assert_eq!(
list.block_codec(0),
Some(PostingCodec::Simd4x),
"count={count}"
);
assert_eq!(collect_postings(&list), docs, "count={count}");
let mut cursor = list.iterator();
for &(doc, tf) in &docs {
assert_eq!(cursor.seek(doc.saturating_sub(1)), doc);
assert_eq!(cursor.term_freq(), tf);
}
assert_eq!(cursor.seek(docs[count - 1].0 + 1), TERMINATED);
assert_eq!(serialize_bpl(&list), bytes, "byte-identical round trip");
let mut out = Vec::new();
let (merged_count, written) =
BlockPostingList::concatenate_streaming(&[(&bytes, 0), (&bytes, 1000)], &mut out)
.unwrap();
assert_eq!((merged_count as usize, written), (2 * count, out.len()));
let merged = BlockPostingList::deserialize(&out).unwrap();
assert_eq!(merged.block_codec(1), Some(PostingCodec::Simd4x));
let expected: Vec<(u32, u32)> = docs
.iter()
.copied()
.chain(docs.iter().map(|&(d, t)| (d + 1000, t)))
.collect();
assert_eq!(collect_postings(&merged), expected, "count={count}");
let typed =
BlockPostingList::concatenate_blocks(&[(list.clone(), 0), (list.clone(), 1000)])
.unwrap();
assert_eq!(serialize_bpl(&typed), out);
let mut short = bytes.clone();
short[0] = (count + 1) as u8;
assert!(BlockPostingList::deserialize(&short).is_err());
}
}
#[test]
fn legacy_footer_whose_max_tf_equals_the_magic_is_rejected_not_misread() {
for count in [1u32, 2, 130, 700] {
let mut postings = PostingList::new();
for i in 0..count {
postings.push(i * 2, if i == 0 { FOOTER_MAGIC } else { 1 });
}
let bpl = BlockPostingList::from_posting_list(&postings).unwrap();
assert_eq!(bpl.max_tf(), FOOTER_MAGIC);
let bytes = serialize_bpl(&bpl);
let footer = Footer::parse(&bytes).unwrap();
let mut legacy = bytes[..footer.l1_end()].to_vec();
for block in 0..bpl.num_blocks() {
let at = footer.l0_start() + block * L0_SIZE + 12;
legacy[at..at + 4]
.copy_from_slice(&(bpl.block_max_tf(block).unwrap() as f32).to_le_bytes());
}
legacy.extend_from_slice(
&bytes[bytes.len() - FOOTER_V2_SIZE..bytes.len() - (FOOTER_V2_SIZE - FOOTER_SIZE)],
);
assert_eq!(
u32::from_le_bytes(legacy[legacy.len() - 4..].try_into().unwrap()),
FOOTER_MAGIC
);
assert!(
BlockPostingList::deserialize(&legacy).is_err(),
"count={count}: ambiguous footer must not be read as either layout"
);
assert!(!BlockPostingList::has_cursors_bytes(&legacy));
let mut out = vec![0xab];
assert!(BlockPostingList::concatenate_streaming(&[(&legacy, 0)], &mut out).is_err());
assert_eq!(out, [0xab]);
assert_eq!(
BlockPostingList::deserialize(&bytes).unwrap().max_tf(),
FOOTER_MAGIC
);
}
}
#[test]
fn mixed_cursor_and_cursorless_sources_fail_loudly_before_output() {
let docs: Vec<(u32, u32)> = (0..300u32).map(|i| (i * 2, i % 3 + 1)).collect();
let with = build_bpl_with_positions(&docs);
let without = build_bpl(&docs);
assert!(with.has_position_cursors() && !without.has_position_cursors());
for sources in [
[(with.clone(), 0), (without.clone(), 1000)],
[(without.clone(), 0), (with.clone(), 1000)],
] {
let error = BlockPostingList::concatenate_blocks(&sources).unwrap_err();
assert!(
error
.to_string()
.contains("with and without position cursors"),
"{error}"
);
let bytes: Vec<Vec<u8>> = sources.iter().map(|(s, _)| serialize_bpl(s)).collect();
let mut out = vec![0xab];
let result = BlockPostingList::concatenate_streaming(
&[(&bytes[0], 0), (&bytes[1], 1000)],
&mut out,
);
match result {
Err(crate::Error::Corruption(message)) => {
assert!(
message.contains("with and without position cursors"),
"{message}"
)
}
other => panic!("expected a loud corruption error, got {other:?}"),
}
assert_eq!(out, [0xab], "nothing is written for a rejected source set");
}
}
}
#[cfg(test)]
#[path = "posting/merge_admission_tests.rs"]
mod merge_admission_tests;
#[cfg(test)]
mod byte_gap_validation_tests {
use super::*;
#[test]
fn strict_gap_validation_agrees_with_full_scan_at_width_and_wrap_boundaries() {
for count in 1..=BLOCK_SIZE {
for width in 0..=32 {
let mask = if width == 0 {
0
} else {
u32::MAX >> (32 - width)
};
for first in [0u32, 17, u32::MAX / 2, u32::MAX - 1] {
for pattern in 0..3 {
let mut docs = vec![first];
for i in 1..count {
let encoded = match pattern {
0 => mask,
1 => 0,
_ => (i as u32).wrapping_mul(0x9e3779b9) & mask,
};
docs.push(docs.last().unwrap().wrapping_add(encoded).wrapping_add(1));
}
let end = *docs.last().unwrap();
for last in [end, end.wrapping_add(1), end.wrapping_sub(1)] {
assert_eq!(
verify_block_docs(&docs, first, last, None, Some(width)),
verify_block_docs(&docs, first, last, None, None),
"count={count}, width={width}, first={first}, pattern={pattern}, last={last}"
);
}
}
}
}
}
}
#[test]
fn byte_gap_validation_agrees_with_decoded_order_for_tails_and_unsigned_wraps() {
for count in 1..=BLOCK_SIZE {
for first in [0u32, 17, u32::MAX - 400, u32::MAX - 1] {
for pattern in 0..5 {
let gaps: Vec<u8> = (1..count)
.map(|i| match pattern {
0 => 1,
1 => 255,
2 => {
if i == count / 2 {
0
} else {
1
}
}
3 => (i * 71) as u8,
_ => ((i * 71) as u8).max(1),
})
.collect();
let mut docs = vec![first];
for &gap in &gaps {
docs.push(docs.last().unwrap().wrapping_add(u32::from(gap)));
}
let end = *docs.last().unwrap();
for last in [end, end.wrapping_add(1), end.wrapping_sub(1)] {
assert_eq!(
verify_block_docs(&docs, first, last, Some(&gaps), None),
verify_block_docs(&docs, first, last, None, None),
"count={count}, first={first}, pattern={pattern}, last={last}"
);
}
}
}
}
}
}