From 6f03abc24554c8559ecb8e107f08147600ffb97c Mon Sep 17 00:00:00 2001 From: hengfeiyang Date: Wed, 4 Nov 2026 11:52:08 +0800 Subject: [PATCH] fix(search): address LIKE pushdown and NO_INDEX review findings - Skip the `_all` lookup for a LIKE on an FTS field added after the query start, since older files have none of its terms. - Read the NO_INDEX hint before track_total_hits wraps DISTINCT and UNION. - Pass use_index through the simple multi-stream request format. - Fall back to BASE_TIME instead of the stream's created_at when a field has no index update time. --- src/api/search/src/search/multi_streams.rs | 22 +++- src/infra/src/schema/mod.rs | 77 +++---------- src/search/src/sql/mod.rs | 36 ++++++- src/search_service/src/grpc/flight/index.rs | 101 ++++++++++++++++-- src/search_service/src/grpc/flight/scan.rs | 20 +++- .../tests/search/test_like_index_pushdown.py | 75 +++++++++---- 6 files changed, 233 insertions(+), 98 deletions(-) diff --git a/src/api/search/src/search/multi_streams.rs b/src/api/search/src/search/multi_streams.rs index 753698e56e..142569c8e4 100644 --- a/src/api/search/src/search/multi_streams.rs +++ b/src/api/search/src/search/multi_streams.rs @@ -997,6 +997,8 @@ fn parse_simple_multi_stream_request( #[derive(serde::Deserialize)] struct SimpleMultiStreamWrapper { query: SimpleMultiStreamQuery, + #[serde(default = "search::default_use_index")] + use_index: bool, } #[derive(serde::Deserialize)] struct SimpleMultiStreamQuery { @@ -1046,7 +1048,7 @@ fn parse_simple_multi_stream_request( uses_zo_fn: simple_req.query.query_fn.is_some(), query_fn: simple_req.query.query_fn, skip_wal: false, - use_index: true, + use_index: simple_req.use_index, regions: vec![], clusters: vec![], search_type: None, @@ -1571,7 +1573,23 @@ mod tests { use openobserve_api_common::extractors::Headers; use openobserve_core::auth::UserEmail; - use super::around_multi; + use super::{around_multi, parse_simple_multi_stream_request}; + + #[test] + fn test_parse_simple_multi_stream_request_use_index() { + let body = |extra: &str| { + format!( + r#"{{"query":{{"sql":["SELECT * FROM t"],"start_time":0,"end_time":1}}{extra}}}"# + ) + }; + assert!( + parse_simple_multi_stream_request(body("").as_bytes()) + .unwrap() + .use_index + ); + let req = parse_simple_multi_stream_request(body(r#","use_index":false"#).as_bytes()); + assert!(!req.unwrap().use_index); + } #[test] fn multi_stream_checks_each_query_after_decoding_it() { diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index f57ba04f6f..79c6a1f1de 100644 --- a/src/infra/src/schema/mod.rs +++ b/src/infra/src/schema/mod.rs @@ -549,25 +549,17 @@ pub fn get_stream_setting_bloom_filter_fields>( settings: &Option, stream_type: StreamType, - created_at: Option, fields: &[String], distinct_fields_updated_at: i64, ) -> i64 { if fields.is_empty() { return 0; } - let created_at = match created_at { - Some(created_at) => created_at, - None => { - log::warn!("created_at not found in schema metadata"); - Utc::now().timestamp_micros() - } - }; let default_updated_at = match settings { Some(settings) if settings.borrow().index_updated_at > 0 => { settings.borrow().index_updated_at } - _ => created_at, + _ => BASE_TIME.timestamp_micros(), }; fields .iter() @@ -1453,7 +1445,6 @@ mod tests { fn test_get_stream_setting_index_updated_at_for_distinct_fields() { use config::meta::stream::DistinctField; - let created_at = 1_700_000_000_000_000; let added_ts = 1_755_000_000_000_000; let settings = StreamSettings { index_updated_at: 1_750_000_000_000_000, @@ -1469,7 +1460,6 @@ mod tests { let result = get_stream_setting_index_updated_at_for_fields( &Some(settings.clone()), StreamType::Logs, - Some(created_at), &fields, 0, ); @@ -1480,7 +1470,6 @@ mod tests { let result = get_stream_setting_index_updated_at_for_fields( &Some(settings.clone()), StreamType::Logs, - Some(created_at), &fields, cutoff, ); @@ -1494,7 +1483,6 @@ mod tests { let result = get_stream_setting_index_updated_at_for_fields( &Some(settings_both), StreamType::Logs, - Some(created_at), &fields, 0, ); @@ -1508,7 +1496,6 @@ mod tests { let result = get_stream_setting_index_updated_at_for_fields( &Some(settings_disabled), StreamType::Logs, - Some(created_at), &fields, 0, ); @@ -1518,7 +1505,6 @@ mod tests { #[test] fn test_get_stream_setting_index_updated_at_for_fields() { let base_time = BASE_TIME.timestamp_micros(); - let created_at = 1_700_000_000_000_000; let mut stream_settings = StreamSettings { index_updated_at: 1_750_000_000_000_000, ..Default::default() @@ -1530,58 +1516,37 @@ mod tests { // no referenced fields -> no cutoff let fields = Vec::new(); - let result = get_stream_setting_index_updated_at_for_fields( - &settings, - StreamType::Logs, - Some(created_at), - &fields, - 0, - ); + let result = + get_stream_setting_index_updated_at_for_fields(&settings, StreamType::Logs, &fields, 0); assert_eq!(result, 0); // a field with its own entry uses it let fields = vec!["user_id".to_string()]; - let result = get_stream_setting_index_updated_at_for_fields( - &settings, - StreamType::Logs, - Some(created_at), - &fields, - 0, - ); + let result = + get_stream_setting_index_updated_at_for_fields(&settings, StreamType::Logs, &fields, 0); assert_eq!(result, 1_760_000_000_000_000); // a field without an entry falls back to the stream-level value let fields = vec!["level".to_string()]; - let result = get_stream_setting_index_updated_at_for_fields( - &settings, - StreamType::Logs, - Some(created_at), - &fields, - 0, - ); + let result = + get_stream_setting_index_updated_at_for_fields(&settings, StreamType::Logs, &fields, 0); assert_eq!(result, 1_750_000_000_000_000); // multiple fields use the max let fields = vec!["user_id".to_string(), "level".to_string()]; - let result = get_stream_setting_index_updated_at_for_fields( - &settings, - StreamType::Logs, - Some(created_at), - &fields, - 0, - ); + let result = + get_stream_setting_index_updated_at_for_fields(&settings, StreamType::Logs, &fields, 0); assert_eq!(result, 1_760_000_000_000_000); - // None settings falls back to created_at + // None settings falls back to BASE_TIME let fields = vec!["user_id".to_string()]; let result = get_stream_setting_index_updated_at_for_fields( &None::, StreamType::Logs, - Some(created_at), &fields, 0, ); - assert_eq!(result, created_at); + assert_eq!(result, base_time); // default-enabled index fields always resolve to BASE_TIME, even with an entry stream_settings @@ -1589,19 +1554,13 @@ mod tests { .insert("trace_id".to_string(), 1_760_000_000_000_000); let settings = Some(stream_settings); let fields = vec!["trace_id".to_string()]; - let result = get_stream_setting_index_updated_at_for_fields( - &settings, - StreamType::Logs, - Some(created_at), - &fields, - 0, - ); + let result = + get_stream_setting_index_updated_at_for_fields(&settings, StreamType::Logs, &fields, 0); assert_eq!(result, base_time); let fields = vec!["service_name".to_string()]; let result = get_stream_setting_index_updated_at_for_fields( &None::, StreamType::Logs, - Some(created_at), &fields, 0, ); @@ -1609,13 +1568,8 @@ mod tests { // a default field combined with a configured field still uses the max let fields = vec!["trace_id".to_string(), "user_id".to_string()]; - let result = get_stream_setting_index_updated_at_for_fields( - &settings, - StreamType::Logs, - Some(created_at), - &fields, - 0, - ); + let result = + get_stream_setting_index_updated_at_for_fields(&settings, StreamType::Logs, &fields, 0); assert_eq!(result, 1_760_000_000_000_000); // metrics streams have no built-in default index fields, so the field's own entry wins @@ -1623,7 +1577,6 @@ mod tests { let result = get_stream_setting_index_updated_at_for_fields( &settings, StreamType::Metrics, - Some(created_at), &fields, 0, ); diff --git a/src/search/src/sql/mod.rs b/src/search/src/sql/mod.rs index 6bc92c5a3a..b03d4bf8c7 100644 --- a/src/search/src/sql/mod.rs +++ b/src/search/src/sql/mod.rs @@ -196,6 +196,8 @@ impl Sql { "only SELECT queries are supported".to_string(), ))); } + // read before track_total_hits wraps DISTINCT and UNION in an unhinted outer SELECT + let no_index_hint = has_no_index_hint(&statement); //********************Change the sql start*********************************// // 2. rewrite track_total_hits @@ -330,7 +332,6 @@ impl Sql { let final_schemas = finalize_schemas(&used_schemas); let is_complex = is_complex_query_stmt(&statement); - let no_index_hint = has_no_index_hint(&statement); Ok(Sql { sql: statement.to_string(), @@ -632,4 +633,37 @@ mod tests { .unwrap(); assert!(parse_no_index_hint(&statement.to_string())); } + + #[tokio::test] + async fn test_no_index_hint_with_track_total_hits() { + let schema = arrow_schema::Schema::new(vec![ + arrow_schema::Field::new("_timestamp", arrow_schema::DataType::Int64, false), + arrow_schema::Field::new("log", arrow_schema::DataType::Utf8, true), + ]); + infra::db_init().await.unwrap(); + infra::schema::merge( + "no_index_hint_test", + "t", + StreamType::Logs, + &schema, + Some(1752660674351000), + ) + .await + .unwrap(); + for sql in [ + "SELECT /*+ NO_INDEX */ DISTINCT log FROM t", + "SELECT /*+ NO_INDEX */ log FROM t UNION ALL SELECT log FROM t", + ] { + let query = proto::cluster_rpc::SearchQuery { + sql: sql.to_string(), + size: 10, + track_total_hits: true, + ..Default::default() + }; + let sql = Sql::new(&query, "no_index_hint_test", StreamType::Logs, None) + .await + .unwrap(); + assert!(sql.no_index_hint, "{}", sql.sql); + } + } } diff --git a/src/search_service/src/grpc/flight/index.rs b/src/search_service/src/grpc/flight/index.rs index fff466b269..9e378e35a1 100644 --- a/src/search_service/src/grpc/flight/index.rs +++ b/src/search_service/src/grpc/flight/index.rs @@ -82,7 +82,11 @@ pub(super) fn optimizer_physical_plan( let index_optimizer_rule_ref = Arc::new(Mutex::new(index_mode)); let index_fields: HashSet = settings.index_fields.iter().cloned().collect(); - let fts_fields: HashSet = settings.fts_fields.iter().cloned().collect(); + // files written before a field joined FTS have none of its terms in `_all` + let fts_fields: HashSet = settings + .fts_fields_indexed_since(time_range.0) + .cloned() + .collect(); let index_rule = IndexRule::new( index_fields.clone(), fts_fields, @@ -170,6 +174,7 @@ mod tests { use std::sync::Arc; use arrow_schema::{DataType, Field, Schema}; + use config::{meta::stream::StreamSettings, utils::time::BASE_TIME}; use datafusion::{ execution::{SessionStateBuilder, runtime_env::RuntimeEnvBuilder}, prelude::SessionConfig, @@ -250,6 +255,25 @@ mod tests { } async fn plan_with_index(sql: &str, use_index: bool) -> (Arc, IndexPlan) { + let settings = StreamSearchSettings { + index_fields: vec!["service".to_string()], + fts_fields: vec!["message".to_string()], + ..Default::default() + }; + let index_info = IndexInfo { + skip_index: !use_index, + ..Default::default() + }; + let start_time = BASE_TIME.timestamp_micros(); + plan_with_settings(sql, (start_time, start_time + 100), &settings, &index_info).await + } + + async fn plan_with_settings( + sql: &str, + time_range: (i64, i64), + settings: &StreamSearchSettings, + index_info: &IndexInfo, + ) -> (Arc, IndexPlan) { let schema = Arc::new(Schema::new(vec![ Field::new("_timestamp", DataType::Int64, false), Field::new("service", DataType::Utf8, false), @@ -269,20 +293,32 @@ mod tests { &ctx, &schema, StreamType::Logs, - (0, 100), - &StreamSearchSettings { - index_fields: vec!["service".to_string()], - fts_fields: vec!["message".to_string()], - ..Default::default() - }, - &IndexInfo { - skip_index: !use_index, - ..Default::default() - }, + time_range, + settings, + index_info, ) .unwrap() } + fn settings_with_fts_added_at( + field_added_at: Option, + index_updated_at: i64, + ) -> StreamSearchSettings { + StreamSearchSettings { + settings: Some(StreamSettings { + index_updated_at, + index_fields_updated_at: field_added_at + .map(|added_at| ("message".to_string(), added_at)) + .into_iter() + .collect(), + ..Default::default() + }), + index_fields: vec!["service".to_string()], + fts_fields: vec!["message".to_string()], + ..Default::default() + } + } + fn plan_text(plan: &Arc) -> String { datafusion::physical_plan::displayable(plan.as_ref()) .indent(false) @@ -311,6 +347,49 @@ mod tests { assert!(plan_text(&plan).contains("LIKE %timeout%")); } + #[tokio::test] + async fn test_optimizer_physical_plan_like_needs_fts_field_older_than_query() { + let sql = "SELECT count(*) FROM default WHERE message LIKE '%timeout%'"; + let start = BASE_TIME.timestamp_micros() + 100; + for (field_added_at, index_updated_at, uses_index) in [ + (None, 0, true), + (Some(start - 1), 0, true), + (Some(start), 0, true), + (Some(start + 1), 0, false), + (None, start, true), + (None, start + 1, false), + (Some(start), start + 1, true), + (Some(start + 1), start, false), + ] { + let settings = settings_with_fts_added_at(field_added_at, index_updated_at); + let (plan, index) = + plan_with_settings(sql, (start, start + 100), &settings, &IndexInfo::default()) + .await; + assert_eq!( + index.condition.is_some(), + uses_index, + "{field_added_at:?} {index_updated_at}" + ); + assert!(plan_text(&plan).contains("LIKE %timeout%")); + } + } + + #[tokio::test] + async fn test_optimizer_physical_plan_late_fts_field_keeps_other_conditions() { + let sql = "SELECT * FROM default WHERE service = 'auth' AND message LIKE '%timeout%'"; + let start = BASE_TIME.timestamp_micros() + 100; + let settings = settings_with_fts_added_at(Some(start + 1), 0); + let (plan, index) = + plan_with_settings(sql, (start, start + 100), &settings, &IndexInfo::default()).await; + assert_eq!( + index.condition, + Some(IndexCondition { + conditions: vec![Condition::Equal("service".to_string(), "auth".to_string(),)], + }) + ); + assert!(plan_text(&plan).contains("LIKE %timeout%")); + } + #[tokio::test] async fn test_optimizer_physical_plan_skips_index_when_disabled() { let sql = "SELECT count(*) FROM default WHERE service = 'auth'"; diff --git a/src/search_service/src/grpc/flight/scan.rs b/src/search_service/src/grpc/flight/scan.rs index 94c13aadf7..ba8b1377c0 100644 --- a/src/search_service/src/grpc/flight/scan.rs +++ b/src/search_service/src/grpc/flight/scan.rs @@ -33,7 +33,7 @@ use infra::{ schema::{ get_stream_setting_bloom_filter_fields, get_stream_setting_fts_fields, get_stream_setting_index_fields, get_stream_setting_index_updated_at_for_fields, - unwrap_stream_created_at, unwrap_stream_settings, + unwrap_stream_settings, }, }; use itertools::Itertools; @@ -54,7 +54,6 @@ use crate::{ pub(super) struct StreamSearchSettings { pub(super) settings: Option, pub(super) stream_type: StreamType, - pub(super) created_at: Option, pub(super) fts_fields: Vec, pub(super) index_fields: Vec, pub(super) bloom_fields: Vec, @@ -106,7 +105,6 @@ impl StreamSearchSettings { .map(|v| (v.key.to_string(), v.value.to_string())) .collect(); Self { - created_at: unwrap_stream_created_at(&db_schema), settings, stream_type: source.stream_type, fts_fields, @@ -124,7 +122,6 @@ impl StreamSearchSettings { let index_updated_at = get_stream_setting_index_updated_at_for_fields( &self.settings, self.stream_type, - self.created_at, &mode.referenced_fields(), distinct_fields_updated_at, ); @@ -143,6 +140,21 @@ impl StreamSearchSettings { index_updated_at } + + /// FTS fields whose terms are in `_all` for every row at or after `start_time`. + pub(super) fn fts_fields_indexed_since( + &self, + start_time: i64, + ) -> impl Iterator { + self.fts_fields.iter().filter(move |field| { + get_stream_setting_index_updated_at_for_fields( + &self.settings, + self.stream_type, + std::slice::from_ref(*field), + 0, + ) <= start_time + }) + } } /// Object-storage files whose aggregate is answered without a storage scan. diff --git a/tests/api-testing/tests/search/test_like_index_pushdown.py b/tests/api-testing/tests/search/test_like_index_pushdown.py index 7823269b5d..ee6d547de9 100644 --- a/tests/api-testing/tests/search/test_like_index_pushdown.py +++ b/tests/api-testing/tests/search/test_like_index_pushdown.py @@ -43,48 +43,87 @@ QUERIES = [ ] -def _count(client: OpenObserveClient, sql: str, *, use_index: bool = True) -> int: - start, end = time_window(minutes=60) - body = search_payload(sql, start_time=start, end_time=end, size=1) +def _count( + client: OpenObserveClient, + sql: str, + *, + start: int | None = None, + use_index: bool = True, +) -> int: + window_start, end = time_window(minutes=60) + body = search_payload(sql, start_time=start or window_start, end_time=end, size=1) body["use_index"] = use_index resp = client.post("_search?type=logs", json=body, raise_for_status=True) hits = resp.json().get("hits", []) return int(hits[0]["n"]) if hits else 0 +def _now_micros() -> int: + return int(time.time() * 1_000_000) + + @pytest.fixture -def seeded_stream(client: OpenObserveClient, temp_stream_name: str) -> str: +def seeded_stream(client: OpenObserveClient, temp_stream_name: str) -> tuple[str, int]: + """Returns the stream and a query start after its FTS field was added, so LIKE uses `_all`.""" stream = temp_stream_name - now = int(time.time() * 1_000_000) - records = [ - {"_timestamp": now - i * 1_000_000, "message": message, "service": service} - for i, (message, service) in enumerate(ROWS) - ] - client.streams.ingest_json(stream, records) + client.streams.ingest_json( + stream, + [{"_timestamp": _now_micros() - 60_000_000, "message": "setup", "service": "setup"}], + ) resp = client.streams.update_settings( stream, full_text_search_keys={"add": ["message"]}, index_fields={"add": ["service"]}, ) assert resp.status_code == 200, resp.text + time.sleep(1) + start = _now_micros() + records = [ + {"_timestamp": start + i, "message": message, "service": service} + for i, (message, service) in enumerate(ROWS) + ] + client.streams.ingest_json(stream, records) # moves the rows from memtable to files so the index path gets exercised where available client.put("node/flush", prefix="") wait_until( - lambda: _count(client, f'SELECT count(*) AS n FROM "{stream}"') == len(ROWS), + lambda: _count(client, f'SELECT count(*) AS n FROM "{stream}"', start=start) == len(ROWS), timeout=60, msg=f"{stream} never returned {len(ROWS)} rows", ) - return stream + return stream, start @pytest.mark.parametrize("where, matches", QUERIES, ids=[q for q, _ in QUERIES]) def test_like_matches_scan_with_and_without_index( - client: OpenObserveClient, seeded_stream: str, where: str, matches + client: OpenObserveClient, seeded_stream: tuple[str, int], where: str, matches ): + stream, start = seeded_stream expected = sum(1 for message, service in ROWS if matches(message, service)) - sql = f'SELECT count(*) AS n FROM "{seeded_stream}" WHERE {where}' - no_index_sql = f'SELECT /*+ NO_INDEX */ count(*) AS n FROM "{seeded_stream}" WHERE {where}' + sql = f'SELECT count(*) AS n FROM "{stream}" WHERE {where}' + no_index_sql = f'SELECT /*+ NO_INDEX */ count(*) AS n FROM "{stream}" WHERE {where}' - assert _count(client, sql) == expected - assert _count(client, no_index_sql) == expected - assert _count(client, sql, use_index=False) == expected + assert _count(client, sql, start=start) == expected + assert _count(client, no_index_sql, start=start) == expected + assert _count(client, sql, start=start, use_index=False) == expected + + +def test_like_finds_rows_written_before_field_joined_fts( + client: OpenObserveClient, temp_stream_name: str +): + stream = temp_stream_name + record = {"_timestamp": _now_micros(), "message": "ok", "payload_text": "timeout"} + client.streams.ingest_json(stream, [record]) + client.put("node/flush", prefix="") + # the file's `_all` must be built without the field for this to catch a miss + time.sleep(5) + resp = client.streams.update_settings( + stream, full_text_search_keys={"add": ["payload_text"]} + ) + assert resp.status_code == 200, resp.text + + sql = f"""SELECT count(*) AS n FROM "{stream}" WHERE payload_text LIKE '%timeout%'""" + wait_until( + lambda: _count(client, sql) == 1, + timeout=60, + msg=f"{stream} missed a row written before payload_text joined FTS", + )