hamelin_legacy 0.3.9

//! CTE-based implementation of MATCH pattern matching.
//!
//! This module provides an alternative to Trino's MATCH_RECOGNIZE using
//! standard SQL CTEs and window functions with regex.
//!
//! ## Strategy: Forward-Looking Window + REGEXP_LIKE
//!
//! The approach uses forward-looking windows to match MATCH_RECOGNIZE semantics:
//! 1. **Build state**: Use `ARRAY_AGG` with a forward time window (`CURRENT ROW AND <within> FOLLOWING`)
//! 2. **Filter on starting rows**: Only check rows where the pattern's first variable matches
//! 3. **Match check**: Use `REGEXP_LIKE(state, '^pattern')` to check if pattern matches
//!
//! ## Semantics: ONE ROW PER MATCH + AFTER MATCH SKIP TO NEXT ROW
//!
//! This implementation returns **ONE ROW PER MATCH** at the pattern start:
//! - Only rows where the pattern starts AND completes are returned
//! - Overlapping matches are allowed (SKIP TO NEXT ROW semantics)
//! - The `^` anchor ensures the pattern starts at the current row
//! - The returned timestamp is the start of the match

use std::rc::Rc;
use std::sync::Arc;

use crate::ast::assignment_clause::HamelinAssignmentClause;
use crate::ast::command::from;
use crate::ast::expression::HamelinExpression;
use crate::ast::from_clause::HamelinFromClause;
use crate::ast::pipeline::HamelinPipeline;
use crate::ast::sort_expression::HamelinSortExpression;
use crate::ast::QueryTranslationContext;
use crate::env::Environment;
use crate::translation::projection_builder::ProjectionBuilder;
use crate::translation::PendingQuery;

use antlr_rust::tree::ParseTree;
use hamelin_lib::antlr::hamelinparser::{
    ExactlyContextAttrs, FromClauseContextAll, GroupClauseContextAttrs, MatchCommandContext,
    MatchCommandContextAttrs, NestedContextAttrs, PatternContextAll, QuantifiedContextAttrs,
    QuantifierContextAll,
};
use hamelin_lib::err::{TranslationError, TranslationErrors};
use hamelin_lib::func::def::{FunctionTranslationContext, SpecialPosition};
use hamelin_lib::func::registry::FunctionRegistry;
use hamelin_lib::sql::expression::apply::{BinaryOperatorApply, FunctionCallApply, Lambda};
use hamelin_lib::sql::expression::identifier::{Identifier, SimpleIdentifier};
use hamelin_lib::sql::expression::literal::{
    ColumnReference, IntegerLiteral, NullLiteral, StringLiteral,
};
use hamelin_lib::sql::expression::operator::Operator;
use hamelin_lib::sql::expression::{Case, Leaf};
use hamelin_lib::sql::expression::{Direction, OrderByExpression, SQLExpression};
use hamelin_lib::sql::query::cte::CTE;
use hamelin_lib::sql::query::projection::{Binding, ColumnProjection, Projection};
use hamelin_lib::sql::query::window::{
    FrameBoundary, FrameType, WindowExpression, WindowFrame, WindowSpecification,
};
use hamelin_lib::sql::query::{BadPatternVariable, PatternVariable, SQLQuery, TableReference};
use hamelin_lib::sql::types::{SQLBaseType, SQLType};
use hamelin_lib::translation::ExpressionTranslation;
use hamelin_lib::types::{Type, CALENDAR_INTERVAL, INTERVAL, TIMESTAMP};
use ordermap::OrderMap;

use super::from::{FromAliasing, FromAnonymous};
use hamelin_lib::sql::expression::apply::UnaryOperatorApply;

/// Check if a SQL expression contains a negative interval (unary minus operator).
/// This is used to reject negative WITHIN intervals like `-5m`.
fn contains_negative_interval(sql: &SQLExpression) -> bool {
    match sql {
        SQLExpression::UnaryOperatorApply(UnaryOperatorApply { operator, .. }) => {
            *operator == Operator::Minus
        }
        _ => false,
    }
}

/// An AGG binding with its translated expression
#[derive(Debug, Clone)]
struct AggBinding {
    /// The output column name
    identifier: Identifier,
    /// The translated expression (contains aggregation function call)
    translation: ExpressionTranslation,
}

/// Represents a pattern element with its quantifier for regex generation
#[derive(Debug, Clone)]
pub struct PatternElement {
    /// The pattern variable name (uppercase, e.g., "A", "B")
    pub variable: String,
    /// The quantifier for this element
    pub quantifier: PatternQuantifier,
}

/// Quantifiers for pattern elements
#[derive(Debug, Clone, PartialEq)]
pub enum PatternQuantifier {
    /// Exactly one (default)
    One,
    /// One or more (+)
    OneOrMore,
    /// Zero or more (*)
    ZeroOrMore,
    /// Zero or one (?)
    ZeroOrOne,
    /// Exactly n times
    Exactly(usize),
}

impl PatternElement {
    pub fn new(variable: String, quantifier: PatternQuantifier) -> Self {
        Self {
            variable,
            quantifier,
        }
    }
}

/// Convert a pattern sequence to a regex string.
///
/// The regex matches a comma-separated string of pattern variable labels.
/// For example, pattern `A B+ A` with variables A="a" and B="b" becomes:
/// `^a,(b,)+a` (with ^ anchor to match at start for forward-looking window)
///
/// The forward-looking approach:
/// - We look ahead from the current row using ARRAY_AGG with FOLLOWING window
/// - The `^` anchor ensures the pattern starts at the current row
/// - Greedy quantifiers naturally find the longest match
///
/// Special handling for optional trailing elements:
/// - When a required element is followed by optional elements (ZeroOrMore, ZeroOrOne),
///   the comma separator must be included in the optional group
/// - For `A B?`: generates `^a(,b)?` not `^a,(b)?` so "a" matches when B is absent
/// - For `A B*`: generates `^a(,b(,b)*)?` not `^a,(b(,b)*)?`
pub fn pattern_to_regex(elements: &[PatternElement]) -> String {
    if elements.is_empty() {
        return String::new();
    }

    // Helper to check if an element can match zero times
    fn is_optional(q: &PatternQuantifier) -> bool {
        matches!(
            q,
            PatternQuantifier::ZeroOrMore
                | PatternQuantifier::ZeroOrOne
                | PatternQuantifier::Exactly(0)
        )
    }

    // Check if all remaining elements from index i are optional
    fn all_remaining_optional(elements: &[PatternElement], from_idx: usize) -> bool {
        elements[from_idx..]
            .iter()
            .all(|e| is_optional(&e.quantifier))
    }

    let mut parts = Vec::new();

    // Track if all preceding elements were optional (could match zero times)
    // This determines if we need a leading comma or not
    let mut all_preceding_optional = true;

    for (i, elem) in elements.iter().enumerate() {
        let is_first = i == 0;
        let is_last = i == elements.len() - 1;
        let var_lower = elem.variable.to_lowercase();

        // Check if all remaining elements after this one are optional
        // If so, we need to include the comma in the optional group
        let next_all_optional = !is_last && all_remaining_optional(elements, i + 1);

        // Determine if we need a leading comma for this element
        // We need a comma if there's a preceding non-optional element that matched
        let needs_leading_comma = !is_first && !all_preceding_optional;

        let part = match &elem.quantifier {
            PatternQuantifier::One => {
                all_preceding_optional = false;
                if is_last {
                    var_lower
                } else if next_all_optional {
                    // Don't add comma here - it will be part of the next optional group
                    var_lower
                } else {
                    format!("{},", var_lower)
                }
            }
            PatternQuantifier::OneOrMore => {
                all_preceding_optional = false;
                if is_last {
                    // B+ at end: match "b" or "b,b" or "b,b,b" = b(,b)*
                    format!("{}(,{})*", var_lower, var_lower)
                } else if next_all_optional {
                    // B+ followed by all-optional: don't add trailing comma
                    format!("{}(,{})*", var_lower, var_lower)
                } else {
                    // B+ not at end: (b,)+
                    format!("({},)+", var_lower)
                }
            }
            PatternQuantifier::ZeroOrMore => {
                // ZeroOrMore stays optional - don't change all_preceding_optional
                if is_last || next_all_optional {
                    // B* at end or followed by all-optional: include comma in optional group
                    if needs_leading_comma {
                        // Preceding required element - need leading comma
                        format!("(,{}(,{})*)?", var_lower, var_lower)
                    } else if is_first {
                        // First element - no comma needed
                        format!("({}(,{})*)?", var_lower, var_lower)
                    } else {
                        // Not first, but all preceding are optional - use optional comma
                        // This handles A? B* where state could be "b" or "a,b"
                        format!("(,?{}(,{})*)?", var_lower, var_lower)
                    }
                } else {
                    format!("({},)*", var_lower)
                }
            }
            PatternQuantifier::ZeroOrOne => {
                // ZeroOrOne stays optional - don't change all_preceding_optional
                if is_last || next_all_optional {
                    // B? at end or followed by all-optional: include comma in optional group
                    if needs_leading_comma {
                        // Preceding required element - need leading comma
                        format!("(,{})?", var_lower)
                    } else if is_first {
                        // First element - no comma needed
                        format!("({})?", var_lower)
                    } else {
                        // Not first, but all preceding are optional - use optional comma
                        // This handles A? B? where state could be "b" or "a,b"
                        format!("(,?{})?", var_lower)
                    }
                } else {
                    format!("({},)?", var_lower)
                }
            }
            PatternQuantifier::Exactly(n) => {
                if *n > 0 {
                    all_preceding_optional = false;
                }
                if *n == 0 {
                    String::new()
                } else if is_last {
                    let repeated: Vec<_> = (0..*n).map(|_| var_lower.clone()).collect();
                    repeated.join(",")
                } else if next_all_optional {
                    // Don't add trailing comma
                    let repeated: Vec<_> = (0..*n).map(|_| var_lower.clone()).collect();
                    repeated.join(",")
                } else {
                    let repeated: Vec<_> = (0..*n).map(|_| format!("{},", var_lower)).collect();
                    repeated.join("")
                }
            }
        };

        if !part.is_empty() {
            parts.push(part);
        }
    }

    // Add ^ anchor to match at start of state string (forward-looking)
    format!("^{}", parts.join(""))
}

