mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-05 08:09:22 +00:00
This commit is contained in:
@@ -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<IndexCondition>,
|
||||
idx_optimize_rule: Option<IndexOptimizeMode>,
|
||||
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
|
||||
|
||||
+123
-11
@@ -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<Field>,
|
||||
) -> anyhow::Result<Box<dyn Query>> {
|
||||
let queries = self
|
||||
.conditions
|
||||
.iter()
|
||||
.map(|condition| {
|
||||
condition
|
||||
.to_tantivy_query(&schema, default_field)
|
||||
.map(|condition| (Occur::Must, condition))
|
||||
})
|
||||
.collect::<anyhow::Result<Vec<_>>>()?;
|
||||
Ok(Box::new(BooleanQuery::from(queries)))
|
||||
) -> anyhow::Result<(Box<dyn Query>, bool)> {
|
||||
let mut has_skipped = false;
|
||||
let mut queries: Vec<(Occur, Box<dyn Query>)> = 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<Field> {
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user