diff --git a/Cargo.lock b/Cargo.lock
index 934545f..1ebbc7b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1613,6 +1613,15 @@ version = "1.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
 
+[[package]]
+name = "elastic_dsl_utilities"
+version = "0.1.0"
+dependencies = [
+ "chrono",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "encoding_rs"
 version = "0.8.35"
@@ -3551,6 +3560,7 @@ dependencies = [
  "datafusion",
  "datafusion_summary_library",
  "dsrs",
+ "elastic_dsl_utilities",
  "flate2",
  "form_urlencoded",
  "futures",
diff --git a/Cargo.toml b/Cargo.toml
index f396d96..a283fea 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -4,6 +4,7 @@ members = [
     "asap-common/sketch-core",
     "asap-common/dependencies/rs/promql_utilities",
     "asap-common/dependencies/rs/sql_utilities",
+    "asap-common/dependencies/rs/elastic_dsl_utilities",
     "asap-common/dependencies/rs/sketch_db_common",
     "asap-common/dependencies/rs/datafusion_summary_library",
     "asap-query-engine",
@@ -34,5 +35,6 @@ promql_utilities = { path = "asap-common/dependencies/rs/promql_utilities" }
 sql_utilities = { path = "asap-common/dependencies/rs/sql_utilities" }
 sketch_db_common = { path = "asap-common/dependencies/rs/sketch_db_common" }
 datafusion_summary_library = { path = "asap-common/dependencies/rs/datafusion_summary_library" }
+elastic_dsl_utilities = { path = "asap-common/dependencies/rs/elastic_dsl_utilities" }
 asap_planner = { path = "asap-planner-rs" }
 indexmap = { version = "2.0", features = ["serde"] }
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
new file mode 100644
index 0000000..3726cab
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/Cargo.toml
@@ -0,0 +1,9 @@
+[package]
+name = "elastic_dsl_utilities"
+edition.workspace = true
+version.workspace = true
+
+[dependencies]
+serde.workspace = true
+serde_json.workspace = true
+chrono.workspace = true
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
new file mode 100644
index 0000000..c03929f
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/lib.rs
@@ -0,0 +1,7 @@
+pub mod parsing;
+pub mod pattern;
+pub mod types;
+
+pub use parsing::*;
+pub use pattern::{classify, parse_and_classify};
+pub use types::*;
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/src/parsing.rs b/asap-common/dependencies/rs/elastic_dsl_utilities/src/parsing.rs
new file mode 100644
index 0000000..3963c1f
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/parsing.rs
@@ -0,0 +1,439 @@
+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
new file mode 100644
index 0000000..2924056
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/pattern.rs
@@ -0,0 +1,403 @@
+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
new file mode 100644
index 0000000..5c1269d
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/src/types.rs
@@ -0,0 +1,286 @@
+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),
+            })
+        );
+    }
+}
diff --git a/asap-common/dependencies/rs/elastic_dsl_utilities/supported_es_queries.md b/asap-common/dependencies/rs/elastic_dsl_utilities/supported_es_queries.md
new file mode 100644
index 0000000..46e2002
--- /dev/null
+++ b/asap-common/dependencies/rs/elastic_dsl_utilities/supported_es_queries.md
@@ -0,0 +1,147 @@
+# (ASAP) Potential Elasticsearch Queries to be Supported
+
+Info dump about what Elasticsearch queries could be supported by ASAP based on my understanding of the query engine.
+
+## Supportable Query Types
+
+Elastic DSL single level aggregation queries that explicitly don't return any records (`"size": 0`), with basic filtering by column label values, seem the most similar to the Prometheus style queries, so they could maybe be translated.
+
+Here is a list of ES aggregation functions that seem to map well to the current sketches available:
+
+- `"percentiles"` (quantiles)  
+- `"min"`  
+- `"max"`  
+- `"sum"`  
+- `"avg"`
+
+## Query Templates
+
+Templates outlining the basic structure for some Elasticsearch queries. For the following examples, we use `${name}` syntax to denote dynamic/user provided variables. In each query, the time range specifier would be optional.
+
+### 1\. Simple Aggregation
+
+Compute summary statistics on all data for one or more data columns (metrics), optionally with a time range/predicate.
+
+```json
+{
+    "size": 0,
+    "query": {
+        "range": {
+            "@timestamp": {
+                "gte": "now-30s",
+                "lte": "now"
+            }
+        }
+    },                                          
+    "aggs": {
+        "${agg_result1}": {
+            "${agg_type1}": {
+                "field": "${metric_name1}",
+            }
+        },
+        "${agg_result2}": {
+            "${agg_type2}": {
+                "field": "${metric_name2}",
+                "${param1}": "${arg1}"
+            }
+        }
+    }
+}
+```
+
+This is semantically equivalent to the following SQL.
+
+```sql
+SELECT 
+    AGG1(metric_name1) AS agg_result1,
+    AGG2(metric_name2) AS agg_result2
+FROM table_name
+WHERE time_created >= NOW() - INTERVAL '30 seconds';
+```
+
+### 2\. Groupby Aggregation
+
+Compute one or more summary statistics, grouped by one or more labels, optionally with a time range/predicate.
+
+#### Group By (One Label)
+```json
+{
+    "size": 0,
+    "query": {
+        "bool": {
+            "filter": {
+                "term": { "${field1}.keyword": "${value1}" },
+                "range": {
+                    "@timestamp": {
+                        "gte": "now-30s",
+                        "lte": "now"
+                    }
+                }
+            }
+        }
+    },                                          
+    "aggs": {
+        "${grouped_result1}": {
+            "terms": {
+                "field": "${field1}.keyword",
+            },
+            "aggs": {
+                "${agg_result1}": {
+                    "${agg_type1}": {
+                        "field": "${metric_name1}",
+                    }
+                }
+            }
+        }
+    }
+}
+```
+
+#### Group By (Multi-Label)
+
+```json
+{
+    "size": 0,
+    "query": {
+        "bool": {
+            "filter": {
+                "term": { "${field1}.keyword": "${value1}" },
+                "range": {
+                    "@timestamp": {
+                        "gte": "now-30s",
+                        "lte": "now"
+                    }
+                }
+            }
+        }
+    },                                          
+    "aggs": {
+        "${grouped_result1}": {
+            "multi_terms": {
+                "terms": [
+                    { "field": "${field1}.keyword"},
+                    { "field": "${field2}.keyword"}
+                ]
+            },
+            "aggs": {
+                "${agg_result1}": {
+                    "${agg_type1}": {
+                        "field": "${metric_name1}",
+                    }
+                }
+            }
+        }
+    }
+}
+```
+
+The corresponding SQL (multi-label case) is as follows.
+
+```sql
+SELECT 
+    AGG1(metric_name1) AS agg_result1,
+    AGG2(metric_name2) AS agg_result2
+FROM table_name
+WHERE time_created >= NOW() - INTERVAL '30 seconds'
+GROUP BY field1, field2;
+```
diff --git a/asap-query-engine/Cargo.toml b/asap-query-engine/Cargo.toml
index acf082c..6c031ae 100644
--- a/asap-query-engine/Cargo.toml
+++ b/asap-query-engine/Cargo.toml
@@ -56,6 +56,7 @@ lazy_static = "1.4"
 zstd = "0.13"
 reqwest = { version = "0.11", features = ["json"] }
 tracing-appender = "0.2"