/// Extract pattern elements from the AST.
///
/// Note: The same pattern variable CAN appear multiple times in a pattern sequence.
/// For example, `A B+ A` is valid and means "event from A, one or more from B, then another from A".
/// The FROM clause bindings are deduplicated separately in `extract_from_clauses_from_pattern`.
pub fn extract_pattern_elements(
    patterns: &[Rc<PatternContextAll<'static>>],
    cte_env: &Arc<Environment>,
    statement_context: &Rc<QueryTranslationContext>,
) -> Result<Vec<PatternElement>, TranslationErrors> {
    let mut result = Vec::new();

    for pctx in patterns {
        match pctx.as_ref() {
            PatternContextAll::QuantifiedContext(qctx) => {
                let fctx = TranslationErrors::expect(qctx, qctx.fromClause())?;
                let fc = HamelinFromClause::new(
                    fctx.clone(),
                    cte_env.clone(),
                    statement_context.clone(),
                );
                let maybe_si = fc.table_alias().map(|si| si.to_sql()).transpose()?;
                let si = match maybe_si {
                    Some(s) => s,
                    None => fc.table_identifier()?.last().clone(),
                };

                // Validate the pattern variable name is clean (no regex metacharacters)
                let pattern_var = PatternVariable::try_from(si.clone()).map_err(
                    |BadPatternVariable { name }| {
                        TranslationError::msg(
                            qctx,
                            &format!(
                                "Pattern variable '{}' contains special characters. \
                                 Pattern variables must be simple identifiers without backticks.",
                                name.name
                            ),
                        )
                        .single()
                    },
                )?;
                let var_name = pattern_var.name.to_uppercase();

                let quantifier = if let Some(q) = qctx.quantifier() {
                    match q.as_ref() {
                        QuantifierContextAll::AtLeastOneContext(_) => PatternQuantifier::OneOrMore,
                        QuantifierContextAll::AnyNumberContext(_) => PatternQuantifier::ZeroOrMore,
                        QuantifierContextAll::ZeroOrOneContext(_) => PatternQuantifier::ZeroOrOne,
                        QuantifierContextAll::ExactlyContext(ectx) => {
                            let text =
                                TranslationErrors::expect(ectx, ectx.INTEGER_VALUE())?.get_text();
                            let n: usize = text.parse().map_err(|_| {
                                TranslationError::msg(
                                    ectx,
                                    &format!("Invalid quantifier value '{}': expected a positive integer", text),
                                )
                                .single()
                            })?;
                            PatternQuantifier::Exactly(n)
                        }
                        QuantifierContextAll::Error(e) => {
                            return Err(TranslationError::msg(e, "parse error").single());
                        }
                    }
                } else {
                    PatternQuantifier::One
                };

                result.push(PatternElement::new(var_name, quantifier));
            }
            PatternContextAll::NestedContext(nctx) => {
                // Nested pattern groups like (a b)+ are not supported because our CTE-based
                // pattern matching uses simple regex on comma-separated state labels.
                // Supporting group quantifiers would require capturing groups in the regex
                // and correlating them back to row positions, which is complex to implement.
                if nctx.quantifier().is_some() {
                    return Err(TranslationError::msg(
                        nctx,
                        "Nested pattern groups with quantifiers (e.g., (a b)+) are not supported. \
                         The CTE-based pattern matching uses regex on state labels and does not \
                         support capturing groups. Use flat patterns instead (e.g., a b+ or a+ b+).",
                    )
                    .single());
                }
                // If no quantifier, just flatten the nested elements
                let nested =
                    extract_pattern_elements(&nctx.pattern_all(), cte_env, statement_context)?;
                result.extend(nested);
            }
            PatternContextAll::Error(e) => {
                return Err(TranslationError::msg(e, "parse error").single());
            }
        }
    }

    Ok(result)
}

/// Build an expression that returns the pattern label for the current row.
/// This is a nested IF that checks each pattern variable's define condition.
fn build_pattern_label_expr(
    from_clauses: &OrderMap<SimpleIdentifier, Rc<FromClauseContextAll<'static>>>,
) -> SQLExpression {
    let mut result: SQLExpression = NullLiteral::default().into();

    // Build nested IF: IF(a IS NOT NULL, 'a', IF(b IS NOT NULL, 'b', ...))
    for si in from_clauses.keys().rev() {
        let var_name = si.name.to_lowercase();
        let condition: SQLExpression = BinaryOperatorApply::new(
            Operator::IsNot,
            ColumnReference::new(si.clone().into()).into(),
            NullLiteral::default().into(),
        )
        .into();

        result = FunctionCallApply::with_three(
            "IF",
            condition,
            StringLiteral::new(&var_name).into(),
            result,
        )
        .into();
    }

    result
}

/// Main translation function using CTE-based approach.
///
/// Uses forward-looking windows to match MATCH_RECOGNIZE semantics.
///
/// Generates SQL of the form:
/// ```sql
/// WITH __ordered_events AS (
///     SELECT *, IF(a IS NOT NULL, 'a', IF(b IS NOT NULL, 'b', NULL)) AS __pattern_label
///     FROM <input>
/// ),
/// __with_state AS (
///     SELECT *,
///         ARRAY_JOIN(ARRAY_AGG(__pattern_label) OVER (
///             [PARTITION BY ...]
///             ORDER BY <sort_col>
///             RANGE BETWEEN CURRENT ROW AND <within> FOLLOWING
///         ), ',') AS __state
///     FROM __ordered_events
///     WHERE __pattern_label IS NOT NULL
/// )
/// SELECT <output_cols>
/// FROM __with_state
/// WHERE __pattern_label = '<first_pattern_var>'
///   AND REGEXP_LIKE(__state, '^<pattern>')
/// ORDER BY <sort_col>
/// ```
pub fn translate_with_cte(
    ctx: &MatchCommandContext<'static>,
    pipeline: &HamelinPipeline,
    previous: &PendingQuery,
) -> Result<PendingQuery, TranslationErrors> {
    let cte = Arc::new(pipeline.cte.clone());

    // Extract FROM clauses from pattern
    let from_clauses =
        extract_from_clauses_from_pattern(&ctx.pattern_all(), &cte, &pipeline.context)?;

    let from = from::uber_from(
        from_clauses.values().cloned().collect(),
        pipeline,
        previous,
        FromAliasing::AliasingAllowed,
        FromAnonymous::DefaultAlias,
    )?;
    let previous_env = Arc::new(from.env.clone());

    // Build defines environment
    let mut defines_env = previous_env.remove_substructure();
    for (si, fctx) in from_clauses.iter() {
        defines_env = defines_env.with_pattern_variable(
            si.clone().into(),
            previous_env
                .lookup(&si.clone().into())
                .map_err(|e| TranslationError::wrap_box(fctx.as_ref(), e.into()))?
                .try_unwrap_struct()
                .map_err(|e| TranslationError::wrap_box(fctx.as_ref(), e.into()))?
                .remove_substructure()
                .into(),
        );
    }

    // Extract pattern elements for regex generation
    let pattern_elements = extract_pattern_elements(&ctx.pattern_all(), &cte, &pipeline.context)?;

    // Generate the regex pattern (includes $ anchor)
    let regex_pattern = pattern_to_regex(&pattern_elements);

    // Extract WITHIN expression (required - defines the time window)
    let within_translation = ctx
        .within
        .as_ref()
        .map(|within_ctx| {
            HamelinExpression::new(
                within_ctx.clone(),
                pipeline.context.expression_translation_context(
                    &defines_env,
                    FunctionTranslationContext::default()
                        .with_special_allowed(SpecialPosition::Agg)
                        .with_special_allowed(SpecialPosition::Match),
                ),
            )
            .translate()
        })
        .transpose()?
        .ok_or_else(|| {
            TranslationError::msg(
                ctx,
                "CTE-based MATCH requires a WITHIN clause to define the time window",
            )
            .single()
        })?;
    let within_expression = within_translation.sql;
    let within_type = within_translation.typ;

    // Validate WITHIN expression - reject negative intervals
    if contains_negative_interval(&within_expression) {
        return Err(TranslationError::msg(ctx, "WITHIN interval cannot be negative").single());
    }

    // Handle SORT BY (or default to timestamp)
    // For complex sort expressions, we need to compute them in CTE 1 and reference by alias
    let (order_by, sort_col_name, sort_col_binding, sort_type) = if ctx
        .sortExpression_all()
        .is_empty()
    {
        let timestamp_ident: Identifier = SimpleIdentifier::new("timestamp").into();
        let ts_type = from.env.lookup(&timestamp_ident).ok().ok_or_else(|| {
            TranslationError::msg(
                ctx,
                "No 'timestamp' field in the environment. You need to specify SORT BY.",
            )
            .single()
        })?;
        if ts_type != TIMESTAMP {
            return Err(TranslationError::msg(
                ctx,
                &format!(
                    "Field 'timestamp' has type {:?}, expected TIMESTAMP. You need to specify SORT BY.",
                    ts_type
                ),
            )
            .single());
        }
        (
            vec![OrderByExpression::new(
                ColumnReference::new(timestamp_ident.clone()).into(),
                Direction::ASC,
            )],
            SimpleIdentifier::new("timestamp"),
            None, // No binding needed - timestamp is already a column
            TIMESTAMP,
        )
    } else {
        let translated = HamelinSortExpression::new(
            ctx.sortExpression_all(),
            pipeline
                .context
                .default_expression_translation_context(&previous_env),
        )
        .translate()?;

        // CTE strategy with WITHIN requires exactly one SORT expression (same as MATCH_RECOGNIZE)
        if translated.0.len() != 1 {
            return Err(TranslationError::msg(
                ctx,
                "MATCH with WITHIN requires exactly one SORT expression",
            )
            .single());
        }

        // CTE strategy uses forward-looking windows (CURRENT ROW AND <within> FOLLOWING)
        // which requires ASC sort order. DESC would make FOLLOWING look backwards.
        if translated.0[0].direction == Direction::DESC {
            return Err(TranslationError::msg(
                    ctx,
                    "MATCH with WITHIN requires ascending sort order (ASC). \
                     DESC is not supported because the forward-looking window would look backwards.",
                )
                .single());
        }

        let sort_type = translated.1[0].clone();

        // Check if it's a simple column reference
        let is_simple_col_ref = matches!(
            translated.0[0].expression.as_ref(),
            SQLExpression::Leaf(Leaf::ColumnReference(_))
        );

        if is_simple_col_ref {
            // Simple column reference - no binding needed
            let col_name = if let SQLExpression::Leaf(Leaf::ColumnReference(cr)) =
                translated.0[0].expression.as_ref()
            {
                cr.identifier.last().clone()
            } else {
                unreachable!()
            };
            (translated.0, col_name, None, sort_type)
        } else {
            // Complex expression - need to compute it in CTE and reference by alias
            let sort_alias = SimpleIdentifier::new("__sort_col");
            let sort_expr = (*translated.0[0].expression).clone();
            let direction = translated.0[0].direction.clone();

            // Create new order_by that references the alias instead of the complex expression
            let new_order_by = vec![OrderByExpression::new(
                ColumnReference::new(sort_alias.clone().into()).into(),
                direction,
            )];

            (new_order_by, sort_alias, Some(sort_expr), sort_type)
        }
    };

    // Validate type compatibility between SORT column and WITHIN expression
    // For TIMESTAMP: TIMESTAMP - TIMESTAMP = INTERVAL (should match WITHIN interval type)
    // For numeric types: INT - INT = INT (should match WITHIN numeric type)
    let types_compatible = if sort_type == TIMESTAMP {
        within_type == INTERVAL || within_type == CALENDAR_INTERVAL
    } else {
        sort_type == within_type
    };

    if !types_compatible {
        return Err(TranslationError::msg(
            ctx,
            &format!(
                "MATCH with WITHIN: SORT column type {:?} is incompatible with WITHIN type {:?}",
                sort_type, within_type
            ),
        )
        .single());
    }

    // Handle BY (partition) clauses
    // We need to store both the expression (for CTE window partitioning) and binding info (for output)
    let mut partition_exprs = Vec::new();
    let mut by_clause_bindings = Vec::new();
    for gc in ctx.groupClause_all() {
        let clause = TranslationErrors::expect(gc.as_ref(), gc.assignmentClause())?;
        let (identifier, translation) = HamelinAssignmentClause::new(
            clause.clone(),
            pipeline
                .context
                .default_expression_translation_context(&previous_env),
        )
        .to_sql()?;

        // Use a prefixed name in the CTE to avoid collisions
        // Use double underscore as separator to avoid collisions between
        // qualified names like `user.name` (__by_user__name) and simple names
        // like `user_name` (__by_user_name)
        let full_name = identifier
            .simples()
            .iter()
            .map(|si| si.name.as_str())
            .collect::<Vec<_>>()
            .join("__");
        let prefixed_name = SimpleIdentifier::new(&format!("__by_{}", full_name));

        partition_exprs.push((prefixed_name.clone(), translation.sql.clone()));
        by_clause_bindings.push(ByClauseBinding {
            identifier,
            prefixed_name,
            typ: translation.typ,
        });
    }

    // Get base columns from environment
    let base_columns: Vec<SimpleIdentifier> = from.env.flatten().fields.keys().cloned().collect();

    // Handle AGG clauses - these are aggregations over matched rows
    let agg_bindings: Vec<AggBinding> = ctx
        .assignmentClause_all()
        .into_iter()
        .map(|clause| {
            HamelinAssignmentClause::new(
                clause.clone(),
                pipeline.context.expression_translation_context(
                    &defines_env,
                    FunctionTranslationContext::default()
                        .with_special_allowed(SpecialPosition::Agg)
                        .with_special_allowed(SpecialPosition::Match),
                ),
            )
            .to_sql()
            .map(|(identifier, translation)| AggBinding {
                identifier,
                translation,
            })
        })
        .collect::<Result<Vec<_>, _>>()?;

    // Validate that all aggregates are supported by CTE strategy
    validate_supported_aggregates(&agg_bindings, ctx, &pipeline.context.registry)?;

    // Extract first required pattern variable for filtering (forward-looking approach)
    // If the first element is optional (?, *, or {0}), we need to include the next element too
    let first_pattern_vars: Vec<String> = {
        let mut vars = Vec::new();
        for elem in &pattern_elements {
            // Skip elements that match zero times (they don't contribute to the pattern start)
            let is_zero = matches!(elem.quantifier, PatternQuantifier::Exactly(0));
            if !is_zero {
                vars.push(elem.variable.to_lowercase());
            }
            // Stop if this element is required (not optional)
            if !matches!(
                elem.quantifier,
                PatternQuantifier::ZeroOrOne
                    | PatternQuantifier::ZeroOrMore
                    | PatternQuantifier::Exactly(0)
            ) {
                break;
            }
        }
        vars
    };
    if first_pattern_vars.is_empty() {
        return Err(TranslationError::msg(ctx, "MATCH pattern cannot be empty").single());
    }

    // CTE names
    let ordered_cte_name = SimpleIdentifier::new("__ordered_events");
    let state_cte_name = SimpleIdentifier::new("__with_state");

    // CTE 1: __ordered_events - base query with pattern labels, BY clause columns, and sort column
    let pattern_label_expr = build_pattern_label_expr(&from_clauses);
    let ordered_query = build_ordered_events_cte(
        &from.query,
        &base_columns,
        pattern_label_expr,
        &partition_exprs,
        sort_col_binding.as_ref().map(|expr| (&sort_col_name, expr)),
    )?;

    // CTE 2: __with_state - accumulate event labels into state string (forward-looking)
    let has_agg = !agg_bindings.is_empty();
    let agg_value_columns = extract_agg_value_columns(&agg_bindings);
    let state_query = build_state_cte(
        &ordered_cte_name,
        &base_columns,
        &partition_exprs,
        &order_by,
        &within_expression,
        sort_col_binding.as_ref().map(|_| &sort_col_name),
        has_agg,
        &agg_value_columns,
    )?;

    // Build final query - filter on starting variable(s) and regex match
    // AGG values are computed here using match length from regex
    let final_query = build_final_query(
        &state_cte_name,
        &regex_pattern,
        &first_pattern_vars,
        &base_columns,
        &by_clause_bindings,
        &agg_bindings,
        &order_by,
        &from.env,
    )?;

    // Assemble CTEs
    let cte_chain = CTE::default()
        .with(ordered_cte_name, ordered_query)
        .with(state_cte_name, state_query);

    let statement = final_query.cte(cte_chain);

    // Build output environment
    // IMPORTANT: Column order must match the SQL SELECT in build_final_query
    let mut output_builder = ProjectionBuilder::default();

    // Collect the first component of BY and AGG identifiers for deduplication.
    // We need to exclude base columns that would conflict with BY/AGG output columns.
    // For nested paths like `user.name`, we exclude the top-level `user` since
    // ProjectionBuilder will create a nested struct for it.
    let by_first_names: Vec<&SimpleIdentifier> = by_clause_bindings
        .iter()
        .map(|b| b.identifier.first())
        .collect();
    let agg_first_names: Vec<&SimpleIdentifier> =
        agg_bindings.iter().map(|b| b.identifier.first()).collect();

    // 1. Base columns first (excluding BY/AGG columns)
    for col in &base_columns {
        if by_first_names.contains(&col) || agg_first_names.contains(&col) {
            continue;
        }
        let col_type = from
            .env
            .lookup(&col.clone().into())
            .map_err(|e| TranslationError::wrap(ctx, e).single())?;
        output_builder.bind(
            col.clone().into(),
            ColumnReference::new(col.clone().into()).into(),
            col_type,
        );
    }

    // 2. BY (partition) columns (using full identifier path)
    for binding in &by_clause_bindings {
        output_builder.bind(
            binding.identifier.clone(),
            ColumnReference::new(binding.identifier.clone()).into(),
            binding.typ.clone(),
        );
    }

    // 3. AGG columns (using full identifier path)
    for binding in &agg_bindings {
        let typ = get_agg_output_type(binding);
        output_builder.bind(
            binding.identifier.clone(),
            ColumnReference::new(binding.identifier.clone()).into(),
            typ,
        );
    }

    Ok(PendingQuery::new(
        statement.into(),
        Environment::new(output_builder.build_hamelin_type()),
    ))
}

/// CTE 1: Add pattern labels, BY clause columns, and complex sort expression to each row
fn build_ordered_events_cte(
    from_query: &SQLQuery,
    base_columns: &[SimpleIdentifier],
    pattern_label_expr: SQLExpression,
    by_clause_exprs: &[(SimpleIdentifier, SQLExpression)],
    sort_col_binding: Option<(&SimpleIdentifier, &SQLExpression)>,
) -> Result<SQLQuery, TranslationErrors> {
    let mut projections: Vec<Projection> = base_columns
        .iter()
        .map(|col| {
            ColumnProjection {
                identifier: col.clone().into(),
            }
            .into()
        })
        .collect();

    projections
        .push(Binding::new(SimpleIdentifier::new("__pattern_label"), pattern_label_expr).into());

    // Add BY clause columns with prefixed names
    for (prefixed_name, expr) in by_clause_exprs {
        projections.push(Binding::new(prefixed_name.clone(), expr.clone()).into());
    }

    // Add complex sort expression if present (for non-simple column references)
    if let Some((sort_name, sort_expr)) = sort_col_binding {
        projections.push(Binding::new(sort_name.clone(), sort_expr.clone()).into());
    }

    // Strip the CTE from from_query - CTEs will be added at the top level
    let mut base_query = from_query.clone();
    base_query.cte = CTE::default();
    Ok(base_query.select(projections))
}

/// CTE 2: Build state string using ARRAY_AGG with forward-looking time window
///
/// Uses `RANGE BETWEEN CURRENT ROW AND <within> FOLLOWING` to look ahead from each row.
/// This matches MATCH_RECOGNIZE forward-looking semantics.
///
/// For AGG support, we also collect __labels_array which contains all pattern labels
/// in the window. This allows us to compute match length in the final query and aggregate
/// only over the matched rows.
fn build_state_cte(
    ordered_cte_name: &SimpleIdentifier,
    base_columns: &[SimpleIdentifier],
    partition_by: &[(SimpleIdentifier, SQLExpression)],
    order_by: &[OrderByExpression],
    within_expr: &SQLExpression,
    sort_col_name: Option<&SimpleIdentifier>,
    has_agg: bool,
    agg_value_columns: &[SimpleIdentifier],
) -> Result<SQLQuery, TranslationErrors> {
    // Window: RANGE BETWEEN CURRENT ROW AND <within> FOLLOWING (forward-looking)
    // Use the prefixed column names for partitioning (they're already computed in CTE 1)
    let partition_exprs: Vec<SQLExpression> = partition_by
        .iter()
        .map(|(prefixed_name, _)| ColumnReference::new(prefixed_name.clone().into()).into())
        .collect();

    let window_spec = WindowSpecification {
        partition_by: partition_exprs,
        order_by: order_by.to_vec(),
        frame: Some(WindowFrame {
            frame_type: FrameType::RANGE,
            preceding: FrameBoundary::CurrentRowBoundary,
            following: FrameBoundary::BoundaryExpression(Box::new(within_expr.clone())),
        }),
    };

    // ARRAY_AGG(__pattern_label) OVER (...) - keep as array for match length computation
    let labels_array: SQLExpression = WindowExpression::new(
        FunctionCallApply::with_one(
            "ARRAY_AGG",
            ColumnReference::new(SimpleIdentifier::new("__pattern_label").into()).into(),
        )
        .into(),
        window_spec.clone().into(),
    )
    .into();

    // ARRAY_JOIN for state string (used in regex matching)
    let state_expr: SQLExpression = FunctionCallApply::with_two(
        "ARRAY_JOIN",
        labels_array.clone(),
        StringLiteral::new(",").into(),
    )
    .into();

    let mut projections: Vec<Projection> = base_columns
        .iter()
        .map(|col| {
            ColumnProjection {
                identifier: col.clone().into(),
            }
            .into()
        })
        .collect();

    projections.push(
        ColumnProjection {
            identifier: SimpleIdentifier::new("__pattern_label").into(),
        }
        .into(),
    );

    // Include BY clause columns (prefixed names)
    for (prefixed_name, _) in partition_by {
        projections.push(
            ColumnProjection {
                identifier: prefixed_name.clone().into(),
            }
            .into(),
        );
    }

    // Include complex sort column if present
    if let Some(sort_name) = sort_col_name {
        projections.push(
            ColumnProjection {
                identifier: sort_name.clone().into(),
            }
            .into(),
        );
    }

    projections.push(Binding::new(SimpleIdentifier::new("__state"), state_expr).into());

    // If we have AGG, also include __labels_array for match length computation
    if has_agg {
        projections
            .push(Binding::new(SimpleIdentifier::new("__labels_array"), labels_array).into());

        // Create value arrays for columns used in aggregates (max, min, first, last, etc.)
        for col in agg_value_columns {
            let value_array: SQLExpression = WindowExpression::new(
                FunctionCallApply::with_one(
                    "ARRAY_AGG",
                    ColumnReference::new(col.clone().into()).into(),
                )
                .into(),
                window_spec.clone().into(),
            )
            .into();
            let array_col_name = agg_array_column_name(col);
            projections.push(Binding::new(array_col_name, value_array).into());
        }
    }

    // Filter: only keep rows where __pattern_label IS NOT NULL
    let filter: SQLExpression = BinaryOperatorApply::new(
        Operator::IsNot,
        ColumnReference::new(SimpleIdentifier::new("__pattern_label").into()).into(),
        NullLiteral::default().into(),
    )
    .into();

    Ok(SQLQuery::default()
        .from(TableReference::new(ordered_cte_name.clone().into()).into())
        .select(projections)
        .where_(filter))
}

/// Binding info for BY clause columns that need aliased projection
struct ByClauseBinding {
    /// The alias/identifier for output
    identifier: Identifier,
    /// The prefixed column name stored in the CTE
    prefixed_name: SimpleIdentifier,
    /// The type of the expression
    typ: Type,
}

/// Check if a function is a valid aggregate for CTE-based MATCH.
///
/// Valid aggregates are functions registered in the function registry with
/// `SpecialPosition::Match`.
fn is_valid_cte_aggregate(func_name: &str, registry: &FunctionRegistry) -> bool {
    let name_lower = func_name.to_lowercase();
    if let Some(defs) = registry.function_defs.get(&name_lower) {
        defs.iter()
            .any(|def| matches!(def.special_position(), Some(SpecialPosition::Match)))
    } else {
        false
    }
}

/// Validate that all aggregate functions in AGG bindings are supported by CTE strategy.
///
/// The CTE strategy collects column values into arrays via ARRAY_AGG, then applies
/// array-based aggregation functions. In principle, any function that has an array
/// equivalent could be supported here.
///
/// Returns an error if:
/// - The function is not a valid match aggregate (SpecialPosition::Match)
/// - The aggregate argument is not a simple column reference (for aggregates that take arguments)
fn validate_supported_aggregates(
    agg_bindings: &[AggBinding],
    ctx: &MatchCommandContext<'static>,
    registry: &FunctionRegistry,
) -> Result<(), TranslationErrors> {
    for binding in agg_bindings {
        if let SQLExpression::FunctionCallApply(fc) = &binding.translation.sql {
            if !is_valid_cte_aggregate(&fc.function_name, registry) {
                return Err(TranslationError::msg(
                    ctx,
                    &format!(
                        "CTE-based MATCH does not support {}() in AGG. \
                         Supported aggregates are those with array function equivalents \
                         (count, sum, avg, min, max, first, last).",
                        fc.function_name
                    ),
                )
                .single());
            }

            // For aggregates with arguments (not count()), verify the argument is a simple column reference
            // Complex expressions like sum(value + 1) or avg(a.b) are not supported
            if !fc.arguments.is_empty() {
                if let Some(arg) = fc.arguments.first() {
                    if extract_simple_column_name(arg).is_none() {
                        return Err(TranslationError::msg(
                            ctx,
                            &format!(
                                "CTE-based MATCH AGG only supports simple column references as arguments. \
                                 Expression '{}' in {}() is not a simple column.",
                                arg, fc.function_name
                            ),
                        )
                        .single());
                    }
                }
            }
        } else {
            // Non-function expressions (column references, literals, arithmetic) are not valid in AGG
            return Err(TranslationError::msg(
                ctx,
                &format!(
                    "MATCH AGG requires aggregate function calls. '{}' is not an aggregate function.",
                    binding.translation.sql
                ),
            )
            .single());
        }
    }
    Ok(())
}

/// Extract unique column names from aggregate expressions that need value arrays.
///
/// Returns column names used inside aggregate function arguments. These columns
/// need ARRAY_AGG to collect values for the match window, which are then sliced
/// to the match length for aggregation.
fn extract_agg_value_columns(agg_bindings: &[AggBinding]) -> Vec<SimpleIdentifier> {
    let mut columns = Vec::new();
    let mut seen = std::collections::HashSet::new();

    for binding in agg_bindings {
        if let SQLExpression::FunctionCallApply(fc) = &binding.translation.sql {
            // Extract columns from function arguments (e.g., max(value) -> value)
            // Skip functions with no arguments like count()
            if !fc.arguments.is_empty() {
                for col_ref in binding.translation.sql.get_column_references() {
                    let col_name = col_ref.identifier.last().clone();
                    if seen.insert(col_name.name.clone()) {
                        columns.push(col_name);
                    }
                }
            }
        }
    }

    columns
}

/// Get the array column name for a given base column.
fn agg_array_column_name(col: &SimpleIdentifier) -> SimpleIdentifier {
    SimpleIdentifier::new(&format!("__agg_values_{}", col.name))
}

/// Transform an aggregation expression to use match length instead of full window.
///
/// For count(): CARDINALITY(SLICE(__labels_array, 1, match_length))
/// For max(col): reduce(SLICE(__agg_values_col, 1, match_length), NULL, (s,n) -> IF(s IS NULL, n, IF(n > s, n, s)), s -> s)
/// For min(col): reduce(SLICE(__agg_values_col, 1, match_length), NULL, (s,n) -> IF(s IS NULL, n, IF(n < s, n, s)), s -> s)
/// For first(col): element_at(__agg_values_col, 1)
/// For last(col): element_at(__agg_values_col, match_length)
///
/// The __labels_array contains all pattern labels in the window. By slicing it
/// to match_length, we get only the labels that participate in the actual match.
/// Result of transforming an aggregate expression for match length.
/// Contains the transformed SQL expression.
struct TransformedAgg {
    sql: SQLExpression,
}

/// Determine the output type for an AGG binding.
/// SUM and AVG return DOUBLE in the CTE strategy (for type-safe aggregation),
/// while other aggregates preserve their original type.
fn get_agg_output_type(binding: &AggBinding) -> Type {
    if let SQLExpression::FunctionCallApply(fc) = &binding.translation.sql {
        let func_name = fc.function_name.to_uppercase();
        if func_name == "SUM" || func_name == "AVG" {
            return Type::Double;
        }
    }
    binding.translation.typ.clone()
}

fn transform_agg_for_match_length(
    agg_expr: &SQLExpression,
    match_length_expr: &SQLExpression,
) -> TransformedAgg {
    if let SQLExpression::FunctionCallApply(fc) = agg_expr {
        let func_name = fc.function_name.to_uppercase();

        match func_name.as_str() {
            "COUNT" => {
                if fc.arguments.is_empty() {
                    // count() - count all rows in the match
                    // Transform to CARDINALITY(SLICE(__labels_array, 1, match_length))
                    return TransformedAgg {
                        sql: FunctionCallApply::with_one(
                            "CARDINALITY",
                            FunctionCallApply::with_three(
                                "SLICE",
                                ColumnReference::new(
                                    SimpleIdentifier::new("__labels_array").into(),
                                )
                                .into(),
                                IntegerLiteral::from_int(1).into(),
                                match_length_expr.clone(),
                            )
                            .into(),
                        )
                        .into(),
                    };
                } else if let Some(arg) = fc.arguments.first() {
                    // count(x) - count non-null values of x in the match
                    // Transform to CARDINALITY(FILTER(SLICE(__agg_values_x, 1, match_length), e -> e IS NOT NULL))
                    if let Some(col_name) = extract_simple_column_name(arg) {
                        let array_col = agg_array_column_name(&col_name);
                        let sliced_array: SQLExpression = FunctionCallApply::with_three(
                            "SLICE",
                            ColumnReference::new(array_col.into()).into(),
                            IntegerLiteral::from_int(1).into(),
                            match_length_expr.clone(),
                        )
                        .into();

                        // filter(array, e -> e IS NOT NULL)
                        let e = SimpleIdentifier::new("e");
                        let filtered_array: SQLExpression = FunctionCallApply::with_two(
                            "filter",
                            sliced_array,
                            Lambda::new(
                                vec![e.clone()],
                                BinaryOperatorApply::new(
                                    Operator::IsNot,
                                    ColumnReference::new(e.into()).into(),
                                    NullLiteral::default().into(),
                                )
                                .into(),
                            )
                            .into(),
                        )
                        .into();

                        return TransformedAgg {
                            sql: FunctionCallApply::with_one("CARDINALITY", filtered_array).into(),
                        };
                    }
                }
            }

            "MAX" | "MIN" => {
                // Get the column being aggregated
                if let Some(arg) = fc.arguments.first() {
                    if let Some(col_name) = extract_simple_column_name(arg) {
                        let array_col = agg_array_column_name(&col_name);
                        let sliced_array = FunctionCallApply::with_three(
                            "SLICE",
                            ColumnReference::new(array_col.into()).into(),
                            IntegerLiteral::from_int(1).into(),
                            match_length_expr.clone(),
                        );

                        // Build reduce expression for max/min
                        // reduce(array, NULL, (s,n) -> CASE WHEN s IS NULL THEN n WHEN n >/< s THEN n ELSE s END, s -> s)
                        let s = SimpleIdentifier::new("s");
                        let n = SimpleIdentifier::new("n");
                        let cmp_op = if func_name == "MAX" {
                            Operator::Gt
                        } else {
                            Operator::Lt
                        };

                        return TransformedAgg {
                            sql: FunctionCallApply::with_four(
                                "reduce",
                                sliced_array.into(),
                                NullLiteral::default().into(),
                                Lambda::new(
                                    vec![s.clone(), n.clone()],
                                    Case::new(
                                        &[(
                                            BinaryOperatorApply::new(
                                                Operator::Is,
                                                ColumnReference::new(s.clone().into()).into(),
                                                NullLiteral::default().into(),
                                            )
                                            .into(),
                                            ColumnReference::new(n.clone().into()).into(),
                                        )],
                                        Some(
                                            Case::new(
                                                &[(
                                                    BinaryOperatorApply::new(
                                                        cmp_op,
                                                        ColumnReference::new(n.clone().into())
                                                            .into(),
                                                        ColumnReference::new(s.clone().into())
                                                            .into(),
                                                    )
                                                    .into(),
                                                    ColumnReference::new(n.clone().into()).into(),
                                                )],
                                                Some(ColumnReference::new(s.clone().into()).into()),
                                            )
                                            .into(),
                                        ),
                                    )
                                    .into(),
                                )
                                .into(),
                                Lambda::new(
                                    vec![s.clone()],
                                    ColumnReference::new(s.clone().into()).into(),
                                )
                                .into(),
                            )
                            .into(),
                        };
                    }
                }
            }

            "FIRST" => {
                // Get the column being aggregated
                if let Some(arg) = fc.arguments.first() {
                    if let Some(col_name) = extract_simple_column_name(arg) {
                        let array_col = agg_array_column_name(&col_name);
                        // element_at(array, 1) - first element
                        return TransformedAgg {
                            sql: FunctionCallApply::with_two(
                                "element_at",
                                ColumnReference::new(array_col.into()).into(),
                                IntegerLiteral::from_int(1).into(),
                            )
                            .into(),
                        };
                    }
                }
            }

            "LAST" => {
                // Get the column being aggregated
                if let Some(arg) = fc.arguments.first() {
                    if let Some(col_name) = extract_simple_column_name(arg) {
                        let array_col = agg_array_column_name(&col_name);
                        // element_at(array, match_length) - last element of matched portion
                        return TransformedAgg {
                            sql: FunctionCallApply::with_two(
                                "element_at",
                                ColumnReference::new(array_col.into()).into(),
                                match_length_expr.clone(),
                            )
                            .into(),
                        };
                    }
                }
            }

            "SUM" => {
                // Get the column being aggregated
                if let Some(arg) = fc.arguments.first() {
                    if let Some(col_name) = extract_simple_column_name(arg) {
                        let array_col = agg_array_column_name(&col_name);
                        let sliced_array = FunctionCallApply::with_three(
                            "SLICE",
                            ColumnReference::new(array_col.into()).into(),
                            IntegerLiteral::from_int(1).into(),
                            match_length_expr.clone(),
                        );

                        // Build reduce expression for sum using DOUBLE for type safety
                        // Standard SQL SUM ignores NULL values and returns NULL if all values are NULL
                        // 1. Filter out NULLs: filter(sliced_array, e -> e IS NOT NULL)
                        // 2. Sum non-null values: reduce(filtered, 0.0, (s,n) -> s + n, s -> s)
                        // 3. Return NULL if empty: NULLIF(CARDINALITY(filtered), 0) to check

                        let e = SimpleIdentifier::new("e");
                        let filtered_array: SQLExpression = FunctionCallApply::with_two(
                            "filter",
                            sliced_array.into(),
                            Lambda::new(
                                vec![e.clone()],
                                BinaryOperatorApply::new(
                                    Operator::IsNot,
                                    ColumnReference::new(e.into()).into(),
                                    NullLiteral::default().into(),
                                )
                                .into(),
                            )
                            .into(),
                        )
                        .into();

                        let s = SimpleIdentifier::new("s");
                        let n = SimpleIdentifier::new("n");
                        let zero_double: SQLExpression = IntegerLiteral::from_int(0)
                            .to_sql_expression()
                            .cast(SQLType::SQLBaseType(SQLBaseType::Double));

                        // Sum the non-null values
                        let sum_expr: SQLExpression = FunctionCallApply::with_four(
                            "reduce",
                            filtered_array.clone(),
                            zero_double,
                            Lambda::new(
                                vec![s.clone(), n.clone()],
                                BinaryOperatorApply::new(
                                    Operator::Plus,
                                    ColumnReference::new(s.clone().into()).into(),
                                    ColumnReference::new(n.clone().into()).into(),
                                )
                                .into(),
                            )
                            .into(),
                            Lambda::new(
                                vec![s.clone()],
                                ColumnReference::new(s.clone().into()).into(),
                            )
                            .into(),
                        )
                        .into();

                        // Return NULL if all values were NULL (filtered array is empty)
                        // IF(CARDINALITY(filtered) = 0, NULL, sum)
                        let count_expr: SQLExpression =
                            FunctionCallApply::with_one("CARDINALITY", filtered_array).into();

                        return TransformedAgg {
                            sql: FunctionCallApply::with_three(
                                "IF",
                                BinaryOperatorApply::new(
                                    Operator::Eq,
                                    count_expr,
                                    IntegerLiteral::from_int(0).into(),
                                )
                                .into(),
                                NullLiteral::default().into(),
                                sum_expr,
                            )
                            .into(),
                        };
                    }
                }
            }

            "AVG" => {
                // Get the column being aggregated
                if let Some(arg) = fc.arguments.first() {
                    if let Some(col_name) = extract_simple_column_name(arg) {
                        let array_col = agg_array_column_name(&col_name);
                        let sliced_array: SQLExpression = FunctionCallApply::with_three(
                            "SLICE",
                            ColumnReference::new(array_col.into()).into(),
                            IntegerLiteral::from_int(1).into(),
                            match_length_expr.clone(),
                        )
                        .into();

                        // Filter out NULLs first - standard SQL AVG ignores NULL values
                        // filter(sliced_array, e -> e IS NOT NULL)
                        let e = SimpleIdentifier::new("e");
                        let filtered_array: SQLExpression = FunctionCallApply::with_two(
                            "filter",
                            sliced_array,
                            Lambda::new(
                                vec![e.clone()],
                                BinaryOperatorApply::new(
                                    Operator::IsNot,
                                    ColumnReference::new(e.into()).into(),
                                    NullLiteral::default().into(),
                                )
                                .into(),
                            )
                            .into(),
                        )
                        .into();

                        // Count non-null values for division
                        let non_null_count: SQLExpression =
                            FunctionCallApply::with_one("CARDINALITY", filtered_array.clone())
                                .into();

                        // Build reduce expression for sum over non-null values
                        // Since we filtered NULLs, no need for COALESCE
                        // reduce(filtered_array, CAST(0 AS DOUBLE), (s,n) -> s + n, s -> s)
                        let s = SimpleIdentifier::new("s");
                        let n = SimpleIdentifier::new("n");
                        let zero_double: SQLExpression = IntegerLiteral::from_int(0)
                            .to_sql_expression()
                            .cast(SQLType::SQLBaseType(SQLBaseType::Double));

                        let sum_expr: SQLExpression = FunctionCallApply::with_four(
                            "reduce",
                            filtered_array,
                            zero_double,
                            Lambda::new(
                                vec![s.clone(), n.clone()],
                                BinaryOperatorApply::new(
                                    Operator::Plus,
                                    ColumnReference::new(s.clone().into()).into(),
                                    ColumnReference::new(n.clone().into()).into(),
                                )
                                .into(),
                            )
                            .into(),
                            Lambda::new(
                                vec![s.clone()],
                                ColumnReference::new(s.clone().into()).into(),
                            )
                            .into(),
                        )
                        .into();

                        // Divide sum by count of non-null values (standard AVG semantics)
                        // If all values are NULL, this returns NULL (0/0 or sum/0)
                        // Use NULLIF to return NULL when count is 0
                        return TransformedAgg {
                            sql: BinaryOperatorApply::new(
                                Operator::Slash,
                                sum_expr,
                                FunctionCallApply::with_two(
                                    "NULLIF",
                                    non_null_count,
                                    IntegerLiteral::from_int(0).into(),
                                )
                                .into(),
                            )
                            .into(),
                        };
                    }
                }
            }

            _ => {}
        }
    }

    // For unsupported aggregates, return the original expression
    TransformedAgg {
        sql: agg_expr.clone(),
    }
}

/// Extract a simple column name from an expression if it's a direct column reference.
fn extract_simple_column_name(expr: &SQLExpression) -> Option<SimpleIdentifier> {
    if let SQLExpression::Leaf(Leaf::ColumnReference(cr)) = expr {
        // Only return Some for single-part identifiers (e.g., "value")
        // Return None for nested identifiers (e.g., "user.name") since we can't
        // reliably extract a column name for ARRAY_AGG collection
        match &cr.identifier {
            Identifier::Simple(simple) => Some(simple.clone()),
            Identifier::Compound(_) => None,
        }
    } else {
        None
    }
}

/// Build the final query that filters on starting pattern variable(s) and regex match.
///
/// AGG values are computed here using the match length extracted from the regex.
/// The match length = CARDINALITY(SLICE(__labels_array, 1, length_of_matched_portion))
/// where length_of_matched_portion is derived from REGEXP_EXTRACT.
fn build_final_query(
    state_cte_name: &SimpleIdentifier,
    regex_pattern: &str,
    first_pattern_vars: &[String],
    base_columns: &[SimpleIdentifier],
    by_clause_bindings: &[ByClauseBinding],
    agg_bindings: &[AggBinding],
    order_by: &[OrderByExpression],
    from_env: &Environment,
) -> Result<SQLQuery, TranslationErrors> {
    let mut output_builder = ProjectionBuilder::default();

    // Collect the first component of BY and AGG identifiers for deduplication.
    // We need to exclude base columns that would conflict with BY/AGG output columns.
    // For nested paths like `user.name`, we exclude the top-level `user` since
    // ProjectionBuilder will create a nested struct for it.
    let by_first_names: Vec<&SimpleIdentifier> = by_clause_bindings
        .iter()
        .map(|b| b.identifier.first())
        .collect();
    let agg_first_names: Vec<&SimpleIdentifier> =
        agg_bindings.iter().map(|b| b.identifier.first()).collect();

    // Add all base columns (pass through all input fields), skipping those that will be
    // added by BY or AGG clauses to avoid ambiguous column references
    for col in base_columns {
        if !by_first_names.contains(&col) && !agg_first_names.contains(&col) {
            let col_type = from_env
                .lookup(&col.clone().into())
                .unwrap_or(Type::Unknown);
            output_builder.bind(
                col.clone().into(),
                ColumnReference::new(col.clone().into()).into(),
                col_type,
            );
        }
    }

    // Add BY clause columns (aliased from prefixed names back to original names)
    // Using ProjectionBuilder properly handles nested identifiers like `user.name`
    for binding in by_clause_bindings {
        output_builder.bind(
            binding.identifier.clone(),
            ColumnReference::new(binding.prefixed_name.clone().into()).into(),
            binding.typ.clone(),
        );
    }

    // Add AGG columns - compute using match length from __labels_array
    // Match length = CARDINALITY(SPLIT(REGEXP_EXTRACT(__state, pattern), ','))
    // This gives us the number of rows in the actual match, not the whole window
    if !agg_bindings.is_empty() {
        // Build expression for match length:
        // CARDINALITY(SPLIT(REGEXP_EXTRACT(__state, '<pattern>'), ','))
        let match_length_expr: SQLExpression = FunctionCallApply::with_one(
            "CARDINALITY",
            FunctionCallApply::with_two(
                "SPLIT",
                FunctionCallApply::with_two(
                    "REGEXP_EXTRACT",
                    ColumnReference::new(SimpleIdentifier::new("__state").into()).into(),
                    StringLiteral::new(regex_pattern).into(),
                )
                .into(),
                StringLiteral::new(",").into(),
            )
            .into(),
        )
        .into();

        for binding in agg_bindings {
            // Transform the AGG expression to use the match length
            let transformed =
                transform_agg_for_match_length(&binding.translation.sql, &match_length_expr);
            let typ = get_agg_output_type(binding);
            output_builder.bind(binding.identifier.clone(), transformed.sql, typ);
        }
    }

    // Build projections from the ProjectionBuilder
    let projections = output_builder
        .build_projections()
        .map_err(|e| TranslationError::fatal("build_final_query", e.into()).single())?;

    // Filter: (__pattern_label IN ('a', 'b', ...)) AND REGEXP_LIKE(__state, '^pattern')
    // For patterns with optional first elements, we need to match any of the possible starting vars
    let pattern_filter: SQLExpression = if first_pattern_vars.len() == 1 {
        BinaryOperatorApply::new(
            Operator::Eq,
            ColumnReference::new(SimpleIdentifier::new("__pattern_label").into()).into(),
            StringLiteral::new(&first_pattern_vars[0]).into(),
        )
        .into()
    } else {
        // Use IN for multiple possible starting variables
        BinaryOperatorApply::new(
            Operator::In,
            ColumnReference::new(SimpleIdentifier::new("__pattern_label").into()).into(),
            FunctionCallApply::with_positional(
                "",
                first_pattern_vars
                    .iter()
                    .map(|v| StringLiteral::new(v).into())
                    .collect(),
            )
            .into(),
        )
        .into()
    };

    let regex_filter: SQLExpression = FunctionCallApply::with_two(
        "REGEXP_LIKE",
        ColumnReference::new(SimpleIdentifier::new("__state").into()).into(),
        StringLiteral::new(regex_pattern).into(),
    )
    .into();

    let filter: SQLExpression =
        BinaryOperatorApply::new(Operator::And, pattern_filter, regex_filter).into();

    Ok(SQLQuery::default()
        .from(TableReference::new(state_cte_name.clone().into()).into())
        .select(projections)
        .where_(filter)
        .order_by(order_by.to_vec()))
}

/// Extract FROM clauses from pattern.
///
/// This extracts unique FROM clause bindings. The same pattern variable can appear
/// multiple times in the pattern sequence (e.g., `A B+ A`), but each variable must
/// be bound to a single data source. Duplicate definitions like `A=table1 A=table2`
/// will produce an error.
fn extract_from_clauses_from_pattern(
    ctx: &Vec<Rc<PatternContextAll<'static>>>,
    cte_env: &Arc<Environment>,
    statement_context: &Rc<QueryTranslationContext>,
) -> Result<OrderMap<SimpleIdentifier, Rc<FromClauseContextAll<'static>>>, TranslationErrors> {
    let mut result: OrderMap<SimpleIdentifier, Rc<FromClauseContextAll<'static>>> = OrderMap::new();

    for pctx in ctx {
        match pctx.as_ref() {
            PatternContextAll::QuantifiedContext(qctx) => {
                let fctx = TranslationErrors::expect(qctx, qctx.fromClause())?;
                let fc = HamelinFromClause::new(
                    fctx.clone(),
                    cte_env.clone(),
                    statement_context.clone(),
                );
                let maybe_si = fc.table_alias().map(|si| si.to_sql()).transpose()?;
                let si = maybe_si.unwrap_or(fc.table_identifier()?.last().clone());

                // Check for duplicate pattern variable definitions
                // The same variable appearing multiple times in the sequence is OK (e.g., A B+ A),
                // but only if it refers to the same FROM clause (same binding)
                if let Some(existing_fctx) = result.get(&si) {
                    // Compare the table identifiers to see if they're the same binding
                    let existing_fc = HamelinFromClause::new(
                        existing_fctx.clone(),
                        cte_env.clone(),
                        statement_context.clone(),
                    );
                    let existing_table = existing_fc.table_identifier()?;
                    let new_table = fc.table_identifier()?;

                    if existing_table != new_table {
                        return Err(TranslationError::msg(
                            qctx,
                            &format!(
                                "Duplicate pattern variable '{}' bound to different data sources",
                                si.name
                            ),
                        )
                        .single());
                    }
                    // Same binding, skip (already in result)
                } else {
                    result.insert(si, fctx);
                }
            }
            PatternContextAll::NestedContext(nctx) => {
                let nested = extract_from_clauses_from_pattern(
                    &nctx.pattern_all(),
                    cte_env,
                    statement_context,
                )?;
                for (si, fctx) in nested {
                    if let Some(existing_fctx) = result.get(&si) {
                        let existing_fc = HamelinFromClause::new(
                            existing_fctx.clone(),
                            cte_env.clone(),
                            statement_context.clone(),
                        );
                        let new_fc = HamelinFromClause::new(
                            fctx.clone(),
                            cte_env.clone(),
                            statement_context.clone(),
                        );
                        let existing_table = existing_fc.table_identifier()?;
                        let new_table = new_fc.table_identifier()?;

                        if existing_table != new_table {
                            return Err(TranslationError::msg(
                                nctx,
                                &format!(
                                    "Duplicate pattern variable '{}' bound to different data sources",
                                    si.name
                                ),
                            )
                            .single());
                        }
                    } else {
                        result.insert(si, fctx);
                    }
                }
            }
            PatternContextAll::Error(e) => {
                return Err(TranslationError::msg(e, "parse error").single())
            }
        }
    }

    Ok(result)
}

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

    #[test]
    fn test_pattern_to_regex_simple_sequence() {
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::One),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a,b");
    }

    #[test]
    fn test_pattern_to_regex_one_or_more_middle() {
        // A B+ A - same variable can appear multiple times in pattern
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::OneOrMore),
            PatternElement::new("A".to_string(), PatternQuantifier::One),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a,(b,)+a");
    }

    #[test]
    fn test_pattern_to_regex_one_or_more_end() {
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::OneOrMore),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a,b(,b)*");
    }

    #[test]
    fn test_pattern_to_regex_optional_at_start() {
        // A? B - optional at start, required at end
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::ZeroOrOne),
            PatternElement::new("B".to_string(), PatternQuantifier::One),
        ];
        assert_eq!(pattern_to_regex(&elements), "^(a,)?b");
    }

    #[test]
    fn test_pattern_to_regex_optional_at_end() {
        // A B? - required at start, optional at end
        // When B matches zero times, state is just "a", so regex must be ^a(,b)?
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::ZeroOrOne),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a(,b)?");
    }

    #[test]
    fn test_pattern_to_regex_zero_or_more_at_end() {
        // A B* - required at start, zero-or-more at end
        // When B matches zero times, state is just "a", so regex must be ^a(,b(,b)*)?
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::ZeroOrMore),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a(,b(,b)*)?");
    }

    #[test]
    fn test_pattern_to_regex_multiple_optional_at_end() {
        // A B? C? - required, then two optional
        // Should match "a", "a,b", "a,c", "a,b,c"
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::ZeroOrOne),
            PatternElement::new("C".to_string(), PatternQuantifier::ZeroOrOne),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a(,b)?(,c)?");
    }

    #[test]
    fn test_pattern_to_regex_one_or_more_followed_by_optional() {
        // A+ B? - one-or-more followed by optional
        // Should match "a", "a,a", "a,b", "a,a,b"
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::OneOrMore),
            PatternElement::new("B".to_string(), PatternQuantifier::ZeroOrOne),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a(,a)*(,b)?");
    }

    #[test]
    fn test_pattern_to_regex_complex() {
        // A B+ A B pattern - variables can repeat in sequence
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::OneOrMore),
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::One),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a,(b,)+a,b");
    }

    #[test]
    fn test_pattern_to_regex_required_optional_required() {
        // A B? C - required, optional, required
        // Must match "a,c" when B is absent
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::One),
            PatternElement::new("B".to_string(), PatternQuantifier::ZeroOrOne),
            PatternElement::new("C".to_string(), PatternQuantifier::One),
        ];
        assert_eq!(pattern_to_regex(&elements), "^a,(b,)?c");
    }

    #[test]
    fn test_pattern_to_regex_all_optional_from_start() {
        // A? B? - both optional from the start
        // Must NOT have leading comma since state string never starts with comma
        // Should match "", "a", "b", "a,b"
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::ZeroOrOne),
            PatternElement::new("B".to_string(), PatternQuantifier::ZeroOrOne),
        ];
        // First element: (a)? - no leading comma
        // Second element: (,?b)? - optional comma because A might or might not have matched
        // This correctly matches "", "a", "b", "a,b"
        assert_eq!(pattern_to_regex(&elements), "^(a)?(,?b)?");
    }

    #[test]
    fn test_pattern_to_regex_zero_or_more_at_start() {
        // A* B - zero or more at start followed by required
        // Should match "b", "a,b", "a,a,b"
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::ZeroOrMore),
            PatternElement::new("B".to_string(), PatternQuantifier::One),
        ];
        assert_eq!(pattern_to_regex(&elements), "^(a,)*b");
    }

    #[test]
    fn test_pattern_to_regex_all_zero_or_more() {
        // A* B* - both zero or more
        // Should match "", "a", "a,a", "b", "b,b", "a,b", "a,a,b,b"
        let elements = vec![
            PatternElement::new("A".to_string(), PatternQuantifier::ZeroOrMore),
            PatternElement::new("B".to_string(), PatternQuantifier::ZeroOrMore),
        ];
        // First element: (a(,a)*)? - no leading comma
        // Second element: (,?b(,b)*)? - optional comma because A might or might not have matched
        // This correctly matches "", "a", "b", "a,b", "a,a,b,b"
        assert_eq!(pattern_to_regex(&elements), "^(a(,a)*)?(,?b(,b)*)?");
    }
}