From 59cb76d12da99d3e27cb205269fe91e7a027dd37 Mon Sep 17 00:00:00 2001 From: Yang Cen Date: Thu, 17 Sep 2026 14:10:27 +0000 Subject: [PATCH 1/5] perf(fts): derive bulk AND prune decisions from a per-window score limit The bulk conjunction path evaluated the exact `score_sum_cannot_compete` bound 64 times per window to fill the frequency prune table and once more per surviving candidate. Both decisions are monotone in the first clause's partial score, so compute the largest rejected f32 score once per (floor, followers' block max) pair and reduce every later decision to a single f32 comparison. Decisions are bit-identical to the exact bound. Co-Authored-By: Claude Fable 5.1 --- rust/lance-index/src/scalar/inverted/wand.rs | 214 +++++++++++++++++-- 1 file changed, 200 insertions(+), 14 deletions(-) diff --git a/rust/lance-index/src/scalar/inverted/wand.rs b/rust/lance-index/src/scalar/inverted/wand.rs index 6a3b575bebe..d03414fad7e 100644 --- a/rust/lance-index/src/scalar/inverted/wand.rs +++ b/rust/lance-index/src/scalar/inverted/wand.rs @@ -1801,6 +1801,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 @@ -4237,6 +4313,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; @@ -4363,15 +4444,23 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { 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, - ) - }) + ); + first_score_limit_key = Some(key); + } + match first_score_limit { + Some(limit) => { + std::array::from_fn(|frequency| freq_bound_lut[frequency] <= limit) + } + None => [false; FREQ_LUT_BUCKETS], + } } else { [false; FREQ_LUT_BUCKETS] }; @@ -4522,13 +4611,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; } } @@ -5891,6 +5983,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; From 944f4f015a5a054ea2b79b6112e7af49c5c5b5f4 Mon Sep 17 00:00:00 2001 From: Yang Cen Date: Thu, 17 Sep 2026 14:10:33 +0000 Subject: [PATCH 2/5] perf(fts): carry decoded frequencies through the lazy doc-id advance Once a block's frequency stream has been decoded, `next_doc_id` can fill the candidate's frequency directly instead of leaving `doc()` to re-materialize it for every candidate. Single-clause conjunctions visit every candidate of a block they decode, so they paid that cost on each one. Co-Authored-By: Claude Fable 5.1 --- rust/lance-index/src/scalar/inverted/wand.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/rust/lance-index/src/scalar/inverted/wand.rs b/rust/lance-index/src/scalar/inverted/wand.rs index d03414fad7e..ec38534a534 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; } From 2e96204dc0bfcd2af580a05a4bb20034f4ce85cf Mon Sep 17 00:00:00 2001 From: Yang Cen Date: Tue, 15 Sep 2026 14:08:32 +0800 Subject: [PATCH 3/5] perf(fts): jump bulk conjunction windows to the next lead document When a clause has no document in the current window, start the next window at that clause's next document instead of the next block boundary. A sparse lead then drives the traversal and dense followers skip whole blocks by metadata rather than being sliced block by block. Co-Authored-By: Claude Fable 5.1 --- rust/lance-index/src/scalar/inverted/wand.rs | 96 +++++++++++++++++++- 1 file changed, 92 insertions(+), 4 deletions(-) diff --git a/rust/lance-index/src/scalar/inverted/wand.rs b/rust/lance-index/src/scalar/inverted/wand.rs index ec38534a534..b86bb221da9 100644 --- a/rust/lance-index/src/scalar/inverted/wand.rs +++ b/rust/lance-index/src/scalar/inverted/wand.rs @@ -2320,6 +2320,7 @@ struct AndWindowStats { candidates_returned: usize, score_first_rejections: usize, pairwise_intersections: usize, + windows_sliced: usize, } impl Eq for TailPosting {} @@ -4337,6 +4338,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 { @@ -4384,6 +4389,12 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { wins.clear(); let mut skip_window = false; let mut exhausted = false; + // 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 posting in &self.lead { let PostingList::Compressed(ref list) = posting.list else { unreachable!("bulk AND requires compressed postings"); @@ -4401,9 +4412,11 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { // 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; @@ -4721,7 +4734,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); @@ -10278,6 +10291,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")] From 3e73de9d362ed58267755b06d0ae170b8c282669 Mon Sep 17 00:00:00 2001 From: Yang Cen Date: Tue, 15 Sep 2026 15:09:34 +0800 Subject: [PATCH 4/5] perf(fts): trim per-window overhead in bulk conjunctions Reuse the previous slice end when a clause stays in the same block, skip the end search when the block ends inside the window, rebuild the frequency prune table only when the floor or the followers' block maxes change, run the generic cursor merge as an AVX2-specialized kernel, and stop preparing score-first windows the bulk path never reads. Co-Authored-By: Claude Fable 5.1 --- rust/lance-index/src/scalar/inverted/wand.rs | 181 ++++++++++++------- 1 file changed, 113 insertions(+), 68 deletions(-) diff --git a/rust/lance-index/src/scalar/inverted/wand.rs b/rust/lance-index/src/scalar/inverted/wand.rs index b86bb221da9..a8157227574 100644 --- a/rust/lance-index/src/scalar/inverted/wand.rs +++ b/rust/lance-index/src/scalar/inverted/wand.rs @@ -3910,6 +3910,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]; @@ -3973,7 +3977,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], @@ -4055,6 +4059,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], @@ -4126,6 +4174,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 @@ -4134,6 +4183,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, @@ -4149,48 +4199,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], @@ -4306,6 +4314,14 @@ 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]; + // 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; @@ -4395,18 +4411,33 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { // 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 posting in &self.lead { + 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 { - lo + state.doc_ids[lo..].partition_point(|&doc| doc <= win_end32) + 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 + 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 @@ -4453,7 +4484,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"); @@ -4462,9 +4493,6 @@ 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"); let others_block_max = others_block_max.expect("positive floor should initialize bounds"); let key = (self.threshold.to_bits(), others_block_max.to_bits()); @@ -4475,16 +4503,21 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { score_sum_upper_bound_factor(num_lists), ); first_score_limit_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], + }; } - match first_score_limit { - Some(limit) => { - std::array::from_fn(|frequency| freq_bound_lut[frequency] <= limit) - } - None => [false; FREQ_LUT_BUCKETS], - } - } else { - [false; FREQ_LUT_BUCKETS] - }; + } else if first_score_limit_key.is_some() { + freq_cannot_beat = [false; FREQ_LUT_BUCKETS]; + first_score_limit = None; + first_score_limit_key = None; + } #[cfg(target_arch = "x86_64")] let use_avx2 = *HAS_AVX2; #[cfg(not(target_arch = "x86_64"))] @@ -4547,13 +4580,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() { From 9834e75527d749b742c5098e16034a68ddfe24c9 Mon Sep 17 00:00:00 2001 From: Yang Cen Date: Fri, 18 Sep 2026 07:57:53 +0000 Subject: [PATCH 5/5] perf(fts): rebuild the bulk AND prune table after a mid-window floor The per-candidate prune refreshes the cached score limit when the floor turns positive in the middle of a window, but it never touched the frequency prune table. A following window with the same floor and the same follower block max then matched the cached key and kept the all-false table, so kernel-level frequency pruning stayed off until the key changed. Results were unaffected; only the pruning was lost. Key the table separately from the limit so each is rebuilt when stale. Co-Authored-By: Claude Fable 5.1 --- rust/lance-index/src/scalar/inverted/wand.rs | 98 +++++++++++++++++++- 1 file changed, 95 insertions(+), 3 deletions(-) diff --git a/rust/lance-index/src/scalar/inverted/wand.rs b/rust/lance-index/src/scalar/inverted/wand.rs index a8157227574..196b1e5d67f 100644 --- a/rust/lance-index/src/scalar/inverted/wand.rs +++ b/rust/lance-index/src/scalar/inverted/wand.rs @@ -2321,6 +2321,7 @@ struct AndWindowStats { score_first_rejections: usize, pairwise_intersections: usize, windows_sliced: usize, + windows_frequency_pruned: usize, } impl Eq for TailPosting {} @@ -4317,6 +4318,10 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { // 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. @@ -4503,6 +4508,8 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { score_sum_upper_bound_factor(num_lists), ); 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"); @@ -4512,11 +4519,15 @@ impl<'a, S: Scorer, D: WandDocuments> Wand<'a, S, D> { } None => [false; FREQ_LUT_BUCKETS], }; + freq_cannot_beat_key = Some(key); } - } else if first_score_limit_key.is_some() { + } else if freq_cannot_beat_key.is_some() { freq_cannot_beat = [false; FREQ_LUT_BUCKETS]; - first_score_limit = None; - first_score_limit_key = None; + 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; @@ -10810,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" + ); + } }