diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index bac6ea35ae..c897a7314b 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -599,7 +599,15 @@ pub async fn tantivy_search( } { // Each result corresponds to a file in the file list match result { - Ok((file_name, result)) => { + Ok((file_name, result, has_skipped_conditions)) => { + // If any AND condition was skipped because its field does + // not exist in this tantivy index (e.g., a newly added + // secondary-index field with no historical data), keep the + // DataFusion filter so it can evaluate the skipped + // predicates on the parquet data. + if has_skipped_conditions { + is_add_filter_back = true; + } if file_name.is_empty() { // no need inverted index for this file, need add filter back let took = start.elapsed().as_millis() as usize; @@ -717,13 +725,18 @@ pub async fn get_tantivy_directory( Ok(PuffinDirReader::from_path(file_account, source).await?) } +/// Returns (file_key, result, has_skipped_conditions). +/// `has_skipped_conditions` is true when one or more AND conditions were skipped +/// because the corresponding field does not exist in this tantivy index (e.g., a +/// newly added secondary-index field). The caller must preserve the DataFusion +/// filter to handle those skipped predicates. async fn search_tantivy_index( trace_id: &str, time_range: (i64, i64), index_condition: Option, idx_optimize_rule: Option, parquet_file: &FileKey, -) -> anyhow::Result<(String, TantivyResult)> { +) -> anyhow::Result<(String, TantivyResult, bool)> { let file_account = parquet_file.account.clone(); // TODO: this convert happen two times, once before cache_files and once in search_tantivy_index let Some(ttv_file_name) = convert_parquet_file_name_to_tantivy_file(&parquet_file.key) else { @@ -744,7 +757,7 @@ async fn search_tantivy_index( metrics::TANTIVY_RESULT_CACHE_HITS_TOTAL .with_label_values::<&str>(&[]) .inc(); - return Ok((parquet_file.key.to_string(), result)); + return Ok((parquet_file.key.to_string(), result, false)); } } @@ -793,7 +806,8 @@ async fn search_tantivy_index( // generate the tantivy query let condition: IndexCondition = index_condition.ok_or(anyhow::anyhow!("IndexCondition not found"))?; - let query = condition.to_tantivy_query(tantivy_schema.clone(), fts_field)?; + let (query, has_skipped_conditions) = + condition.to_tantivy_query(tantivy_schema.clone(), fts_field)?; let need_all_term_fields = condition .need_all_term_fields() .into_iter() @@ -872,7 +886,7 @@ async fn search_tantivy_index( TantivyResult::Distinct(distinct) => TantivyResult::Distinct(distinct), TantivyResult::RowIds(row_ids) => { if row_ids.is_empty() || parquet_file.meta.records == 0 { - return Ok((key, TantivyResult::RowIdsBitVec(0, BitVec::EMPTY))); + return Ok((key, TantivyResult::RowIdsBitVec(0, BitVec::EMPTY), false)); } // return early if the number of matched docs is too large let skip_threshold = cfg.limit.inverted_index_skip_threshold; @@ -886,6 +900,7 @@ async fn search_tantivy_index( return Ok(( "".to_string(), TantivyResult::RowIdsBitVec(row_ids_percent as usize, BitVec::EMPTY), + true, )); } percent = row_ids_percent; @@ -910,15 +925,17 @@ async fn search_tantivy_index( }; // cache the result if the memory size is less than the limit + // Do not cache when conditions were skipped — the result is incomplete. if cfg.common.inverted_index_result_cache_enabled && !cache_key.is_empty() + && !has_skipped_conditions && (result.get_memory_size() < cfg.limit.inverted_index_result_cache_max_entry_size || percent < 1.0) { let entry = get_cache_entry(result.clone(), percent, parquet_file.meta.records as usize); tantivy_result_cache::GLOBAL_CACHE.put(cache_key, entry); } - Ok((key, result)) + Ok((key, result, has_skipped_conditions)) } /// if simple distinct without filter, we need to warm up the field diff --git a/src/service/search/index.rs b/src/service/search/index.rs index 48e4d31eb4..753b57450d 100644 --- a/src/service/search/index.rs +++ b/src/service/search/index.rs @@ -127,21 +127,42 @@ impl IndexCondition { } // get the tantivy query for the index condition + // Returns (query, has_skipped): + // has_skipped = true means some conditions were skipped because the field + // does not exist in this tantivy index (e.g., a newly added index field + // that has no historical data). The caller must keep the DataFusion filter + // so that the skipped predicates are still evaluated. pub fn to_tantivy_query( &self, schema: Schema, default_field: Option, - ) -> anyhow::Result> { - let queries = self - .conditions - .iter() - .map(|condition| { - condition - .to_tantivy_query(&schema, default_field) - .map(|condition| (Occur::Must, condition)) - }) - .collect::>>()?; - Ok(Box::new(BooleanQuery::from(queries))) + ) -> anyhow::Result<(Box, bool)> { + let mut has_skipped = false; + let mut queries: Vec<(Occur, Box)> = Vec::with_capacity(self.conditions.len()); + for condition in &self.conditions { + match condition.to_tantivy_query(&schema, default_field) { + Ok(query) => { + queries.push((Occur::Must, query)); + } + Err(e) => { + // Field not present in this tantivy index file (e.g., a + // secondary index field that was added after this file was + // created). Skip the condition and let DataFusion handle it. + log::debug!("to_tantivy_query: skipping condition due to missing field: {e}"); + has_skipped = true; + } + } + } + if queries.is_empty() { + // All conditions were skipped because none of the queried fields + // exist in this tantivy index. Return an error so the caller falls + // back to a direct parquet scan instead of doing a full-index scan. + Err(anyhow::anyhow!( + "All AND conditions reference fields missing from the tantivy index" + )) + } else { + Ok((Box::new(BooleanQuery::from(queries)), has_skipped)) + } } // get the fields use for search in datafusion(for add filter back logical) @@ -2103,4 +2124,95 @@ mod tests { ]))); assert_eq!(get_arg_name(&unnamed_other), UNKNOWN_NAME); } + + /// Build a minimal tantivy schema containing only the given text field names. + fn build_tantivy_schema(fields: &[&str]) -> Schema { + let mut builder = Schema::builder(); + for name in fields { + builder.add_text_field(name, tantivy::schema::TEXT); + } + builder.build() + } + + /// Collect the distinct tantivy fields referenced by a query's terms. + fn collect_query_fields(query: &dyn Query) -> HashSet { + let mut fields = HashSet::new(); + query.query_terms(&mut |term, _| { + fields.insert(term.field()); + }); + fields + } + + #[test] + fn test_to_tantivy_query_all_fields_present() { + // Schema has both A and B. + let schema = build_tantivy_schema(&["A", "B"]); + let field_a = schema.get_field("A").unwrap(); + let field_b = schema.get_field("B").unwrap(); + + let mut cond = IndexCondition::new(); + cond.add_condition(Condition::Equal("A".into(), "a".into())); + cond.add_condition(Condition::Equal("B".into(), "b".into())); + + let (query, has_skipped) = cond + .to_tantivy_query(schema, None) + .expect("query build should succeed"); + assert!(!has_skipped, "no field is missing, should not skip"); + + let referenced = collect_query_fields(query.as_ref()); + assert!(referenced.contains(&field_a)); + assert!(referenced.contains(&field_b)); + } + + #[test] + fn test_to_tantivy_query_missing_field_is_skipped() { + // Schema only has A; condition references B (newly added index field, + // no historical data in this tantivy file). + let schema = build_tantivy_schema(&["A"]); + let field_a = schema.get_field("A").unwrap(); + + let mut cond = IndexCondition::new(); + cond.add_condition(Condition::Equal("A".into(), "a".into())); + cond.add_condition(Condition::Equal("B".into(), "b".into())); + + let (query, has_skipped) = cond + .to_tantivy_query(schema, None) + .expect("query build should succeed even when a field is missing"); + assert!(has_skipped, "missing field B should be reported as skipped"); + + let referenced = collect_query_fields(query.as_ref()); + // Only A should be referenced; B was dropped. + assert_eq!(referenced.len(), 1); + assert!(referenced.contains(&field_a)); + } + + #[test] + fn test_to_tantivy_query_all_fields_missing_returns_error() { + // Schema has no matching fields for any condition. + // Should return an error so the caller falls back to parquet scan. + let schema = build_tantivy_schema(&["other_field"]); + let mut cond = IndexCondition::new(); + cond.add_condition(Condition::Equal("A".into(), "a".into())); + cond.add_condition(Condition::Equal("B".into(), "b".into())); + + let result = cond.to_tantivy_query(schema, None); + assert!( + result.is_err(), + "should return error when all fields are missing" + ); + } + + #[test] + fn test_to_tantivy_query_single_condition_missing_field() { + // Single condition whose field is missing — should return an error. + let schema = build_tantivy_schema(&["A"]); + let mut cond = IndexCondition::new(); + cond.add_condition(Condition::Equal("B".into(), "b".into())); + + let result = cond.to_tantivy_query(schema, None); + assert!( + result.is_err(), + "should return error when the only field is missing" + ); + } }