fix(alerts): resolve each joined table by its own stream type [v0.40.0] (#15642)

src/handler/http/request/search/utils.rs

@@ -127,59 +127,57 @@ pub async fn validate_query_fields(
     stream_type: StreamType,
     sql: &str,
 ) -> Result<(), Error> {
-    // Step 1: Parse SQL to get columns (lightweight parsing)
     let search_query = proto::cluster_rpc::SearchQuery {
         sql: sql.to_string(),
         ..Default::default()
     };
-
     let sql = Sql::new_with_options(&search_query, org_id, stream_type, None, false).await?;
 
-    // Step 2: Resolve which stream and fields to validate.
-    //
-    // For multi-stream CTE queries sql.stream_names lists every real stream referenced
-    // (e.g. ["default", "k8s_events"]). Each stream has its own set of fields in
-    // sql.columns. Look up the entry matching stream_name so that fields belonging to
-    // a different stream are not incorrectly checked against stream_name's schema.
-    let (target_stream, target_fields) = sql
+    // A bare stream name can refer to multiple types within the same query.
+    let mut targets: Vec<_> = sql
         .stream_names
         .iter()
-        .find(|r| r.stream_name() == stream_name)
+        .filter(|r| r.stream_name() == stream_name)
         .map(|r| {
-            let fields = sql.columns.get(r).cloned().unwrap_or_default();
-            (stream_name.to_string(), fields)
+            (
+                r.get_stream_type(stream_type),
+                sql.columns.get(r).cloned().unwrap_or_default(),
+            )
         })
-        .unwrap_or_else(|| {
-            // stream_name not found in sql.stream_names (aliased or unresolved table).
-            // Fall back to original behaviour: all tracked columns against stream_name.
-            let fields = sql.columns.values().flatten().cloned().collect();
-            (stream_name.to_string(), fields)
-        });
+        .collect();
+    if targets.is_empty() {
+        targets.push((
+            stream_type,
+            sql.columns.values().flatten().cloned().collect(),
+        ));
+    }
 
-    // Step 3: Validate target stream's fields against its own schema and UDS.
-    let schema = infra::schema::get(org_id, &target_stream, stream_type)
-        .await
-        .map_err(|_| Error::ErrorCode(ErrorCodes::SearchStreamNotFound(target_stream.clone())))?;
+    for (target_stream_type, target_fields) in targets {
+        let schema = infra::schema::get(org_id, stream_name, target_stream_type)
+            .await
+            .map_err(|_| {
+                Error::ErrorCode(ErrorCodes::SearchStreamNotFound(stream_name.to_string()))
+            })?;
+        let settings = infra::schema::get_settings(org_id, stream_name, target_stream_type).await;
+        let uds_fields = infra::schema::get_stream_setting_defined_schema_fields(&settings);
 
-    let settings = infra::schema::get_settings(org_id, &target_stream, stream_type).await;
-    let uds_fields = infra::schema::get_stream_setting_defined_schema_fields(&settings);
+        for field in target_fields {
+            if is_system_field(&field) {
+                continue;
+            }
 
-    for field in target_fields {
-        if is_system_field(&field) {
-            continue;
-        }
+            if schema.field_with_name(&field).is_err() {
+                return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!(
+                    "{}. Field not found in stream schema.",
+                    field
+                ))));
+            }
 
-        if schema.field_with_name(&field).is_err() {
-            return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!(
-                "{}. Field not found in stream schema.",
-                field
-            ))));
-        }
-
-        if !uds_fields.is_empty() && !uds_fields.contains(&field) {
-            return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!(
-                "{field}. Field exists but not in User-Defined Schema (UDS)",
-            ))));
+            if !uds_fields.is_empty() && !uds_fields.contains(&field) {
+                return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!(
+                    "{field}. Field exists but not in User-Defined Schema (UDS)",
+                ))));
+            }
         }
     }
 
@@ -380,6 +378,135 @@ JOIN "test1" AS b"#;
             err_msg
         );
     }
