tellaro-query-language 2.0.0

//! TQL stats AST to OpenSearch aggregation DSL translator.
//!
//! Converts TQL stats expressions (e.g., `| stats count(*) by event.code`)
//! into OpenSearch aggregation DSL with proper nested terms aggregations
//! for groupby, metric aggregation mappings, and `.keyword` field resolution.

use super::field_mappings::{FieldMappings, FieldType};
use crate::parser::{Aggregation, GroupBy, StatsNode};
use serde_json::{json, Value as JsonValue};

/// The size a `terms` aggregation is asked for when the query requests no limit
/// at all.
///
/// `| stats sum(n) by g` returns EVERY bucket in memory, and `terms` has no
/// unbounded size, so a cap is unavoidable -- but the cap that used to stand in
/// for it was 5 (a "matches Kibana/Grafana/Splunk defaults" bucket size), which
/// is a silent answer to a different question. 10000 matches the size the
/// listing family (`values`/`unique`/`distinct`) has always used here for the
/// same reason, and sits under the `search.max_buckets` default of 65536.
const UNLIMITED_BUCKET_SIZE: usize = 10000;

/// Name of the `bucket_sort` that applies the group-by bucket limit as a SECOND
/// pass after the aggregation modifier. Leading underscore so it can never
/// collide with an alias: TQL aliases come from the `identifier` rule.
const BUCKET_LIMIT_AGG_NAME: &str = "_tql_bucket_limit";

/// Whether OpenSearch will order a `terms` aggregation by this aggregation type
/// with a bare `{name: direction}` path.
///
/// Keyed on the OpenSearch type rather than the TQL function name so
/// `map_aggregation` stays the only place the synonyms are listed.
fn is_single_value_metric(os_type: &str) -> bool {
    matches!(
        os_type,
        "value_count" | "cardinality" | "sum" | "min" | "max" | "avg"
    )
}

/// The name a metric aggregation is emitted under, when no alias was written.
///
/// Single-sourced because it is read TWICE: once to name the metric aggregation
/// and once to build the `terms` order path that ranks buckets by it. The two
/// copies were spelled differently -- the metric key from the LOWERCASED
/// function, the order path from `agg.function` raw -- which is only harmless
/// because `TqlParser::parse_aggregation` lowercases the name as it builds the
/// AST. One edit to that line would have made `SUM(x) top 3 by g` emit an order
/// path naming an aggregation that does not exist, which OpenSearch answers with
/// a 500 rather than a wrong number.
fn agg_alias(agg: &Aggregation, index: usize) -> String {
    agg.alias
        .clone()
        .unwrap_or_else(|| format!("{}_{}", agg.function.to_lowercase(), index))
}

/// Map TQL aggregation function names to OpenSearch aggregation types.
fn map_aggregation(func: &str) -> Option<&'static str> {
    match func {
        "count" => Some("value_count"),
        "unique_count" | "cardinality" => Some("cardinality"),
        "sum" => Some("sum"),
        "min" => Some("min"),
        "max" => Some("max"),
        "average" | "avg" | "mean" => Some("avg"),
        "median" | "med" => Some("percentiles"),
        "std" | "standard_deviation" => Some("extended_stats"),
        "percentile" | "percentiles" | "p" | "pct" => Some("percentiles"),
        "percentile_rank" | "percentile_ranks" | "pct_rank" | "pct_ranks" => {
            Some("percentile_ranks")
        }
        "values" | "unique" | "distinct" => Some("terms"),
        _ => None,
    }
}

/// Resolve a field name to its `.keyword` variant if it's a text field.
///
/// OpenSearch cannot aggregate on text fields — terms aggregations require
/// keyword fields.
fn resolve_aggregation_field(field: &str, field_mappings: Option<&FieldMappings>) -> String {
    if let Some(mappings) = field_mappings {
        // If the base field is text type, use .keyword for aggregation
        if let Some(ft) = mappings.get_field_type(field) {
            if *ft == FieldType::Text {
                return format!("{}.keyword", field);
            }
        }
        // Also check if a .keyword subfield exists
        let keyword_field = format!("{}.keyword", field);
        if mappings.get_field_type(&keyword_field).is_some() {
            return keyword_field;
        }
    }
    field.to_string()
}