+elastic_dsl_utilities.workspace = true
 sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust", rev = "440427438fdaf3ac2298b53ee148f9e12a64ffcc" }
 
 [dev-dependencies]
diff --git a/asap-query-engine/src/engines/simple_engine.rs b/asap-query-engine/src/engines/simple_engine.rs
index be8ceba..5f431d5 100644
--- a/asap-query-engine/src/engines/simple_engine.rs
+++ b/asap-query-engine/src/engines/simple_engine.rs
@@ -31,6 +31,9 @@ use sqlparser::parser::Parser as parser;
 
 // SQL issue: refactor simpleengine to create matchresult similar to SQLquerydata
 
+use elastic_dsl_utilities::pattern::parse_and_classify;
+use elastic_dsl_utilities::types::{EsDslQueryPattern, GroupBySpec, MetricAggType};
+
 // Type alias for merged outputs (single aggregate per key after merging)
 type MergedOutputsMap = HashMap<Option<KeyByLabelValues>, Box<dyn AggregateCore>>;
 
@@ -1472,13 +1475,191 @@ impl SimpleEngine {
         match self.query_language {
             QueryLanguage::promql => self.handle_query_promql(query, time),
             QueryLanguage::sql => self.handle_query_sql(query, time),
-            QueryLanguage::elastic_querydsl => self.handle_query_elastic(),
+            QueryLanguage::elastic_querydsl => self.handle_query_elastic(query, time),
             QueryLanguage::elastic_sql => self.handle_query_sql(query, time),
         }
     }
 
