From a1e707f0290526e41094226843e27ff39922dbe6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=B6ren=20Brunk?= Date: Mon, 17 Aug 2026 12:07:44 +0200 Subject: [PATCH 1/3] feat(fts): score unindexed and partially indexed fragments in combined_fields The previous commit refuses a `combined_fields` query whose target columns do not all cover every scanned fragment, so a default full-text search fails on any dataset with rows appended since the indexes were built. `MatchQuery` already unions in a flat scan for its unindexed fragments; this does the same for BM25F. Coverage is per column here, which makes it more than a copy of the single-column path. `dl'` sums each column's document length and a row absent from a column's `DocSet` contributes 0, so a fragment indexed for `title` but not `body` cannot be scored from the index at all. The indexed scan is therefore restricted to the intersection of per-column coverage and everything else goes to the flat scan, rather than splitting on the union. Both sides then score against one shared corpus. The flat side alone sees the rows no index covers, so it measures their contribution and publishes the blend; the indexed side waits for it instead of folding only its own `docCount'`/`docFreq'`/`avgdl'`. Without that, a row reached through either path would rank differently depending on which side happened to score it. Data overlays are handled by measuring rather than patching. When a target column carries an overlay-stale index entry, folding the flat row into the index statistics would double count it against the entry it replaces, and the flat scan cannot subtract what it replaced. So the corpus is measured from current data instead: every target fragment is scanned, every row folded into every column, and the index statistics left out. That costs a full scan of the target columns, so it stays confined to the stale case. `fast_search` is unchanged, being index-only by contract. --- rust/lance-index/src/scalar/inverted.rs | 4 +- .../src/scalar/inverted/combined.rs | 11 +- .../src/scalar/inverted/combined/flat.rs | 201 ++++++ .../src/scalar/inverted/combined/search.rs | 15 +- .../src/scalar/inverted/combined/stats.rs | 260 +++++++- .../src/scalar/inverted/combined/testing.rs | 65 +- rust/lance-index/src/scalar/inverted/index.rs | 6 +- .../src/scalar/inverted/index/flat_search.rs | 222 ++++++- .../src/scalar/inverted/index/partition.rs | 58 ++ .../src/scalar/inverted/index/search.rs | 70 ++ .../src/scalar/inverted/index/tests/stats.rs | 9 +- rust/lance/src/dataset/scanner.rs | 221 ++++++- rust/lance/src/index/scalar/inverted.rs | 65 ++ rust/lance/src/io/exec/fts.rs | 606 ++++++++++++++++-- rust/lance/tests/query/inverted.rs | 20 +- 15 files changed, 1717 insertions(+), 116 deletions(-) create mode 100644 rust/lance-index/src/scalar/inverted/combined/flat.rs diff --git a/rust/lance-index/src/scalar/inverted.rs b/rust/lance-index/src/scalar/inverted.rs index 11a2b6e513f..a5c7dd05f8b 100644 --- a/rust/lance-index/src/scalar/inverted.rs +++ b/rust/lance-index/src/scalar/inverted.rs @@ -30,8 +30,8 @@ use arrow_schema::{DataType, Field}; use async_trait::async_trait; pub use builder::InvertedIndexBuilder; pub use combined::{ - CombinedFieldColumn, build_combined_bm25_scorer, combined_fields_search, - validate_combined_tokenizers, + CombinedCorpusStats, CombinedFieldColumn, FlatFieldStats, build_combined_bm25_scorer, + combined_fields_search, flat_combined_fields_search_stream, validate_combined_tokenizers, }; pub use compound::{ compound_search, compound_search_prepared_match, diff --git a/rust/lance-index/src/scalar/inverted/combined.rs b/rust/lance-index/src/scalar/inverted/combined.rs index 9de77185ce7..e48f245b8a0 100644 --- a/rust/lance-index/src/scalar/inverted/combined.rs +++ b/rust/lance-index/src/scalar/inverted/combined.rs @@ -23,6 +23,7 @@ //! scored; see [`combined_fields_search`]. mod cursor; +mod flat; mod search; mod stats; #[cfg(test)] @@ -31,9 +32,11 @@ mod testing; use std::sync::Arc; use lance_core::{Error, Result}; +use lance_select::RowAddrTreeMap; +pub use flat::flat_combined_fields_search_stream; pub use search::combined_fields_search; -pub use stats::build_combined_bm25_scorer; +pub use stats::{CombinedCorpusStats, FlatFieldStats, build_combined_bm25_scorer}; use super::index::InvertedIndex; use super::query::Tokens; @@ -47,6 +50,12 @@ pub struct CombinedFieldColumn { pub weight: f32, /// Opened inverted-index segments for this column. pub indices: Vec>, + /// Rows an overlay made stale, in the row-id domain index results use. + /// + /// The flat scan folds their current values into the corpus, so their old + /// values are subtracted from `indices` to avoid counting them twice. + /// `None` when no overlay touched this column. + pub stale_rows: Option>, } /// Deduplicate the query tokens into the unique terms of the virtual field, diff --git a/rust/lance-index/src/scalar/inverted/combined/flat.rs b/rust/lance-index/src/scalar/inverted/combined/flat.rs new file mode 100644 index 00000000000..0e0d6ac3155 --- /dev/null +++ b/rust/lance-index/src/scalar/inverted/combined/flat.rs @@ -0,0 +1,201 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +//! The unindexed `combined_fields` plan: blend the scanned column values into +//! `dl'`/`tf'` and score the rows no target column's index covers. + +use std::sync::Arc; + +use arrow::array::{Float32Builder, UInt64Builder}; +use arrow_array::{ArrayRef, RecordBatch}; +use datafusion::execution::SendableRecordBatchStream; +use datafusion::physical_plan::metrics::Time; +use datafusion::physical_plan::stream::RecordBatchStreamAdapter; +use futures::{FutureExt, stream}; +use lance_core::error::DataFusionResult; +use lance_core::utils::tokio::spawn_cpu; +use lance_select::RowAddrMask; + +use super::super::index::{BlendedRows, FTS_SCHEMA, slice_into_batches, tokenize_and_blend_multi}; +use super::super::query::{Operator, Tokens}; +use super::super::scorer::CombinedFieldsBM25Scorer; +use super::super::tokenizer::document_tokenizer::LanceTokenizer; +use super::stats::{CombinedCorpusStats, FlatFieldStats, build_combined_bm25_scorer}; +use super::{CombinedFieldColumn, unique_terms}; +use crate::metrics::MetricsCollector; + +/// Exact cross-field BM25F search over rows that no index fully covers, scored +/// straight from their column values. Together with the indexed +/// [`combined_fields_search`](super::combined_fields_search) this covers the +/// whole dataset, with the same term deduplication and `operator` semantics. +/// +/// - `input` carries `_rowid` plus every target column; `doc_col_indices` locates +/// them, in `columns` order. +/// - `stats_masks[column]` selects the rows folded into that column's corpus +/// statistics, so a row indexed for some columns is not counted twice. +/// - `emit_mask`, when set, selects the rows to emit from an unfiltered `input`. +/// - `flat_covers_whole_corpus` selects [`CombinedCorpusStats::FlatOnly`]. +/// - `metrics` receives only the scorer build's index reads. +/// +/// Returns the scorer alongside the stream. The whole input is consumed before +/// the first output batch, so an indexed sibling can score against the same +/// statistics. +#[allow(clippy::too_many_arguments)] +pub async fn flat_combined_fields_search_stream( + input: SendableRecordBatchStream, + columns: &[CombinedFieldColumn], + doc_col_indices: Vec, + stats_masks: &[Arc], + emit_mask: Option>, + flat_covers_whole_corpus: bool, + tokens: &Tokens, + tokenizer: Box, + operator: Operator, + target_batch_size: usize, + elapsed_compute: Option