/// Build OpenSearch aggregation DSL from a TQL stats AST.
///
/// Returns a JSON object with the `aggs` key containing the aggregation DSL.
pub fn translate_stats(
    stats: &StatsNode,
    field_mappings: Option<&FieldMappings>,
) -> Result<JsonValue, String> {
    if stats.aggregations.is_empty() {
        return Err("No aggregations specified in stats query".to_string());
    }

    let aggs_dsl = if stats.group_by.is_empty() {
        build_simple_aggregations(&stats.aggregations)?
    } else {
        build_grouped_aggregations(&stats.aggregations, &stats.group_by, field_mappings)?
    };

    Ok(json!({ "aggs": aggs_dsl }))
}

/// Build simple aggregations without grouping.
fn build_simple_aggregations(aggregations: &[Aggregation]) -> Result<JsonValue, String> {
    let mut aggs = serde_json::Map::new();

    for (i, agg) in aggregations.iter().enumerate() {
        let func = agg.function.to_lowercase();
        let field = agg.field.as_deref().unwrap_or("*");
        let alias = agg_alias(agg, i);

        let agg_dsl = build_single_aggregation(&func, field, agg)?;
        aggs.insert(alias, agg_dsl);
    }

    Ok(JsonValue::Object(aggs))
}

/// Build a single aggregation DSL object.
fn build_single_aggregation(
    func: &str,
    field: &str,
    agg: &Aggregation,
) -> Result<JsonValue, String> {
    // Special case: count(*)
    if func == "count" && field == "*" {
        return Ok(json!({ "value_count": { "field": "_id" } }));
    }

    let os_type =
        map_aggregation(func).ok_or_else(|| format!("Unknown aggregation function: {}", func))?;

    match func {
        "median" | "med" => Ok(json!({ "percentiles": { "field": field, "percents": [50.0] } })),
        "std" | "standard_deviation" => Ok(json!({ "extended_stats": { "field": field } })),
        "percentile" | "percentiles" | "p" | "pct" => {
            let percents = agg
                .percentile_values
                .as_ref()
                .cloned()
                .unwrap_or_else(|| vec![50.0]);
            Ok(json!({ "percentiles": { "field": field, "percents": percents } }))
        }
        "percentile_rank" | "percentile_ranks" | "pct_rank" | "pct_ranks" => {
            let values = agg.rank_values.as_ref().cloned().unwrap_or_default();
            if values.is_empty() {
                return Err("percentile_rank requires at least one value".to_string());
            }
            Ok(json!({ "percentile_ranks": { "field": field, "values": values } }))
        }
        "values" | "unique" | "distinct" => Ok(
            // The same number for the same reason as every other terms bucket
            // in this file; spelled out here it was a second, silent definition
            // 145 lines from the one that is named.
            json!({ "terms": { "field": field, "size": UNLIMITED_BUCKET_SIZE } }),
        ),
        _ => {
            // Direct mapping (count, sum, min, max, avg, cardinality, etc.)
            Ok(json!({ os_type: { "field": field } }))
        }
    }
}

