#[cfg(test)]
use std::cell::Cell;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, LazyLock};
use std::{
cell::{OnceCell, RefCell, UnsafeCell},
collections::{BinaryHeap, VecDeque},
};
use std::{cmp::Reverse, fmt::Debug};
use arrow::array::AsArray;
use arrow::datatypes::Int32Type;
use arrow_array::Array;
use itertools::Itertools;
use lance_core::utils::address::RowAddress;
use lance_core::{Error, Result};
use lance_select::RowAddrMask;
use smallvec::SmallVec;
use crate::metrics::MetricsCollector;
#[path = "wand_intersection.rs"]
mod intersection;
use super::{
CompressedPositionStorage,
documents::{DocId, DocLengths, DocVisibility},
impact::{IMPACT_LEVEL1_BLOCKS, ImpactScoreCache, ImpactSkipData},
index::{PositionStreamCodec, dequantize_doc_length},
query::Operator,
scorer::{BM25_DOC_WEIGHT_UPPER_BOUND, MemBM25Scorer, bm25_doc_weight_with_norm, idf},
};
use super::{
CompressedPostingList, DocSet, PostingList, RawDocInfo,
builder::ScoredDoc,
encoding::{
MAX_POSTING_BLOCK_SIZE, decode_position_stream_block, decompress_positions,
decompress_posting_block_doc_ids, decompress_posting_block_frequencies,
decompress_posting_remainder_doc_ids, decompress_posting_remainder_frequencies,
seek_packed_doc_positions,
},
query::FtsSearchParams,
scorer::Scorer,
};
use super::{DocInfo, builder::BLOCK_SIZE};
const TERMINATED_DOC_ID: u64 = u64::MAX;
type TopKHeap = BinaryHeap<Reverse<(ScoredDoc, u32, u64, u32)>>;
type NormKCache<'a> = (&'a [u8], Box<[f32; 256]>);
struct FrequencySlots {
slots: Vec<Vec<(u32, u32)>>,
}
impl FrequencySlots {
fn with_capacity(capacity: usize) -> Self {
Self {
slots: Vec::with_capacity(capacity),
}
}
fn push(&mut self, pairs: impl Iterator<Item = (u32, u32)>) -> Result<u32> {
let slot = u32::try_from(self.slots.len()).map_err(|_| {
Error::internal(format!(
"FTS top-k frequency slot count {} exceeds u32::MAX",
self.slots.len()
))
})?;
self.slots.push(pairs.collect());
Ok(slot)
}
fn replace(&mut self, slot: u32, pairs: impl Iterator<Item = (u32, u32)>) -> Result<()> {
let num_slots = self.slots.len();
let slot = self.slots.get_mut(slot as usize).ok_or_else(|| {
Error::internal(format!(
"FTS top-k frequency slot {slot} is out of bounds for {num_slots} slots"
))
})?;
slot.clear();
slot.extend(pairs);
Ok(())
}
fn take(&mut self, slot: u32) -> Result<Vec<(u32, u32)>> {
let num_slots = self.slots.len();
self.slots
.get_mut(slot as usize)
.map(std::mem::take)
.ok_or_else(|| {
Error::internal(format!(
"FTS top-k frequency slot {slot} is out of bounds for {num_slots} slots"
))
})
}
}
struct TopKCollector {
limit: usize,
heap: TopKHeap,
frequency_slots: FrequencySlots,
}
impl TopKCollector {
fn new(limit: usize, initial_capacity: usize) -> Self {
let initial_capacity = initial_capacity.min(limit);
Self {
limit,
heap: BinaryHeap::with_capacity(initial_capacity),
frequency_slots: FrequencySlots::with_capacity(initial_capacity),
}
}
fn insert(
&mut self,
doc: ScoredDoc,
doc_length: u32,
posting_doc_id: u64,
pairs: impl Iterator<Item = (u32, u32)>,
) -> Result<bool> {
if self.limit == 0 {
return Ok(false);
}
if self.heap.len() > self.limit {
return Err(Error::internal(format!(
"FTS top-k heap length {} exceeds limit {}",
self.heap.len(),
self.limit
)));
}
let frequency_slot = if self.heap.len() == self.limit {
let Some(kth_score) = self.heap.peek().map(|entry| entry.0.0.score.0) else {
return Err(Error::internal(
"FTS top-k heap is empty while its nonzero limit is reached",
));
};
if doc.score.0.partial_cmp(&kth_score) != Some(std::cmp::Ordering::Greater) {
return Ok(false);
}
let Some(Reverse((_, _, _, frequency_slot))) = self.heap.pop() else {
return Err(Error::internal(
"FTS top-k heap entry disappeared during replacement",
));
};
self.frequency_slots.replace(frequency_slot, pairs)?;
frequency_slot
} else {
self.frequency_slots.push(pairs)?
};
self.heap
.push(Reverse((doc, doc_length, posting_doc_id, frequency_slot)));
Ok(true)
}
fn kth_score_if_full(&self) -> Option<f32> {
if self.heap.len() == self.limit {
self.heap.peek().map(|entry| entry.0.0.score.0)
} else {
None
}
}
fn into_candidates<C>(
self,
mut to_candidate: impl FnMut(u64) -> C,
) -> Result<Vec<DocCandidate<C>>> {
let Self {
heap,
mut frequency_slots,
..
} = self;
heap.into_iter()
.map(
|Reverse((doc, doc_length, posting_doc_id, frequency_slot))| {
Ok(DocCandidate {
document: to_candidate(doc.row_id),
posting_doc_id,
freqs: frequency_slots.take(frequency_slot)?,
doc_length,
})
},
)
.collect()
}
#[cfg(test)]
fn num_frequency_slots(&self) -> usize {
self.frequency_slots.slots.len()
}
}
const LINEAR_BLOCK_SKIP_LIMIT: usize = 8;
pub static FLAT_SEARCH_PERCENT_THRESHOLD: LazyLock<u64> = LazyLock::new(|| {
std::env::var("LANCE_FLAT_SEARCH_PERCENT_THRESHOLD")
.unwrap_or_else(|_| "10".to_string())
.parse::<u64>()
.unwrap_or(10)
});
static USE_MAXSCORE_SEARCH: LazyLock<bool> =
LazyLock::new(|| std::env::var("LANCE_FTS_MAXSCORE").as_deref() != Ok("0"));
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
enum CompetitiveFloorMode {
#[default]
Exclusive,
Inclusive,
}
impl CompetitiveFloorMode {
#[inline]
fn rejects_upper_bound(self, upper_bound: f64, floor: f32) -> bool {
match self {
Self::Exclusive => upper_bound <= f64::from(floor),
Self::Inclusive => upper_bound < f64::from(floor),
}
}
#[inline]
fn accepts_score(self, score: f32, floor: f32) -> bool {
match self {
Self::Exclusive => score > floor,
Self::Inclusive => score >= floor,
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
enum BulkAndMode {
#[default]
Auto,
On,
Off,
}
impl BulkAndMode {
fn parse(value: &str) -> Option<Self> {
let value = value.trim();
if value.eq_ignore_ascii_case("auto") {
Some(Self::Auto)
} else if value.eq_ignore_ascii_case("on") || value == "1" {
Some(Self::On)
} else if value.eq_ignore_ascii_case("off") || value == "0" {
Some(Self::Off)
} else {
None
}
}
const fn enabled_for(self, num_clauses: usize) -> bool {
match self {
Self::Auto => matches!(num_clauses, 2 | 3),
Self::On => true,
Self::Off => false,
}
}
}
fn bulk_and_mode_from_env() -> BulkAndMode {
match std::env::var("LANCE_FTS_BULK_AND") {
Ok(value) => BulkAndMode::parse(&value).unwrap_or_else(|| {
log::warn!(
"Invalid LANCE_FTS_BULK_AND value {value:?}; expected auto, on/1, or off/0; \
falling back to auto"
);
BulkAndMode::Auto
}),
Err(std::env::VarError::NotPresent) => BulkAndMode::Auto,
Err(std::env::VarError::NotUnicode(value)) => {
log::warn!(
"Invalid non-Unicode LANCE_FTS_BULK_AND value {value:?}; expected auto, on/1, \
or off/0; falling back to auto"
);
BulkAndMode::Auto
}
}
}
static BULK_AND_MODE: LazyLock<BulkAndMode> = LazyLock::new(bulk_and_mode_from_env);
#[cfg(target_arch = "x86_64")]
static HAS_AVX2: LazyLock<bool> = LazyLock::new(|| std::arch::is_x86_feature_detected!("avx2"));
#[inline]
unsafe fn find_next_geq_scalar(docs: *const u32, mut pos: usize, end: usize, target: u32) -> usize {
unsafe {
while pos < end && *docs.add(pos) < target {
pos += 1;
}
}
pos
}
#[cfg(target_arch = "x86_64")]
#[target_feature(enable = "avx2")]
unsafe fn find_next_geq_avx2(docs: *const u32, mut pos: usize, end: usize, target: u32) -> usize {
use core::arch::x86_64::*;
unsafe {
let sign_bit = _mm256_set1_epi32(i32::MIN);
let target_lanes = _mm256_xor_si256(_mm256_set1_epi32(target as i32), sign_bit);
while pos + 8 < end {
if *docs.add(pos + 8) < target {
pos += 9;
continue;
}
let docs_lanes = _mm256_xor_si256(
_mm256_loadu_si256(docs.add(pos) as *const __m256i),
sign_bit,
);
let below = _mm256_cmpgt_epi32(target_lanes, docs_lanes);
let mask = _mm256_movemask_ps(_mm256_castsi256_ps(below)) as u32;
if mask != 0xFF {
return pos + mask.trailing_ones() as usize;
}
return pos + 8;
}
find_next_geq_scalar(docs, pos, end, target)
}
}
#[inline]
unsafe fn find_next_geq(docs: *const u32, pos: usize, end: usize, target: u32) -> usize {
#[cfg(target_arch = "x86_64")]
if *HAS_AVX2 {
return unsafe { find_next_geq_avx2(docs, pos, end, target) };
}
unsafe { find_next_geq_scalar(docs, pos, end, target) }
}
#[inline]
fn find_next_geq_in_block(docs: &[u32], pos: usize, target: u32) -> usize {
debug_assert!(pos <= docs.len());
#[cfg(target_arch = "x86_64")]
if *HAS_AVX2 {
return unsafe { find_next_geq_avx2(docs.as_ptr(), pos, docs.len(), target) };
}
pos + docs[pos..].partition_point(|&doc_id| doc_id < target)
}
#[inline]
fn conservative_bm25_upper_bound(query_weight: f32) -> f32 {
if query_weight <= 0.0 {
0.0
} else {
query_weight * BM25_DOC_WEIGHT_UPPER_BOUND
}
}
#[inline]
fn scorer_upper_bound<S: Scorer + ?Sized>(query_weight: f32, scorer: &S) -> f32 {
if query_weight.is_nan() {
return f32::INFINITY;
}
if query_weight <= 0.0 {
return 0.0;
}
match scorer.doc_weight_upper_bound() {
Some(bound) if bound.is_finite() && bound >= 0.0 => query_weight * bound,
_ => f32::INFINITY,
}
}
#[derive(Debug)]
pub(super) struct GroupedTermScorer {
query_weight: f32,
freqs_by_posting_doc_id: Vec<(u64, u32)>,
}
impl GroupedTermScorer {
pub(super) fn new(query_weight: f32, posting: &PostingList) -> Self {
let freqs_by_posting_doc_id = posting
.iter()
.map(|(posting_doc_id, freq, _)| (posting_doc_id, freq))
.collect();
Self {
query_weight,
freqs_by_posting_doc_id,
}
}
pub(super) fn query_weight(&self) -> f32 {
self.query_weight
}
pub(super) fn frequency(&self, posting_doc_id: u64) -> Option<u32> {
self.freqs_by_posting_doc_id
.binary_search_by_key(&posting_doc_id, |(doc_id, _)| *doc_id)
.ok()
.map(|index| self.freqs_by_posting_doc_id[index].1)
}
fn score<S: Scorer + ?Sized>(&self, posting_doc_id: u64, doc_length: u32, scorer: &S) -> f32 {
self.frequency(posting_doc_id)
.map(|freq| self.query_weight * scorer.doc_weight(freq, doc_length))
.unwrap_or_default()
}
}
pub struct PostingIterator {
token: String,
token_id: u32,
position: u32,
query_weight: f32,
list: PostingList,
index: usize,
block_idx: usize,
current_doc: Option<DocInfo>,
approximate_upper_bound: f32,
use_scorer_upper_bound: bool,
grouped_terms: Option<Arc<[GroupedTermScorer]>>,
position_scratch: RefCell<Option<Vec<u32>>>,
compressed: Option<UnsafeCell<CompressedState>>,
}
#[derive(Clone)]
struct CompressedState {
block_idx: usize,
doc_ids: Vec<u32>,
freqs: Vec<u32>,
frequency_offset: usize,
frequency_block_idx: Option<usize>,
#[cfg(test)]
frequency_blocks_decoded: usize,
#[cfg(test)]
impact_bound_computations: usize,
buffer: Box<[u32; MAX_POSTING_BLOCK_SIZE]>,
position_block_idx: Option<usize>,
position_values: Vec<u32>,
position_offsets: Vec<usize>,
position_group_offsets: Vec<usize>,
position_unpacked_group: Box<[u32; BLOCK_SIZE]>,
position_unpacked_group_idx: Option<usize>,
position_tail: Vec<u32>,
position_total_deltas: usize,
block_max_window: BlockMaxWindow,
level0_cache: Option<(usize, u32, f32)>,
level1_cache: Option<(usize, u32, f32)>,
}
impl CompressedState {
fn new(block_size: usize) -> Self {
Self {
block_idx: 0,
doc_ids: Vec::with_capacity(block_size),
freqs: Vec::with_capacity(block_size),
frequency_offset: 0,
frequency_block_idx: None,
#[cfg(test)]
frequency_blocks_decoded: 0,
#[cfg(test)]
impact_bound_computations: 0,
buffer: Box::new([0; MAX_POSTING_BLOCK_SIZE]),
position_block_idx: None,
position_values: Vec::new(),
position_offsets: Vec::new(),
position_group_offsets: Vec::new(),
position_unpacked_group: Box::new([0; BLOCK_SIZE]),
position_unpacked_group_idx: None,
position_tail: Vec::new(),
position_total_deltas: 0,
block_max_window: BlockMaxWindow::new(),
level0_cache: None,
level1_cache: None,
}
}
#[inline]
fn decompress_doc_ids(
&mut self,
block: &[u8],
block_idx: usize,
num_blocks: usize,
length: u32,
tail_codec: super::PostingTailCodec,
block_size: usize,
) {
self.doc_ids.clear();
self.freqs.clear();
self.frequency_block_idx = None;
let remainder = length as usize % block_size;
self.frequency_offset = if block_idx + 1 == num_blocks && remainder != 0 {
decompress_posting_remainder_doc_ids(
block,
remainder,
tail_codec,
block_size,
&mut self.doc_ids,
)
} else {
decompress_posting_block_doc_ids(
block,
&mut self.buffer[..],
&mut self.doc_ids,
block_size,
)
};
self.block_idx = block_idx;
self.position_block_idx = None;
self.position_values.clear();
self.position_offsets.clear();
}
#[inline]
fn decompress_frequencies(
&mut self,
block: &[u8],
block_idx: usize,
num_blocks: usize,
length: u32,
tail_codec: super::PostingTailCodec,
block_size: usize,
) {
debug_assert_eq!(self.block_idx, block_idx);
if self.frequency_block_idx == Some(block_idx) {
return;
}
self.freqs.clear();
let remainder = length as usize % block_size;
if block_idx + 1 == num_blocks && remainder != 0 {
decompress_posting_remainder_frequencies(
block,
self.frequency_offset,
remainder,
tail_codec,
&mut self.freqs,
);
} else {
decompress_posting_block_frequencies(
block,
self.frequency_offset,
&mut self.buffer[..],
&mut self.freqs,
block_size,
);
}
self.frequency_block_idx = Some(block_idx);
#[cfg(test)]
{
self.frequency_blocks_decoded += 1;
}
}
}
#[derive(Clone)]
struct BlockMaxWindow {
start_block_idx: usize,
next_block_idx: usize,
max_scores: VecDeque<(usize, f32)>,
impact_score_cache: ImpactScoreCache,
}
struct BlockMaxScore {
score: f32,
#[cfg(test)]
blocks_scanned: usize,
}
impl BlockMaxWindow {
fn new() -> Self {
Self {
start_block_idx: 0,
next_block_idx: 0,
max_scores: VecDeque::new(),
impact_score_cache: ImpactScoreCache::default(),
}
}
fn reset(&mut self, start_block_idx: usize) {
self.start_block_idx = start_block_idx;
self.next_block_idx = start_block_idx;
self.max_scores.clear();
}
fn max_score_up_to<S: Scorer + ?Sized>(
&mut self,
list: &CompressedPostingList,
start_block_idx: usize,
up_to: u64,
query_weight: f32,
scorer: &S,
) -> BlockMaxScore {
if start_block_idx >= list.blocks.len() {
self.reset(start_block_idx);
return BlockMaxScore {
score: 0.0,
#[cfg(test)]
blocks_scanned: 0,
};
}
if start_block_idx < self.start_block_idx || start_block_idx > self.next_block_idx {
self.reset(start_block_idx);
}
self.start_block_idx = start_block_idx;
while matches!(self.max_scores.front(), Some((block_idx, _)) if *block_idx < start_block_idx)
{
self.max_scores.pop_front();
}
if list.block_least_doc_id(start_block_idx) as u64 > up_to {
self.reset(start_block_idx);
return BlockMaxScore {
score: 0.0,
#[cfg(test)]
blocks_scanned: 0,
};
}
if list.block_size == MAX_POSTING_BLOCK_SIZE {
self.reset(start_block_idx);
return BlockMaxScore {
score: scorer_upper_bound(query_weight, scorer),
#[cfg(test)]
blocks_scanned: 0,
};
}
self.next_block_idx = self.next_block_idx.max(start_block_idx);
#[cfg(test)]
let mut blocks_scanned = 0;
while self.next_block_idx < list.blocks.len()
&& list.block_least_doc_id(self.next_block_idx) as u64 <= up_to
{
let score = match list.impacts.as_ref() {
Some(impacts) => impacts.level0_score_cached(
self.next_block_idx,
query_weight,
scorer,
&mut self.impact_score_cache,
),
None => list.block_max_score(self.next_block_idx),
};
while matches!(self.max_scores.back(), Some((_, old_score)) if *old_score <= score) {
self.max_scores.pop_back();
}
self.max_scores.push_back((self.next_block_idx, score));
self.next_block_idx += 1;
#[cfg(test)]
{
blocks_scanned += 1;
}
}
let score = self
.max_scores
.front()
.map(|(_, score)| *score)
.unwrap_or(0.0);
BlockMaxScore {
score,
#[cfg(test)]
blocks_scanned,
}
}
}
impl Debug for PostingIterator {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PostingIterator")
.field(
"doc",
&self
.doc()
.map(|doc| doc.doc_id())
.unwrap_or(TERMINATED_DOC_ID),
)
.field("approximate_upper_bound", &self.approximate_upper_bound)
.field("token_id", &self.token_id)
.finish()
}
}
impl PartialEq for PostingIterator {
fn eq(&self, other: &Self) -> bool {
self.token_id == other.token_id && self.position == other.position
}
}
impl Eq for PostingIterator {}
impl PartialOrd for PostingIterator {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for PostingIterator {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
match (self.doc(), other.doc()) {
(Some(doc1), Some(doc2)) => doc1
.cmp(&doc2)
.then(
self.approximate_upper_bound
.total_cmp(&other.approximate_upper_bound),
)
.then(self.token_id.cmp(&other.token_id))
.then(self.position.cmp(&other.position)),
(Some(_), None) => std::cmp::Ordering::Less,
(None, Some(_)) => std::cmp::Ordering::Greater,
(None, None) => self
.approximate_upper_bound
.total_cmp(&other.approximate_upper_bound)
.then(self.token_id.cmp(&other.token_id))
.then(self.position.cmp(&other.position)),
}
}
}
impl PostingIterator {
fn block_idx_for_doc(
&self,
list: &CompressedPostingList,
mut block_idx: usize,
least_id: u32,
) -> usize {
let mut linear_skips = 0;
while block_idx + 1 < list.blocks.len() && linear_skips < LINEAR_BLOCK_SKIP_LIMIT {
if list.block_least_doc_id(block_idx + 1) > least_id {
return block_idx;
}
block_idx += 1;
linear_skips += 1;
}
if block_idx + 1 >= list.blocks.len() {
return block_idx;
}
if let Some(impacts) = list.impacts.as_ref()
&& let Some(block_idx) =
self.block_idx_for_doc_with_impacts(list, impacts, block_idx, least_id)
{
return block_idx;
}
self.block_idx_for_doc_by_least_doc_id(list, block_idx, least_id, list.blocks.len())
}
fn block_idx_for_doc_with_impacts(
&self,
list: &CompressedPostingList,
impacts: &ImpactSkipData,
mut block_idx: usize,
least_id: u32,
) -> Option<usize> {
while block_idx + 1 < list.blocks.len() {
let group_idx = (block_idx + 1) / IMPACT_LEVEL1_BLOCKS;
let group_end = ((group_idx + 1) * IMPACT_LEVEL1_BLOCKS).min(list.blocks.len());
let group_doc_up_to = impacts.level1_doc_up_to(group_idx)?;
if group_doc_up_to < least_id {
block_idx = group_end - 1;
continue;
}
if group_doc_up_to == least_id {
return Some(group_end - 1);
}
return Some(
self.block_idx_for_doc_by_least_doc_id(list, block_idx, least_id, group_end),
);
}
Some(block_idx)
}
fn block_idx_for_doc_by_least_doc_id(
&self,
list: &CompressedPostingList,
block_idx: usize,
least_id: u32,
right: usize,
) -> usize {
let mut left = block_idx + 1;
let mut right = right;
while left < right {
let mid = left + (right - left) / 2;
if list.block_least_doc_id(mid) <= least_id {
left = mid + 1;
} else {
right = mid;
}
}
left - 1
}
#[inline]
fn block_end_doc(&self) -> u64 {
self.next_block_first_doc()
.map(|doc| doc.saturating_sub(1))
.unwrap_or(TERMINATED_DOC_ID)
}
fn impact_group_bound<S: Scorer + ?Sized>(&self, scorer: &S) -> Option<(u64, f32)> {
match self.list {
PostingList::Compressed(ref list) => {
let impacts = list.impacts.as_ref()?;
let (doc_up_to, score) = self.impact_level1(impacts, scorer);
if doc_up_to == u32::MAX {
return None;
}
Some((u64::from(doc_up_to), score))
}
PostingList::Plain(_) => None,
}
}
#[inline]
fn compressed_state_ptr(&self) -> *mut CompressedState {
debug_assert!(self.compressed.is_some());
self.compressed.as_ref().unwrap().get()
}
#[inline]
fn ensure_compressed_block_ptr(
&self,
list: &CompressedPostingList,
block_idx: usize,
) -> *mut CompressedState {
let compressed = unsafe { &mut *self.ensure_compressed_doc_ids_ptr(list, block_idx) };
if compressed.frequency_block_idx != Some(block_idx) {
let block = list.blocks.value(block_idx);
compressed.decompress_frequencies(
block,
block_idx,
list.blocks.len(),
list.length,
list.posting_tail_codec,
list.block_size,
);
}
compressed as *mut CompressedState
}
#[inline]
fn ensure_compressed_doc_ids_ptr(
&self,
list: &CompressedPostingList,
block_idx: usize,
) -> *mut CompressedState {
let compressed = unsafe { &mut *self.compressed_state_ptr() };
if compressed.block_idx != block_idx || compressed.doc_ids.is_empty() {
let block = list.blocks.value(block_idx);
compressed.decompress_doc_ids(
block,
block_idx,
list.blocks.len(),
list.length,
list.posting_tail_codec,
list.block_size,
);
}
compressed as *mut CompressedState
}
#[cfg(test)]
pub(crate) fn new(
token: String,
token_id: u32,
position: u32,
list: PostingList,
num_doc: usize,
) -> Self {
Self::with_query_weight(token, token_id, position, 1.0, list, num_doc)
}
pub(crate) fn with_query_weight(
token: String,
token_id: u32,
position: u32,
query_weight: f32,
list: PostingList,
num_doc: usize,
) -> Self {
let approximate_upper_bound = match &list {
PostingList::Compressed(posting) if posting.impacts.is_some() => f32::INFINITY,
PostingList::Compressed(posting) if posting.block_size == MAX_POSTING_BLOCK_SIZE => {
conservative_bm25_upper_bound(query_weight)
}
_ => match list.max_score() {
Some(max_score) => max_score,
None => conservative_bm25_upper_bound(idf(list.len(), num_doc)),
},
};
let compressed = match &list {
PostingList::Compressed(list) => {
Some(UnsafeCell::new(CompressedState::new(list.block_size)))
}
PostingList::Plain(_) => None,
};
let mut posting = Self {
token,
token_id,
position,
query_weight,
list,
index: 0,
block_idx: 0,
current_doc: None,
approximate_upper_bound,
use_scorer_upper_bound: false,
grouped_terms: None,
position_scratch: RefCell::new(Some(Vec::new())),
compressed,
};
posting.refresh_current_doc();
posting
}
pub(super) fn with_scorer_upper_bound(mut self) -> Self {
self.use_scorer_upper_bound = true;
self
}
pub(super) fn with_grouped_terms(mut self, terms: Arc<[GroupedTermScorer]>) -> Self {
self.approximate_upper_bound = self.list.max_score().unwrap_or(f32::INFINITY);
self.grouped_terms = Some(terms);
self
}
pub(super) fn fork_from_start(&self) -> Self {
let compressed = match &self.list {
PostingList::Compressed(list) => {
Some(UnsafeCell::new(CompressedState::new(list.block_size)))
}
PostingList::Plain(_) => None,
};
let mut posting = Self {
token: self.token.clone(),
token_id: self.token_id,
position: self.position,
query_weight: self.query_weight,
list: self.list.clone(),
index: 0,
block_idx: 0,
current_doc: None,
approximate_upper_bound: self.approximate_upper_bound,
use_scorer_upper_bound: self.use_scorer_upper_bound,
grouped_terms: self.grouped_terms.clone(),
position_scratch: RefCell::new(Some(Vec::new())),
compressed,
};
posting.refresh_current_doc();
posting
}
#[inline]
pub(crate) fn term_index(&self) -> u32 {
self.position
}
#[inline]
pub(crate) fn token(&self) -> &str {
&self.token
}
#[inline]
fn approximate_upper_bound(&self) -> f32 {
self.approximate_upper_bound
}
#[inline]
fn frequency_clamp_upper_bound<S: Scorer + ?Sized>(&self, scorer: &S) -> f32 {
if self.use_scorer_upper_bound {
scorer_upper_bound(self.query_weight, scorer)
} else {
self.approximate_upper_bound
}
}
#[inline]
fn global_upper_bound<S: Scorer + ?Sized>(&self, scorer: &S) -> f32 {
if self.has_grouped_terms() {
return self.approximate_upper_bound;
}
if self.query_weight <= 0.0 {
return 0.0;
}
if let PostingList::Compressed(ref list) = self.list
&& let Some(impacts) = list.impacts.as_ref()
{
let compressed = unsafe { &mut *self.compressed_state_ptr() };
return self.query_weight
* impacts.global_max_doc_weight_cached(
scorer,
&mut compressed.block_max_window.impact_score_cache,
);
}
if self.use_scorer_upper_bound {
return scorer_upper_bound(self.query_weight, scorer);
}
if let PostingList::Compressed(ref list) = self.list
&& list.block_size == MAX_POSTING_BLOCK_SIZE
{
return scorer_upper_bound(self.query_weight, scorer);
}
self.approximate_upper_bound
}
#[inline]
fn score<S: Scorer + ?Sized>(&self, scorer: &S, freq: u32, doc_length: u32) -> f32 {
if let (Some(grouped_terms), Some(doc)) = (&self.grouped_terms, self.current_doc) {
return grouped_terms
.iter()
.map(|term| term.score(doc.doc_id(), doc_length, scorer))
.sum();
}
self.query_weight * scorer.doc_weight(freq, doc_length)
}
#[inline]
fn has_grouped_terms(&self) -> bool {
self.grouped_terms.is_some()
}
#[inline]
fn cost(&self) -> usize {
self.list.len()
}
#[inline]
fn empty(&self) -> bool {
self.index >= self.list.len()
}
#[inline]
fn doc(&self) -> Option<DocInfo> {
let current_doc = self.current_doc?;
match self.list {
PostingList::Compressed(ref list) => {
if current_doc.frequency() != 0 {
return Some(current_doc);
}
let block_idx = self.index >> list.block_shift();
let block_offset = self.index & list.block_mask();
let compressed = unsafe { &mut *self.ensure_compressed_block_ptr(list, block_idx) };
Some(DocInfo::Raw(RawDocInfo {
doc_id: current_doc.doc_id() as u32,
frequency: compressed.freqs[block_offset],
}))
}
PostingList::Plain(_) => Some(current_doc),
}
}
#[cfg(test)]
fn frequency_blocks_decoded(&self) -> usize {
self.compressed
.as_ref()
.map(|compressed| unsafe { &*compressed.get() }.frequency_blocks_decoded)
.unwrap_or_default()
}
#[cfg(test)]
fn impact_bound_computations(&self) -> usize {
self.compressed
.as_ref()
.map(|compressed| unsafe { &*compressed.get() }.impact_bound_computations)
.unwrap_or_default()
}
pub(super) fn current_doc_id(&self) -> Option<u64> {
self.current_doc.map(|doc| doc.doc_id())
}
fn refresh_current_doc(&mut self) {
if self.empty() {
self.current_doc = None;
return;
}
let current_doc = match self.list {
PostingList::Compressed(ref list) => {
let block_idx = self.index >> list.block_shift();
let block_offset = self.index & list.block_mask();
let compressed =
unsafe { &mut *self.ensure_compressed_doc_ids_ptr(list, block_idx) };
let doc_id = compressed.doc_ids[block_offset];
let doc = DocInfo::Raw(RawDocInfo {
doc_id,
frequency: 0,
});
Some(doc)
}
PostingList::Plain(ref list) => Some(DocInfo::Located(list.doc(self.index))),
};
self.current_doc = current_doc;
}
fn position_cursor(&self) -> Result<PositionCursor<'_>> {
match self.list {
PostingList::Plain(ref list) => {
let positions = list.positions.as_ref().ok_or_else(|| {
Error::index(format!(
"positions are missing for token {:?} (token id {}, query position {})",
self.token, self.token_id, self.position
))
})?;
let start = positions.value_offsets()[self.index] as usize;
let end = positions.value_offsets()[self.index + 1] as usize;
Ok(PositionCursor::new(
PositionValues::Owned(
positions.values().as_primitive::<Int32Type>().values()[start..end]
.iter()
.map(|value| *value as u32)
.collect(),
),
self.position as i32,
))
}
PostingList::Compressed(ref list) => match list.positions.as_ref().ok_or_else(|| {
Error::index(format!(
"positions are missing for token {:?} (token id {}, query position {})",
self.token, self.token_id, self.position
))
})? {
CompressedPositionStorage::LegacyPerDoc(positions) => {
let positions = positions.value(self.index);
let positions = decompress_positions(positions.as_binary());
Ok(PositionCursor::new(
PositionValues::Owned(positions),
self.position as i32,
))
}
CompressedPositionStorage::SharedStream(stream) => {
let block_idx = self.index >> list.block_shift();
let block_offset = self.index & list.block_mask();
let compressed =
unsafe { &mut *self.ensure_compressed_block_ptr(list, block_idx) };
match stream.codec() {
PositionStreamCodec::PackedDelta => {
if compressed.position_block_idx != Some(block_idx) {
compressed.position_group_offsets.clear();
compressed.position_group_offsets.push(0);
compressed.position_tail.clear();
compressed.position_unpacked_group_idx = None;
compressed.position_offsets.clear();
compressed
.position_offsets
.reserve(compressed.freqs.len() + 1);
compressed.position_offsets.push(0);
let mut offset = 0usize;
for &freq in &compressed.freqs {
offset += freq as usize;
compressed.position_offsets.push(offset);
}
compressed.position_total_deltas = offset;
compressed.position_block_idx = Some(block_idx);
}
let delta_start = compressed.position_offsets[block_offset];
let delta_end = compressed.position_offsets[block_offset + 1];
let mut position_values = self
.position_scratch
.borrow_mut()
.take()
.unwrap_or_default();
if let Err(error) = seek_packed_doc_positions(
stream.block(block_idx),
compressed.position_total_deltas,
delta_start..delta_end,
&mut compressed.position_group_offsets,
&mut compressed.position_unpacked_group,
&mut compressed.position_unpacked_group_idx,
&mut compressed.position_tail,
&mut position_values,
) {
*self.position_scratch.borrow_mut() = Some(position_values);
return Err(Error::index(format!(
"failed to decode positions for token {:?} (token id {}, query position {}) at posting index {}: {error}",
self.token, self.token_id, self.position, self.index
)));
}
Ok(PositionCursor::new(
PositionValues::Recycled(RecycledPositionValues::new(
position_values,
&self.position_scratch,
)),
self.position as i32,
))
}
PositionStreamCodec::VarintDocDelta => {
if compressed.position_block_idx != Some(block_idx) {
compressed.position_values.clear();
decode_position_stream_block(
stream.block(block_idx),
compressed.freqs.as_slice(),
stream.codec(),
&mut compressed.position_values,
)
.map_err(|error| {
Error::index(format!(
"failed to decode positions for token {:?} (token id {}, query position {}) in block {block_idx}: {error}",
self.token, self.token_id, self.position
))
})?;
compressed.position_offsets.clear();
compressed
.position_offsets
.reserve(compressed.freqs.len() + 1);
compressed.position_offsets.push(0);
let mut offset = 0usize;
for &freq in &compressed.freqs {
offset += freq as usize;
compressed.position_offsets.push(offset);
}
compressed.position_block_idx = Some(block_idx);
}
let start = compressed.position_offsets[block_offset];
let end = compressed.position_offsets[block_offset + 1];
let mut position_values = self
.position_scratch
.borrow_mut()
.take()
.unwrap_or_default();
position_values.clear();
position_values
.extend_from_slice(&compressed.position_values[start..end]);
Ok(PositionCursor::new(
PositionValues::Recycled(RecycledPositionValues::new(
position_values,
&self.position_scratch,
)),
self.position as i32,
))
}
}
}
},
}
}
fn next(&mut self, least_id: u64) {
match self.list {
PostingList::Compressed(ref list) => {
debug_assert!(least_id <= u32::MAX as u64);
let least_id = least_id as u32;
let shift = list.block_shift();
let block_idx = self.block_idx_for_doc(list, self.index >> shift, least_id);
self.index = self.index.max(block_idx << shift);
let length = list.length as usize;
while self.index < length {
let block_idx = self.index >> shift;
let block_offset = self.index & list.block_mask();
let compressed =
unsafe { &mut *self.ensure_compressed_block_ptr(list, block_idx) };
let in_block = &compressed.doc_ids[block_offset..];
let offset_in_block = in_block.partition_point(|&doc_id| doc_id < least_id);
let new_offset = block_offset + offset_in_block;
if new_offset < compressed.doc_ids.len() {
self.index = (block_idx << shift) + new_offset;
self.block_idx = block_idx;
self.current_doc = Some(DocInfo::Raw(RawDocInfo {
doc_id: compressed.doc_ids[new_offset],
frequency: compressed.freqs[new_offset],
}));
return;
}
if block_idx + 1 >= list.blocks.len() {
self.index = length;
self.block_idx = self.index >> shift;
self.current_doc = None;
break;
}
self.index = (block_idx + 1) << shift;
}
self.block_idx = self.index >> shift;
self.current_doc = None;
}
PostingList::Plain(ref list) => {
self.index += list.row_ids[self.index..].partition_point(|&id| id < least_id);
self.current_doc = (!self.empty()).then(|| DocInfo::Located(list.doc(self.index)));
}
}
}
fn next_doc_id(&mut self, least_id: u64, is_vectorized_search_enabled: bool) {
match self.list {
PostingList::Compressed(ref list) => {
debug_assert!(least_id <= u32::MAX as u64);
let least_id = least_id as u32;
let shift = list.block_shift();
let block_idx = self.block_idx_for_doc(list, self.index >> shift, least_id);
self.index = self.index.max(block_idx << shift);
let length = list.length as usize;
while self.index < length {
let block_idx = self.index >> shift;
let block_offset = self.index & list.block_mask();
let compressed =
unsafe { &mut *self.ensure_compressed_doc_ids_ptr(list, block_idx) };
let new_offset = if is_vectorized_search_enabled {
find_next_geq_in_block(&compressed.doc_ids, block_offset, least_id)
} else {
block_offset
+ compressed.doc_ids[block_offset..]
.partition_point(|&doc_id| doc_id < least_id)
};
if new_offset < compressed.doc_ids.len() {
self.index = (block_idx << shift) + new_offset;
self.block_idx = block_idx;
self.current_doc = Some(DocInfo::Raw(RawDocInfo {
doc_id: compressed.doc_ids[new_offset],
frequency: 0,
}));
return;
}
if block_idx + 1 >= list.blocks.len() {
self.index = length;
self.block_idx = self.index >> shift;
self.current_doc = None;
break;
}
self.index = (block_idx + 1) << shift;
}
self.block_idx = self.index >> shift;
self.current_doc = None;
}
PostingList::Plain(ref list) => {
self.index += list.row_ids[self.index..].partition_point(|&id| id < least_id);
self.current_doc = (!self.empty()).then(|| DocInfo::Located(list.doc(self.index)));
}
}
}
fn shallow_next(&mut self, least_id: u64) {
match self.list {
PostingList::Compressed(ref list) => {
debug_assert!(least_id <= u32::MAX as u64);
let least_id = least_id as u32;
self.block_idx = self.block_idx_for_doc(list, self.block_idx, least_id);
}
PostingList::Plain(_) => {
}
}
}
#[inline]
fn impact_level0<S: Scorer + ?Sized>(
&self,
impacts: &ImpactSkipData,
scorer: &S,
) -> (u32, f32) {
let compressed = unsafe { &mut *self.compressed_state_ptr() };
if let Some((block_idx, doc_up_to, score)) = compressed.level0_cache
&& block_idx == self.block_idx
{
return (doc_up_to, score);
}
let doc_up_to = impacts.level0_doc_up_to(self.block_idx).unwrap_or(u32::MAX);
let score = impacts.level0_score_cached(
self.block_idx,
self.query_weight,
scorer,
&mut compressed.block_max_window.impact_score_cache,
);
#[cfg(test)]
{
compressed.impact_bound_computations += 1;
}
compressed.level0_cache = Some((self.block_idx, doc_up_to, score));
(doc_up_to, score)
}
#[inline]
fn impact_level1<S: Scorer + ?Sized>(
&self,
impacts: &ImpactSkipData,
scorer: &S,
) -> (u32, f32) {
let group_idx = self.block_idx / IMPACT_LEVEL1_BLOCKS;
let compressed = unsafe { &mut *self.compressed_state_ptr() };
if let Some((cached_group_idx, doc_up_to, score)) = compressed.level1_cache
&& cached_group_idx == group_idx
{
return (doc_up_to, score);
}
let doc_up_to = impacts.level1_doc_up_to(group_idx).unwrap_or(u32::MAX);
let score = impacts.level1_score_cached(
group_idx,
self.query_weight,
scorer,
&mut compressed.block_max_window.impact_score_cache,
);
#[cfg(test)]
{
compressed.impact_bound_computations += 1;
}
compressed.level1_cache = Some((group_idx, doc_up_to, score));
(doc_up_to, score)
}
#[inline]
fn block_max_score<S: Scorer + ?Sized>(&self, scorer: &S) -> f32 {
match self.list {
PostingList::Compressed(ref list) => {
if self.has_grouped_terms() && list.block_size == MAX_POSTING_BLOCK_SIZE {
return self.approximate_upper_bound;
}
if let Some(impacts) = list.impacts.as_ref() {
return self.impact_level0(impacts, scorer).1;
}
if self.use_scorer_upper_bound {
return scorer_upper_bound(self.query_weight, scorer);
}
if list.block_size == MAX_POSTING_BLOCK_SIZE {
return scorer_upper_bound(self.query_weight, scorer);
}
list.block_max_score(self.block_idx)
}
PostingList::Plain(_) if self.use_scorer_upper_bound => {
scorer_upper_bound(self.query_weight, scorer)
}
PostingList::Plain(_) => self.approximate_upper_bound,
}
}
#[inline]
fn block_max_score_up_to_with_stats<S: Scorer + ?Sized>(
&self,
up_to: u64,
scorer: &S,
) -> BlockMaxScore {
match self.list {
PostingList::Compressed(ref list) => {
if self.has_grouped_terms() && list.block_size == MAX_POSTING_BLOCK_SIZE {
return BlockMaxScore {
score: self.approximate_upper_bound,
#[cfg(test)]
blocks_scanned: 0,
};
}
if let Some(impacts) = list.impacts.as_ref() {
let (level0_up_to, level0_score) = self.impact_level0(impacts, scorer);
if up_to <= u64::from(level0_up_to) {
return BlockMaxScore {
score: level0_score,
#[cfg(test)]
blocks_scanned: 0,
};
}
}
if self.use_scorer_upper_bound {
return BlockMaxScore {
score: scorer_upper_bound(self.query_weight, scorer),
#[cfg(test)]
blocks_scanned: 0,
};
}
let compressed = unsafe { &mut *self.compressed_state_ptr() };
compressed.block_max_window.max_score_up_to(
list,
self.block_idx,
up_to,
self.query_weight,
scorer,
)
}
PostingList::Plain(_) => BlockMaxScore {
score: if self.use_scorer_upper_bound {
scorer_upper_bound(self.query_weight, scorer)
} else {
self.approximate_upper_bound
},
#[cfg(test)]
blocks_scanned: 0,
},
}
}
fn window_max_score<S: Scorer + ?Sized>(&self, up_to: Option<u64>, scorer: &S) -> f32 {
if let Some(up_to) = up_to
&& let PostingList::Compressed(ref list) = self.list
&& list.impacts.is_some()
{
return self.block_max_score_up_to_with_stats(up_to, scorer).score;
}
self.block_max_score(scorer)
}
#[inline]
fn is_compressed(&self) -> bool {
matches!(self.list, PostingList::Compressed(_))
}
#[inline]
fn has_next_compressed_block(&self) -> bool {
match self.list {
PostingList::Compressed(ref list) => self.block_idx + 1 < list.blocks.len(),
PostingList::Plain(_) => false,
}
}
fn block_first_doc(&self) -> Option<u64> {
match self.list {
PostingList::Compressed(ref list) => {
Some(list.block_least_doc_id(self.block_idx) as u64)
}
PostingList::Plain(ref plain) => plain.row_ids.get(self.index).cloned(),
}
}
#[allow(clippy::too_many_arguments)]
fn collect_window_scores<S: Scorer + ?Sized, D: WandDocuments>(
&mut self,
window_min: u64,
up_to: u64,
clause_idx: usize,
docs: &D,
scorer: &S,
norm_k: Option<(&[u8], &[f32; 256])>,
acc: &mut WindowAccumulator,
) {
if self.doc().is_some_and(|doc| doc.doc_id() < window_min) {
self.next(window_min);
}
match self.list {
PostingList::Compressed(ref list) => {
let shift = list.block_shift();
let mask = list.block_mask();
'blocks: while let Some(doc) = self.current_doc {
if doc.doc_id() > up_to {
break;
}
let block_idx = self.index >> shift;
let block_offset = self.index & mask;
let compressed =
unsafe { &mut *self.ensure_compressed_block_ptr(list, block_idx) };
for offset in block_offset..compressed.doc_ids.len() {
let doc_id = compressed.doc_ids[offset];
if u64::from(doc_id) > up_to {
self.index = (block_idx << shift) + offset;
self.block_idx = block_idx;
self.current_doc = Some(DocInfo::Raw(RawDocInfo {
doc_id,
frequency: compressed.freqs[offset],
}));
break 'blocks;
}
let freq = compressed.freqs[offset];
let doc_weight = match norm_k {
Some((norms, cache)) => bm25_doc_weight_with_norm(
freq,
cache[norms[doc_id as usize] as usize],
),
None => scorer.doc_weight(freq, docs.scoring_num_tokens(doc_id)),
};
let score = self.query_weight * doc_weight;
let slot = (u64::from(doc_id) - window_min) as usize;
acc.add(clause_idx, slot, score, freq);
}
let next_start = (block_idx + 1) << shift;
if next_start >= list.length as usize {
self.index = list.length as usize;
self.block_idx = self.index >> shift;
self.current_doc = None;
break;
}
self.index = next_start;
self.block_idx = block_idx + 1;
let compressed =
unsafe { &mut *self.ensure_compressed_block_ptr(list, block_idx + 1) };
self.current_doc = Some(DocInfo::Raw(RawDocInfo {
doc_id: compressed.doc_ids[0],
frequency: compressed.freqs[0],
}));
}
}
PostingList::Plain(_) => {
while let Some(doc) = self.doc() {
let doc_id = doc.doc_id();
if doc_id > up_to {
break;
}
let doc_length = docs.doc_length(&doc);
let score = self.score(scorer, doc.frequency(), doc_length);
let slot = (doc_id - window_min) as usize;
acc.add(clause_idx, slot, score, doc.frequency());
self.next(doc_id + 1);
}
}
}
}
#[inline]
fn next_block_first_doc(&self) -> Option<u64> {
match self.list {
PostingList::Compressed(ref list) => {
if self.block_idx + 1 >= list.blocks.len() {
return None;
}
Some(list.block_least_doc_id(self.block_idx + 1) as u64)
}
PostingList::Plain(ref plain) => plain.row_ids.get(self.index + 1).cloned(),
}
}
#[cfg(test)]
fn validate_modern_doc_ids(&self, num_docs: usize) -> Result<()> {
validate_modern_posting_doc_ids(&self.list, &self.token, num_docs)
}
}
pub(super) fn validate_modern_posting_doc_ids(
posting: &PostingList,
token: &str,
num_docs: usize,
) -> Result<()> {
let PostingList::Compressed(list) = posting else {
return Err(Error::index(format!(
"modern FTS posting for token {token:?} uses a legacy row-address layout"
)));
};
if list.length == 0 {
return Ok(());
}
let last_block_idx = list.blocks.len().checked_sub(1).ok_or_else(|| {
Error::index(format!(
"modern FTS posting for token {token:?} has length {} but no blocks",
list.length
))
})?;
let mut state = CompressedState::new(list.block_size);
state.decompress_doc_ids(
list.blocks.value(last_block_idx),
last_block_idx,
list.blocks.len(),
list.length,
list.posting_tail_codec,
list.block_size,
);
let max_doc_id = state.doc_ids.last().copied().ok_or_else(|| {
Error::index(format!(
"modern FTS posting for token {token:?} has an empty final block"
))
})?;
if max_doc_id as usize >= num_docs {
return Err(Error::index(format!(
"modern FTS posting for token {token:?} contains DocId {max_doc_id}, outside [0, {num_docs})"
)));
}
Ok(())
}
const MAXSCORE_INNER_WINDOW: usize = 1 << 12;
#[inline]
pub(super) fn score_sum_upper_bound_factor(num_values: usize) -> f64 {
if num_values <= 2 {
1.0
} else {
let relative_error_bound = (num_values - 1) as f64 * f64::from(f32::EPSILON);
1.0 + 2.0 * relative_error_bound
}
}
#[inline]
fn score_sum_cannot_compete(
partial_score: f32,
remaining_upper_bound: f64,
floor: f32,
upper_bound_factor: f64,
floor_mode: CompetitiveFloorMode,
) -> bool {
let upper_bound = (f64::from(partial_score) + remaining_upper_bound) * upper_bound_factor;
floor_mode.rejects_upper_bound(f64::from(outward_f32_upper_bound(upper_bound)), floor)
}
type ScoreContribution = ((u32, u32), f32);
#[inline]
fn score_contributions_in_query_order(mut contributions: SmallVec<[ScoreContribution; 8]>) -> f32 {
if !contributions
.windows(2)
.all(|window| window[0].0 <= window[1].0)
{
contributions.sort_unstable_by_key(|contribution| contribution.0);
}
contributions
.into_iter()
.fold(0.0_f32, |score, (_, contribution)| score + contribution)
}
struct WindowAccumulator {
scores: Vec<f32>,
freqs: Vec<u32>,
words: Vec<u64>,
num_clauses: usize,
}
impl WindowAccumulator {
fn new(num_clauses: usize) -> Self {
Self {
scores: vec![0.0; MAXSCORE_INNER_WINDOW],
freqs: vec![0; num_clauses * MAXSCORE_INNER_WINDOW],
words: vec![0; MAXSCORE_INNER_WINDOW / 64],
num_clauses,
}
}
#[inline]
fn add(&mut self, clause_idx: usize, slot: usize, score: f32, freq: u32) {
self.scores[slot] += score;
self.freqs[slot * self.num_clauses + clause_idx] = freq;
self.words[slot >> 6] |= 1u64 << (slot & 63);
}
#[inline]
fn clause_freq(&self, clause_idx: usize, slot: usize) -> u32 {
self.freqs[slot * self.num_clauses + clause_idx]
}
#[inline]
fn clear_slot(&mut self, slot: usize) {
self.scores[slot] = 0.0;
self.freqs[slot * self.num_clauses..(slot + 1) * self.num_clauses].fill(0);
}
}
#[derive(Debug)]
pub struct DocCandidate<C> {
pub document: C,
pub posting_doc_id: u64,
pub freqs: Vec<(u32, u32)>,
pub doc_length: u32,
}
pub(super) type FlatDocuments<'a> = (usize, Box<dyn Iterator<Item = (u64, u64)> + 'a>);
pub(super) trait WandDocuments {
type Candidate: Copy + Debug;
fn len(&self) -> usize;
fn visible_cost_upper_bound(&self) -> usize {
self.len()
}
fn scoring_norms(&self) -> Option<&[u8]>;
fn scoring_num_tokens(&self, doc_id: u32) -> u32;
fn doc_length(&self, doc: &DocInfo) -> u32;
fn document_key(&self, doc: &DocInfo) -> Option<u64>;
fn document_key_for_doc_id(&self, doc_id: u32) -> Option<u64>;
fn candidate_from_key(&self, key: u64) -> Self::Candidate;
fn flat_documents(&self) -> Option<FlatDocuments<'_>>;
fn flat_doc_length(&self, doc_id: u64, document_key: u64, compressed: bool) -> u32;
}
pub(super) trait ModernVisibility {
fn selected(&self, doc_id: DocId) -> bool;
fn len(&self, total_docs: usize) -> usize;
fn iter(&self) -> Option<Box<dyn Iterator<Item = DocId> + '_>>;
}
pub(super) struct AllModernDocuments;
impl ModernVisibility for AllModernDocuments {
#[inline]
fn selected(&self, _doc_id: DocId) -> bool {
true
}
fn len(&self, total_docs: usize) -> usize {
total_docs
}
fn iter(&self) -> Option<Box<dyn Iterator<Item = DocId> + '_>> {
None
}
}
impl ModernVisibility for &DocVisibility {
#[inline]
fn selected(&self, doc_id: DocId) -> bool {
DocVisibility::selected(self, doc_id)
}
fn len(&self, total_docs: usize) -> usize {
DocVisibility::len(self, total_docs)
}
fn iter(&self) -> Option<Box<dyn Iterator<Item = DocId> + '_>> {
DocVisibility::iter(self)
.map(|doc_ids| Box::new(doc_ids) as Box<dyn Iterator<Item = DocId>>)
}
}
pub(super) struct ModernWandDocuments<'a, V> {
lengths: &'a DocLengths,
visibility: V,
}
impl<'a> ModernWandDocuments<'a, AllModernDocuments> {
pub(crate) fn all(lengths: &'a DocLengths) -> Self {
Self {
lengths,
visibility: AllModernDocuments,
}
}
}
impl<'a> ModernWandDocuments<'a, &'a DocVisibility> {
pub(crate) fn filtered(lengths: &'a DocLengths, visibility: &'a DocVisibility) -> Self {
Self {
lengths,
visibility,
}
}
}
impl<V: ModernVisibility> WandDocuments for ModernWandDocuments<'_, V> {
type Candidate = DocId;
fn len(&self) -> usize {
self.lengths.len()
}
fn visible_cost_upper_bound(&self) -> usize {
self.visibility.len(self.lengths.len())
}
fn scoring_norms(&self) -> Option<&[u8]> {
self.lengths.scoring_norms()
}
fn scoring_num_tokens(&self, doc_id: u32) -> u32 {
self.lengths.scoring(DocId::new(doc_id))
}
fn doc_length(&self, doc: &DocInfo) -> u32 {
match doc {
DocInfo::Raw(doc) => self.scoring_num_tokens(doc.doc_id),
DocInfo::Located(_) => unreachable!("modern posting lists contain dense DocIds"),
}
}
fn document_key(&self, doc: &DocInfo) -> Option<u64> {
match doc {
DocInfo::Raw(doc) if self.visibility.selected(DocId::new(doc.doc_id)) => {
Some(u64::from(doc.doc_id))
}
DocInfo::Raw(_) => None,
DocInfo::Located(_) => unreachable!("modern posting lists contain dense DocIds"),
}
}
fn document_key_for_doc_id(&self, doc_id: u32) -> Option<u64> {
self.visibility
.selected(DocId::new(doc_id))
.then_some(u64::from(doc_id))
}
fn candidate_from_key(&self, key: u64) -> Self::Candidate {
DocId::new(key as u32)
}
fn flat_documents(&self) -> Option<(usize, Box<dyn Iterator<Item = (u64, u64)> + '_>)> {
self.visibility.iter().map(|doc_ids| {
let len = self.visibility.len(self.lengths.len());
let docs = doc_ids.map(|doc_id| {
let value = u64::from(doc_id.get());
(value, value)
});
(len, Box::new(docs) as Box<dyn Iterator<Item = (u64, u64)>>)
})
}
fn flat_doc_length(&self, doc_id: u64, _document_key: u64, _compressed: bool) -> u32 {
self.scoring_num_tokens(doc_id as u32)
}
}
pub(super) struct LegacyWandDocuments<'a> {
docs: &'a DocSet,
mask: &'a RowAddrMask,
}
impl<'a> LegacyWandDocuments<'a> {
pub(crate) fn new(docs: &'a DocSet, mask: &'a RowAddrMask) -> Self {
Self { docs, mask }
}
}
impl WandDocuments for LegacyWandDocuments<'_> {
type Candidate = u64;
fn len(&self) -> usize {
self.docs.len()
}
fn scoring_norms(&self) -> Option<&[u8]> {
self.docs.scoring_norms()
}
fn scoring_num_tokens(&self, doc_id: u32) -> u32 {
self.docs.scoring_num_tokens(doc_id)
}
fn doc_length(&self, doc: &DocInfo) -> u32 {
match doc {
DocInfo::Raw(doc) => self.docs.scoring_num_tokens(doc.doc_id),
DocInfo::Located(doc) => self.docs.num_tokens_by_row_id(doc.row_id),
}
}
fn document_key(&self, doc: &DocInfo) -> Option<u64> {
let row_id = match doc {
DocInfo::Raw(doc) => self.docs.row_id(doc.doc_id),
DocInfo::Located(doc) => doc.row_id,
};
(row_id != RowAddress::TOMBSTONE_ROW && self.mask.selected(row_id)).then_some(row_id)
}
fn document_key_for_doc_id(&self, doc_id: u32) -> Option<u64> {
let row_id = self.docs.row_id(doc_id);
(row_id != RowAddress::TOMBSTONE_ROW && self.mask.selected(row_id)).then_some(row_id)
}
fn candidate_from_key(&self, key: u64) -> Self::Candidate {
key
}
fn flat_documents(&self) -> Option<(usize, Box<dyn Iterator<Item = (u64, u64)> + '_>)> {
let count = self.mask.max_len()? as usize;
let row_ids = self.mask.iter_addrs()?;
let docs = row_ids.flat_map(|row_addr| {
let row_id: u64 = row_addr.into();
self.docs
.doc_ids(row_id)
.map(move |doc_id| (doc_id, row_id))
});
Some((count, Box::new(docs)))
}
fn flat_doc_length(&self, doc_id: u64, document_key: u64, compressed: bool) -> u32 {
if compressed {
self.docs.scoring_num_tokens(doc_id as u32)
} else {
self.docs.num_tokens_by_row_id(document_key)
}
}
}
#[cfg(test)]
impl WandDocuments for DocSet {
type Candidate = u64;
fn len(&self) -> usize {
self.len()
}
fn scoring_norms(&self) -> Option<&[u8]> {
self.scoring_norms()
}
fn scoring_num_tokens(&self, doc_id: u32) -> u32 {
self.scoring_num_tokens(doc_id)
}
fn doc_length(&self, doc: &DocInfo) -> u32 {
match doc {
DocInfo::Raw(doc) => self.scoring_num_tokens(doc.doc_id),
DocInfo::Located(doc) => self.num_tokens_by_row_id(doc.row_id),
}
}
fn document_key(&self, doc: &DocInfo) -> Option<u64> {
Some(match doc {
DocInfo::Raw(doc) if self.has_row_ids() => self.row_id(doc.doc_id),
DocInfo::Raw(doc) => u64::from(doc.doc_id),
DocInfo::Located(doc) => doc.row_id,
})
}
fn document_key_for_doc_id(&self, doc_id: u32) -> Option<u64> {
Some(if self.has_row_ids() {
self.row_id(doc_id)
} else {
u64::from(doc_id)
})
}
fn candidate_from_key(&self, key: u64) -> Self::Candidate {
key
}
fn flat_documents(&self) -> Option<(usize, Box<dyn Iterator<Item = (u64, u64)> + '_>)> {
None
}
fn flat_doc_length(&self, doc_id: u64, document_key: u64, compressed: bool) -> u32 {
if compressed {
self.scoring_num_tokens(doc_id as u32)
} else {
self.num_tokens_by_row_id(document_key)
}
}
}
struct HeadPosting {
doc_id: u64,
posting: Box<PostingIterator>,
}
impl HeadPosting {
fn new(posting: Box<PostingIterator>) -> Self {
let doc_id = posting
.doc()
.map(|doc| doc.doc_id())
.unwrap_or(TERMINATED_DOC_ID);
Self { doc_id, posting }
}
fn doc_id(&self) -> u64 {
self.doc_id
}
}
impl PartialEq for HeadPosting {
fn eq(&self, other: &Self) -> bool {
self.doc_id == other.doc_id
&& self.posting.approximate_upper_bound().to_bits()
== other.posting.approximate_upper_bound().to_bits()
&& self.posting.token_id == other.posting.token_id
&& self.posting.position == other.posting.position
}
}
impl Eq for HeadPosting {}
impl PartialOrd for HeadPosting {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for HeadPosting {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
other
.doc_id
.cmp(&self.doc_id)
.then_with(|| {
self.posting
.approximate_upper_bound()
.total_cmp(&other.posting.approximate_upper_bound())
})
.then_with(|| self.posting.token_id.cmp(&other.posting.token_id))
.then_with(|| self.posting.position.cmp(&other.posting.position))
}
}
struct TailPosting {
upper_bound: f32,
cost: usize,
posting: Box<PostingIterator>,
}
impl TailPosting {
fn new(upper_bound: f32, cost: usize, posting: Box<PostingIterator>) -> Self {
Self {
upper_bound,
cost,
posting,
}
}
}
impl PartialEq for TailPosting {
fn eq(&self, other: &Self) -> bool {
self.upper_bound.to_bits() == other.upper_bound.to_bits()
&& self.cost == other.cost
&& self.posting.token_id == other.posting.token_id
&& self.posting.position == other.posting.position
}
}
#[cfg(test)]
#[derive(Default)]
struct AndWindowStats {
windows_wide: usize,
windows_narrow: usize,
windows_skipped: usize,
range_blocks_scanned: usize,
candidates_returned: usize,
score_first_rejections: usize,
pairwise_intersections: usize,
}
impl Eq for TailPosting {}
impl PartialOrd for TailPosting {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for TailPosting {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.upper_bound
.total_cmp(&other.upper_bound)
.then_with(|| other.cost.cmp(&self.cost))
.then_with(|| other.posting.token_id.cmp(&self.posting.token_id))
.then_with(|| other.posting.position.cmp(&self.posting.position))
}
}
pub struct Wand<'a, S: Scorer, D: WandDocuments> {
threshold: f32, floor_mode: CompetitiveFloorMode,
operator: Operator,
num_terms: usize,
head: BinaryHeap<HeadPosting>,
#[allow(clippy::vec_box)]
lead: Vec<Box<PostingIterator>>,
tail: BinaryHeap<TailPosting>,
tail_max_score: f64,
up_to: Option<u64>,
and_max_score: f32,
and_last_doc: Option<u64>,
and_candidate_score: Option<f32>,
score_first_and_enabled: bool,
#[cfg(test)]
disable_score_first_and: bool,
score_first_and_suffix_bounds: SmallVec<[f64; 8]>,
score_first_and_bounds_up_to: Option<u64>,
score_first_and_dense_range: Option<(u64, u64)>,
score_first_and_norm_k: OnceCell<Option<NormKCache<'a>>>,
#[cfg(test)]
and_window_stats: AndWindowStats,
bulk_and_mode_override: Option<BulkAndMode>,
#[cfg(test)]
bulk_and_searches: usize,
#[cfg(test)]
maxscore_single_essential_windows: usize,
#[cfg(test)]
maxscore_general_windows: usize,
#[cfg(test)]
phrase_position_checks: Cell<usize>,
documents: &'a D,
scorer: S,
shared_threshold: Option<Arc<AtomicU32>>,
}
fn atomic_store_max_f32(slot: &AtomicU32, val: f32) {
let mut cur = slot.load(Ordering::Relaxed);
while val > f32::from_bits(cur) {
match slot.compare_exchange_weak(cur, val.to_bits(), Ordering::Relaxed, Ordering::Relaxed) {
Ok(_) => break,
Err(actual) => cur = actual,
}
}
}
impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> {
pub(crate) fn new(
operator: Operator,
postings: impl Iterator<Item = PostingIterator>,
documents: &'a D,
scorer: S,
) -> Self {
let mut head = BinaryHeap::new();
let mut lead = Vec::new();
for posting in postings {
if posting.current_doc_id().is_none() {
continue;
}
let posting = Box::new(posting);
if operator == Operator::And {
lead.push(posting);
} else {
head.push(HeadPosting::new(posting));
}
}
if operator == Operator::And {
lead.sort_unstable_by_key(|posting| posting.cost());
}
let score_first_and_enabled = operator == Operator::And
&& matches!(lead.len(), 4 | 5)
&& lead.iter().enumerate().all(|(index, posting)| {
lead[..index]
.iter()
.all(|prior| prior.token_id != posting.token_id)
})
&& lead.iter().all(|posting| {
!posting.has_grouped_terms()
&& !posting.use_scorer_upper_bound
&& posting.query_weight.is_finite()
&& posting.query_weight >= 0.0
&& matches!(
&posting.list,
PostingList::Compressed(list)
if list.block_size == MAX_POSTING_BLOCK_SIZE
&& list.impacts.is_some()
)
})
&& scorer
.doc_norm(1)
.is_some_and(|norm| norm.is_finite() && norm >= 0.0);
Self {
threshold: 0.0,
floor_mode: CompetitiveFloorMode::Exclusive,
operator,
num_terms: if operator == Operator::And {
lead.len()
} else {
head.len()
},
head,
lead,
tail: BinaryHeap::new(),
tail_max_score: 0.0,
up_to: None,
and_max_score: f32::INFINITY,
and_last_doc: None,
and_candidate_score: None,
score_first_and_enabled,
#[cfg(test)]
disable_score_first_and: false,
score_first_and_suffix_bounds: SmallVec::new(),
score_first_and_bounds_up_to: None,
score_first_and_dense_range: None,
score_first_and_norm_k: OnceCell::new(),
#[cfg(test)]
and_window_stats: AndWindowStats::default(),
bulk_and_mode_override: None,
#[cfg(test)]
bulk_and_searches: 0,
#[cfg(test)]
maxscore_single_essential_windows: 0,
#[cfg(test)]
maxscore_general_windows: 0,
#[cfg(test)]
phrase_position_checks: Cell::new(0),
documents,
scorer,
shared_threshold: None,
}
}
#[cfg(test)]
fn with_bulk_and_mode(mut self, mode: BulkAndMode) -> Self {
self.bulk_and_mode_override = Some(mode);
self
}
#[cfg(test)]
fn with_score_first_and(mut self, enabled: bool) -> Self {
self.disable_score_first_and = !enabled;
self
}
fn with_floor_mode(mut self, mode: CompetitiveFloorMode) -> Self {
self.floor_mode = mode;
self
}
pub(crate) fn with_shared_threshold(mut self, shared: Arc<AtomicU32>) -> Self {
self.shared_threshold = Some(shared);
self
}
fn norm_k_cache(&self) -> Option<(&'a [u8], Box<[f32; 256]>)> {
let norms = self.documents.scoring_norms()?;
let mut cache = Box::new([0f32; 256]);
for (code, slot) in cache.iter_mut().enumerate() {
*slot = self.scorer.doc_norm(dequantize_doc_length(code as u8))?;
}
Some((norms, cache))
}
fn update_threshold(&mut self, local_kth: f32, wand_factor: f32) {
let mut t = local_kth * wand_factor;
if let Some(shared) = self.shared_threshold.as_ref() {
atomic_store_max_f32(shared, local_kth);
let g = f32::from_bits(shared.load(Ordering::Relaxed)) * wand_factor;
if g > t {
t = g;
}
}
if self.score_first_and_enabled && self.threshold <= 0.0 && t > 0.0 {
self.up_to = None;
self.invalidate_score_first_and_window();
}
self.threshold = t;
}
fn raise_to_shared_floor(&mut self, wand_factor: f32) {
if let Some(shared) = self.shared_threshold.as_ref() {
let g = f32::from_bits(shared.load(Ordering::Relaxed)) * wand_factor;
if g > self.threshold {
if self.score_first_and_enabled && self.threshold <= 0.0 && g > 0.0 {
self.up_to = None;
self.invalidate_score_first_and_window();
}
self.threshold = g;
}
}
}
pub(crate) fn search(
&mut self,
params: &FtsSearchParams,
metrics: &dyn MetricsCollector,
) -> Result<Vec<DocCandidate<D::Candidate>>> {
let limit = params.limit.unwrap_or(usize::MAX);
if limit == 0 {
return Ok(vec![]);
}
if params.phrase_slop.is_some() {
self.invalidate_score_first_and_window();
self.score_first_and_enabled = false;
}
if self.operator == Operator::Or
&& let Some((num_docs_selected, documents)) = self.documents.flat_documents()
&& num_docs_selected.saturating_mul(100)
<= (*FLAT_SEARCH_PERCENT_THRESHOLD as usize).saturating_mul(self.documents.len())
{
return self.flat_search(params, documents, metrics);
}
if *USE_MAXSCORE_SEARCH
&& self.operator == Operator::Or
&& params.phrase_slop.is_none()
&& !self.head.is_empty()
&& self.head.iter().all(|posting| {
posting.posting.is_compressed() && !posting.posting.has_grouped_terms()
})
{
return self.maxscore_search(params, metrics);
}
if self.operator == Operator::And
&& !self.lead.is_empty()
&& self
.lead
.iter()
.all(|posting| posting.is_compressed() && !posting.has_grouped_terms())
&& {
let mode = self
.bulk_and_mode_override
.unwrap_or_else(|| *BULK_AND_MODE);
mode.enabled_for(self.lead.len())
|| (mode == BulkAndMode::Auto
&& self.lead.len() >= 4
&& self.lead.iter().all(|posting| {
matches!(&posting.list, PostingList::Compressed(list)
if list.block_size == MAX_POSTING_BLOCK_SIZE && list.impacts.is_some())
}))
}
{
#[cfg(test)]
{
self.bulk_and_searches += 1;
}
return self.and_bulk_search(params, metrics);
}
let mut candidates = TopKCollector::new(limit, std::cmp::min(limit, BLOCK_SIZE * 10));
let mut num_comparisons = 0;
loop {
self.raise_to_shared_floor(params.wand_factor);
let Some((doc, and_score)) = self.next()? else {
break;
};
num_comparisons += 1;
let posting_doc_id = doc.doc_id();
let Some(document_key) = self.documents.document_key(&doc) else {
if self.operator == Operator::Or {
self.push_back_leads(doc.doc_id() + 1);
}
continue;
};
let doc_length = self.documents.doc_length(&doc);
let score = if self.operator == Operator::Or {
self.advance_all_tail(doc.doc_id(), None, None);
if let Some(slop) = params.phrase_slop {
let early_score =
(self.threshold > 0.0).then(|| self.score_in_query_order(doc_length));
if early_score
.is_some_and(|score| self.exclusive_score_cannot_beat_floor(score))
{
self.push_back_leads(doc.doc_id() + 1);
continue;
}
if !self.check_positions(slop as i32)? {
self.push_back_leads(doc.doc_id() + 1);
continue;
}
early_score.unwrap_or_else(|| self.score_in_query_order(doc_length))
} else {
self.score_in_query_order(doc_length)
}
} else {
self.advance_all_tail(doc.doc_id(), None, None);
if let Some(slop) = params.phrase_slop {
let early_score = (self.threshold > 0.0).then(|| {
if self.and_candidate_score.is_some() {
and_score
} else {
self.score_in_query_order(doc_length)
}
});
if early_score
.is_some_and(|score| self.exclusive_score_cannot_beat_floor(score))
{
continue;
}
if !self.check_positions(slop as i32)? {
continue;
}
early_score.unwrap_or_else(|| {
if self.and_candidate_score.is_some() {
and_score
} else {
self.score_in_query_order(doc_length)
}
})
} else if self.and_candidate_score.is_some() {
and_score
} else {
self.score_in_query_order(doc_length)
}
};
if candidates.insert(
ScoredDoc::new(document_key, score),
doc_length,
posting_doc_id,
self.iter_term_freqs(),
)? && let Some(kth) = candidates.kth_score_if_full()
{
self.update_threshold(kth, params.wand_factor);
}
if self.operator == Operator::Or {
self.push_back_leads(doc.doc_id() + 1);
}
}
metrics.record_comparisons(num_comparisons);
candidates.into_candidates(|key| self.documents.candidate_from_key(key))
}
fn flat_search(
&mut self,
params: &FtsSearchParams,
documents: Box<dyn Iterator<Item = (u64, u64)> + '_>,
metrics: &dyn MetricsCollector,
) -> Result<Vec<DocCandidate<D::Candidate>>> {
let limit = params.limit.unwrap_or(usize::MAX);
if limit == 0 {
return Ok(vec![]);
}
let documents = documents.sorted_unstable().collect::<Vec<_>>();
let is_compressed = self
.head
.peek()
.map(|posting| matches!(posting.posting.list, PostingList::Compressed(_)))
.or_else(|| {
self.lead
.first()
.map(|posting| matches!(posting.list, PostingList::Compressed(_)))
})
.unwrap_or(false);
let mut num_comparisons = 0;
let mut candidates = TopKCollector::new(limit, 0);
for (doc_id, document_key) in documents {
num_comparisons += 1;
self.move_head_before_target_to_tail(doc_id);
self.move_head_doc_to_lead(doc_id);
if self.lead.is_empty() && self.tail.is_empty() {
continue;
}
if !self.can_target_beat_threshold(doc_id) {
self.advance_tail_and_lead_to_head(doc_id + 1);
continue;
}
self.collect_tail_matches(doc_id);
if self.operator == Operator::And && self.lead.len() < self.num_terms {
self.advance_lead_to_head(doc_id + 1);
continue;
}
let doc_length = self
.documents
.flat_doc_length(doc_id, document_key, is_compressed);
if self.operator == Operator::Or && !self.refine_or_candidate(doc_id, doc_length) {
self.advance_tail_and_lead_to_head(doc_id + 1);
continue;
}
self.collect_tail_matches(doc_id);
let early_score = (self.threshold > 0.0).then(|| self.score_in_query_order(doc_length));
if let Some(slop) = params.phrase_slop {
if early_score.is_some_and(|score| self.exclusive_score_cannot_beat_floor(score)) {
self.advance_lead_to_head(doc_id + 1);
continue;
}
if !self.check_positions(slop as i32)? {
self.advance_lead_to_head(doc_id + 1);
continue;
}
}
let score = early_score.unwrap_or_else(|| self.score_in_query_order(doc_length));
if candidates.insert(
ScoredDoc::new(document_key, score),
doc_length,
doc_id,
self.iter_term_freqs(),
)? && let Some(kth) = candidates.kth_score_if_full()
{
self.update_threshold(kth, params.wand_factor);
}
self.advance_lead_to_head(doc_id + 1);
}
metrics.record_comparisons(num_comparisons);
candidates.into_candidates(|key| self.documents.candidate_from_key(key))
}
fn maxscore_search(
&mut self,
params: &FtsSearchParams,
metrics: &dyn MetricsCollector,
) -> Result<Vec<DocCandidate<D::Candidate>>> {
struct MaxScoreClause {
posting: Box<PostingIterator>,
query_rank: usize,
bound: f32,
prefix_bound: f64,
}
let limit = params.limit.unwrap_or(usize::MAX);
let mut clauses = std::mem::take(&mut self.head)
.into_vec()
.into_iter()
.map(|head| MaxScoreClause {
posting: head.posting,
query_rank: 0,
bound: 0.0,
prefix_bound: 0.0,
})
.collect::<Vec<_>>();
clauses.sort_unstable_by_key(|clause| (clause.posting.position, clause.posting.token_id));
for (query_rank, clause) in clauses.iter_mut().enumerate() {
clause.query_rank = query_rank;
}
let num_query_terms = clauses.len();
let total_sum_upper_bound_factor = score_sum_upper_bound_factor(num_query_terms);
let mut acc = WindowAccumulator::new(clauses.len());
let mut candidates = TopKCollector::new(limit, std::cmp::min(limit, BLOCK_SIZE * 10));
let norm_k = self.norm_k_cache();
let norm_k_ref = norm_k
.as_ref()
.map(|(norms, cache)| (*norms, cache.as_ref()));
let mut num_comparisons = 0usize;
let mut min_window_size = 1u64;
let mut num_windows = 0u64;
let mut prev_first_essential = 0usize;
let mut window_min = clauses
.iter()
.filter_map(|clause| clause.posting.doc().map(|doc| doc.doc_id()))
.min()
.unwrap_or(TERMINATED_DOC_ID);
while window_min < TERMINATED_DOC_ID {
clauses.retain(|clause| clause.posting.doc().is_some());
if clauses.is_empty() {
break;
}
self.raise_to_shared_floor(params.wand_factor);
let first_window_lead = prev_first_essential.min(clauses.len() - 1);
let mut window_max = TERMINATED_DOC_ID;
for clause in &mut clauses {
let doc = clause
.posting
.doc()
.map(|doc| doc.doc_id())
.expect("exhausted clauses were retained out");
clause.posting.shallow_next(doc.max(window_min));
}
for clause in &clauses[first_window_lead..] {
window_max = window_max.min(clause.posting.block_end_doc());
}
if clauses.len() > 1 {
if (num_comparisons as u64) < num_windows * 32 * clauses.len() as u64 {
min_window_size = (min_window_size * 2).min(MAXSCORE_INNER_WINDOW as u64);
} else {
min_window_size = 1;
}
window_max = window_max.max(window_min.saturating_add(min_window_size - 1));
}
for clause in &mut clauses {
let doc = clause
.posting
.doc()
.map(|doc| doc.doc_id())
.expect("exhausted clauses were retained out");
clause.bound = if doc > window_max {
0.0
} else {
clause
.posting
.block_max_score_up_to_with_stats(window_max, &self.scorer)
.score
};
}
let mut clauses_in_query_order = true;
let mut first_essential = 0;
let mut prefix = 0.0_f64;
if self.threshold > 0.0 {
clauses.sort_unstable_by(|a, b| a.bound.total_cmp(&b.bound));
clauses_in_query_order = clauses
.windows(2)
.all(|window| window[0].query_rank <= window[1].query_rank);
for (i, clause) in clauses.iter_mut().enumerate() {
let next_prefix = prefix + f64::from(clause.bound);
let widened = next_prefix * score_sum_upper_bound_factor(i + 1);
if widened > f64::from(self.threshold) {
break;
}
prefix = next_prefix;
clause.prefix_bound = prefix;
first_essential = i + 1;
}
}
prev_first_essential = first_essential;
num_windows += 1;
if first_essential == clauses.len() {
window_min = match window_max {
TERMINATED_DOC_ID => TERMINATED_DOC_ID,
max => max + 1,
};
if clauses.len() == 1 && window_min != TERMINATED_DOC_ID && self.threshold > 0.0 {
let posting = &clauses[0].posting;
if let PostingList::Compressed(ref list) = posting.list
&& let Some(impacts) = list.impacts.as_ref()
{
let compressed = unsafe { &mut *posting.compressed_state_ptr() };
let bounds = impacts.level0_doc_weight_bounds_cached(
&self.scorer,
&mut compressed.block_max_window.impact_score_cache,
);
let query_weight = posting.query_weight;
let first_docs = list.block_first_docs();
let mut block_idx = first_docs
.partition_point(|&first| u64::from(first) <= window_min)
.saturating_sub(1);
while block_idx < bounds.len()
&& query_weight * bounds[block_idx] <= self.threshold
{
block_idx += 1;
}
window_min = if block_idx < bounds.len() {
window_min.max(u64::from(list.block_least_doc_id(block_idx)))
} else {
TERMINATED_DOC_ID
};
}
}
continue;
}
let total_non_essential_bound = if first_essential > 0 {
clauses[first_essential - 1].prefix_bound
} else {
0.0
};
if first_essential + 1 == clauses.len() {
#[cfg(test)]
{
self.maxscore_single_essential_windows += 1;
}
let (non_essential, essential) = clauses.split_at_mut(first_essential);
let essential_query_rank = essential[0].query_rank;
let posting = &mut essential[0].posting;
if posting.doc().is_some_and(|doc| doc.doc_id() < window_min) {
posting.next(window_min);
}
let essential_term = posting.term_index();
let essential_weight = posting.query_weight;
let sole_query_clause = non_essential.is_empty();
macro_rules! consider_candidate {
($doc:expr, $freq:expr) => {{
let doc = $doc;
let freq = $freq;
num_comparisons += 1;
let norm_addend =
norm_k_ref.map(|(norms, cache)| cache[norms[doc as usize] as usize]);
let score = match norm_addend {
Some(addend) => {
essential_weight * bm25_doc_weight_with_norm(freq, addend)
}
None => {
essential_weight
* self.scorer.doc_weight(
freq,
self.documents.scoring_num_tokens(doc as u32),
)
}
};
if !(self.threshold > 0.0
&& score_sum_cannot_compete(
score,
total_non_essential_bound,
self.threshold,
total_sum_upper_bound_factor,
CompetitiveFloorMode::Exclusive,
))
{
if let Some(document_key) =
self.documents.document_key_for_doc_id(doc as u32)
{
let mut total = score;
let mut scores_by_query_rank = SmallVec::<[f32; 8]>::new();
if !sole_query_clause {
scores_by_query_rank.resize(num_query_terms, 0.0);
scores_by_query_rank[essential_query_rank] = score;
}
let mut rejected = false;
for i in (0..non_essential.len()).rev() {
if self.threshold > 0.0
&& score_sum_cannot_compete(
total,
non_essential[i].prefix_bound,
self.threshold,
total_sum_upper_bound_factor,
CompetitiveFloorMode::Exclusive,
)
{
rejected = true;
break;
}
let query_rank = non_essential[i].query_rank;
let probe = &mut non_essential[i].posting;
if probe.doc().is_some_and(|d| d.doc_id() < doc) {
probe.next(doc);
}
if let Some(d) = probe.doc()
&& d.doc_id() == doc
{
let contribution = match norm_addend {
Some(addend) => {
probe.query_weight
* bm25_doc_weight_with_norm(
d.frequency(),
addend,
)
}
None => probe.score(
&self.scorer,
d.frequency(),
self.documents.scoring_num_tokens(doc as u32),
),
};
total += contribution;
scores_by_query_rank[query_rank] = contribution;
}
}
if !rejected {
let canonical_score = if sole_query_clause {
score
} else {
scores_by_query_rank
.into_iter()
.fold(0.0_f32, |sum, contribution| sum + contribution)
};
if canonical_score > self.threshold {
let doc_length =
self.documents.scoring_num_tokens(doc as u32);
if candidates.insert(
ScoredDoc::new(document_key, canonical_score),
doc_length,
doc,
std::iter::once((essential_term, freq)).chain(
non_essential.iter().filter_map(|clause| {
clause.posting.doc().and_then(|d| {
(d.doc_id() == doc).then(|| {
(
clause.posting.term_index(),
d.frequency(),
)
})
})
}),
),
)? && let Some(kth) = candidates.kth_score_if_full()
{
self.update_threshold(kth, params.wand_factor);
}
}
}
}
}
}};
}
match posting.list {
PostingList::Compressed(ref list) => {
let shift = list.block_shift();
let mask = list.block_mask();
'stream: while let Some(cur) = posting.current_doc {
if cur.doc_id() > window_max {
break;
}
let block_idx = posting.index >> shift;
let block_offset = posting.index & mask;
let compressed = unsafe {
&mut *posting.ensure_compressed_block_ptr(list, block_idx)
};
for offset in block_offset..compressed.doc_ids.len() {
let doc_id = compressed.doc_ids[offset];
if u64::from(doc_id) > window_max {
posting.index = (block_idx << shift) + offset;
posting.block_idx = block_idx;
posting.current_doc = Some(DocInfo::Raw(RawDocInfo {
doc_id,
frequency: compressed.freqs[offset],
}));
break 'stream;
}
consider_candidate!(u64::from(doc_id), compressed.freqs[offset]);
}
let next_start = (block_idx + 1) << shift;
if next_start >= list.length as usize {
posting.index = list.length as usize;
posting.block_idx = posting.index >> shift;
posting.current_doc = None;
break;
}
posting.index = next_start;
posting.block_idx = block_idx + 1;
let compressed = unsafe {
&mut *posting.ensure_compressed_block_ptr(list, block_idx + 1)
};
posting.current_doc = Some(DocInfo::Raw(RawDocInfo {
doc_id: compressed.doc_ids[0],
frequency: compressed.freqs[0],
}));
}
}
PostingList::Plain(_) => {
while let Some(cur) = posting.doc() {
let doc = cur.doc_id();
if doc > window_max {
break;
}
consider_candidate!(doc, cur.frequency());
posting.next(doc + 1);
}
}
}
window_min = match window_max {
TERMINATED_DOC_ID => TERMINATED_DOC_ID,
max => max + 1,
};
continue;
}
#[cfg(test)]
{
self.maxscore_general_windows += 1;
}
let mut inner_min = window_min;
loop {
let mut next_essential_doc = TERMINATED_DOC_ID;
for clause in &clauses[first_essential..] {
if let Some(doc) = clause.posting.doc() {
next_essential_doc = next_essential_doc.min(doc.doc_id());
}
}
inner_min = inner_min.max(next_essential_doc);
if inner_min == TERMINATED_DOC_ID || inner_min > window_max {
break;
}
let inner_max =
window_max.min(inner_min.saturating_add(MAXSCORE_INNER_WINDOW as u64 - 1));
for (clause_idx, clause) in clauses.iter_mut().enumerate().skip(first_essential) {
clause.posting.collect_window_scores(
inner_min,
inner_max,
clause_idx,
self.documents,
&self.scorer,
norm_k_ref,
&mut acc,
);
}
for word_idx in 0..acc.words.len() {
let mut word = acc.words[word_idx];
if word == 0 {
continue;
}
acc.words[word_idx] = 0;
while word != 0 {
let bit = word.trailing_zeros() as usize;
word &= word - 1;
let slot = (word_idx << 6) | bit;
let doc = inner_min + slot as u64;
let mut score = acc.scores[slot];
num_comparisons += 1;
if self.threshold > 0.0
&& score_sum_cannot_compete(
score,
total_non_essential_bound,
self.threshold,
total_sum_upper_bound_factor,
CompetitiveFloorMode::Exclusive,
)
{
acc.clear_slot(slot);
continue;
}
let Some(document_key) = self.documents.document_key_for_doc_id(doc as u32)
else {
acc.clear_slot(slot);
continue;
};
let norm_addend =
norm_k_ref.map(|(norms, cache)| cache[norms[doc as usize] as usize]);
let mut doc_length_cell: Option<u32> = None;
let needs_canonical_rescore =
first_essential != 0 || !clauses_in_query_order;
let mut scores_by_query_rank = SmallVec::<[f32; 8]>::new();
if needs_canonical_rescore {
scores_by_query_rank.resize(num_query_terms, 0.0);
}
let mut rejected = false;
for i in (0..first_essential).rev() {
if self.threshold > 0.0
&& score_sum_cannot_compete(
score,
clauses[i].prefix_bound,
self.threshold,
total_sum_upper_bound_factor,
CompetitiveFloorMode::Exclusive,
)
{
rejected = true;
break;
}
let query_rank = clauses[i].query_rank;
let posting = &mut clauses[i].posting;
if posting.doc().is_some_and(|d| d.doc_id() < doc) {
posting.next(doc);
}
if let Some(d) = posting.doc()
&& d.doc_id() == doc
{
let contribution = match norm_addend {
Some(addend) => {
posting.query_weight
* bm25_doc_weight_with_norm(d.frequency(), addend)
}
None => {
let doc_length =
*doc_length_cell.get_or_insert_with(|| {
self.documents.scoring_num_tokens(doc as u32)
});
posting.score(&self.scorer, d.frequency(), doc_length)
}
};
score += contribution;
if needs_canonical_rescore {
scores_by_query_rank[query_rank] = contribution;
}
}
}
if !rejected {
let doc_length = doc_length_cell
.unwrap_or_else(|| self.documents.scoring_num_tokens(doc as u32));
let score = if !needs_canonical_rescore {
score
} else {
for (i, clause) in clauses.iter().enumerate().skip(first_essential)
{
let freq = acc.clause_freq(i, slot);
if freq == 0 {
continue;
}
let contribution = match norm_addend {
Some(addend) => {
clause.posting.query_weight
* bm25_doc_weight_with_norm(freq, addend)
}
None => {
clause.posting.score(&self.scorer, freq, doc_length)
}
};
scores_by_query_rank[clause.query_rank] = contribution;
}
scores_by_query_rank
.iter()
.fold(0.0_f32, |sum, contribution| sum + contribution)
};
if score <= self.threshold {
acc.clear_slot(slot);
continue;
}
if candidates.insert(
ScoredDoc::new(document_key, score),
doc_length,
doc,
clauses.iter().enumerate().filter_map(|(i, clause)| {
if i >= first_essential {
let freq = acc.clause_freq(i, slot);
(freq > 0).then(|| (clause.posting.term_index(), freq))
} else {
clause.posting.doc().and_then(|d| {
(d.doc_id() == doc).then(|| {
(clause.posting.term_index(), d.frequency())
})
})
}
}),
)? && let Some(kth) = candidates.kth_score_if_full()
{
self.update_threshold(kth, params.wand_factor);
}
}
acc.clear_slot(slot);
}
}
if inner_max >= window_max {
break;
}
inner_min = inner_max + 1;
}
window_min = match window_max {
TERMINATED_DOC_ID => TERMINATED_DOC_ID,
max => max + 1,
};
}
metrics.record_comparisons(num_comparisons);
candidates.into_candidates(|key| self.documents.candidate_from_key(key))
}
#[inline]
fn exclusive_score_cannot_beat_floor(&self, score: f32) -> bool {
self.threshold > 0.0
&& score.is_finite()
&& !self.floor_mode.accepts_score(score, self.threshold)
}
fn score_in_query_order(&self, doc_length: u32) -> f32 {
if self.lead.windows(2).all(|window| {
(window[0].position, window[0].token_id) <= (window[1].position, window[1].token_id)
}) {
return self.lead.iter().fold(0.0_f32, |score, posting| {
score
+ posting.doc().map_or(0.0, |doc| {
posting.score(&self.scorer, doc.frequency(), doc_length)
})
});
}
let contributions = self
.lead
.iter()
.filter_map(|posting| {
posting.doc().map(|doc| {
(
(posting.position, posting.token_id),
posting.score(&self.scorer, doc.frequency(), doc_length),
)
})
})
.collect::<SmallVec<[ScoreContribution; 8]>>();
score_contributions_in_query_order(contributions)
}
fn iter_term_freqs(&self) -> impl Iterator<Item = (u32, u32)> + '_ {
self.lead.iter().filter_map(|posting| {
posting
.doc()
.map(|doc| (posting.term_index(), doc.frequency()))
})
}
fn and_candidate_cannot_beat_threshold(&self, doc_length: u32) -> bool {
if self.operator != Operator::And
|| self.threshold <= 0.0
|| self.num_terms < 2
|| self.lead.len() != self.num_terms
{
return false;
}
let Some((first, remaining)) = self.lead.split_first() else {
return false;
};
let Some(doc) = first.doc() else {
return false;
};
let remaining_upper_bound = remaining
.iter()
.map(|posting| f64::from(posting.block_max_score(&self.scorer)))
.sum::<f64>();
score_sum_cannot_compete(
first.score(&self.scorer, doc.frequency(), doc_length),
remaining_upper_bound,
self.threshold,
score_sum_upper_bound_factor(self.num_terms),
self.floor_mode,
)
}
fn invalidate_score_first_and_window(&mut self) {
if self.score_first_and_enabled {
self.score_first_and_suffix_bounds.clear();
self.score_first_and_bounds_up_to = None;
self.score_first_and_dense_range = None;
}
self.and_candidate_score = None;
}
fn prepare_score_first_and_window(&mut self) {
if !self.score_first_and_enabled {
return;
}
self.score_first_and_suffix_bounds.clear();
self.score_first_and_bounds_up_to = None;
self.score_first_and_dense_range = None;
if self.threshold <= 0.0 {
return;
}
#[cfg(test)]
if self.disable_score_first_and {
return;
}
let Some(up_to) = self.up_to else {
return;
};
if self.and_max_score < self.threshold {
return;
}
let mut dense_from = 0;
let mut dense_to = up_to;
for posting in &self.lead {
let Some(current_doc) = posting.current_doc_id() else {
return;
};
let PostingList::Compressed(list) = &posting.list else {
return;
};
if list.block_size == 0 {
return;
}
let expected_blocks = (list.length as usize).div_ceil(list.block_size);
if list.blocks.len() != expected_blocks || posting.block_idx >= list.blocks.len() {
return;
}
let Some(block_start) = posting.block_idx.checked_mul(list.block_size) else {
return;
};
let Some(remaining) = (list.length as usize).checked_sub(block_start) else {
return;
};
let block_count = remaining.min(list.block_size);
if block_count == 0 {
return;
}
let block_first = list.block_least_doc_id(posting.block_idx);
let Some(block_last) = list
.impacts
.as_ref()
.and_then(|impacts| impacts.level0_doc_up_to(posting.block_idx))
else {
return;
};
if block_last < block_first
|| u64::from(block_last) - u64::from(block_first) + 1 != block_count as u64
{
return;
}
dense_from = dense_from.max(current_doc).max(u64::from(block_first));
dense_to = dense_to.min(u64::from(block_last));
}
if dense_from > dense_to {
return;
}
self.score_first_and_suffix_bounds
.resize(self.lead.len() + 1, 0.0);
for index in (0..self.lead.len()).rev() {
let bound = self.lead[index].block_max_score(&self.scorer);
if !bound.is_finite() || bound < 0.0 {
self.score_first_and_suffix_bounds.clear();
return;
}
self.score_first_and_suffix_bounds[index] =
next_up_f64(f64::from(bound) + self.score_first_and_suffix_bounds[index + 1]);
}
self.score_first_and_bounds_up_to = Some(up_to);
self.score_first_and_dense_range = Some((dense_from, dense_to));
}
#[inline]
fn score_first_and_cannot_compete(&self, partial_score: f32, next_clause: usize) -> bool {
self.score_first_and_bounds_up_to == self.up_to
&& self
.score_first_and_suffix_bounds
.get(next_clause)
.is_some_and(|remaining_upper_bound| {
score_sum_cannot_compete(
partial_score,
*remaining_upper_bound,
self.threshold,
score_sum_upper_bound_factor(self.num_terms),
self.floor_mode,
)
})
}
fn and_next_target_after(&self, doc: u64) -> Option<u64> {
if self
.lead
.first()
.is_some_and(|posting| posting.is_compressed())
&& doc >= u64::from(u32::MAX)
{
None
} else {
doc.checked_add(1)
}
}
fn next(&mut self) -> Result<Option<(DocInfo, f32)>> {
if self.operator == Operator::And {
let candidate = self.next_and_candidate();
#[cfg(test)]
{
if candidate.is_some() {
self.and_window_stats.candidates_returned += 1;
}
}
return Ok(candidate.map(|doc| (doc, self.and_candidate_score.unwrap_or(0.0))));
}
loop {
let Some(target) = self.head_doc() else {
if self.advance_tail_to_next_or_window() {
continue;
}
return Ok(None);
};
if self.up_to.is_none_or(|up_to| target > up_to) {
self.update_max_scores(target);
}
self.move_head_doc_to_lead(target);
if self.lead.is_empty() {
continue;
}
if self.threshold > 0.0
&& self
.floor_mode
.rejects_upper_bound(f64::from(self.or_block_window_max()), self.threshold)
{
let mut skip_to = match self.up_to {
Some(up_to) if up_to < u32::MAX as u64 => up_to + 1,
_ => target + 1,
};
let group_skip = self.or_group_skip_to();
if let Some(group_skip_to) = group_skip {
skip_to = skip_to.max(group_skip_to);
}
self.push_back_leads(skip_to);
continue;
}
let Some(first_doc) = self.lead.first().and_then(|posting| posting.doc()) else {
self.push_back_leads(target + 1);
continue;
};
let doc_length = self.documents.doc_length(&first_doc);
let mut lead_score = 0.0;
if let Some(first_posting) = self.lead.first() {
lead_score += first_posting.score(&self.scorer, first_doc.frequency(), doc_length);
}
for posting in self.lead.iter().skip(1) {
if let Some(lead_doc) = posting.doc() {
lead_score += posting.score(&self.scorer, lead_doc.frequency(), doc_length);
}
}
while !self.floor_mode.accepts_score(lead_score, self.threshold) {
if score_sum_cannot_compete(
lead_score,
self.tail_upper_bound_sum(),
self.threshold,
score_sum_upper_bound_factor(self.num_terms),
self.floor_mode,
) {
self.push_back_leads(first_doc.doc_id() + 1);
break;
}
if !self.advance_tail_top(target, doc_length, &mut lead_score) {
break;
}
}
if !self.lead.is_empty() {
return Ok(Some((first_doc, lead_score)));
}
}
}
fn next_and_candidate(&mut self) -> Option<DocInfo> {
self.and_candidate_score = None;
if self.lead.len() < self.num_terms {
return None;
}
let is_vectorized_search_enabled = matches!(self.lead.len(), 4 | 5);
if let Some(last_doc) = self.and_last_doc
&& self
.lead
.first()
.and_then(|posting| posting.current_doc_id())
== Some(last_doc)
{
let next_target = self.and_advance_target(self.and_next_target_after(last_doc)?);
if next_target == TERMINATED_DOC_ID {
return None;
}
self.lead[0].next_doc_id(next_target, is_vectorized_search_enabled);
}
'advance_head: loop {
let doc = self
.lead
.first()
.and_then(|posting| posting.current_doc_id())?;
if self.up_to.is_none_or(|up_to| doc > up_to) {
let next_target = self.and_advance_target(doc);
if next_target == TERMINATED_DOC_ID {
return None;
}
if next_target != doc {
self.lead[0].next_doc_id(next_target, is_vectorized_search_enabled);
continue;
}
}
let mut is_score_first = self.score_first_and_bounds_up_to == self.up_to
&& self.score_first_and_suffix_bounds.len() == self.lead.len() + 1
&& self
.score_first_and_dense_range
.is_some_and(|(from, to)| doc >= from && doc <= to);
let mut partial_score = 0.0_f32;
let mut score_contributions = SmallVec::<[ScoreContribution; 8]>::new();
let mut score_first_doc_norm = 0.0;
if is_score_first {
let lead_doc = self.lead.first().and_then(|posting| posting.doc())?;
let cached_doc_norm = self
.score_first_and_norm_k
.get_or_init(|| {
self.norm_k_cache().filter(|(_, cache)| {
cache
.iter()
.all(|addend| addend.is_finite() && *addend >= 0.0)
})
})
.as_ref()
.and_then(|(norms, cache)| {
norms.get(doc as usize).map(|&code| cache[code as usize])
});
let doc_norm = cached_doc_norm
.or_else(|| self.scorer.doc_norm(self.documents.doc_length(&lead_doc)));
if let Some(doc_norm) = doc_norm
&& doc_norm.is_finite()
&& doc_norm >= 0.0
{
score_first_doc_norm = doc_norm;
let contribution = self.lead[0].query_weight
* bm25_doc_weight_with_norm(lead_doc.frequency(), doc_norm);
partial_score = contribution;
score_contributions
.push(((self.lead[0].position, self.lead[0].token_id), contribution));
if self.score_first_and_cannot_compete(partial_score, 1) {
#[cfg(test)]
{
self.and_window_stats.score_first_rejections += 1;
}
let next_target = self.and_advance_target(self.and_next_target_after(doc)?);
if next_target == TERMINATED_DOC_ID {
return None;
}
self.lead[0].next_doc_id(next_target, is_vectorized_search_enabled);
continue;
}
} else {
is_score_first = false;
}
}
for index in 1..self.lead.len() {
let posting = &mut self.lead[index];
if posting.current_doc_id()? < doc {
posting.next_doc_id(doc, is_vectorized_search_enabled);
}
let next = posting.current_doc_id()?;
if next > doc {
let next_target = self.and_advance_target(next);
if next_target == TERMINATED_DOC_ID {
return None;
}
self.lead[0].next_doc_id(next_target, is_vectorized_search_enabled);
continue 'advance_head;
}
if is_score_first {
let posting_doc = posting.doc()?;
let contribution = posting.query_weight
* bm25_doc_weight_with_norm(posting_doc.frequency(), score_first_doc_norm);
partial_score += contribution;
score_contributions.push(((posting.position, posting.token_id), contribution));
if self.score_first_and_cannot_compete(partial_score, index + 1) {
#[cfg(test)]
{
self.and_window_stats.score_first_rejections += 1;
}
let next_target = self.and_advance_target(self.and_next_target_after(doc)?);
if next_target == TERMINATED_DOC_ID {
return None;
}
self.lead[0].next_doc_id(next_target, is_vectorized_search_enabled);
continue 'advance_head;
}
}
}
let lead_doc = self.lead.first().and_then(|posting| posting.doc())?;
let doc_length = self.documents.doc_length(&lead_doc);
if !is_score_first && self.and_candidate_cannot_beat_threshold(doc_length) {
let next_target = self.and_advance_target(self.and_next_target_after(doc)?);
if next_target == TERMINATED_DOC_ID {
return None;
}
self.lead[0].next_doc_id(next_target, is_vectorized_search_enabled);
continue;
}
if is_score_first {
self.and_candidate_score =
Some(score_contributions_in_query_order(score_contributions));
}
self.and_last_doc = Some(doc);
return Some(lead_doc);
}
}
fn posting_block_up_to(posting: &PostingIterator, target: u64) -> u64 {
posting
.next_block_first_doc()
.map(|doc| doc.saturating_sub(1))
.unwrap_or(TERMINATED_DOC_ID)
.max(target)
}
fn and_bulk_search(
&mut self,
params: &FtsSearchParams,
metrics: &dyn MetricsCollector,
) -> Result<Vec<DocCandidate<D::Candidate>>> {
let limit = params.limit.unwrap_or(usize::MAX);
if limit == 0 {
return Ok(vec![]);
}
let num_lists = self.lead.len();
let use_pairwise_intersection = num_lists >= 4
&& self.lead.iter().all(|posting| {
matches!(&posting.list, PostingList::Compressed(list)
if list.block_size == MAX_POSTING_BLOCK_SIZE && list.impacts.is_some())
});
let phrase_slop = params.phrase_slop;
let mut score_order = (0..num_lists).collect::<Vec<_>>();
score_order.sort_unstable_by_key(|&index| {
let posting = &self.lead[index];
(posting.position, posting.token_id)
});
struct WindowList {
docs: *const u32,
freqs: *const u32,
pos: usize,
end: usize,
block_start: usize,
}
#[inline]
fn window_bm25_score<S: Scorer + ?Sized>(
lead: &[Box<PostingIterator>],
wins: &[WindowList],
offs: &[u8],
score_order: &[usize],
scorer: &S,
norm_addend: Option<f32>,
doc_length: u32,
) -> f32 {
let mut score = 0.0_f32;
for &clause_index in score_order {
let win = &wins[clause_index];
let posting = &lead[clause_index];
let off = offs[clause_index];
let freq = unsafe { *win.freqs.add(off as usize) };
score += match norm_addend {
Some(addend) => posting.query_weight * bm25_doc_weight_with_norm(freq, addend),
None => posting.score(scorer, freq, doc_length),
};
}
score
}
macro_rules! merge_kernels {
($name2:ident, $name3:ident, $docs_name2:ident, $docs_name3:ident, $geq:ident $(, #[$feat:meta])?) => {
$(#[$feat])?
unsafe fn $name2(
wins: &[WindowList],
freq_cannot_beat: &[bool; FREQ_LUT_BUCKETS],
docs_out: &mut Vec<u32>,
offs_out: &mut Vec<u8>,
) {
let (d0, mut p0, e0) = (wins[0].docs, wins[0].pos, wins[0].end);
let (d1, mut p1, e1) = (wins[1].docs, wins[1].pos, wins[1].end);
let f0 = wins[0].freqs;
unsafe {
while p0 < e0 {
let doc = *d0.add(p0);
let freq = (*f0.add(p0) as usize).min(FREQ_LUT_BUCKETS - 1);
if freq_cannot_beat[freq] {
p0 += 1;
continue;
}
p1 = $geq(d1, p1, e1, doc);
if p1 >= e1 {
return;
}
let second = *d1.add(p1);
if second > doc {
p0 = $geq(d0, p0 + 1, e0, second);
continue;
}
docs_out.push(doc);
offs_out.push(p0 as u8);
offs_out.push(p1 as u8);
p0 += 1;
}
}
}
$(#[$feat])?
unsafe fn $name3(
wins: &[WindowList],
freq_cannot_beat: &[bool; FREQ_LUT_BUCKETS],
docs_out: &mut Vec<u32>,
offs_out: &mut Vec<u8>,
) {
let (d0, mut p0, e0) = (wins[0].docs, wins[0].pos, wins[0].end);
let (d1, mut p1, e1) = (wins[1].docs, wins[1].pos, wins[1].end);
let (d2, mut p2, e2) = (wins[2].docs, wins[2].pos, wins[2].end);
let f0 = wins[0].freqs;
unsafe {
'outer: while p0 < e0 {
let doc = *d0.add(p0);
let freq = (*f0.add(p0) as usize).min(FREQ_LUT_BUCKETS - 1);
if freq_cannot_beat[freq] {
p0 += 1;
continue 'outer;
}
p1 = $geq(d1, p1, e1, doc);
if p1 >= e1 {
return;
}
let second = *d1.add(p1);
if second > doc {
p0 = $geq(d0, p0 + 1, e0, second);
continue 'outer;
}
p2 = $geq(d2, p2, e2, doc);
if p2 >= e2 {
return;
}
let third = *d2.add(p2);
if third > doc {
p0 = $geq(d0, p0 + 1, e0, third);
continue 'outer;
}
docs_out.push(doc);
offs_out.push(p0 as u8);
offs_out.push(p1 as u8);
offs_out.push(p2 as u8);
p0 += 1;
}
}
}
$(#[$feat])?
unsafe fn $docs_name2(
wins: &[WindowList],
docs_out: &mut Vec<u32>,
offs_out: &mut Vec<u8>,
) {
let (d0, mut p0, e0) = (wins[0].docs, wins[0].pos, wins[0].end);
let (d1, mut p1, e1) = (wins[1].docs, wins[1].pos, wins[1].end);
unsafe {
while p0 < e0 {
let doc = *d0.add(p0);
p1 = $geq(d1, p1, e1, doc);
if p1 >= e1 {
return;
}
let second = *d1.add(p1);
if second > doc {
p0 = $geq(d0, p0 + 1, e0, second);
continue;
}
docs_out.push(doc);
offs_out.push(p0 as u8);
offs_out.push(p1 as u8);
p0 += 1;
}
}
}
$(#[$feat])?
unsafe fn $docs_name3(
wins: &[WindowList],
docs_out: &mut Vec<u32>,
offs_out: &mut Vec<u8>,
) {
let (d0, mut p0, e0) = (wins[0].docs, wins[0].pos, wins[0].end);
let (d1, mut p1, e1) = (wins[1].docs, wins[1].pos, wins[1].end);
let (d2, mut p2, e2) = (wins[2].docs, wins[2].pos, wins[2].end);
unsafe {
'outer: while p0 < e0 {
let doc = *d0.add(p0);
p1 = $geq(d1, p1, e1, doc);
if p1 >= e1 {
return;
}
let second = *d1.add(p1);
if second > doc {
p0 = $geq(d0, p0 + 1, e0, second);
continue 'outer;
}
p2 = $geq(d2, p2, e2, doc);
if p2 >= e2 {
return;
}
let third = *d2.add(p2);
if third > doc {
p0 = $geq(d0, p0 + 1, e0, third);
continue 'outer;
}
docs_out.push(doc);
offs_out.push(p0 as u8);
offs_out.push(p1 as u8);
offs_out.push(p2 as u8);
p0 += 1;
}
}
}
};
}
merge_kernels!(
merge_window_2,
merge_window_3,
merge_window_docs_2,
merge_window_docs_3,
find_next_geq_scalar
);
#[cfg(target_arch = "x86_64")]
merge_kernels!(
merge_window_2_avx2,
merge_window_3_avx2,
merge_window_docs_2_avx2,
merge_window_docs_3_avx2,
find_next_geq_avx2,
#[target_feature(enable = "avx2")]
);
#[inline]
fn merge_window_1(wins: &[WindowList], docs_out: &mut Vec<u32>, offs_out: &mut Vec<u8>) {
let win = &wins[0];
for pos in win.pos..win.end {
docs_out.push(unsafe { *win.docs.add(pos) });
offs_out.push(pos as u8);
}
}
#[inline]
#[allow(clippy::too_many_arguments)]
fn merge_window_n(
wins: &[WindowList],
freq_cannot_beat: &[bool; FREQ_LUT_BUCKETS],
cursors: &mut Vec<usize>,
docs_out: &mut Vec<u32>,
offs_out: &mut Vec<u8>,
) {
cursors.clear();
cursors.extend(wins.iter().map(|win| win.pos));
'outer: while cursors[0] < wins[0].end {
let doc = unsafe { *wins[0].docs.add(cursors[0]) };
let freq =
unsafe { *wins[0].freqs.add(cursors[0]) as usize }.min(FREQ_LUT_BUCKETS - 1);
if freq_cannot_beat[freq] {
cursors[0] += 1;
continue 'outer;
}
for j in 1..wins.len() {
let win = &wins[j];
let pos = unsafe { find_next_geq(win.docs, cursors[j], win.end, doc) };
cursors[j] = pos;
if pos >= win.end {
return;
}
let clause_doc = unsafe { *win.docs.add(pos) };
if clause_doc > doc {
cursors[0] = unsafe {
find_next_geq(wins[0].docs, cursors[0] + 1, wins[0].end, clause_doc)
};
continue 'outer;
}
}
docs_out.push(doc);
for &pos in cursors.iter() {
offs_out.push(pos as u8);
}
cursors[0] += 1;
}
}
#[inline]
fn merge_window_docs_n(
wins: &[WindowList],
cursors: &mut Vec<usize>,
docs_out: &mut Vec<u32>,
offs_out: &mut Vec<u8>,
) {
cursors.clear();
cursors.extend(wins.iter().map(|win| win.pos));
'outer: while cursors[0] < wins[0].end {
let doc = unsafe { *wins[0].docs.add(cursors[0]) };
for j in 1..wins.len() {
let win = &wins[j];
let pos = unsafe { find_next_geq(win.docs, cursors[j], win.end, doc) };
cursors[j] = pos;
if pos >= win.end {
return;
}
let clause_doc = unsafe { *win.docs.add(pos) };
if clause_doc > doc {
cursors[0] = unsafe {
find_next_geq(wins[0].docs, cursors[0] + 1, wins[0].end, clause_doc)
};
continue 'outer;
}
}
docs_out.push(doc);
for &pos in cursors.iter() {
offs_out.push(pos as u8);
}
cursors[0] += 1;
}
}
#[inline]
fn merge_window_docs_pairwise(
wins: &[WindowList],
scratch: &mut Vec<u32>,
docs_out: &mut Vec<u32>,
offs_out: &mut Vec<u8>,
) {
let first = &wins[0];
docs_out.extend_from_slice(unsafe {
std::slice::from_raw_parts(first.docs.add(first.pos), first.end - first.pos)
});
for win in &wins[1..] {
let slice =
unsafe { std::slice::from_raw_parts(win.docs.add(win.pos), win.end - win.pos) };
let first_doc = slice[0];
let last_doc = slice[slice.len() - 1];
if u64::from(last_doc) - u64::from(first_doc) + 1 == slice.len() as u64 {
if docs_out[0] >= first_doc && docs_out[docs_out.len() - 1] <= last_doc {
continue;
}
let lo = docs_out.partition_point(|&doc| doc < first_doc);
let hi = docs_out.partition_point(|&doc| doc <= last_doc);
if lo != 0 {
docs_out.copy_within(lo..hi, 0);
}
docs_out.truncate(hi - lo);
} else {
intersection::intersect(docs_out, slice, scratch);
std::mem::swap(docs_out, scratch);
}
if docs_out.is_empty() {
return;
}
}
offs_out.resize(docs_out.len() * wins.len(), 0);
for (clause, win) in wins.iter().enumerate() {
let first_doc = unsafe { *win.docs.add(win.pos) };
let last_doc = unsafe { *win.docs.add(win.end - 1) };
let is_dense =
u64::from(last_doc) - u64::from(first_doc) + 1 == (win.end - win.pos) as u64;
let mut pos = win.pos;
for (candidate, &doc) in docs_out.iter().enumerate() {
pos = if is_dense {
win.pos + (doc - first_doc) as usize
} else {
unsafe { find_next_geq(win.docs, pos, win.end, doc) }
};
debug_assert!(
pos < win.end,
"intersection must be present in every clause"
);
offs_out[candidate * wins.len() + clause] = pos as u8;
}
}
}
let mut candidates = TopKCollector::new(limit, std::cmp::min(limit, BLOCK_SIZE * 10));
let mut num_comparisons: usize = 0;
let mut wins: Vec<WindowList> = Vec::with_capacity(num_lists);
let mut batch_docs: Vec<u32> = Vec::with_capacity(MAX_POSTING_BLOCK_SIZE);
let mut batch_offs: Vec<u8> = Vec::with_capacity(MAX_POSTING_BLOCK_SIZE * num_lists);
let mut batch_lens: Vec<u32> = Vec::with_capacity(MAX_POSTING_BLOCK_SIZE);
let mut batch_norms: Vec<u8> = Vec::with_capacity(MAX_POSTING_BLOCK_SIZE);
let mut cursor_scratch: Vec<usize> = Vec::with_capacity(num_lists);
let mut intersection_scratch: Vec<u32> = Vec::new();
let mut norm_k = None;
let mut norm_k_initialized = false;
const FREQ_LUT_BUCKETS: usize = 64;
let mut freq_bound_lut: Option<[f32; FREQ_LUT_BUCKETS]> = None;
let mut target: u64 = 0;
for posting in &self.lead {
match posting.current_doc_id() {
Some(doc) => target = target.max(doc),
None => return Ok(vec![]),
}
}
'window: loop {
self.raise_to_shared_floor(params.wand_factor);
let window_started_with_floor = self.threshold > 0.0;
if window_started_with_floor {
if num_lists >= 2 && freq_bound_lut.is_none() {
let mut lut = [f32::INFINITY; FREQ_LUT_BUCKETS];
for (freq, slot) in lut.iter_mut().enumerate().take(FREQ_LUT_BUCKETS - 1) {
*slot = self.lead[0].score(&self.scorer, freq as u32, 0);
}
lut[FREQ_LUT_BUCKETS - 1] =
self.lead[0].frequency_clamp_upper_bound(&self.scorer);
freq_bound_lut = Some(lut);
}
let advanced = self.and_advance_target(target);
if advanced == TERMINATED_DOC_ID {
break;
}
target = advanced;
}
debug_assert!(target <= u32::MAX as u64);
let target32 = target as u32;
let mut win_end = TERMINATED_DOC_ID;
for j in 0..num_lists {
let (block_idx, block_up_to) = {
let posting = &self.lead[j];
let PostingList::Compressed(ref list) = posting.list else {
unreachable!("bulk AND requires compressed postings");
};
let block_idx = posting.block_idx_for_doc(list, posting.block_idx, target32);
let block_up_to = if block_idx + 1 < list.blocks.len() {
u64::from(list.block_least_doc_id(block_idx + 1)).saturating_sub(1)
} else {
TERMINATED_DOC_ID
};
(block_idx, block_up_to.max(target))
};
self.lead[j].block_idx = block_idx;
win_end = win_end.min(block_up_to);
}
let win_end32 = u32::try_from(win_end).unwrap_or(u32::MAX);
wins.clear();
let mut skip_window = false;
let mut exhausted = false;
for posting in &self.lead {
let PostingList::Compressed(ref list) = posting.list else {
unreachable!("bulk AND requires compressed postings");
};
let block_idx = posting.block_idx;
let state = unsafe { &mut *posting.ensure_compressed_doc_ids_ptr(list, block_idx) };
let lo = state.doc_ids.partition_point(|&doc| doc < target32);
let hi = if win_end32 == u32::MAX {
state.doc_ids.len()
} else {
lo + state.doc_ids[lo..].partition_point(|&doc| doc <= win_end32)
};
if lo == hi {
if block_idx + 1 >= list.blocks.len()
&& state.doc_ids.last().is_none_or(|&doc| doc < target32)
{
exhausted = true;
}
skip_window = true;
break;
}
wins.push(WindowList {
docs: state.doc_ids.as_ptr(),
freqs: std::ptr::null(),
pos: lo,
end: hi,
block_start: block_idx << list.block_shift(),
});
}
if exhausted {
break 'window;
}
if !skip_window {
let mut others_block_max = if window_started_with_floor {
Some(
self.lead[1..]
.iter()
.map(|posting| f64::from(posting.block_max_score(&self.scorer)))
.sum::<f64>(),
)
} else {
None
};
batch_docs.clear();
batch_offs.clear();
let freq_cannot_beat = if window_started_with_floor && num_lists >= 2 {
let posting = &self.lead[0];
let PostingList::Compressed(ref list) = posting.list else {
unreachable!("bulk AND requires compressed postings");
};
let state = unsafe {
&mut *posting.ensure_compressed_block_ptr(list, posting.block_idx)
};
wins[0].freqs = state.freqs.as_ptr();
let freq_bound_lut = freq_bound_lut
.as_ref()
.expect("positive threshold should initialize the frequency bound LUT");
std::array::from_fn(|frequency| {
score_sum_cannot_compete(
freq_bound_lut[frequency],
others_block_max.expect("positive floor should initialize bounds"),
self.threshold,
score_sum_upper_bound_factor(num_lists),
CompetitiveFloorMode::Exclusive,
)
})
} else {
[false; FREQ_LUT_BUCKETS]
};
#[cfg(target_arch = "x86_64")]
let use_avx2 = *HAS_AVX2;
#[cfg(not(target_arch = "x86_64"))]
let use_avx2 = false;
match (num_lists, use_avx2, window_started_with_floor) {
(1, _, _) => merge_window_1(&wins, &mut batch_docs, &mut batch_offs),
#[cfg(target_arch = "x86_64")]
(2, true, false) => unsafe {
merge_window_docs_2_avx2(&wins, &mut batch_docs, &mut batch_offs)
},
#[cfg(target_arch = "x86_64")]
(3, true, false) => unsafe {
merge_window_docs_3_avx2(&wins, &mut batch_docs, &mut batch_offs)
},
(2, _, false) => unsafe {
merge_window_docs_2(&wins, &mut batch_docs, &mut batch_offs)
},
(3, _, false) => unsafe {
merge_window_docs_3(&wins, &mut batch_docs, &mut batch_offs)
},
(_, _, false) if use_pairwise_intersection => {
#[cfg(test)]
{
self.and_window_stats.pairwise_intersections += 1;
}
merge_window_docs_pairwise(
&wins,
&mut intersection_scratch,
&mut batch_docs,
&mut batch_offs,
)
}
(_, _, false) => merge_window_docs_n(
&wins,
&mut cursor_scratch,
&mut batch_docs,
&mut batch_offs,
),
#[cfg(target_arch = "x86_64")]
(2, true, true) => unsafe {
merge_window_2_avx2(
&wins,
&freq_cannot_beat,
&mut batch_docs,
&mut batch_offs,
)
},
#[cfg(target_arch = "x86_64")]
(3, true, true) => unsafe {
merge_window_3_avx2(
&wins,
&freq_cannot_beat,
&mut batch_docs,
&mut batch_offs,
)
},
(2, _, true) => unsafe {
merge_window_2(&wins, &freq_cannot_beat, &mut batch_docs, &mut batch_offs)
},
(3, _, true) => unsafe {
merge_window_3(&wins, &freq_cannot_beat, &mut batch_docs, &mut batch_offs)
},
(_, _, true) => merge_window_n(
&wins,
&freq_cannot_beat,
&mut cursor_scratch,
&mut batch_docs,
&mut batch_offs,
),
}
if !batch_docs.is_empty() {
for (win, posting) in wins.iter_mut().zip(self.lead.iter()) {
let PostingList::Compressed(ref list) = posting.list else {
unreachable!("bulk AND requires compressed postings");
};
let state = unsafe {
&mut *posting.ensure_compressed_block_ptr(list, posting.block_idx)
};
win.freqs = state.freqs.as_ptr();
}
if !norm_k_initialized {
norm_k = self.norm_k_cache();
norm_k_initialized = true;
}
}
batch_lens.clear();
batch_norms.clear();
let norm_k_ref = norm_k
.as_ref()
.map(|(norms, cache)| (*norms, cache.as_ref()));
match norm_k_ref {
Some((norms, _)) => {
for &doc in batch_docs.iter() {
batch_norms.push(norms[doc as usize]);
}
}
None => match self.documents.scoring_norms() {
Some(norms) => {
for &doc in batch_docs.iter() {
batch_lens.push(dequantize_doc_length(norms[doc as usize]));
}
}
None => {
for &doc in batch_docs.iter() {
batch_lens.push(self.documents.scoring_num_tokens(doc));
}
}
},
}
for (index, &doc) in batch_docs.iter().enumerate() {
let (norm_addend, doc_length) = match norm_k_ref {
Some((_, cache)) => {
let code = batch_norms[index];
(Some(cache[code as usize]), dequantize_doc_length(code))
}
None => (None, batch_lens[index]),
};
let offs = &batch_offs[index * num_lists..(index + 1) * num_lists];
if self.threshold > 0.0 && num_lists >= 2 && others_block_max.is_none() {
others_block_max = Some(
self.lead[1..]
.iter()
.map(|posting| f64::from(posting.block_max_score(&self.scorer)))
.sum::<f64>(),
);
}
if self.threshold > 0.0
&& num_lists >= 2
&& let Some(others_block_max) = others_block_max
{
let first_freq = unsafe { *wins[0].freqs.add(offs[0] as usize) };
let first_score = match norm_addend {
Some(addend) => {
self.lead[0].query_weight
* bm25_doc_weight_with_norm(first_freq, addend)
}
None => self.lead[0].score(&self.scorer, first_freq, doc_length),
};
if score_sum_cannot_compete(
first_score,
others_block_max,
self.threshold,
score_sum_upper_bound_factor(num_lists),
CompetitiveFloorMode::Exclusive,
) {
continue;
}
}
#[cfg(test)]
{
self.and_window_stats.candidates_returned += 1;
}
num_comparisons += 1;
let Some(document_key) = self.documents.document_key_for_doc_id(doc) else {
continue;
};
let early_score = (self.threshold > 0.0).then(|| {
window_bm25_score(
&self.lead,
&wins,
offs,
&score_order,
&self.scorer,
norm_addend,
doc_length,
)
});
if let Some(slop) = phrase_slop {
if early_score
.is_some_and(|score| self.exclusive_score_cannot_beat_floor(score))
{
continue;
}
for ((win, posting), &off) in
wins.iter().zip(self.lead.iter_mut()).zip(offs.iter())
{
posting.index = win.block_start + off as usize;
posting.current_doc = Some(DocInfo::Raw(RawDocInfo {
doc_id: doc,
frequency: unsafe { *win.freqs.add(off as usize) },
}));
}
let matched = if slop == 0 {
self.check_exact_positions_bulk()?
} else {
self.check_positions(slop as i32)?
};
if !matched {
continue;
}
}
let score = early_score.unwrap_or_else(|| {
window_bm25_score(
&self.lead,
&wins,
offs,
&score_order,
&self.scorer,
norm_addend,
doc_length,
)
});
if candidates.insert(
ScoredDoc::new(document_key, score),
doc_length,
u64::from(doc),
wins.iter().zip(self.lead.iter()).zip(offs.iter()).map(
|((win, posting), &off)| {
(posting.term_index(), unsafe {
*win.freqs.add(off as usize)
})
},
),
)? && let Some(kth) = candidates.kth_score_if_full()
{
self.update_threshold(kth, params.wand_factor);
}
}
}
if win_end == TERMINATED_DOC_ID {
break;
}
target = win_end + 1;
}
metrics.record_comparisons(num_comparisons);
candidates.into_candidates(|key| self.documents.candidate_from_key(key))
}
fn and_move_to_next_block(&mut self, target: u64) {
self.invalidate_score_first_and_window();
if self.threshold <= 0.0 {
self.up_to = Some(if self.score_first_and_enabled {
TERMINATED_DOC_ID
} else {
target
});
self.and_max_score = f32::INFINITY;
return;
}
if self.lead.is_empty() {
self.up_to = Some(TERMINATED_DOC_ID);
self.and_max_score = 0.0;
return;
}
for posting in &mut self.lead {
posting.shallow_next(target);
}
let narrow_up_to = self
.lead
.iter()
.map(|posting| Self::posting_block_up_to(posting, target))
.min()
.unwrap_or(TERMINATED_DOC_ID);
let narrow_max_score = conservative_score_sum(
self.lead
.iter()
.map(|posting| posting.block_max_score(&self.scorer)),
);
if narrow_max_score >= self.threshold {
self.up_to = Some(narrow_up_to);
self.and_max_score = narrow_max_score;
self.prepare_score_first_and_window();
#[cfg(test)]
{
self.and_window_stats.windows_narrow += 1;
}
return;
}
let lead_up_to = self
.lead
.first()
.map(|posting| Self::posting_block_up_to(posting, target))
.unwrap_or(TERMINATED_DOC_ID);
let can_try_wide = lead_up_to > narrow_up_to
&& lead_up_to != TERMINATED_DOC_ID
&& self.lead.iter().all(|posting| posting.is_compressed());
if can_try_wide {
let mut wide_bounds = SmallVec::<[f32; 8]>::new();
#[cfg(test)]
let mut range_blocks_scanned = 0;
for posting in &mut self.lead {
let block_max = posting.block_max_score_up_to_with_stats(lead_up_to, &self.scorer);
wide_bounds.push(block_max.score);
#[cfg(test)]
{
range_blocks_scanned += block_max.blocks_scanned;
}
}
let wide_max_score = conservative_score_sum(wide_bounds.into_iter());
#[cfg(test)]
{
self.and_window_stats.range_blocks_scanned += range_blocks_scanned;
}
if wide_max_score < self.threshold {
self.up_to = Some(lead_up_to);
self.and_max_score = wide_max_score;
#[cfg(test)]
{
self.and_window_stats.windows_wide += 1;
}
return;
}
}
self.up_to = Some(narrow_up_to);
self.and_max_score = narrow_max_score;
self.prepare_score_first_and_window();
#[cfg(test)]
{
self.and_window_stats.windows_narrow += 1;
}
}
fn and_advance_target(&mut self, mut target: u64) -> u64 {
if self.up_to.is_none_or(|up_to| target > up_to) {
self.and_move_to_next_block(target);
}
loop {
let Some(up_to) = self.up_to else {
return TERMINATED_DOC_ID;
};
if self.and_max_score >= self.threshold {
return target;
}
#[cfg(test)]
{
self.and_window_stats.windows_skipped += 1;
}
if up_to == TERMINATED_DOC_ID {
return TERMINATED_DOC_ID;
}
target = up_to + 1;
self.and_move_to_next_block(target);
}
}
#[allow(clippy::vec_box)]
fn head_doc(&self) -> Option<u64> {
self.head.peek().map(HeadPosting::doc_id)
}
fn push_head(&mut self, posting: Box<PostingIterator>) {
if posting.doc().is_some() {
self.head.push(HeadPosting::new(posting));
}
}
fn move_head_doc_to_lead(&mut self, target: u64) {
while self.head_doc() == Some(target) {
if let Some(posting) = self.head.pop() {
self.lead.push(posting.posting);
}
}
}
fn move_head_before_target_to_tail(&mut self, target: u64) {
if self.threshold <= 0.0 {
while matches!(self.head_doc(), Some(doc_id) if doc_id < target) {
if let Some(mut posting) = self.head.pop().map(|posting| posting.posting) {
posting.next(target);
self.push_head(posting);
}
}
return;
}
while matches!(self.head_doc(), Some(doc_id) if doc_id < target) {
if let Some(posting) = self.head.pop() {
let upper_bound = posting.posting.global_upper_bound(&self.scorer);
if let Some(mut evicted) =
self.insert_tail_with_overflow(posting.posting, upper_bound)
{
evicted.next(target);
self.push_head(evicted);
}
}
}
}
fn or_block_window_max(&self) -> f32 {
conservative_score_sum(
self.lead
.iter()
.map(|posting| posting.window_max_score(self.up_to, &self.scorer))
.chain(
self.head
.iter()
.map(|posting| posting.posting.window_max_score(self.up_to, &self.scorer)),
)
.chain(self.tail.iter().map(|posting| posting.upper_bound)),
)
}
fn can_target_beat_threshold(&mut self, target: u64) -> bool {
if self.up_to.is_none_or(|up_to| target > up_to) {
self.update_max_scores(target);
}
let mut possible_matches = self.lead.len();
for posting in &self.tail {
if matches!(posting.posting.block_first_doc(), Some(block_doc) if block_doc <= target) {
possible_matches += 1;
}
}
let sum = conservative_score_sum(
self.lead
.iter()
.map(|posting| posting.window_max_score(self.up_to, &self.scorer))
.chain(
self.tail
.iter()
.filter(|posting| {
matches!(
posting.posting.block_first_doc(),
Some(block_doc) if block_doc <= target
)
})
.map(|posting| posting.posting.window_max_score(self.up_to, &self.scorer)),
),
);
match self.operator {
Operator::And => {
possible_matches >= self.num_terms
&& !self
.floor_mode
.rejects_upper_bound(f64::from(sum), self.threshold)
}
Operator::Or => !self
.floor_mode
.rejects_upper_bound(f64::from(sum), self.threshold),
}
}
fn update_max_scores(&mut self, target: u64) {
let lead_cost = self
.lead
.iter()
.map(|posting| posting.cost())
.min()
.unwrap_or(usize::MAX);
let mut narrow_up_to = TERMINATED_DOC_ID;
for posting in &mut self.lead {
posting.shallow_next(target);
narrow_up_to = narrow_up_to.min(posting.block_end_doc());
}
let mut head_postings = std::mem::take(&mut self.head).into_vec();
for posting in &mut head_postings {
let doc_id = posting.doc_id();
posting.posting.shallow_next(doc_id);
narrow_up_to = narrow_up_to.min(posting.posting.block_end_doc());
}
let mut tail_postings = std::mem::take(&mut self.tail).into_vec();
for tail_posting in &mut tail_postings {
tail_posting.posting.shallow_next(target);
}
if narrow_up_to == TERMINATED_DOC_ID
&& let Some(top) = tail_postings
.iter()
.min_by_key(|posting| posting.posting.cost())
&& top.posting.cost() <= lead_cost
{
narrow_up_to = narrow_up_to.min(top.posting.block_end_doc().max(target));
}
self.up_to = Some(narrow_up_to);
self.head = BinaryHeap::from(head_postings);
self.tail_max_score = 0.0;
for tail_posting in tail_postings {
let posting = tail_posting.posting;
let upper_bound = match posting.block_first_doc() {
Some(block_doc) if block_doc <= target => {
posting.window_max_score(self.up_to, &self.scorer)
}
_ => 0.0,
};
if let Some(mut evicted) = self.insert_tail_with_overflow(posting, upper_bound) {
evicted.next(target);
self.push_head(evicted);
}
}
}
fn or_group_skip_to(&self) -> Option<u64> {
let mut group_up_to = TERMINATED_DOC_ID;
for posting in &self.lead {
let (doc_up_to, _) = posting.impact_group_bound(&self.scorer)?;
group_up_to = group_up_to.min(doc_up_to);
}
for posting in self.head.iter() {
let (doc_up_to, _) = posting.posting.impact_group_bound(&self.scorer)?;
group_up_to = group_up_to.min(doc_up_to);
}
for tail_posting in self.tail.iter() {
let (doc_up_to, _) = tail_posting.posting.impact_group_bound(&self.scorer)?;
group_up_to = group_up_to.min(doc_up_to);
}
if self.up_to.is_some_and(|up_to| group_up_to <= up_to) {
return None;
}
let mut bounds = SmallVec::<[f32; 8]>::new();
for posting in &self.lead {
bounds.push(posting.impact_group_bound(&self.scorer)?.1);
}
for posting in self
.head
.iter()
.filter(|posting| posting.doc_id() <= group_up_to)
{
bounds.push(posting.posting.impact_group_bound(&self.scorer)?.1);
}
for tail_posting in self.tail.iter().filter(|tail_posting| {
matches!(
tail_posting.posting.block_first_doc(),
Some(block_doc) if block_doc <= group_up_to
)
}) {
bounds.push(tail_posting.posting.impact_group_bound(&self.scorer)?.1);
}
let bounds_sum = conservative_score_sum(bounds.into_iter());
self.floor_mode
.rejects_upper_bound(f64::from(bounds_sum), self.threshold)
.then_some(group_up_to.saturating_add(1))
}
fn refine_or_candidate(&mut self, target: u64, doc_length: u32) -> bool {
if self.threshold <= 0.0 {
return true;
}
let mut lead_score = self
.lead
.iter()
.filter_map(|posting| {
posting
.doc()
.map(|doc| posting.score(&self.scorer, doc.frequency(), doc_length))
})
.sum::<f32>();
while !self.floor_mode.accepts_score(lead_score, self.threshold) {
if score_sum_cannot_compete(
lead_score,
self.tail_upper_bound_sum(),
self.threshold,
score_sum_upper_bound_factor(self.num_terms),
self.floor_mode,
) {
return false;
}
if !self.advance_tail_top(target, doc_length, &mut lead_score) {
return true;
}
}
true
}
fn collect_tail_matches(&mut self, target: u64) {
let mut remaining = Vec::with_capacity(self.tail.len());
let tail = std::mem::take(&mut self.tail);
self.tail_max_score = 0.0;
for tail_posting in tail.into_vec() {
let mut posting = tail_posting.posting;
posting.next(target);
match posting.doc().map(|doc| doc.doc_id()) {
Some(doc_id) if doc_id == target => self.lead.push(posting),
Some(_) => remaining.push(posting),
None => {}
}
}
for posting in remaining {
self.push_head(posting);
}
}
fn advance_tail_and_lead_to_head(&mut self, least_id: u64) {
let mut postings = Vec::with_capacity(self.tail.len() + self.lead.len());
while let Some(tail) = self.tail.pop() {
postings.push(tail.posting);
}
self.tail_max_score = 0.0;
postings.append(&mut self.lead);
for mut posting in postings {
posting.next(least_id);
self.push_head(posting);
}
}
fn advance_lead_to_head(&mut self, least_id: u64) {
let lead = std::mem::take(&mut self.lead);
for mut posting in lead {
posting.next(least_id);
self.push_head(posting);
}
debug_assert!(self.tail.is_empty());
self.clear_tail();
}
fn clear_tail(&mut self) {
self.tail.clear();
self.tail_max_score = 0.0;
}
fn insert_tail(&mut self, posting: Box<PostingIterator>, upper_bound: f32) {
self.tail_max_score = next_up_f64(self.tail_max_score + f64::from(upper_bound));
self.tail
.push(TailPosting::new(upper_bound, posting.cost(), posting));
}
fn insert_tail_with_overflow(
&mut self,
posting: Box<PostingIterator>,
upper_bound: f32,
) -> Option<Box<PostingIterator>> {
if self.threshold <= 0.0 || upper_bound <= 0.0 {
return Some(posting);
}
let parked_upper_bound = self.tail_upper_bound_sum() + f64::from(upper_bound);
if score_sum_cannot_compete(
0.0,
parked_upper_bound,
self.threshold,
score_sum_upper_bound_factor(self.num_terms),
self.floor_mode,
) {
self.insert_tail(posting, upper_bound);
return None;
}
if self.tail.is_empty() {
return Some(posting);
}
let candidate = TailPosting::new(upper_bound, posting.cost(), posting);
if let Some(top) = self.tail.peek()
&& top > &candidate
{
let evicted = self.tail.pop().expect("peeked tail posting should exist");
self.remove_tail_upper_bound(evicted.upper_bound);
self.tail_max_score = next_up_f64(self.tail_max_score + f64::from(upper_bound));
self.tail.push(candidate);
return Some(evicted.posting);
}
Some(candidate.posting)
}
fn lead_to_tail_upper_bound(&self, posting: &PostingIterator, target: u64) -> f32 {
if self.operator == Operator::Or
&& posting.is_compressed()
&& self.up_to.is_some_and(|up_to| target <= up_to)
{
posting.window_max_score(self.up_to, &self.scorer)
} else {
posting.global_upper_bound(&self.scorer)
}
}
fn advance_tail_to_next_or_window(&mut self) -> bool {
if self.operator != Operator::Or || self.tail.is_empty() {
return false;
}
let Some(up_to) = self.up_to else {
return false;
};
if up_to >= u32::MAX as u64 {
return false;
}
if !self
.tail
.iter()
.any(|tail| tail.posting.has_next_compressed_block())
{
return false;
}
self.update_max_scores(up_to + 1);
true
}
fn push_back_leads(&mut self, target: u64) {
if self.threshold <= 0.0 {
while let Some(mut posting) = self.lead.pop() {
posting.next(target);
self.push_head(posting);
}
return;
}
while let Some(posting) = self.lead.pop() {
let upper_bound = self.lead_to_tail_upper_bound(&posting, target);
if let Some(mut evicted) = self.insert_tail_with_overflow(posting, upper_bound) {
evicted.next(target);
self.push_head(evicted);
}
}
}
fn advance_tail_top(&mut self, target: u64, doc_length: u32, lead_score: &mut f32) -> bool {
let Some(TailPosting {
upper_bound,
cost: _,
mut posting,
}) = self.tail.pop()
else {
return false;
};
self.remove_tail_upper_bound(upper_bound);
posting.next(target);
match posting.doc() {
Some(doc) if doc.doc_id() == target => {
*lead_score += posting.score(&self.scorer, doc.frequency(), doc_length);
self.lead.push(posting);
}
Some(_) => self.push_head(posting),
None => {}
}
true
}
#[inline]
fn tail_upper_bound_sum(&self) -> f64 {
self.tail_max_score
}
#[inline]
fn remove_tail_upper_bound(&mut self, upper_bound: f32) {
if self.tail.is_empty() {
self.tail_max_score = 0.0;
return;
}
self.tail_max_score = next_up_f64((self.tail_max_score - f64::from(upper_bound)).max(0.0));
}
fn advance_all_tail(
&mut self,
target: u64,
doc_length: Option<u32>,
mut score: Option<&mut f32>,
) {
let tail = std::mem::take(&mut self.tail);
self.tail_max_score = 0.0;
for tail_posting in tail.into_vec() {
let mut posting = tail_posting.posting;
posting.next(target);
match posting.doc() {
Some(doc) if doc.doc_id() == target => {
if let (Some(doc_length), Some(score)) = (doc_length, score.as_deref_mut()) {
*score += posting.score(&self.scorer, doc.frequency(), doc_length);
}
self.lead.push(posting)
}
Some(_) => self.push_head(posting),
None => {}
}
}
}
fn current_doc_postings(&self) -> Vec<&PostingIterator> {
if !self.lead.is_empty() {
return self.lead.iter().map(|posting| posting.as_ref()).collect();
}
let Some(target) = self.head_doc() else {
return Vec::new();
};
self.head
.iter()
.filter(|posting| posting.doc_id() == target)
.map(|posting| posting.posting.as_ref())
.collect()
}
fn check_positions(&self, slop: i32) -> Result<bool> {
#[cfg(test)]
{
self.phrase_position_checks
.set(self.phrase_position_checks.get() + 1);
}
if slop == 0 {
return self.check_exact_positions();
}
let mut position_iters = self
.current_doc_postings()
.into_iter()
.map(PostingIterator::position_cursor)
.collect::<Result<Vec<_>>>()?;
position_iters.sort_unstable_by_key(|iter| iter.position_in_query);
loop {
let mut max_relative_pos = None;
let mut all_same = true;
for window in position_iters.windows(2) {
let last = window[0].relative_position();
let next = window[1].relative_position();
let (Some(last), Some(next)) = (last, next) else {
return Ok(false);
};
let move_to = if last > next {
last
} else {
std::cmp::max(last + 1, next - slop)
};
max_relative_pos = max_relative_pos.max(Some(move_to));
if !(last <= next && next <= last + slop) {
all_same = false;
break;
}
}
if all_same {
return Ok(true);
}
position_iters.iter_mut().for_each(|iter| {
iter.advance_to_relative(max_relative_pos.unwrap());
});
}
}
fn check_exact_positions_bulk(&self) -> Result<bool> {
#[cfg(test)]
{
self.phrase_position_checks
.set(self.phrase_position_checks.get() + 1);
}
const MAX_INLINE_CLAUSES: usize = 16;
let num_clauses = self.lead.len();
if num_clauses > MAX_INLINE_CLAUSES {
return self.check_exact_positions();
}
let mut cursors: [Option<PositionCursor<'_>>; MAX_INLINE_CLAUSES] =
std::array::from_fn(|_| None);
let mut anchor_idx = 0usize;
let mut anchor_len = usize::MAX;
for (index, (slot, posting)) in cursors.iter_mut().zip(self.lead.iter()).enumerate() {
let cursor = posting.position_cursor()?;
if cursor.len() < anchor_len {
anchor_len = cursor.len();
anchor_idx = index;
}
*slot = Some(cursor);
}
let anchor = cursors[anchor_idx]
.as_ref()
.expect("anchor cursor was just populated");
let anchor_offset = anchor.position_in_query as u32;
'anchor: for &anchor_position in anchor.positions.as_slice() {
let Some(base) = anchor_position.checked_sub(anchor_offset) else {
continue;
};
for (index, slot) in cursors[..num_clauses].iter().enumerate() {
if index == anchor_idx {
continue;
}
let cursor = slot.as_ref().expect("clause cursor was just populated");
let Some(target) = base.checked_add(cursor.position_in_query as u32) else {
return Ok(false);
};
if cursor.positions.as_slice().binary_search(&target).is_err() {
continue 'anchor;
}
}
return Ok(true);
}
Ok(false)
}
fn check_exact_positions(&self) -> Result<bool> {
let mut position_iters = self
.current_doc_postings()
.into_iter()
.map(PostingIterator::position_cursor)
.collect::<Result<Vec<_>>>()?;
position_iters.sort_unstable_by_key(|iter| iter.len());
let Some(lead) = position_iters.first() else {
return Ok(false);
};
let lead_position = lead.position_in_query;
loop {
let Some(anchor) = position_iters[0].absolute_position() else {
return Ok(false);
};
let Some(base) = anchor.checked_sub(lead_position as u32) else {
position_iters[0].advance_next();
continue;
};
let mut next_lead_relative = None;
let mut matched = true;
for follower in position_iters.iter_mut().skip(1) {
let Some(target) = base.checked_add(follower.position_in_query as u32) else {
return Ok(false);
};
let Some(position) = follower.advance_to_absolute(target) else {
return Ok(false);
};
if position != target {
next_lead_relative = Some(position as i32 - follower.position_in_query);
matched = false;
break;
}
}
if matched {
return Ok(true);
}
position_iters[0].advance_to_relative(next_lead_relative.unwrap());
}
}
}
#[derive(Debug)]
struct RecycledPositionValues<'a> {
values: Option<Vec<u32>>,
pool: &'a RefCell<Option<Vec<u32>>>,
}
impl<'a> RecycledPositionValues<'a> {
fn new(values: Vec<u32>, pool: &'a RefCell<Option<Vec<u32>>>) -> Self {
Self {
values: Some(values),
pool,
}
}
fn as_slice(&self) -> &[u32] {
self.values
.as_deref()
.expect("position values are present until drop")
}
}
impl Drop for RecycledPositionValues<'_> {
fn drop(&mut self) {
let values = self
.values
.take()
.expect("position values are present until drop");
let mut pool = self.pool.borrow_mut();
if pool.is_none() {
*pool = Some(values);
}
}
}
#[derive(Debug)]
enum PositionValues<'a> {
Recycled(RecycledPositionValues<'a>),
Owned(Vec<u32>),
}
impl<'a> PositionValues<'a> {
fn as_slice(&self) -> &[u32] {
match self {
Self::Recycled(values) => values.as_slice(),
Self::Owned(values) => values.as_slice(),
}
}
fn len(&self) -> usize {
self.as_slice().len()
}
}
#[derive(Debug)]
struct PositionCursor<'a> {
positions: PositionValues<'a>,
pub position_in_query: i32,
index: usize,
}
impl<'a> PositionCursor<'a> {
fn new(positions: PositionValues<'a>, position_in_query: i32) -> Self {
Self {
positions,
position_in_query,
index: 0,
}
}
fn len(&self) -> usize {
self.positions.len()
}
fn absolute_position(&self) -> Option<u32> {
self.positions.as_slice().get(self.index).copied()
}
fn relative_position(&self) -> Option<i32> {
self.positions
.as_slice()
.get(self.index)
.map(|position| *position as i32 - self.position_in_query)
}
fn advance_to_relative(&mut self, least_relative_pos: i32) {
if self.index >= self.len() {
return;
}
let least_pos = least_relative_pos + self.position_in_query;
let least_pos = least_pos.max(0) as u32;
let values = self.positions.as_slice();
self.index += values[self.index..].partition_point(|&pos| pos < least_pos);
}
fn advance_to_absolute(&mut self, least_pos: u32) -> Option<u32> {
if self.index >= self.len() {
return None;
}
let values = self.positions.as_slice();
self.index += values[self.index..].partition_point(|&pos| pos < least_pos);
self.absolute_position()
}
fn advance_next(&mut self) {
self.index = self.index.saturating_add(1).min(self.len());
}
}
pub(super) struct WandCursor<'a, D: WandDocuments> {
wand: Wand<'a, Arc<MemBM25Scorer>, D>,
phrase_slop: Option<u32>,
wand_factor: f32,
cost: usize,
global_score_upper_bound: OnceCell<Option<f32>>,
current_doc: Option<DocInfo>,
current_document_key: Option<u64>,
current_score: f32,
confirmation: Option<bool>,
shallow: Option<(u64, u64, f32)>,
comparisons: usize,
metrics_recorded: bool,
metrics: &'a dyn MetricsCollector,
}
impl<'a, D: WandDocuments> WandCursor<'a, D> {
pub(super) fn new(
operator: Operator,
postings: Vec<PostingIterator>,
documents: &'a D,
scorer: Arc<MemBM25Scorer>,
params: &FtsSearchParams,
metrics: &'a dyn MetricsCollector,
) -> Self {
let cost = match operator {
Operator::And => postings.iter().map(PostingIterator::cost).min(),
Operator::Or => Some(
postings
.iter()
.map(PostingIterator::cost)
.fold(0, usize::saturating_add),
),
}
.unwrap_or_default()
.min(documents.visible_cost_upper_bound());
let mut wand = Wand::new(operator, postings.into_iter(), documents, scorer)
.with_floor_mode(CompetitiveFloorMode::Inclusive);
if params.phrase_slop.is_some() {
wand.score_first_and_enabled = false;
}
Self {
wand,
phrase_slop: params.phrase_slop,
wand_factor: params.wand_factor,
cost,
global_score_upper_bound: OnceCell::new(),
current_doc: None,
current_document_key: None,
current_score: 0.0,
confirmation: None,
shallow: None,
comparisons: 0,
metrics_recorded: false,
metrics,
}
}
pub(super) fn doc(&self) -> Option<u64> {
self.current_doc.map(|doc| doc.doc_id())
}
pub(super) fn document_key(&self) -> Option<u64> {
self.current_document_key
}
fn clear_current(&mut self) {
self.current_doc = None;
self.current_document_key = None;
self.current_score = 0.0;
self.confirmation = None;
self.shallow = None;
}
fn record_metrics(&mut self) {
if !self.metrics_recorded {
self.metrics.record_comparisons(self.comparisons);
self.metrics_recorded = true;
}
}
fn position_next(&mut self) -> Result<Option<u64>> {
loop {
let Some((doc, and_score)) = self.wand.next()? else {
self.clear_current();
self.record_metrics();
return Ok(None);
};
self.comparisons += 1;
let doc_id = doc.doc_id();
let Some(document_key) = self.wand.documents.document_key(&doc) else {
if self.wand.operator == Operator::Or {
self.wand.push_back_leads(doc_id.saturating_add(1));
}
continue;
};
let doc_length = self.wand.documents.doc_length(&doc);
self.wand.advance_all_tail(doc_id, None, None);
let score = if self.wand.and_candidate_score.is_some() {
and_score
} else {
self.wand.score_in_query_order(doc_length)
};
self.current_doc = Some(doc);
self.current_document_key = Some(document_key);
self.current_score = score;
self.confirmation = self.phrase_slop.is_none().then_some(true);
self.shallow = None;
return Ok(Some(doc_id));
}
}
pub(super) fn next(&mut self) -> Result<Option<u64>> {
if let Some(doc) = self.current_doc
&& self.wand.operator == Operator::Or
{
self.wand.push_back_leads(doc.doc_id().saturating_add(1));
}
self.clear_current();
self.position_next()
}
pub(super) fn advance(&mut self, target: u64) -> Result<Option<u64>> {
if self.doc().is_some_and(|doc| doc >= target) {
return Ok(self.doc());
}
self.clear_current();
self.wand.seek(target);
self.position_next()
}
pub(super) fn cost(&self) -> usize {
self.cost
}
pub(super) fn global_score_upper_bound(&self) -> Option<f32> {
*self
.global_score_upper_bound
.get_or_init(|| self.wand.compound_global_score_upper_bound())
}
pub(super) fn current_score(&self) -> Result<f32> {
self.current_doc
.map(|_| self.current_score)
.ok_or_else(|| Error::internal("posting FTS scorer is not positioned on a document"))
}
pub(super) fn matches(&mut self) -> Result<bool> {
let Some(_) = self.current_doc else {
return Ok(false);
};
if let Some(confirmed) = self.confirmation {
return Ok(confirmed);
}
let phrase_slop = self.phrase_slop.ok_or_else(|| {
Error::internal("posting FTS scorer requires phrase slop for position confirmation")
})?;
let confirmed = self.wand.check_positions(phrase_slop as i32)?;
self.confirmation = Some(confirmed);
Ok(confirmed)
}
pub(super) fn match_cost(&self) -> Option<f32> {
self.phrase_slop.map(|_| self.wand.num_terms.max(1) as f32)
}
pub(super) fn advance_shallow(&mut self, target: u64) -> Result<u64> {
let (up_to, upper) = self.wand.compound_shallow_bound(target);
self.shallow = Some((target, up_to, upper));
Ok(up_to)
}
pub(super) fn score_upper_bound(&self, up_to: u64) -> Result<f32> {
let (target, shallow_up_to, upper) = self.shallow.ok_or_else(|| {
Error::internal("score bound requires advance_shallow on the posting FTS scorer")
})?;
if up_to < target || up_to > shallow_up_to {
return Err(Error::internal(format!(
"posting FTS score bound up_to={up_to} is outside shallow range [{target}, {shallow_up_to}]"
)));
}
Ok(upper)
}
pub(super) fn set_min_competitive_score(&mut self, min_score: f32) -> Result<()> {
if min_score.is_nan() {
return Err(Error::invalid_input(
"minimum competitive FTS score cannot be NaN",
));
}
let floor = min_score * self.wand_factor;
if floor > self.wand.threshold {
if self.wand.score_first_and_enabled && self.wand.threshold <= 0.0 && floor > 0.0 {
self.wand.up_to = None;
self.wand.invalidate_score_first_and_window();
}
self.wand.threshold = floor;
}
Ok(())
}
}
impl<D: WandDocuments> Drop for WandCursor<'_, D> {
fn drop(&mut self) {
self.record_metrics();
}
}
impl<S: Scorer, D: WandDocuments> Wand<'_, S, D> {
fn compound_global_score_upper_bound(&self) -> Option<f32> {
if self.lead.len() + self.head.len() + self.tail.len() != self.num_terms {
return None;
}
if self.lead.iter().any(|posting| posting.has_grouped_terms())
|| self
.head
.iter()
.any(|posting| posting.posting.has_grouped_terms())
|| self
.tail
.iter()
.any(|posting| posting.posting.has_grouped_terms())
{
return None;
}
let upper = conservative_score_sum(
self.lead
.iter()
.map(|posting| posting.global_upper_bound(&self.scorer))
.chain(
self.head
.iter()
.map(|posting| posting.posting.global_upper_bound(&self.scorer)),
)
.chain(
self.tail
.iter()
.map(|posting| posting.posting.global_upper_bound(&self.scorer)),
),
);
(upper.is_finite() && upper >= 0.0).then_some(upper)
}
fn seek(&mut self, target: u64) {
self.up_to = None;
self.and_max_score = f32::INFINITY;
self.and_last_doc = None;
self.invalidate_score_first_and_window();
if self.operator == Operator::And {
for posting in &mut self.lead {
if posting.doc().is_some_and(|doc| doc.doc_id() < target) {
posting.next(target);
}
}
return;
}
let mut postings = std::mem::take(&mut self.head)
.into_vec()
.into_iter()
.map(|posting| posting.posting)
.chain(self.lead.drain(..))
.chain(
std::mem::take(&mut self.tail)
.into_vec()
.into_iter()
.map(|posting| posting.posting),
)
.collect::<Vec<_>>();
self.tail_max_score = 0.0;
for posting in &mut postings {
if posting.doc().is_some_and(|doc| doc.doc_id() < target) {
posting.next(target);
}
}
self.head = postings
.into_iter()
.filter(|posting| posting.doc().is_some())
.map(HeadPosting::new)
.collect();
}
fn compound_shallow_bound(&mut self, target: u64) -> (u64, f32) {
if self.operator == Operator::Or {
self.update_max_scores(target);
return (
self.up_to.unwrap_or(TERMINATED_DOC_ID),
conservative_score_sum(
self.lead
.iter()
.map(|posting| posting.window_max_score(self.up_to, &self.scorer))
.chain(self.head.iter().map(|posting| {
posting.posting.window_max_score(self.up_to, &self.scorer)
}))
.chain(self.tail.iter().map(|posting| posting.upper_bound)),
),
);
}
let saved_block_indices = self.score_first_and_enabled.then(|| {
self.lead
.iter()
.map(|posting| posting.block_idx)
.collect::<SmallVec<[usize; 8]>>()
});
let mut up_to = TERMINATED_DOC_ID;
for posting in &mut self.lead {
posting.shallow_next(target);
up_to = up_to.min(Self::posting_block_up_to(posting, target));
}
let upper = conservative_score_sum(
self.lead
.iter()
.map(|posting| posting.block_max_score(&self.scorer)),
);
if let Some(saved_block_indices) = saved_block_indices {
for (posting, block_idx) in self.lead.iter_mut().zip(saved_block_indices) {
posting.block_idx = block_idx;
}
}
(up_to, upper)
}
}
fn conservative_score_sum(scores: impl Iterator<Item = f32>) -> f32 {
let (num_scores, exact) = scores.fold((0, 0.0_f64), |(count, sum), score| {
(count + 1, sum + f64::from(score))
});
let widened = exact * score_sum_upper_bound_factor(num_scores);
outward_f32_upper_bound(widened)
}
#[inline]
pub(super) fn outward_f32_upper_bound(value: f64) -> f32 {
let rounded = value as f32;
if f64::from(rounded) < value {
next_up_f32(rounded)
} else {
rounded
}
}
fn next_up_f32(value: f32) -> f32 {
if !value.is_finite() {
return value;
}
if value == 0.0 {
return f32::from_bits(1);
}
let bits = value.to_bits();
if value > 0.0 {
f32::from_bits(bits + 1)
} else {
f32::from_bits(bits - 1)
}
}
#[inline]
fn next_up_f64(value: f64) -> f64 {
if !value.is_finite() {
return value;
}
if value == 0.0 {
return f64::from_bits(1);
}
let bits = value.to_bits();
if value > 0.0 {
f64::from_bits(bits + 1)
} else {
f64::from_bits(bits - 1)
}
}
#[cfg(test)]
mod tests {
use arrow::buffer::ScalarBuffer;
use rstest::rstest;
use std::sync::atomic::{AtomicUsize, Ordering};
use super::super::documents::resident_row_address_projection_for_test;
use super::super::impact::build_impact_skip_data;
use super::*;
use crate::scalar::inverted::scorer::{IndexBM25Scorer, MemBM25Scorer};
use crate::{
metrics::{LocalMetricsCollector, NoOpMetricsCollector},
scalar::inverted::{
CompressedPostingList, PlainPostingList, PostingListBuilder, SharedPositionStream,
builder::PositionRecorder,
encoding::{
compress_posting_list, compress_posting_list_with_tail_codec_and_block_size,
encode_position_stream_block_into,
},
},
};
struct CostOnlyDocuments {
total_docs: usize,
visible_cost_upper_bound: usize,
}
impl WandDocuments for CostOnlyDocuments {
type Candidate = u64;
fn len(&self) -> usize {
self.total_docs
}
fn visible_cost_upper_bound(&self) -> usize {
self.visible_cost_upper_bound
}
fn scoring_norms(&self) -> Option<&[u8]> {
None
}
fn scoring_num_tokens(&self, _doc_id: u32) -> u32 {
1
}
fn doc_length(&self, _doc: &DocInfo) -> u32 {
1
}
fn document_key(&self, doc: &DocInfo) -> Option<u64> {
Some(doc.doc_id())
}
fn document_key_for_doc_id(&self, doc_id: u32) -> Option<u64> {
Some(u64::from(doc_id))
}
fn candidate_from_key(&self, key: u64) -> Self::Candidate {
key
}
fn flat_documents(&self) -> Option<FlatDocuments<'_>> {
None
}
fn flat_doc_length(&self, _doc_id: u64, _document_key: u64, _compressed: bool) -> u32 {
1
}
}
#[test]
fn wand_cursor_cost_uses_materialized_visibility_upper_bound() {
let total_docs = 10;
let selected = DocVisibility::Selected(roaring::RoaringBitmap::from_iter([1, 7]));
let selected_bound = <&DocVisibility as ModernVisibility>::len(&&selected, total_docs);
let filtered = DocVisibility::Filtered {
projection: resident_row_address_projection_for_test((0..total_docs as u64).collect()),
mask: Arc::new(RowAddrMask::all_rows()),
};
let filtered_bound = <&DocVisibility as ModernVisibility>::len(&&filtered, total_docs);
let all_bound = AllModernDocuments.len(total_docs);
assert_eq!(selected_bound, 2);
assert_eq!(filtered_bound, total_docs);
assert_eq!(all_bound, total_docs);
let scorer = Arc::new(MemBM25Scorer::new(
total_docs as u64,
total_docs,
std::collections::HashMap::from([("term".to_owned(), 8)]),
));
let posting = || {
PostingIterator::new(
"term".to_owned(),
0,
0,
generate_posting_list((0..8).collect(), 1.0, None, true),
total_docs,
)
};
let params = FtsSearchParams::default();
let metrics = NoOpMetricsCollector;
let selected_documents = CostOnlyDocuments {
total_docs,
visible_cost_upper_bound: selected_bound,
};
let selected_cursor = WandCursor::new(
Operator::Or,
vec![posting()],
&selected_documents,
scorer.clone(),
¶ms,
&metrics,
);
assert_eq!(selected_cursor.cost(), 2);
for visible_cost_upper_bound in [filtered_bound, all_bound] {
let documents = CostOnlyDocuments {
total_docs,
visible_cost_upper_bound,
};
let cursor = WandCursor::new(
Operator::Or,
vec![posting()],
&documents,
scorer.clone(),
¶ms,
&metrics,
);
assert_eq!(cursor.cost(), 8);
}
let mut complete_docs = DocSet::default();
for doc_id in 0..total_docs {
complete_docs.append(doc_id as u64, 1);
}
assert_eq!(
WandDocuments::visible_cost_upper_bound(&complete_docs),
total_docs
);
}
#[test]
fn conservative_score_sum_covers_query_order_f32_rounding() {
let values = [
f32::from_bits(0x3e65_15bd),
f32::from_bits(0x34b4_3b11),
f32::from_bits(0x35e9_48ed),
f32::from_bits(0x3203_3773),
];
let exact_score = values
.into_iter()
.fold(0.0_f32, |score, value| score + value);
assert_eq!(exact_score.to_bits(), 0x3e65_164a);
assert!(conservative_score_sum(values.into_iter()) >= exact_score);
let mut reordered = [
f32::from_bits(0x3c87_b63e),
f32::from_bits(0x3d28_d10b),
f32::from_bits(0x3cc4_29c0),
];
let bound = conservative_score_sum(reordered.into_iter());
reordered.sort_by(|left, right| right.total_cmp(left));
let reordered_score = reordered
.into_iter()
.fold(0.0_f32, |score, value| score + value);
assert_eq!(reordered_score.to_bits(), 0x3da7_6086);
assert!(bound >= reordered_score);
}
#[test]
fn compound_global_bound_rejects_grouped_term_scoring() {
let mut docs = DocSet::default();
docs.append(0, 1);
let list = generate_posting_list(vec![0], 1.0, None, false);
let grouped_terms = Arc::<[GroupedTermScorer]>::from([GroupedTermScorer::new(1.0, &list)]);
let posting =
PostingIterator::with_query_weight(String::from("term"), 0, 0, 1.0, list, docs.len())
.with_grouped_terms(grouped_terms);
let wand = Wand::new(Operator::Or, std::iter::once(posting), &docs, UnitScorer);
assert_eq!(wand.compound_global_score_upper_bound(), None);
}
#[test]
fn compound_global_bound_rejects_late_partial_posting_state() {
let mut docs = DocSet::default();
docs.append(0, 1);
let posting = PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_posting_list(vec![0], 1.0, None, false),
docs.len(),
);
let mut wand = Wand::new(Operator::Or, std::iter::once(posting), &docs, UnitScorer);
assert!(wand.next().unwrap().is_some());
wand.push_back_leads(1);
assert_eq!(wand.compound_global_score_upper_bound(), None);
}
#[test]
fn test_maxscore_prefix_bound_covers_f32_summation_rounding() {
let remaining_bounds = [6.286_838_4e-7_f32, 0.015_441_144_f32];
let essential_score = 2.762_496_2_f32;
let threshold = f32::from_bits(0x4031_c9bc);
let raw_prefix = remaining_bounds.into_iter().sum::<f32>();
assert_eq!(essential_score + raw_prefix, threshold);
let actual_score = remaining_bounds
.into_iter()
.rev()
.fold(essential_score, |score, bound| score + bound);
assert!(actual_score > threshold);
let prefix_bound = remaining_bounds.into_iter().map(f64::from).sum::<f64>();
assert!(!score_sum_cannot_compete(
essential_score,
prefix_bound,
threshold,
score_sum_upper_bound_factor(3),
CompetitiveFloorMode::Exclusive,
));
}
#[test]
fn two_clause_inclusive_bound_rounds_outward_at_f32_boundary() {
let partial_score = 1.0_f32;
let remaining_upper_bound = f32::from_bits(0x3380_0001);
let floor = partial_score + remaining_upper_bound;
assert_eq!(floor.to_bits(), 0x3f80_0001);
assert!(!score_sum_cannot_compete(
partial_score,
f64::from(remaining_upper_bound),
floor,
score_sum_upper_bound_factor(2),
CompetitiveFloorMode::Inclusive,
));
}
struct UnitScorer;
impl Scorer for UnitScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, _doc_tokens: u32) -> f32 {
freq as f32
}
}
struct AdjacentScoreScorer;
impl Scorer for AdjacentScoreScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, _doc_tokens: u32) -> f32 {
match freq {
1 => f32::from_bits(1.0_f32.to_bits() - 1),
2 => 1.0,
_ => unreachable!("test only defines two score buckets"),
}
}
}
#[test]
fn inclusive_floor_keeps_tie_without_admitting_lower_ulp() {
let posting = PostingIterator::with_query_weight(
"term".to_owned(),
0,
0,
1.0,
generate_posting_list_with_freqs(vec![0, 1], vec![1, 2], 1.0, None, false),
2,
);
let mut docs = DocSet::default();
docs.append(0, 1);
docs.append(1, 1);
let mut wand = Wand::new(
Operator::Or,
std::iter::once(posting),
&docs,
AdjacentScoreScorer,
)
.with_floor_mode(CompetitiveFloorMode::Inclusive);
wand.update_threshold(1.0, 1.0);
let (doc, score) = wand.next().unwrap().unwrap();
assert_eq!(doc.doc_id(), 1);
assert_eq!(score, 1.0);
}
#[test]
fn final_score_uses_query_order_and_keeps_floor_ties() {
let contributions = [0.002_972_301_6_f32, 0.001_982_450_7_f32, 0.001_882_293_f32];
let bounds = [2.0_f32, 1.0, 3.0];
let postings = || {
contributions
.into_iter()
.zip(bounds)
.enumerate()
.map(|(position, (query_weight, max_score))| {
PostingIterator::with_query_weight(
format!("t{position}"),
position as u32,
position as u32,
query_weight,
generate_posting_list(vec![0], max_score, None, false),
1,
)
})
.collect::<Vec<_>>()
};
let mut docs = DocSet::default();
docs.append(0, 1);
let mut wand = Wand::new(Operator::Or, postings().into_iter(), &docs, UnitScorer);
let (_, heap_order_score) = wand.next().unwrap().unwrap();
let query_order_score = contributions
.into_iter()
.fold(0.0_f32, |score, contribution| score + contribution);
let wide_score = contributions
.into_iter()
.fold(0.0_f64, |score, contribution| {
score + f64::from(contribution)
}) as f32;
assert_eq!(query_order_score.to_bits(), 0x3be0_094c);
assert_eq!(wide_score.to_bits(), 0x3be0_094b);
assert_eq!(heap_order_score.to_bits(), 0x3be0_094a);
assert_eq!(wand.score_in_query_order(1), query_order_score);
let params = FtsSearchParams::default();
let metrics = NoOpMetricsCollector;
let mut cursor = WandCursor::new(
Operator::Or,
postings(),
&docs,
Arc::new(MemBM25Scorer::new(1, 1, std::collections::HashMap::new())),
¶ms,
&metrics,
);
cursor.set_min_competitive_score(query_order_score).unwrap();
assert_eq!(cursor.next().unwrap(), Some(0));
assert_eq!(cursor.current_score().unwrap(), query_order_score);
}
#[rstest]
#[case::initial_floor(false)]
#[case::explicit_zero_floor(true)]
fn wand_cursor_preserves_zero_score_membership(
#[case] set_zero_floor: bool,
#[values(Operator::Or, Operator::And)] operator: Operator,
#[values(1, 4, 5)] num_terms: u32,
) {
let postings = (0..num_terms)
.map(|term| {
let list = if operator == Operator::And {
generate_impact_posting_list_with_freqs_and_block_size(
vec![0],
vec![1],
vec![1],
MAX_POSTING_BLOCK_SIZE,
)
} else {
generate_posting_list(vec![0], 0.0, None, false)
};
PostingIterator::with_query_weight(
format!("common{term}"),
term,
term,
0.0,
list,
1,
)
})
.collect();
let mut docs = DocSet::default();
docs.append(0, 1);
let params = FtsSearchParams::default();
let metrics = NoOpMetricsCollector;
let mut cursor = WandCursor::new(
operator,
postings,
&docs,
Arc::new(MemBM25Scorer::new(1, 1, std::collections::HashMap::new())),
¶ms,
&metrics,
);
if set_zero_floor {
cursor.set_min_competitive_score(0.0).unwrap();
}
assert_eq!(cursor.next().unwrap(), Some(0));
assert_eq!(cursor.current_score().unwrap(), 0.0);
}
#[rstest]
#[case::all_essential(false)]
#[case::single_essential(true)]
fn maxscore_publishes_query_order_score_bits(#[case] single_essential: bool) {
let contributions = [0.002_972_301_6_f32, 0.001_982_450_7_f32, 0.001_882_293_f32];
let bounds = [0.002_99_f32, 0.001_99_f32, 0.003_f32];
let postings = contributions
.into_iter()
.zip(bounds)
.enumerate()
.map(|(position, (query_weight, max_score))| {
PostingIterator::with_query_weight(
format!("t{position}"),
position as u32,
position as u32,
query_weight,
generate_posting_list(vec![0], max_score, None, true),
1,
)
})
.collect::<Vec<_>>();
let mut docs = DocSet::default();
docs.append(0, 1);
let shared_floor = Arc::new(AtomicU32::new(0.0_f32.to_bits()));
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::Or,
postings.into_iter(),
&docs,
CountingScorer {
scored: scored.clone(),
},
)
.with_shared_threshold(shared_floor.clone());
let query_order_score = contributions
.into_iter()
.fold(0.0_f32, |score, contribution| score + contribution);
if single_essential {
wand.threshold = f32::from_bits(query_order_score.to_bits() - 1);
}
let hits = wand
.maxscore_search(
&FtsSearchParams::default().with_limit(Some(1)),
&NoOpMetricsCollector,
)
.unwrap();
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].document, 0);
assert_eq!(shared_floor.load(Ordering::Relaxed), 0x3be0_094c);
assert_eq!(scored.load(Ordering::Relaxed), contributions.len());
assert_eq!(wand.maxscore_single_essential_windows > 0, single_essential);
assert_eq!(wand.maxscore_general_windows > 0, !single_essential);
}
#[rstest]
fn bulk_and_and_classic_publish_identical_query_order_score_bits(
#[values(3, 4, 5, 8, 16)] num_clauses: usize,
) {
let contributions = [0.002_972_301_6_f32, 0.001_982_450_7_f32, 0.001_882_293_f32]
.into_iter()
.cycle()
.take(num_clauses)
.collect::<Vec<_>>();
let bounds = [0.002_99_f32, 0.001_99_f32, 0.003_f32];
let clause_docs = [vec![0, 1, 2], vec![0], vec![0, 1]];
let mut docs = DocSet::default();
for doc_id in 0..3 {
docs.append(doc_id, 1);
}
let run = |mode| {
let postings = contributions
.iter()
.copied()
.zip(bounds.into_iter().cycle())
.zip(clause_docs.iter().cycle())
.enumerate()
.map(|(position, ((query_weight, max_score), doc_ids))| {
PostingIterator::with_query_weight(
format!("t{position}"),
position as u32,
position as u32,
query_weight,
generate_posting_list(doc_ids.clone(), max_score, None, true),
docs.len(),
)
})
.collect::<Vec<_>>();
let shared_floor = Arc::new(AtomicU32::new(0.0_f32.to_bits()));
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer)
.with_bulk_and_mode(mode)
.with_shared_threshold(shared_floor.clone());
let hits = wand
.search(
&FtsSearchParams::default().with_limit(Some(1)),
&NoOpMetricsCollector,
)
.unwrap();
(
hits,
shared_floor.load(Ordering::Relaxed),
wand.bulk_and_searches > 0,
)
};
let (bulk, bulk_score, bulk_used) = run(BulkAndMode::On);
let (classic, classic_score, classic_used) = run(BulkAndMode::Off);
assert!(bulk_used);
assert!(!classic_used);
assert_eq!(bulk.len(), 1);
assert_eq!(classic.len(), 1);
assert_eq!(bulk[0].document, 0);
assert_eq!(classic[0].document, 0);
assert_eq!(
bulk_score,
contributions
.iter()
.fold(0.0_f32, |sum, value| sum + value)
.to_bits()
);
assert_eq!(classic_score, bulk_score);
}
struct PartialNormScorer;
impl Scorer for PartialNormScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, _doc_tokens: u32) -> f32 {
freq as f32
}
fn doc_norm(&self, doc_tokens: u32) -> Option<f32> {
(doc_tokens == 0).then_some(0.0)
}
}
#[test]
fn test_norm_cache_requires_all_norm_codes() {
let mut docs = DocSet::default();
docs.append(0, 1);
docs.set_quantized_scoring(true);
let wand = Wand::new(Operator::Or, std::iter::empty(), &docs, PartialNormScorer);
assert!(wand.norm_k_cache().is_none());
}
#[test]
fn test_top_k_collector_reuses_frequency_slots() -> Result<()> {
const LIMIT: usize = 8;
const NUM_DOCS: usize = 10_000;
let mut collector = TopKCollector::new(LIMIT, LIMIT);
for doc in 0..NUM_DOCS {
let num_terms = doc % 4 + 1;
let inserted = collector.insert(
ScoredDoc::new(doc as u64, doc as f32),
num_terms as u32,
doc as u64,
(0..num_terms).map(|term| (term as u32, doc as u32)),
)?;
assert!(inserted);
assert!(collector.num_frequency_slots() <= LIMIT);
}
assert_eq!(collector.num_frequency_slots(), LIMIT);
let mut candidates = collector.into_candidates(|key| key)?;
candidates.sort_unstable_by_key(|candidate| candidate.posting_doc_id);
assert_eq!(candidates.len(), LIMIT);
for (candidate, expected_doc) in candidates.iter().zip(NUM_DOCS - LIMIT..NUM_DOCS) {
assert_eq!(candidate.posting_doc_id, expected_doc as u64);
assert_eq!(candidate.document, expected_doc as u64);
let expected_freqs = (0..expected_doc % 4 + 1)
.map(|term| (term as u32, expected_doc as u32))
.collect::<Vec<_>>();
assert_eq!(candidate.freqs, expected_freqs);
}
Ok(())
}
#[rstest]
#[case::auto("auto", Some(BulkAndMode::Auto))]
#[case::auto_case_and_whitespace(" AUTO ", Some(BulkAndMode::Auto))]
#[case::on("on", Some(BulkAndMode::On))]
#[case::on_legacy("1", Some(BulkAndMode::On))]
#[case::off("off", Some(BulkAndMode::Off))]
#[case::off_legacy("0", Some(BulkAndMode::Off))]
#[case::invalid("true", None)]
#[case::empty("", None)]
fn test_bulk_and_mode_parse(#[case] value: &str, #[case] expected: Option<BulkAndMode>) {
assert_eq!(BulkAndMode::parse(value), expected);
}
#[rstest]
#[case::auto_one(BulkAndMode::Auto, 1, false)]
#[case::auto_two(BulkAndMode::Auto, 2, true)]
#[case::auto_three(BulkAndMode::Auto, 3, true)]
#[case::auto_four(BulkAndMode::Auto, 4, false)]
#[case::on_one(BulkAndMode::On, 1, true)]
#[case::on_five(BulkAndMode::On, 5, true)]
#[case::off_two(BulkAndMode::Off, 2, false)]
#[case::off_five(BulkAndMode::Off, 5, false)]
fn test_bulk_and_mode_enabled_for(
#[case] mode: BulkAndMode,
#[case] num_clauses: usize,
#[case] expected: bool,
) {
assert_eq!(mode.enabled_for(num_clauses), expected);
}
#[test]
fn find_next_geq_matches_partition_point_for_full_u32_domain() {
let cases = [
vec![],
vec![7],
vec![0, 1, 2, 3, 4, 5, 6, 7],
vec![0, 1, 2, 3, 4, 5, 6, 7, 8],
(0..32).collect(),
vec![
0,
1,
2,
3,
4,
5,
6,
7,
i32::MAX as u32,
i32::MAX as u32 + 1,
i32::MAX as u32 + 2,
i32::MAX as u32 + 3,
i32::MAX as u32 + 4,
i32::MAX as u32 + 5,
i32::MAX as u32 + 6,
u32::MAX - 2,
u32::MAX - 1,
u32::MAX,
],
];
let targets = [
0,
1,
6,
7,
8,
i32::MAX as u32 - 1,
i32::MAX as u32,
i32::MAX as u32 + 1,
u32::MAX - 1,
u32::MAX,
];
for docs in &cases {
for pos in 0..=docs.len() {
for &target in &targets {
let expected = pos + docs[pos..].partition_point(|&doc_id| doc_id < target);
let actual = unsafe { find_next_geq(docs.as_ptr(), pos, docs.len(), target) };
assert_eq!(
actual, expected,
"docs={docs:?}, pos={pos}, target={target}"
);
assert_eq!(
find_next_geq_in_block(docs, pos, target),
expected,
"block search: docs={docs:?}, pos={pos}, target={target}"
);
}
}
}
}
struct PanicQueryWeightScorer;
impl Scorer for PanicQueryWeightScorer {
fn query_weight(&self, _token: &str) -> f32 {
panic!("query_weight should be precomputed before WAND construction");
}
fn doc_weight(&self, freq: u32, _doc_tokens: u32) -> f32 {
freq as f32
}
}
struct InverseDocLengthScorer;
impl Scorer for InverseDocLengthScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, doc_tokens: u32) -> f32 {
freq as f32 / doc_tokens as f32
}
}
struct CountingScorer {
scored: Arc<AtomicUsize>,
}
impl Scorer for CountingScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, doc_tokens: u32) -> f32 {
self.scored.fetch_add(1, Ordering::Relaxed);
freq as f32 / doc_tokens as f32
}
}
struct CountingBm25ShapeScorer {
scored: Arc<AtomicUsize>,
}
impl Scorer for CountingBm25ShapeScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, _doc_tokens: u32) -> f32 {
self.scored.fetch_add(1, Ordering::Relaxed);
bm25_doc_weight_with_norm(freq, 1.2)
}
fn doc_weight_upper_bound(&self) -> Option<f32> {
Some(BM25_DOC_WEIGHT_UPPER_BOUND)
}
fn doc_norm(&self, _doc_tokens: u32) -> Option<f32> {
Some(1.2)
}
}
struct CandidateOnlyBm25ShapeScorer {
norm_calls: Arc<AtomicUsize>,
unsupported_norm: Option<f32>,
}
impl Scorer for CandidateOnlyBm25ShapeScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, _doc_tokens: u32) -> f32 {
bm25_doc_weight_with_norm(freq, 0.4)
}
fn doc_weight_upper_bound(&self) -> Option<f32> {
Some(BM25_DOC_WEIGHT_UPPER_BOUND)
}
fn doc_norm(&self, doc_tokens: u32) -> Option<f32> {
self.norm_calls.fetch_add(1, Ordering::Relaxed);
if doc_tokens == 1 {
Some(0.4)
} else {
self.unsupported_norm
}
}
}
struct NonFiniteBoundBm25ShapeScorer;
impl Scorer for NonFiniteBoundBm25ShapeScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, _freq: u32, _doc_tokens: u32) -> f32 {
f32::INFINITY
}
fn doc_norm(&self, _doc_tokens: u32) -> Option<f32> {
Some(1.0)
}
}
struct VariedBm25ShapeScorer;
impl Scorer for VariedBm25ShapeScorer {
fn query_weight(&self, _token: &str) -> f32 {
1.0
}
fn doc_weight(&self, freq: u32, doc_tokens: u32) -> f32 {
bm25_doc_weight_with_norm(freq, self.doc_norm(doc_tokens).unwrap())
}
fn doc_weight_upper_bound(&self) -> Option<f32> {
Some(BM25_DOC_WEIGHT_UPPER_BOUND)
}
fn doc_norm(&self, doc_tokens: u32) -> Option<f32> {
Some(0.3 + doc_tokens as f32 * 0.1)
}
}
#[cfg_attr(coverage, coverage(off))]
fn unit_length_docs(num_docs: usize) -> DocSet {
let mut docs = DocSet::default();
for doc_id in 0..num_docs {
docs.append(doc_id as u64, 1);
}
docs
}
#[cfg_attr(coverage, coverage(off))]
fn unit_length_impact_posting(
term: u32,
doc_ids: Vec<u32>,
frequencies: Vec<u32>,
num_docs: usize,
) -> PostingIterator {
let doc_lengths = vec![1; doc_ids.len()];
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids,
frequencies,
doc_lengths,
MAX_POSTING_BLOCK_SIZE,
),
num_docs,
)
}
fn generate_posting_list(
doc_ids: Vec<u32>,
max_score: f32,
block_max_scores: Option<Vec<f32>>,
is_compressed: bool,
) -> PostingList {
let freqs = vec![1; doc_ids.len()];
generate_posting_list_with_freqs(doc_ids, freqs, max_score, block_max_scores, is_compressed)
}
fn generate_posting_list_with_freqs(
doc_ids: Vec<u32>,
freqs: Vec<u32>,
max_score: f32,
block_max_scores: Option<Vec<f32>>,
is_compressed: bool,
) -> PostingList {
assert_eq!(doc_ids.len(), freqs.len());
let block_max_scores = block_max_scores.unwrap_or_else(|| vec![max_score; doc_ids.len()]);
if is_compressed {
let blocks = compress_posting_list(
doc_ids.len(),
doc_ids.iter(),
freqs.iter(),
block_max_scores.into_iter(),
)
.unwrap();
PostingList::Compressed(CompressedPostingList::new(
blocks,
max_score,
doc_ids.len() as u32,
crate::scalar::inverted::PostingTailCodec::VarintDelta,
crate::scalar::inverted::LEGACY_BLOCK_SIZE,
None,
None,
))
} else {
PostingList::Plain(PlainPostingList::new(
ScalarBuffer::from_iter(doc_ids.iter().map(|id| *id as u64)),
ScalarBuffer::from_iter(freqs.iter().map(|freq| *freq as f32)),
Some(max_score),
None,
))
}
}
fn generate_impact_posting_list_with_freqs(
doc_ids: Vec<u32>,
freqs: Vec<u32>,
doc_lengths: Vec<u32>,
) -> PostingList {
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids,
freqs,
doc_lengths,
crate::scalar::inverted::LEGACY_BLOCK_SIZE,
)
}
fn generate_impact_posting_list_with_freqs_and_block_size(
doc_ids: Vec<u32>,
freqs: Vec<u32>,
doc_lengths: Vec<u32>,
block_size: usize,
) -> PostingList {
assert_eq!(doc_ids.len(), freqs.len());
assert_eq!(doc_ids.len(), doc_lengths.len());
let block_max_scores = vec![0.0; doc_ids.len().div_ceil(block_size)];
let blocks = compress_posting_list_with_tail_codec_and_block_size(
doc_ids.len(),
doc_ids.iter(),
freqs.iter(),
block_max_scores.into_iter(),
crate::scalar::inverted::PostingTailCodec::VarintDelta,
block_size,
)
.unwrap();
let impact_blocks = doc_ids
.chunks(block_size)
.zip(freqs.chunks(block_size))
.zip(doc_lengths.chunks(block_size))
.map(|((doc_ids, freqs), doc_lengths)| {
doc_ids
.iter()
.copied()
.zip(freqs.iter().copied())
.zip(doc_lengths.iter().copied())
.map(|((doc_id, freq), doc_length)| (doc_id, freq, doc_length))
.collect::<Vec<_>>()
})
.collect::<Vec<_>>();
let impacts = build_impact_skip_data(impact_blocks.as_slice()).unwrap();
PostingList::Compressed(CompressedPostingList::new(
blocks,
0.0,
doc_ids.len() as u32,
crate::scalar::inverted::PostingTailCodec::VarintDelta,
block_size,
None,
Some(impacts),
))
}
fn generate_contiguous_impact_posting_list_with_block_size(
total: usize,
block_size: usize,
) -> PostingList {
generate_impact_posting_list_with_freqs_and_block_size(
(0..total as u32).collect(),
vec![1; total],
vec![1; total],
block_size,
)
}
fn generate_posting_list_with_positions(
doc_ids: Vec<u32>,
positions_by_doc: Vec<Vec<u32>>,
max_score: f32,
is_compressed: bool,
) -> PostingList {
let freqs = positions_by_doc
.iter()
.map(|positions| positions.len() as u32)
.collect::<Vec<_>>();
if is_compressed {
let mut builder = PostingListBuilder::new(true);
for (doc_id, positions) in doc_ids.iter().copied().zip(positions_by_doc) {
builder.add(doc_id, PositionRecorder::Position(positions.into()));
}
let batch = builder
.to_batch(vec![max_score; doc_ids.len().div_ceil(BLOCK_SIZE)])
.unwrap();
PostingList::from_batch(&batch, Some(max_score), Some(doc_ids.len() as u32)).unwrap()
} else {
let mut position_builder =
arrow::array::ListBuilder::new(arrow::array::Int32Builder::new());
for positions in positions_by_doc {
for position in positions {
position_builder.values().append_value(position as i32);
}
position_builder.append(true);
}
PostingList::Plain(PlainPostingList::new(
ScalarBuffer::from_iter(doc_ids.iter().map(|id| *id as u64)),
ScalarBuffer::from_iter(freqs.iter().map(|freq| *freq as f32)),
Some(max_score),
Some(position_builder.finish()),
))
}
}
#[rstest]
#[case::packed_delta(PositionStreamCodec::PackedDelta)]
#[case::varint_doc_delta(PositionStreamCodec::VarintDocDelta)]
fn test_shared_position_cursors_use_independent_scratch(
#[case] codec: PositionStreamCodec,
) -> Result<()> {
let mut posting_list =
generate_posting_list_with_positions(vec![0], vec![vec![1_u32, 3, 10]], 1.0, true);
let PostingList::Compressed(ref mut list) = posting_list else {
unreachable!("the helper was asked for a compressed posting list");
};
let mut encoded = Vec::new();
encode_position_stream_block_into(&[1, 3, 10], &[3], codec, &mut encoded)?;
list.positions = Some(CompressedPositionStorage::SharedStream(
SharedPositionStream::new(codec, vec![0], bytes::Bytes::from(encoded)),
));
let posting = PostingIterator::new(String::from("term"), 0, 0, posting_list, 1);
let first = posting.position_cursor()?;
let second = posting.position_cursor()?;
assert_eq!(second.positions.as_slice(), &[1, 3, 10]);
assert_eq!(first.positions.as_slice(), &[1, 3, 10]);
assert!(posting.position_scratch.borrow().is_none());
drop(second);
assert!(posting.position_scratch.borrow().is_some());
drop(first);
assert!(posting.position_scratch.borrow().is_some());
Ok(())
}
#[test]
fn test_phrase_search_propagates_corrupt_packed_positions() {
let mut docs = DocSet::default();
docs.append(0, BLOCK_SIZE as u32 + 1);
let mut corrupt_list = generate_posting_list_with_positions(
vec![0],
vec![(0..BLOCK_SIZE as u32).collect()],
1.0,
true,
);
let PostingList::Compressed(ref mut list) = corrupt_list else {
unreachable!("the helper was asked for a compressed posting list");
};
list.positions = Some(CompressedPositionStorage::SharedStream(
SharedPositionStream::new(
PositionStreamCodec::PackedDelta,
vec![0],
bytes::Bytes::from_static(&[1]),
),
));
let postings = vec![
PostingIterator::new(String::from("corrupt"), 0, 0, corrupt_list, docs.len()),
PostingIterator::new(
String::from("valid"),
1,
1,
generate_posting_list_with_positions(
vec![0],
vec![vec![BLOCK_SIZE as u32]],
1.0,
true,
),
docs.len(),
),
];
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer);
let mut params = FtsSearchParams::default().with_limit(Some(10));
params.phrase_slop = Some(0);
let error = wand
.search(¶ms, &NoOpMetricsCollector)
.expect_err("corrupt packed positions should fail the phrase search");
let message = error.to_string();
assert!(
message.contains("packed position group payload"),
"{message}"
);
assert!(message.contains("corrupt"), "{message}");
}
fn sorted_candidate_row_ids(candidates: Vec<DocCandidate<u64>>) -> Vec<u64> {
let mut row_ids = candidates
.into_iter()
.map(|candidate| candidate.document)
.collect::<Vec<_>>();
row_ids.sort_unstable();
row_ids
}
#[rstest]
#[tokio::test]
async fn test_wand(#[values(false, true)] is_compressed: bool) {
let mut docs = DocSet::default();
for i in 0..2 * BLOCK_SIZE {
docs.append(i as u64, 1);
}
let postings = vec![
PostingIterator::new(
String::from("test"),
0,
0,
generate_posting_list(
Vec::from_iter(0..=BLOCK_SIZE as u32 + 1),
1.0,
None,
is_compressed,
),
docs.len(),
),
PostingIterator::new(
String::from("full"),
1,
1,
generate_posting_list(vec![BLOCK_SIZE as u32 + 2], 1.0, None, is_compressed),
docs.len(),
),
];
let bm25 = IndexBM25Scorer::new(std::iter::empty());
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, bm25);
let result = wand
.search(&FtsSearchParams::default(), &NoOpMetricsCollector)
.unwrap();
assert_eq!(result.len(), 0); }
#[test]
fn cross_partition_threshold_sharing_prunes() {
use crate::metrics::MetricsCollector;
use std::sync::atomic::AtomicUsize;
#[derive(Default)]
struct CountComparisons(AtomicUsize);
impl MetricsCollector for CountComparisons {
fn record_parts_loaded(&self, _: usize) {}
fn record_index_loads(&self, _: usize) {}
fn record_comparisons(&self, n: usize) {
self.0.fetch_add(n, Ordering::Relaxed);
}
}
let params = FtsSearchParams::default().with_limit(Some(10));
let part_docs = 4 * BLOCK_SIZE as u32;
let parts: Vec<(f32, std::ops::Range<u32>)> = std::iter::once((10.0, 0..part_docs))
.chain((1..8).map(|i| (1.0, i * part_docs..(i + 1) * part_docs)))
.collect();
let new_floor = || Arc::new(AtomicU32::new(f32::NEG_INFINITY.to_bits()));
let total_comparisons = |shared_floor: Option<&Arc<AtomicU32>>| -> usize {
let metrics = CountComparisons::default();
for (qw, rows) in &parts {
let mut docs = DocSet::default();
for d in rows.clone() {
docs.append(d as u64, 1);
}
let postings = vec![PostingIterator::with_query_weight(
String::from("t"),
0,
0,
*qw,
generate_posting_list(rows.clone().collect(), *qw, None, false),
docs.len(),
)];
let floor = shared_floor.cloned().unwrap_or_else(new_floor);
Wand::new(Operator::Or, postings.into_iter(), &docs, UnitScorer)
.with_shared_threshold(floor)
.search(¶ms, &metrics)
.unwrap();
}
metrics.0.load(Ordering::Relaxed)
};
let one_floor = new_floor();
let with_shared_floor = total_comparisons(Some(&one_floor));
let with_private_floors = total_comparisons(None);
assert!(
with_shared_floor < with_private_floors,
"shared floor should prune comparisons: \
shared={with_shared_floor} private={with_private_floors}"
);
}
#[test]
fn test_posting_iterator_next_compressed_partition_point() {
let mut docs = DocSet::default();
let num_docs = (BLOCK_SIZE * 2 + 5) as u32;
for i in 0..num_docs {
docs.append(i as u64, 1);
}
let doc_ids = (0..num_docs).collect::<Vec<_>>();
let posting = generate_posting_list(doc_ids, 1.0, None, true);
let mut iter = PostingIterator::new(String::from("term"), 0, 0, posting, docs.len());
iter.next(10);
assert_eq!(iter.doc().unwrap().doc_id(), 10);
let target = BLOCK_SIZE as u64 + 3;
iter.next(target);
assert_eq!(iter.doc().unwrap().doc_id(), target);
iter.next(num_docs as u64 + 10);
assert!(iter.doc().is_none());
}
#[test]
fn posting_iterator_fork_restarts_shared_compressed_payload() {
let num_docs = (BLOCK_SIZE * 2 + 5) as u32;
let posting = generate_posting_list((0..num_docs).collect(), 1.0, None, true);
let mut original =
PostingIterator::new(String::from("term"), 7, 3, posting, num_docs as usize);
original.next(BLOCK_SIZE as u64 + 3);
assert_eq!(
original.doc().map(|doc| doc.doc_id()),
Some(BLOCK_SIZE as u64 + 3)
);
let mut replay = original.fork_from_start();
assert_eq!(replay.doc().map(|doc| doc.doc_id()), Some(0));
replay.next(5);
assert_eq!(replay.doc().map(|doc| doc.doc_id()), Some(5));
assert_eq!(
original.doc().map(|doc| doc.doc_id()),
Some(BLOCK_SIZE as u64 + 3),
"fork advancement must not mutate the membership cursor"
);
}
#[test]
fn test_wand_skip_to_next_block() {
let mut docs = DocSet::default();
for i in 0..201 {
docs.append(i as u64, 1);
}
let large_posting_docs1: Vec<u32> = (0..=200).collect();
let postings = vec![
PostingIterator::new(
String::from("full"),
0,
0,
generate_posting_list(large_posting_docs1, 1.0, Some(vec![0.5, 0.5]), true),
docs.len(),
),
PostingIterator::new(
String::from("text"),
1,
1,
generate_posting_list(vec![0], 1.0, Some(vec![0.5]), true),
docs.len(),
),
];
let bm25 = IndexBM25Scorer::new(std::iter::empty());
let mut wand = Wand::new(Operator::Or, postings.into_iter(), &docs, bm25);
wand.threshold = 1.5;
let result = wand.search(&FtsSearchParams::default(), &NoOpMetricsCollector);
assert!(result.is_ok());
}
#[test]
fn test_or_single_term_block_skip_matches_and() {
let total = 3 * BLOCK_SIZE as u32;
let hot = BLOCK_SIZE as u32..BLOCK_SIZE as u32 + 12;
let mut docs = DocSet::default();
for row_id in 0..total {
let doc_tokens = if hot.contains(&row_id) {
row_id - hot.start + 1
} else {
1000
};
docs.append(row_id as u64, doc_tokens);
}
let params = FtsSearchParams::new().with_limit(Some(10));
let run = |operator| {
let scored = Arc::new(AtomicUsize::new(0));
let posting = PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_posting_list(
(0..total).collect(),
1.0,
Some(vec![0.001, 1.0, 0.001]),
true,
),
docs.len(),
);
let mut wand = Wand::new(
operator,
std::iter::once(posting),
&docs,
CountingScorer {
scored: scored.clone(),
},
);
let hits = wand.search(¶ms, &NoOpMetricsCollector).unwrap();
let mut row_ids = hits.iter().map(|hit| hit.document).collect::<Vec<_>>();
row_ids.sort_unstable();
(row_ids, scored.load(Ordering::Relaxed))
};
let (or_hits, or_scored) = run(Operator::Or);
let (and_hits, _) = run(Operator::And);
let expected = (hot.start..hot.start + 10)
.map(u64::from)
.collect::<Vec<_>>();
assert_eq!(or_hits, expected, "OR must return the top-k");
assert_eq!(or_hits, and_hits, "OR and AND must agree for a single term");
assert!(
or_scored <= 2 * BLOCK_SIZE,
"expected pruning to skip a block, but scored {or_scored} of {total}",
);
}
#[rstest]
fn test_or_search_records_comparisons(#[values(false, true)] is_compressed: bool) {
let mut docs = DocSet::default();
for row_id in 0..6 {
docs.append(row_id, 1);
}
let postings = vec![
PostingIterator::with_query_weight(
String::from("alpha"),
0,
0,
1.0,
generate_posting_list(vec![0, 1, 4], 1.0, None, is_compressed),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("beta"),
1,
1,
1.0,
generate_posting_list(vec![1, 2, 5], 1.0, None, is_compressed),
docs.len(),
),
];
let mut wand = Wand::new(Operator::Or, postings.into_iter(), &docs, UnitScorer);
let metrics = LocalMetricsCollector::default();
let candidates = wand.search(&FtsSearchParams::default(), &metrics).unwrap();
assert_eq!(sorted_candidate_row_ids(candidates), vec![0, 1, 2, 4, 5]);
assert!(metrics.comparisons.load(Ordering::Relaxed) > 0);
}
#[test]
fn test_wand_new_uses_precomputed_query_weight() {
let mut docs = DocSet::default();
docs.append(1, 1);
let postings = vec![PostingIterator::with_query_weight(
String::from("term"),
0,
0,
2.0,
generate_posting_list(vec![0], 1.0, None, false),
docs.len(),
)];
let wand = Wand::new(
Operator::Or,
postings.into_iter(),
&docs,
PanicQueryWeightScorer,
);
assert_eq!(wand.head.len(), 1);
}
#[test]
fn test_and_search_terminates_for_disjoint_postings() {
let mut docs = DocSet::default();
for i in 0..6 {
docs.append(i, 1);
}
let postings = vec![
PostingIterator::with_query_weight(
String::from("a"),
0,
0,
1.0,
generate_posting_list(vec![0, 2, 4], 1.0, None, false),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("b"),
1,
1,
1.0,
generate_posting_list(vec![1, 3, 5], 1.0, None, false),
docs.len(),
),
];
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer);
assert!(wand.next().unwrap().is_none());
}
#[test]
fn test_up_to_refreshes_on_first_candidate() {
let mut docs = DocSet::default();
for i in 0..=(BLOCK_SIZE as u64 + 1) {
docs.append(i, 1);
}
let postings = vec![PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_posting_list(
(0..=(BLOCK_SIZE as u32 + 1)).collect(),
1.0,
Some(vec![1.0, 1.0]),
true,
),
docs.len(),
)];
let mut wand = Wand::new(Operator::Or, postings.into_iter(), &docs, UnitScorer);
assert!(wand.up_to.is_none());
let _ = wand.next().unwrap();
assert!(wand.up_to.is_some());
}
#[test]
fn test_or_push_back_lead_uses_current_block_max_for_tail_bound() {
let total = 2 * BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for doc_id in 0..total {
docs.append(doc_id as u64, 1);
}
let postings = vec![PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_posting_list((0..total).collect(), 10.0, Some(vec![1.0, 10.0]), true),
docs.len(),
)];
let mut wand = Wand::new(Operator::Or, postings.into_iter(), &docs, UnitScorer);
wand.threshold = 1.5;
wand.update_max_scores(0);
wand.move_head_doc_to_lead(0);
assert_eq!(wand.up_to, Some((BLOCK_SIZE - 1) as u64));
wand.push_back_leads(1);
assert_eq!(wand.tail.len(), 1);
assert!(
(wand.tail_max_score - 1.0).abs() < 1e-6,
"tail should use the current block max, got {}",
wand.tail_max_score
);
assert!(wand.head_doc().is_none());
}
#[test]
fn test_or_push_back_lead_falls_back_after_block_window_expires() {
let total = 2 * BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for doc_id in 0..total {
docs.append(doc_id as u64, 1);
}
let freqs = (0..total)
.map(|doc_id| if doc_id >= BLOCK_SIZE as u32 { 10 } else { 1 })
.collect::<Vec<_>>();
let mut posting = PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_posting_list_with_freqs(
(0..total).collect(),
freqs,
10.0,
Some(vec![1.0, 10.0]),
true,
),
docs.len(),
);
posting.next((BLOCK_SIZE - 1) as u64);
let mut wand = Wand::new(Operator::Or, std::iter::once(posting), &docs, UnitScorer);
wand.threshold = 1.5;
let block_end = (BLOCK_SIZE - 1) as u64;
wand.update_max_scores(block_end);
wand.move_head_doc_to_lead(block_end);
assert_eq!(wand.up_to, Some(block_end));
wand.push_back_leads(BLOCK_SIZE as u64);
assert!(wand.tail.is_empty());
assert_eq!(wand.head_doc(), Some(BLOCK_SIZE as u64));
let candidate = wand.next().unwrap().unwrap();
assert_eq!(candidate.0.doc_id(), BLOCK_SIZE as u64);
}
#[test]
fn test_non_positive_threshold_advances_without_impact_bound_scoring() {
let mut docs = DocSet::default();
for doc_id in 0..3 {
docs.append(doc_id, 1);
}
let make_posting = || {
PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_impact_posting_list_with_freqs(
vec![0, 1, 2],
vec![1, 1, 1],
vec![1, 1, 1],
),
docs.len(),
)
};
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::Or,
std::iter::once(make_posting()),
&docs,
CountingScorer {
scored: scored.clone(),
},
);
wand.move_head_before_target_to_tail(1);
assert_eq!(wand.head_doc(), Some(1));
assert!(wand.tail.is_empty());
assert_eq!(scored.load(Ordering::Relaxed), 0);
let mut wand = Wand::new(
Operator::Or,
std::iter::once(make_posting()),
&docs,
CountingScorer {
scored: scored.clone(),
},
);
wand.move_head_doc_to_lead(0);
wand.push_back_leads(1);
assert_eq!(wand.head_doc(), Some(1));
assert!(wand.tail.is_empty());
assert_eq!(scored.load(Ordering::Relaxed), 0);
}
#[test]
fn test_or_plain_tail_does_not_advance_headless_window() {
let mut docs = DocSet::default();
for doc_id in 0..4 {
docs.append(doc_id, 1);
}
let postings = vec![PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_posting_list(vec![0, 1, 2, 3], 1.0, None, false),
docs.len(),
)];
let mut wand = Wand::new(Operator::Or, postings.into_iter(), &docs, UnitScorer);
wand.threshold = 2.0;
wand.update_max_scores(0);
wand.move_head_doc_to_lead(0);
wand.push_back_leads(1);
assert_eq!(wand.tail.len(), 1);
assert!(wand.head_doc().is_none());
assert!(!wand.advance_tail_to_next_or_window());
}
#[test]
fn test_or_headless_tail_window_scans_past_final_top_tail() {
let total = 3 * BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for doc_id in 0..total {
docs.append(doc_id as u64, 1);
}
let future_docs = (0..BLOCK_SIZE as u32)
.chain(2 * BLOCK_SIZE as u32..3 * BLOCK_SIZE as u32)
.collect::<Vec<_>>();
let mut future_freqs = vec![1; future_docs.len()];
future_freqs[0] = 4;
future_freqs[BLOCK_SIZE] = 20;
let postings = vec![
PostingIterator::with_query_weight(
String::from("future"),
0,
0,
1.0,
generate_posting_list_with_freqs(
future_docs,
future_freqs,
20.0,
Some(vec![4.0, 20.0]),
true,
),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("final_tail"),
1,
1,
1.0,
generate_posting_list_with_freqs(vec![0], vec![6], 6.0, Some(vec![6.0]), true),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("booster"),
2,
2,
1.0,
generate_posting_list_with_freqs(vec![0], vec![7], 7.0, Some(vec![7.0]), true),
docs.len(),
),
];
let mut wand = Wand::new(Operator::Or, postings.into_iter(), &docs, UnitScorer);
let result = wand
.search(
&FtsSearchParams::new().with_limit(Some(1)),
&NoOpMetricsCollector,
)
.unwrap();
assert_eq!(
sorted_candidate_row_ids(result),
vec![(2 * BLOCK_SIZE) as u64]
);
}
#[test]
fn test_and_search_prunes_with_threshold_and_keeps_candidate() {
let mut docs = DocSet::default();
for i in 0..(2 * BLOCK_SIZE as u64) {
let doc_tokens = if i < BLOCK_SIZE as u64 { 100 } else { 1 };
docs.append(i, doc_tokens);
}
let all_docs = (0..2 * BLOCK_SIZE as u32).collect::<Vec<_>>();
let postings = vec![
PostingIterator::with_query_weight(
String::from("a"),
0,
0,
1.0,
generate_posting_list(all_docs.clone(), 1.0, Some(vec![0.02, 1.0]), true),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("b"),
1,
1,
1.0,
generate_posting_list(all_docs, 1.0, Some(vec![0.02, 1.0]), true),
docs.len(),
),
];
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
InverseDocLengthScorer,
);
wand.threshold = 0.5;
let candidate = wand.next().unwrap().unwrap();
assert_eq!(candidate.0.doc_id(), BLOCK_SIZE as u64);
}
#[test]
fn test_and_advance_falls_back_to_narrow_when_range_max_loosens_bound() {
let total = 4 * BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for i in 0..total {
docs.append(i as u64, 1);
}
let lead_docs = (0..total).step_by(2).collect::<Vec<_>>();
let follower_docs = (0..total).collect::<Vec<_>>();
let postings = vec![
PostingIterator::with_query_weight(
String::from("lead"),
0,
0,
1.0,
generate_posting_list(lead_docs, 1.0, Some(vec![1.0, 1.0]), true),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("follower"),
1,
1,
1.0,
generate_posting_list(follower_docs, 10.0, Some(vec![0.1, 10.0, 0.1, 0.1]), true),
docs.len(),
),
];
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer);
wand.threshold = 5.0;
let target = wand.and_advance_target(0);
assert_eq!(target, BLOCK_SIZE as u64);
assert_eq!(wand.up_to, Some((2 * BLOCK_SIZE - 1) as u64));
assert!(
(wand.and_max_score - 11.0).abs() < 1e-6,
"expected the second narrow window to include the high follower block, got {}",
wand.and_max_score
);
assert_eq!(wand.and_window_stats.windows_wide, 0);
assert_eq!(wand.and_window_stats.windows_narrow, 2);
assert_eq!(wand.and_window_stats.windows_skipped, 1);
}
#[test]
fn test_and_advance_uses_narrow_window_for_candidate_ranges() {
let total = 4 * BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for i in 0..total {
docs.append(i as u64, 1);
}
let lead_docs = (0..total).step_by(2).collect::<Vec<_>>();
let follower_docs = (0..total).collect::<Vec<_>>();
let postings = vec![
PostingIterator::with_query_weight(
String::from("lead"),
0,
0,
1.0,
generate_posting_list(lead_docs, 1.0, Some(vec![1.0, 1.0]), true),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("follower"),
1,
1,
1.0,
generate_posting_list(follower_docs, 1.0, Some(vec![1.0, 1.0, 1.0, 1.0]), true),
docs.len(),
),
];
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer);
wand.threshold = 1.5;
let target = wand.and_advance_target(0);
assert_eq!(target, 0);
assert_eq!(wand.up_to, Some((BLOCK_SIZE - 1) as u64));
assert!((wand.and_max_score - 2.0).abs() < 1e-6);
assert_eq!(wand.and_window_stats.windows_wide, 0);
assert_eq!(wand.and_window_stats.windows_narrow, 1);
assert_eq!(wand.and_window_stats.range_blocks_scanned, 0);
}
#[test]
fn test_and_wide_window_only_skips_and_does_not_return_candidates() {
let total = 4 * BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for i in 0..total {
docs.append(i as u64, 1);
}
let lead_docs = (0..total).step_by(2).collect::<Vec<_>>();
let follower_docs = (0..total).collect::<Vec<_>>();
let postings = vec![
PostingIterator::with_query_weight(
String::from("lead"),
0,
0,
1.0,
generate_posting_list(lead_docs, 3.0, Some(vec![1.0, 3.0]), true),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("follower"),
1,
1,
1.0,
generate_posting_list(follower_docs, 3.0, Some(vec![0.1, 0.1, 3.0, 3.0]), true),
docs.len(),
),
];
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer);
wand.threshold = 2.0;
let candidate = wand.next().unwrap().unwrap();
assert_eq!(candidate.0.doc_id(), (2 * BLOCK_SIZE) as u64);
assert_eq!(wand.up_to, Some((3 * BLOCK_SIZE - 1) as u64));
assert_eq!(wand.and_window_stats.windows_wide, 1);
assert_eq!(wand.and_window_stats.windows_skipped, 1);
assert_eq!(wand.and_window_stats.windows_narrow, 1);
assert_eq!(wand.and_window_stats.candidates_returned, 1);
}
#[test]
fn test_and_range_max_preserves_exact_top_k() {
let total = 4 * BLOCK_SIZE as u32;
let hot = BLOCK_SIZE as u32..BLOCK_SIZE as u32 + 16;
let mut docs = DocSet::default();
for doc_id in 0..total {
let doc_tokens = if hot.contains(&doc_id) { 1 } else { 1000 };
docs.append(doc_id as u64, doc_tokens);
}
let params = FtsSearchParams::new().with_limit(Some(8));
let run = |is_compressed: bool| {
let lead_docs = (0..total).step_by(2).collect::<Vec<_>>();
let follower_docs = (0..total).collect::<Vec<_>>();
let lead_scores = is_compressed.then_some(vec![1.0, 0.001]);
let follower_scores = is_compressed.then_some(vec![0.001, 1.0, 0.001, 0.001]);
let postings = vec![
PostingIterator::with_query_weight(
String::from("lead"),
0,
0,
1.0,
generate_posting_list(lead_docs, 1.0, lead_scores, is_compressed),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("follower"),
1,
1,
1.0,
generate_posting_list(follower_docs, 1.0, follower_scores, is_compressed),
docs.len(),
),
];
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
InverseDocLengthScorer,
);
sorted_candidate_row_ids(wand.search(¶ms, &NoOpMetricsCollector).unwrap())
};
let compressed = run(true);
let plain = run(false);
let expected = hot.step_by(2).map(u64::from).collect::<Vec<_>>();
assert_eq!(compressed, expected);
assert_eq!(compressed, plain);
}
#[test]
fn test_block_max_score_up_to_slides_and_expires_old_max() {
let total = 5 * BLOCK_SIZE as u32;
let posting = generate_posting_list(
(0..total).collect(),
5.0,
Some(vec![1.0, 4.0, 2.0, 5.0, 3.0]),
true,
);
let mut posting = PostingIterator::new(String::from("term"), 0, 0, posting, total as usize);
posting.shallow_next(0);
assert_eq!(
posting
.block_max_score_up_to_with_stats((3 * BLOCK_SIZE - 1) as u64, &UnitScorer)
.score,
4.0
);
posting.shallow_next((2 * BLOCK_SIZE) as u64);
assert_eq!(
posting
.block_max_score_up_to_with_stats((4 * BLOCK_SIZE - 1) as u64, &UnitScorer)
.score,
5.0
);
posting.shallow_next((4 * BLOCK_SIZE) as u64);
assert_eq!(
posting
.block_max_score_up_to_with_stats((5 * BLOCK_SIZE - 1) as u64, &UnitScorer)
.score,
3.0
);
}
#[test]
fn test_impact_level1_skip_keeps_boundary_equality_in_group() {
for block_size in [crate::scalar::inverted::LEGACY_BLOCK_SIZE, 256] {
let total = (IMPACT_LEVEL1_BLOCKS + 1) * block_size;
let mut posting = PostingIterator::new(
String::from("term"),
0,
0,
generate_contiguous_impact_posting_list_with_block_size(total, block_size),
total,
);
let target = (IMPACT_LEVEL1_BLOCKS * block_size - 1) as u64;
posting.shallow_next(target);
assert_eq!(posting.block_idx, IMPACT_LEVEL1_BLOCKS - 1);
posting.next(target);
assert_eq!(posting.block_idx, IMPACT_LEVEL1_BLOCKS - 1);
assert_eq!(posting.doc().map(|doc| doc.doc_id()), Some(target));
}
}
#[test]
fn test_impact_level1_skip_handles_partial_final_group() {
for block_size in [crate::scalar::inverted::LEGACY_BLOCK_SIZE, 256] {
let total = (IMPACT_LEVEL1_BLOCKS + 3) * block_size + 17;
let mut posting = PostingIterator::new(
String::from("term"),
0,
0,
generate_contiguous_impact_posting_list_with_block_size(total, block_size),
total,
);
let target = (total - 1) as u64;
let expected_block = total.div_ceil(block_size) - 1;
posting.shallow_next(target);
assert_eq!(posting.block_idx, expected_block);
posting.next(target);
assert_eq!(posting.block_idx, expected_block);
assert_eq!(posting.doc().map(|doc| doc.doc_id()), Some(target));
}
}
#[test]
fn test_impact_level1_skip_reaches_far_target_doc() {
for block_size in [crate::scalar::inverted::LEGACY_BLOCK_SIZE, 256] {
let total = (IMPACT_LEVEL1_BLOCKS * 3 + 5) * block_size;
let target_block = IMPACT_LEVEL1_BLOCKS * 2 + 2;
let target = (target_block * block_size + 17) as u64;
let mut posting = PostingIterator::new(
String::from("term"),
0,
0,
generate_contiguous_impact_posting_list_with_block_size(total, block_size),
total,
);
posting.shallow_next(target);
assert_eq!(posting.block_idx, target_block);
posting.next(target);
assert_eq!(posting.block_idx, target_block);
assert_eq!(posting.doc().map(|doc| doc.doc_id()), Some(target));
}
}
#[test]
fn test_or_impact_level1_window_skips_low_group_with_single_score() {
let total = (IMPACT_LEVEL1_BLOCKS + 1) * BLOCK_SIZE;
let target = (IMPACT_LEVEL1_BLOCKS * BLOCK_SIZE) as u64;
let mut docs = DocSet::default();
for doc_id in 0..total as u64 {
docs.append(doc_id, 1);
}
let doc_ids = (0..total as u32).collect::<Vec<_>>();
let freqs = doc_ids
.iter()
.map(|doc_id| if u64::from(*doc_id) < target { 1 } else { 10 })
.collect::<Vec<_>>();
let posting_list = generate_impact_posting_list_with_freqs(doc_ids, freqs, vec![1; total]);
let mut probe = PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
posting_list.clone(),
docs.len(),
);
probe.shallow_next(0);
let counting_scorer = CountingScorer {
scored: Arc::new(AtomicUsize::new(0)),
};
let (group_up_to, group_score) = probe.impact_group_bound(&counting_scorer).unwrap();
assert_eq!(group_up_to, target - 1);
assert_eq!(group_score, 1.0);
assert_eq!(
probe.window_max_score(Some(target - 1), &counting_scorer),
1.0
);
let posting = PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
posting_list,
docs.len(),
);
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::Or,
std::iter::once(posting),
&docs,
CountingScorer {
scored: scored.clone(),
},
);
wand.threshold = 2.0;
let (candidate, score) = wand.next().unwrap().unwrap();
assert_eq!(candidate.doc_id(), target);
assert_eq!(score, 10.0);
let total_entries = (IMPACT_LEVEL1_BLOCKS + 1) + 2;
assert!(
scored.load(Ordering::Relaxed) <= total_entries + 8,
"bounds should be baked once instead of recomputed per window; scored={}",
scored.load(Ordering::Relaxed)
);
}
#[test]
fn test_compressed_impact_block_max_score_memoizes_current_block() {
let total = 2 * BLOCK_SIZE as u32;
let doc_ids = (0..total).collect::<Vec<_>>();
let freqs = doc_ids
.iter()
.map(|doc_id| if *doc_id < BLOCK_SIZE as u32 { 1 } else { 2 })
.collect::<Vec<_>>();
let doc_lengths = vec![1; total as usize];
let posting_list = generate_impact_posting_list_with_freqs(doc_ids, freqs, doc_lengths);
let mut posting =
PostingIterator::new(String::from("term"), 0, 0, posting_list, total as usize);
let scored = Arc::new(AtomicUsize::new(0));
let scorer = CountingScorer {
scored: scored.clone(),
};
let first_score = posting.block_max_score(&scorer);
assert_eq!(first_score, 1.0);
let baked = scored.load(Ordering::Relaxed);
assert!(baked >= 2);
{
let compressed = unsafe { &mut *posting.compressed_state_ptr() };
assert_eq!(
compressed.level0_cache,
Some((0, BLOCK_SIZE as u32 - 1, first_score))
);
}
let second_score = posting.block_max_score(&scorer);
assert_eq!(second_score, first_score);
assert_eq!(
scored.load(Ordering::Relaxed),
baked,
"repeated block max scores must not recompute doc weights"
);
posting.shallow_next(BLOCK_SIZE as u64);
let next_block_score = posting.block_max_score(&scorer);
assert_eq!(next_block_score, 2.0);
assert_eq!(
scored.load(Ordering::Relaxed),
baked,
"other blocks answer from the baked bounds without rescoring"
);
}
#[rstest]
#[case(0.0)]
#[case(-1.0)]
fn test_non_positive_query_weight_skips_global_impact_bound(#[case] query_weight: f32) {
let posting = PostingIterator::with_query_weight(
String::from("term"),
0,
0,
query_weight,
generate_impact_posting_list_with_freqs(vec![0], vec![1], vec![1]),
1,
);
let scored = Arc::new(AtomicUsize::new(0));
let scorer = CountingScorer {
scored: scored.clone(),
};
assert_eq!(posting.global_upper_bound(&scorer), 0.0);
assert_eq!(scored.load(Ordering::Relaxed), 0);
}
#[test]
fn test_and_candidate_prune_scores_first_term_before_full_score() {
let total_docs = 2 * BLOCK_SIZE as u32 + 1;
let mut docs = DocSet::default();
for doc_id in 0..total_docs {
let doc_tokens = if doc_id == 0 { 1 } else { 1000 };
docs.append(doc_id as u64, doc_tokens);
}
let first_docs = (0..2 * BLOCK_SIZE as u32).collect::<Vec<_>>();
let second_docs = (0..total_docs).collect::<Vec<_>>();
let postings = vec![
PostingIterator::with_query_weight(
String::from("a"),
0,
0,
1.0,
generate_posting_list(first_docs, 1.0, Some(vec![1.0, 0.001]), true),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("b"),
1,
1,
1.0,
generate_posting_list(second_docs, 1.0, Some(vec![1.0, 0.001, 0.001]), true),
docs.len(),
),
];
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
CountingScorer {
scored: scored.clone(),
},
);
let result = wand
.search(
&FtsSearchParams::new().with_limit(Some(1)),
&NoOpMetricsCollector,
)
.unwrap();
let addrs = result
.into_iter()
.map(|doc| doc.document)
.collect::<Vec<_>>();
assert_eq!(addrs, vec![0]);
let scored = scored.load(Ordering::Relaxed);
assert!(
scored <= BLOCK_SIZE + 1 + 63,
"expected candidate pruning to avoid full scoring in the first block, scored {scored}"
);
}
#[test]
fn test_and_candidate_prune_keeps_top_candidate() {
let total_docs = 2 * BLOCK_SIZE as u32 + 1;
let mut docs = DocSet::default();
for doc_id in 0..total_docs {
let doc_tokens = if doc_id == 0 { 1 } else { 1000 };
docs.append(doc_id as u64, doc_tokens);
}
let first_docs = (0..2 * BLOCK_SIZE as u32).collect::<Vec<_>>();
let second_docs = (0..total_docs).collect::<Vec<_>>();
let postings = vec![
PostingIterator::with_query_weight(
String::from("a"),
0,
0,
1.0,
generate_posting_list(first_docs, 1.0, Some(vec![1.0, 0.001]), true),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("b"),
1,
1,
1.0,
generate_posting_list(second_docs, 1.0, Some(vec![1.0, 0.001, 0.001]), true),
docs.len(),
),
];
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
InverseDocLengthScorer,
);
let metrics = LocalMetricsCollector::default();
let result = wand
.search(&FtsSearchParams::new().with_limit(Some(1)), &metrics)
.unwrap();
let addrs = result
.into_iter()
.map(|doc| doc.document)
.collect::<Vec<_>>();
assert_eq!(addrs, vec![0]);
assert_eq!(metrics.comparisons.load(Ordering::Relaxed), 1);
}
#[test]
fn test_and_candidate_prune_keeps_later_high_score_candidate() {
let mut docs = DocSet::default();
for doc_id in 0..3 {
docs.append(doc_id, 1);
}
let postings = vec![
PostingIterator::with_query_weight(
String::from("a"),
0,
0,
1.0,
generate_posting_list_with_freqs(
vec![0, 1],
vec![10, 1],
10.0,
Some(vec![10.0]),
true,
),
docs.len(),
),
PostingIterator::with_query_weight(
String::from("b"),
1,
1,
1.0,
generate_posting_list_with_freqs(
vec![0, 1, 2],
vec![1, 20, 1],
20.0,
Some(vec![20.0]),
true,
),
docs.len(),
),
];
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer);
let result = wand
.search(
&FtsSearchParams::new().with_limit(Some(1)),
&NoOpMetricsCollector,
)
.unwrap();
let addrs = result
.into_iter()
.map(|doc| doc.document)
.collect::<Vec<_>>();
assert_eq!(addrs, vec![1]);
}
#[test]
fn score_first_and_checks_first_follower_before_decoding_lead_frequency() {
let postings = [
vec![0],
vec![100, 101],
vec![100, 101, 102],
vec![100, 101, 102, 103],
]
.into_iter()
.enumerate()
.map(|(term, doc_ids)| {
unit_length_impact_posting(term as u32, doc_ids.clone(), vec![1; doc_ids.len()], 104)
});
let docs = CostOnlyDocuments {
total_docs: 104,
visible_cost_upper_bound: 104,
};
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off);
assert!(wand.score_first_and_enabled);
wand.threshold = f32::from_bits(1);
assert!(wand.next().unwrap().is_none());
assert!(wand.score_first_and_dense_range.is_none());
assert!(wand.score_first_and_norm_k.get().is_none());
assert!(
wand.lead
.iter()
.all(|posting| posting.frequency_blocks_decoded() == 0),
"the first follower miss must leapfrog before any frequency decode"
);
}
#[test]
fn score_first_and_checks_all_followers_before_decoding_frequencies() {
let postings = [
vec![0],
vec![0, 200],
vec![100, 101, 102],
vec![100, 101, 102, 103],
]
.into_iter()
.enumerate()
.map(|(term, doc_ids)| {
unit_length_impact_posting(term as u32, doc_ids.clone(), vec![1; doc_ids.len()], 201)
});
let docs = CostOnlyDocuments {
total_docs: 201,
visible_cost_upper_bound: 201,
};
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off);
assert!(wand.score_first_and_enabled);
wand.threshold = f32::from_bits(1);
assert!(wand.next().unwrap().is_none());
assert!(wand.score_first_and_dense_range.is_none());
assert!(wand.score_first_and_norm_k.get().is_none());
let decoded_by_term = wand
.lead
.iter()
.map(|posting| (posting.token_id, posting.frequency_blocks_decoded()))
.sorted_unstable()
.collect::<Vec<_>>();
assert_eq!(decoded_by_term, vec![(0, 0), (1, 0), (2, 0), (3, 0)]);
}
#[rstest]
#[case::dense(false, true, true, Some((0, MAX_POSTING_BLOCK_SIZE as u64 - 1)))]
#[case::hole(true, true, true, None)]
#[case::dead_window(false, false, true, None)]
#[case::initial_dead_window(false, false, false, None)]
fn score_first_and_certifies_only_dense_blocks(
#[case] has_hole: bool,
#[case] is_live_window: bool,
#[case] has_prior_live_window: bool,
#[case] expected: Option<(u64, u64)>,
) {
let postings = (0..4)
.map(|term| {
let doc_ids: Vec<u32> = if term == 1 && has_hole {
(0..128).chain(129..=256).chain([300]).collect()
} else {
(0..MAX_POSTING_BLOCK_SIZE as u32 + term).collect()
};
unit_length_impact_posting(term, doc_ids.clone(), vec![1; doc_ids.len()], 301)
})
.collect::<Vec<_>>();
let docs = CostOnlyDocuments {
total_docs: 301,
visible_cost_upper_bound: 301,
};
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_bulk_and_mode(BulkAndMode::Off);
wand.threshold = f32::from_bits(1);
if has_prior_live_window {
assert_eq!(wand.and_advance_target(0), 0);
} else {
wand.threshold = f32::MAX;
wand.and_move_to_next_block(0);
}
let decoded_doc_ids = wand
.lead
.iter()
.map(|posting| {
let compressed = unsafe { &mut *posting.compressed.as_ref().unwrap().get() };
(
compressed.block_idx,
std::mem::take(&mut compressed.doc_ids),
)
})
.collect::<Vec<_>>();
if !is_live_window {
wand.threshold = f32::MAX;
}
wand.prepare_score_first_and_window();
assert_eq!(wand.score_first_and_dense_range, expected);
assert!(wand.score_first_and_norm_k.get().is_none());
if expected.is_some() {
assert_eq!(wand.score_first_and_bounds_up_to, wand.up_to);
assert_eq!(
wand.score_first_and_suffix_bounds.len(),
wand.lead.len() + 1
);
} else {
assert!(wand.score_first_and_suffix_bounds.is_empty());
assert!(wand.score_first_and_bounds_up_to.is_none());
}
for (posting, (block_idx, doc_ids)) in wand.lead.iter().zip(decoded_doc_ids) {
let compressed = unsafe { &mut *posting.compressed.as_ref().unwrap().get() };
assert_eq!(compressed.block_idx, block_idx);
assert!(compressed.doc_ids.is_empty());
compressed.doc_ids = doc_ids;
}
assert!(
wand.lead
.iter()
.all(|posting| posting.frequency_blocks_decoded() == 0),
"metadata certification must not decode frequencies"
);
if !has_hole && is_live_window {
wand.seek(100);
assert!(wand.score_first_and_dense_range.is_none());
assert_eq!(wand.and_advance_target(100), 100);
assert_eq!(wand.score_first_and_dense_range, Some((100, 255)));
}
}
#[test]
fn score_first_and_certifies_short_tail_without_decoding_it() {
let postings = (0..4)
.map(|term| {
let doc_ids = (0..MAX_POSTING_BLOCK_SIZE as u32 + 1 + term).collect::<Vec<_>>();
unit_length_impact_posting(
term,
doc_ids.clone(),
vec![1; doc_ids.len()],
MAX_POSTING_BLOCK_SIZE + 4,
)
})
.collect::<Vec<_>>();
let docs = CostOnlyDocuments {
total_docs: MAX_POSTING_BLOCK_SIZE + 4,
visible_cost_upper_bound: MAX_POSTING_BLOCK_SIZE + 4,
};
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_bulk_and_mode(BulkAndMode::Off);
for posting in &mut wand.lead {
posting.next_doc_id(MAX_POSTING_BLOCK_SIZE as u64, true);
}
wand.threshold = f32::from_bits(1);
assert_eq!(
wand.and_advance_target(MAX_POSTING_BLOCK_SIZE as u64),
MAX_POSTING_BLOCK_SIZE as u64
);
assert_eq!(
wand.score_first_and_dense_range,
Some((MAX_POSTING_BLOCK_SIZE as u64, MAX_POSTING_BLOCK_SIZE as u64))
);
assert!(
wand.lead
.iter()
.all(|posting| posting.frequency_blocks_decoded() == 0)
);
}
#[test]
fn score_first_and_intersects_differently_aligned_dense_blocks() {
let starts = [0, 100, 50, 80];
let postings = starts.into_iter().enumerate().map(|(term, start)| {
let doc_ids =
(start..start + MAX_POSTING_BLOCK_SIZE as u32 + term as u32).collect::<Vec<_>>();
unit_length_impact_posting(term as u32, doc_ids.clone(), vec![1; doc_ids.len()], 400)
});
let docs = CostOnlyDocuments {
total_docs: 400,
visible_cost_upper_bound: 400,
};
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off);
wand.threshold = f32::from_bits(1);
assert_eq!(wand.and_advance_target(0), 0);
assert_eq!(wand.score_first_and_dense_range, Some((100, 255)));
assert_eq!(wand.next().unwrap().unwrap().0.doc_id(), 100);
}
#[test]
fn score_first_and_certifies_first_live_window_after_impact_skip() {
let total_docs = 2 * MAX_POSTING_BLOCK_SIZE as u32;
let postings = (0..4).map(|term| {
let doc_ids = (0..total_docs + term).collect::<Vec<_>>();
let freqs = doc_ids
.iter()
.map(|doc_id| {
if *doc_id < MAX_POSTING_BLOCK_SIZE as u32 {
1
} else {
20
}
})
.collect::<Vec<_>>();
unit_length_impact_posting(term, doc_ids, freqs, total_docs as usize + 4)
});
let docs = CostOnlyDocuments {
total_docs: total_docs as usize + 4,
visible_cost_upper_bound: total_docs as usize + 4,
};
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off);
wand.threshold = 7.0;
assert_eq!(wand.and_advance_target(0), MAX_POSTING_BLOCK_SIZE as u64);
assert!(wand.and_window_stats.windows_skipped > 0);
assert_eq!(
wand.score_first_and_dense_range,
Some((
MAX_POSTING_BLOCK_SIZE as u64,
2 * MAX_POSTING_BLOCK_SIZE as u64 - 1
))
);
}
#[test]
fn score_first_and_rejects_malformed_block_count_certificate() {
let postings = (0..4).map(|term| {
let doc_ids = (0..MAX_POSTING_BLOCK_SIZE as u32 + term).collect::<Vec<_>>();
let mut list = generate_impact_posting_list_with_freqs_and_block_size(
doc_ids.clone(),
vec![1; doc_ids.len()],
vec![1; doc_ids.len()],
MAX_POSTING_BLOCK_SIZE,
);
if term == 3 {
let PostingList::Compressed(list) = &mut list else {
unreachable!();
};
list.length += MAX_POSTING_BLOCK_SIZE as u32;
}
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
list,
2 * MAX_POSTING_BLOCK_SIZE,
)
});
let docs = CostOnlyDocuments {
total_docs: 2 * MAX_POSTING_BLOCK_SIZE,
visible_cost_upper_bound: 2 * MAX_POSTING_BLOCK_SIZE,
};
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off);
wand.threshold = f32::from_bits(1);
assert_eq!(wand.and_advance_target(0), 0);
assert!(wand.score_first_and_dense_range.is_none());
}
#[test]
fn dense_certificate_does_not_replace_membership_checks() {
let postings = [
vec![1],
vec![0, 0, 2],
vec![0, 1, 2, 3],
vec![0, 1, 2, 3, 4],
]
.into_iter()
.enumerate()
.map(|(term, doc_ids)| {
unit_length_impact_posting(term as u32, doc_ids.clone(), vec![1; doc_ids.len()], 5)
});
let docs = CostOnlyDocuments {
total_docs: 5,
visible_cost_upper_bound: 5,
};
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off);
wand.threshold = f32::from_bits(1);
assert_eq!(wand.and_advance_target(1), 1);
assert_eq!(wand.score_first_and_dense_range, Some((1, 1)));
assert!(wand.next().unwrap().is_none());
}
#[rstest]
#[case::unsupported(None, 6)]
#[case::negative(Some(-1.0), 261)]
#[case::non_finite(Some(f32::INFINITY), 261)]
fn score_first_and_norm_cache_matches_uncached_score_bits(
#[case] unsupported_norm: Option<f32>,
#[case] expected_norm_calls: usize,
) {
fn run<S: Scorer>(scorer: S, docs: &DocSet) -> (Vec<(u64, u32)>, Option<bool>) {
let postings = (0..4).map(|term| {
let doc_ids = (0..4 + term).collect::<Vec<_>>();
let freqs = doc_ids
.iter()
.map(|doc_id| (doc_id + term) % 4 + 1)
.collect::<Vec<_>>();
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
0.001_882_293 + term as f32 * 0.000_173_205,
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids.clone(),
freqs,
vec![1; doc_ids.len()],
MAX_POSTING_BLOCK_SIZE,
),
docs.len(),
)
});
let mut wand = Wand::new(Operator::And, postings, docs, scorer)
.with_bulk_and_mode(BulkAndMode::Off);
assert!(wand.score_first_and_enabled);
assert!(wand.score_first_and_norm_k.get().is_none());
wand.threshold = f32::from_bits(1);
let mut scored = Vec::new();
while let Some((doc, score)) = wand.next().unwrap() {
scored.push((doc.doc_id(), score.to_bits()));
if doc.doc_id() == 0 {
let norm_cache_ptr = wand
.score_first_and_norm_k
.get()
.unwrap()
.as_ref()
.map(|(_, cache)| cache.as_ptr());
wand.seek(1);
assert!(wand.score_first_and_suffix_bounds.is_empty());
assert!(wand.score_first_and_bounds_up_to.is_none());
assert!(wand.score_first_and_dense_range.is_none());
assert_eq!(
wand.score_first_and_norm_k
.get()
.unwrap()
.as_ref()
.map(|(_, cache)| cache.as_ptr()),
norm_cache_ptr
);
}
}
let cache_state = wand
.score_first_and_norm_k
.get()
.map(|cache| cache.is_some());
(scored, cache_state)
}
let mut docs = unit_length_docs(7);
docs.set_quantized_scoring(true);
let (cached, cached_state) = run(VariedBm25ShapeScorer, &docs);
let norm_calls = Arc::new(AtomicUsize::new(0));
let (uncached, uncached_state) = run(
CandidateOnlyBm25ShapeScorer {
norm_calls: norm_calls.clone(),
unsupported_norm,
},
&docs,
);
assert_eq!(cached_state, Some(true));
assert_eq!(uncached_state, Some(false));
assert_eq!(cached, uncached);
assert_eq!(cached.len(), 4);
assert_eq!(
norm_calls.load(Ordering::Relaxed),
expected_norm_calls,
"unsupported norm tables must be attempted once, then use the per-candidate fallback"
);
}
#[rstest]
fn score_first_and_clears_certified_window_before_phrase_search(
#[values(4, 5)] num_terms: u32,
) {
let mut docs = unit_length_docs(2);
docs.set_quantized_scoring(true);
let postings = (0..num_terms).map(|term| {
let mut list = generate_impact_posting_list_with_freqs_and_block_size(
vec![0, 1],
vec![1, 1],
vec![1, 1],
MAX_POSTING_BLOCK_SIZE,
);
let PostingList::Compressed(compressed) = &mut list else {
unreachable!();
};
let codec = PositionStreamCodec::PackedDelta;
let mut encoded = Vec::new();
encode_position_stream_block_into(&[term, term], &[1, 1], codec, &mut encoded).unwrap();
compressed.positions = Some(CompressedPositionStorage::SharedStream(
SharedPositionStream::new(codec, vec![0], bytes::Bytes::from(encoded)),
));
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
list,
docs.len(),
)
});
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off);
wand.threshold = f32::from_bits(1);
assert_eq!(wand.next().unwrap().unwrap().0.doc_id(), 0);
assert!(wand.score_first_and_enabled);
assert_eq!(wand.score_first_and_dense_range, Some((0, 1)));
assert!(wand.score_first_and_bounds_up_to.is_some());
assert!(!wand.score_first_and_suffix_bounds.is_empty());
assert!(wand.and_candidate_score.is_some());
assert!(wand.score_first_and_norm_k.get().unwrap().is_some());
let mut params = FtsSearchParams::new().with_limit(Some(10));
params.phrase_slop = Some(0);
let results = wand.search(¶ms, &NoOpMetricsCollector).unwrap();
assert_eq!(results.len(), 1);
assert_eq!(results[0].document, 1);
assert!(!wand.score_first_and_enabled);
assert!(wand.score_first_and_suffix_bounds.is_empty());
assert!(wand.score_first_and_bounds_up_to.is_none());
assert!(wand.score_first_and_dense_range.is_none());
assert!(wand.and_candidate_score.is_none());
assert!(wand.score_first_and_norm_k.get().unwrap().is_some());
}
#[test]
fn score_first_and_prunes_before_decoding_final_follower() {
let total_docs = 2 * MAX_POSTING_BLOCK_SIZE as u32;
let doc_ids = (0..total_docs).collect::<Vec<_>>();
let docs = unit_length_docs(doc_ids.len());
let run = |is_score_first_enabled| {
let postings = (0..4)
.map(|term| {
let term_doc_ids = (4 - term as u32..total_docs).collect::<Vec<_>>();
let mut freqs = vec![1; term_doc_ids.len()];
freqs[term] = 2;
if term > 0 {
freqs[MAX_POSTING_BLOCK_SIZE - 4] = 3;
freqs[term_doc_ids.len() - 4 + term] = 3;
}
unit_length_impact_posting(term as u32, term_doc_ids, freqs, docs.len())
})
.collect::<Vec<_>>();
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
CountingBm25ShapeScorer { scored },
)
.with_bulk_and_mode(BulkAndMode::Off)
.with_score_first_and(is_score_first_enabled);
let result = wand
.search(
&FtsSearchParams::new().with_limit(Some(1)),
&NoOpMetricsCollector,
)
.unwrap();
(
sorted_candidate_row_ids(result),
wand.lead
.iter()
.map(|posting| posting.frequency_blocks_decoded())
.sum::<usize>(),
wand.and_window_stats.score_first_rejections,
)
};
let (classic_docs, classic_decodes, classic_rejections) = run(false);
let (score_first_docs, score_first_decodes, score_first_rejections) = run(true);
assert_eq!(classic_docs, vec![4]);
assert_eq!(score_first_docs, classic_docs);
assert_eq!(classic_rejections, 0);
assert!(score_first_rejections > MAX_POSTING_BLOCK_SIZE);
assert!(
score_first_decodes < classic_decodes,
"partial exact scoring should avoid decoding the final follower block: score-first={score_first_decodes}, classic={classic_decodes}"
);
}
#[rstest]
#[case::four_k1(4, 1)]
#[case::four_k10(4, 10)]
#[case::four_underfilled(4, 1024)]
#[case::five_k1(5, 1)]
#[case::five_k10(5, 10)]
#[case::five_underfilled(5, 1024)]
fn score_first_and_matches_classic_exact_scores(
#[case] num_terms: usize,
#[case] limit: usize,
#[values(false, true)] is_quantized_scoring: bool,
) {
let run = |is_score_first_enabled| {
let total_docs = 3 * MAX_POSTING_BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for doc_id in 0..total_docs {
docs.append(u64::from(doc_id), doc_id % 17 + 1);
}
docs.set_quantized_scoring(is_quantized_scoring);
let postings = (0..num_terms)
.map(|term| {
let doc_ids = (term as u32 * 3..total_docs).collect::<Vec<_>>();
let freqs = doc_ids
.iter()
.map(|doc_id| {
if *doc_id + 1 == total_docs {
20 + term as u32
} else {
(*doc_id + term as u32) % 5 + 1
}
})
.collect::<Vec<_>>();
let doc_lengths = doc_ids
.iter()
.map(|doc_id| *doc_id % 17 + 1)
.collect::<Vec<_>>();
let position = term as u32;
PostingIterator::with_query_weight(
format!("t{term}"),
term as u32,
position,
0.001_882_293 + position as f32 * 0.000_173_205,
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids,
freqs,
doc_lengths,
MAX_POSTING_BLOCK_SIZE,
),
docs.len(),
)
})
.collect::<Vec<_>>();
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_bulk_and_mode(BulkAndMode::Off)
.with_score_first_and(is_score_first_enabled);
let mut top = Vec::<(u64, f32)>::with_capacity(limit.min(total_docs as usize));
while let Some((doc, cached_score)) = wand.next().unwrap() {
let score = if wand.and_candidate_score.is_some() {
cached_score
} else {
wand.score_in_query_order(wand.documents.doc_length(&doc))
};
top.push((doc.doc_id(), score));
top.sort_unstable_by(|left, right| {
right
.1
.total_cmp(&left.1)
.then_with(|| left.0.cmp(&right.0))
});
top.truncate(limit);
if top.len() == limit {
wand.update_threshold(top[limit - 1].1, 1.0);
}
}
(
top.into_iter()
.map(|(doc, score)| (doc, score.to_bits()))
.collect::<Vec<_>>(),
wand.and_window_stats.score_first_rejections,
)
};
let (classic, classic_rejections) = run(false);
let (score_first, score_first_rejections) = run(true);
assert_eq!(score_first, classic);
assert_eq!(classic_rejections, 0);
if limit == 1 {
assert_eq!(score_first[0].0, 3 * MAX_POSTING_BLOCK_SIZE as u64 - 1);
}
if limit > 3 * MAX_POSTING_BLOCK_SIZE {
assert_eq!(score_first_rejections, 0);
} else {
assert!(score_first_rejections > 0);
}
}
#[rstest]
fn score_first_and_matches_doc_first_across_dense_and_sparse_windows(
#[values(4, 5)] num_terms: u32,
) {
let total_docs = 3 * MAX_POSTING_BLOCK_SIZE as u32 + 1;
let missing_doc = MAX_POSTING_BLOCK_SIZE as u32 + 128;
let mut docs = unit_length_docs(total_docs as usize);
docs.set_quantized_scoring(true);
let run = |is_score_first_enabled| {
let postings = (0..num_terms).map(|term| {
let doc_ids = (0..total_docs)
.filter(|doc_id| *doc_id != missing_doc)
.collect::<Vec<_>>();
let freqs = doc_ids
.iter()
.map(|doc_id| (doc_id + term) % 4 + 1)
.collect::<Vec<_>>();
PostingIterator::with_query_weight(
format!("t{term}"),
term,
num_terms - term - 1,
0.001_882_293 + term as f32 * 0.000_173_205,
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids.clone(),
freqs,
vec![1; doc_ids.len()],
MAX_POSTING_BLOCK_SIZE,
),
docs.len(),
)
});
let mut wand = Wand::new(Operator::And, postings, &docs, VariedBm25ShapeScorer)
.with_bulk_and_mode(BulkAndMode::Off)
.with_score_first_and(is_score_first_enabled);
let mut scored = Vec::new();
while let Some((doc, cached_score)) = wand.next().unwrap() {
let doc_id = doc.doc_id();
let is_dense = doc_id < MAX_POSTING_BLOCK_SIZE as u64
|| doc_id > 2 * MAX_POSTING_BLOCK_SIZE as u64;
assert_eq!(
wand.and_candidate_score.is_some(),
is_score_first_enabled && doc_id > 0 && is_dense
);
let score = if wand.and_candidate_score.is_some() {
cached_score
} else {
wand.score_in_query_order(1)
};
scored.push((doc_id, score.to_bits()));
if doc_id == 0 {
assert!(wand.score_first_and_norm_k.get().is_none());
wand.update_threshold(f32::from_bits(1), 1.0);
}
}
scored
};
let classic = run(false);
let score_first = run(true);
assert_eq!(score_first, classic);
assert_eq!(score_first.len(), total_docs as usize - 1);
}
#[test]
fn score_first_and_restores_shallow_probe_and_keeps_later_window() {
let total_docs = 3 * MAX_POSTING_BLOCK_SIZE as u32;
let mut docs = DocSet::default();
for doc_id in 0..total_docs {
docs.append(u64::from(doc_id), doc_id % 11 + 1);
}
let postings = (0..4)
.map(|term| {
let doc_ids = (0..total_docs).collect::<Vec<_>>();
let freqs = doc_ids
.iter()
.map(|doc_id| {
if *doc_id + 1 == total_docs {
32
} else {
(*doc_id + term) % 4 + 1
}
})
.collect::<Vec<_>>();
let doc_lengths = doc_ids
.iter()
.map(|doc_id| *doc_id % 11 + 1)
.collect::<Vec<_>>();
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
0.002 + term as f32 * 0.000_31,
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids,
freqs,
doc_lengths,
MAX_POSTING_BLOCK_SIZE,
),
docs.len(),
)
})
.collect::<Vec<_>>();
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_score_first_and(true)
.with_floor_mode(CompetitiveFloorMode::Inclusive);
wand.threshold = f32::from_bits(1);
assert_eq!(wand.and_advance_target(0), 0);
assert!(wand.score_first_and_bounds_up_to.is_some());
let internal_up_to = wand.up_to;
let internal_suffix_bounds = wand.score_first_and_suffix_bounds.clone();
let internal_dense_range = wand.score_first_and_dense_range;
let (shallow_up_to, _) = wand.compound_shallow_bound(300);
assert!(shallow_up_to >= 300);
assert_eq!(wand.up_to, internal_up_to);
assert_eq!(wand.score_first_and_suffix_bounds, internal_suffix_bounds);
assert_eq!(wand.score_first_and_dense_range, internal_dense_range);
wand.seek(300);
assert!(wand.up_to.is_none());
assert!(wand.score_first_and_suffix_bounds.is_empty());
assert!(wand.score_first_and_dense_range.is_none());
let mut best_score = 0.0_f32;
let mut saw_late_winner = false;
while let Some((doc, score)) = wand.next().unwrap() {
let expected = wand.score_in_query_order(wand.documents.doc_length(&doc));
assert_eq!(score.to_bits(), expected.to_bits());
if score > best_score {
best_score = score;
wand.threshold = score;
}
if doc.doc_id() + 1 == u64::from(total_docs) {
saw_late_winner = true;
}
}
assert!(saw_late_winner);
assert!(wand.and_window_stats.score_first_rejections > 0);
}
#[test]
fn score_first_and_reanchors_after_future_shallow_probe() {
let run = |is_score_first_enabled| {
let total_docs = 3 * MAX_POSTING_BLOCK_SIZE as u32;
let postings = (0..4)
.map(|term| {
let doc_ids = (0..total_docs).collect::<Vec<_>>();
let mut freqs = vec![1; doc_ids.len()];
freqs[0] = 20;
unit_length_impact_posting(term, doc_ids, freqs, total_docs as usize)
})
.collect::<Vec<_>>();
let docs = unit_length_docs(total_docs as usize);
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_score_first_and(is_score_first_enabled);
wand.threshold = 7.0;
let _ = wand.compound_shallow_bound(600);
let _ = wand.compound_shallow_bound(512);
let (doc, cached_score) = wand.next().unwrap().unwrap();
let score = if wand.and_candidate_score.is_some() {
cached_score
} else {
wand.score_in_query_order(wand.documents.doc_length(&doc))
};
(doc.doc_id(), score.to_bits())
};
let classic = run(false);
let score_first = run(true);
assert_eq!(score_first, classic);
assert_eq!(score_first.0, 0);
}
#[rstest]
fn score_first_and_inclusive_tie_uses_query_order(
#[values(4_usize, 5_usize)] num_terms: usize,
#[values(false, true)] is_quantized_scoring: bool,
) {
let mut contributions = SmallVec::<[ScoreContribution; 8]>::new();
let postings = (0..num_terms)
.map(|term| {
let position = (num_terms - term - 1) as u32;
let doc_ids = (100 - term as u32..=100).collect::<Vec<_>>();
let mut freqs = vec![1; doc_ids.len()];
freqs[term] = term as u32 + 1;
let query_weight = 0.001_882_293 + position as f32 * 0.000_173_205;
contributions.push((
(position, term as u32),
query_weight * bm25_doc_weight_with_norm(term as u32 + 1, 0.4),
));
PostingIterator::with_query_weight(
format!("t{term}"),
term as u32,
position,
query_weight,
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids.clone(),
freqs,
vec![1; doc_ids.len()],
MAX_POSTING_BLOCK_SIZE,
),
101,
)
})
.collect::<Vec<_>>();
let expected_score = score_contributions_in_query_order(contributions);
let mut docs = unit_length_docs(101);
docs.set_quantized_scoring(is_quantized_scoring);
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_floor_mode(CompetitiveFloorMode::Inclusive);
assert!(wand.score_first_and_enabled);
wand.threshold = expected_score;
let (doc, score) = wand.next().unwrap().unwrap();
assert_eq!(doc.doc_id(), 100);
assert_eq!(score.to_bits(), expected_score.to_bits());
}
#[rstest]
fn score_first_and_handles_u32_max_doc_id(#[values(0.0, 0.1)] floor: f32) {
let postings = (0..4)
.map(|term| unit_length_impact_posting(term, vec![u32::MAX], vec![1], 1))
.collect::<Vec<_>>();
let docs = CostOnlyDocuments {
total_docs: 1,
visible_cost_upper_bound: 1,
};
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_score_first_and(true);
wand.threshold = floor;
assert_eq!(
wand.next().unwrap().unwrap().0.doc_id(),
u64::from(u32::MAX)
);
assert!(wand.score_first_and_dense_range.is_none());
if floor == 0.0 {
assert_eq!(wand.up_to, Some(TERMINATED_DOC_ID));
}
assert!(wand.next().unwrap().is_none());
}
#[rstest]
#[case::local_heap("local")]
#[case::shared_floor("shared")]
#[case::compound_floor("compound")]
fn score_first_and_replaces_zero_floor_window_after_floor_update(
#[case] floor_source: &str,
#[values(4, 5)] num_terms: u32,
) {
let total_docs = MAX_POSTING_BLOCK_SIZE as u32 + 3;
let docs = unit_length_docs(total_docs as usize);
let postings = (0..num_terms)
.map(|term| {
let doc_ids = (0..total_docs).collect::<Vec<_>>();
unit_length_impact_posting(term, doc_ids, vec![1; total_docs as usize], docs.len())
})
.collect();
let params = FtsSearchParams::default();
let metrics = NoOpMetricsCollector;
let mut cursor = WandCursor::new(
Operator::And,
postings,
&docs,
Arc::new(MemBM25Scorer::new(
total_docs as u64,
total_docs as usize,
std::collections::HashMap::new(),
)),
¶ms,
&metrics,
);
assert!(cursor.wand.score_first_and_enabled);
assert_eq!(cursor.next().unwrap(), Some(0));
assert_eq!(cursor.wand.up_to, Some(TERMINATED_DOC_ID));
assert_eq!(cursor.next().unwrap(), Some(1));
assert_eq!(cursor.wand.up_to, Some(TERMINATED_DOC_ID));
assert_eq!(cursor.advance(10).unwrap(), Some(10));
assert_eq!(cursor.wand.up_to, Some(TERMINATED_DOC_ID));
let current_score = cursor.current_score().unwrap();
let shallow_up_to = cursor
.advance_shallow(MAX_POSTING_BLOCK_SIZE as u64)
.unwrap();
let shallow_bound = cursor.score_upper_bound(shallow_up_to).unwrap();
assert_eq!(cursor.wand.up_to, Some(TERMINATED_DOC_ID));
assert!(cursor.wand.score_first_and_suffix_bounds.is_empty());
assert!(cursor.wand.score_first_and_norm_k.get().is_none());
let floor = f32::from_bits(1);
match floor_source {
"local" => cursor.wand.update_threshold(floor, 1.0),
"shared" => {
cursor.wand.shared_threshold = Some(Arc::new(AtomicU32::new(floor.to_bits())));
cursor.wand.raise_to_shared_floor(1.0);
}
"compound" => cursor.set_min_competitive_score(floor).unwrap(),
_ => unreachable!(),
}
assert!(cursor.wand.up_to.is_none());
assert_eq!(cursor.doc(), Some(10));
assert_eq!(
cursor.current_score().unwrap().to_bits(),
current_score.to_bits()
);
assert_eq!(
cursor.score_upper_bound(shallow_up_to).unwrap(),
shallow_bound
);
assert_eq!(cursor.next().unwrap(), Some(11));
assert_eq!(cursor.wand.up_to, Some(MAX_POSTING_BLOCK_SIZE as u64 - 1));
assert!(cursor.wand.score_first_and_dense_range.is_some());
assert!(cursor.wand.and_candidate_score.is_some());
assert_eq!(
cursor.current_score().unwrap().to_bits(),
current_score.to_bits()
);
}
#[rstest]
fn score_first_and_underfilled_window_stays_unbounded(
#[values(4, 5)] num_terms: u32,
#[values(0, 5)] num_matches: u32,
) {
let sparse_end = 5 * MAX_POSTING_BLOCK_SIZE as u32;
let total_docs = sparse_end + num_matches;
let docs = unit_length_docs(total_docs as usize);
let make_postings = || {
(0..num_terms)
.map(|term| {
let doc_ids = (term..sparse_end)
.step_by(num_terms as usize)
.chain(sparse_end..total_docs)
.collect::<Vec<_>>();
unit_length_impact_posting(
term,
doc_ids.clone(),
vec![1; doc_ids.len()],
docs.len(),
)
})
.collect::<Vec<_>>()
};
let run = |is_eligible| {
let mut wand = Wand::new(
Operator::And,
make_postings().into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_bulk_and_mode(BulkAndMode::Off);
wand.score_first_and_enabled = is_eligible;
let mut scored = Vec::new();
while let Some((doc, _)) = wand.next().unwrap() {
scored.push((doc.doc_id(), wand.score_in_query_order(1).to_bits()));
if is_eligible {
assert_eq!(wand.up_to, Some(TERMINATED_DOC_ID));
}
}
if is_eligible {
assert_eq!(wand.up_to, Some(TERMINATED_DOC_ID));
}
assert!(wand.score_first_and_suffix_bounds.is_empty());
assert!(wand.score_first_and_dense_range.is_none());
assert!(wand.score_first_and_norm_k.get().is_none());
assert!(
wand.lead
.iter()
.all(|posting| posting.impact_bound_computations() == 0)
);
scored
};
let expected = run(false);
assert_eq!(run(true), expected);
assert_eq!(expected.len(), num_matches as usize);
let mut wand = Wand::new(
Operator::And,
make_postings().into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.with_bulk_and_mode(BulkAndMode::Off);
let results = wand
.search(
&FtsSearchParams::new().with_limit(Some(10)),
&NoOpMetricsCollector,
)
.unwrap();
assert_eq!(results.len(), num_matches as usize);
assert_eq!(wand.threshold, 0.0);
assert_eq!(wand.up_to, Some(TERMINATED_DOC_ID));
let mut actual_docs = results
.into_iter()
.map(|candidate| candidate.document)
.collect::<Vec<_>>();
actual_docs.sort_unstable();
assert_eq!(
actual_docs,
expected.into_iter().map(|(doc, _)| doc).collect::<Vec<_>>()
);
}
#[test]
fn score_first_and_requires_bm25_shape_and_positive_floor() {
let doc_ids = vec![0, 1];
let make_postings = || {
(0..4)
.map(|term| {
unit_length_impact_posting(term, doc_ids.clone(), vec![1, 1], doc_ids.len())
})
.collect::<Vec<_>>()
};
let docs = unit_length_docs(2);
let mut wand = Wand::new(
Operator::And,
make_postings().into_iter(),
&docs,
UnitScorer,
);
assert!(!wand.score_first_and_enabled);
assert_eq!(wand.next().unwrap().unwrap().0.doc_id(), 0);
assert_eq!(wand.and_window_stats.score_first_rejections, 0);
let scored = Arc::new(AtomicUsize::new(0));
let mut underfilled = Wand::new(
Operator::And,
make_postings().into_iter(),
&docs,
CountingBm25ShapeScorer { scored },
);
assert!(underfilled.score_first_and_enabled);
let results = underfilled
.search(
&FtsSearchParams::new().with_limit(Some(3)),
&NoOpMetricsCollector,
)
.unwrap();
assert_eq!(results.len(), 2);
assert_eq!(underfilled.threshold, 0.0);
assert_eq!(underfilled.and_window_stats.score_first_rejections, 0);
assert!(underfilled.score_first_and_dense_range.is_none());
assert!(underfilled.score_first_and_norm_k.get().is_none());
let mut non_finite = Wand::new(
Operator::And,
make_postings().into_iter(),
&docs,
NonFiniteBoundBm25ShapeScorer,
);
assert!(non_finite.score_first_and_enabled);
non_finite.threshold = 1.0;
assert_eq!(non_finite.next().unwrap().unwrap().0.doc_id(), 0);
assert!(non_finite.score_first_and_suffix_bounds.is_empty());
assert!(non_finite.score_first_and_bounds_up_to.is_none());
assert!(non_finite.score_first_and_dense_range.is_none());
assert!(non_finite.and_candidate_score.is_none());
assert_eq!(non_finite.and_window_stats.score_first_rejections, 0);
}
#[test]
fn score_first_and_selection_stays_within_modern_four_or_five_clause_path() {
let docs = unit_length_docs(2);
let modern = |num_terms, block_size| {
(0..num_terms)
.map(|term| {
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
generate_impact_posting_list_with_freqs_and_block_size(
vec![0, 1],
vec![1, 1],
vec![1, 1],
block_size,
),
docs.len(),
)
})
.collect::<Vec<_>>()
};
for num_terms in [2, 3] {
let wand = Wand::new(
Operator::And,
modern(num_terms, MAX_POSTING_BLOCK_SIZE).into_iter(),
&docs,
VariedBm25ShapeScorer,
);
assert!(!wand.score_first_and_enabled);
}
let plain = (0..4).map(|term| {
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
generate_posting_list(vec![0, 1], 1.0, None, false),
docs.len(),
)
});
assert!(
!Wand::new(Operator::And, plain, &docs, VariedBm25ShapeScorer).score_first_and_enabled
);
let no_impact_non_256 = (0..4).map(|term| {
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
generate_posting_list(vec![0, 1], 1.0, None, true),
docs.len(),
)
});
assert!(
!Wand::new(
Operator::And,
no_impact_non_256,
&docs,
VariedBm25ShapeScorer,
)
.score_first_and_enabled
);
assert!(
!Wand::new(
Operator::And,
modern(4, crate::scalar::inverted::LEGACY_BLOCK_SIZE).into_iter(),
&docs,
VariedBm25ShapeScorer,
)
.score_first_and_enabled
);
}
#[rstest]
fn test_wand_batches_lagging_iterators(#[values(false, true)] is_compressed: bool) {
let mut docs = DocSet::default();
for i in 0..16 {
docs.append(i as u64, 1);
}
let postings = vec![
PostingIterator::new(
String::from("a"),
0,
0,
generate_posting_list(vec![1, 10], 1.0, None, is_compressed),
docs.len(),
),
PostingIterator::new(
String::from("b"),
1,
1,
generate_posting_list(vec![2, 10], 1.0, None, is_compressed),
docs.len(),
),
PostingIterator::new(
String::from("c"),
2,
2,
generate_posting_list(vec![10], 1.0, None, is_compressed),
docs.len(),
),
];
let mut wand = Wand::new(Operator::Or, postings.into_iter(), &docs, UnitScorer);
wand.threshold = 2.5;
let candidate = wand.next().unwrap().unwrap();
assert_eq!(candidate.0.doc_id(), 10);
assert_eq!(wand.lead.len(), 3);
}
#[test]
fn test_flat_search_or_keeps_masked_docs_in_same_block() {
let mut docs = DocSet::default();
for i in 0..=(BLOCK_SIZE as u64 + 1) {
let doc_tokens = if i == 1 { 100 } else { 1 };
docs.append(i, doc_tokens);
}
let posting = PostingIterator::with_query_weight(
String::from("term"),
0,
0,
1.0,
generate_posting_list(
(1..=(BLOCK_SIZE as u32 + 1)).collect(),
1.0,
Some(vec![1.0, 1.0]),
true,
),
docs.len(),
);
let mut wand = Wand::new(
Operator::Or,
vec![posting].into_iter(),
&docs,
InverseDocLengthScorer,
);
wand.threshold = 0.5;
let selected = vec![(1_u64, 1_u64), (2_u64, 2_u64)];
let result = wand
.flat_search(
&FtsSearchParams::default(),
Box::new(selected.into_iter()),
&NoOpMetricsCollector,
)
.unwrap();
let matched = result
.into_iter()
.map(|doc| doc.document)
.collect::<Vec<_>>();
assert_eq!(matched, vec![2]);
}
#[test]
fn test_doc_ids_resolves_every_document_a_row_owns() {
let row_id_col = arrow_array::UInt64Array::from(vec![100_u64, 100, 101]);
let num_tokens_col = arrow_array::UInt32Array::from(vec![1_u32, 1, 1]);
let docs = DocSet::from_columns(&row_id_col, &num_tokens_col, false, None).unwrap();
assert_eq!(docs.doc_ids(100).collect::<Vec<_>>(), vec![0, 1]);
assert_eq!(docs.doc_ids(101).collect::<Vec<_>>(), vec![2]);
assert!(docs.doc_ids(999).next().is_none());
let mut legacy = DocSet::default();
legacy.append(7, 1);
assert_eq!(legacy.doc_ids(7).collect::<Vec<_>>(), vec![7]);
assert!(legacy.doc_ids(8).next().is_none());
}
#[rstest]
fn test_flat_search_finds_list_row_with_match_at_non_last_position(
#[values(false, true)] is_compressed: bool,
) {
let row_id_col = arrow_array::UInt64Array::from(vec![100_u64, 100, 101]);
let num_tokens_col = arrow_array::UInt32Array::from(vec![1_u32, 1, 1]);
let docs = DocSet::from_columns(&row_id_col, &num_tokens_col, false, None).unwrap();
let posting = PostingIterator::with_query_weight(
String::from("needle"),
0,
0,
1.0,
generate_posting_list(vec![0], 1.0, None, is_compressed),
docs.len(),
);
let mut wand = Wand::new(
Operator::Or,
vec![posting].into_iter(),
&docs,
InverseDocLengthScorer,
);
wand.threshold = 0.5;
let selected = docs
.doc_ids(100)
.map(|doc_id| (doc_id, 100_u64))
.collect::<Vec<_>>();
let result = wand
.flat_search(
&FtsSearchParams::default(),
Box::new(selected.into_iter()),
&NoOpMetricsCollector,
)
.unwrap();
let addrs = result
.into_iter()
.map(|doc| doc.document)
.collect::<Vec<_>>();
assert!(
addrs.as_slice() == [100],
"expected exactly row 100, got {addrs:?}"
);
}
#[test]
fn test_block_max_score_matches_stored_value() {
let doc_ids = vec![0_u32];
let block_max_scores = vec![0.7_f32];
let posting_list = generate_posting_list(doc_ids, 0.7, Some(block_max_scores), true);
let expected = match &posting_list {
PostingList::Compressed(list) => list.block_max_score(0),
PostingList::Plain(_) => unreachable!("expected compressed posting list"),
};
let posting = PostingIterator::new(String::from("test"), 0, 0, posting_list, 1);
let actual = posting.block_max_score(&UnitScorer);
assert!(
(actual - expected).abs() < 1e-6,
"block max score should match stored value"
);
}
#[test]
fn test_modern_doc_id_validation_checks_layout_and_upper_bound() {
let compressed = generate_posting_list(vec![0, 4], 1.0, None, true);
let posting = PostingIterator::new(String::from("term"), 0, 0, compressed, 5);
posting
.validate_modern_doc_ids(5)
.expect("largest DocId is inside the document table");
let error = posting.validate_modern_doc_ids(4).unwrap_err();
assert!(error.to_string().contains("DocId 4"));
assert!(error.to_string().contains("[0, 4)"));
let plain = generate_posting_list(vec![0], 1.0, None, false);
let posting = PostingIterator::new(String::from("legacy"), 0, 0, plain, 1);
let error = posting.validate_modern_doc_ids(1).unwrap_err();
assert!(error.to_string().contains("legacy row-address layout"));
}
#[test]
fn test_256_document_blocks_without_impacts_use_conservative_quantized_score_bound() {
let exact_doc_length = 300;
let quantized_doc_length = super::super::index::dequantize_doc_length(
super::super::index::quantize_doc_length(exact_doc_length),
);
assert!(quantized_doc_length < exact_doc_length);
let scorer = Arc::new(MemBM25Scorer::new(100, 1, Default::default()));
let stored_exact_score = scorer.doc_weight(1, exact_doc_length);
let quantized_score = scorer.doc_weight(1, quantized_doc_length);
assert!(quantized_score > stored_exact_score);
let doc_ids = [0_u32];
let frequencies = [1_u32];
let blocks = compress_posting_list_with_tail_codec_and_block_size(
doc_ids.len(),
doc_ids.iter(),
frequencies.iter(),
std::iter::once(stored_exact_score),
crate::scalar::inverted::PostingTailCodec::VarintDelta,
MAX_POSTING_BLOCK_SIZE,
)
.unwrap();
let posting_list = PostingList::Compressed(CompressedPostingList::new(
blocks,
stored_exact_score,
doc_ids.len() as u32,
crate::scalar::inverted::PostingTailCodec::VarintDelta,
MAX_POSTING_BLOCK_SIZE,
None,
None,
));
let posting = PostingIterator::new(String::from("term"), 0, 0, posting_list, doc_ids.len());
let expected_bound = BM25_DOC_WEIGHT_UPPER_BOUND;
assert_eq!(posting.approximate_upper_bound(), expected_bound);
assert_eq!(posting.global_upper_bound(&scorer), expected_bound);
assert_eq!(posting.block_max_score(&scorer), expected_bound);
assert_eq!(
posting.block_max_score_up_to_with_stats(0, &scorer).score,
expected_bound
);
assert!(expected_bound >= quantized_score);
}
#[test]
fn test_256_document_blocks_without_impacts_unknown_scorer_uses_infinite_bound() {
let doc_ids = [0_u32];
let frequencies = [10_u32];
let blocks = compress_posting_list_with_tail_codec_and_block_size(
doc_ids.len(),
doc_ids.iter(),
frequencies.iter(),
std::iter::once(10.0),
crate::scalar::inverted::PostingTailCodec::VarintDelta,
MAX_POSTING_BLOCK_SIZE,
)
.unwrap();
let posting_list = PostingList::Compressed(CompressedPostingList::new(
blocks,
10.0,
doc_ids.len() as u32,
crate::scalar::inverted::PostingTailCodec::VarintDelta,
MAX_POSTING_BLOCK_SIZE,
None,
None,
));
let posting = PostingIterator::new(String::from("term"), 0, 0, posting_list, doc_ids.len());
assert!(posting.global_upper_bound(&UnitScorer).is_infinite());
assert!(posting.block_max_score(&UnitScorer).is_infinite());
assert!(
posting
.block_max_score_up_to_with_stats(0, &UnitScorer)
.score
.is_infinite()
);
}
#[rstest]
fn test_exact_phrase_with_repeated_terms(#[values(false, true)] is_compressed: bool) {
let mut docs = DocSet::default();
docs.append(0, 16);
let token_a_positions = vec![vec![1_u32, 3, 10]];
let token_b_positions = vec![vec![2_u32, 11]];
let postings = vec![
PostingIterator::new(
String::from("a"),
0,
0,
generate_posting_list_with_positions(
vec![0],
token_a_positions.clone(),
1.0,
is_compressed,
),
docs.len(),
),
PostingIterator::new(
String::from("b"),
1,
1,
generate_posting_list_with_positions(
vec![0],
token_b_positions,
1.0,
is_compressed,
),
docs.len(),
),
PostingIterator::new(
String::from("a"),
2,
2,
generate_posting_list_with_positions(
vec![0],
token_a_positions,
1.0,
is_compressed,
),
docs.len(),
),
];
let bm25 = IndexBM25Scorer::new(std::iter::empty());
let wand = Wand::new(Operator::And, postings.into_iter(), &docs, bm25);
assert!(wand.check_exact_positions().unwrap());
assert!(wand.check_positions(0).unwrap());
}
#[rstest]
fn test_exact_phrase_respects_query_position_gaps(#[values(false, true)] is_compressed: bool) {
let mut docs = DocSet::default();
docs.append(0, 16);
let postings = vec![
PostingIterator::new(
String::from("want"),
0,
0,
generate_posting_list_with_positions(
vec![0],
vec![vec![0_u32]],
1.0,
is_compressed,
),
docs.len(),
),
PostingIterator::new(
String::from("apple"),
1,
2,
generate_posting_list_with_positions(
vec![0],
vec![vec![2_u32]],
1.0,
is_compressed,
),
docs.len(),
),
];
let bm25 = IndexBM25Scorer::new(std::iter::empty());
let wand = Wand::new(Operator::And, postings.into_iter(), &docs, bm25);
assert!(wand.check_exact_positions().unwrap());
assert!(wand.check_positions(0).unwrap());
}
#[rstest]
fn test_and_phrase_miss_advances_to_next_candidate(#[values(false, true)] is_compressed: bool) {
let mut docs = DocSet::default();
docs.append(0, 8);
docs.append(1, 8);
let postings = vec![
PostingIterator::new(
String::from("a"),
0,
0,
generate_posting_list_with_positions(
vec![0, 1],
vec![vec![1_u32], vec![10_u32]],
1.0,
is_compressed,
),
docs.len(),
),
PostingIterator::new(
String::from("b"),
1,
1,
generate_posting_list_with_positions(
vec![0, 1],
vec![vec![3_u32], vec![11_u32]],
1.0,
is_compressed,
),
docs.len(),
),
];
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer);
let first = wand.next().unwrap().unwrap();
assert_eq!(first.0.doc_id(), 0);
assert!(!wand.check_positions(0).unwrap());
wand.threshold = 1.5;
let second = wand.next().unwrap().unwrap();
assert_eq!(second.0.doc_id(), 1);
assert!(wand.check_positions(0).unwrap());
}
#[rstest]
#[case::and_k10(false, 0, 10, 3)]
#[case::and_k3(false, 0, 3, 3)]
#[case::and_two_clauses(false, 0, 10, 2)]
#[case::and_one_clause(false, 0, 10, 1)]
#[case::and_four_clauses(false, 0, 10, 4)]
#[case::and_five_clauses(false, 0, 10, 5)]
#[case::and_six_clauses(false, 0, 10, 6)]
#[case::and_eight_clauses(false, 0, 10, 8)]
#[case::and_twelve_clauses(false, 0, 10, 12)]
#[case::and_sixteen_clauses(false, 0, 10, 16)]
#[case::phrase_k10(true, 0, 10, 3)]
#[case::phrase_k3(true, 0, 3, 3)]
#[case::phrase_slop_three(true, 3, 10, 3)]
#[case::phrase_two_clauses(true, 0, 10, 2)]
#[case::phrase_four_clauses(true, 0, 10, 4)]
#[case::phrase_five_clauses(true, 0, 10, 5)]
#[case::phrase_six_clauses(true, 0, 10, 6)]
#[case::phrase_eight_clauses(true, 0, 10, 8)]
#[case::phrase_sixteen_clauses(true, 0, 10, 16)]
fn test_bulk_and_matches_classic(
#[case] phrase: bool,
#[case] slop: u32,
#[case] limit: usize,
#[case] num_clauses: usize,
) {
let num_docs = (BLOCK_SIZE * 8 + 37) as u32;
let mut docs = DocSet::default();
for doc_id in 0..num_docs {
docs.append(u64::from(doc_id), 32 + doc_id % 57);
}
let clause_docs = |modulus: u32, salt: u32| -> Vec<u32> {
(0..num_docs)
.filter(|doc| (doc.wrapping_mul(2654435761).wrapping_add(salt)) % modulus < 2)
.collect()
};
let clauses = [
clause_docs(3, 7),
clause_docs(4, 13),
clause_docs(5, 29),
clause_docs(3, 41),
clause_docs(3, 7),
clause_docs(4, 13),
]
.into_iter()
.cycle()
.take(num_clauses)
.collect::<Vec<_>>();
let build_postings = || {
clauses
.iter()
.enumerate()
.map(|(term_pos, doc_ids)| {
let list = if phrase {
let positions = doc_ids
.iter()
.map(|&doc| {
if doc % 2 == 0 {
vec![5 + term_pos as u32, 40 + (doc % 3)]
} else {
vec![20 + (term_pos as u32) * 4]
}
})
.collect::<Vec<_>>();
generate_posting_list_with_positions(doc_ids.clone(), positions, 8.0, true)
} else {
generate_posting_list(doc_ids.clone(), 8.0, None, true)
};
PostingIterator::with_query_weight(
format!("t{term_pos}"),
term_pos as u32,
term_pos as u32,
1.0 + term_pos as f32 * 0.5,
list,
docs.len(),
)
})
.collect::<Vec<_>>()
};
let mut params = FtsSearchParams::default().with_limit(Some(limit));
if phrase {
params.phrase_slop = Some(slop);
}
let normalize = |result: Vec<DocCandidate<u64>>| {
let mut rows = result
.into_iter()
.map(|candidate| {
(
candidate.posting_doc_id,
candidate.doc_length,
candidate.freqs,
candidate.document,
)
})
.collect::<Vec<_>>();
rows.sort_unstable();
rows
};
let run = |mode| {
let shared_floor = Arc::new(AtomicU32::new(0.0_f32.to_bits()));
let mut wand = Wand::new(
Operator::And,
build_postings().into_iter(),
&docs,
UnitScorer,
)
.with_bulk_and_mode(mode)
.with_shared_threshold(shared_floor.clone());
let rows = normalize(wand.search(¶ms, &NoOpMetricsCollector).unwrap());
let used_bulk = wand.bulk_and_searches > 0;
((rows, shared_floor.load(Ordering::Relaxed)), used_bulk)
};
let (bulk, bulk_used) = run(BulkAndMode::On);
let (classic, classic_used) = run(BulkAndMode::Off);
let (auto, auto_used) = run(BulkAndMode::Auto);
assert!(bulk_used, "on should use bulk conjunction search");
assert!(!classic_used, "off should use classic conjunction search");
assert_eq!(auto_used, matches!(num_clauses, 2 | 3));
assert!(!bulk.0.is_empty(), "test corpus should produce matches");
assert_eq!(bulk, classic);
assert_eq!(auto, classic);
}
#[rstest]
#[case::bulk_two(BulkAndMode::On, 2)]
#[case::bulk_four(BulkAndMode::On, 4)]
#[case::bulk_eight(BulkAndMode::On, 8)]
#[case::bulk_sixteen(BulkAndMode::On, 16)]
#[case::classic_four(BulkAndMode::Off, 4)]
#[case::classic_five(BulkAndMode::Off, 5)]
fn underfilled_disjoint_and_decodes_no_frequencies_or_bounds(
#[case] mode: BulkAndMode,
#[case] num_clauses: usize,
) {
let num_docs = (BLOCK_SIZE * 3) as u32;
let mut docs = DocSet::default();
for doc_id in 0..num_docs {
docs.append(u64::from(doc_id), 1);
}
let even = (0..num_docs).filter(|doc| doc % 2 == 0).collect::<Vec<_>>();
let odd = (0..num_docs).filter(|doc| doc % 2 == 1).collect::<Vec<_>>();
let all = (0..num_docs).collect::<Vec<_>>();
let postings = (0..num_clauses)
.map(|term| {
let doc_ids = match term {
0 => even.clone(),
1 => odd.clone(),
_ => all.clone(),
};
let len = doc_ids.len();
PostingIterator::with_query_weight(
format!("t{term}"),
term as u32,
term as u32,
1.0,
generate_impact_posting_list_with_freqs(doc_ids, vec![1; len], vec![1; len]),
docs.len(),
)
})
.collect::<Vec<_>>();
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
CountingScorer {
scored: scored.clone(),
},
)
.with_bulk_and_mode(mode);
let hits = wand
.search(
&FtsSearchParams::default().with_limit(Some(10)),
&NoOpMetricsCollector,
)
.unwrap();
assert!(hits.is_empty());
assert_eq!(scored.load(Ordering::Relaxed), 0);
assert_eq!(
wand.lead
.iter()
.map(|posting| posting.frequency_blocks_decoded())
.sum::<usize>(),
0
);
assert_eq!(
wand.lead
.iter()
.map(|posting| posting.impact_bound_computations())
.sum::<usize>(),
0
);
}
#[rstest]
#[case::legacy_full(crate::scalar::inverted::LEGACY_BLOCK_SIZE, 0)]
#[case::legacy_tail(crate::scalar::inverted::LEGACY_BLOCK_SIZE, 7)]
#[case::modern_full(MAX_POSTING_BLOCK_SIZE, 0)]
#[case::modern_tail(MAX_POSTING_BLOCK_SIZE, 7)]
fn underfilled_sparse_and_decodes_only_matching_frequency_blocks(
#[values(BulkAndMode::On, BulkAndMode::Off)] mode: BulkAndMode,
#[case] block_size: usize,
#[case] tail_len: usize,
) {
let posting_len = block_size * 2 + tail_len;
let first = (0..posting_len)
.map(|index| (index * 2) as u32)
.collect::<Vec<_>>();
let mut second = (0..posting_len)
.map(|index| (index * 2 + 1) as u32)
.collect::<Vec<_>>();
let last_doc = *first.last().unwrap();
*second.last_mut().unwrap() = last_doc;
let num_docs = posting_len * 2;
let mut docs = DocSet::default();
for doc_id in 0..num_docs {
docs.append(doc_id as u64, 1);
}
let postings = [first, second]
.into_iter()
.enumerate()
.map(|(term, doc_ids)| {
let len = doc_ids.len();
PostingIterator::with_query_weight(
format!("t{term}"),
term as u32,
term as u32,
1.0,
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids,
vec![1; len],
vec![1; len],
block_size,
),
docs.len(),
)
})
.collect::<Vec<_>>();
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
CountingScorer {
scored: scored.clone(),
},
)
.with_bulk_and_mode(mode);
let hits = wand
.search(
&FtsSearchParams::default().with_limit(Some(10)),
&NoOpMetricsCollector,
)
.unwrap();
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].document, u64::from(last_doc));
assert_eq!(scored.load(Ordering::Relaxed), 2);
assert_eq!(
wand.lead
.iter()
.map(|posting| posting.frequency_blocks_decoded())
.sum::<usize>(),
2,
"only the two blocks containing the surviving match need frequencies"
);
assert_eq!(
wand.lead
.iter()
.map(|posting| posting.impact_bound_computations())
.sum::<usize>(),
0
);
}
#[rstest]
#[case::dense("dense")]
#[case::sparse("sparse")]
#[case::shifted("shifted")]
#[case::underfilled_five("five")]
#[case::underfilled_zero("zero")]
#[case::max_doc_id("max")]
fn pairwise_and_matches_classic(
#[case] shape: &str,
#[values(4, 5, 8, 16)] num_clauses: usize,
#[values(0.0, 0.1)] initial_floor: f32,
) {
let posting_len = MAX_POSTING_BLOCK_SIZE * 3 + 7;
let docs = CostOnlyDocuments {
total_docs: posting_len * 2 + num_clauses,
visible_cost_upper_bound: posting_len * 2 + num_clauses,
};
let run = |mode| {
let postings = (0..num_clauses)
.map(|term| {
let doc_ids = (0..posting_len)
.map(|index| match shape {
"dense" => index as u32,
"shifted" => (index + term) as u32,
"sparse" => (index * 3) as u32,
"max" => u32::MAX - ((posting_len - index - 1) * 1_000_003) as u32,
"zero" | "five" => {
let is_last_clause = term + 1 == num_clauses;
let is_match = shape == "five" && index >= posting_len - 5;
(index * 2 + usize::from(is_last_clause && !is_match)) as u32
}
_ => unreachable!(),
})
.collect::<Vec<_>>();
let frequencies = (0..posting_len)
.map(|index| 1 + ((index + term) % 7) as u32)
.collect::<Vec<_>>();
PostingIterator::with_query_weight(
format!("t{term}"),
term as u32,
term as u32,
[0.002_972_301_6, 0.001_982_450_7, 0.001_882_293][term % 3],
generate_impact_posting_list_with_freqs_and_block_size(
doc_ids,
frequencies,
vec![1; posting_len],
MAX_POSTING_BLOCK_SIZE,
),
docs.len(),
)
})
.collect::<Vec<_>>();
let shared_floor = Arc::new(AtomicU32::new(initial_floor.to_bits()));
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer)
.with_bulk_and_mode(mode)
.with_shared_threshold(shared_floor.clone());
let hits = wand
.search(
&FtsSearchParams::default().with_limit(Some(10)),
&NoOpMetricsCollector,
)
.unwrap();
let mut rows = hits
.into_iter()
.map(|hit| (hit.document, hit.doc_length, hit.freqs))
.collect::<Vec<_>>();
rows.sort_unstable();
if mode == BulkAndMode::On {
assert_eq!(
wand.and_window_stats.pairwise_intersections > 0,
initial_floor == 0.0
);
if initial_floor == 0.0 && matches!(shape, "five" | "zero") {
assert_eq!(rows.len(), if shape == "five" { 5 } else { 0 });
assert_eq!(
wand.lead
.iter()
.map(|posting| posting.frequency_blocks_decoded())
.sum::<usize>(),
if shape == "five" { num_clauses } else { 0 },
"only final matching blocks need frequency decoding"
);
assert_eq!(
wand.lead
.iter()
.map(|posting| posting.impact_bound_computations())
.sum::<usize>(),
0
);
}
}
(rows, shared_floor.load(Ordering::Relaxed))
};
let classic = run(BulkAndMode::Off);
assert_eq!(run(BulkAndMode::On), classic);
assert_eq!(run(BulkAndMode::Auto), classic);
}
#[rstest]
fn pairwise_and_preserves_phrase_and_visibility(
#[values(4, 8, 16)] num_clauses: u32,
#[values(None, Some(0), Some(3))] phrase_slop: Option<u32>,
#[values(false, true)] is_filtered: bool,
) {
let num_docs = MAX_POSTING_BLOCK_SIZE * 2 + 7;
let mut docs = DocSet::default();
for doc in 0..num_docs {
docs.append(doc as u64, 1);
}
let mask = if is_filtered {
RowAddrMask::from_block([0, (num_docs - 1) as u64].into_iter().collect())
} else {
RowAddrMask::all_rows()
};
let documents = LegacyWandDocuments::new(&docs, &mask);
let run = |mode| {
let postings = (0..num_clauses)
.map(|term| {
let mut list = generate_impact_posting_list_with_freqs_and_block_size(
(0..num_docs as u32).collect(),
vec![1; num_docs],
vec![1; num_docs],
MAX_POSTING_BLOCK_SIZE,
);
let mut bytes = Vec::new();
let mut offsets = Vec::new();
for start in (0..num_docs).step_by(MAX_POSTING_BLOCK_SIZE) {
let end = (start + MAX_POSTING_BLOCK_SIZE).min(num_docs);
let positions = (start..end)
.map(|doc| {
if doc >= num_docs - 5 {
5 + term
} else {
5 + term * 8
}
})
.collect::<Vec<_>>();
offsets.push(bytes.len() as u32);
encode_position_stream_block_into(
&positions,
&vec![1; end - start],
PositionStreamCodec::VarintDocDelta,
&mut bytes,
)
.unwrap();
}
if let PostingList::Compressed(list) = &mut list {
list.positions = Some(CompressedPositionStorage::SharedStream(
SharedPositionStream::new(
PositionStreamCodec::VarintDocDelta,
offsets,
bytes.into(),
),
));
}
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
list,
docs.len(),
)
})
.collect::<Vec<_>>();
let mut wand = Wand::new(Operator::And, postings.into_iter(), &documents, UnitScorer)
.with_bulk_and_mode(mode);
let mut params = FtsSearchParams::default().with_limit(Some(10));
params.phrase_slop = phrase_slop;
let mut hits = wand
.search(¶ms, &NoOpMetricsCollector)
.unwrap()
.into_iter()
.map(|hit| (hit.document, hit.freqs))
.collect::<Vec<_>>();
hits.sort_unstable();
if mode == BulkAndMode::On {
assert!(wand.and_window_stats.pairwise_intersections > 0);
}
assert_eq!(
hits.len(),
if phrase_slop.is_some() {
5 - usize::from(is_filtered)
} else {
10
}
);
(hits, wand.threshold.to_bits())
};
let classic = run(BulkAndMode::Off);
assert_eq!(run(BulkAndMode::On), classic);
assert_eq!(run(BulkAndMode::Auto), classic);
}
#[rstest]
fn phrase_skips_position_confirm_when_complete_score_cannot_beat_floor(
#[values(BulkAndMode::Off, BulkAndMode::On)] mode: BulkAndMode,
#[values(0_u32, 3)] phrase_slop: u32,
#[values(true, false)] phrase_first: bool,
#[values(1_usize, 32)] limit: usize,
) {
let num_docs = 32_usize;
let phrase_doc = if phrase_first { 0 } else { num_docs - 1 };
let mut docs = DocSet::default();
for doc in 0..num_docs {
docs.append(doc as u64, 1);
}
let postings = (0..2_u32)
.map(|term| {
let mut list = generate_impact_posting_list_with_freqs_and_block_size(
(0..num_docs as u32).collect(),
vec![1; num_docs],
vec![1; num_docs],
MAX_POSTING_BLOCK_SIZE,
);
let positions = (0..num_docs as u32)
.map(|doc| {
if term == 0 {
0
} else if doc == phrase_doc as u32 {
1
} else {
50
}
})
.collect::<Vec<_>>();
let mut bytes = Vec::new();
encode_position_stream_block_into(
&positions,
&vec![1; num_docs],
PositionStreamCodec::VarintDocDelta,
&mut bytes,
)
.unwrap();
if let PostingList::Compressed(list) = &mut list {
list.positions = Some(CompressedPositionStorage::SharedStream(
SharedPositionStream::new(
PositionStreamCodec::VarintDocDelta,
vec![0],
bytes.into(),
),
));
}
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
list,
docs.len(),
)
})
.collect::<Vec<_>>();
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer)
.with_bulk_and_mode(mode);
let mut params = FtsSearchParams::default().with_limit(Some(limit));
params.phrase_slop = Some(phrase_slop);
let hits = wand.search(¶ms, &NoOpMetricsCollector).unwrap();
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].posting_doc_id, phrase_doc as u64);
assert_eq!(wand.bulk_and_searches, usize::from(mode == BulkAndMode::On));
let expected_checks = if phrase_first && limit == 1 {
1
} else {
num_docs
};
assert_eq!(wand.phrase_position_checks.get(), expected_checks);
}
#[rstest]
fn pairwise_and_requires_current_posting_blocks(
#[values(BulkAndMode::On, BulkAndMode::Auto)] mode: BulkAndMode,
#[values(2, 3, 4, 16)] num_clauses: usize,
#[values(crate::scalar::inverted::LEGACY_BLOCK_SIZE, MAX_POSTING_BLOCK_SIZE)]
block_size: usize,
#[values(false, true)] has_impacts: bool,
#[values(false, true)] has_grouped_terms: bool,
) {
let docs = CostOnlyDocuments {
total_docs: 5,
visible_cost_upper_bound: 5,
};
let postings = (0..num_clauses)
.map(|term| {
let mut list = generate_impact_posting_list_with_freqs_and_block_size(
vec![0, 2, 4, 6, 8],
vec![1; 5],
vec![1; 5],
block_size,
);
if !has_impacts && let PostingList::Compressed(list) = &mut list {
list.impacts = None;
}
let grouped_terms = has_grouped_terms.then(|| {
Arc::<[GroupedTermScorer]>::from([GroupedTermScorer::new(1.0, &list)])
});
let posting = PostingIterator::with_query_weight(
format!("t{term}"),
term as u32,
term as u32,
1.0,
list,
docs.len(),
);
if let Some(grouped_terms) = grouped_terms {
posting.with_grouped_terms(grouped_terms)
} else {
posting
}
})
.collect::<Vec<_>>();
let mut wand = Wand::new(Operator::And, postings.into_iter(), &docs, UnitScorer)
.with_bulk_and_mode(mode);
let hits = wand
.search(
&FtsSearchParams::default().with_limit(Some(10)),
&NoOpMetricsCollector,
)
.unwrap();
assert_eq!(hits.len(), 5);
assert_eq!(
wand.and_window_stats.pairwise_intersections > 0,
num_clauses >= 4
&& block_size == MAX_POSTING_BLOCK_SIZE
&& has_impacts
&& !has_grouped_terms
);
if has_grouped_terms {
assert_eq!(wand.bulk_and_searches, 0);
}
}
#[rstest]
fn bulk_and_activates_frequencies_and_bounds_after_heap_fills(
#[values(2, 4, 8, 16)] num_clauses: u32,
) {
let num_docs = (BLOCK_SIZE * 2) as u32;
let mut docs = DocSet::default();
for doc_id in 0..num_docs {
docs.append(u64::from(doc_id), 1);
}
let frequencies = (0..num_docs)
.map(|doc| if doc < BLOCK_SIZE as u32 { 1 } else { 10 })
.collect::<Vec<_>>();
let postings = (0..num_clauses)
.map(|term| {
PostingIterator::with_query_weight(
format!("t{term}"),
term,
term,
1.0,
generate_impact_posting_list_with_freqs(
(0..num_docs).collect(),
frequencies.clone(),
vec![1; num_docs as usize],
),
docs.len(),
)
})
.collect::<Vec<_>>();
let scored = Arc::new(AtomicUsize::new(0));
let mut wand = Wand::new(
Operator::And,
postings.into_iter(),
&docs,
CountingScorer {
scored: scored.clone(),
},
)
.with_bulk_and_mode(BulkAndMode::On);
let hits = wand
.search(
&FtsSearchParams::default().with_limit(Some(1)),
&NoOpMetricsCollector,
)
.unwrap();
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].document, BLOCK_SIZE as u64);
assert!(wand.threshold > 0.0);
assert!(
wand.lead
.iter()
.map(|posting| posting.frequency_blocks_decoded())
.sum::<usize>()
>= 4,
"the competitive second window must decode its frequency blocks"
);
assert!(
wand.lead
.iter()
.map(|posting| posting.impact_bound_computations())
.sum::<usize>()
> 0,
"impact bounds must activate on the first window after the heap fills"
);
assert!(scored.load(Ordering::Relaxed) > 0);
}
}