From 1c65d26dafb61f1c8c1d96e097dec2a29902be46 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Wed, 8 Jul 2026 17:49:22 +0800 Subject: [PATCH 01/19] feat: add code analyzer for FTS --- protos/index_old.proto | 30 ++ python/python/tests/test_scalar_index.py | 246 +++++++++ python/src/dataset.rs | 150 ++++-- rust/lance-index/src/scalar/inverted/index.rs | 307 +++++++++-- .../src/scalar/inverted/tokenizer.rs | 482 ++++++++++++++++-- rust/lance-tokenizer/src/code_tokenizer.rs | 140 +++++ rust/lance-tokenizer/src/lib.rs | 4 + .../src/word_delimiter_filter.rs | 279 ++++++++++ rust/lance/src/dataset/mem_wal/index.rs | 13 +- rust/lance/src/dataset/mem_wal/index/fts.rs | 148 +++++- rust/lance/src/index/scalar/inverted.rs | 44 +- rust/lance/src/io/exec/fts.rs | 5 +- 12 files changed, 1684 insertions(+), 164 deletions(-) create mode 100644 rust/lance-tokenizer/src/code_tokenizer.rs create mode 100644 rust/lance-tokenizer/src/word_delimiter_filter.rs diff --git a/protos/index_old.proto b/protos/index_old.proto index 601aa2681da..d9024183b55 100644 --- a/protos/index_old.proto +++ b/protos/index_old.proto @@ -26,6 +26,19 @@ message LabelListIndexDetails {} message NGramIndexDetails {} message ZoneMapIndexDetails {} message InvertedIndexDetails { + // High-level analyzer profile. It selects the semantic defaults for the FTS + // pipeline, e.g. "text" for natural-language text or "code" for code search. + // Absent is a legacy "text" profile unless base_tokenizer is the legacy + // "code" alias. Readers must canonicalize this before comparing details. + optional string analyzer = 12; + // Document-level tokenizer. It extracts searchable text from the stored value, + // e.g. "text" for a plain string or "json" for JSON content. The extracted + // text is then passed to base_tokenizer. Absent means text / auto-inferred + // legacy behavior. + optional string lance_tokenizer = 13; + // Lexical tokenizer used after the analyzer/document layer has selected the + // text to tokenize. This is an implementation component such as "simple", + // "icu", "ngram", or "code"; analyzer="code" owns the code-search defaults. // Marking this field as optional as old versions of the index store blank details and we // need to make sure we have a proper optional field to detect this. optional string base_tokenizer = 1; @@ -39,4 +52,21 @@ message InvertedIndexDetails { uint32 min_ngram_length = 9; uint32 max_ngram_length = 10; bool prefix_only = 11; + // Code analyzer flag. Split one lexical identifier into subwords, e.g. + // getUserName -> get/user/name. This changes the indexed term space and must + // be persisted for query-time tokenizer reconstruction and segment checks. + bool split_identifiers = 14; + // Code analyzer flag. Split identifier subwords across letter/number + // boundaries, e.g. HTML2JSON -> html/2/json. This changes the indexed term + // space and must be persisted with the index. + bool split_on_numerics = 15; + // Code analyzer flag. Keep the complete lexical identifier in addition to + // subwords, e.g. user_name plus user/name. This affects BM25 term statistics + // and must be persisted with the index. + bool preserve_original = 16; + // Code analyzer flag. Index operator tokens such as "::", "->", and "!=". + // Disabled by default because operators are often high-frequency noise, but + // when enabled it changes query semantics and must be persisted. + bool index_operators = 17; + repeated string custom_stop_words = 19; } diff --git a/python/python/tests/test_scalar_index.py b/python/python/tests/test_scalar_index.py index a2e35e6749f..52af81bc7f7 100644 --- a/python/python/tests/test_scalar_index.py +++ b/python/python/tests/test_scalar_index.py @@ -21,6 +21,7 @@ from lance.query import ( BooleanQuery, BoostQuery, + FullTextOperator, MatchQuery, MultiMatchQuery, Occur, @@ -806,6 +807,239 @@ def test_full_text_search(dataset, with_position, base_tokenizer): ) +def test_code_analyzer_full_text_search(tmp_path): + table = pa.table( + { + "code": [ + "getUserName", + "set_user_name", + "user-name", + "username", + "other", + ] + } + ) + ds = lance.write_dataset(table, tmp_path) + ds.create_scalar_index("code", index_type="INVERTED", analyzer="code") + + results = ds.to_table( + columns=["code"], + full_text_query=MatchQuery("user", "code"), + ) + assert set(results["code"].to_pylist()) == { + "getUserName", + "set_user_name", + "user-name", + } + + stats = ds.stats.index_stats("code_idx")["indices"][0] + params = stats["params"] + assert params["analyzer"] == "code" + assert params["base_tokenizer"] == "code" + assert params["split_identifiers"] is True + assert params["split_on_numerics"] is True + assert params["preserve_original"] is True + assert params["stem"] is False + assert params["remove_stop_words"] is False + + +def test_code_analyzer_complex_code_constructs(tmp_path): + table = pa.table( + { + "path": [ + "edge/trait.rs", + "edge/impl.rs", + "edge/fn_pointer.rs", + "edge/unit_result.rs", + "edge/hrtb.rs", + "edge/associated.rs", + "edge/operators.rs", + ], + "code": [ + """ +pub trait EdgeAsyncRepository<'a, T: Send + Sync> +where + T: TryFrom<&'a str, Error = EdgeParseError>, +{ + type Output<'b>: Iterator> + where + Self: 'b; + + async fn fetch_by_key( + &'a self, + key: [u8; N], + ) -> Result, EdgeRepoError>; +} +""", + """ +impl<'a, T, S> EdgeAsyncRepository<'a, T> for EdgeStore +where + T: TryFrom<&'a str, Error = EdgeParseError> + Clone + Send + Sync, + S: EdgeBackend + ?Sized, +{ + type Output<'b> = std::vec::IntoIter> where Self: 'b; + + async fn fetch_by_key( + &'a self, + key: [u8; N], + ) -> Result, EdgeRepoError> { + self.backend.fetch::(key).await + } +} +""", + """ +pub fn build_edge_handler( + factory: F, +) -> impl Fn() -> Result, EdgeError> +where + F: FnOnce() -> Result + Send + 'static, + T: Default + Send + Sync + 'static, +{ + move || factory().map(EdgeHandler::new) +} +""", + """ +pub fn edge_unit_result_callback() -> Result<()> { + Ok(()) +} +""", + """ +pub fn edge_higher_ranked<'a, T>( + visitor: impl for<'b> Fn(&'b T) -> Result<&'b str, EdgeVisitError>, + value: &'a T, +) -> Result<&'a str, EdgeVisitError> { + visitor(value) +} +""", + """ +pub fn edge_collect_stream(items: I) -> Result, E::Error> +where + I: IntoIterator, + E: EdgeExtract, +{ + items.into_iter().map(E::extract).collect() +} +""", + """ +pub fn edge_operator_arrow() -> Result { + let variant = EdgeModule::EdgeVariant; + if variant != EdgeModule::Default && EdgeMask::enabled() { + return Ok(EdgeArrow::new(variant)); + } + Err(EdgeError::empty()) +} +""", + ], + } + ) + table = table.append_column("code_ops", table["code"]) + ds = lance.write_dataset(table, tmp_path) + ds.create_scalar_index("code", index_type="INVERTED", analyzer="code") + ds.create_scalar_index( + "code_ops", + index_type="INVERTED", + analyzer="code", + index_operators=True, + ) + + ds.insert( + pa.table( + { + "path": ["edge/flat_unindexed.rs"], + "code": [ + """ +pub async fn edge_flat_generic_return() -> Result +where + T: TryFrom + Send, + E: Into, +{ + T::try_from(String::new()).map_err(Into::into) +} +""" + ], + "code_ops": [ + """ +pub async fn edge_flat_operator() -> Result { + EdgeFlat::try_new() -> Result +} +""" + ], + } + ) + ) + ds = lance.dataset(tmp_path) + + def assert_search(column, query, expected_path, operator=FullTextOperator.AND): + result = ds.scanner( + columns=["path", "_score"], + full_text_query=MatchQuery(query, column, operator=operator), + limit=50, + ).to_table() + assert expected_path in result["path"].to_pylist() + + assert_search( + "code", + "EdgeAsyncRepository fetch_by_key TryFrom EdgeRepoError", + "edge/trait.rs", + ) + assert_search( + "code", + "EdgeStore fetch_by_key const usize where Result", + "edge/impl.rs", + ) + assert_search( + "code", + "build_edge_handler FnOnce Result EdgeHandler", + "edge/fn_pointer.rs", + ) + assert_search( + "code", + "edge_unit_result_callback fn () -> Result", + "edge/unit_result.rs", + ) + assert_search( + "code", + "edge_higher_ranked for Fn EdgeVisitError Result", + "edge/hrtb.rs", + ) + assert_search( + "code", + "edge_collect_stream IntoIterator Item Error Result", + "edge/associated.rs", + ) + assert_search( + "code", + "edge_flat_generic_return TryFrom EdgeFlatError Result", + "edge/flat_unindexed.rs", + ) + assert_search( + "code_ops", + "edge_operator_arrow -> Result", + "edge/operators.rs", + ) + assert_search( + "code_ops", + "EdgeModule :: EdgeVariant !=", + "edge/operators.rs", + ) + assert_search( + "code_ops", + "edge_flat_operator -> Result EdgeFlatError", + "edge/flat_unindexed.rs", + ) + + default_operator_results = ds.scanner( + columns=["path", "_score"], + full_text_query=MatchQuery("->", "code", operator=FullTextOperator.OR), + ).to_table() + operator_results = ds.scanner( + columns=["path", "_score"], + full_text_query=MatchQuery("->", "code_ops", operator=FullTextOperator.OR), + ).to_table() + assert default_operator_results.num_rows == 0 + assert operator_results.num_rows > 0 + + def test_unindexed_full_text_search_on_empty_index(tmp_path): # Create fts index on empty table. schema = pa.schema({"text": pa.string()}) @@ -871,6 +1105,18 @@ def test_fts_custom_stop_words(tmp_path): assert len(results["_rowid"].to_pylist()) == 1 +def test_fts_custom_stop_words_none(tmp_path): + data = pa.table({"text": ["alpha beta", "gamma"]}) + ds = lance.write_dataset(data, tmp_path) + ds.create_scalar_index("text", "INVERTED", custom_stop_words=None) + + results = ds.to_table(full_text_query="alpha") + assert results.num_rows == 1 + + stats = ds.stats.index_stats("text_idx")["indices"][0] + assert stats["params"]["custom_stop_words"] is None + + def test_fts_stop_words_respect_language_for_simple_tokenizer(tmp_path): data = pa.table({"text": ["the lance data", "的 lance data"]}) ds = lance.write_dataset(data, tmp_path, mode="overwrite") diff --git a/python/src/dataset.rs b/python/src/dataset.rs index a7e8b52fb1f..3442071e965 100644 --- a/python/src/dataset.rs +++ b/python/src/dataset.rs @@ -78,7 +78,6 @@ use lance_index::{ FtsPrewarmOptions, IndexParams, IndexType, PrewarmOptions, optimize::OptimizeOptions, progress::{IndexBuildProgress, NoopIndexBuildProgress}, - scalar::inverted::InvertedListFormatVersion, scalar::{FullTextSearchQuery, InvertedIndexParams, ScalarIndexParams}, vector::{ ApproxMode, DEFAULT_QUERY_PARALLELISM, Query as VectorQuery, @@ -2402,75 +2401,130 @@ impl Dataset { }) } "INVERTED" | "FTS" => { - let mut params = InvertedIndexParams::default(); + let mut params_json = serde_json::Map::new(); if let Some(kwargs) = kwargs { - if let Some(with_position) = kwargs.get_item("with_position")? { - params = params.with_position(with_position.extract()?); + let allowed_kwargs = [ + "analyzer", + "with_position", + "base_tokenizer", + "language", + "max_token_length", + "lower_case", + "stem", + "remove_stop_words", + "custom_stop_words", + "ascii_folding", + "min_ngram_length", + "max_ngram_length", + "prefix_only", + "split_identifiers", + "split_on_numerics", + "preserve_original", + "index_operators", + "memory_limit", + "num_workers", + "format_version", + "fragment_ids", + "index_uuid", + "progress_callback", + ]; + for (key, _) in kwargs.iter() { + let key: String = key.extract()?; + if !allowed_kwargs.contains(&key.as_str()) { + return Err(PyValueError::new_err(format!( + "unknown FTS index parameter '{}'", + key + ))); + } + } + macro_rules! insert_param { + ($name:literal, $ty:ty) => { + if let Some(value) = kwargs.get_item($name)? { + let value: $ty = value.extract()?; + params_json.insert( + $name.to_string(), + serde_json::to_value(value).map_err(|err| { + PyValueError::new_err(format!( + "failed to serialize FTS parameter {}: {}", + $name, err + )) + })?, + ); + } + }; } - if let Some(base_tokenizer) = kwargs.get_item("base_tokenizer")? { - params = params.base_tokenizer(base_tokenizer.extract()?); + insert_param!("analyzer", String); + if let Some(with_position) = kwargs.get_item("with_position")? { + params_json.insert( + "with_position".to_string(), + serde_json::Value::Bool(with_position.extract()?), + ); } + insert_param!("base_tokenizer", String); if let Some(language) = kwargs.get_item("language")? { let language: PyBackedStr = language.cast::()?.clone().try_into()?; - params = params.language(&language).map_err(|e| { - PyValueError::new_err(format!( - "can't set tokenizer language to {}: {:?}", - language, e - )) - })?; + params_json.insert( + "language".to_string(), + serde_json::Value::String(language.to_string()), + ); } if let Some(max_token_length) = kwargs.get_item("max_token_length")? { - params = params.max_token_length(max_token_length.extract()?); + let max_token_length: Option = max_token_length.extract()?; + params_json.insert( + "max_token_length".to_string(), + serde_json::to_value(max_token_length).map_err(|err| { + PyValueError::new_err(format!( + "failed to serialize FTS parameter max_token_length: {}", + err + )) + })?, + ); } - if let Some(lower_case) = kwargs.get_item("lower_case")? { - params = params.lower_case(lower_case.extract()?); - } - if let Some(stem) = kwargs.get_item("stem")? { - params = params.stem(stem.extract()?); - } - if let Some(remove_stop_words) = kwargs.get_item("remove_stop_words")? { - params = params.remove_stop_words(remove_stop_words.extract()?); - } - if let Some(stop_words_file) = kwargs.get_item("custom_stop_words")? { - params = params.custom_stop_words(stop_words_file.extract()?); - } - if let Some(ascii_folding) = kwargs.get_item("ascii_folding")? { - params = params.ascii_folding(ascii_folding.extract()?); - } - if let Some(min_ngram_length) = kwargs.get_item("min_ngram_length")? { - params = params.ngram_min_length(min_ngram_length.extract()?); - } - if let Some(max_ngram_length) = kwargs.get_item("max_ngram_length")? { - params = params.ngram_max_length(max_ngram_length.extract()?); - } - if let Some(prefix_only) = kwargs.get_item("prefix_only")? { - params = params.ngram_prefix_only(prefix_only.extract()?); - } - if let Some(memory_limit) = kwargs.get_item("memory_limit")? { - params = params.memory_limit_mb(memory_limit.extract()?); - } - if let Some(num_workers) = kwargs.get_item("num_workers")? { - params = params.num_workers(num_workers.extract()?); + insert_param!("lower_case", bool); + insert_param!("stem", bool); + insert_param!("remove_stop_words", bool); + if let Some(custom_stop_words) = kwargs.get_item("custom_stop_words")? { + let custom_stop_words: Option> = + custom_stop_words.extract()?; + params_json.insert( + "custom_stop_words".to_string(), + serde_json::to_value(custom_stop_words).map_err(|err| { + PyValueError::new_err(format!( + "failed to serialize FTS parameter custom_stop_words: {}", + err + )) + })?, + ); } + insert_param!("ascii_folding", bool); + insert_param!("min_ngram_length", u32); + insert_param!("max_ngram_length", u32); + insert_param!("prefix_only", bool); + insert_param!("split_identifiers", bool); + insert_param!("split_on_numerics", bool); + insert_param!("preserve_original", bool); + insert_param!("index_operators", bool); + insert_param!("memory_limit", u64); + insert_param!("num_workers", usize); if let Some(format_version) = kwargs.get_item("format_version")? && !format_version.is_none() { let value = if let Ok(value) = format_version.cast::() { - value.to_string_lossy().to_string() + serde_json::Value::String(value.to_string_lossy().to_string()) } else if let Ok(value) = format_version.extract::() { - value.to_string() + serde_json::Value::from(value) } else { return Err(PyValueError::new_err( "format_version must be 1, 2, 'v1', or 'v2'", )); }; - let format_version = value - .parse::() - .map_err(|err| PyValueError::new_err(err.to_string()))?; - params = params.format_version(format_version); + params_json.insert("format_version".to_string(), value); } } + let params: InvertedIndexParams = + serde_json::from_value(serde_json::Value::Object(params_json)) + .map_err(|err| PyValueError::new_err(err.to_string()))?; Box::new(params) } _ => { diff --git a/rust/lance-index/src/scalar/inverted/index.rs b/rust/lance-index/src/scalar/inverted/index.rs index e1aab1c2843..0fff5b15c38 100644 --- a/rust/lance-index/src/scalar/inverted/index.rs +++ b/rust/lance-index/src/scalar/inverted/index.rs @@ -1758,7 +1758,6 @@ impl InvertedPartition { operator: Operator, metrics: &dyn MetricsCollector, ) -> Result { - let is_fuzzy = matches!(params.fuzziness, Some(n) if n != 0); 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(|| { @@ -1784,9 +1783,6 @@ impl InvertedPartition { matched_positions.insert(position); } token_ids.push((token_id, token, position)); - } else if is_phrase_query || is_and_query { - // if the token is not found, we can't do phrase or AND query - return Ok(LoadedPostings::empty()); } } if token_ids.is_empty() { @@ -1799,16 +1795,8 @@ impl InvertedPartition { return Ok(LoadedPostings::empty()); } - let is_fuzzy_and_query = is_fuzzy && is_and_query && !is_phrase_query; - if !is_phrase_query { - if is_fuzzy_and_query { - 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); - } else { - token_ids.sort_unstable_by_key(|(token_id, _, _)| *token_id); - token_ids.dedup_by_key(|(token_id, _, _)| *token_id); - } - } + 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 num_docs = self.docs.len(); let loaded_postings = stream::iter(token_ids) @@ -1824,35 +1812,6 @@ impl InvertedPartition { .try_collect::>() .await?; - if (is_and_query || is_phrase_query) - && !is_fuzzy_and_query - && loaded_postings - .iter() - .any(|(_, _, _, posting)| posting.is_empty()) - { - return Ok(LoadedPostings::empty()); - } - - if !is_fuzzy_and_query { - return Ok(LoadedPostings { - postings: loaded_postings - .into_iter() - .map(|(token_id, token, position, posting)| { - let query_weight = idf(posting.len(), num_docs); - PostingIterator::with_query_weight( - token, - token_id, - position, - query_weight, - posting, - num_docs, - ) - }) - .collect(), - grouped_expansions: Vec::new(), - }); - } - let needs_union = loaded_postings .windows(2) .any(|window| window[0].2 == window[1].2); @@ -1880,6 +1839,10 @@ impl InvertedPartition { } else { let token_id = group[0].0; let token = group[0].1.clone(); + let query_weight = group + .iter() + .map(|(_, _, posting)| idf(posting.len(), num_docs)) + .sum::(); grouped_expansions.push(GroupedExpansionTerms { position, terms: group @@ -1895,12 +1858,26 @@ impl InvertedPartition { postings, docs_for_union .as_deref() - .expect("union docs must be loaded for grouped fuzzy AND"), + .expect("union docs must be loaded for grouped query terms"), )?; - (token_id, token, posting) + if posting.is_empty() && (is_and_query || is_phrase_query) { + return Ok(LoadedPostings::empty()); + } + grouped_postings.push(PostingIterator::with_query_weight( + token, + token_id, + position, + query_weight, + posting, + num_docs, + )); + continue; }; if posting.is_empty() { - return Ok(LoadedPostings::empty()); + if is_and_query || is_phrase_query { + return Ok(LoadedPostings::empty()); + } + continue; } let query_weight = idf(posting.len(), num_docs); @@ -5575,6 +5552,7 @@ async fn tokenize_and_count( ), ])); let output_schema_clone = output_schema.clone(); + let query_token_indices = Arc::new(query_token_indices(query_tokens.as_ref())); let bytes_accumulated = Arc::new(AtomicU64::new(0)); let bytes_warning_emitted = Arc::new(AtomicBool::new(false)); @@ -5583,6 +5561,7 @@ async fn tokenize_and_count( let mut tokenizer = tokenizer.box_clone(); let output_schema = output_schema.clone(); let query_tokens = query_tokens.clone(); + let query_token_indices = query_token_indices.clone(); let bytes_accumulated = bytes_accumulated.clone(); let bytes_warning_emitted = bytes_warning_emitted.clone(); let elapsed_compute = elapsed_compute.clone(); @@ -5606,8 +5585,10 @@ async fn tokenize_and_count( let mut all_tokens = 0; while let Some(token) = stream.next() { all_tokens += 1; - if let Some(token_index) = query_tokens.token_index(&token.text) { - temp_query_token_counts[token_index] += 1; + if let Some(token_indices) = query_token_indices.get(&token.text) { + for token_index in token_indices { + temp_query_token_counts[*token_index] += 1; + } } } all_tokens @@ -5737,6 +5718,17 @@ fn tokenize_and_count_list( Ok(()) } +fn query_token_indices(query_tokens: &Tokens) -> HashMap> { + let mut indices = HashMap::new(); + for idx in 0..query_tokens.len() { + indices + .entry(query_tokens.get_token(idx).to_string()) + .or_insert_with(Vec::new) + .push(idx); + } + indices +} + /// Initialize the BM25 scorer /// /// In order to calculate BM25 scores we need to know token counts for the entire corpus. We extract these from the @@ -5800,9 +5792,11 @@ fn flat_bm25_score( query_tokens: &Tokens, counted_input: &RecordBatch, scorer: &MemBM25Scorer, + operator: Operator, ) -> Result { let mut row_ids_builder = UInt64Builder::with_capacity(counted_input.num_rows()); let mut scores_builder = Float32Builder::with_capacity(counted_input.num_rows()); + let query_groups = query_position_groups(query_tokens); let mut row_ids_iter = counted_input .column(FLAT_ROW_ID_COL_IDX) @@ -5827,16 +5821,24 @@ fn flat_bm25_score( for _ in 0..counted_input.num_rows() { let num_tokens_in_doc = all_token_counts_iter.next().expect_ok()?; let row_id = row_ids_iter.next().expect_ok()?; + let mut query_token_counts = Vec::with_capacity(query_tokens.len()); + for _ in query_tokens { + query_token_counts.push(query_token_counts_iter.next().expect_ok()?); + } if num_tokens_in_doc == 0 { - for _ in query_tokens { - query_token_counts_iter.next().expect_ok()?; - } + continue; + } + if operator == Operator::And + && !query_groups + .iter() + .all(|group| group.iter().any(|idx| query_token_counts[*idx] > 0)) + { continue; } let doc_norm = K1 * (1.0 - B + B * num_tokens_in_doc as f32 / scorer.avg_doc_length()); let mut score = 0.0; - for token in query_tokens { - let freq = query_token_counts_iter.next().expect_ok()? as f32; + for (token, freq) in query_tokens.into_iter().zip(query_token_counts.into_iter()) { + let freq = freq as f32; let idf = idf(scorer.num_docs_containing_token(token), scorer.num_docs()); score += idf * (freq * (K1 + 1.0) / (freq + doc_norm)); } @@ -5855,6 +5857,23 @@ fn flat_bm25_score( Ok(batch) } +fn query_position_groups(query_tokens: &Tokens) -> Vec> { + let mut groups = Vec::new(); + let mut current_position = None; + for idx in 0..query_tokens.len() { + let position = query_tokens.position(idx); + if current_position != Some(position) { + current_position = Some(position); + groups.push(Vec::new()); + } + groups + .last_mut() + .expect("a group should exist after pushing for position") + .push(idx); + } + groups +} + #[deprecated( note = "use `flat_bm25_search_stream_with_metrics` to record CPU compute \ time on a metric handle; pass `None` for the old behavior" @@ -5892,6 +5911,32 @@ pub async fn flat_bm25_search_stream_with_metrics( base_scorer: Option, target_batch_size: usize, elapsed_compute: Option