/// The `terms` order path for the AGGREGATION-level top-N modifier, or a refusal.
///
/// Mirrors `StatsEvaluator::apply_modifiers`: the FIRST aggregation carrying a
/// modifier wins and the rest are ignored, `top` is descending and `bottom`
/// ascending, and the limit defaults to 10. The key it ranks on is the
/// aggregate VALUE -- which is the entire difference between this modifier and
/// the group-by bucket limit, which ranks by `doc_count`.
///
/// Returns `Ok(None)` when no aggregation carries a modifier.
fn resolve_aggregation_modifier(
    aggregations: &[Aggregation],
    group_levels: usize,
) -> Result<Option<(String, &'static str, usize)>, String> {
    for (i, agg) in aggregations.iter().enumerate() {
        let Some(modifier) = agg.modifier.as_deref() else {
            continue;
        };
        let limit = agg.limit.unwrap_or(10);

        // A `terms` aggregation ranks its OWN buckets. With several group-by
        // levels the in-memory engine ranks the flattened CROSS PRODUCT -- a
        // global ordering over (department, role) pairs -- and nested `terms`
        // cannot produce one. Measured against OpenSearch 2.19.4: ordering the
        // outer level by an inner metric is rejected outright ("Invalid
        // aggregation order path [g2>s] ... [g2] is not single-bucket"), and so
        // is ordering it by a name that only exists one level down -- which is
        // exactly what this function replaced, so the DSL emitted here for a
        // multi-level grouping with a modifier was a 500, not a wrong number.
        if group_levels > 1 {
            return Err(format!(
                "'{modifier} {limit}' on an aggregation cannot be pushed down alongside \
                 {group_levels} group-by fields: OpenSearch ranks the buckets of one `terms` \
                 aggregation, while the in-memory engine ranks the flattened cross product of \
                 all group-by levels. Group by a single field, or move the limit onto the \
                 group-by field (`by <field> {modifier} {limit}`), which ranks by doc_count."
            ));
        }

        // `top 0` means ZERO buckets in memory (`results[..0]`). OpenSearch
        // rejects both ways of asking for that: `terms` replies "[size] must be
        // greater than 0" and `bucket_sort` "[size] must be a positive integer".
        if limit == 0 {
            return Err(format!(
                "'{modifier} 0' on an aggregation asks for zero buckets, which OpenSearch cannot \
                 express: `terms` requires size > 0 and `bucket_sort` requires a positive size."
            ));
        }

        // Ordering a `terms` aggregation by a MULTI-VALUE metric needs the
        // sub-metric spelled out (`m.50`, `m.std_deviation`); OpenSearch rejects
        // the bare name with "When ordering on a multi-value metrics aggregation
        // a metric name must be specified". Which sub-metric is not a formatting
        // question -- `median` is one percentile of possibly several and `std`
        // has six `std_deviation*` fields -- so choosing one here would be
        // choosing a ranking the in-memory engine may not share. The listing
        // family (`values`/`unique`/`distinct`) maps to a bucket aggregation and
        // is not orderable at all.
        let func = agg.function.to_lowercase();
        let os_type = map_aggregation(&func)
            .ok_or_else(|| format!("Unknown aggregation function: {}", func))?;
        if !is_single_value_metric(os_type) {
            return Err(format!(
                "'{modifier} {limit}' cannot rank buckets by '{func}': OpenSearch cannot order a \
                 `terms` aggregation by a value that is not a single-value metric. Rank by count, \
                 sum, min, max, avg or cardinality, or move the limit onto the group-by field \
                 (`by <field> {modifier} {limit}`)."
            ));
        }

        let direction = if modifier == "top" { "desc" } else { "asc" };
        return Ok(Some((agg_alias(agg, i), direction, limit)));
    }

    Ok(None)
}

