From 85af02b098f5f28a108983220efdab858835dee4 Mon Sep 17 00:00:00 2001 From: Yang Cen Date: Thu, 20 Aug 2026 17:43:42 +0800 Subject: [PATCH] refactor(fts): add exact posting load policies --- .../src/scalar/inverted/index/partition.rs | 242 +++++++++++++--- .../scalar/inverted/index/posting_reader.rs | 127 +++++++-- .../src/scalar/inverted/index/tests/stats.rs | 259 ++++++++++++++++++ 3 files changed, 567 insertions(+), 61 deletions(-) diff --git a/rust/lance-index/src/scalar/inverted/index/partition.rs b/rust/lance-index/src/scalar/inverted/index/partition.rs index 59169efd394..902d809330b 100644 --- a/rust/lance-index/src/scalar/inverted/index/partition.rs +++ b/rust/lance-index/src/scalar/inverted/index/partition.rs @@ -2,6 +2,105 @@ // SPDX-FileCopyrightText: Copyright The Lance Authors use super::*; +use smallvec::SmallVec; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct PositionMatchSummary { + exact_scoring_required: bool, + every_position_matched: bool, +} + +#[derive(Debug, Clone, Copy)] +pub(in super::super) struct PostingLoadOptions { + force_global_scorer: bool, + read_policy: PostingReadPolicy, +} + +impl PostingLoadOptions { + const fn read_ahead(force_global_scorer: bool) -> Self { + Self { + force_global_scorer, + read_policy: PostingReadPolicy::ReadAhead, + } + } + + #[cfg(test)] + pub(in super::super) const fn cache_aware_exact(force_global_scorer: bool) -> Self { + Self { + force_global_scorer, + read_policy: PostingReadPolicy::CacheAwareExact, + } + } +} + +fn summarize_position_matches(mut positions: SmallVec<[(u32, bool); 8]>) -> PositionMatchSummary { + positions.sort_unstable_by_key(|(position, _)| *position); + + let mut exact_scoring_required = false; + let mut every_position_matched = true; + let mut group_start = 0; + while group_start < positions.len() { + let position = positions[group_start].0; + let mut group_end = group_start + 1; + let mut group_matched = positions[group_start].1; + while group_end < positions.len() && positions[group_end].0 == position { + exact_scoring_required = true; + group_matched |= positions[group_end].1; + group_end += 1; + } + every_position_matched &= group_matched; + group_start = group_end; + } + + PositionMatchSummary { + exact_scoring_required, + every_position_matched, + } +} + +fn posting_group_demand_counts( + inverted_list: &PostingListReader, + token_ids: &[(u32, String, u32)], +) -> HashMap<(u32, u32), usize> { + let mut demanded_token_ids = token_ids + .iter() + .map(|(token_id, _, _)| *token_id) + .collect::>(); + demanded_token_ids.sort_unstable(); + demanded_token_ids.dedup(); + + let mut counts = HashMap::new(); + for token_id in demanded_token_ids { + if let Some(group) = inverted_list.group_range_for_token(token_id) { + *counts.entry(group).or_default() += 1; + } + } + counts +} + +fn effective_posting_read_policy( + inverted_list: &PostingListReader, + requested_policy: PostingReadPolicy, + group_demand_counts: &HashMap<(u32, u32), usize>, + token_id: u32, +) -> PostingReadPolicy { + if requested_policy == PostingReadPolicy::ReadAhead { + return PostingReadPolicy::ReadAhead; + } + let Some(group) = inverted_list.group_range_for_token(token_id) else { + return PostingReadPolicy::CacheAwareExact; + }; + let demand_count = group_demand_counts.get(&group).copied(); + debug_assert!( + demand_count.is_some(), + "posting group {group:?} must have a demand count for token {token_id}" + ); + if demand_count == Some(1) { + PostingReadPolicy::CacheAwareExact + } else { + PostingReadPolicy::ReadAhead + } +} /// Query-level inputs for a grouped-term score upper bound. /// @@ -510,7 +609,6 @@ impl InvertedPartition { // bounds must share corpus-level statistics before the global collector // can safely propagate its threshold. Old posting formats without impacts // fall back to a scorer-derived global upper bound in that mode. - #[instrument(level = "debug", skip_all)] pub(in super::super) async fn load_posting_lists( &self, tokens: &Tokens, @@ -520,59 +618,96 @@ impl InvertedPartition { metrics: &dyn MetricsCollector, force_global_scorer: bool, ) -> Result { + self.load_posting_lists_with_policy( + tokens, + params, + operator, + impact_scorer, + metrics, + PostingLoadOptions::read_ahead(force_global_scorer), + ) + .await + } + + #[instrument(name = "load_posting_lists", level = "debug", skip_all)] + pub(in super::super) async fn load_posting_lists_with_policy( + &self, + tokens: &Tokens, + params: &FtsSearchParams, + operator: Operator, + impact_scorer: &MemBM25Scorer, + metrics: &dyn MetricsCollector, + options: PostingLoadOptions, + ) -> Result { + let PostingLoadOptions { + force_global_scorer, + read_policy: requested_read_policy, + } = options; let is_phrase_query = params.phrase_slop.is_some(); let is_and_query = operator == Operator::And; - let required_positions = (is_and_query || is_phrase_query).then(|| { - (0..tokens.len()) - .map(|index| tokens.position(index)) - .collect::>() - }); // Fuzzy expansion already ran once at the index level (see // `InvertedIndex::bm25_search`) under the global `max_expansions` // budget. Positions identify alternatives that must share one posting // iterator, including code identifier subwords and fuzzy expansions. - let tokens = tokens.clone(); - let token_positions = (0..tokens.len()) - .map(|index| tokens.position(index)) - .collect::>(); - let mut seen_positions = HashSet::with_capacity(token_positions.len()); - let exact_scoring_required = token_positions - .iter() - .any(|position| !seen_positions.insert(*position)); let mut token_ids = Vec::with_capacity(tokens.len()); - let mut matched_positions = required_positions.as_ref().map(|_| HashSet::new()); - for (index, token) in tokens.into_iter().enumerate() { - let token_id = self.map(&token); + let mut position_matches = SmallVec::<[(u32, bool); 8]>::new(); + for index in 0..tokens.len() { + let token = tokens.get_token(index); + let position = tokens.position(index); + let token_id = self.map(token); + position_matches.push((position, token_id.is_some())); if let Some(token_id) = token_id { - let position = token_positions[index]; - if let Some(matched_positions) = matched_positions.as_mut() { - matched_positions.insert(position); - } - token_ids.push((token_id, token, position)); + token_ids.push((token_id, token.to_owned(), position)); } } + let position_summary = summarize_position_matches(position_matches); + let exact_scoring_required = position_summary.exact_scoring_required; if token_ids.is_empty() { return Ok(LoadedPostings::empty()); } - if let Some(required_positions) = required_positions.as_ref() - && let Some(matched_positions) = matched_positions.as_ref() - && !required_positions.is_subset(matched_positions) - { + if (is_and_query || is_phrase_query) && !position_summary.every_position_matched { return Ok(LoadedPostings::empty()); } token_ids.sort_unstable_by_key(|(token_id, _, position)| (*position, *token_id)); token_ids.dedup_by(|lhs, rhs| lhs.0 == rhs.0 && lhs.2 == rhs.2); + let group_demand_counts = if requested_read_policy == PostingReadPolicy::CacheAwareExact { + posting_group_demand_counts(self.inverted_list.as_ref(), &token_ids) + } else { + HashMap::new() + }; + let num_docs = self.docs.len(); let loaded_postings = stream::iter(token_ids) - .map(|(token_id, token, position)| async move { - let posting = self - .inverted_list - .posting_list(token_id, is_phrase_query, metrics) - .await?; + .map(|(token_id, token, position)| { + let read_policy = effective_posting_read_policy( + self.inverted_list.as_ref(), + requested_read_policy, + &group_demand_counts, + token_id, + ); + async move { + let posting = match read_policy { + PostingReadPolicy::ReadAhead => { + self.inverted_list + .posting_list(token_id, is_phrase_query, metrics) + .await? + } + PostingReadPolicy::CacheAwareExact => { + self.inverted_list + .posting_list_with_policy( + token_id, + is_phrase_query, + metrics, + read_policy, + ) + .await? + } + }; - Result::Ok((token_id, token, position, posting)) + Result::Ok((token_id, token, position, posting)) + } }) .buffered(self.store.io_parallelism()) .try_collect::>() @@ -857,6 +992,49 @@ mod tests { use super::*; + fn position_summary(entries: &[(u32, bool)]) -> PositionMatchSummary { + summarize_position_matches(entries.iter().copied().collect()) + } + + #[test] + fn position_summary_marks_or_duplicates_for_exact_scoring() { + let summary = position_summary(&[(0, true), (0, false), (1, false)]); + + assert!(summary.exact_scoring_required); + assert!(!summary.every_position_matched); + } + + #[test] + fn position_summary_requires_a_match_in_every_and_group() { + let complete = position_summary(&[(0, false), (0, true), (1, true)]); + let incomplete = position_summary(&[(0, true), (1, false), (1, false)]); + + assert!(complete.exact_scoring_required); + assert!(complete.every_position_matched); + assert!(incomplete.exact_scoring_required); + assert!(!incomplete.every_position_matched); + } + + #[test] + fn position_summary_groups_nonadjacent_positions() { + let summary = position_summary(&[(9, false), (1, true), (4, true), (9, true)]); + + assert!(summary.exact_scoring_required); + assert!(summary.every_position_matched); + } + + #[test] + fn position_summary_spills_past_eight_tokens_without_losing_exactness() { + let mut positions = SmallVec::<[(u32, bool); 8]>::new(); + positions.extend((0..10).map(|position| (position, true))); + positions.push((3, false)); + assert!(positions.spilled()); + + let summary = summarize_position_matches(positions); + assert!(summary.exact_scoring_required); + assert!(summary.every_position_matched); + } + #[rstest] #[case::plain(false)] #[case::v3_compressed(true)] diff --git a/rust/lance-index/src/scalar/inverted/index/posting_reader.rs b/rust/lance-index/src/scalar/inverted/index/posting_reader.rs index bc7202db682..190e8df64c7 100644 --- a/rust/lance-index/src/scalar/inverted/index/posting_reader.rs +++ b/rust/lance-index/src/scalar/inverted/index/posting_reader.rs @@ -67,6 +67,14 @@ pub(super) enum PositionsLayout { SharedStream(PositionStreamCodec), } +/// Selects whether a posting lookup may read neighboring token rows on a +/// cache miss. Exact reads still reuse an already-resident read-ahead group. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(in super::super) enum PostingReadPolicy { + ReadAhead, + CacheAwareExact, +} + impl std::fmt::Debug for PostingListReader { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let mut s = f.debug_struct("InvertedListReader"); @@ -401,47 +409,59 @@ impl PostingListReader { Ok(batch) } - #[instrument(level = "debug", skip(self, metrics))] pub(crate) async fn posting_list( &self, token_id: u32, is_phrase_query: bool, metrics: &dyn MetricsCollector, + ) -> Result { + self.posting_list_with_policy( + token_id, + is_phrase_query, + metrics, + PostingReadPolicy::ReadAhead, + ) + .await + } + + #[instrument(name = "posting_list", level = "debug", skip(self, metrics))] + pub(in super::super) async fn posting_list_with_policy( + &self, + token_id: u32, + is_phrase_query: bool, + metrics: &dyn MetricsCollector, + read_policy: PostingReadPolicy, ) -> Result { let mut posting = match self.group_range_for_token(token_id) { // Grouped path (issue #7040): one cache entry covers rows // [start, end), so neighbouring rare terms share a single read. Some((start, end)) => { - let result = self - .index_cache - .get_or_insert_with_key_hit( - posting_list_group_cache_key(start, end, self.has_impacts), - || async move { - metrics.record_part_load(); - info!(target: TRACE_IO_EVENTS, r#type=IO_TYPE_LOAD_SCALAR_PART, index_type="inverted", part_id=start); - self.load_posting_list_group(start, end).await - }, - ) - .await; - match &result { - Ok((_, true)) => metrics.record_index_cache_hit(), - _ => metrics.record_index_cache_miss(), - } - let (group, _) = result?; - let (max_score, length) = if group.needs_external_metadata() { - self.posting_metadata_for_token(token_id, Some(metrics)) + let exact_end = token_id.checked_add(1).ok_or_else(|| { + Error::index(format!( + "posting token id {token_id} cannot form an exclusive singleton range" + )) + })?; + if read_policy == PostingReadPolicy::CacheAwareExact + && (start != token_id || end != exact_end) + && let Some(group) = self + .index_cache + .get_with_key(&posting_list_group_cache_key(start, end, self.has_impacts)) + .await + { + // This cache-only probe never invokes the posting loader. + // Report the group hit because it is the path that serves + // the posting; a probe miss is not a query-cache miss. + metrics.record_index_cache_hit(); + self.posting_from_group(token_id, start, end, group.as_ref(), metrics) .await? } else { - (None, None) - }; - let slot = (token_id - start) as usize; - group - .posting_list(slot, max_score, length)? - .ok_or_else(|| { - Error::index(format!( - "token {token_id} maps to slot {slot} outside posting group [{start}, {end})" - )) - })? + let (selected_start, selected_end) = match read_policy { + PostingReadPolicy::ReadAhead => (start, end), + PostingReadPolicy::CacheAwareExact => (token_id, exact_end), + }; + self.load_cached_posting_group(token_id, selected_start, selected_end, metrics) + .await? + } } // Fallback for layouts that cannot use row-based groups: one cache // entry per token. @@ -487,6 +507,55 @@ impl PostingListReader { Ok(posting) } + async fn load_cached_posting_group( + &self, + token_id: u32, + start: u32, + end: u32, + metrics: &dyn MetricsCollector, + ) -> Result { + let result = self + .index_cache + .get_or_insert_with_key_hit( + posting_list_group_cache_key(start, end, self.has_impacts), + || async move { + metrics.record_part_load(); + info!(target: TRACE_IO_EVENTS, r#type=IO_TYPE_LOAD_SCALAR_PART, index_type="inverted", part_id=start); + self.load_posting_list_group(start, end).await + }, + ) + .await; + match &result { + Ok((_, true)) => metrics.record_index_cache_hit(), + _ => metrics.record_index_cache_miss(), + } + let (group, _) = result?; + self.posting_from_group(token_id, start, end, group.as_ref(), metrics) + .await + } + + async fn posting_from_group( + &self, + token_id: u32, + start: u32, + end: u32, + group: &PostingListGroup, + metrics: &dyn MetricsCollector, + ) -> Result { + let (max_score, length) = if group.needs_external_metadata() { + self.posting_metadata_for_token(token_id, Some(metrics)) + .await? + } else { + (None, None) + }; + let slot = (token_id - start) as usize; + group.posting_list(slot, max_score, length)?.ok_or_else(|| { + Error::index(format!( + "token {token_id} maps to slot {slot} outside posting group [{start}, {end})" + )) + }) + } + pub(super) async fn ensure_modern_posting_validated( &self, token_id: u32, diff --git a/rust/lance-index/src/scalar/inverted/index/tests/stats.rs b/rust/lance-index/src/scalar/inverted/index/tests/stats.rs index 55969d6f6ec..6bdd31919e6 100644 --- a/rust/lance-index/src/scalar/inverted/index/tests/stats.rs +++ b/rust/lance-index/src/scalar/inverted/index/tests/stats.rs @@ -10,6 +10,8 @@ use super::*; struct PostingMetadataCounter { rows_read: std::sync::atomic::AtomicUsize, metadata_rows_read: std::sync::atomic::AtomicUsize, + posting_rows_read: std::sync::atomic::AtomicUsize, + impact_rows_read: std::sync::atomic::AtomicUsize, read_range_calls: std::sync::atomic::AtomicUsize, } @@ -21,6 +23,14 @@ impl PostingMetadataCounter { self.metadata_rows_read .load(std::sync::atomic::Ordering::Relaxed) } + fn posting_rows_read(&self) -> usize { + self.posting_rows_read + .load(std::sync::atomic::Ordering::Relaxed) + } + fn impact_rows_read(&self) -> usize { + self.impact_rows_read + .load(std::sync::atomic::Ordering::Relaxed) + } fn read_range_calls(&self) -> usize { self.read_range_calls .load(std::sync::atomic::Ordering::Relaxed) @@ -60,6 +70,22 @@ impl IndexReader for CountingPostingReader { .metadata_rows_read .fetch_add(n, std::sync::atomic::Ordering::Relaxed); } + let touches_posting = projection + .map(|columns| columns.contains(&POSTING_COL)) + .unwrap_or(false); + if touches_posting { + self.counter + .posting_rows_read + .fetch_add(n, std::sync::atomic::Ordering::Relaxed); + } + let touches_impacts = projection + .map(|columns| columns.contains(&IMPACT_COL)) + .unwrap_or(false); + if touches_impacts { + self.counter + .impact_rows_read + .fetch_add(n, std::sync::atomic::Ordering::Relaxed); + } self.inner.read_range(range, projection).await } async fn num_batches(&self, batch_size: u64) -> u32 { @@ -215,6 +241,15 @@ async fn load_counted_v2_index( (index, counter, tmpdir) } +fn set_test_posting_group_size(index: &mut Arc, group_size: u32) { + let index = Arc::get_mut(index).expect("test index should have one owner"); + let partition = + Arc::get_mut(&mut index.partitions[0]).expect("test partition should have one owner"); + let inverted_list = Arc::get_mut(&mut partition.inverted_list) + .expect("test posting reader should have one owner"); + inverted_list.grouping = PostingGrouping::SyntheticFixed { group_size }; +} + /// IO regression test for the lazy posting-metadata refactor. Builds a /// v2 InvertedIndex with `num_tokens` tokens in a single partition, /// wraps the IndexStore so reads against the posting file are counted, @@ -488,6 +523,230 @@ async fn test_grouped_posting_lists_read_one_group_per_neighborhood() { ); } +#[tokio::test] +async fn test_cache_aware_exact_cold_read_uses_one_singleton_group() { + let cache = LanceCache::with_capacity(1024 * 1024); + let (mut index, counter, _tmpdir) = load_counted_v2_index(8, cache.clone()).await; + set_test_posting_group_size(&mut index, 4); + let inverted_list = index.partitions[0].inverted_list.clone(); + let metrics = LocalMetricsCollector::default(); + + let posting = inverted_list + .posting_list_with_policy(0, false, &metrics, PostingReadPolicy::CacheAwareExact) + .await + .unwrap(); + + assert_eq!(posting.len(), 1); + assert_eq!(counter.read_range_calls(), 1); + assert_eq!(counter.rows_read(), 1); + assert_eq!(counter.metadata_rows_read(), 1); + assert_eq!(counter.posting_rows_read(), 1); + assert_eq!(counter.impact_rows_read(), 1); + assert_eq!(metrics.index_cache_hits(), 0); + assert_eq!(metrics.index_cache_misses(), 1); + assert_eq!(metrics.parts_loaded.load(Ordering::Relaxed), 1); +} + +#[tokio::test] +async fn test_cache_aware_exact_singleton_is_singleflight_and_cached() { + let cache = LanceCache::with_capacity(1024 * 1024); + let (mut index, counter, _tmpdir) = load_counted_v2_index(8, cache.clone()).await; + set_test_posting_group_size(&mut index, 4); + let inverted_list = index.partitions[0].inverted_list.clone(); + let metrics = Arc::new(LocalMetricsCollector::default()); + + let postings = futures::future::join_all((0..8).map(|_| { + let inverted_list = inverted_list.clone(); + let metrics = metrics.clone(); + async move { + inverted_list + .posting_list_with_policy( + 0, + false, + metrics.as_ref(), + PostingReadPolicy::CacheAwareExact, + ) + .await + } + })) + .await + .into_iter() + .collect::>>() + .unwrap(); + + assert!(postings.iter().all(|posting| posting.len() == 1)); + assert_eq!(counter.read_range_calls(), 1); + assert_eq!(counter.rows_read(), 1); + assert_eq!(metrics.index_cache_misses(), 1); + assert_eq!(metrics.index_cache_hits(), 7); + assert_eq!(metrics.parts_loaded.load(Ordering::Relaxed), 1); + + let warm_metrics = LocalMetricsCollector::default(); + inverted_list + .posting_list_with_policy(0, false, &warm_metrics, PostingReadPolicy::CacheAwareExact) + .await + .unwrap(); + assert_eq!(counter.read_range_calls(), 1); + assert_eq!(warm_metrics.index_cache_hits(), 1); + assert_eq!(warm_metrics.index_cache_misses(), 0); + assert_eq!(warm_metrics.parts_loaded.load(Ordering::Relaxed), 0); +} + +#[tokio::test] +async fn test_cache_aware_exact_reuses_prewarmed_read_ahead_group() { + let cache = LanceCache::with_capacity(1024 * 1024); + let (mut index, counter, _tmpdir) = load_counted_v2_index(8, cache.clone()).await; + set_test_posting_group_size(&mut index, 4); + let inverted_list = index.partitions[0].inverted_list.clone(); + inverted_list.prewarm_posting_lists(false, 1).await.unwrap(); + let calls_after_prewarm = counter.read_range_calls(); + let rows_after_prewarm = counter.rows_read(); + let metrics = LocalMetricsCollector::default(); + + let posting = inverted_list + .posting_list_with_policy(0, false, &metrics, PostingReadPolicy::CacheAwareExact) + .await + .unwrap(); + + assert_eq!(posting.len(), 1); + assert_eq!(counter.read_range_calls(), calls_after_prewarm); + assert_eq!(counter.rows_read(), rows_after_prewarm); + assert_eq!(metrics.index_cache_hits(), 1); + assert_eq!(metrics.index_cache_misses(), 0); + assert_eq!(metrics.parts_loaded.load(Ordering::Relaxed), 0); +} + +#[tokio::test] +async fn test_cache_aware_exact_prefers_group_after_singleton_then_prewarm() { + let cache = LanceCache::with_capacity(1024 * 1024); + let (mut index, counter, _tmpdir) = load_counted_v2_index(8, cache.clone()).await; + set_test_posting_group_size(&mut index, 4); + let inverted_list = index.partitions[0].inverted_list.clone(); + + inverted_list + .posting_list_with_policy( + 0, + false, + &NoOpMetricsCollector, + PostingReadPolicy::CacheAwareExact, + ) + .await + .unwrap(); + assert_eq!(counter.read_range_calls(), 1); + assert_eq!(counter.rows_read(), 1); + + inverted_list.prewarm_posting_lists(false, 1).await.unwrap(); + let calls_after_prewarm = counter.read_range_calls(); + let rows_after_prewarm = counter.rows_read(); + assert!(calls_after_prewarm > 1); + assert!(rows_after_prewarm > 1); + + let metrics = LocalMetricsCollector::default(); + inverted_list + .posting_list_with_policy(0, false, &metrics, PostingReadPolicy::CacheAwareExact) + .await + .unwrap(); + assert_eq!(counter.read_range_calls(), calls_after_prewarm); + assert_eq!(counter.rows_read(), rows_after_prewarm); + assert_eq!(metrics.index_cache_hits(), 1); + assert_eq!(metrics.index_cache_misses(), 0); +} + +#[tokio::test] +async fn test_default_posting_read_keeps_read_ahead_group() { + let cache = LanceCache::with_capacity(1024 * 1024); + let (mut index, counter, _tmpdir) = load_counted_v2_index(8, cache.clone()).await; + set_test_posting_group_size(&mut index, 4); + let inverted_list = index.partitions[0].inverted_list.clone(); + + let posting = inverted_list + .posting_list(0, false, &NoOpMetricsCollector) + .await + .unwrap(); + + assert_eq!(posting.len(), 1); + assert_eq!(counter.read_range_calls(), 1); + assert_eq!(counter.rows_read(), 4); + assert_eq!(counter.metadata_rows_read(), 4); + assert_eq!(counter.posting_rows_read(), 4); + assert_eq!(counter.impact_rows_read(), 4); +} + +#[tokio::test] +async fn test_cache_aware_same_group_expansions_share_one_read_ahead_group() { + let cache = LanceCache::with_capacity(1024 * 1024); + let (mut index, counter, _tmpdir) = load_counted_v2_index(8, cache.clone()).await; + set_test_posting_group_size(&mut index, 4); + let partition = index.partitions[0].clone(); + let tokens = Tokens::with_positions( + vec!["t0".to_owned(), "t1".to_owned()], + vec![0, 0], + DocType::Text, + ); + let scorer = MemBM25Scorer::new( + 8, + 8, + HashMap::from([("t0".to_owned(), 1), ("t1".to_owned(), 1)]), + ); + let metrics = LocalMetricsCollector::default(); + + let loaded = partition + .load_posting_lists_with_policy( + &tokens, + &FtsSearchParams::new(), + Operator::Or, + &scorer, + &metrics, + PostingLoadOptions::cache_aware_exact(true), + ) + .await + .unwrap(); + + assert_eq!(loaded.postings.len(), 1); + assert_eq!(counter.read_range_calls(), 1); + assert_eq!(counter.rows_read(), 4); + assert_eq!(counter.metadata_rows_read(), 4); + assert_eq!(counter.posting_rows_read(), 4); + assert_eq!(counter.impact_rows_read(), 4); + assert_eq!(metrics.index_cache_misses(), 1); + assert_eq!(metrics.index_cache_hits(), 1); + assert_eq!(metrics.parts_loaded.load(Ordering::Relaxed), 1); +} + +#[tokio::test] +async fn test_cache_aware_repeated_token_positions_share_one_singleton() { + let cache = LanceCache::with_capacity(1024 * 1024); + let (mut index, counter, _tmpdir) = load_counted_v2_index(8, cache.clone()).await; + set_test_posting_group_size(&mut index, 4); + let partition = index.partitions[0].clone(); + let tokens = Tokens::with_positions( + vec!["t0".to_owned(), "t0".to_owned()], + vec![0, 1], + DocType::Text, + ); + let scorer = MemBM25Scorer::new(8, 8, HashMap::from([("t0".to_owned(), 1)])); + let metrics = LocalMetricsCollector::default(); + + let loaded = partition + .load_posting_lists_with_policy( + &tokens, + &FtsSearchParams::new(), + Operator::And, + &scorer, + &metrics, + PostingLoadOptions::cache_aware_exact(true), + ) + .await + .unwrap(); + + assert_eq!(loaded.postings.len(), 2); + assert_eq!(counter.read_range_calls(), 1); + assert_eq!(counter.rows_read(), 1); + assert_eq!(metrics.index_cache_misses(), 1); + assert_eq!(metrics.index_cache_hits(), 1); + assert_eq!(metrics.parts_loaded.load(Ordering::Relaxed), 1); +} + /// Build a single-partition v2 index where every token's posting list spans /// `docs_per_token` docs. Runtime grouping packs consecutive token rows /// into shared cache groups.