-    pub fn handle_query_elastic(&self) -> Option<(KeyByLabelNames, QueryResult)> {
-        None
+    pub fn handle_query_elastic(
+        &self,
+        query: String,
+        time: f64,
+    ) -> Option<(KeyByLabelNames, QueryResult)> {
+        let context = self.build_query_execution_context_elastic(query, time)?;
+        debug!(
+            "Built execution context for ElasticSearch query {:?}",
+            context
+        );
+        // Execute complete query pipeline
+        let results = self
+            .execute_query_pipeline(&context, false) // SQL: topk disabled
+            .map_err(|e| {
+                warn!("Query execution failed: {}", e);
+                e
+            })
+            .ok()?;
+
+        Some((
+            context.metadata.query_output_labels,
+            QueryResult::vector(results, context.query_time),
+        ))
+    }
+
+    pub fn build_query_execution_context_elastic(
+        &self,
+        query: String,
+        time: f64,
+    ) -> Option<QueryExecutionContext> {
+        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);
+            }
+        }
+
+        // 3. Convert parsed query into execution context components (labels, statistic, kwargs, metadata, store query plan, etc.)
+
+        // TODO: Figure out how to handle query configuration for ElasticSearch queries.
+        let query_config = self.find_query_config(&query)?;
+        let agg_info = self.get_aggregation_id_info(query_config);
+
+        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 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);
+
+        let query_plan = self
+            .create_store_query_plan(&metric, &timestamps, &agg_info)
+            .map_err(|e| {
+                warn!("Failed to create store query plan: {}", e);
+                e
+            })
+            .ok()?;
+
+        let grouping_labels = self
+            .streaming_config
+            .get_aggregation_config(agg_info.aggregation_id_for_value)
+            .map(|config| config.grouping_labels.clone())
+            .unwrap_or_else(|| query_metadata.query_output_labels.clone());
+
+        let aggregated_labels = self
+            .streaming_config
+            .get_aggregation_config(agg_info.aggregation_id_for_key)
+            .map(|config| config.aggregated_labels.clone())
+            .unwrap_or_else(KeyByLabelNames::empty);
+
+        Some(QueryExecutionContext {
+            metric,
+            metadata: query_metadata,
+            store_plan: query_plan.clone(),
+            agg_info: agg_info.clone(),
+            do_merge,
+            spatial_filter,
+            query_time,
+            grouping_labels,
+            aggregated_labels,
+        })
+    }
+
+    fn build_query_metadata_elastic(
+        &self,
+        query_pattern: &EsDslQueryPattern,
+    ) -> 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).
+
+        // Figure out aggregation type and what labels are included in output.
+        // 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();
+
+        // 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()),
+            None => KeyByLabelNames::empty(),
+        };
+
+        let metric = aggregation.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 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 metadata = QueryMetadata {
+            query_output_labels: query_output_labels.clone(),
+            statistic_to_compute,
+            query_kwargs: query_kwargs.clone(),
+        };
+        Some((metric, metadata))
+    }
+
+    pub fn resolve_query_time_range_elastic(
+        &self,
+        query_time: u64,
+        query_pattern: EsDslQueryPattern,
+    ) -> 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).
+        // If no time range is specified, default to entire history up to query_time.
+
+        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");
+            }
+        };
+
+        QueryTimestamps {
+            start_timestamp,
+            end_timestamp,
+        }
     }
 
     // /// Try to extract sketch query components from a PromQL query string.