+
+    #[tokio::test]
+    async fn test_validate_query_fields_join_enrichment_table() {
+        use arrow_schema::{DataType, Field, Schema};
+        use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache};
+
+        let org_id = "test_org_join_enrich";
+        let logs_key = format!("{org_id}/{}/app", StreamType::Logs);
+        let enrich_key = format!("{org_id}/{}/geo", StreamType::EnrichmentTables);
+        {
+            let mut w = STREAM_SCHEMAS_LATEST.write().await;
+            w.insert(
+                logs_key.clone(),
+                SchemaCache::new(Schema::new(vec![
+                    Field::new(config::TIMESTAMP_COL_NAME, DataType::Int64, false),
+                    Field::new("ip", DataType::Utf8, true),
+                ])),
+            );
+            w.insert(
+                enrich_key.clone(),
+                SchemaCache::new(Schema::new(vec![
+                    Field::new(config::TIMESTAMP_COL_NAME, DataType::Int64, false),
+                    Field::new("addr", DataType::Utf8, true),
+                    Field::new("country", DataType::Utf8, true),
+                ])),
+            );
+        }
+
+        let sql = r#"SELECT a.ip, b.country FROM "app" AS a JOIN "enrichment_tables"."geo" AS b ON a.ip = b.addr"#;
+        let logs_res = validate_query_fields(org_id, "app", StreamType::Logs, sql).await;
+        let enrich_res = validate_query_fields(org_id, "geo", StreamType::Logs, sql).await;
+
+        {
+            let mut w = STREAM_SCHEMAS_LATEST.write().await;
+            w.remove(&logs_key);
+            w.remove(&enrich_key);
+        }
+
+        assert!(logs_res.is_ok(), "logs side failed: {logs_res:?}");
+        assert!(enrich_res.is_ok(), "enrichment side failed: {enrich_res:?}");
+    }
+
+    #[tokio::test]
+    async fn test_validate_query_fields_same_name_logs_uds() {
+        check_same_name_join_uds(
+            StreamType::Logs,
+            r#"SELECT l.secret FROM "logs"."app" l JOIN "enrichment_tables"."app" e ON l.id = e.id"#,
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn test_validate_query_fields_same_name_enrichment_uds() {
+        check_same_name_join_uds(
+            StreamType::EnrichmentTables,
+            r#"SELECT e.secret FROM "app" l JOIN "enrichment_tables"."app" e ON l.id = e.id"#,
+        )
+        .await;
+    }
+
+    async fn check_same_name_join_uds(restricted_type: StreamType, sql: &str) {
+        use arrow_schema::{DataType, Field, Schema};
+        use config::meta::stream::StreamSettings;
+        use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache};
+
+        let org_id = format!("test_org_same_name_{restricted_type}_uds");
+        let logs_key = format!("{org_id}/{}/app", StreamType::Logs);
+        let enrich_key = format!("{org_id}/{}/app", StreamType::EnrichmentTables);
+        let schema = Schema::new(vec![
+            Field::new(config::TIMESTAMP_COL_NAME, DataType::Int64, false),
+            Field::new("id", DataType::Utf8, true),
+            Field::new("secret", DataType::Utf8, true),
+        ]);
+        {
+            let mut cache = STREAM_SCHEMAS_LATEST.write().await;
+            for (stream_type, key) in [
+                (StreamType::Logs, &logs_key),
+                (StreamType::EnrichmentTables, &enrich_key),
+            ] {
+                let settings = StreamSettings {
+                    defined_schema_fields: if stream_type == restricted_type {
+                        vec!["id".to_string()]
+                    } else {
+                        Vec::new()
+                    },
+                    ..Default::default()
+                };
+                let metadata = std::collections::HashMap::from([(
+                    "settings".to_string(),
+                    serde_json::to_string(&settings).unwrap(),
+                )]);
+                cache.insert(
+                    key.clone(),
+                    SchemaCache::new(schema.clone().with_metadata(metadata)),
+                );
+            }
+        }
+
+        let invalid = validate_query_fields(&org_id, "app", StreamType::Logs, sql).await;
+        let valid = validate_query_fields(
+            &org_id,
+            "app",
+            StreamType::Logs,
+            &sql.replace("secret", "id"),
+        )
+        .await;
+
+        {
+            let mut cache = STREAM_SCHEMAS_LATEST.write().await;
+            cache.remove(&logs_key);
+            cache.remove(&enrich_key);
+        }
+        {
+            let mut settings = infra::schema::STREAM_SETTINGS.write().await;
+            settings.remove(&logs_key);
+            settings.remove(&enrich_key);
+            infra::schema::set_stream_settings_atomic(settings.clone());
+        }
+
+        assert!(
+            matches!(
+                &invalid,
+                Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(message)))
+                    if message.contains("secret") && message.contains("User-Defined Schema")
+            ),
+            "{invalid:?}"
+        );
+        assert!(valid.is_ok(), "{valid:?}");
+    }
 }
 
 // ============================================================================

