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

@@ -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() {

src/infra/src/schema/mod.rs

@@ -549,25 +549,17 @@ pub fn get_stream_setting_bloom_filter_fields<T: std::borrow::Borrow<StreamSetti
 pub fn get_stream_setting_index_updated_at_for_fields<T: std::borrow::Borrow<StreamSettings>>(
     settings: &Option<T>,
     stream_type: StreamType,
-    created_at: Option<i64>,
     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::<StreamSettings>,
             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::<StreamSettings>,
             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,
         );

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);
+        }
+    }
 }

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<String> = settings.index_fields.iter().cloned().collect();
 
-    let fts_fields: HashSet<String> = settings.fts_fields.iter().cloned().collect();
+    // files written before a field joined FTS have none of its terms in `_all`
+    let fts_fields: HashSet<String> = 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<dyn ExecutionPlan>, 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<dyn ExecutionPlan>, 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<i64>,
+        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<dyn ExecutionPlan>) -> 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'";

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<StreamSettings>,
     pub(super) stream_type: StreamType,
-    pub(super) created_at: Option<i64>,
     pub(super) fts_fields: Vec<String>,
     pub(super) index_fields: Vec<String>,
     pub(super) bloom_fields: Vec<String>,
@@ -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<Item = &String> {
+        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.

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",
+    )