From 283dcb1c19d1a48c2e349b9510af8d1893eb619d Mon Sep 17 00:00:00 2001
From: Eric Wang <eswang1202@gmail.com>
Date: Thu, 23 Apr 2026 16:45:19 -0400
Subject: [PATCH 1/7] Initial updated parsing utilities w/ new AST module
 (Elastic DSL).

---
 .../src/ast_parsing/extract_info.rs           | 319 ++++++++++++++++++
 .../src/ast_parsing/mod.rs                    |   2 +
 .../src/ast_parsing/query_info.rs             |  67 ++++
 .../rs/elastic_dsl_utilities/src/datemath.rs  | 206 +++++++++++
 .../rs/elastic_dsl_utilities/src/helpers.rs   |  15 +
 .../rs/elastic_dsl_utilities/src/lib.rs       |   3 +
 6 files changed, 612 insertions(+)
 create mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
 create mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
 create mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
 create mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs
 create mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs

diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
new file mode 100644
index 00000000..221e727f
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
@@ -0,0 +1,319 @@
+use opensearch_dsl::{self as dsl};
+use serde_json;
+use crate::ast_parsing::query_info::{ElasticDSLQuery, Predicate, GroupBySpec, AggregationType, FieldName, TermValue};
+use crate::helpers::{strip_keyword_suffix};
+
+
+pub fn parse_query_to_ast(query: &str) -> Option<dsl::Search> {
+    let search_request = serde_json::from_str(query).ok()?;
+    search_request
+}
+
+pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQuery> {
+    // Placeholder for walking the AST and extracting relevant information
+    // This would involve traversing the AST nodes and applying logic to determine query patterns, labels, statistics, etc.
+    let predicates = match ast.query.clone()? {
+        dsl::Query::Bool(bool_query) => {
+            // Extract information from the bool query
+            walk_bool_query_and_extract_info(&bool_query)
+        }
+        _ => {
+            // Handle other query types
+            Vec::new() // Return an empty vector of predicates for unsupported query types
+        }
+    };
+    let (target_field, aggregation_type, group_by_spec) = walk_aggregations_and_extract_info(&ast.aggs)?;
+    Some(ElasticDSLQuery::new(target_field, predicates, group_by_spec, aggregation_type))
+}
+
+pub fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Predicate> {
+    // Placeholder for walking the filter context of the AST and extracting relevant information
+    // This would involve traversing the filter nodes and applying logic to determine label filters, time ranges, etc.
+    let dsl::QueryCollection(filters) = bool_query.filter.clone();
+    let mut predicates = Vec::new();
+    for query in filters {
+        match query {
+            dsl::Query::Term(term_query) => {
+                // Extract information from the term query
+                let field = strip_keyword_suffix(&term_query.field).to_owned();
+                let Some(value) = term_query.value else {
+                    continue; // Skip if term query does not have a value
+                };
+                let Some(term_value) = map_term_to_json_value(&value) else {
+                    continue; // Skip if term query value cannot be mapped to a JSON value
+                };
+                // Process the term query information as needed
+                predicates.push(Predicate::Term { field, value: term_value });
+            }
+
+            dsl::Query::Range(range_query) => {
+                // Extract information from the range query
+                let field = strip_keyword_suffix(&range_query.field).to_owned();
+                let gte = range_query.gte.clone();
+                let lte = range_query.lte.clone();
+                // Process the range query information as needed
+                let gte_value = gte.as_ref().and_then(|gte_term| map_term_to_json_value(gte_term));
+                let lte_value = lte.as_ref().and_then(|lte_term| map_term_to_json_value(lte_term));
+                predicates.push(Predicate::Range { field, gte: gte_value, lte: lte_value });
+            }
+            _ => {
+                // Handle other query types
+                todo!()
+            }
+        }
+    }
+    predicates
+}
+
+pub fn walk_aggregations_and_extract_info(aggregations: &dsl::Aggregations) -> Option<(FieldName, AggregationType, Option<GroupBySpec>)> {
+    // Traverse the aggregations in the AST and extracting relevant information. Extract the first valid aggregation type found, along with any associated group by specifications.
+    for (_, agg) in aggregations {
+        match agg {
+            dsl::Aggregation::MultiTerms(terms_agg) => {
+                // Extract information from the terms aggregation
+                let field_names: Vec<String> = terms_agg.multi_terms.terms.iter().filter_map(|multi_term| multi_term.field.clone()).collect();
+                let field_names: Vec<String> = field_names.iter().map(|s| strip_keyword_suffix(s).to_owned()).collect();
+                if field_names.is_empty() {
+                    return None; // Return None if no valid field names are found in the multi-terms aggregation.
+                }
+                let group_by_spec = Some(GroupBySpec::Fields(field_names));
+                let (target_field, aggregation_type) = find_aggregation_info(&terms_agg.aggs.clone())?;
+                return Some((target_field, aggregation_type, group_by_spec));
+            },
+            dsl::Aggregation::Terms(terms_agg) => {
+                // Extract information from the terms aggregation
+                 if let Some(field) = terms_agg.terms.field.clone() {
+                    let field = strip_keyword_suffix(&field).to_owned();
+                    // Process the terms aggregation information as needed
+                    let group_by_spec = Some(GroupBySpec::Fields(vec![field]));
+                    let (target_field, aggregation_type) = find_aggregation_info(&terms_agg.aggs.clone())?;
+                    return Some((target_field, aggregation_type, group_by_spec));
+                }
+            }
+            other => {
+                // Handle other aggregation types
+                let (target_field, aggregation_type) = extract_aggregation_info(&other)?;
+                return Some((target_field, aggregation_type, None));
+            }
+        }
+    }
+    None // Return None if no relevant aggregation information is found
+}
+
+fn find_aggregation_info(aggregations: &dsl::Aggregations) -> Option<(FieldName, AggregationType)> {
+    // Placeholder for extracting specific information from an aggregation node
+    for (_, agg) in aggregations {
+        let (field, aggregation_type) = extract_aggregation_info(&agg)?;
+        return Some((field, aggregation_type));
+    }
+    None // Return None if no relevant aggregation information is found
+}
+
+pub fn extract_aggregation_info(agg: &dsl::Aggregation) -> Option<(FieldName, AggregationType)> {
+    // Extracts the specific aggregation type and target field from the given aggregation node, if it matches supported types (avg, sum, min, max, percentiles).
+    match agg {
+        dsl::Aggregation::Avg(avg_agg) => {
+            let field = strip_keyword_suffix(&avg_agg.avg.field).to_owned();
+            let aggregation_type = AggregationType::Avg;
+            Some((field, aggregation_type))
+        },
+        dsl::Aggregation::Sum(sum_agg) => {
+            let field = strip_keyword_suffix(&sum_agg.sum.field.clone()?).to_owned();
+            let aggregation_type = AggregationType::Sum;
+            Some((field, aggregation_type))
+        },
+        dsl::Aggregation::Min(min_agg) => {
+            let field = strip_keyword_suffix(&min_agg.min.field.clone()?).to_owned();
+            let aggregation_type = AggregationType::Min;
+            Some((field, aggregation_type))
+        },
+        dsl::Aggregation::Max(max_agg) => {
+            let field = strip_keyword_suffix(&max_agg.max.field.clone()?).to_owned();
+            let aggregation_type = AggregationType::Max;
+            Some((field, aggregation_type))
+        },
+        dsl::Aggregation::Percentiles(percentiles_agg) => {
+            let field = percentiles_agg.percentiles.field.clone();
+            let percents = percentiles_agg.percentiles.percents.clone().unwrap_or_default();
+            let aggregation_type = AggregationType::Percentiles(percents);
+            Some((field, aggregation_type))
+        },
+        _ => None, // Return None for unsupported aggregation types
+    }
+}
+
+fn map_term_to_json_value(term: &dsl::Term) -> Option<TermValue> {
+    // Placeholder for extracting field and value from a term query
+    match term {
+        dsl::Term::String(value) => {
+            let value_str = value.to_string(); // Convert the term value to a string representation
+            Some(TermValue::String(value_str))
+        },
+        dsl::Term::Float32(value) => {
+            Some(TermValue::Float(value.clone() as f64))
+        },
+        dsl::Term::Float64(value) => {
+            Some(TermValue::Float(value.clone()))
+        },
+        dsl::Term::PositiveNumber(value) => Some(TermValue::UnsignedInt(value.clone())),
+        dsl::Term::NegativeNumber(value) => Some(TermValue::Int(value.clone())),
+        dsl::Term::Boolean(value) => Some(TermValue::Boolean(value.clone()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+        use super::*;
+
+        #[test]
+        fn parse_query_to_ast_parses_valid_search_request() {
+                let query = r#"
+                {
+                    "query": {
+                        "bool": {
+                            "filter": [
+                                { "term": { "service.keyword": { "value": "frontend" } } }
+                            ]
+                        }
+                    },
+                    "aggs": {
+                        "avg_latency": { "avg": { "field": "latency_ms" } }
+                    }
+                }
+                "#;
+
+                let ast = parse_query_to_ast(query);
+                assert!(ast.is_some());
+        }
+
+        #[test]
+        fn parse_query_to_ast_returns_none_for_invalid_json() {
+                let query = r#"{ "query": { "bool": { "filter": [ } }"#;
+                assert!(parse_query_to_ast(query).is_none());
+        }
+
+        #[test]
+        fn walk_bool_query_and_extract_info_extracts_term_and_range_predicates() {
+                let bool_query = dsl::Query::bool()
+                        .filter(dsl::Query::term("service.keyword", "frontend"))
+                        .filter(dsl::Query::term("is_canary", true))
+                        .filter(dsl::Query::range("@timestamp").gte("now-30s").lte("now"));
+
+                let predicates = walk_bool_query_and_extract_info(&bool_query);
+                assert_eq!(predicates.len(), 3);
+                assert_eq!(
+                        predicates[0],
+                        Predicate::Term {
+                                field: "service".to_string(),
+                                value: TermValue::String("frontend".to_string()),
+                        }
+                );
+                assert_eq!(
+                        predicates[1],
+                        Predicate::Term {
+                                field: "is_canary".to_string(),
+                                value: TermValue::Boolean(true),
+                        }
+                );
+                assert_eq!(
+                        predicates[2],
+                        Predicate::Range {
+                                field: "@timestamp".to_string(),
+                                gte: Some(TermValue::String("now-30s".to_string())),
+                                lte: Some(TermValue::String("now".to_string())),
+                        }
+                );
+        }
+
+        #[test]
+        fn walk_aggregations_and_extract_info_extracts_terms_group_by_and_metric() {
+                let query = r#"
+                {
+                    "aggs": {
+                        "by_service": {
+                            "terms": { "field": "service.keyword" },
+                            "aggs": {
+                                "avg_latency": { "avg": { "field": "latency_ms" } }
+                            }
+                        }
+                    }
+                }
+                "#;
+                let ast = parse_query_to_ast(query).expect("query should parse");
+
+                let (target_field, agg_type, group_by) =
+                        walk_aggregations_and_extract_info(&ast.aggs).expect("aggregation info should parse");
+                assert_eq!(target_field, "latency_ms");
+                assert_eq!(agg_type, AggregationType::Avg);
+                assert_eq!(
+                        group_by,
+                        Some(GroupBySpec::Fields(vec!["service".to_string()]))
+                );
+        }
+
+        #[test]
+        fn walk_aggregations_and_extract_info_extracts_multi_terms_and_percentiles() {
+                let query = r#"
+                {
+                    "aggs": {
+                        "by_labels": {
+                            "multi_terms": {
+                                "terms": [
+                                    { "field": "service.keyword" },
+                                    { "field": "env.keyword" }
+                                ]
+                            },
+                            "aggs": {
+                                "latency_percentiles": {
+                                    "percentiles": {
+                                        "field": "latency_ms",
+                                        "percents": [50.0, 95.0]
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+                "#;
+                let ast = parse_query_to_ast(query).expect("query should parse");
+
+                let (target_field, agg_type, group_by) =
+                        walk_aggregations_and_extract_info(&ast.aggs).expect("aggregation info should parse");
+                assert_eq!(target_field, "latency_ms");
+                assert_eq!(
+                        agg_type,
+                        AggregationType::Percentiles(vec![50.0, 95.0])
+                );
+                assert_eq!(
+                        group_by,
+                        Some(GroupBySpec::Fields(vec![
+                                "service".to_string(),
+                                "env".to_string()
+                        ]))
+                );
+        }
+
+        #[test]
+        fn walk_ast_and_extract_info_builds_elastic_dsl_query() {
+                let ast = dsl::Search::new()
+                        .query(
+                                dsl::Query::bool()
+                                        .filter(dsl::Query::term("service.keyword", "frontend"))
+                                        .filter(dsl::Query::range("@timestamp").gte("now-30s").lte("now")),
+                        )
+                        .aggregate(
+                                "by_service",
+                                dsl::Aggregation::terms("service.keyword")
+                                        .aggregate("max_latency", dsl::Aggregation::max("latency_ms")),
+                        );
+                let info = walk_ast_and_extract_info(&ast).expect("info should parse");
+
+                assert_eq!(info.target_field, "latency_ms");
+                assert_eq!(info.aggregation, AggregationType::Max);
+                assert_eq!(info.predicates.len(), 2);
+                assert_eq!(
+                        info.group_by_buckets,
+                        Some(GroupBySpec::Fields(vec!["service".to_string()]))
+                );
+        }
+}
\ No newline at end of file
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
new file mode 100644
index 00000000..77c57c6b
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
@@ -0,0 +1,2 @@
+pub mod extract_info;
+pub mod query_info;
\ No newline at end of file
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
new file mode 100644
index 00000000..8eb60f69
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
@@ -0,0 +1,67 @@
+
+use serde::{Deserialize, Serialize};
+
+
+pub type FieldName = String;
+
+pub struct ElasticDSLQuery {
+    // A distilled representation of an ElasticSearch DSL query, capturing the essential logic and structure.
+    pub target_field: FieldName, // List of metrics being queried
+    pub predicates: Vec<Predicate>, // Predicates applied to the query (e.g. filters in bool.filter)
+    pub group_by_buckets: Option<GroupBySpec>, // Grouping specification if the query includes a group by clause
+    pub aggregation: AggregationType, // The statistic being computed (e.g. avg, sum, percentiles)
+}
+
+impl ElasticDSLQuery {
+    // Additional methods for processing or analyzing the query can be added here
+    
+    pub fn new(
+        target_field: FieldName,
+        predicates: Vec<Predicate>,
+        group_by_buckets: Option<GroupBySpec>,
+        aggregation: AggregationType,
+    ) -> Self {
+        Self {
+            target_field,
+            predicates,
+            group_by_buckets,
+            aggregation,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum AggregationType {
+    Avg,
+    Sum,
+    Min,
+    Max,
+    Percentiles(Vec<f64>), // List of percentiles being computed
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum Predicate {
+    Term { field: FieldName, value: TermValue },
+    Range { field: FieldName, gte: Option<TermValue>, lte: Option<TermValue> },
+    // Other predicate types can be added here (e.g. exists, wildcard, etc.)
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum TermValue {
+    String(String),
+    Float(f64),
+    Int(i64),
+    UnsignedInt(u64),
+    Boolean(bool),
+    // Other term value types can be added here (e.g. boolean, date, etc.)
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum GroupBySpec {
+    Fields(Vec<FieldName>),
+    Filters(Vec<Predicate>), // Grouping by filters (e.g. group by whether a field matches a certain value)
+}
\ No newline at end of file
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs
new file mode 100644
index 00000000..075526f9
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs
@@ -0,0 +1,206 @@
+use serde::{Deserialize, Serialize};
+use crate::ast_parsing::query_info::{FieldName, TermValue, Predicate};
+
+
+/// Time range bounds resolved into epoch milliseconds.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ResolvedTimeRange {
+    pub field: FieldName,
+    pub gte_ms: Option<i64>,
+    pub lte_ms: Option<i64>,
+}
+
+/// An optional time range applied to a timestamp field.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TimeRange {
+    pub field: FieldName,
+    pub gte: Option<String>,
+    pub lte: Option<String>,
+}
+
+impl TimeRange {
+    /// Parse a date-math expression into epoch milliseconds using the provided
+    /// `now_ms` as reference for `now`-relative expressions.
+    ///
+    /// Supported forms:
+    /// - `now`
+    /// - `now-30s`, `now+5m`, `now-1h`, `now-2d`, `now-1w`, `now-500ms`
+    /// - RFC3339 timestamps (e.g. `2026-03-22T12:34:56Z`)
+    /// - Plain integer timestamps (returned as-is)
+    pub fn parse_date_math(expr: &str, now_ms: i64) -> Option<i64> {
+        if expr == "now" {
+            return Some(now_ms);
+        }
+
+        if let Some(delta) = Self::parse_now_delta_ms(expr) {
+            return now_ms.checked_add(delta);
+        }
+
+        if let Ok(v) = expr.parse::<i64>() {
+            return Some(v);
+        }
+
+        chrono::DateTime::parse_from_rfc3339(expr)
+            .ok()
+            .map(|dt| dt.timestamp_millis())
+    }
+
+    /// Resolve `gte`/`lte` date-math strings into numeric epoch-millisecond
+    /// values relative to `now_ms`.
+    pub fn resolve_epoch_millis(&self, now_ms: i64) -> Option<ResolvedTimeRange> {
+        let gte_ms = match &self.gte {
+            Some(v) => Some(Self::parse_date_math(v, now_ms)?),
+            None => None,
+        };
+        let lte_ms = match &self.lte {
+            Some(v) => Some(Self::parse_date_math(v, now_ms)?),
+            None => None,
+        };
+
+        Some(ResolvedTimeRange {
+            field: self.field.clone(),
+            gte_ms,
+            lte_ms,
+        })
+    }
+
+    fn parse_now_delta_ms(expr: &str) -> Option<i64> {
+        let rest = expr.strip_prefix("now")?;
+        if rest.is_empty() {
+            return Some(0);
+        }
+
+        let sign_char = rest.chars().next()?;
+        let sign = match sign_char {
+            '+' => 1_i64,
+            '-' => -1_i64,
+            _ => return None,
+        };
+
+        let offset = &rest[1..];
+        if offset.is_empty() {
+            return None;
+        }
+
+        let digit_count = offset.chars().take_while(|c| c.is_ascii_digit()).count();
+        if digit_count == 0 || digit_count == offset.len() {
+            return None;
+        }
+
+        let qty = offset[..digit_count].parse::<i64>().ok()?;
+        let unit = &offset[digit_count..];
+        let unit_ms = match unit {
+            "ms" => 1_i64,
+            "s" => 1_000_i64,
+            "m" => 60_000_i64,
+            "h" => 3_600_000_i64,
+            "d" => 86_400_000_i64,
+            "w" => 604_800_000_i64,
+            _ => return None,
+        };
+
+        qty.checked_mul(unit_ms)?.checked_mul(sign)
+    }
+}
+
+pub fn range_query_to_time_range(predicate: &Predicate, now_ms: i64) -> Option<ResolvedTimeRange> {
+    match predicate {
+        Predicate::Range { field, gte, lte } => {
+            let tr = TimeRange {
+                field: field.clone(),
+                gte: gte.as_ref().and_then(|v| match v {
+                    TermValue::String(s) => Some(s.clone()),
+                    _ => None,
+                }),
+                lte: lte.as_ref().and_then(|v| match v {
+                    TermValue::String(s) => Some(s.clone()),
+                    _ => None,
+                }),
+            };
+            let resolved =tr.resolve_epoch_millis(now_ms)?;
+            Some(resolved)
+        },
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_date_math_supports_now_relative_and_numeric() {
+        let now_ms = 1_700_000_000_000_i64;
+        assert_eq!(TimeRange::parse_date_math("now", now_ms), Some(now_ms));
+        assert_eq!(
+            TimeRange::parse_date_math("now-30s", now_ms),
+            Some(now_ms - 30_000)
+        );
+        assert_eq!(
+            TimeRange::parse_date_math("now+5m", now_ms),
+            Some(now_ms + 300_000)
+        );
+        assert_eq!(
+            TimeRange::parse_date_math("1700000000123", now_ms),
+            Some(1_700_000_000_123)
+        );
+    }
+
+    #[test]
+    fn parse_date_math_supports_rfc3339() {
+        let now_ms = 0;
+        let value = TimeRange::parse_date_math("2026-03-22T12:34:56Z", now_ms)
+            .expect("RFC3339 timestamp should parse");
+        assert_eq!(value, 1_774_182_896_000);
+    }
+
+    #[test]
+    fn parse_date_math_rejects_invalid_expressions() {
+        let now_ms = 1_700_000_000_000_i64;
+        assert_eq!(TimeRange::parse_date_math("now+", now_ms), None);
+        assert_eq!(TimeRange::parse_date_math("now-10", now_ms), None);
+        assert_eq!(TimeRange::parse_date_math("yesterday", now_ms), None);
+    }
+
+    #[test]
+    fn resolve_epoch_millis_resolves_both_bounds() {
+        let range = TimeRange {
+            field: "@timestamp".to_string(),
+            gte: Some("now-1m".to_string()),
+            lte: Some("now".to_string()),
+        };
+        let now_ms = 2_000_000_i64;
+
+        let resolved = range
+            .resolve_epoch_millis(now_ms)
+            .expect("range should resolve");
+        assert_eq!(resolved.field, "@timestamp");
+        assert_eq!(resolved.gte_ms, Some(1_940_000));
+        assert_eq!(resolved.lte_ms, Some(2_000_000));
+    }
+
+    #[test]
+    fn range_query_to_time_range_converts_range_predicate() {
+        let predicate = Predicate::Range {
+            field: "@timestamp".to_string(),
+            gte: Some(TermValue::String("now-30s".to_string())),
+            lte: Some(TermValue::String("now".to_string())),
+        };
+        let now_ms = 1_000_000_i64;
+
+        let resolved = range_query_to_time_range(&predicate, now_ms)
+            .expect("range predicate should convert");
+        assert_eq!(resolved.field, "@timestamp");
+        assert_eq!(resolved.gte_ms, Some(970_000));
+        assert_eq!(resolved.lte_ms, Some(1_000_000));
+    }
+
+    #[test]
+    fn range_query_to_time_range_returns_none_for_non_range_predicates() {
+        let predicate = Predicate::Term {
+            field: "service".to_string(),
+            value: TermValue::String("frontend".to_string()),
+        };
+        assert!(range_query_to_time_range(&predicate, 100).is_none());
+    }
+}
\ No newline at end of file
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs
new file mode 100644
index 00000000..cdd5aeac
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs
@@ -0,0 +1,15 @@
+/// Strip the `.keyword` suffix from a field name, if present.
+pub fn strip_keyword_suffix(field: &str) -> &str {
+    field.strip_suffix(".keyword").unwrap_or(field)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn strip_keyword_suffix_removes_only_trailing_keyword() {
+        assert_eq!(strip_keyword_suffix("service.keyword"), "service");
+        assert_eq!(strip_keyword_suffix("env"), "env");
+    }
+}
\ No newline at end of file
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
index c03929fb..aff9be4c 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
@@ -1,6 +1,9 @@
 pub mod parsing;
 pub mod pattern;
 pub mod types;
+pub mod ast_parsing;
+pub mod datemath;
+pub mod helpers;
 
 pub use parsing::*;
 pub use pattern::{classify, parse_and_classify};

From 2accb8cc61a30ed90b1061cf3ef4d824989eb0bc Mon Sep 17 00:00:00 2001
From: Eric Wang <eswang1202@gmail.com>
Date: Thu, 23 Apr 2026 16:45:49 -0400
Subject: [PATCH 2/7] Add modded opensearch-dsl crate Cargo.toml.

---
 Cargo.lock                                            | 11 +++++++++++
 .../dependencies/rs/elastic_dsl_utilities/Cargo.toml  |  1 +
 2 files changed, 12 insertions(+)

diff --git a/Cargo.lock b/Cargo.lock
index 78b6cfd2..6344e25e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1679,6 +1679,7 @@ name = "elastic_dsl_utilities"
 version = "0.3.0"
 dependencies = [
  "chrono",
+ "opensearch-dsl",
  "serde",
  "serde_json",
 ]
@@ -3028,6 +3029,16 @@ version = "11.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
 
+[[package]]
+name = "opensearch-dsl"
+version = "0.3.1"
+dependencies = [
+ "chrono",
+ "num-traits",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "openssl"
 version = "0.10.76"
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
index 3726cabb..92e1278f 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
@@ -7,3 +7,4 @@ version.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 chrono.workspace = true
+opensearch-dsl = { version = "0.3.1", path = "../../../../../opensearch-client-rs/opensearch-dsl" }

From 86f968accc1bb79624a562e891d4bde632016031 Mon Sep 17 00:00:00 2001
From: Eric Wang <eswang1202@gmail.com>
Date: Thu, 23 Apr 2026 16:47:53 -0400
Subject: [PATCH 3/7] Cargo format.

---
 .../src/ast_parsing/extract_info.rs           | 285 ++++++++++--------
 .../src/ast_parsing/mod.rs                    |   2 +-
 .../src/ast_parsing/query_info.rs             |  19 +-
 .../rs/elastic_dsl_utilities/src/datemath.rs  |  13 +-
 .../rs/elastic_dsl_utilities/src/helpers.rs   |   2 +-
 .../rs/elastic_dsl_utilities/src/lib.rs       |   6 +-
 6 files changed, 179 insertions(+), 148 deletions(-)

diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
index 221e727f..e140af97 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
@@ -1,8 +1,9 @@
+use crate::ast_parsing::query_info::{
+    AggregationType, ElasticDSLQuery, FieldName, GroupBySpec, Predicate, TermValue,
+};
+use crate::helpers::strip_keyword_suffix;
 use opensearch_dsl::{self as dsl};
 use serde_json;
-use crate::ast_parsing::query_info::{ElasticDSLQuery, Predicate, GroupBySpec, AggregationType, FieldName, TermValue};
-use crate::helpers::{strip_keyword_suffix};
-
 
 pub fn parse_query_to_ast(query: &str) -> Option<dsl::Search> {
     let search_request = serde_json::from_str(query).ok()?;
@@ -22,8 +23,14 @@ pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQuery> {
             Vec::new() // Return an empty vector of predicates for unsupported query types
         }
     };
-    let (target_field, aggregation_type, group_by_spec) = walk_aggregations_and_extract_info(&ast.aggs)?;
-    Some(ElasticDSLQuery::new(target_field, predicates, group_by_spec, aggregation_type))
+    let (target_field, aggregation_type, group_by_spec) =
+        walk_aggregations_and_extract_info(&ast.aggs)?;
+    Some(ElasticDSLQuery::new(
+        target_field,
+        predicates,
+        group_by_spec,
+        aggregation_type,
+    ))
 }
 
 pub fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Predicate> {
@@ -43,7 +50,10 @@ pub fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Pred
                     continue; // Skip if term query value cannot be mapped to a JSON value
                 };
                 // Process the term query information as needed
-                predicates.push(Predicate::Term { field, value: term_value });
+                predicates.push(Predicate::Term {
+                    field,
+                    value: term_value,
+                });
             }
 
             dsl::Query::Range(range_query) => {
@@ -52,41 +62,61 @@ pub fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Pred
                 let gte = range_query.gte.clone();
                 let lte = range_query.lte.clone();
                 // Process the range query information as needed
-                let gte_value = gte.as_ref().and_then(|gte_term| map_term_to_json_value(gte_term));
-                let lte_value = lte.as_ref().and_then(|lte_term| map_term_to_json_value(lte_term));
-                predicates.push(Predicate::Range { field, gte: gte_value, lte: lte_value });
+                let gte_value = gte
+                    .as_ref()
+                    .and_then(|gte_term| map_term_to_json_value(gte_term));
+                let lte_value = lte
+                    .as_ref()
+                    .and_then(|lte_term| map_term_to_json_value(lte_term));
+                predicates.push(Predicate::Range {
+                    field,
+                    gte: gte_value,
+                    lte: lte_value,
+                });
             }
             _ => {
                 // Handle other query types
-                todo!()
+                continue; // Skip unsupported query types
             }
         }
     }
     predicates
 }
 
-pub fn walk_aggregations_and_extract_info(aggregations: &dsl::Aggregations) -> Option<(FieldName, AggregationType, Option<GroupBySpec>)> {
+pub fn walk_aggregations_and_extract_info(
+    aggregations: &dsl::Aggregations,
+) -> Option<(FieldName, AggregationType, Option<GroupBySpec>)> {
     // Traverse the aggregations in the AST and extracting relevant information. Extract the first valid aggregation type found, along with any associated group by specifications.
     for (_, agg) in aggregations {
         match agg {
             dsl::Aggregation::MultiTerms(terms_agg) => {
                 // Extract information from the terms aggregation
-                let field_names: Vec<String> = terms_agg.multi_terms.terms.iter().filter_map(|multi_term| multi_term.field.clone()).collect();
-                let field_names: Vec<String> = field_names.iter().map(|s| strip_keyword_suffix(s).to_owned()).collect();
+                let field_names: Vec<String> = terms_agg
+                    .multi_terms
+                    .terms
+                    .iter()
+                    .filter_map(|multi_term| multi_term.field.clone())
+                    .collect();
+                let field_names: Vec<String> = field_names
+                    .iter()
+                    .map(|s| strip_keyword_suffix(s).to_owned())
+                    .collect();
                 if field_names.is_empty() {
                     return None; // Return None if no valid field names are found in the multi-terms aggregation.
                 }
                 let group_by_spec = Some(GroupBySpec::Fields(field_names));
-                let (target_field, aggregation_type) = find_aggregation_info(&terms_agg.aggs.clone())?;
+                let (target_field, aggregation_type) =
+                    find_aggregation_info(&terms_agg.aggs.clone())?;
                 return Some((target_field, aggregation_type, group_by_spec));
-            },
+            }
             dsl::Aggregation::Terms(terms_agg) => {
                 // Extract information from the terms aggregation
-                 if let Some(field) = terms_agg.terms.field.clone() {
+                if let Some(field) = terms_agg.terms.field.clone() {
                     let field = strip_keyword_suffix(&field).to_owned();
                     // Process the terms aggregation information as needed
                     let group_by_spec = Some(GroupBySpec::Fields(vec![field]));
-                    let (target_field, aggregation_type) = find_aggregation_info(&terms_agg.aggs.clone())?;
+                    let (target_field, aggregation_type) =
+                        find_aggregation_info(&terms_agg.aggs.clone())?;
                     return Some((target_field, aggregation_type, group_by_spec));
                 }
             }
@@ -116,28 +146,32 @@ pub fn extract_aggregation_info(agg: &dsl::Aggregation) -> Option<(FieldName, Ag
             let field = strip_keyword_suffix(&avg_agg.avg.field).to_owned();
             let aggregation_type = AggregationType::Avg;
             Some((field, aggregation_type))
-        },
+        }
         dsl::Aggregation::Sum(sum_agg) => {
             let field = strip_keyword_suffix(&sum_agg.sum.field.clone()?).to_owned();
             let aggregation_type = AggregationType::Sum;
             Some((field, aggregation_type))
-        },
+        }
         dsl::Aggregation::Min(min_agg) => {
             let field = strip_keyword_suffix(&min_agg.min.field.clone()?).to_owned();
             let aggregation_type = AggregationType::Min;
             Some((field, aggregation_type))
-        },
+        }
         dsl::Aggregation::Max(max_agg) => {
             let field = strip_keyword_suffix(&max_agg.max.field.clone()?).to_owned();
             let aggregation_type = AggregationType::Max;
             Some((field, aggregation_type))
-        },
+        }
         dsl::Aggregation::Percentiles(percentiles_agg) => {
             let field = percentiles_agg.percentiles.field.clone();
-            let percents = percentiles_agg.percentiles.percents.clone().unwrap_or_default();
+            let percents = percentiles_agg
+                .percentiles
+                .percents
+                .clone()
+                .unwrap_or_default();
             let aggregation_type = AggregationType::Percentiles(percents);
             Some((field, aggregation_type))
-        },
+        }
         _ => None, // Return None for unsupported aggregation types
     }
 }
@@ -148,26 +182,22 @@ fn map_term_to_json_value(term: &dsl::Term) -> Option<TermValue> {
         dsl::Term::String(value) => {
             let value_str = value.to_string(); // Convert the term value to a string representation
             Some(TermValue::String(value_str))
-        },
-        dsl::Term::Float32(value) => {
-            Some(TermValue::Float(value.clone() as f64))
-        },
-        dsl::Term::Float64(value) => {
-            Some(TermValue::Float(value.clone()))
-        },
+        }
+        dsl::Term::Float32(value) => Some(TermValue::Float(value.clone() as f64)),
+        dsl::Term::Float64(value) => Some(TermValue::Float(value.clone())),
         dsl::Term::PositiveNumber(value) => Some(TermValue::UnsignedInt(value.clone())),
         dsl::Term::NegativeNumber(value) => Some(TermValue::Int(value.clone())),
-        dsl::Term::Boolean(value) => Some(TermValue::Boolean(value.clone()))
+        dsl::Term::Boolean(value) => Some(TermValue::Boolean(value.clone())),
     }
 }
 
 #[cfg(test)]
 mod tests {
-        use super::*;
+    use super::*;
 
-        #[test]
-        fn parse_query_to_ast_parses_valid_search_request() {
-                let query = r#"
+    #[test]
+    fn parse_query_to_ast_parses_valid_search_request() {
+        let query = r#"
                 {
                     "query": {
                         "bool": {
@@ -182,52 +212,52 @@ mod tests {
                 }
                 "#;
 
-                let ast = parse_query_to_ast(query);
-                assert!(ast.is_some());
-        }
+        let ast = parse_query_to_ast(query);
+        assert!(ast.is_some());
+    }
 
-        #[test]
-        fn parse_query_to_ast_returns_none_for_invalid_json() {
-                let query = r#"{ "query": { "bool": { "filter": [ } }"#;
-                assert!(parse_query_to_ast(query).is_none());
-        }
+    #[test]
+    fn parse_query_to_ast_returns_none_for_invalid_json() {
+        let query = r#"{ "query": { "bool": { "filter": [ } }"#;
+        assert!(parse_query_to_ast(query).is_none());
+    }
 
-        #[test]
-        fn walk_bool_query_and_extract_info_extracts_term_and_range_predicates() {
-                let bool_query = dsl::Query::bool()
-                        .filter(dsl::Query::term("service.keyword", "frontend"))
-                        .filter(dsl::Query::term("is_canary", true))
-                        .filter(dsl::Query::range("@timestamp").gte("now-30s").lte("now"));
+    #[test]
+    fn walk_bool_query_and_extract_info_extracts_term_and_range_predicates() {
+        let bool_query = dsl::Query::bool()
+            .filter(dsl::Query::term("service.keyword", "frontend"))
+            .filter(dsl::Query::term("is_canary", true))
+            .filter(dsl::Query::range("@timestamp").gte("now-30s").lte("now"));
 
-                let predicates = walk_bool_query_and_extract_info(&bool_query);
-                assert_eq!(predicates.len(), 3);
-                assert_eq!(
-                        predicates[0],
-                        Predicate::Term {
-                                field: "service".to_string(),
-                                value: TermValue::String("frontend".to_string()),
-                        }
-                );
-                assert_eq!(
-                        predicates[1],
-                        Predicate::Term {
-                                field: "is_canary".to_string(),
-                                value: TermValue::Boolean(true),
-                        }
-                );
-                assert_eq!(
-                        predicates[2],
-                        Predicate::Range {
-                                field: "@timestamp".to_string(),
-                                gte: Some(TermValue::String("now-30s".to_string())),
-                                lte: Some(TermValue::String("now".to_string())),
-                        }
-                );
-        }
+        let predicates = walk_bool_query_and_extract_info(&bool_query);
+        assert_eq!(predicates.len(), 3);
+        assert_eq!(
+            predicates[0],
+            Predicate::Term {
+                field: "service".to_string(),
+                value: TermValue::String("frontend".to_string()),
+            }
+        );
+        assert_eq!(
+            predicates[1],
+            Predicate::Term {
+                field: "is_canary".to_string(),
+                value: TermValue::Boolean(true),
+            }
+        );
+        assert_eq!(
+            predicates[2],
+            Predicate::Range {
+                field: "@timestamp".to_string(),
+                gte: Some(TermValue::String("now-30s".to_string())),
+                lte: Some(TermValue::String("now".to_string())),
+            }
+        );
+    }
 
-        #[test]
-        fn walk_aggregations_and_extract_info_extracts_terms_group_by_and_metric() {
-                let query = r#"
+    #[test]
+    fn walk_aggregations_and_extract_info_extracts_terms_group_by_and_metric() {
+        let query = r#"
                 {
                     "aggs": {
                         "by_service": {
@@ -239,21 +269,21 @@ mod tests {
                     }
                 }
                 "#;
-                let ast = parse_query_to_ast(query).expect("query should parse");
+        let ast = parse_query_to_ast(query).expect("query should parse");
 
-                let (target_field, agg_type, group_by) =
-                        walk_aggregations_and_extract_info(&ast.aggs).expect("aggregation info should parse");
-                assert_eq!(target_field, "latency_ms");
-                assert_eq!(agg_type, AggregationType::Avg);
-                assert_eq!(
-                        group_by,
-                        Some(GroupBySpec::Fields(vec!["service".to_string()]))
-                );
-        }
+        let (target_field, agg_type, group_by) =
+            walk_aggregations_and_extract_info(&ast.aggs).expect("aggregation info should parse");
+        assert_eq!(target_field, "latency_ms");
+        assert_eq!(agg_type, AggregationType::Avg);
+        assert_eq!(
+            group_by,
+            Some(GroupBySpec::Fields(vec!["service".to_string()]))
+        );
+    }
 
-        #[test]
-        fn walk_aggregations_and_extract_info_extracts_multi_terms_and_percentiles() {
-                let query = r#"
+    #[test]
+    fn walk_aggregations_and_extract_info_extracts_multi_terms_and_percentiles() {
+        let query = r#"
                 {
                     "aggs": {
                         "by_labels": {
@@ -275,45 +305,42 @@ mod tests {
                     }
                 }
                 "#;
-                let ast = parse_query_to_ast(query).expect("query should parse");
+        let ast = parse_query_to_ast(query).expect("query should parse");
 
-                let (target_field, agg_type, group_by) =
-                        walk_aggregations_and_extract_info(&ast.aggs).expect("aggregation info should parse");
-                assert_eq!(target_field, "latency_ms");
-                assert_eq!(
-                        agg_type,
-                        AggregationType::Percentiles(vec![50.0, 95.0])
-                );
-                assert_eq!(
-                        group_by,
-                        Some(GroupBySpec::Fields(vec![
-                                "service".to_string(),
-                                "env".to_string()
-                        ]))
-                );
-        }
+        let (target_field, agg_type, group_by) =
+            walk_aggregations_and_extract_info(&ast.aggs).expect("aggregation info should parse");
+        assert_eq!(target_field, "latency_ms");
+        assert_eq!(agg_type, AggregationType::Percentiles(vec![50.0, 95.0]));
+        assert_eq!(
+            group_by,
+            Some(GroupBySpec::Fields(vec![
+                "service".to_string(),
+                "env".to_string()
+            ]))
+        );
+    }
 
-        #[test]
-        fn walk_ast_and_extract_info_builds_elastic_dsl_query() {
-                let ast = dsl::Search::new()
-                        .query(
-                                dsl::Query::bool()
-                                        .filter(dsl::Query::term("service.keyword", "frontend"))
-                                        .filter(dsl::Query::range("@timestamp").gte("now-30s").lte("now")),
-                        )
-                        .aggregate(
-                                "by_service",
-                                dsl::Aggregation::terms("service.keyword")
-                                        .aggregate("max_latency", dsl::Aggregation::max("latency_ms")),
-                        );
-                let info = walk_ast_and_extract_info(&ast).expect("info should parse");
+    #[test]
+    fn walk_ast_and_extract_info_builds_elastic_dsl_query() {
+        let ast = dsl::Search::new()
+            .query(
+                dsl::Query::bool()
+                    .filter(dsl::Query::term("service.keyword", "frontend"))
+                    .filter(dsl::Query::range("@timestamp").gte("now-30s").lte("now")),
+            )
+            .aggregate(
+                "by_service",
+                dsl::Aggregation::terms("service.keyword")
+                    .aggregate("max_latency", dsl::Aggregation::max("latency_ms")),
+            );
+        let info = walk_ast_and_extract_info(&ast).expect("info should parse");
 
-                assert_eq!(info.target_field, "latency_ms");
-                assert_eq!(info.aggregation, AggregationType::Max);
-                assert_eq!(info.predicates.len(), 2);
-                assert_eq!(
-                        info.group_by_buckets,
-                        Some(GroupBySpec::Fields(vec!["service".to_string()]))
-                );
-        }
-}
\ No newline at end of file
+        assert_eq!(info.target_field, "latency_ms");
+        assert_eq!(info.aggregation, AggregationType::Max);
+        assert_eq!(info.predicates.len(), 2);
+        assert_eq!(
+            info.group_by_buckets,
+            Some(GroupBySpec::Fields(vec!["service".to_string()]))
+        );
+    }
+}
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
index 77c57c6b..ad07eed3 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
@@ -1,2 +1,2 @@
 pub mod extract_info;
-pub mod query_info;
\ No newline at end of file
+pub mod query_info;
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
index 8eb60f69..63b5cb51 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
@@ -1,12 +1,10 @@
-
 use serde::{Deserialize, Serialize};
 
-
 pub type FieldName = String;
 
 pub struct ElasticDSLQuery {
     // A distilled representation of an ElasticSearch DSL query, capturing the essential logic and structure.
-    pub target_field: FieldName, // List of metrics being queried
+    pub target_field: FieldName,    // List of metrics being queried
     pub predicates: Vec<Predicate>, // Predicates applied to the query (e.g. filters in bool.filter)
     pub group_by_buckets: Option<GroupBySpec>, // Grouping specification if the query includes a group by clause
     pub aggregation: AggregationType, // The statistic being computed (e.g. avg, sum, percentiles)
@@ -14,7 +12,7 @@ pub struct ElasticDSLQuery {
 
 impl ElasticDSLQuery {
     // Additional methods for processing or analyzing the query can be added here
-    
+
     pub fn new(
         target_field: FieldName,
         predicates: Vec<Predicate>,
@@ -43,8 +41,15 @@ pub enum AggregationType {
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum Predicate {
-    Term { field: FieldName, value: TermValue },
-    Range { field: FieldName, gte: Option<TermValue>, lte: Option<TermValue> },
+    Term {
+        field: FieldName,
+        value: TermValue,
+    },
+    Range {
+        field: FieldName,
+        gte: Option<TermValue>,
+        lte: Option<TermValue>,
+    },
     // Other predicate types can be added here (e.g. exists, wildcard, etc.)
 }
 
@@ -64,4 +69,4 @@ pub enum TermValue {
 pub enum GroupBySpec {
     Fields(Vec<FieldName>),
     Filters(Vec<Predicate>), // Grouping by filters (e.g. group by whether a field matches a certain value)
-}
\ No newline at end of file
+}
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs
index 075526f9..64f2cf87 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/datemath.rs
@@ -1,6 +1,5 @@
+use crate::ast_parsing::query_info::{FieldName, Predicate, TermValue};
 use serde::{Deserialize, Serialize};
-use crate::ast_parsing::query_info::{FieldName, TermValue, Predicate};
-
 
 /// Time range bounds resolved into epoch milliseconds.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -117,9 +116,9 @@ pub fn range_query_to_time_range(predicate: &Predicate, now_ms: i64) -> Option<R
                     _ => None,
                 }),
             };
-            let resolved =tr.resolve_epoch_millis(now_ms)?;
+            let resolved = tr.resolve_epoch_millis(now_ms)?;
             Some(resolved)
-        },
+        }
         _ => None,
     }
 }
@@ -188,8 +187,8 @@ mod tests {
         };
         let now_ms = 1_000_000_i64;
 
-        let resolved = range_query_to_time_range(&predicate, now_ms)
-            .expect("range predicate should convert");
+        let resolved =
+            range_query_to_time_range(&predicate, now_ms).expect("range predicate should convert");
         assert_eq!(resolved.field, "@timestamp");
         assert_eq!(resolved.gte_ms, Some(970_000));
         assert_eq!(resolved.lte_ms, Some(1_000_000));
@@ -203,4 +202,4 @@ mod tests {
         };
         assert!(range_query_to_time_range(&predicate, 100).is_none());
     }
-}
\ No newline at end of file
+}
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs
index cdd5aeac..c2cf7a2c 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/helpers.rs
@@ -12,4 +12,4 @@ mod tests {
         assert_eq!(strip_keyword_suffix("service.keyword"), "service");
         assert_eq!(strip_keyword_suffix("env"), "env");
     }
-}
\ No newline at end of file
+}
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
index aff9be4c..1e504b4e 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
@@ -1,9 +1,9 @@
-pub mod parsing;
-pub mod pattern;
-pub mod types;
 pub mod ast_parsing;
 pub mod datemath;
 pub mod helpers;
+pub mod parsing;
+pub mod pattern;
+pub mod types;
 
 pub use parsing::*;
 pub use pattern::{classify, parse_and_classify};

From 8059342494ea15403e290feec047a6cdd6ed1140 Mon Sep 17 00:00:00 2001
From: Eric Wang <eswang1202@gmail.com>
Date: Thu, 23 Apr 2026 18:31:15 -0400
Subject: [PATCH 4/7] Export updated DSL utilities and remove old code. Rename
 ElasticDSLQuery -> ElasticDSLQueryInfo to reflect its role as query metadata
 carrier.

---
 .../src/ast_parsing/extract_info.rs           |  20 +-
 .../src/ast_parsing/mod.rs                    |   3 +
 .../src/ast_parsing/query_info.rs             |   5 +-
 .../rs/elastic_dsl_utilities/src/lib.rs       |   9 +-
 .../rs/elastic_dsl_utilities/src/parsing.rs   | 439 ------------------
 .../rs/elastic_dsl_utilities/src/pattern.rs   | 403 ----------------
 .../rs/elastic_dsl_utilities/src/types.rs     | 286 ------------
 7 files changed, 22 insertions(+), 1143 deletions(-)
 delete mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/parsing.rs
 delete mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/pattern.rs
 delete mode 100644 asap-common/dependencies/rs/elastic_dsl_utilities/src/types.rs

diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
index e140af97..59835d35 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
@@ -1,17 +1,23 @@
 use crate::ast_parsing::query_info::{
-    AggregationType, ElasticDSLQuery, FieldName, GroupBySpec, Predicate, TermValue,
+    AggregationType, ElasticDSLQueryInfo, FieldName, GroupBySpec, Predicate, TermValue,
 };
 use crate::helpers::strip_keyword_suffix;
 use opensearch_dsl::{self as dsl};
 use serde_json;
 
+pub fn extract_query_info(query: &str) -> Option<ElasticDSLQueryInfo> {
+    // Main entry point for extracting relevant information from the parsed query pattern. This function would contain the core logic for traversing the AST and applying rules to determine the target field, predicates, group by specifications, and aggregation type.
+    let search_request = parse_query_to_ast(query)?;
+    walk_ast_and_extract_info(&search_request)
+ }
+
 pub fn parse_query_to_ast(query: &str) -> Option<dsl::Search> {
     let search_request = serde_json::from_str(query).ok()?;
     search_request
 }
 
-pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQuery> {
-    // Placeholder for walking the AST and extracting relevant information
+pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQueryInfo> {
+    // Traverses the AST and extracts relevant information for answering sketchable aggregations within ASAPQuery.
     // This would involve traversing the AST nodes and applying logic to determine query patterns, labels, statistics, etc.
     let predicates = match ast.query.clone()? {
         dsl::Query::Bool(bool_query) => {
@@ -25,7 +31,7 @@ pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQuery> {
     };
     let (target_field, aggregation_type, group_by_spec) =
         walk_aggregations_and_extract_info(&ast.aggs)?;
-    Some(ElasticDSLQuery::new(
+    Some(ElasticDSLQueryInfo::new(
         target_field,
         predicates,
         group_by_spec,
@@ -33,7 +39,7 @@ pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQuery> {
     ))
 }
 
-pub fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Predicate> {
+fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Predicate> {
     // Placeholder for walking the filter context of the AST and extracting relevant information
     // This would involve traversing the filter nodes and applying logic to determine label filters, time ranges, etc.
     let dsl::QueryCollection(filters) = bool_query.filter.clone();
@@ -83,7 +89,7 @@ pub fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Pred
     predicates
 }
 
-pub fn walk_aggregations_and_extract_info(
+fn walk_aggregations_and_extract_info(
     aggregations: &dsl::Aggregations,
 ) -> Option<(FieldName, AggregationType, Option<GroupBySpec>)> {
     // Traverse the aggregations in the AST and extracting relevant information. Extract the first valid aggregation type found, along with any associated group by specifications.
@@ -139,7 +145,7 @@ fn find_aggregation_info(aggregations: &dsl::Aggregations) -> Option<(FieldName,
     None // Return None if no relevant aggregation information is found
 }
 
-pub fn extract_aggregation_info(agg: &dsl::Aggregation) -> Option<(FieldName, AggregationType)> {
+fn extract_aggregation_info(agg: &dsl::Aggregation) -> Option<(FieldName, AggregationType)> {
     // Extracts the specific aggregation type and target field from the given aggregation node, if it matches supported types (avg, sum, min, max, percentiles).
     match agg {
         dsl::Aggregation::Avg(avg_agg) => {
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
index ad07eed3..a50b887d 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/mod.rs
@@ -1,2 +1,5 @@
 pub mod extract_info;
 pub mod query_info;
+
+pub use extract_info::*;
+pub use query_info::*;
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
index 63b5cb51..ce82d705 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/query_info.rs
@@ -2,7 +2,8 @@ use serde::{Deserialize, Serialize};
 
 pub type FieldName = String;
 
-pub struct ElasticDSLQuery {
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ElasticDSLQueryInfo {
     // A distilled representation of an ElasticSearch DSL query, capturing the essential logic and structure.
     pub target_field: FieldName,    // List of metrics being queried
     pub predicates: Vec<Predicate>, // Predicates applied to the query (e.g. filters in bool.filter)
@@ -10,7 +11,7 @@ pub struct ElasticDSLQuery {
     pub aggregation: AggregationType, // The statistic being computed (e.g. avg, sum, percentiles)
 }
 
-impl ElasticDSLQuery {
+impl ElasticDSLQueryInfo {
     // Additional methods for processing or analyzing the query can be added here
 
     pub fn new(
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
index 1e504b4e..4a86ba8b 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
@@ -1,10 +1,7 @@
 pub mod ast_parsing;
 pub mod datemath;
 pub mod helpers;
-pub mod parsing;
-pub mod pattern;
-pub mod types;
 
-pub use parsing::*;
-pub use pattern::{classify, parse_and_classify};
-pub use types::*;
+pub use ast_parsing::*;
+pub use datemath::*;
+pub use helpers::*;
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/parsing.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/parsing.rs
deleted file mode 100644
index 3963c1f7..00000000
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/parsing.rs
+++ /dev/null
@@ -1,439 +0,0 @@
-use serde_json::Value;
-
-use crate::types::{GroupBySpec, LabelFilter, MetricAggType, MetricAggregation, TimeRange};
-
-// ---------------------------------------------------------------------------
-// Metric aggregation helpers
-// ---------------------------------------------------------------------------
-
-/// Try to extract a list of metric aggregations from the top-level `"aggs"`
-/// object of a query.  Returns `None` if *any* aggregation entry is not one of
-/// the recognised metric types (avg / min / max / sum / percentiles).
-pub fn extract_metric_aggs(aggs: &Value) -> Option<Vec<MetricAggregation>> {
-    let obj = aggs.as_object()?;
-    if obj.is_empty() {
-        return None;
-    }
-
-    let mut result = Vec::with_capacity(obj.len());
-    for (result_name, agg_body) in obj {
-        // Each aggregation body is an object that should contain exactly one
-        // recognised metric aggregation key.
-        let body_obj = agg_body.as_object()?;
-        let mut found = None;
-        for (key, inner) in body_obj {
-            if let Some(agg_type) = MetricAggType::from_json_str(key) {
-                let field = inner.get("field")?.as_str()?.to_owned();
-                let kwargs_map = inner
-                    .as_object()?
-                    .iter()
-                    .filter(|(k, _)| *k != "field")
-                    .map(|(k, v)| (k.clone(), v.clone()))
-                    .collect();
-                let kwargs = serde_json::Value::Object(kwargs_map);
-                found = Some(MetricAggregation {
-                    result_name: result_name.clone(),
-                    agg_type,
-                    field,
-                    params: if kwargs.as_object().is_some_and(|o| o.is_empty()) {
-                        None
-                    } else {
-                        Some(kwargs)
-                    },
-                });
-                break;
-            }
-        }
-        result.push(found?);
-    }
-    Some(result)
-}
-
-// ---------------------------------------------------------------------------
-// Time range helpers
-// ---------------------------------------------------------------------------
-
-/// Try to extract a `TimeRange` from a bare `{"range": {"<field>": {...}}}`
-/// query value.  Accepts either string or numeric values for gte/lte.
-pub fn extract_time_range(query: &Value) -> Option<TimeRange> {
-    let range_obj = query.get("range")?.as_object()?;
-    // There should be exactly one field entry in the range object.
-    if range_obj.len() != 1 {
-        return None;
-    }
-    let (field, bounds) = range_obj.iter().next()?;
-    let gte = bounds.get("gte").and_then(value_to_string);
-    let lte = bounds.get("lte").and_then(value_to_string);
-    Some(TimeRange {
-        field: field.clone(),
-        gte,
-        lte,
-    })
-}
-
-fn value_to_string(v: &Value) -> Option<String> {
-    match v {
-        Value::String(s) => Some(s.clone()),
-        Value::Number(n) => Some(n.to_string()),
-        _ => None,
-    }
-}
-
-// ---------------------------------------------------------------------------
-// Term / label-filter helpers
-// ---------------------------------------------------------------------------
-
-/// Strip the `.keyword` suffix from a field name, if present.
-fn strip_keyword_suffix(field: &str) -> &str {
-    field.strip_suffix(".keyword").unwrap_or(field)
-}
-
-/// Try to extract a `LabelFilter` from a single `"term"` query object.
-///
-/// Handles both the opensearch-dsl long form:
-/// ```json
-/// { "term": { "field": { "value": "val" } } }
-/// ```
-/// and the ES shorthand:
-/// ```json
-/// { "term": { "field": "val" } }
-/// ```
-pub fn extract_label_filter_from_term(term_query: &Value) -> Option<LabelFilter> {
-    let term_obj = term_query.get("term")?.as_object()?;
-    if term_obj.len() != 1 {
-        return None;
-    }
-    let (raw_field, field_value) = term_obj.iter().next()?;
-    let field = strip_keyword_suffix(raw_field).to_owned();
-    let value = if let Some(s) = field_value.as_str() {
-        // Shorthand: "field": "value"
-        s.to_owned()
-    } else if let Some(inner) = field_value.as_object() {
-        // Long form: "field": { "value": "..." }
-        inner.get("value")?.as_str()?.to_owned()
-    } else {
-        return None;
-    };
-    Some(LabelFilter { field, value })
-}
-
-// ---------------------------------------------------------------------------
-// Bool filter helpers
-// ---------------------------------------------------------------------------
-
-/// Try to extract a list of label filters (and optionally a time range) from a
-/// `{"bool": {"filter": [...]}}` query structure.
-///
-/// The `filter` array must contain at least a term query, and may also contain
-/// a range query.  Additional (unrecognised) entries in the array cause this
-/// function to return `None`.
-pub fn extract_label_filters(query: &Value) -> Option<(Vec<LabelFilter>, Option<TimeRange>)> {
-    let filter_clauses = query.get("bool")?.get("filter")?;
-
-    // The filter value may be an array (multiple clauses) or a single object.
-    let clauses: Vec<&Value> = if let Some(arr) = filter_clauses.as_array() {
-        arr.iter().collect()
-    } else if filter_clauses.is_object() {
-        vec![filter_clauses]
-    } else {
-        return None;
-    };
-
-    let mut label_filters: Vec<LabelFilter> = Vec::new();
-    let mut time_range: Option<TimeRange> = None;
-
-    for clause in clauses {
-        if clause.get("term").is_some() {
-            label_filters.push(extract_label_filter_from_term(clause)?);
-        } else if clause.get("range").is_some() {
-            if time_range.is_some() {
-                return None;
-            }
-            time_range = Some(extract_time_range(clause)?);
-        } else {
-            // Unknown clause type in the filter.
-            return None;
-        }
-    }
-
-    Some((label_filters, time_range))
-}
-
-// ---------------------------------------------------------------------------
-// Query predicate helpers
-// ---------------------------------------------------------------------------
-
-/// Extract optional predicates from top-level query:
-/// - `{"range": ...}` -> `(label_filters=[], time_range=Some(...))`
-/// - `{"bool": {"filter": ...}}` -> label filters + optional time range
-/// - `None`/`null` query is represented by caller as `(vec![], None)`.
-pub fn extract_predicates_from_query(
-    query: &Value,
-) -> Option<(Vec<LabelFilter>, Option<TimeRange>)> {
-    if query.is_null() {
-        return Some((Vec::new(), None));
-    }
-
-    if let Some(time_range) = extract_time_range(query) {
-        return Some((Vec::new(), Some(time_range)));
-    }
-
-    if query.get("bool").is_some() {
-        return extract_label_filters(query);
-    }
-
-    None
-}
-
-// ---------------------------------------------------------------------------
-// Group-by helpers
-// ---------------------------------------------------------------------------
-
-/// Try to extract a grouped aggregation from top-level `"aggs"` object.
-///
-/// Expected shape:
-/// ```json
-/// {
-///   "aggs": {
-///     "<grouped_result>": {
-///       "terms": { "field": "<label>.keyword" },
-///       "aggs": { ...metric aggs... }
-///     }
-///   }
-/// }
-/// ```
-/// or
-/// ```json
-/// {
-///   "aggs": {
-///     "<grouped_result>": {
-///       "multi_terms": {
-///         "terms": [{"field": "a.keyword"}, {"field": "b.keyword"}]
-///       },
-///       "aggs": { ...metric aggs... }
-///     }
-///   }
-/// }
-/// ```
-pub fn extract_group_by_agg(aggs: &Value) -> Option<(String, GroupBySpec, Vec<MetricAggregation>)> {
-    let obj = aggs.as_object()?;
-    // There must be exactly one top-level aggregation entry.
-    if obj.len() != 1 {
-        return None;
-    }
-    let (grouped_result_name, agg_body) = obj.iter().next()?;
-
-    let group_by = if let Some(terms_obj) = agg_body.get("terms") {
-        let raw_field = terms_obj.get("field")?.as_str()?;
-        GroupBySpec::Terms {
-            field: strip_keyword_suffix(raw_field).to_owned(),
-        }
-    } else if let Some(multi_terms_obj) = agg_body.get("multi_terms") {
-        let terms = multi_terms_obj.get("terms")?.as_array()?;
-        if terms.is_empty() {
-            return None;
-        }
-        let mut fields = Vec::with_capacity(terms.len());
-        for term in terms {
-            let raw_field = term.get("field")?.as_str()?;
-            fields.push(strip_keyword_suffix(raw_field).to_owned());
-        }
-        GroupBySpec::MultiTerms { fields }
-    } else {
-        return None;
-    };
-
-    // The nested "aggs" holds the metric sub-aggregations.
-    let nested_aggs = agg_body.get("aggs").unwrap_or(&Value::Null);
-    let metric_aggs = extract_metric_aggs(nested_aggs)?;
-
-    Some((grouped_result_name.clone(), group_by, metric_aggs))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn test_extract_metric_aggs_basic() {
-        let aggs = json!({
-            "avg_latency": { "avg": { "field": "latency_ms" } },
-            "max_latency": { "max": { "field": "latency_ms" } },
-            "p95_latency": { "percentiles": { "field": "latency_ms", "percents": [95] } }
-        });
-        let result = extract_metric_aggs(&aggs).unwrap();
-        assert_eq!(result.len(), 3);
-        let avg = result
-            .iter()
-            .find(|a| a.result_name == "avg_latency")
-            .unwrap();
-        assert_eq!(avg.agg_type, MetricAggType::Avg);
-        assert_eq!(avg.field, "latency_ms");
-        let p95 = result
-            .iter()
-            .find(|a| a.result_name == "p95_latency")
-            .unwrap();
-        assert_eq!(p95.agg_type, MetricAggType::Percentiles);
-        assert_eq!(p95.field, "latency_ms");
-        assert_eq!(
-            p95.params.as_ref().unwrap().get("percents").unwrap(),
-            &json!([95])
-        );
-
-        let avg = result
-            .iter()
-            .find(|a| a.result_name == "avg_latency")
-            .unwrap();
-        assert!(avg.params.is_none());
-    }
-
-    #[test]
-    fn test_extract_metric_aggs_rejects_unknown_type() {
-        let aggs = json!({
-            "by_service": { "terms": { "field": "service" } }
-        });
-        assert!(extract_metric_aggs(&aggs).is_none());
-    }
-
-    #[test]
-    fn test_extract_time_range() {
-        let query = json!({
-            "range": {
-                "@timestamp": { "gte": "now-30s", "lte": "now" }
-            }
-        });
-        let tr = extract_time_range(&query).unwrap();
-        assert_eq!(tr.field, "@timestamp");
-        assert_eq!(tr.gte.as_deref(), Some("now-30s"));
-        assert_eq!(tr.lte.as_deref(), Some("now"));
-    }
-
-    #[test]
-    fn test_extract_label_filter_long_form() {
-        let term = json!({ "term": { "service.keyword": { "value": "frontend" } } });
-        let f = extract_label_filter_from_term(&term).unwrap();
-        assert_eq!(f.field, "service");
-        assert_eq!(f.value, "frontend");
-    }
-
-    #[test]
-    fn test_extract_label_filter_shorthand() {
-        let term = json!({ "term": { "env": "production" } });
-        let f = extract_label_filter_from_term(&term).unwrap();
-        assert_eq!(f.field, "env");
-        assert_eq!(f.value, "production");
-    }
-
-    #[test]
-    fn test_extract_bool_filter_term_and_range() {
-        let query = json!({
-            "bool": {
-                "filter": [
-                    { "term": { "service.keyword": { "value": "frontend" } } },
-                    { "term": { "env.keyword": { "value": "production" } } },
-                    { "range": { "@timestamp": { "gte": "now-30s", "lte": "now" } } }
-                ]
-            }
-        });
-        let (lf, tr) = extract_label_filters(&query).unwrap();
-        assert_eq!(lf[0].field, "service");
-        assert_eq!(lf[0].value, "frontend");
-        assert_eq!(lf[1].field, "env");
-        assert_eq!(lf[1].value, "production");
-        let tr = tr.unwrap();
-        assert_eq!(tr.field, "@timestamp");
-    }
-
-    #[test]
-    fn test_extract_bool_filter_term_only() {
-        let query = json!({
-            "bool": {
-                "filter": [
-                    { "term": { "env": "staging" } }
-                ]
-            }
-        });
-        let (lf, tr) = extract_label_filters(&query).unwrap();
-        assert_eq!(lf[0].field, "env");
-        assert_eq!(lf[0].value, "staging");
-        assert!(tr.is_none());
-    }
-
-    #[test]
-    fn test_extract_bool_filter_single_object() {
-        // filter as a plain object (not array)
-        let query = json!({
-            "bool": {
-                "filter": { "term": { "region": "us-east-1" } }
-            }
-        });
-        let (lf, tr) = extract_label_filters(&query).unwrap();
-        assert_eq!(lf[0].field, "region");
-        assert_eq!(lf[0].value, "us-east-1");
-        assert!(tr.is_none());
-    }
-
-    #[test]
-    fn test_extract_batched_filters() {
-        let aggs = json!({
-            "by_service": {
-                "terms": {
-                    "field": "service.keyword"
-                },
-                "aggs": {
-                    "avg_latency": { "avg": { "field": "latency_ms" } }
-                }
-            }
-        });
-        let (name, group_by, metric_aggs) = extract_group_by_agg(&aggs).unwrap();
-        assert_eq!(name, "by_service");
-        assert_eq!(
-            group_by,
-            GroupBySpec::Terms {
-                field: "service".to_string()
-            }
-        );
-        assert_eq!(metric_aggs.len(), 1);
-        assert_eq!(metric_aggs[0].agg_type, MetricAggType::Avg);
-    }
-
-    #[test]
-    fn test_extract_group_by_multi_terms() {
-        let aggs = json!({
-            "by_service_region": {
-                "multi_terms": {
-                    "terms": [
-                        { "field": "service.keyword" },
-                        { "field": "region.keyword" }
-                    ]
-                },
-                "aggs": {
-                    "p95_latency": { "percentiles": { "field": "latency_ms", "percents": [95] } }
-                }
-            }
-        });
-        let (_, group_by, metric_aggs) = extract_group_by_agg(&aggs).unwrap();
-        assert_eq!(
-            group_by,
-            GroupBySpec::MultiTerms {
-                fields: vec!["service".to_string(), "region".to_string()]
-            }
-        );
-        assert_eq!(metric_aggs.len(), 1);
-        assert_eq!(metric_aggs[0].agg_type, MetricAggType::Percentiles);
-    }
-
-    #[test]
-    fn test_extract_predicates_from_range() {
-        let query = json!({
-            "range": {
-                "@timestamp": { "gte": "now-30s", "lte": "now" }
-            }
-        });
-        let (filters, time_range) = extract_predicates_from_query(&query).unwrap();
-        assert!(filters.is_empty());
-        let tr = time_range.unwrap();
-        assert_eq!(tr.field, "@timestamp");
-    }
-}
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/pattern.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/pattern.rs
deleted file mode 100644
index 29240565..00000000
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/pattern.rs
+++ /dev/null
@@ -1,403 +0,0 @@
-use serde_json::Value;
-
-use crate::{
-    parsing::{extract_group_by_agg, extract_metric_aggs, extract_predicates_from_query},
-    types::EsDslQueryPattern,
-};
-
-/// Classify a parsed ES DSL query `Value` into one of the recognised
-/// sketch-acceleratable patterns (or `Unknown` if it does not match any).
-///
-/// The classification logic follows the templates documented in
-/// `supported_es_queries.md`:
-///
-/// - **Template 1** (`SimpleAggregation`): `size=0`, top-level metric `aggs`
-///   (avg/min/max/sum/percentiles), optional bare `range` query.
-/// - **Template 2** (`GroupByAggregation`): `size=0`, one top-level grouped
-///   aggregation (`terms` or `multi_terms`) with nested metric sub-aggregations,
-///   optional `bool.filter` predicates (term labels + optional range).
-pub fn classify(value: &Value) -> EsDslQueryPattern {
-    // Gate: size must be explicitly 0.
-    match value.get("size") {
-        Some(Value::Number(n)) => {
-            if n.as_u64() != Some(0) {
-                return EsDslQueryPattern::Unknown;
-            }
-        }
-        _ => return EsDslQueryPattern::Unknown,
-    }
-
-    let aggs = value.get("aggs").unwrap_or(&Value::Null);
-    let query = value.get("query");
-
-    let (label_filters, time_range) = match query {
-        None => (Vec::new(), None),
-        Some(q) => match extract_predicates_from_query(q) {
-            Some(predicates) => predicates,
-            None => return EsDslQueryPattern::Unknown,
-        },
-    };
-
-    // ------------------------------------------------------------------
-    // Template 2: grouped aggregation (`terms` or `multi_terms`).
-    // ------------------------------------------------------------------
-    if let Some((grouped_result_name, group_by, aggregations)) = extract_group_by_agg(aggs) {
-        return EsDslQueryPattern::GroupByAggregation {
-            grouped_result_name,
-            group_by,
-            label_filters,
-            time_range,
-            aggregations,
-        };
-    }
-
-    // ------------------------------------------------------------------
-    // Templates 1 & 2 require metric-only top-level aggregations.
-    // ------------------------------------------------------------------
-    let aggregations = match extract_metric_aggs(aggs) {
-        Some(a) => a,
-        None => return EsDslQueryPattern::Unknown,
-    };
-
-    EsDslQueryPattern::SimpleAggregation {
-        label_filters,
-        time_range,
-        aggregations,
-    }
-}
-
-/// Parse a raw JSON string as an ES DSL query and classify it into a sketch-acceleratable pattern, returning the extracted structured components if successful.
-///
-/// Returns a `serde_json::Error` if the input is not valid JSON.
-pub fn parse_and_classify(json: &str) -> Result<EsDslQueryPattern, serde_json::Error> {
-    let value: Value = serde_json::from_str(json)?;
-    Ok(classify(&value))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::types::{GroupBySpec, LabelFilter, MetricAggType};
-    use serde_json::json;
-
-    // -----------------------------------------------------------------------
-    // Template 1 — Simple Aggregation
-    // -----------------------------------------------------------------------
-
-    #[test]
-    fn test_t1_simple_agg_with_time_range() {
-        let query = json!({
-            "size": 0,
-            "query": {
-                "range": {
-                    "@timestamp": { "gte": "now-30s", "lte": "now" }
-                }
-            },
-            "aggs": {
-                "avg_latency": { "avg": { "field": "latency_ms" } },
-                "max_latency": { "max": { "field": "latency_ms" } }
-            }
-        });
-
-        let pattern = classify(&query);
-        match pattern {
-            EsDslQueryPattern::SimpleAggregation {
-                label_filters,
-                time_range,
-                aggregations,
-            } => {
-                assert!(label_filters.is_empty());
-                let tr = time_range.unwrap();
-                assert_eq!(tr.field, "@timestamp");
-                assert_eq!(tr.gte.as_deref(), Some("now-30s"));
-                assert_eq!(tr.lte.as_deref(), Some("now"));
-                assert_eq!(aggregations.len(), 2);
-            }
-            other => panic!("Expected SimpleAggregation, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_t1_simple_agg_no_query() {
-        let query = json!({
-            "size": 0,
-            "aggs": {
-                "total_bytes": { "sum": { "field": "bytes" } }
-            }
-        });
-
-        let pattern = classify(&query);
-        match pattern {
-            EsDslQueryPattern::SimpleAggregation {
-                label_filters,
-                time_range,
-                aggregations,
-            } => {
-                assert!(label_filters.is_empty());
-                assert!(time_range.is_none());
-                assert_eq!(aggregations.len(), 1);
-                assert_eq!(aggregations[0].agg_type, MetricAggType::Sum);
-                assert_eq!(aggregations[0].field, "bytes");
-            }
-            other => panic!("Expected SimpleAggregation, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_t1_percentiles_aggregation() {
-        let query = json!({
-            "size": 0,
-            "aggs": {
-                "p95_latency": { "percentiles": { "field": "latency_ms", "percents": [95] } }
-            }
-        });
-
-        let pattern = classify(&query);
-        match pattern {
-            EsDslQueryPattern::SimpleAggregation {
-                label_filters,
-                time_range,
-                aggregations,
-            } => {
-                assert!(label_filters.is_empty());
-                assert!(time_range.is_none());
-                assert_eq!(aggregations.len(), 1);
-                assert_eq!(aggregations[0].agg_type, MetricAggType::Percentiles);
-                assert_eq!(aggregations[0].field, "latency_ms");
-            }
-            other => panic!("Expected SimpleAggregation, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_simple_agg_with_bool_filter_predicates() {
-        let query = json!({
-            "size": 0,
-            "query": {
-                "bool": {
-                    "filter": [
-                        { "term": { "service.keyword": { "value": "frontend" } } },
-                        { "term": { "env.keyword": { "value": "staging" } } },
-                        { "range": { "@timestamp": { "gte": "now-30s", "lte": "now" } } }
-                    ]
-                }
-            },
-            "aggs": {
-                "avg_latency": { "avg": { "field": "latency_ms" } }
-            }
-        });
-
-        let pattern = classify(&query);
-        match pattern {
-            EsDslQueryPattern::SimpleAggregation {
-                label_filters,
-                time_range,
-                aggregations,
-            } => {
-                assert_eq!(label_filters.len(), 2);
-                assert_eq!(
-                    label_filters[0],
-                    LabelFilter {
-                        field: "service".into(),
-                        value: "frontend".into()
-                    }
-                );
-                assert_eq!(
-                    label_filters[1],
-                    LabelFilter {
-                        field: "env".into(),
-                        value: "staging".into()
-                    }
-                );
-                assert!(time_range.is_some());
-                assert_eq!(aggregations.len(), 1);
-            }
-            other => panic!("Expected SimpleAggregation, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_neg_size_absent_is_unknown() {
-        let query = json!({
-            "aggs": {
-                "min_val": { "min": { "field": "response_time" } }
-            }
-        });
-
-        assert_eq!(classify(&query), EsDslQueryPattern::Unknown);
-    }
-
-    // -----------------------------------------------------------------------
-    // Template 2 — GroupBy Aggregation
-    // -----------------------------------------------------------------------
-
-    #[test]
-    fn test_t2_groupby_terms_with_filters_and_range() {
-        let query = json!({
-            "size": 0,
-            "query": {
-                "bool": {
-                    "filter": [
-                        { "term": { "service.keyword": { "value": "frontend" } } },
-                        { "term": { "env.keyword": { "value": "staging" } } },
-                        { "range": { "@timestamp": { "gte": "now-30s", "lte": "now" } } }
-                    ]
-                }
-            },
-            "aggs": {
-                "grouped_result": {
-                    "terms": {
-                        "field": "service.keyword"
-                    },
-                    "aggs": {
-                        "avg_latency": { "avg": { "field": "latency_ms" } }
-                    }
-                }
-            }
-        });
-
-        let pattern = classify(&query);
-        match pattern {
-            EsDslQueryPattern::GroupByAggregation {
-                grouped_result_name,
-                group_by,
-                label_filters,
-                time_range,
-                aggregations,
-            } => {
-                assert_eq!(grouped_result_name, "grouped_result");
-                assert_eq!(
-                    group_by,
-                    GroupBySpec::Terms {
-                        field: "service".into()
-                    }
-                );
-                assert_eq!(label_filters[0].field, "service");
-                assert_eq!(label_filters[0].value, "frontend");
-                assert_eq!(label_filters[1].field, "env");
-                assert_eq!(label_filters[1].value, "staging");
-                let tr = time_range.unwrap();
-                assert_eq!(tr.field, "@timestamp");
-                assert_eq!(aggregations.len(), 1);
-            }
-            other => panic!("Expected GroupByAggregation, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_t2_groupby_multi_terms() {
-        let query = json!({
-            "size": 0,
-            "query": {
-                "bool": {
-                    "filter": [
-                        { "term": { "env": "staging" } }
-                    ]
-                }
-            },
-            "aggs": {
-                "grouped_result": {
-                    "multi_terms": {
-                        "terms": [
-                            { "field": "service.keyword" },
-                            { "field": "region.keyword" }
-                        ]
-                    },
-                    "aggs": {
-                        "p99_latency": { "max": { "field": "latency_ms" } }
-                    }
-                }
-            }
-        });
-
-        let pattern = classify(&query);
-        match pattern {
-            EsDslQueryPattern::GroupByAggregation {
-                grouped_result_name,
-                group_by,
-                label_filters,
-                time_range,
-                aggregations,
-            } => {
-                assert_eq!(grouped_result_name, "grouped_result");
-                assert_eq!(
-                    group_by,
-                    GroupBySpec::MultiTerms {
-                        fields: vec!["service".into(), "region".into()]
-                    }
-                );
-                assert_eq!(
-                    label_filters[0],
-                    LabelFilter {
-                        field: "env".into(),
-                        value: "staging".into()
-                    }
-                );
-                assert!(time_range.is_none());
-                assert_eq!(aggregations.len(), 1);
-            }
-            other => panic!("Expected GroupByAggregation, got {:?}", other),
-        }
-    }
-
-    // -----------------------------------------------------------------------
-    // Negative cases
-    // -----------------------------------------------------------------------
-
-    #[test]
-    fn test_neg_size_nonzero_is_unknown() {
-        let query = json!({
-            "size": 10,
-            "aggs": {
-                "avg_val": { "avg": { "field": "cpu" } }
-            }
-        });
-        assert_eq!(classify(&query), EsDslQueryPattern::Unknown);
-    }
-
-    #[test]
-    fn test_neg_unknown_agg_type_is_unknown() {
-        let query = json!({
-            "size": 0,
-            "aggs": {
-                "by_service": {
-                    "terms": { "field": "service" },
-                    "aggs": {
-                        "foo": { "median": { "field": "latency" } }
-                    }
-                }
-            }
-        });
-        assert_eq!(classify(&query), EsDslQueryPattern::Unknown);
-    }
-
-    #[test]
-    fn test_neg_unsupported_query_type_is_unknown() {
-        // A match query in the top-level query is not a supported pattern.
-        let query = json!({
-            "size": 0,
-            "query": {
-                "match": { "message": "error" }
-            },
-            "aggs": {
-                "count": { "sum": { "field": "bytes" } }
-            }
-        });
-        assert_eq!(classify(&query), EsDslQueryPattern::Unknown);
-    }
-
-    #[test]
-    fn test_parse_and_classify_roundtrip() {
-        let json = r#"{"size":0,"aggs":{"avg_cpu":{"avg":{"field":"cpu_usage"}}}}"#;
-        let result = parse_and_classify(json).unwrap();
-        assert!(matches!(
-            result,
-            EsDslQueryPattern::SimpleAggregation { .. }
-        ));
-    }
-
-    #[test]
-    fn test_parse_and_classify_invalid_json() {
-        assert!(parse_and_classify("{invalid}").is_err());
-    }
-}
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/types.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/types.rs
deleted file mode 100644
index 5c1269d9..00000000
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/types.rs
+++ /dev/null
@@ -1,286 +0,0 @@
-use serde::{Deserialize, Serialize};
-
-/// Time range bounds resolved into epoch milliseconds.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct ResolvedTimeRange {
-    pub field: String,
-    pub gte_ms: Option<i64>,
-    pub lte_ms: Option<i64>,
-}
-
-/// The metric aggregation function type.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "lowercase")]
-pub enum MetricAggType {
-    Avg,
-    Min,
-    Max,
-    Sum,
-    Percentiles,
-}
-
-impl MetricAggType {
-    /// Returns the JSON key name for this aggregation type.
-    pub fn as_json_str(&self) -> &'static str {
-        match self {
-            MetricAggType::Avg => "avg",
-            MetricAggType::Min => "min",
-            MetricAggType::Max => "max",
-            MetricAggType::Sum => "sum",
-            MetricAggType::Percentiles => "percentiles",
-        }
-    }
-
-    /// Try to parse from a string key.
-    pub fn from_json_str(s: &str) -> Option<Self> {
-        match s {
-            "avg" => Some(MetricAggType::Avg),
-            "min" => Some(MetricAggType::Min),
-            "max" => Some(MetricAggType::Max),
-            "sum" => Some(MetricAggType::Sum),
-            "percentiles" => Some(MetricAggType::Percentiles),
-            _ => None,
-        }
-    }
-}
-
-/// A simple equality filter on a label (string-valued field).
-/// The `.keyword` suffix is stripped from the field name.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct LabelFilter {
-    pub field: String,
-    pub value: String,
-}
-
-/// An optional time range applied to a timestamp field.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct TimeRange {
-    pub field: String,
-    pub gte: Option<String>,
-    pub lte: Option<String>,
-}
-
-impl TimeRange {
-    /// Parse a date-math expression into epoch milliseconds using the provided
-    /// `now_ms` as reference for `now`-relative expressions.
-    ///
-    /// Supported forms:
-    /// - `now`
-    /// - `now-30s`, `now+5m`, `now-1h`, `now-2d`, `now-1w`, `now-500ms`
-    /// - RFC3339 timestamps (e.g. `2026-03-22T12:34:56Z`)
-    /// - Plain integer timestamps (returned as-is)
-    pub fn parse_date_math(expr: &str, now_ms: i64) -> Option<i64> {
-        if expr == "now" {
-            return Some(now_ms);
-        }
-
-        if let Some(delta) = Self::parse_now_delta_ms(expr) {
-            return now_ms.checked_add(delta);
-        }
-
-        if let Ok(v) = expr.parse::<i64>() {
-            return Some(v);
-        }
-
-        chrono::DateTime::parse_from_rfc3339(expr)
-            .ok()
-            .map(|dt| dt.timestamp_millis())
-    }
-
-    /// Resolve `gte`/`lte` date-math strings into numeric epoch-millisecond
-    /// values relative to `now_ms`.
-    pub fn resolve_epoch_millis(&self, now_ms: i64) -> Option<ResolvedTimeRange> {
-        let gte_ms = match &self.gte {
-            Some(v) => Some(Self::parse_date_math(v, now_ms)?),
-            None => None,
-        };
-        let lte_ms = match &self.lte {
-            Some(v) => Some(Self::parse_date_math(v, now_ms)?),
-            None => None,
-        };
-
-        Some(ResolvedTimeRange {
-            field: self.field.clone(),
-            gte_ms,
-            lte_ms,
-        })
-    }
-
-    fn parse_now_delta_ms(expr: &str) -> Option<i64> {
-        let rest = expr.strip_prefix("now")?;
-        if rest.is_empty() {
-            return Some(0);
-        }
-
-        let sign_char = rest.chars().next()?;
-        let sign = match sign_char {
-            '+' => 1_i64,
-            '-' => -1_i64,
-            _ => return None,
-        };
-
-        let offset = &rest[1..];
-        if offset.is_empty() {
-            return None;
-        }
-
-        let digit_count = offset.chars().take_while(|c| c.is_ascii_digit()).count();
-        if digit_count == 0 || digit_count == offset.len() {
-            return None;
-        }
-
-        let qty = offset[..digit_count].parse::<i64>().ok()?;
-        let unit = &offset[digit_count..];
-        let unit_ms = match unit {
-            "ms" => 1_i64,
-            "s" => 1_000_i64,
-            "m" => 60_000_i64,
-            "h" => 3_600_000_i64,
-            "d" => 86_400_000_i64,
-            "w" => 604_800_000_i64,
-            _ => return None,
-        };
-
-        qty.checked_mul(unit_ms)?.checked_mul(sign)
-    }
-}
-
-/// A single metric aggregation extracted from an ES query.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct MetricAggregation {
-    /// The top-level aggregation result key (the name given by the user).
-    pub result_name: String,
-    pub agg_type: MetricAggType,
-    /// The document field being aggregated over.
-    pub field: String,
-    pub params: Option<serde_json::Value>, // Optional additional parameters (e.g. percentiles values)
-}
-
-/// Group-by shape in a grouped aggregation.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub enum GroupBySpec {
-    /// `{"terms": {"field": "..."}}`
-    Terms { field: String },
-    /// `{"multi_terms": {"terms": [{"field": "..."}, ...]}}`
-    MultiTerms { fields: Vec<String> },
-}
-
-/// The classified pattern of an ES DSL query, along with the extracted
-/// structured components needed to route it to a sketch fast-path.
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub enum EsDslQueryPattern {
-    /// Template 1: metric aggregations over all data, with an optional time
-    /// range and optional label filters in `bool.filter`.
-    ///
-    /// ES: `{ "size": 0, "query": { "range": {...} }, "aggs": { ... } }`
-    SimpleAggregation {
-        label_filters: Vec<LabelFilter>,
-        time_range: Option<TimeRange>,
-        aggregations: Vec<MetricAggregation>,
-    },
-
-    /// Template 2: grouped aggregation by one or more labels (`terms` or
-    /// `multi_terms`) with nested metric aggregations, and optional
-    /// `bool.filter` predicates.
-    ///
-    /// ES: `{ "size": 0, "aggs": { "<name>": { "terms"|"multi_terms": ...,
-    ///         "aggs": { ... } } } }`
-    GroupByAggregation {
-        grouped_result_name: String,
-        group_by: GroupBySpec,
-        label_filters: Vec<LabelFilter>,
-        time_range: Option<TimeRange>,
-        aggregations: Vec<MetricAggregation>,
-    },
-
-    /// The query did not match any recognised sketch-acceleratable pattern.
-    Unknown,
-}
-
-impl EsDslQueryPattern {
-    pub fn get_time_range(&self) -> Option<&TimeRange> {
-        match self {
-            EsDslQueryPattern::SimpleAggregation { time_range, .. } => time_range.as_ref(),
-            EsDslQueryPattern::GroupByAggregation { time_range, .. } => time_range.as_ref(),
-            EsDslQueryPattern::Unknown => None,
-        }
-    }
-
-    pub fn get_groupby_spec(&self) -> Option<&GroupBySpec> {
-        match self {
-            EsDslQueryPattern::GroupByAggregation { group_by, .. } => Some(group_by),
-            _ => None,
-        }
-    }
-
-    pub fn get_metric_aggs(&self) -> Option<&Vec<MetricAggregation>> {
-        match self {
-            EsDslQueryPattern::SimpleAggregation { aggregations, .. } => Some(aggregations),
-            EsDslQueryPattern::GroupByAggregation { aggregations, .. } => Some(aggregations),
-            EsDslQueryPattern::Unknown => None,
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{ResolvedTimeRange, TimeRange};
-
-    #[test]
-    fn parse_date_math_now_relative() {
-        let now = 1_700_000_000_000_i64;
-        assert_eq!(TimeRange::parse_date_math("now", now), Some(now));
-        assert_eq!(
-            TimeRange::parse_date_math("now-30s", now),
-            Some(now - 30_000)
-        );
-        assert_eq!(
-            TimeRange::parse_date_math("now+2m", now),
-            Some(now + 120_000)
-        );
-        assert_eq!(
-            TimeRange::parse_date_math("now-500ms", now),
-            Some(now - 500)
-        );
-    }
-
-    #[test]
-    fn parse_date_math_rfc3339_and_integer() {
-        let now = 0_i64;
-        assert_eq!(
-            TimeRange::parse_date_math("2026-03-22T00:00:00Z", now),
-            Some(1_774_137_600_000)
-        );
-        assert_eq!(
-            TimeRange::parse_date_math("1774137600000", now),
-            Some(1_774_137_600_000)
-        );
-    }
-
-    #[test]
-    fn parse_date_math_invalid_expressions() {
-        let now = 1_700_000_000_000_i64;
-        assert_eq!(TimeRange::parse_date_math("now-", now), None);
-        assert_eq!(TimeRange::parse_date_math("now-10q", now), None);
-        assert_eq!(TimeRange::parse_date_math("yesterday", now), None);
-    }
-
-    #[test]
-    fn resolve_epoch_millis_for_range() {
-        let tr = TimeRange {
-            field: "@timestamp".to_string(),
-            gte: Some("now-30s".to_string()),
-            lte: Some("now".to_string()),
-        };
-        let now = 1_700_000_000_000_i64;
-
-        assert_eq!(
-            tr.resolve_epoch_millis(now),
-            Some(ResolvedTimeRange {
-                field: "@timestamp".to_string(),
-                gte_ms: Some(now - 30_000),
-                lte_ms: Some(now),
-            })
-        );
-    }
-}

From 45f8a7ace12ace7d2d9d389b4f7bf1a0c3377607 Mon Sep 17 00:00:00 2001
From: Eric Wang <eswang1202@gmail.com>
Date: Thu, 23 Apr 2026 22:47:10 -0400
Subject: [PATCH 5/7] Fix issue where ES DSL parser does not extract predicates
 (e.g. filter for time range query) when it is not enclosed in a bool context.
 Refactor Elastic DSL handler to use new parsing utilities.

---
 .../src/ast_parsing/extract_info.rs           | 105 +++++++++--------
 .../src/engines/simple_engine/elastic.rs      | 110 +++++++++---------
 .../src/tests/elastic_dsl_query_tests.rs      |  24 ++--
 3 files changed, 124 insertions(+), 115 deletions(-)

diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
index 59835d35..b9551d36 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
@@ -19,15 +19,21 @@ pub fn parse_query_to_ast(query: &str) -> Option<dsl::Search> {
 pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQueryInfo> {
     // Traverses the AST and extracts relevant information for answering sketchable aggregations within ASAPQuery.
     // This would involve traversing the AST nodes and applying logic to determine query patterns, labels, statistics, etc.
-    let predicates = match ast.query.clone()? {
-        dsl::Query::Bool(bool_query) => {
+    let query = ast.query.clone();
+    let predicates = match query {
+        Some(dsl::Query::Bool(bool_query)) => {
             // Extract information from the bool query
             walk_bool_query_and_extract_info(&bool_query)
-        }
-        _ => {
-            // Handle other query types
-            Vec::new() // Return an empty vector of predicates for unsupported query types
-        }
+        },
+        Some(other) => {
+            // Predicates may just be specified directly without enclosing bool context.
+            if let Some(predicate) = extract_predicates_from_query(&other) {
+                vec![predicate]
+            } else {
+                Vec::new()
+            }
+        },
+        None => Vec::new(), // Return an empty vector of predicates if no query is specified
     };
     let (target_field, aggregation_type, group_by_spec) =
         walk_aggregations_and_extract_info(&ast.aggs)?;
@@ -45,50 +51,57 @@ fn walk_bool_query_and_extract_info(bool_query: &dsl::BoolQuery) -> Vec<Predicat
     let dsl::QueryCollection(filters) = bool_query.filter.clone();
     let mut predicates = Vec::new();
     for query in filters {
-        match query {
-            dsl::Query::Term(term_query) => {
-                // Extract information from the term query
-                let field = strip_keyword_suffix(&term_query.field).to_owned();
-                let Some(value) = term_query.value else {
-                    continue; // Skip if term query does not have a value
-                };
-                let Some(term_value) = map_term_to_json_value(&value) else {
-                    continue; // Skip if term query value cannot be mapped to a JSON value
-                };
-                // Process the term query information as needed
-                predicates.push(Predicate::Term {
-                    field,
-                    value: term_value,
-                });
-            }
-
-            dsl::Query::Range(range_query) => {
-                // Extract information from the range query
-                let field = strip_keyword_suffix(&range_query.field).to_owned();
-                let gte = range_query.gte.clone();
-                let lte = range_query.lte.clone();
-                // Process the range query information as needed
-                let gte_value = gte
-                    .as_ref()
-                    .and_then(|gte_term| map_term_to_json_value(gte_term));
-                let lte_value = lte
-                    .as_ref()
-                    .and_then(|lte_term| map_term_to_json_value(lte_term));
-                predicates.push(Predicate::Range {
-                    field,
-                    gte: gte_value,
-                    lte: lte_value,
-                });
-            }
-            _ => {
-                // Handle other query types
-                continue; // Skip unsupported query types
-            }
+        if let Some(predicate) = extract_predicates_from_query(&query) {
+            predicates.push(predicate);
         }
     }
     predicates
 }
 
+fn extract_predicates_from_query(query: &dsl::Query) -> Option<Predicate> {
+    // Extract predicate information from a given query node, if it matches supported patterns (term or range queries).
+    match query {
+        dsl::Query::Term(term_query) => {
+            // Extract information from the term query
+            let field = strip_keyword_suffix(&term_query.field).to_owned();
+            let Some(value) = term_query.value.clone() else {
+                return None; // Skip if term query does not have a value
+            };
+            let Some(term_value) = map_term_to_json_value(&value) else {
+                return None; // Skip if term query value cannot be mapped to a JSON value
+            };
+            // Process the term query information as needed
+            return Some(Predicate::Term {
+                field,
+                value: term_value,
+            })
+        }
+
+        dsl::Query::Range(range_query) => {
+            // Extract information from the range query
+            let field = strip_keyword_suffix(&range_query.field).to_owned();
+            let gte = range_query.gte.clone();
+            let lte = range_query.lte.clone();
+            // Process the range query information as needed
+            let gte_value = gte
+                .as_ref()
+                .and_then(|gte_term| map_term_to_json_value(gte_term));
+            let lte_value = lte
+                .as_ref()
+                .and_then(|lte_term| map_term_to_json_value(lte_term));
+            return Some(Predicate::Range {
+                field,
+                gte: gte_value,
+                lte: lte_value,
+            })
+        },
+        _ => {
+            // Handle other query types
+            return None; // Skip unsupported query types
+        }
+    }
+}
+
 fn walk_aggregations_and_extract_info(
     aggregations: &dsl::Aggregations,
 ) -> Option<(FieldName, AggregationType, Option<GroupBySpec>)> {
diff --git a/asap-query-engine/src/engines/simple_engine/elastic.rs b/asap-query-engine/src/engines/simple_engine/elastic.rs
index 3cdf1682..1b6e1c14 100644
--- a/asap-query-engine/src/engines/simple_engine/elastic.rs
+++ b/asap-query-engine/src/engines/simple_engine/elastic.rs
@@ -5,8 +5,8 @@
 use super::SimpleEngine;
 use super::{QueryExecutionContext, QueryMetadata, QueryTimestamps};
 use crate::engines::query_result::QueryResult;
-use elastic_dsl_utilities::pattern::parse_and_classify;
-use elastic_dsl_utilities::types::{EsDslQueryPattern, GroupBySpec, MetricAggType};
+use elastic_dsl_utilities::ast_parsing::{self, ElasticDSLQueryInfo, GroupBySpec, AggregationType, Predicate};
+use elastic_dsl_utilities::datemath::{range_query_to_time_range};
 use promql_utilities::data_model::KeyByLabelNames;
 use promql_utilities::query_logics::enums::Statistic;
 use std::collections::HashMap;
@@ -34,18 +34,8 @@ impl SimpleEngine {
         let query_time = Self::convert_query_time_to_data_time(time);
 
         // 1. Parse query DSL somehow. Elasticsearch DSL crate does not support deserializing, but maybe can use Opensearch instead?
-        // 2. Determine whether query is supported using some AST representation or hardcoded pattern matching.
-        let query_pattern: EsDslQueryPattern =
-            parse_and_classify(&query).unwrap_or(EsDslQueryPattern::Unknown);
-        match query_pattern {
-            EsDslQueryPattern::Unknown => {
-                debug!("Could not parse query into known pattern");
-                return None;
-            }
-            _ => {
-                debug!("Parsed query pattern: {:?}", query_pattern);
-            }
-        }
+        // 2. Determine whether query is supported using some AST representation or hardcoded pattern matching (do this throughout context) - parsing validates basic structure.
+        let query_info = ast_parsing::extract_info::extract_query_info(&query)?;
 
         // 3. Convert parsed query into execution context components (labels, statistic, kwargs, metadata, store query plan, etc.)
 
@@ -61,12 +51,12 @@ impl SimpleEngine {
 
         let do_merge = true; // No "instant" queries in ElasticSearch supported for now, so we always need to merge.
 
-        let (metric, query_metadata) = self.build_query_metadata_elastic(&query_pattern)?;
+        let (metric, query_metadata) = self.build_query_metadata_elastic(&query_info)?;
 
         let spatial_filter = String::new(); // Placeholder - extract from query if applicable
 
-        // TODO: Need way to parse ES DSL "date math".
-        let timestamps = self.resolve_query_time_range_elastic(query_time, query_pattern);
+        // Parse time range information from first query predicate if available, otherwise default to entire history up to query_time. 
+        let timestamps = self.resolve_query_time_range_elastic(query_time, query_info);
 
         let query_plan = self
             .create_store_query_plan(&metric, &timestamps, &agg_info)
@@ -102,7 +92,7 @@ impl SimpleEngine {
 
     fn build_query_metadata_elastic(
         &self,
-        query_pattern: &EsDslQueryPattern,
+        query_info: &ElasticDSLQueryInfo,
     ) -> Option<(String, QueryMetadata)> {
         // Constructs QueryMetadata based on the parsed ES DSL query pattern. This includes determining the
         // metric to query, the statistic to compute, and any relevant query kwargs (e.g. quantile value for percentiles).
@@ -111,40 +101,32 @@ impl SimpleEngine {
         // By default, we only include grouping labels in the output for ES DSL.
 
         // Take first aggregation by default since current engine doesn't support multiple aggregations in a single query.
-        let aggregation = query_pattern.get_metric_aggs()?.first()?.clone();
+        let aggregation = query_info.aggregation.clone();
 
         // By default, we only include grouping labels in the output for ES DSL.
-        let query_output_labels = match query_pattern.get_groupby_spec() {
-            Some(GroupBySpec::Terms { field }) => KeyByLabelNames::new(vec![field.clone()]),
-            Some(GroupBySpec::MultiTerms { fields }) => KeyByLabelNames::new(fields.to_vec()),
+        let query_output_labels = match &query_info.group_by_buckets {
+            Some(GroupBySpec::Fields(fields)) => KeyByLabelNames::new(fields.clone()),
+            Some(GroupBySpec::Filters(_)) => { return None; } // We don't support filter-based group by in ES DSL for now, so return None to indicate unsupported query pattern.
             None => KeyByLabelNames::empty(),
         };
 
-        let metric = aggregation.field.clone();
+        let metric = query_info.target_field.clone();
 
         // Map ElasticSearch aggregation types to our internal Statistic enum.
-        let statistic_to_compute = match aggregation.agg_type {
-            MetricAggType::Percentiles => Statistic::Quantile,
-            MetricAggType::Avg => Statistic::Rate,
-            MetricAggType::Sum => Statistic::Sum,
-            MetricAggType::Min => Statistic::Min,
-            MetricAggType::Max => Statistic::Max,
+        let statistic_to_compute = match aggregation {
+            AggregationType::Percentiles(_) => Statistic::Quantile,
+            AggregationType::Avg => Statistic::Rate,
+            AggregationType::Sum => Statistic::Sum,
+            AggregationType::Min => Statistic::Min,
+            AggregationType::Max => Statistic::Max,
         };
 
-        let mut query_kwargs = HashMap::new(); // Placeholder - build based on query and statistic
-        if aggregation.agg_type == MetricAggType::Percentiles {
-            // Extract quantile value from aggregation parameters and add to query_kwargs
-            if let Some(params) = &aggregation.params {
-                if let Some(percents) = params.get("percents") {
-                    // Get first value from percents array since we only support one quantile argument for now.
-                    let quantile = percents
-                        .as_array()
-                        .and_then(|arr| arr.first())
-                        .and_then(|v| v.as_f64());
-                    // ES percentiles are specified as values between 0 and 100, but we want to convert to 0-1 range for our internal representation.
-                    query_kwargs.insert("quantile".to_string(), (quantile? / 100.0).to_string());
-                }
-            }
+        let mut query_kwargs = HashMap::new();
+        if let AggregationType::Percentiles(percents) = aggregation {
+            // Get first value from percents array since we only support one quantile argument for now.
+                let quantile = percents.first()?;
+                // ES percentiles are specified as values between 0 and 100, but we want to convert to 0-1 range for our internal representation.
+                query_kwargs.insert("quantile".to_string(), (quantile / 100.0).to_string());
         }
 
         let metadata = QueryMetadata {
@@ -158,7 +140,7 @@ impl SimpleEngine {
     pub fn resolve_query_time_range_elastic(
         &self,
         query_time: u64,
-        query_pattern: EsDslQueryPattern,
+        query_info: ElasticDSLQueryInfo,
     ) -> QueryTimestamps {
         // Resolves the actual start and end timestamps into milliseconds for an ElasticSearch query
         // based on the provided query_time and the time range specified in the ES DSL query pattern (if any).
@@ -167,20 +149,34 @@ impl SimpleEngine {
         let mut start_timestamp: u64 = 0;
         let mut end_timestamp: u64 = query_time;
 
-        let time_range = query_pattern.get_time_range();
-        if let Some(tr) = time_range {
-            if let Some(resolved_range) = tr.resolve_epoch_millis(query_time as i64) {
-                debug!(
-                    "Parsed time range from query: start={} end={}",
-                    resolved_range.gte_ms.unwrap_or(0),
-                    resolved_range.lte_ms.unwrap_or(0)
-                );
-                start_timestamp = resolved_range.gte_ms.unwrap_or(0) as u64;
-                end_timestamp = resolved_range.lte_ms.unwrap_or(query_time as i64) as u64;
-            } else {
-                debug!("Failed to resolve time range from query");
+        let predicate = query_info
+            .predicates
+            .first().clone(); // For now, we only look at the first predicate for time range information. We can extend this to support multiple predicates and more complex logic in the future.
+
+        match predicate {
+            Some(Predicate::Range { field: _, gte: _, lte: _ }) => {
+                // If we have a range predicate, we can try to extract time range information from it.
+                // For now, we assume that any range predicate applies to the timestamp field, but we could add more complex logic here to determine which field is the timestamp field.
+                debug!("Found range predicate in query, attempting to extract time range information");
+                if let Some(resolved_range) = range_query_to_time_range(predicate.unwrap(), query_time as i64) {
+                    debug!(
+                        "Parsed time range from range predicate: start={} end={}",
+                        resolved_range.gte_ms.unwrap_or(0),
+                        resolved_range.lte_ms.unwrap_or(0)
+                    );
+                    start_timestamp = resolved_range.gte_ms.unwrap_or(0) as u64;
+                    end_timestamp = resolved_range.lte_ms.unwrap_or(query_time as i64) as u64;
+                } else {
+                    debug!("Failed to resolve time range from range predicate");
+                }
             }
-        };
+            Some(_) => {
+                debug!("First predicate is not a range predicate, so we won't be able to extract time range information from it");
+            }
+            None => {
+                debug!("No predicates found in query, so we won't be able to extract time range information");
+            }
+        }
 
         QueryTimestamps {
             start_timestamp,
diff --git a/asap-query-engine/src/tests/elastic_dsl_query_tests.rs b/asap-query-engine/src/tests/elastic_dsl_query_tests.rs
index c11b2a54..68a241bd 100644
--- a/asap-query-engine/src/tests/elastic_dsl_query_tests.rs
+++ b/asap-query-engine/src/tests/elastic_dsl_query_tests.rs
@@ -100,10 +100,10 @@ mod tests {
 
     #[test]
     fn test_esdsl_single_label_groupby_aggregation_quantile() {
-        // let _ = tracing_subscriber::fmt()
-        //     .with_max_level(tracing::Level::DEBUG)
-        //     .with_test_writer() // Routes output through the test runner's capture mechanism
-        //     .try_init();
+        let _ = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::DEBUG)
+            .with_test_writer() // Routes output through the test runner's capture mechanism
+            .try_init();
 
         // Elastic DSL query (batch filtered).
         let elastic_query = json!({
@@ -173,10 +173,10 @@ mod tests {
 
     #[test]
     fn test_esdsl_multi_label_groupby_aggregation_quantile() {
-        // let _ = tracing_subscriber::fmt()
-        //     .with_max_level(tracing::Level::DEBUG)
-        //     .with_test_writer() // Routes output through the test runner's capture mechanism
-        //     .try_init();
+        let _ = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::DEBUG)
+            .with_test_writer() // Routes output through the test runner's capture mechanism
+            .try_init();
 
         // Elastic DSL query (batch filtered).
         let elastic_query = json!({
@@ -320,10 +320,10 @@ mod tests {
 
     #[test]
     fn test_esdsl_time_range_query() {
-        // let _ = tracing_subscriber::fmt()
-        //     .with_max_level(tracing::Level::DEBUG)
-        //     .with_test_writer() // Routes output through the test runner's capture mechanism
-        //     .try_init();
+        let _ = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::DEBUG)
+            .with_test_writer() // Routes output through the test runner's capture mechanism
+            .try_init();
 
         // Elastic DSL query (batch filtered).
         let elastic_query = json!({

From ff8e564aa5be5100247b62e4f87349cd18438800 Mon Sep 17 00:00:00 2001
From: Eric Wang <eswang1202@gmail.com>
Date: Fri, 24 Apr 2026 00:07:52 -0400
Subject: [PATCH 6/7] Cargo fmt and clippy. Point elastic_dsl_utilities to
 custom copy of opensearch_dsl.

---
 Cargo.lock                                    |  2 +-
 .../rs/elastic_dsl_utilities/Cargo.toml       |  2 +-
 .../src/ast_parsing/extract_info.rs           | 47 +++++++++----------
 .../src/engines/simple_engine/elastic.rs      | 36 +++++++++-----
 4 files changed, 46 insertions(+), 41 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 6344e25e..5630c4f8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3031,7 +3031,7 @@ checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
 
 [[package]]
 name = "opensearch-dsl"
-version = "0.3.1"
+version = "0.1.0"
 dependencies = [
  "chrono",
  "num-traits",
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
index 92e1278f..d07f01a2 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
@@ -7,4 +7,4 @@ version.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 chrono.workspace = true
-opensearch-dsl = { version = "0.3.1", path = "../../../../../opensearch-client-rs/opensearch-dsl" }
+opensearch-dsl = { version = "0.1.0", path = "../../../../../opensearch-dsl" }
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
index b9551d36..754f437c 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
@@ -6,14 +6,13 @@ use opensearch_dsl::{self as dsl};
 use serde_json;
 
 pub fn extract_query_info(query: &str) -> Option<ElasticDSLQueryInfo> {
-    // Main entry point for extracting relevant information from the parsed query pattern. This function would contain the core logic for traversing the AST and applying rules to determine the target field, predicates, group by specifications, and aggregation type.
-    let search_request = parse_query_to_ast(query)?;
+    // Main entry point for extracting relevant information from the parsed query pattern.
+    let search_request = serde_json::from_str(query).ok()?;
     walk_ast_and_extract_info(&search_request)
- }
+}
 
 pub fn parse_query_to_ast(query: &str) -> Option<dsl::Search> {
-    let search_request = serde_json::from_str(query).ok()?;
-    search_request
+    serde_json::from_str(query).ok()?
 }
 
 pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQueryInfo> {
@@ -24,7 +23,7 @@ pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQueryInf
         Some(dsl::Query::Bool(bool_query)) => {
             // Extract information from the bool query
             walk_bool_query_and_extract_info(&bool_query)
-        },
+        }
         Some(other) => {
             // Predicates may just be specified directly without enclosing bool context.
             if let Some(predicate) = extract_predicates_from_query(&other) {
@@ -32,7 +31,7 @@ pub fn walk_ast_and_extract_info(ast: &dsl::Search) -> Option<ElasticDSLQueryInf
             } else {
                 Vec::new()
             }
-        },
+        }
         None => Vec::new(), // Return an empty vector of predicates if no query is specified
     };
     let (target_field, aggregation_type, group_by_spec) =
@@ -71,7 +70,7 @@ fn extract_predicates_from_query(query: &dsl::Query) -> Option<Predicate> {
                 return None; // Skip if term query value cannot be mapped to a JSON value
             };
             // Process the term query information as needed
-            return Some(Predicate::Term {
+            Some(Predicate::Term {
                 field,
                 value: term_value,
             })
@@ -83,21 +82,17 @@ fn extract_predicates_from_query(query: &dsl::Query) -> Option<Predicate> {
             let gte = range_query.gte.clone();
             let lte = range_query.lte.clone();
             // Process the range query information as needed
-            let gte_value = gte
-                .as_ref()
-                .and_then(|gte_term| map_term_to_json_value(gte_term));
-            let lte_value = lte
-                .as_ref()
-                .and_then(|lte_term| map_term_to_json_value(lte_term));
-            return Some(Predicate::Range {
+            let gte_value = gte.as_ref().and_then(map_term_to_json_value);
+            let lte_value = lte.as_ref().and_then(map_term_to_json_value);
+            Some(Predicate::Range {
                 field,
                 gte: gte_value,
                 lte: lte_value,
             })
-        },
+        }
         _ => {
             // Handle other query types
-            return None; // Skip unsupported query types
+            None // Skip unsupported query types
         }
     }
 }
@@ -106,7 +101,7 @@ fn walk_aggregations_and_extract_info(
     aggregations: &dsl::Aggregations,
 ) -> Option<(FieldName, AggregationType, Option<GroupBySpec>)> {
     // Traverse the aggregations in the AST and extracting relevant information. Extract the first valid aggregation type found, along with any associated group by specifications.
-    for (_, agg) in aggregations {
+    for agg in aggregations.values() {
         match agg {
             dsl::Aggregation::MultiTerms(terms_agg) => {
                 // Extract information from the terms aggregation
@@ -141,7 +136,7 @@ fn walk_aggregations_and_extract_info(
             }
             other => {
                 // Handle other aggregation types
-                let (target_field, aggregation_type) = extract_aggregation_info(&other)?;
+                let (target_field, aggregation_type) = extract_aggregation_info(other)?;
                 return Some((target_field, aggregation_type, None));
             }
         }
@@ -151,8 +146,8 @@ fn walk_aggregations_and_extract_info(
 
 fn find_aggregation_info(aggregations: &dsl::Aggregations) -> Option<(FieldName, AggregationType)> {
     // Placeholder for extracting specific information from an aggregation node
-    for (_, agg) in aggregations {
-        let (field, aggregation_type) = extract_aggregation_info(&agg)?;
+    if let Some((_, agg)) = aggregations.iter().next() {
+        let (field, aggregation_type) = extract_aggregation_info(agg)?;
         return Some((field, aggregation_type));
     }
     None // Return None if no relevant aggregation information is found
@@ -202,11 +197,11 @@ fn map_term_to_json_value(term: &dsl::Term) -> Option<TermValue> {
             let value_str = value.to_string(); // Convert the term value to a string representation
             Some(TermValue::String(value_str))
         }
-        dsl::Term::Float32(value) => Some(TermValue::Float(value.clone() as f64)),
-        dsl::Term::Float64(value) => Some(TermValue::Float(value.clone())),
-        dsl::Term::PositiveNumber(value) => Some(TermValue::UnsignedInt(value.clone())),
-        dsl::Term::NegativeNumber(value) => Some(TermValue::Int(value.clone())),
-        dsl::Term::Boolean(value) => Some(TermValue::Boolean(value.clone())),
+        dsl::Term::Float32(value) => Some(TermValue::Float(*value as f64)),
+        dsl::Term::Float64(value) => Some(TermValue::Float(*value)),
+        dsl::Term::PositiveNumber(value) => Some(TermValue::UnsignedInt(*value)),
+        dsl::Term::NegativeNumber(value) => Some(TermValue::Int(*value)),
+        dsl::Term::Boolean(value) => Some(TermValue::Boolean(*value)),
     }
 }
 
diff --git a/asap-query-engine/src/engines/simple_engine/elastic.rs b/asap-query-engine/src/engines/simple_engine/elastic.rs
index 1b6e1c14..572840b9 100644
--- a/asap-query-engine/src/engines/simple_engine/elastic.rs
+++ b/asap-query-engine/src/engines/simple_engine/elastic.rs
@@ -5,8 +5,10 @@
 use super::SimpleEngine;
 use super::{QueryExecutionContext, QueryMetadata, QueryTimestamps};
 use crate::engines::query_result::QueryResult;
-use elastic_dsl_utilities::ast_parsing::{self, ElasticDSLQueryInfo, GroupBySpec, AggregationType, Predicate};
-use elastic_dsl_utilities::datemath::{range_query_to_time_range};
+use elastic_dsl_utilities::ast_parsing::{
+    self, AggregationType, ElasticDSLQueryInfo, GroupBySpec, Predicate,
+};
+use elastic_dsl_utilities::datemath::range_query_to_time_range;
 use promql_utilities::data_model::KeyByLabelNames;
 use promql_utilities::query_logics::enums::Statistic;
 use std::collections::HashMap;
@@ -55,7 +57,7 @@ impl SimpleEngine {
 
         let spatial_filter = String::new(); // Placeholder - extract from query if applicable
 
-        // Parse time range information from first query predicate if available, otherwise default to entire history up to query_time. 
+        // Parse time range information from first query predicate if available, otherwise default to entire history up to query_time.
         let timestamps = self.resolve_query_time_range_elastic(query_time, query_info);
 
         let query_plan = self
@@ -106,7 +108,9 @@ impl SimpleEngine {
         // By default, we only include grouping labels in the output for ES DSL.
         let query_output_labels = match &query_info.group_by_buckets {
             Some(GroupBySpec::Fields(fields)) => KeyByLabelNames::new(fields.clone()),
-            Some(GroupBySpec::Filters(_)) => { return None; } // We don't support filter-based group by in ES DSL for now, so return None to indicate unsupported query pattern.
+            Some(GroupBySpec::Filters(_)) => {
+                return None;
+            } // We don't support filter-based group by in ES DSL for now, so return None to indicate unsupported query pattern.
             None => KeyByLabelNames::empty(),
         };
 
@@ -124,9 +128,9 @@ impl SimpleEngine {
         let mut query_kwargs = HashMap::new();
         if let AggregationType::Percentiles(percents) = aggregation {
             // Get first value from percents array since we only support one quantile argument for now.
-                let quantile = percents.first()?;
-                // ES percentiles are specified as values between 0 and 100, but we want to convert to 0-1 range for our internal representation.
-                query_kwargs.insert("quantile".to_string(), (quantile / 100.0).to_string());
+            let quantile = percents.first()?;
+            // ES percentiles are specified as values between 0 and 100, but we want to convert to 0-1 range for our internal representation.
+            query_kwargs.insert("quantile".to_string(), (quantile / 100.0).to_string());
         }
 
         let metadata = QueryMetadata {
@@ -149,16 +153,22 @@ impl SimpleEngine {
         let mut start_timestamp: u64 = 0;
         let mut end_timestamp: u64 = query_time;
 
-        let predicate = query_info
-            .predicates
-            .first().clone(); // For now, we only look at the first predicate for time range information. We can extend this to support multiple predicates and more complex logic in the future.
+        let predicate = query_info.predicates.first(); // For now, we only look at the first predicate for time range information. We can extend this to support multiple predicates and more complex logic in the future.
 
         match predicate {
-            Some(Predicate::Range { field: _, gte: _, lte: _ }) => {
+            Some(Predicate::Range {
+                field: _,
+                gte: _,
+                lte: _,
+            }) => {
                 // If we have a range predicate, we can try to extract time range information from it.
                 // For now, we assume that any range predicate applies to the timestamp field, but we could add more complex logic here to determine which field is the timestamp field.
-                debug!("Found range predicate in query, attempting to extract time range information");
-                if let Some(resolved_range) = range_query_to_time_range(predicate.unwrap(), query_time as i64) {
+                debug!(
+                    "Found range predicate in query, attempting to extract time range information"
+                );
+                if let Some(resolved_range) =
+                    range_query_to_time_range(predicate.unwrap(), query_time as i64)
+                {
                     debug!(
                         "Parsed time range from range predicate: start={} end={}",
                         resolved_range.gte_ms.unwrap_or(0),

From ccee90cbe53ac74e56dfe70f607fb8bf56f687b1 Mon Sep 17 00:00:00 2001
From: Eric Wang <eswang1202@gmail.com>
Date: Sun, 26 Apr 2026 20:18:34 -0400
Subject: [PATCH 7/7] Point elastic_dsl_utilities to renamed DSL parser crate.

---
 Cargo.lock                                    | 23 ++++++++++---------
 .../rs/elastic_dsl_utilities/Cargo.toml       |  2 +-
 .../src/ast_parsing/extract_info.rs           |  2 +-
 3 files changed, 14 insertions(+), 13 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 5630c4f8..3bd79354 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1679,7 +1679,18 @@ name = "elastic_dsl_utilities"
 version = "0.3.0"
 dependencies = [
  "chrono",
- "opensearch-dsl",
+ "elasticsearch-dsl-ast",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "elasticsearch-dsl-ast"
+version = "0.1.0"
+source = "git+https://github.com/ProjectASAP/elasticsearch-dsl-ast#f1e1e3565be1c04d64ba0234def1d3b1042838e9"
+dependencies = [
+ "chrono",
+ "num-traits",
  "serde",
  "serde_json",
 ]
@@ -3029,16 +3040,6 @@ version = "11.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
 
-[[package]]
-name = "opensearch-dsl"
-version = "0.1.0"
-dependencies = [
- "chrono",
- "num-traits",
- "serde",
- "serde_json",
-]
-
 [[package]]
 name = "openssl"
 version = "0.10.76"
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
index d07f01a2..db353488 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
@@ -7,4 +7,4 @@ version.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 chrono.workspace = true
-opensearch-dsl = { version = "0.1.0", path = "../../../../../opensearch-dsl" }
+elasticsearch-dsl-ast = { version = "0.1.0", git = "https://github.com/ProjectASAP/elasticsearch-dsl-ast" }
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
index 754f437c..299af3b1 100644
--- a/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/ast_parsing/extract_info.rs
@@ -2,7 +2,7 @@ use crate::ast_parsing::query_info::{
     AggregationType, ElasticDSLQueryInfo, FieldName, GroupBySpec, Predicate, TermValue,
 };
 use crate::helpers::strip_keyword_suffix;
-use opensearch_dsl::{self as dsl};
+use elasticsearch_dsl_ast::{self as dsl};
 use serde_json;
 
 pub fn extract_query_info(query: &str) -> Option<ElasticDSLQueryInfo> {