diff --git a/rust/lance-index/src/scalar/inverted/wand.rs b/rust/lance-index/src/scalar/inverted/wand.rs index 6a3b575bebe..196b1e5d67f 100644 --- a/rust/lance-index/src/scalar/inverted/wand.rs +++ b/rust/lance-index/src/scalar/inverted/wand.rs @@ -1402,9 +1402,17 @@ impl PostingIterator { if new_offset < compressed.doc_ids.len() { self.index = (block_idx << shift) + new_offset; self.block_idx = block_idx; + // Frequencies stay lazy, but once this block's stream is + // decoded every later candidate in it can carry its + // frequency for free instead of re-materializing in `doc`. + let frequency = if compressed.frequency_block_idx == Some(block_idx) { + compressed.freqs[new_offset] + } else { + 0 + }; self.current_doc = Some(DocInfo::Raw(RawDocInfo { doc_id: compressed.doc_ids[new_offset], - frequency: 0, + frequency, })); return; } @@ -1801,6 +1809,82 @@ fn score_sum_cannot_compete( type ScoreContribution = ((u32, u32), f32); +/// Largest `f32` partial score that `score_sum_cannot_compete` still rejects +/// under the exclusive floor, given the other clauses' upper bound. The +/// rejection predicate is monotone in the partial score, so every score at or +/// below the returned limit is rejected and every score above it competes. +/// Returns `None` when no finite score is rejected. +fn exclusive_partial_score_limit( + remaining_upper_bound: f64, + floor: f32, + upper_bound_factor: f64, +) -> Option { + let rejects = |score: f32| { + score_sum_cannot_compete( + score, + remaining_upper_bound, + floor, + upper_bound_factor, + CompetitiveFloorMode::Exclusive, + ) + }; + if !remaining_upper_bound.is_finite() || !floor.is_finite() { + return None; + } + // Start from the real-valued boundary and walk at most a few ULPs to the + // exact f32 boundary of the (monotone) predicate. + let mut limit = ((f64::from(floor) / upper_bound_factor) - remaining_upper_bound) as f32; + if !limit.is_finite() { + return None; + } + if rejects(limit) { + for _ in 0..8 { + let up = next_up_f32(limit); + if up == limit || !rejects(up) { + return Some(limit); + } + limit = up; + } + } else { + for _ in 0..8 { + let down = next_down_f32(limit); + if down == limit { + return None; + } + if rejects(down) { + return Some(down); + } + limit = down; + } + } + // The walk did not converge (pathological inputs); fall back to an exact + // binary search over the ordered f32 bit patterns. + let mut lo = f32::MIN; + let mut hi = f32::MAX; + if !rejects(lo) { + return None; + } + if rejects(hi) { + return Some(hi); + } + for _ in 0..64 { + let mid = f32::from_bits(((lo.to_bits() as i64 + hi.to_bits() as i64) / 2) as u32); + if mid == lo || mid == hi { + break; + } + if rejects(mid) { + lo = mid; + } else { + hi = mid; + } + } + Some(lo) +} + +fn next_down_f32(value: f32) -> f32 { + -next_up_f32(-value) +} + #[inline] fn score_contributions_in_query_order(mut contributions: SmallVec<[ScoreContribution; 8]>) -> f32 { if !contributions @@ -2236,6 +2320,8 @@ struct AndWindowStats { candidates_returned: usize, score_first_rejections: usize, pairwise_intersections: usize, + windows_sliced: usize, + windows_frequency_pruned: usize, } impl Eq for TailPosting {} @@ -3825,6 +3911,10 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { if list.block_size == MAX_POSTING_BLOCK_SIZE && list.impacts.is_some()) }); let phrase_slop = params.phrase_slop; + // The bulk path never evaluates score-first windows, so keep the block + // advance from preparing suffix bounds it would not use. + self.invalidate_score_first_and_window(); + self.score_first_and_enabled = false; let mut score_order = (0..num_lists).collect::>(); score_order.sort_unstable_by_key(|&index| { let posting = &self.lead[index]; @@ -3888,7 +3978,7 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { // the clause sup), so every skipped doc would also fail the exact // per-candidate prune: results are unchanged, the work never happens. macro_rules! merge_kernels { - ($name2:ident, $name3:ident, $docs_name2:ident, $docs_name3:ident, $geq:ident $(, #[$feat:meta])?) => { + ($name2:ident, $name3:ident, $name_n:ident, $docs_name2:ident, $docs_name3:ident, $geq:ident $(, #[$feat:meta])?) => { $(#[$feat])? unsafe fn $name2( wins: &[WindowList], @@ -3970,6 +4060,50 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { } } + $(#[$feat])? + #[allow(clippy::too_many_arguments)] + unsafe fn $name_n( + wins: &[WindowList], + freq_cannot_beat: &[bool; FREQ_LUT_BUCKETS], + cursors: &mut Vec, + docs_out: &mut Vec, + offs_out: &mut Vec, + ) { + cursors.clear(); + cursors.extend(wins.iter().map(|win| win.pos)); + let (d0, mut p0, e0) = (wins[0].docs, wins[0].pos, wins[0].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; + } + for j in 1..wins.len() { + let win = wins.get_unchecked(j); + let pos = $geq(win.docs, *cursors.get_unchecked(j), win.end, doc); + *cursors.get_unchecked_mut(j) = pos; + if pos >= win.end { + return; + } + let clause_doc = *win.docs.add(pos); + if clause_doc > doc { + p0 = $geq(d0, p0 + 1, e0, clause_doc); + continue 'outer; + } + } + docs_out.push(doc); + offs_out.push(p0 as u8); + for j in 1..wins.len() { + offs_out.push(*cursors.get_unchecked(j) as u8); + } + p0 += 1; + } + } + } + $(#[$feat])? unsafe fn $docs_name2( wins: &[WindowList], @@ -4041,6 +4175,7 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { merge_kernels!( merge_window_2, merge_window_3, + merge_window_n, merge_window_docs_2, merge_window_docs_3, find_next_geq_scalar @@ -4049,6 +4184,7 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { merge_kernels!( merge_window_2_avx2, merge_window_3_avx2, + merge_window_n_avx2, merge_window_docs_2_avx2, merge_window_docs_3_avx2, find_next_geq_avx2, @@ -4064,48 +4200,6 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { } } - #[inline] - #[allow(clippy::too_many_arguments)] - fn merge_window_n( - wins: &[WindowList], - freq_cannot_beat: &[bool; FREQ_LUT_BUCKETS], - cursors: &mut Vec, - docs_out: &mut Vec, - offs_out: &mut Vec, - ) { - 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], @@ -4221,6 +4315,18 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { let mut batch_norms: Vec = Vec::with_capacity(MAX_POSTING_BLOCK_SIZE); let mut cursor_scratch: Vec = Vec::with_capacity(num_lists); let mut intersection_scratch: Vec = Vec::new(); + // The frequency-bucket prune decisions only depend on the floor and + // the followers' block maxes, which rarely change between windows. + let mut freq_cannot_beat = [false; FREQ_LUT_BUCKETS]; + // Keyed separately from the score limit: the per-candidate prune can + // refresh the limit in the middle of a window without touching the + // table, so an unchanged limit key does not mean the table is current. + let mut freq_cannot_beat_key: Option<(u32, u64)> = None; + // Per clause: the block that was sliced last and the end of that + // slice. Windows are contiguous, so the next slice of the same block + // starts where the previous one ended instead of at a binary search. + let mut slice_prev: SmallVec<[(usize, usize); 8]> = + std::iter::repeat_n((usize::MAX, 0), num_lists).collect(); // Norm cache: one byte-norm load plus a cached addend replaces the // per-clause BM25 denominator recompute in pass B. let mut norm_k = None; @@ -4237,6 +4343,11 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { // candidate still goes through the exact per-candidate prune below. const FREQ_LUT_BUCKETS: usize = 64; let mut freq_bound_lut: Option<[f32; FREQ_LUT_BUCKETS]> = None; + // Exclusive limit on the first clause's partial score for the current + // window: scores at or below it cannot beat the floor. Recomputed only + // when the floor or the followers' block maxes change. + let mut first_score_limit: Option = None; + let mut first_score_limit_key: Option<(u32, u64)> = None; // The conjunction can only start at the max of the clauses' first docs. let mut target: u64 = 0; @@ -4248,6 +4359,10 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { } 'window: loop { + #[cfg(test)] + { + self.and_window_stats.windows_sliced += 1; + } self.raise_to_shared_floor(params.wand_factor); let window_started_with_floor = self.threshold > 0.0; if window_started_with_floor { @@ -4295,26 +4410,49 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { wins.clear(); let mut skip_window = false; let mut exhausted = false; - for posting in &self.lead { + // When a clause has no document in the window, the conjunction + // cannot match before that clause's next document, so the next + // window starts there instead of at the next block boundary. A + // sparse lead then drives the traversal and dense followers skip + // whole blocks by metadata instead of being sliced block by block. + let mut jump_to: u64 = 0; + for (j, posting) in self.lead.iter().enumerate() { 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() + let doc_ids = state.doc_ids.as_slice(); + let len = doc_ids.len(); + // Everything before the previous slice's end of this block is + // below the current target, so resume from there. + let (prev_block, prev_hi) = slice_prev[j]; + let start = if prev_block == block_idx { + prev_hi.min(len) + } else { + 0 + }; + let lo = if start < len && doc_ids[start] >= target32 { + start + } else { + start + doc_ids[start..].partition_point(|&doc| doc < target32) + }; + let hi = if win_end32 == u32::MAX || len == 0 || doc_ids[len - 1] <= win_end32 { + len } else { - lo + state.doc_ids[lo..].partition_point(|&doc| doc <= win_end32) + lo + doc_ids[lo..].partition_point(|&doc| doc <= win_end32) }; + slice_prev[j] = (block_idx, hi); if lo == hi { // No docs of this clause in the window: the whole window // has no conjunction match. If this was the clause's last // block and it is fully behind the target, the clause is // exhausted and the conjunction is done. - if block_idx + 1 >= list.blocks.len() - && state.doc_ids.last().is_none_or(|&doc| doc < target32) - { + if lo < state.doc_ids.len() { + jump_to = u64::from(state.doc_ids[lo]); + } else if block_idx + 1 < list.blocks.len() { + jump_to = u64::from(list.block_least_doc_id(block_idx + 1)); + } else { exhausted = true; } skip_window = true; @@ -4351,7 +4489,7 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { // Precompute the conservative decision once per frequency // bucket so the scalar and SIMD merge kernels stay // floating-point-free. - let freq_cannot_beat = if window_started_with_floor && num_lists >= 2 { + 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"); @@ -4360,21 +4498,37 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { &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"), + let others_block_max = + others_block_max.expect("positive floor should initialize bounds"); + let key = (self.threshold.to_bits(), others_block_max.to_bits()); + if first_score_limit_key != Some(key) { + first_score_limit = exclusive_partial_score_limit( + others_block_max, self.threshold, score_sum_upper_bound_factor(num_lists), - CompetitiveFloorMode::Exclusive, - ) - }) - } else { - [false; FREQ_LUT_BUCKETS] - }; + ); + first_score_limit_key = Some(key); + } + if freq_cannot_beat_key != Some(key) { + let freq_bound_lut = freq_bound_lut + .as_ref() + .expect("positive threshold should initialize the frequency bound LUT"); + freq_cannot_beat = match first_score_limit { + Some(limit) => { + std::array::from_fn(|frequency| freq_bound_lut[frequency] <= limit) + } + None => [false; FREQ_LUT_BUCKETS], + }; + freq_cannot_beat_key = Some(key); + } + } else if freq_cannot_beat_key.is_some() { + freq_cannot_beat = [false; FREQ_LUT_BUCKETS]; + freq_cannot_beat_key = None; + } + #[cfg(test)] + if freq_cannot_beat.iter().any(|&pruned| pruned) { + self.and_window_stats.windows_frequency_pruned += 1; + } #[cfg(target_arch = "x86_64")] let use_avx2 = *HAS_AVX2; #[cfg(not(target_arch = "x86_64"))] @@ -4437,13 +4591,25 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { (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, - ), + #[cfg(target_arch = "x86_64")] + (_, true, true) => unsafe { + merge_window_n_avx2( + &wins, + &freq_cannot_beat, + &mut cursor_scratch, + &mut batch_docs, + &mut batch_offs, + ) + }, + (_, _, true) => unsafe { + merge_window_n( + &wins, + &freq_cannot_beat, + &mut cursor_scratch, + &mut batch_docs, + &mut batch_offs, + ) + }, } if !batch_docs.is_empty() { @@ -4522,13 +4688,16 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { } 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, - ) { + let key = (self.threshold.to_bits(), others_block_max.to_bits()); + if first_score_limit_key != Some(key) { + first_score_limit = exclusive_partial_score_limit( + others_block_max, + self.threshold, + score_sum_upper_bound_factor(num_lists), + ); + first_score_limit_key = Some(key); + } + if first_score_limit.is_some_and(|limit| first_score <= limit) { continue; } } @@ -4621,7 +4790,7 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { if win_end == TERMINATED_DOC_ID { break; } - target = win_end + 1; + target = (win_end + 1).max(jump_to); } metrics.record_comparisons(num_comparisons); @@ -5891,6 +6060,100 @@ fn next_up_f64(value: f64) -> f64 { #[cfg(test)] mod tests { + + /// `exclusive_partial_score_limit` must agree with `score_sum_cannot_compete` + /// for every f32 partial score: scores at or below the limit are rejected, + /// scores above it compete. Sweeps ULP neighbourhoods of the boundary plus + /// coarse samples, over floors, remaining bounds and clause counts. + #[test] + fn exclusive_partial_score_limit_matches_predicate() { + let mut seed = 0x9E37_79B9_7F4A_7C15_u64; + let mut next = || { + seed ^= seed << 13; + seed ^= seed >> 7; + seed ^= seed << 17; + seed + }; + let mut checked = 0usize; + for case in 0..2_000 { + let floor = if case % 7 == 0 { + f32::from_bits((next() as u32) & 0x7F7F_FFFF) + } else { + (next() % 10_000) as f32 / 37.0 + 1e-6 + }; + let remaining = if case % 5 == 0 { + f64::from(f32::from_bits((next() as u32) & 0x7F7F_FFFF)) + } else { + (next() % 10_000) as f64 / 41.0 + }; + let num_lists = 2 + (next() % 6) as usize; + let factor = score_sum_upper_bound_factor(num_lists); + let rejects = |score: f32| { + score_sum_cannot_compete( + score, + remaining, + floor, + factor, + CompetitiveFloorMode::Exclusive, + ) + }; + let limit = exclusive_partial_score_limit(remaining, floor, factor); + let mut probes = vec![ + 0.0_f32, + f32::MIN_POSITIVE, + floor, + f32::MAX, + f32::MIN, + ((f64::from(floor) / factor) - remaining) as f32, + ]; + if let Some(limit) = limit { + let mut up = limit; + let mut down = limit; + for _ in 0..4 { + probes.push(up); + probes.push(down); + up = next_up_f32(up); + down = next_down_f32(down); + } + } + for _ in 0..16 { + probes.push(f32::from_bits((next() as u32) & 0x7F7F_FFFF)); + } + for score in probes { + let expected = rejects(score); + let actual = limit.is_some_and(|limit| score <= limit); + assert_eq!( + expected, actual, + "score={score:?} limit={limit:?} floor={floor:?} remaining={remaining:?} num_lists={num_lists}" + ); + checked += 1; + } + } + assert!(checked > 50_000); + } + + #[test] + fn exclusive_partial_score_limit_handles_non_finite_inputs() { + let factor = score_sum_upper_bound_factor(3); + assert_eq!( + exclusive_partial_score_limit(f64::INFINITY, 1.0, factor), + None + ); + assert_eq!(exclusive_partial_score_limit(1.0, f32::NAN, factor), None); + assert_eq!( + exclusive_partial_score_limit(1.0, f32::INFINITY, factor), + None + ); + // A floor no partial score can fail to beat rejects nothing. + let limit = exclusive_partial_score_limit(1.0e30, 1.0, factor); + assert!(limit.is_none_or(|limit| !score_sum_cannot_compete( + next_up_f32(limit), + 1.0e30, + 1.0, + factor, + CompetitiveFloorMode::Exclusive + ))); + } use arrow::buffer::ScalarBuffer; use rstest::rstest; @@ -10084,6 +10347,81 @@ mod tests { ); } + /// A sparse lead over dense followers must drive the window traversal: + /// after an empty window the next one starts at the lead's next document, + /// so the number of sliced windows tracks the lead's documents rather than + /// the followers' block count. Results stay identical to the classic loop. + #[rstest] + fn bulk_and_windows_jump_to_next_lead_document( + #[values(2, 4)] num_clauses: usize, + #[values(0.0, 0.05)] initial_floor: f32, + ) { + let num_docs = MAX_POSTING_BLOCK_SIZE * 64; + let lead_stride = 2_000; + let docs = CostOnlyDocuments { + total_docs: num_docs, + visible_cost_upper_bound: num_docs, + }; + let run = |mode| { + let postings = (0..num_clauses) + .map(|term| { + let doc_ids = if term == 0 { + (0..num_docs as u32) + .step_by(lead_stride) + .collect::>() + } else { + (0..num_docs as u32) + .filter(|doc| doc % 3 != term as u32) + .collect() + }; + 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, + (0..len).map(|index| 1 + (index % 5) as u32).collect(), + vec![1; len], + MAX_POSTING_BLOCK_SIZE, + ), + docs.len(), + ) + }) + .collect::>(); + 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 mut rows = wand + .search( + &FtsSearchParams::default().with_limit(Some(10)), + &NoOpMetricsCollector, + ) + .unwrap() + .into_iter() + .map(|hit| (hit.document, hit.doc_length, hit.freqs)) + .collect::>(); + rows.sort_unstable(); + ( + rows, + shared_floor.load(Ordering::Relaxed), + wand.and_window_stats.windows_sliced, + ) + }; + let (classic, classic_floor, _) = run(BulkAndMode::Off); + let (bulk, bulk_floor, windows) = run(BulkAndMode::On); + assert_eq!(bulk, classic); + assert_eq!(bulk_floor, classic_floor); + let lead_docs = num_docs / lead_stride + 1; + let follower_blocks = (num_docs * 2 / 3) / MAX_POSTING_BLOCK_SIZE; + assert!( + windows <= 2 * lead_docs + 2, + "sliced {windows} windows for {lead_docs} lead docs; followers have {follower_blocks} blocks each" + ); + } + #[rstest] #[case::dense("dense")] #[case::sparse("sparse")] @@ -10483,4 +10821,85 @@ mod tests { ); assert!(scored.load(Ordering::Relaxed) > 0); } + + /// The floor turns positive in the middle of the first window, where only + /// the per-candidate prune refreshes the score limit. The second window + /// has the same floor and the same follower block max, so the frequency + /// prune table must still be built for it instead of staying disabled. + #[test] + fn bulk_and_builds_frequency_prune_table_after_mid_window_floor() { + let num_docs = (MAX_POSTING_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 run = |mode| { + // One block spanning both windows; only document 0 can win. + let sparse_docs = (0..num_docs).step_by(2).collect::>(); + let sparse_freqs = sparse_docs + .iter() + .map(|&doc| if doc == 0 { 10 } else { 1 }) + .collect::>(); + // Two blocks with the same block max, so the follower bound is + // identical in both windows. Document 257 keeps the second window + // competitive by block max without matching the sparse clause. + let dense_docs = (0..num_docs).collect::>(); + let dense_freqs = dense_docs + .iter() + .map(|&doc| if doc == 0 || doc == 257 { 10 } else { 1 }) + .collect::>(); + let postings = [(sparse_docs, sparse_freqs), (dense_docs, dense_freqs)] + .into_iter() + .enumerate() + .map(|(term, (doc_ids, freqs))| { + 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, + freqs, + vec![1; len], + MAX_POSTING_BLOCK_SIZE, + ), + docs.len(), + ) + }) + .collect::>(); + 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(1)), + &NoOpMetricsCollector, + ) + .unwrap() + .into_iter() + .map(|hit| (hit.document, hit.doc_length, hit.freqs)) + .collect::>(); + ( + hits, + wand.threshold, + wand.and_window_stats.windows_sliced, + wand.and_window_stats.windows_frequency_pruned, + ) + }; + let (classic, classic_floor, _, _) = run(BulkAndMode::Off); + let (bulk, bulk_floor, windows_sliced, windows_frequency_pruned) = run(BulkAndMode::On); + assert_eq!(bulk, classic); + assert_eq!(bulk.len(), 1); + assert_eq!(bulk[0].0, 0); + assert_eq!(bulk_floor, 20.0); + assert_eq!(bulk_floor, classic_floor); + assert_eq!( + windows_sliced, 2, + "the dense clause's block boundary splits two windows" + ); + assert_eq!( + windows_frequency_pruned, 1, + "the second window must apply the frequency prune table" + ); + } }