src/service/alerts/alert.rs

@@ -34,7 +34,7 @@ use config::{
         },
         folder::{DEFAULT_FOLDER, Folder, FolderType},
         search::{SearchEventContext, SearchEventType},
-        sql::resolve_stream_names,
+        sql::{TableReferenceExt, resolve_stream_names_with_type},
         stream::StreamType,
     },
     utils::{
@@ -391,8 +391,8 @@ async fn prepare_alert(
             }
 
             let sql = alert.query_condition.sql.as_ref().unwrap();
-            let stream_names = match resolve_stream_names(sql) {
-                Ok(stream_names) => stream_names,
+            let stream_refs = match resolve_stream_names_with_type(sql) {
+                Ok(stream_refs) => stream_refs,
                 Err(e) => {
                     return Err(AlertError::ResolveStreamNameError(e));
                 }
@@ -400,10 +400,12 @@ async fn prepare_alert(
 
             // SQL may contain multiple stream names, check for each stream
             // if the alert period is greater than the max query range
-            for stream in stream_names.iter() {
-                if !stream.eq(stream_name)
+            for stream_ref in stream_refs.iter() {
+                let stream = stream_ref.stream_name();
+                let ref_type = stream_ref.get_stream_type(stream_type);
+                if !(stream == stream_name && ref_type == stream_type)
                     && let Some(settings) =
-                        infra::schema::get_settings(org_id, stream, stream_type).await
+                        infra::schema::get_settings(org_id, &stream, ref_type).await
                 {
                     let max_query_range = settings.max_query_range;
                     if max_query_range > 0
@@ -412,7 +414,7 @@ async fn prepare_alert(
                     {
                         return Err(AlertError::PeriodExceedsMaxQueryRange {
                             max_query_range_hours: max_query_range,
-                            stream_name: stream_name.to_owned(),
+                            stream_name: stream,
                         });
                     }
                 }
@@ -1885,6 +1887,75 @@ mod tests {
     use super::*;
     use crate::service::alerts::{Condition, build_expr};
 
+    #[tokio::test]
+    async fn prepare_alert_checks_a_joined_stream_range_by_its_own_type() {
+        use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache};
+
+        use crate::common::infra::config::DESTINATIONS;
+
+        let org_id = "test_org_alert_save_join_range";
+        let logs_key = format!("{org_id}/{}/app", StreamType::Logs);
+        let traces_key = format!("{org_id}/{}/spans", StreamType::Traces);
+        let dest_key = format!("{org_id}/dest");
+        STREAM_SCHEMAS_LATEST.write().await.insert(
+            logs_key.clone(),
+            SchemaCache::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+                "id",
+                DataType::Utf8,
+                true,
+            )])),
+        );
+        {
+            let mut w = infra::schema::STREAM_SETTINGS.write().await;
+            w.insert(
+                traces_key.clone(),
+                config::meta::stream::StreamSettings {
+                    max_query_range: 1,
+                    ..Default::default()
+                },
+            );
+            infra::schema::set_stream_settings_atomic(
+                w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
+            );
+        }
+        DESTINATIONS.insert(
+            dest_key.clone(),
+            config::meta::destinations::Destination {
+                org_id: org_id.to_string(),
+                name: "dest".to_string(),
+                ..Default::default()
+            },
+        );
+
+        let mut alert = Alert::default();
+        alert.destinations = vec!["dest".to_string()];
+        alert.trigger_condition.frequency = 60;
+        alert.trigger_condition.period = 120;
+        alert.query_condition.query_type = QueryType::SQL;
+        alert.query_condition.sql = Some(
+            r#"SELECT a.id FROM "app" AS a JOIN "traces"."spans" AS b ON a.id = b.id"#.to_string(),
+        );
+        let res = prepare_alert(org_id, "app", "join", &mut alert, true, false).await;
+
+        STREAM_SCHEMAS_LATEST.write().await.remove(&logs_key);
+        {
+            let mut w = infra::schema::STREAM_SETTINGS.write().await;
+            w.remove(&traces_key);
+            infra::schema::set_stream_settings_atomic(
+                w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
+            );
+        }
+        DESTINATIONS.remove(&dest_key);
+
+        assert!(
+            matches!(
+                &res,
+                Err(AlertError::PeriodExceedsMaxQueryRange { stream_name, .. }) if stream_name == "spans"
+            ),
+            "{res:?}"
+        );
+    }
+
     #[test]
     fn test_format_variable_value() {
         // Test common control characters

src/service/alerts/derived_streams.rs

@@ -23,7 +23,7 @@ use config::{
         alerts::{FrequencyType, QueryType, TriggerEvalResults},
         pipeline::components::DerivedStream,
         search::{SearchEventContext, SearchEventType},
-        sql::resolve_stream_names,
+        sql::{TableReferenceExt, resolve_stream_names_with_type},
     },
     utils::sql::is_timestamp_selected,
 };
@@ -57,8 +57,8 @@ pub async fn save(
                 }
 
                 // Check the max_query_range of streams in the sql query
-                let stream_names = match resolve_stream_names(sql) {
-                    Ok(stream_names) => stream_names,
+                let stream_refs = match resolve_stream_names_with_type(sql) {
+                    Ok(stream_refs) => stream_refs,
                     Err(e) => {
                         return Err(anyhow::anyhow!(
                             "Error resolving stream names in SQL query: {e}"
@@ -66,9 +66,14 @@ pub async fn save(
                     }
                 };
                 let (org_id, stream_type) = (&derived_stream.org_id, derived_stream.stream_type);
-                for stream in stream_names.iter() {
-                    if let Some(settings) =
-                        infra::schema::get_settings(org_id, stream, stream_type).await
+                for stream_ref in stream_refs.iter() {
+                    let stream = stream_ref.stream_name();
+                    if let Some(settings) = infra::schema::get_settings(
+                        org_id,
+                        &stream,
+                        stream_ref.get_stream_type(stream_type),
+                    )
+                    .await
                     {
                         let max_query_range = settings.max_query_range;
                         if max_query_range > 0
@@ -239,3 +244,65 @@ impl DerivedStreamExt for DerivedStream {
             .await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Runs the real `save`; every rejection here happens before the DB is touched.
+    async fn save_err(mutate: impl FnOnce(&mut DerivedStream)) -> String {
+        let mut derived_stream = DerivedStream::default();
+        mutate(&mut derived_stream);
+        save(derived_stream, "p", "p1", true)
+            .await
+            .unwrap_err()
+            .to_string()
+    }
+
+    #[tokio::test]
+    async fn save_checks_a_joined_stream_range_by_its_own_type() {
+        use config::meta::stream::{StreamSettings, StreamType};
+
+        let org_id = "test_org_ds_save_join_range";
+        let logs_key = format!("{org_id}/{}/app", StreamType::Logs);
+        let traces_key = format!("{org_id}/{}/spans", StreamType::Traces);
+        let limited = StreamSettings {
+            max_query_range: 1,
+            ..Default::default()
+        };
+        {
+            let mut w = infra::schema::STREAM_SETTINGS.write().await;
+            w.insert(logs_key.clone(), StreamSettings::default());
+            w.insert(traces_key.clone(), limited);
+            infra::schema::set_stream_settings_atomic(
+                w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
+            );
+        }
+
+        let err = save_err(|d| {
+            d.org_id = org_id.to_string();
+            d.stream_type = StreamType::Logs;
+            d.query_condition.query_type = QueryType::SQL;
+            d.query_condition.sql = Some(
+                r#"SELECT a._timestamp AS _timestamp FROM "app" AS a JOIN "traces"."spans" AS b ON a.id = b.id"#
+                    .to_string(),
+            );
+            d.trigger_condition.period = 120;
+        })
+        .await;
+
+        {
+            let mut w = infra::schema::STREAM_SETTINGS.write().await;
+            w.remove(&logs_key);
+            w.remove(&traces_key);
+            infra::schema::set_stream_settings_atomic(
+                w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
+            );
+        }
+
+        assert!(
+            err.contains("max query range") && err.contains("\"spans\""),
+            "{err}"
+        );
+    }
+}

src/service/alerts/mod.rs

@@ -26,7 +26,7 @@ use config::{
         },
         cluster::RoleGroup,
         search::{SearchEventContext, SearchEventType, SqlQuery},
-        sql::resolve_stream_names,
+        sql::{TableReferenceExt, resolve_stream_names_with_type},
         stream::StreamType,
     },
     utils::{
@@ -244,30 +244,7 @@ impl QueryConditionExt for QueryCondition {
             }
         };
 
-        let stream_names = resolve_stream_names(&sql)
-            .map_err(|e| anyhow::anyhow!("Error resolving stream names in SQL query: {e}"))?;
-
-        // SQL may contain multiple stream names, check for each stream
-        // if the query period is greater than the max query range
-        for stream in stream_names.iter() {
-            if infra::schema::get_stream_schema_from_cache(org_id, stream, stream_type)
-                .await
-                .is_none()
-            {
-                return Err(anyhow::anyhow!(
-                    "Stream \"{stream}\" not found in schema, skipping alert evaluation"
-                ));
-            };
-
-            if let Some(settings) = infra::schema::get_settings(org_id, stream, stream_type).await {
-                let max_query_range = settings.max_query_range;
-                if max_query_range > 0 && trigger_condition.period > max_query_range * 60 {
-                    return Err(anyhow::anyhow!(
-                        "Query period is greater than max query range of {max_query_range} hours for stream \"{stream}\""
-                    ));
-                }
-            }
-        }
+        check_query_streams(org_id, stream_type, &sql, trigger_condition.period).await?;
 
         let mut time_diff = Duration::try_minutes(trigger_condition.period)
             .unwrap()
@@ -845,6 +822,41 @@ impl ConditionGroupExt for config::meta::alerts::ConditionItem {
     }
 }
 
+/// Checks that every stream in the alert SQL exists and allows the alert period.
+async fn check_query_streams(
+    org_id: &str,
+    stream_type: StreamType,
+    sql: &str,
+    period: i64,
+) -> Result<(), anyhow::Error> {
+    let stream_refs = resolve_stream_names_with_type(sql)
+        .map_err(|e| anyhow::anyhow!("Error resolving stream names in SQL query: {e}"))?;
+
+    for stream_ref in stream_refs.iter() {
+        let stream = stream_ref.stream_name();
+        // A JOIN may name another type, e.g. "enrichment_tables".tbl in a logs alert.
+        let stream_type = stream_ref.get_stream_type(stream_type);
+        if infra::schema::get_stream_schema_from_cache(org_id, &stream, stream_type)
+            .await
+            .is_none()
+        {
+            return Err(anyhow::anyhow!(
+                "Stream \"{stream}\" not found in schema, skipping alert evaluation"
+            ));
+        };
+
+        if let Some(settings) = infra::schema::get_settings(org_id, &stream, stream_type).await {
+            let max_query_range = settings.max_query_range;
+            if max_query_range > 0 && period > max_query_range * 60 {
+                return Err(anyhow::anyhow!(
+                    "Query period is greater than max query range of {max_query_range} hours for stream \"{stream}\""
+                ));
+            }
+        }
+    }
+    Ok(())
+}
+
 /// Evaluates a list of condition items with left-to-right logical operator application
 ///
 /// Algorithm:
@@ -2012,4 +2024,40 @@ mod tests {
 
         println!("✓ All deeply nested group tests with operator precedence passed!");
     }
+
+    #[tokio::test]
+    async fn test_check_query_streams_join_enrichment_table() {
+        use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache};
+
+        let org_id = "test_org_alert_join_enrich";
+        let logs_key = format!("{org_id}/{}/app", StreamType::Logs);
+        let enrich_key = format!("{org_id}/{}/geo", StreamType::EnrichmentTables);
+        let schema = Schema::new(vec![Field::new("ip", DataType::Utf8, true)]);
+        {
+            let mut w = STREAM_SCHEMAS_LATEST.write().await;
+            w.insert(logs_key.clone(), SchemaCache::new(schema.clone()));
+            w.insert(enrich_key.clone(), SchemaCache::new(schema));
+        }
+
+        let sql =
+            r#"SELECT a.ip FROM "app" AS a JOIN "enrichment_tables"."geo" AS b ON a.ip = b.ip"#;
+        let missing_sql =
+            r#"SELECT a.ip FROM "app" AS a JOIN "enrichment_tables"."nope" AS b ON a.ip = b.ip"#;
+        let ok = check_query_streams(org_id, StreamType::Logs, sql, 10).await;
+        let missing = check_query_streams(org_id, StreamType::Logs, missing_sql, 10).await;
+
+        {
+            let mut w = STREAM_SCHEMAS_LATEST.write().await;
+            w.remove(&logs_key);
+            w.remove(&enrich_key);
+        }
+
+        assert!(ok.is_ok(), "join with enrichment table rejected: {ok:?}");
+        assert!(
+            missing
+                .unwrap_err()
+                .to_string()
+                .contains("\"nope\" not found")
+        );
+    }
 }