/// Build aggregations with grouping (nested terms aggregations).
///
/// # The two top-N modifiers, and what this emits for each
///
/// `Aggregation::limit` was never read here at all, so `| stats sum(n) top 3 by g`
/// emitted `"size": 5` -- three buckets in memory, five against a cluster, and no
/// error either way. The un-modified form diverged too: in memory it returns EVERY
/// bucket and the DSL asked for five.
///
/// Emitted DSL, single group-by level:
///
/// ```text
/// | stats sum(n) by g                terms{size: UNLIMITED_BUCKET_SIZE}
/// | stats sum(n) top 3 by g          terms{size: 3, order:{sum_0: desc}}
/// | stats sum(n, bottom 3) by g      terms{size: 3, order:{sum_0: asc}}
/// | stats count() by g top 3         terms{size: 3, order:{_count: desc}}
/// | stats sum(n) bottom 3 by g top 2 terms{size: 3, order:{sum_0: asc}}
///                                      + bucket_sort{_count desc, size 2}
/// ```
///
/// The `bucket_sort` sub-aggregation is the SECOND pass, and its presence is what
/// preserves the pass ORDER: the in-memory engine applies the aggregation modifier
/// first and the bucket limit second, and swapping them yields a disjoint SET
/// rather than a re-ordering. `bucket_sort` contributes nothing to the response --
/// it re-orders and truncates its parent's bucket list in place.
///
/// # Where the cluster CANNOT be made to agree
///
/// Refused outright, because the DSL would be rejected by the cluster or would
/// answer a different question silently -- see `resolve_aggregation_modifier`.
///
/// Emitted, but KNOWN to differ. Measured against OpenSearch 2.19.4 over
/// `cross_language_tests/fixtures/data/user_records.json` and pinned by
/// `tql/tests/stats_top_n_pushdown_live.rs`, so the difference cannot drift
/// unobserved:
///
/// * **Tie-breaks.** `terms` breaks a tie on its order key with `_key` ascending;
///   both in-memory engines break it in first-appearance (record) order.
///   `| stats sum(salary) by department top 5` cuts inside a four-way
///   `doc_count = 2` tie, so the fifth bucket is `Data Science` from the cluster
///   and `Finance` in memory. No `terms` order expresses "the order the documents
///   were indexed in", so this cannot be reconciled from this side.
/// * **Un-modified bucket ORDER.** With no modifier the SET now agrees, but
///   `terms` returns buckets by `_count` descending while the in-memory engines
///   return them in first-appearance order.
/// * **Multi-level bucket limits.** `| stats count() by department top 3, role top 2`
///   returns the three departments with the highest `doc_count` from nested `terms`
///   (Engineering, Customer Service, Marketing) where the in-memory engine sorts the
///   flattened pairs by `doc_count` and reserves a department slot for whichever
///   department each surviving pair belongs to (Engineering, Customer Service,
///   Finance).
fn build_grouped_aggregations(
    aggregations: &[Aggregation],
    group_by: &[GroupBy],
    field_mappings: Option<&FieldMappings>,
) -> Result<JsonValue, String> {
    // Start with innermost aggregations (the metric aggs)
    let inner_aggs = build_simple_aggregations(aggregations)?;

    let modifier = resolve_aggregation_modifier(aggregations, group_by.len())?;

    // `top 0` on a group-by field is a no-op on a SINGLE level (Python guards
    // with `if bucket_size:`, falsy at zero, and the Rust evaluator reproduces
    // that) but EMPTIES the level on a multi-level grouping. OpenSearch has no
    // `size: 0`, so the emptying reading cannot be expressed.
    if group_by.len() > 1 {
        if let Some(gb) = group_by.iter().find(|gb| gb.bucket_size == Some(0)) {
            return Err(format!(
                "'top 0' on group-by field '{}' asks for zero buckets at that level, which \
                 OpenSearch cannot express: `terms` requires size > 0.",
                gb.field
            ));
        }
    }

    // Build nested terms aggregations, innermost first.
    let mut current_aggs = inner_aggs;
    let last_level = group_by.len() - 1;

    for (depth, gb) in group_by.iter().rev().enumerate() {
        let level = last_level - depth;
        let is_outermost = level == 0;
        let resolved_field = resolve_aggregation_field(&gb.field, field_mappings);

        let mut terms_body = serde_json::Map::new();
        terms_body.insert("field".to_string(), json!(resolved_field));

        match (is_outermost, modifier.as_ref()) {
            (true, Some((alias, direction, limit))) => {
                // The aggregation modifier owns the outermost level: rank by the
                // aggregate VALUE and cut at its own limit.
                terms_body.insert("size".to_string(), json!(limit));
                terms_body.insert("order".to_string(), json!({ alias.as_str(): direction }));
            }
            _ => match gb.bucket_size {
                Some(n) if n > 0 => {
                    // The group-by bucket limit ranks by doc_count. Stated
                    // explicitly even though it is the `terms` default, so the
                    // ranking key is readable in the emitted DSL rather than
                    // inherited.
                    terms_body.insert("size".to_string(), json!(n));
                    terms_body.insert("order".to_string(), json!({ "_count": "desc" }));
                }
                _ => {
                    terms_body.insert("size".to_string(), json!(UNLIMITED_BUCKET_SIZE));
                }
            },
        }

        let mut terms_agg = json!({ "terms": JsonValue::Object(terms_body) });

        let mut sub_aggs = current_aggs
            .as_object()
            .cloned()
            .unwrap_or_else(serde_json::Map::new);

        if is_outermost {
            if let (Some(_), Some(n)) = (modifier.as_ref(), gb.bucket_size) {
                if n > 0 {
                    // Both modifiers on one query: the bucket limit is the
                    // second pass over what the first one selected.
                    sub_aggs.insert(
                        BUCKET_LIMIT_AGG_NAME.to_string(),
                        json!({
                            "bucket_sort": {
                                "sort": [{ "_count": { "order": "desc" } }],
                                "size": n
                            }
                        }),
                    );
                }
            }
        }

        if !sub_aggs.is_empty() {
            terms_agg["aggs"] = JsonValue::Object(sub_aggs);
        }

        // The aggregation NAME uses the RAW field, not the `.keyword`-resolved
        // one, because that is the name every consumer of the response looks the
        // buckets up under.
        let key = format!("group_by_{}", gb.field);
        current_aggs = json!({ key: terms_agg });
    }

    Ok(current_aggs)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::parser::TqlParser;

    /// Every terms bucket this translator emits uses the ONE named size.
    ///
    /// `values`/`unique`/`distinct` spelled the number out, 145 lines from the
    /// constant that holds the same number for the same reason. A second silent
    /// definition of a limit is how the two drift apart.
    #[test]
    fn the_values_aggregation_uses_the_named_bucket_size() {
        for spelling in ["values", "unique", "distinct"] {
            let stats = parse_stats(&format!("| stats {spelling}(user.name)"));
            let dsl = translate_stats(&stats, None).expect("translation failed");
            let size = dsl
                .pointer("/aggs")
                .and_then(|aggs| aggs.as_object())
                .and_then(|aggs| aggs.values().next())
                .and_then(|agg| agg.pointer("/terms/size"))
                .unwrap_or_else(|| panic!("`{spelling}` emitted no terms bucket: {dsl}"));
            assert_eq!(
                *size,
                serde_json::json!(UNLIMITED_BUCKET_SIZE),
                "`{spelling}` disagrees with the named size"
            );
        }
    }

    fn parse_stats(query: &str) -> StatsNode {
        let parser = TqlParser::new();
        let ast = parser.parse(query).expect("parse failed");
        match ast {
            crate::parser::AstNode::StatsExpr(s) => s,
            crate::parser::AstNode::QueryWithStats(qws) => qws.stats,
            _ => panic!("expected stats AST, got {:?}", ast),
        }
    }

    #[test]
    fn test_simple_count_star() {
        let stats = parse_stats("| stats count(*)");
        let dsl = translate_stats(&stats, None).unwrap();
        let aggs = &dsl["aggs"];
        // Should have a value_count on _id
        assert!(aggs
            .as_object()
            .unwrap()
            .values()
            .any(|v| { v.get("value_count").is_some() }));
    }

    #[test]
    fn test_count_by_field() {
        let stats = parse_stats("| stats count(*) by event.code");
        let dsl = translate_stats(&stats, None).unwrap();
        let aggs = &dsl["aggs"];
        // Should have group_by_event.code with nested aggs
        let group = aggs.get("group_by_event.code").expect("missing group_by");
        assert!(group.get("terms").is_some());
        assert!(group.get("aggs").is_some());
    }

    #[test]
    fn test_multiple_group_by() {
        let stats = parse_stats("| stats count(*) by host.name, event.code");
        let dsl = translate_stats(&stats, None).unwrap();
        let aggs = &dsl["aggs"];
        // Outermost should be group_by_host.name
        let outer = aggs.get("group_by_host.name").expect("missing outer group");
        assert!(outer.get("terms").is_some());
        // Should have nested group_by_event.code
        let inner_aggs = outer.get("aggs").expect("missing inner aggs");
        assert!(inner_aggs.get("group_by_event.code").is_some());
    }

    // ---------------------------------------------------------------------
    // The two top-N modifiers rank by DIFFERENT keys. That is the whole
    // distinction, and it is the thing the emitted DSL used to lose: `limit`
    // was never read and `bucket_size` of `None` became a hard-coded 5.
    // ---------------------------------------------------------------------

    /// The outermost `terms` body for a grouped stats query.
    fn terms_body(query: &str) -> JsonValue {
        let stats = parse_stats(query);
        let dsl = translate_stats(&stats, None).unwrap_or_else(|e| panic!("{query:?}: {e}"));
        dsl["aggs"]["group_by_department"]["terms"].clone()
    }

    fn outer(query: &str) -> JsonValue {
        let stats = parse_stats(query);
        let dsl = translate_stats(&stats, None).unwrap_or_else(|e| panic!("{query:?}: {e}"));
        dsl["aggs"]["group_by_department"].clone()
    }

    #[test]
    fn aggregation_modifier_ranks_by_the_aggregate() {
        // Both spellings of the AGGREGATION-level modifier order by the metric.
        for query in [
            "| stats sum(salary) top 3 by department",
            "| stats sum(salary, top 3) by department",
        ] {
            assert_eq!(
                terms_body(query),
                json!({"field": "department", "size": 3, "order": {"sum_0": "desc"}}),
                "{query:?}"
            );
        }
    }

    #[test]
    fn bottom_reverses_the_direction() {
        // A pushdown that hard-coded `desc` would pass every `top` case above.
        for query in [
            "| stats sum(salary) bottom 3 by department",
            "| stats sum(salary, bottom 3) by department",
        ] {
            assert_eq!(
                terms_body(query),
                json!({"field": "department", "size": 3, "order": {"sum_0": "asc"}}),
                "{query:?}"
            );
        }
    }

    #[test]
    fn group_by_bucket_limit_ranks_by_doc_count() {
        // `by <field> top N` is a DIFFERENT modifier: it ranks by `doc_count`.
        // Stated explicitly in the DSL even though `_count` descending is the
        // `terms` default, so the ranking key is readable rather than inherited
        // -- the two modifiers are told apart by exactly this line.
        assert_eq!(
            terms_body("| stats count() by department top 3"),
            json!({"field": "department", "size": 3, "order": {"_count": "desc"}})
        );
    }

    #[test]
    fn unmodified_group_by_asks_for_every_bucket() {
        // `"size": 5` here is a silent answer to a question nobody asked, and is
        // indistinguishable from a dataset that really does have five
        // departments.
        let body = terms_body("| stats sum(salary) by department");
        assert_eq!(body["size"], UNLIMITED_BUCKET_SIZE);
        assert!(body.get("order").is_none(), "{body}");
    }

    #[test]
    fn both_modifiers_emit_two_passes_not_one() {
        // Swapping the passes yields a different SET, so one `terms` cannot
        // carry both. Collapsing them -- `size` from the bucket limit, `order`
        // from the aggregation modifier -- is what was emitted before, and on
        // this query it selects Engineering and Product where both engines
        // answer Engineering and Customer Service.
        let agg = outer("| stats sum(salary) top 4 by department top 2");
        assert_eq!(
            agg["terms"],
            json!({"field": "department", "size": 4, "order": {"sum_0": "desc"}})
        );
        assert_eq!(
            agg["aggs"][BUCKET_LIMIT_AGG_NAME],
            json!({"bucket_sort": {"sort": [{"_count": {"order": "desc"}}], "size": 2}})
        );
        // The metric survives alongside the pipeline aggregation; the order path
        // names it, so losing it would make the query a 500 rather than a wrong
        // number.
        assert_eq!(agg["aggs"]["sum_0"], json!({"sum": {"field": "salary"}}));
    }

    #[test]
    fn bucket_sort_is_absent_when_only_one_modifier_is_present() {
        // A second pass that always ran would re-order the single-modifier
        // answers.
        for query in [
            "| stats sum(salary) top 3 by department",
            "| stats count() by department top 3",
            "| stats sum(salary) by department",
        ] {
            assert!(
                outer(query)["aggs"].get(BUCKET_LIMIT_AGG_NAME).is_none(),
                "{query:?} emitted a second pass it does not need"
            );
        }
    }

    #[test]
    fn top_zero_on_a_group_by_field_is_a_no_op() {
        // Reproduced, not corrected: Python guards its single-level bucket pass
        // with `if bucket_size:`, falsy at zero, and the Rust evaluator
        // reproduces that. `top 0` on a single group-by field returns every
        // bucket -- not zero, and not even a re-order. Note the asymmetry with
        // the aggregation modifier, where `top 0` means zero buckets and is
        // refused.
        let stats = parse_stats("| stats count() by role top 0");
        let dsl = translate_stats(&stats, None).unwrap();
        let body = &dsl["aggs"]["group_by_role"]["terms"];
        assert_eq!(body["size"], UNLIMITED_BUCKET_SIZE);
        assert!(body.get("order").is_none(), "{body}");
    }

    #[test]
    fn multilevel_bucket_limits_nest_per_level() {
        let stats = parse_stats("| stats count() by department top 3, role top 2");
        let dsl = translate_stats(&stats, None).unwrap();
        let outer_agg = &dsl["aggs"]["group_by_department"];
        assert_eq!(
            outer_agg["terms"],
            json!({"field": "department", "size": 3, "order": {"_count": "desc"}})
        );
        assert_eq!(
            outer_agg["aggs"]["group_by_role"]["terms"],
            json!({"field": "role", "size": 2, "order": {"_count": "desc"}})
        );
    }

    #[test]
    fn multilevel_without_limits_asks_for_every_bucket_at_every_level() {
        let stats = parse_stats("| stats count() by department, role");
        let dsl = translate_stats(&stats, None).unwrap();
        let outer_agg = &dsl["aggs"]["group_by_department"];
        assert_eq!(outer_agg["terms"]["size"], UNLIMITED_BUCKET_SIZE);
        assert_eq!(
            outer_agg["aggs"]["group_by_role"]["terms"]["size"],
            UNLIMITED_BUCKET_SIZE
        );
    }

    #[test]
    fn the_order_path_follows_the_aggregation_position() {
        // A modifier on the SECOND aggregation must name the second
        // aggregation. This is the shape that made the Python translator emit
        // an order path naming an aggregation that does not exist.
        let agg = outer("| stats count(), sum(salary) top 3 by department");
        let order = agg["terms"]["order"].as_object().expect("an order");
        let (key, _) = order.iter().next().expect("one order key");
        assert_eq!(key, "sum_1");
        assert!(agg["aggs"].get(key).is_some(), "{}", agg["aggs"]);
    }

    #[test]
    fn an_uppercase_function_name_still_names_the_aggregation_it_emits() {
        // `agg_func_name` is a case-INSENSITIVE pest rule, so `SUM(x) TOP 3`
        // is well-formed. The metric key was built from the LOWERCASED function
        // and the order path from `agg.function` raw -- two spellings of one
        // name, harmless only because `parse_aggregation` lowercases as it
        // builds the AST. Both now come from `agg_alias`, and this pins that a
        // change to either side cannot reintroduce the mismatch: an order path
        // naming an aggregation that does not exist is a 500, not a wrong
        // number.
        let agg = outer("| stats SUM(salary) TOP 3 by department");
        assert_eq!(
            agg["terms"],
            json!({"field": "department", "size": 3, "order": {"sum_0": "desc"}})
        );
        assert!(agg["aggs"].get("sum_0").is_some(), "{}", agg["aggs"]);
    }

    #[test]
    fn test_avg_aggregation() {
        let stats = parse_stats("| stats avg(response_time)");
        let dsl = translate_stats(&stats, None).unwrap();
        let aggs = &dsl["aggs"];
        assert!(aggs
            .as_object()
            .unwrap()
            .values()
            .any(|v| { v.get("avg").is_some() }));
    }
}