diff --git a/asap-query-engine/src/tests/elastic_dsl_query_tests.rs b/asap-query-engine/src/tests/elastic_dsl_query_tests.rs
new file mode 100644
index 0000000..09c7ac5
--- /dev/null
+++ b/asap-query-engine/src/tests/elastic_dsl_query_tests.rs
@@ -0,0 +1,384 @@
+//! Tests for Elastic DSL query handling in the ASAP Query Engine.
+//!
+//! Verifies that various patterns of Elastic DSL queries are correctly classified and processed by the engine,
+//! including correct handling of time ranges, label groupings, and unsupported query patterns.
+
+#[cfg(test)]
+mod tests {
+    use crate::data_model::{AggregateCore, KeyByLabelValues};
+    use crate::precompute_operators::DatasketchesKLLAccumulator;
+    use crate::tests::test_utilities::create_engine_multi_timestamp;
+    use crate::QueryResult;
+    use serde_json::json;
+
+    fn create_kll_accumulator_with_values(values: &[f64]) -> DatasketchesKLLAccumulator {
+        let mut kll = DatasketchesKLLAccumulator::new(200);
+        for &v in values {
+            kll._update(v);
+        }
+        kll
+    }
+
+    #[allow(clippy::type_complexity)]
+    fn create_kll_data_with_timestamps(
+        timestamps: &[u64],
+        label_values: Vec<Option<Vec<String>>>,
+    ) -> Vec<(u64, Option<Vec<String>>, Box<dyn AggregateCore>)> {
+        let mut result = Vec::new();
+        for label_value in label_values {
+            println!("Creating KLL histogram for label value: {label_value:?}");
+            result.extend(timestamps.iter().enumerate().map(|(i, &timestamp)| {
+                (
+                    timestamp,
+                    label_value.clone(),
+                    Box::new(create_kll_accumulator_with_values(
+                        (i * 100 + 1..=i * 100 + 100)
+                            .map(|v| v as f64)
+                            .collect::<Vec<f64>>()
+                            .as_slice(),
+                    )) as Box<dyn AggregateCore>,
+                )
+            }))
+        }
+        result
+    }
+
+    #[test]
+    fn test_esdsl_simple_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();
+
+        // Elastic DSL query (batch filtered).
+        let elastic_query = json!({
+            "size": 0,
+            "aggs": {
+                "out": {
+                    "percentiles": {
+                        "field": "http_requests",
+                        "percents": [90]
+                    }
+                }
+            }
+        });
+
+        // Create data. Engine expects 1 second (1000 ms) intervals.
+        let timestamps = vec![999_000, 1_000_000];
+        let label_values = vec![
+            Some(Vec::new()), // No labels for this test
+        ];
+        let kll_data = create_kll_data_with_timestamps(&timestamps, label_values);
+
+        let engine = create_engine_multi_timestamp(
+            "http_requests",
+            "DatasketchesKLLAccumulator",
+            Vec::new(), // No labels for this test
+            kll_data,
+            &elastic_query.to_string(),
+        );
+
+        let time = 1_000.0; // Arbitrary timestamp for testing
+        let output = engine.handle_query_elastic(elastic_query.to_string(), time);
+        if let Some((_, result)) = output {
+            match &result {
+                QueryResult::Vector(instant) => {
+                    assert_eq!(instant.values.len(), 1);
+                    let sample = &instant.values[0];
+                    assert_eq!(sample.labels, KeyByLabelValues::new()); // No labels expected
+                }
+                _ => {
+                    panic!("Expected Vector result");
+                }
+            }
+            let result_json = serde_json::to_string(&result).unwrap();
+            println!("Query Result: {result_json}");
+        } else {
+            panic!("Expected query result, got None");
+        }
+    }
+
+    #[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();
+
+        // Elastic DSL query (batch filtered).
+        let elastic_query = json!({
+            "size": 0,
+            "aggs": {
+                "out": {
+                    "terms": {
+                        "field": "host.keyword"
+                    },
+                    "aggs": {
+                        "out": {
+                            "percentiles": {
+                                "field": "http_requests",
+                                "percents": [90]
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
+        // Create data. Engine expects 1 second (1000 ms) intervals.
+        let timestamps = vec![999_000, 1_000_000];
+        let label_values = vec![
+            Some(vec!["host-a".to_string()]),
+            Some(vec!["host-b".to_string()]),
+        ];
+        let kll_data = create_kll_data_with_timestamps(&timestamps, label_values);
+
+        let engine = create_engine_multi_timestamp(
+            "http_requests",
+            "DatasketchesKLLAccumulator",
+            vec!["host"],
+            kll_data,
+            &elastic_query.to_string(),
+        );
+
+        let time = 1_000.0; // Arbitrary timestamp for testing
+        let output = engine.handle_query_elastic(elastic_query.to_string(), time);
+        if let Some((_, result)) = output {
+            match &result {
+                QueryResult::Vector(instant) => {
+                    assert_eq!(instant.values.len(), 2);
+                    let label_combinations = vec!["host-a", "host-b"];
+                    let mut found_combinations = Vec::new();
+                    for sample in instant.values.iter() {
+                        let label_string = sample.labels.to_semicolon_str();
+                        found_combinations.push(label_string);
+                    }
+                    for expected in label_combinations {
+                        assert!(
+                            found_combinations.contains(&expected.to_string()),
+                            "Expected label combination not found: {expected}"
+                        );
+                    }
+                }
+                _ => {
+                    panic!("Expected Vector result");
+                }
+            }
+            let result_json = serde_json::to_string(&result).unwrap();
+            println!("Query Result: {result_json}");
+        } else {
+            panic!("Expected query result, got None");
+        }
+    }
+
+    #[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();
+
+        // Elastic DSL query (batch filtered).
+        let elastic_query = json!({
+            "size": 0,
+            "aggs": {
+                "out": {
+                    "multi_terms": {
+                        "terms": [
+                            {
+                                "field": "host.keyword"
+                            },
+                            {
+                                "field": "region.keyword"
+                            }
+                        ]
+                    },
+                    "aggs": {
+                        "out": {
+                            "percentiles": {
+                                "field": "http_requests",
+                                "percents": [90]
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
+        // Create data. Engine expects 1 second (1000 ms) intervals.
+        let timestamps = vec![998_000, 999_000, 1_000_000];
+        let label_values = vec![
+            Some(vec!["host-a".to_string(), "region-a".to_string()]),
+            Some(vec!["host-b".to_string(), "region-b".to_string()]),
+            Some(vec!["host-c".to_string(), "region-c".to_string()]),
+            Some(vec!["host-b".to_string(), "region-c".to_string()]),
+        ];
+        let kll_data = create_kll_data_with_timestamps(&timestamps, label_values);
+
+        let engine = create_engine_multi_timestamp(
+            "http_requests",
+            "DatasketchesKLLAccumulator",
+            vec!["host", "region"],
+            kll_data,
+            &elastic_query.to_string(),
+        );
+
+        let time = 1_000.0; // Arbitrary timestamp for testing
+        let output = engine.handle_query_elastic(elastic_query.to_string(), time);
+        if let Some((_, result)) = output {
+            match &result {
+                QueryResult::Vector(instant) => {
+                    assert_eq!(instant.values.len(), 4);
+                    let label_combinations = vec![
+                        "host-a;region-a",
+                        "host-b;region-b",
+                        "host-c;region-c",
+                        "host-b;region-c",
+                    ];
+                    let mut found_combinations = Vec::new();
+                    for sample in instant.values.iter() {
+                        let label_string = sample.labels.to_semicolon_str();
+                        found_combinations.push(label_string);
+                    }
+                    for expected in label_combinations {
+                        assert!(
+                            found_combinations.contains(&expected.to_string()),
+                            "Expected label combination not found: {expected}"
+                        );
+                    }
+                }
+                _ => {
+                    panic!("Expected Vector result");
+                }
+            }
+            let result_json = serde_json::to_string(&result).unwrap();
+            println!("Query Result: {result_json}");
+        } else {
+            panic!("Expected query result, got None");
+        }
+    }
+
+    #[test]
+    fn test_esdsl_unsupported_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();
+
+        // Elastic DSL query (batch filtered).
+        let elastic_query = json!({
+            "size": 0,
+            "aggs": {
+                "out": {
+                    "multi_terms": {
+                        "terms": [
+                            {
+                                "field": "host.keyword"
+                            },
+                            {
+                                "field": "region.keyword"
+                            }
+                        ]
+                    },
+                    "aggs": {
+                        "out": {
+                            "fake_aggregation": {
+                                "field": "http_requests"
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
+        // Create data. Engine expects 1 second (1000 ms) intervals.
+        let timestamps = vec![998_000, 999_000, 1_000_000];
+        let label_values = vec![
+            Some(vec!["host-a".to_string(), "region-a".to_string()]),
+            Some(vec!["host-b".to_string(), "region-b".to_string()]),
+            Some(vec!["host-c".to_string(), "region-c".to_string()]),
+            Some(vec!["host-b".to_string(), "region-c".to_string()]),
+        ];
+        let kll_data = create_kll_data_with_timestamps(&timestamps, label_values);
+
+        let engine = create_engine_multi_timestamp(
+            "http_requests",
+            "DatasketchesKLLAccumulator",
+            vec!["host", "region"],
+            kll_data,
+            &elastic_query.to_string(),
+        );
+
+        let time = 1_000.0; // Arbitrary timestamp for testing
+        let output = engine.handle_query_elastic(elastic_query.to_string(), time);
+        assert!(
+            output.is_none(),
+            "Expected None for unsupported query, got Some({:?})",
+            output
+        );
+    }
+
+    #[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();
+
+        // Elastic DSL query (batch filtered).
+        let elastic_query = json!({
+            "size": 0,
+            "query": {
+                "range": {
+                    "timestamp": {
+                        "gte": "now-1s",
+                        "lte": "now"
+                    }
+                }
+            },
+            "aggs": {
+                "out": {
+                    "percentiles": {
+                        "field": "http_requests",
+                        "percents": [90]
+                    }
+                }
+            }
+        });
+
+        // Create data. Engine expects 1 second (1000 ms) intervals.
+        let timestamps = vec![998_000, 999_000, 1_000_000];
+        let label_values = vec![
+            Some(Vec::new()), // No labels for this test
+        ];
+        let kll_data = create_kll_data_with_timestamps(&timestamps, label_values);
+
+        let engine = create_engine_multi_timestamp(
+            "http_requests",
+            "DatasketchesKLLAccumulator",
+            Vec::new(), // No labels for this test
+            kll_data,
+            &elastic_query.to_string(),
+        );
+
+        let time = 1_000.0; // Arbitrary timestamp for testing
+        let output = engine.handle_query_elastic(elastic_query.to_string(), time);
+        if let Some((_, result)) = output {
+            match &result {
+                QueryResult::Vector(instant) => {
+                    assert_eq!(instant.values.len(), 1);
+                    let sample = &instant.values[0];
+                    assert_eq!(sample.labels, KeyByLabelValues::new()); // No labels expected
+                    assert_eq!(sample.value, 291.0); // 90th percentile of 200..300 as reported by KLL (skip first two KLL buckets which are outside the resolved time range)
+                }
+                _ => {
+                    panic!("Expected Vector result");
+                }
+            }
+            let result_json = serde_json::to_string(&result).unwrap();
+            println!("Query Result: {result_json}");
+        } else {
+            panic!("Expected query result, got None");
+        }
+    }
+}
diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs
index 4355e45..9988d5f 100644
--- a/asap-query-engine/src/tests/mod.rs
+++ b/asap-query-engine/src/tests/mod.rs
@@ -1,5 +1,6 @@
 pub mod clickhouse_forwarding_tests;
 pub mod datafusion;
+pub mod elastic_dsl_query_tests;
 pub mod elastic_forwarding_tests;
 pub mod prometheus_forwarding_tests;
 pub mod query_equivalence_tests;