use std::ops::RangeInclusive;
use std::sync::Arc;

use derive_more::derive::{From, TryUnwrap};

use crate::{
    err::TranslationError,
    func::def::{FunctionTranslationContext, SpecialPosition},
    tree::{
        ast::{
            clause::{
                Assignment, AssignmentClause, FromClause, SortExpression, SortOrder, TableAlias,
            },
            command::{
                AggCommand, AppendCommand, Command, CommandKind, DropCommand, ExplodeCommand,
                FromCommand, JoinCommand, LetCommand, LimitCommand, LookupCommand, MatchCommand,
                NestCommand, ParseCommand, RowsCommand, SelectCommand, SortCommand, UnionCommand,
                UnnestCommand, WhereCommand, WindowCommand, WithinCommand,
            },
            expression::{Expression, ExpressionKind, FieldReference, UnaryPrefixOperator},
            identifier::{Identifier, ParsedIdentifier, ParsedSimpleIdentifier, SimpleIdentifier},
            node::{Span, Spannable},
            ops::UnaryPrefixOp,
        },
        typed_ast::{
            clause::TypedTableAlias,
            context::ExpressionTranslationContext,
            environment::TypeEnvironment,
            expression::{
                MapExpressionAlgebra, TypedErrorExpression, TypedExpression, TypedExpressionKind,
            },
        },
    },
    types::{
        matcher::{
            AnyMatcher, ArrayMatcher, BaseMatcher, ExactMatcher, IntervalMatcher, OrMatcher,
            RangeMatcher,
        },
        struct_type::Struct,
        Type, BOOLEAN, CALENDAR_INTERVAL, INT, INTERVAL, ROWS, STRING, TIMESTAMP, UNKNOWN,
    },
};

use super::{
    clause::{
        find_identifier_conflict, Projections, ResolvedEnvironment, TypedAssignment,
        TypedFromClause, TypedSelection, TypedTableReference,
    },
    context::StatementTranslationContext,
    pattern::TypedPattern,
};

// ============================================================================
// Side Effect - Declared by commands that have side effects
// ============================================================================

/// Side effect of a command/statement
///
/// Commands that perform mutations (DML/DDL) declare their side effect.
/// This is extracted during type-checking and propagated to the pipeline/statement.
#[derive(Debug, Clone)]
pub enum SideEffect {
    /// No side effect - pure query that returns data
    None,

    /// DML - appends data to a table
    Append {
        table: TypedTableReference,
        distinct_by: Vec<TypedSelection>,
    },
    // Future: Insert, Update, Delete, CreateTable, DropTable, etc.
}

impl SideEffect {
    /// Check if this is a pure query (no side effects)
    pub fn is_none(&self) -> bool {
        matches!(self, SideEffect::None)
    }

    /// Check if this is a DML operation
    pub fn is_dml(&self) -> bool {
        matches!(self, SideEffect::Append { .. })
    }
}

/// A typed command in a pipeline
///
/// TypedCommand represents a type-checked command that transforms data.
/// Each command tracks:
/// - The input schema (bindings available coming into this command)
/// - The output schema (bindings available going out of this command)
/// - The specific typed command kind with type-checked sub-structures
/// - Any errors encountered during type checking (for error recovery)
#[derive(Debug, Clone)]
pub struct TypedCommand {
    /// Reference to the original AST command (for error reporting and IDE features)
    pub ast: Arc<Command>,

    /// Schema available as input to this command
    pub input_schema: Arc<TypeEnvironment>,

    /// Schema available as output from this command
    pub output_schema: Arc<TypeEnvironment>,

    /// The specific typed command
    pub kind: TypedCommandKind,
}

impl TypedCommand {
    /// Type-check a command and produce a typed command
    pub fn from_ast_with_context(
        cmd: Arc<Command>,
        input_schema: Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Self {
        let (kind, output_schema) =
            TypedCommandKind::from_ast_with_context(&cmd, &input_schema, ctx);

        Self {
            ast: cmd,
            input_schema,
            output_schema,
            kind,
        }
    }

    /// Find the first expression in this command matching the predicate.
    ///
    /// Traverses all expressions in the command and their sub-expressions,
    /// returning immediately when a match is found (early exit).
    pub fn find_expression<F>(&self, predicate: &mut F) -> Option<&TypedExpression>
    where
        F: FnMut(&TypedExpression) -> bool,
    {
        self.kind.find_expression(predicate)
    }

    /// Transform all expressions in this command using a MapExpressionAlgebra.
    ///
    /// Returns untyped `Arc<Command>` - caller must re-typecheck the pipeline.
    /// The command's structure is preserved, but all expressions are transformed
    /// through the algebra and rebuilt as untyped AST.
    pub fn cata_expressions(&self, alg: &mut impl MapExpressionAlgebra) -> Arc<Command> {
        Arc::new(Command {
            span: self.ast.span.clone(),
            kind: self.kind.cata_expressions(alg, &self.ast),
        })
    }
}

/// Typed command kinds
///
/// Each variant represents a type-checked command.
/// Commands that fail semantic validation become Error variants,
/// but still compute their best-effort output schema for IDE support.
#[derive(Debug, Clone, From, TryUnwrap)]
pub enum TypedCommandKind {
    /// WHERE command - filters rows based on predicates
    Where(TypedWhereCommand),

    /// WITHIN command - filters rows within a time window
    Within(TypedWithinCommand),

    /// LIMIT command - limits row count
    Limit(TypedLimitCommand),

    /// SORT command - orders rows
    Sort(TypedSortCommand),

    /// LET command - adds new fields to the schema
    Let(TypedLetCommand),

    /// SELECT command - replaces or extends the schema
    Select(TypedSelectCommand),

    /// DROP command - removes fields from the schema
    Drop(TypedDropCommand),

    /// PARSE command - extracts fields from strings
    Parse(TypedParseCommand),

    /// FROM command - loads data from tables
    From(TypedFromCommand),

    /// UNION command - combines multiple table sources
    Union(TypedUnionCommand),

    /// JOIN command - performs INNER join
    Join(TypedJoinCommand),

    /// LOOKUP command - performs LEFT join
    Lookup(TypedLookupCommand),

    /// AGG command - aggregates data with GROUP BY
    Agg(TypedAggCommand),

    /// WINDOW command - applies window functions with PARTITION BY and ORDER BY
    Window(TypedWindowCommand),

    /// MATCH command - performs ordered pattern matching over event sequences
    Match(TypedMatchCommand),

    /// APPEND command - writes pipeline data to a table (DML)
    Append(TypedAppendCommand),

    /// EXPLODE command - explodes array fields into multiple rows
    Explode(TypedExplodeCommand),

    /// UNNEST command - unnests nested struct/array-of-struct fields into columns
    Unnest(TypedUnnestCommand),

    /// NEST command - wraps all input fields into a single nested struct field
    Nest(TypedNestCommand),

    /// Fatal command error - command failed semantic validation
    /// Still computes output schema for downstream IDE features
    Error(Arc<TranslationError>),
}

impl TypedCommandKind {
    /// Type-check a command and produce a typed command with output schema
    ///
    /// Returns the typed command kind and the output schema.
    /// For schema-preserving commands, output_schema = input_schema.
    /// For schema-transforming commands, output_schema is computed from the command's semantics.
    /// For error commands, output_schema is best-effort based on what we can determine.
    pub fn from_ast_with_context(
        cmd: &Arc<Command>,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> (Self, Arc<TypeEnvironment>) {
        let result: Result<(TypedCommandKind, Arc<TypeEnvironment>), Arc<TranslationError>> =
            match &cmd.kind {
                CommandKind::Where(where_cmd) => {
                    TypedWhereCommand::from_ast_with_context(where_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Within(within_cmd) => {
                    TypedWithinCommand::from_ast_with_context(within_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Limit(limit_cmd) => {
                    TypedLimitCommand::from_ast_with_context(limit_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Sort(sort_cmd) => {
                    TypedSortCommand::from_ast_with_context(sort_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Let(let_cmd) => {
                    TypedLetCommand::from_ast_with_context(let_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Select(select_cmd) => {
                    TypedSelectCommand::from_ast_with_context(select_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Drop(drop_cmd) => {
                    TypedDropCommand::from_ast_with_context(drop_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Parse(parse_cmd) => {
                    TypedParseCommand::from_ast_with_context(parse_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::From(from_cmd) => {
                    TypedFromCommand::from_ast_with_context(from_cmd, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Union(union_cmd) => {
                    TypedUnionCommand::from_ast_with_context(union_cmd, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Join(join_cmd) => {
                    TypedJoinCommand::from_ast_with_context(join_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Lookup(lookup_cmd) => {
                    TypedLookupCommand::from_ast_with_context(lookup_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Agg(agg_cmd) => {
                    TypedAggCommand::from_ast_with_context(agg_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Window(window_cmd) => {
                    TypedWindowCommand::from_ast_with_context(window_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Match(match_cmd) => {
                    TypedMatchCommand::from_ast_with_context(match_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Append(append_cmd) => {
                    TypedAppendCommand::from_ast_with_context(append_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Explode(explode_cmd) => {
                    TypedExplodeCommand::from_ast_with_context(explode_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Unnest(unnest_cmd) => {
                    TypedUnnestCommand::from_ast_with_context_unnest(unnest_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Nest(nest_cmd) => {
                    TypedNestCommand::from_ast_with_context(nest_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Rows(rows_cmd) => {
                    TypedUnnestCommand::from_ast_with_context_rows(rows_cmd, input_schema, ctx)
                        .map(|(typed, schema)| (typed.into(), schema))
                }

                CommandKind::Error(err) => {
                    // Emit AST-level error to diagnostic sink
                    ctx.add_error(err.clone());
                    return (TypedCommandKind::Error(err.clone()), input_schema.clone());
                }
            };

        match result {
            Ok((kind, schema)) => (kind, schema),
            Err(err) => (TypedCommandKind::Error(err), input_schema.clone()),
        }
    }

    /// Get the side effect of this command, if any
    ///
    /// Only APPEND commands have side effects currently.
    /// All other commands are pure queries that don't modify data.
    pub fn side_effect(&self) -> Option<SideEffect> {
        match self {
            TypedCommandKind::Append(append_cmd) => Some(SideEffect::Append {
                table: append_cmd.table.clone(),
                distinct_by: append_cmd.distinct_by.clone(),
            }),
            // All other commands are pure queries
            _ => None,
        }
    }

    /// Find the first expression in this command matching the predicate.
    ///
    /// Traverses all expressions in the command and their sub-expressions.
    pub fn find_expression<F>(&self, predicate: &mut F) -> Option<&TypedExpression>
    where
        F: FnMut(&TypedExpression) -> bool,
    {
        match self {
            TypedCommandKind::Where(cmd) => cmd.predicate.find(predicate),
            TypedCommandKind::Within(cmd) => cmd.duration.find(predicate),
            TypedCommandKind::Limit(cmd) => cmd.count.find(predicate),
            TypedCommandKind::Sort(cmd) => cmd
                .expressions
                .iter()
                .find_map(|s| s.expression.find(predicate)),
            TypedCommandKind::Let(cmd) => cmd
                .projections
                .assignments
                .iter()
                .find_map(|a| a.expression.find(predicate)),
            TypedCommandKind::Select(cmd) => cmd
                .projections
                .assignments
                .iter()
                .find_map(|a| a.expression.find(predicate)),
            TypedCommandKind::Drop(_) => None, // No expressions
            TypedCommandKind::Parse(cmd) => cmd.source.as_ref().and_then(|s| s.find(predicate)),
            TypedCommandKind::From(_) => None, // No expressions in FROM
            TypedCommandKind::Union(_) => None, // No expressions in UNION
            TypedCommandKind::Join(cmd) => cmd.condition.as_ref().and_then(|c| c.find(predicate)),
            TypedCommandKind::Lookup(cmd) => cmd.condition.as_ref().and_then(|c| c.find(predicate)),
            TypedCommandKind::Agg(cmd) => cmd
                .aggregates
                .assignments
                .iter()
                .find_map(|a| a.expression.find(predicate))
                .or_else(|| {
                    cmd.group_by
                        .assignments
                        .iter()
                        .find_map(|a| a.expression.find(predicate))
                })
                .or_else(|| {
                    cmd.sort_by
                        .iter()
                        .find_map(|s| s.expression.find(predicate))
                }),
            TypedCommandKind::Window(cmd) => cmd
                .projections
                .assignments
                .iter()
                .find_map(|a| a.expression.find(predicate))
                .or_else(|| {
                    cmd.group_by
                        .assignments
                        .iter()
                        .find_map(|a| a.expression.find(predicate))
                })
                .or_else(|| {
                    cmd.sort_by
                        .iter()
                        .find_map(|s| s.expression.find(predicate))
                })
                .or_else(|| cmd.within.as_ref().and_then(|w| w.find(predicate))),
            TypedCommandKind::Match(cmd) => cmd
                .agg
                .assignments
                .iter()
                .find_map(|a| a.expression.find(predicate))
                .or_else(|| {
                    cmd.group_by
                        .assignments
                        .iter()
                        .find_map(|a| a.expression.find(predicate))
                })
                .or_else(|| cmd.sort.iter().find_map(|s| s.expression.find(predicate)))
                .or_else(|| cmd.within.as_ref().and_then(|w| w.find(predicate))),
            TypedCommandKind::Append(_) => None, // Selections don't have expressions
            TypedCommandKind::Explode(cmd) => cmd
                .items
                .iter()
                .find_map(|item| item.assignment.expression.find(predicate)),
            TypedCommandKind::Unnest(_) => None, // Just identifiers
            TypedCommandKind::Nest(_) => None,   // Just identifiers
            TypedCommandKind::Error(_) => None,
        }
    }

    /// Transform all expressions in this command using a MapExpressionAlgebra.
    ///
    /// Returns the untyped CommandKind with transformed expressions.
    pub fn cata_expressions(
        &self,
        alg: &mut impl MapExpressionAlgebra,
        original_ast: &Command,
    ) -> CommandKind {
        match self {
            TypedCommandKind::Where(cmd) => WhereCommand {
                condition: cmd.predicate.cata(alg),
            }
            .into(),

            TypedCommandKind::Within(cmd) => WithinCommand {
                duration: cmd.duration.cata(alg),
            }
            .into(),

            TypedCommandKind::Limit(cmd) => LimitCommand {
                count: cmd.count.cata(alg),
            }
            .into(),

            TypedCommandKind::Sort(cmd) => SortCommand {
                expressions: cmd
                    .expressions
                    .iter()
                    .map(|s| {
                        Arc::new(SortExpression {
                            span: s.ast.span.clone(),
                            expression: s.expression.cata(alg),
                            order: Some(s.order.clone()),
                        })
                    })
                    .collect(),
            }
            .into(),

            TypedCommandKind::Let(cmd) => LetCommand {
                assignments: cmd
                    .projections
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        })
                    })
                    .collect(),
            }
            .into(),

            TypedCommandKind::Select(cmd) => SelectCommand {
                clauses: cmd
                    .projections
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
            }
            .into(),

            TypedCommandKind::Drop(_) => {
                // DROP has no expressions - preserve original AST
                original_ast.kind.clone()
            }

            TypedCommandKind::Parse(cmd) => ParseCommand {
                source: cmd.source.as_ref().map(|s| s.cata(alg)),
                pattern: cmd.pattern_ast.clone(),
                identifiers: cmd.identifiers.clone(),
                nodrop: cmd.nodrop,
            }
            .into(),

            TypedCommandKind::From(_) => original_ast.kind.clone(),
            TypedCommandKind::Union(_) => original_ast.kind.clone(),

            TypedCommandKind::Join(cmd) => JoinCommand {
                other: Arc::new(FromClause::TableAlias(cmd.right.ast.clone())),
                on_condition: cmd.condition.as_ref().map(|c| c.cata(alg)),
            }
            .into(),

            TypedCommandKind::Lookup(cmd) => LookupCommand {
                other: Arc::new(FromClause::TableAlias(cmd.right.ast.clone())),
                on_condition: cmd.condition.as_ref().map(|c| c.cata(alg)),
            }
            .into(),

            TypedCommandKind::Agg(cmd) => AggCommand {
                clauses: cmd
                    .aggregates
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
                group_by: cmd
                    .group_by
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
                sort: cmd
                    .sort_by
                    .iter()
                    .map(|s| {
                        Arc::new(SortExpression {
                            span: s.ast.span.clone(),
                            expression: s.expression.cata(alg),
                            order: Some(s.order.clone()),
                        })
                    })
                    .collect(),
            }
            .into(),

            TypedCommandKind::Window(cmd) => WindowCommand {
                clauses: cmd
                    .projections
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
                group_by: cmd
                    .group_by
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
                sort: cmd
                    .sort_by
                    .iter()
                    .map(|s| {
                        Arc::new(SortExpression {
                            span: s.ast.span.clone(),
                            expression: s.expression.cata(alg),
                            order: Some(s.order.clone()),
                        })
                    })
                    .collect(),
                within: cmd.within.as_ref().map(|w| w.cata(alg)),
            }
            .into(),

            TypedCommandKind::Match(cmd) => MatchCommand {
                pattern: cmd.patterns.iter().map(|p| p.to_ast()).collect(),
                agg: cmd
                    .agg
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
                group_by: cmd
                    .group_by
                    .assignments
                    .iter()
                    .map(|a| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
                sort: cmd
                    .sort
                    .iter()
                    .map(|s| {
                        Arc::new(SortExpression {
                            span: s.ast.span.clone(),
                            expression: s.expression.cata(alg),
                            order: Some(s.order.clone()),
                        })
                    })
                    .collect(),
                within: cmd.within.as_ref().map(|w| w.cata(alg)),
            }
            .into(),

            TypedCommandKind::Append(_) => {
                // Selections don't have expressions - preserve original AST
                original_ast.kind.clone()
            }

            TypedCommandKind::Explode(cmd) => {
                let clauses = cmd
                    .items
                    .iter()
                    .map(|item| {
                        Arc::new(AssignmentClause::Assignment(Assignment {
                            span: item.assignment.span().into(),
                            identifier: item.assignment.identifier.clone(),
                            expression: item.assignment.expression.cata(alg),
                        }))
                    })
                    .collect();

                ExplodeCommand::multiple(clauses).into()
            }

            TypedCommandKind::Unnest(_) => original_ast.kind.clone(),
            TypedCommandKind::Nest(_) => original_ast.kind.clone(),
            TypedCommandKind::Error(_) => original_ast.kind.clone(),
        }
    }
}

// ============================================================================
// WHERE Command - Preserves Schema
// ============================================================================

/// WHERE command - filters rows based on predicates
///
/// Preserves the schema unchanged.
///
/// Example:
/// ```hamelin
/// FROM events | WHERE severity > 3
/// ```
#[derive(Debug, Clone)]
pub struct TypedWhereCommand {
    /// The boolean predicate expression
    pub predicate: Arc<TypedExpression>,
}

impl TypedWhereCommand {
    fn from_ast_with_context(
        cmd: &WhereCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking the predicate
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        // Type-check the predicate expression
        let predicate =
            TypedExpression::from_ast_with_context(cmd.condition.clone(), &mut expr_ctx);

        // WHERE preserves the schema unchanged
        let output_schema = input_schema.clone();

        Ok((
            Self {
                predicate: Arc::new(predicate),
            },
            output_schema,
        ))
    }
}

// ============================================================================
// WITHIN Command - Preserves Schema
// ============================================================================

/// WITHIN command - filters rows within a time window
///
/// Preserves the schema unchanged.
///
/// Example:
/// ```hamelin
/// FROM events | WITHIN 1h
/// ```
#[derive(Debug, Clone)]
pub struct TypedWithinCommand {
    /// The duration or time range expression
    pub duration: Arc<TypedExpression>,
}

impl TypedWithinCommand {
    fn from_ast_with_context(
        cmd: &WithinCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        // Type-check the duration expression
        let interval_or_timestamp = || {
            OrMatcher::default()
                .with(IntervalMatcher::default())
                .with(ExactMatcher::of(TIMESTAMP))
        };
        let range_or_single = OrMatcher::default()
            .with(interval_or_timestamp())
            .with(RangeMatcher::of(interval_or_timestamp()));

        let duration = TypedExpression::from_ast_with_expected_type(
            cmd.duration.clone(),
            &mut expr_ctx,
            &range_or_single,
        );

        // WITHIN preserves the schema unchanged
        let output_schema = input_schema.clone();

        Ok((
            Self {
                duration: Arc::new(duration),
            },
            output_schema,
        ))
    }
}

// ============================================================================
// LIMIT Command - Preserves Schema
// ============================================================================

/// LIMIT command - limits row count
///
/// Example:
/// ```hamelin
/// FROM events | LIMIT 100
/// ```
#[derive(Debug, Clone)]
pub struct TypedLimitCommand {
    /// The count expression (must be integer type)
    pub count: Arc<TypedExpression>,
}

impl TypedLimitCommand {
    fn from_ast_with_context(
        cmd: &LimitCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking the count
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        // Type-check the count expression (must be INT type)
        let count = TypedExpression::from_ast_with_expected_type(
            cmd.count.clone(),
            &mut expr_ctx,
            &ExactMatcher::of(INT),
        );

        // LIMIT preserves the schema unchanged
        let output_schema = input_schema.clone();

        Ok((
            Self {
                count: Arc::new(count),
            },
            output_schema,
        ))
    }
}

// ============================================================================
// SORT Command - Preserves Schema
// ============================================================================

/// SORT command - orders rows
///
/// Example:
/// ```hamelin
/// FROM events | SORT severity DESC, timestamp ASC
/// ```
#[derive(Debug, Clone)]
pub struct TypedSortCommand {
    /// The sort expressions with their ordering
    pub expressions: Vec<TypedSortExpression>,
}

/// A single sort expression with its ordering
#[derive(Debug, Clone)]
pub struct TypedSortExpression {
    /// Reference to the AST sort expression
    pub ast: Arc<SortExpression>,

    /// The expression to sort by
    pub expression: Arc<TypedExpression>,

    /// The sort order (ASC or DESC)
    pub order: SortOrder,
}

impl TypedSortExpression {
    /// Type-check a sort expression from AST
    pub fn from_ast_with_context(
        sort_expr: Arc<SortExpression>,
        ctx: &mut ExpressionTranslationContext,
    ) -> Self {
        let expression = TypedExpression::from_ast_with_expected_type(
            sort_expr.expression.clone(),
            ctx,
            &BaseMatcher::default(),
        );

        Self {
            ast: sort_expr.clone(),
            expression: Arc::new(expression),
            // Default to Asc when no order specified
            order: sort_expr.order.unwrap_or(SortOrder::Asc),
        }
    }
}

impl TypedSortCommand {
    fn from_ast_with_context(
        cmd: &SortCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking sort expressions
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        // Type-check each sort expression using the shared method
        let expressions = cmd
            .expressions
            .iter()
            .map(|sort_expr| {
                TypedSortExpression::from_ast_with_context(sort_expr.clone(), &mut expr_ctx)
            })
            .collect();

        // SORT preserves the schema unchanged
        let output_schema = input_schema.clone();

        Ok((Self { expressions }, output_schema))
    }
}

// ============================================================================
// LET Command - Extends Schema
// ============================================================================

/// LET command - adds new fields to the schema
///
/// Example:
/// ```hamelin
/// FROM events | LET severity_label = if(severity > 3, "high", "low")
/// ```
#[derive(Debug, Clone)]
pub struct TypedLetCommand {
    /// The typed assignments (name = expression)
    pub projections: Projections,
}

impl TypedLetCommand {
    fn from_ast_with_context(
        cmd: &LetCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking assignments
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        let projections =
            Projections::from_assignments_with_context(&cmd.assignments, &mut expr_ctx);

        let valid = projections.check_conflicts(ctx, &[]);

        let assignments: Vec<_> = projections
            .assignments
            .iter()
            .filter_map(|a| {
                a.identifier
                    .clone()
                    .valid()
                    .ok()
                    .filter(|id| valid.contains(id))
                    .map(|id| (id, a.expression.resolved_type.clone()))
            })
            .collect();

        let output_schema = Arc::new(TypeEnvironment::extend(&assignments, &input_schema));

        Ok((Self { projections }, output_schema))
    }
}

// ============================================================================
// SELECT Command - Replaces or Extends Schema
// ============================================================================

/// SELECT command - replaces or extends the schema
///
/// Example:
/// ```hamelin
/// FROM events | SELECT severity, message
/// FROM events | SELECT *, severity_label = if(severity > 3, "high", "low")
/// ```
#[derive(Debug, Clone)]
pub struct TypedSelectCommand {
    /// The typed projections
    pub projections: Projections,
}

impl TypedSelectCommand {
    fn from_ast_with_context(
        cmd: &SelectCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking projections
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        let projections =
            Projections::from_assignment_clauses_with_context(&cmd.clauses, &mut expr_ctx);

        let valid = projections.check_conflicts(ctx, &[]);

        // SELECT replaces schema entirely — build fresh env from projections only
        let mut output_env = TypeEnvironment::default();
        for typed_assignment in &projections.assignments {
            if let Ok(identifier) = typed_assignment.identifier.clone().valid() {
                if valid.contains(&identifier) {
                    output_env.bind_arc(
                        identifier,
                        typed_assignment.expression.resolved_type.clone(),
                    );
                }
            }
        }
        let output_schema = Arc::new(output_env);

        Ok((Self { projections }, output_schema))
    }
}

// ============================================================================
// DROP Command - Removes Fields from Schema
// ============================================================================

/// DROP command - removes fields from the schema
///
/// Example:
/// ```hamelin
/// FROM events | DROP temp_field, debug_info
/// ```
#[derive(Debug, Clone)]
pub struct TypedDropCommand {
    /// The identifiers that were successfully dropped (valid only)
    pub dropped_fields: Vec<Identifier>,
}

impl TypedDropCommand {
    fn from_ast_with_context(
        cmd: &DropCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        let mut dropped_fields = Vec::new();
        let mut output_env = (**input_schema).clone();

        for selection in &cmd.selections {
            let identifier = match selection.identifier.clone().valid() {
                Ok(id) => id,
                Err(_) => continue,
            };

            if input_schema.lookup(&identifier).is_some() {
                if let Err(e) = output_env.drop(&identifier) {
                    ctx.error("Cannot drop field")
                        .at(selection.as_ref())
                        .with_source(e)
                        .emit();
                } else {
                    dropped_fields.push(identifier);
                }
            } else {
                ctx.error(format!("Field '{}' not found", identifier))
                    .at(selection.as_ref())
                    .emit();
            }
        }

        let output_schema = Arc::new(output_env);

        Ok((Self { dropped_fields }, output_schema))
    }
}

// ============================================================================
// PARSE Command - Extends Schema
// ============================================================================

/// PARSE command - extracts fields from strings
///
/// Example:
/// ```hamelin
/// FROM logs | PARSE message "%{INT:status} %{WORD:method} %{PATH:path}"
/// ```
#[derive(Debug, Clone)]
pub struct TypedParseCommand {
    /// The source expression to parse (optional, defaults to event.original)
    pub source: Option<Arc<TypedExpression>>,

    /// The pattern string (must be a string literal)
    pub pattern: String,

    /// The pattern AST expression (for reconstructing untyped AST)
    pub pattern_ast: Arc<Expression>,

    /// The identifiers for extracted fields
    pub identifiers: Vec<ParsedIdentifier>,

    /// Whether to keep non-matching rows (NODROP flag)
    pub nodrop: bool,
}

impl TypedParseCommand {
    fn from_ast_with_context(
        cmd: &ParseCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        // Extract the pattern string literal (must be a literal, not a computed expression)
        // TODO: Add TypedExpression::from_ast_with_expected_literal to validate this properly
        let pattern = match &cmd.pattern.kind {
            ExpressionKind::StringLiteral(lit) => lit.value.clone(),
            _ => {
                // Pattern must be a string literal - use empty string as placeholder
                // The type error will be embedded in the TypedExpression
                String::new()
            }
        };

        // Type-check the source expression if provided (must be STRING type)
        let source = cmd.source.as_ref().map(|src| {
            Arc::new(TypedExpression::from_ast_with_expected_type(
                src.clone(),
                &mut expr_ctx,
                &ExactMatcher::of(STRING),
            ))
        });

        // Build PARSE fields, then prepend to input
        let mut parse_fields = Struct::default();
        let mut seen: Vec<(Identifier, RangeInclusive<usize>)> = Vec::new();

        for parsed_identifier in &cmd.identifiers {
            if let Ok(identifier) = parsed_identifier.clone().valid() {
                let is_throwaway = match &identifier {
                    Identifier::Simple(simple) => simple == "_",
                    _ => false,
                };

                if !is_throwaway {
                    if let Some(conflict) = find_identifier_conflict(&seen, &identifier) {
                        ctx.error(format!("conflicting definition for '{}'", identifier))
                            .at(parsed_identifier)
                            .add_context(conflict.existing_span, "previously defined here")
                            .with_note(conflict.note)
                            .emit();
                    } else {
                        seen.push((
                            identifier.clone(),
                            parsed_identifier.span().unwrap_or(0..=0),
                        ));
                        parse_fields.insert(identifier.clone(), STRING);
                    }
                }
            }
        }

        let output_schema = Arc::new(TypeEnvironment::new(
            input_schema.as_struct().prepend_overwrite(&parse_fields),
        ));

        // Return the TypedParseCommand with all identifiers (including parse errors)
        Ok((
            Self {
                source,
                pattern,
                pattern_ast: cmd.pattern.clone(),
                identifiers: cmd.identifiers.clone(),
                nodrop: cmd.nodrop,
            },
            output_schema,
        ))
    }
}

// ============================================================================
// FROM Command - Loads Data from Tables
// ============================================================================

/// FROM command - loads data from tables
///
/// Example:
/// ```hamelin
/// FROM events
/// FROM x = events, y = logs
/// ```
///
/// Key semantics:
/// - `FROM table` - fields merged at top level
/// - `FROM x = table1, y = table2` - creates both nested fields (x.field1, y.field1)
///   AND merged top-level fields (field1)
#[derive(Debug, Clone)]
pub struct TypedFromCommand {
    /// The FROM clauses with their resolved schemas
    pub clauses: Vec<TypedFromClause>,
}

impl TypedFromCommand {
    fn from_ast_with_context(
        cmd: &FromCommand,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        let mut typed_clauses = Vec::new();
        let mut merged_struct = Struct::default();

        for clause in &cmd.clauses {
            // Type-check this clause (includes catalog resolution)
            let typed_clause = TypedFromClause::from_ast_with_context(clause.clone(), ctx);

            // Get the resolved environment (empty if error)
            let env = typed_clause.environment();

            // For aliased tables, create prefixed bindings AND expose top-level fields
            let struct_to_merge = match &typed_clause {
                TypedFromClause::Reference(_) => {
                    // Plain reference - get the struct from the environment
                    env.as_struct().clone()
                }
                TypedFromClause::Alias(alias_clause) => {
                    // Get base schema struct
                    let base_struct = env.as_struct().clone();

                    // Create a struct with both:
                    // 1. Top-level fields (x, y) for direct access
                    // 2. Prefixed fields (t1.x, t1.y) for qualified access
                    let mut aliased_struct = base_struct.clone();

                    // Convert alias to identifier for struct field (extract valid identifier)
                    if let Ok(alias) = alias_clause.alias.clone().valid() {
                        // Add prefixed bindings by converting to nested struct
                        // This will create t1.x, t1.y, etc. when flattened
                        aliased_struct = aliased_struct.with(alias.into(), base_struct.into());
                    }
                    // Parse errors: skip prefixed bindings, only top-level fields available

                    aliased_struct
                }
                TypedFromClause::Error(_) => {
                    continue;
                }
            };

            // Merge with accumulated struct
            match merged_struct.merge(&struct_to_merge) {
                Ok(new_struct) => {
                    merged_struct = new_struct;
                    typed_clauses.push(typed_clause);
                }
                Err(e) => {
                    // Schema merge failed - create Error variant with context
                    let error = match &e.at {
                        Some(at) => ctx
                            .error(format!("Field {} is of type {}.", at, e.other))
                            .at(clause.as_ref())
                            .emit(),
                        None => ctx.error("Cannot merge schemas").at(clause.as_ref()).emit(),
                    };
                    typed_clauses.push(error.into());
                }
            }
        }

        // Convert merged Struct to TypeEnvironment
        let output_schema = Arc::new(merged_struct.into());

        Ok((
            Self {
                clauses: typed_clauses,
            },
            output_schema,
        ))
    }
}

// ============================================================================
// UNION Command - Combines Multiple Table Sources
// ============================================================================

/// UNION command - combines multiple table sources
///
/// Example:
/// ```hamelin
/// UNION events, logs
/// ```
///
/// Key differences from FROM:
/// - Does NOT support aliasing (no `x = table` syntax)
/// - Only merges fields at top level (no nesting)
#[derive(Debug, Clone)]
pub struct TypedUnionCommand {
    /// The FROM clauses with their resolved schemas
    /// All must be TableReference (no TableAlias allowed)
    pub clauses: Vec<TypedFromClause>,
}

impl TypedUnionCommand {
    fn from_ast_with_context(
        cmd: &UnionCommand,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        let mut typed_clauses = Vec::new();
        let mut merged_struct = Struct::default();

        for clause in &cmd.clauses {
            // Type-check this clause (includes catalog resolution)
            let typed_clause = TypedFromClause::from_ast_with_context(clause.clone(), ctx);

            // Validate that no aliasing is present - create Error variant
            if matches!(&typed_clause, TypedFromClause::Alias(_)) {
                let error = ctx
                    .error("UNION does not support aliased tables (x = table syntax). Use plain table references only.")
                    .at(clause.as_ref())
                    .emit();
                typed_clauses.push(error.into());
                continue;
            }

            // Get the resolved environment (empty if error)
            let env = typed_clause.environment();

            // Skip Error clauses without attempting merge
            if matches!(&typed_clause, TypedFromClause::Error(_)) {
                typed_clauses.push(typed_clause);
                continue;
            }

            // Convert to Struct for merging
            let struct_to_merge = env.as_struct().clone();

            // Merge with accumulated struct
            match merged_struct.merge(&struct_to_merge) {
                Ok(new_struct) => {
                    merged_struct = new_struct;
                    typed_clauses.push(typed_clause);
                }
                Err(e) => {
                    // Schema merge failed - create Error variant with context
                    let error = match &e.at {
                        Some(at) => ctx
                            .error(format!("Field {} is of type {}.", at, e.other))
                            .at(clause.as_ref())
                            .emit(),
                        None => ctx.error("Cannot merge schemas").at(clause.as_ref()).emit(),
                    };
                    typed_clauses.push(error.into());
                }
            }
        }

        // Convert merged Struct to TypeEnvironment
        let output_schema = Arc::new(merged_struct.into());

        Ok((
            Self {
                clauses: typed_clauses,
            },
            output_schema,
        ))
    }
}

// ============================================================================
// JOIN Command - INNER join
// ============================================================================

/// JOIN command - performs INNER join
///
/// Example:
/// ```hamelin
/// FROM events | JOIN alerts ON events.alert_id == alerts.id
/// FROM events | JOIN x = users ON events.user_id == x.id
/// ```
///
/// Key semantics:
/// - Creates INNER join (only matching rows)
/// - Supports aliasing (e.g., `JOIN x = table`)
/// - Implicit aliasing: `JOIN events` infers alias from table name
/// - ON condition is optional (defaults to CROSS JOIN behavior)
/// - Right side merged into output schema as nested struct
#[derive(Debug, Clone)]
pub struct TypedJoinCommand {
    /// The right side table alias (always has alias, inferred if not explicit)
    pub right: TypedTableAlias,

    /// The join condition (None = CROSS JOIN)
    pub condition: Option<Arc<TypedExpression>>,
}

impl TypedJoinCommand {
    /// Type-check a JOIN command
    fn from_ast_with_context(
        cmd: &JoinCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        let (right, condition, output_schema) =
            typecheck_join_or_lookup(&cmd.other, cmd.on_condition.as_ref(), input_schema, ctx)?;

        Ok((Self { right, condition }, output_schema))
    }
}

// ============================================================================
// LOOKUP Command - LEFT join
// ============================================================================

/// LOOKUP command - performs LEFT join
///
/// Example:
/// ```hamelin
/// FROM events | LOOKUP threat_intel ON events.ip == threat_intel.ip
/// FROM events | LOOKUP x = users ON events.user_id == x.id
/// ```
///
/// Key semantics:
/// - Creates LEFT join (all left rows preserved, right may be NULL)
/// - Supports aliasing (e.g., `LOOKUP x = table`)
/// - Implicit aliasing: `LOOKUP events` infers alias from table name
/// - ON condition is optional (defaults to CROSS JOIN behavior)
/// - Right side merged into output schema as nested struct
#[derive(Debug, Clone)]
pub struct TypedLookupCommand {
    /// The right side table alias (always has alias, inferred if not explicit)
    pub right: TypedTableAlias,

    /// The join condition (None = CROSS JOIN)
    pub condition: Option<Arc<TypedExpression>>,
}

impl TypedLookupCommand {
    /// Type-check a LOOKUP command
    fn from_ast_with_context(
        cmd: &LookupCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        let (right, condition, output_schema) =
            typecheck_join_or_lookup(&cmd.other, cmd.on_condition.as_ref(), input_schema, ctx)?;

        Ok((Self { right, condition }, output_schema))
    }
}

/// Shared helper for type-checking JOIN and LOOKUP commands
fn typecheck_join_or_lookup(
    other: &Arc<FromClause>,
    on_condition: Option<&Arc<Expression>>,
    input_schema: &Arc<TypeEnvironment>,
    ctx: &mut StatementTranslationContext,
) -> Result<
    (
        TypedTableAlias,
        Option<Arc<TypedExpression>>,
        Arc<TypeEnvironment>,
    ),
    Arc<TranslationError>,
> {
    // Type-check the right side (includes catalog resolution)
    let typed_clause = TypedFromClause::from_ast_with_context((*other).clone(), ctx);

    // JOIN/LOOKUP always requires an alias. Convert Reference to TypedTableAlias with inferred alias.
    let right_table_alias = match typed_clause {
        TypedFromClause::Reference(ref_clause) => {
            // Convert Reference to implicit TypedTableAlias by inferring alias from table name
            // Extract valid identifier, or return error if parse failed
            let valid_identifier = match ref_clause.ast.identifier.clone().valid() {
                Ok(id) => id,
                Err(e) => return Err(e.clone()), // Parse error, propagate
            };

            // Get the last component as the alias (e.g., "users" from "db.schema.users")
            let inferred_alias: SimpleIdentifier = valid_identifier.last().clone();

            TypedTableAlias {
                ast: TableAlias {
                    span: Span::NONE,
                    alias: ParsedSimpleIdentifier::Valid(inferred_alias.clone()),
                    table: (*ref_clause.ast).clone(),
                }
                .into(),
                alias: ParsedSimpleIdentifier::Valid(inferred_alias),
                resolved: ref_clause.resolved,
            }
        }
        TypedFromClause::Alias(alias_clause) => {
            // Already has explicit alias
            alias_clause
        }
        TypedFromClause::Error(err) => {
            // Propagate error, use input schema as fallback
            return Err(err);
        }
    };

    // Get the resolved right environment (empty if error)
    let right_env = right_table_alias.resolved.environment();

    // Convert alias to identifier for merging
    let alias = match right_table_alias.alias.clone().valid() {
        Ok(alias) => alias,
        Err(_) => {
            // Error on alias identifier - return error
            return Err(ctx
                .error("Could not determine alias for right side of join")
                .at(other.as_ref())
                .emit());
        }
    };

    // Append right-side fields nested under alias after input fields
    // (legacy behavior: left fields first, then right alias at end)
    let right_struct = right_env.as_struct().clone();
    let mut output = input_schema.as_struct().clone();
    output.insert(alias.into(), right_struct.into());
    let output_schema = Arc::new(TypeEnvironment::new(output));

    // Type-check the ON condition if present
    let typed_condition = on_condition.map(|cond| {
        let mut expr_ctx = ctx.default_expression_context(&output_schema);
        let typed_expr = TypedExpression::from_ast_with_expected_type(
            cond.clone(),
            &mut expr_ctx,
            &ExactMatcher::of(BOOLEAN),
        );

        Arc::new(typed_expr)
    });

    Ok((right_table_alias, typed_condition, output_schema))
}

// ============================================================================
// TypedAggCommand - Aggregation with GROUP BY
// ============================================================================

/// A type-checked AGG command
///
/// AGG performs aggregation over the input data, optionally grouping by expressions.
/// The expression context allows aggregate functions (count, sum, avg, etc.).
///
/// Schema transformation:
/// - Replaces the schema with aggregation results and group-by fields
/// - Each aggregate expression produces a field in the output
/// - Each group-by expression produces a field in the output
#[derive(Debug, Clone)]
pub struct TypedAggCommand {
    /// Aggregation expressions (assignments with aggregate functions)
    pub aggregates: Projections,

    /// GROUP BY expressions
    pub group_by: Projections,

    /// SORT BY expressions (for ordered aggregation)
    pub sort_by: Vec<TypedSortExpression>,
}

impl TypedAggCommand {
    fn from_ast_with_context(
        cmd: &AggCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context that allows aggregate functions
        let agg_fctx =
            FunctionTranslationContext::default().with_special_allowed(SpecialPosition::Agg);
        let mut agg_expr_ctx = ctx.expression_context(input_schema, agg_fctx);

        // Type-check aggregate expressions (aggregate functions allowed)
        let aggregates =
            Projections::from_assignment_clauses_with_context(&cmd.clauses, &mut agg_expr_ctx);

        // Type-check group-by expressions (regular expressions, no aggregates)
        let mut group_by_expr_ctx = ctx.default_expression_context(input_schema);
        let group_by = Projections::from_assignment_clauses_with_context(
            &cmd.group_by,
            &mut group_by_expr_ctx,
        );

        // AGG builds output from scratch: group-by fields + aggregate fields
        let mut output_env = TypeEnvironment::default();

        // Type-check sort-by expressions (regular expressions, no aggregates)
        let sort_by: Vec<TypedSortExpression> = {
            let mut sort_by_expr_ctx = ctx.default_expression_context(input_schema);
            cmd.sort
                .iter()
                .map(|sort_expr| {
                    TypedSortExpression::from_ast_with_context(
                        sort_expr.clone(),
                        &mut sort_by_expr_ctx,
                    )
                })
                .collect()
        };

        // Check for conflicts across group_by + aggregate assignments
        let valid = group_by.check_conflicts(ctx, &[&aggregates]);

        // Bind group_by fields first, then aggregate fields
        for assignment in group_by
            .assignments
            .iter()
            .chain(aggregates.assignments.iter())
        {
            if let Ok(identifier) = assignment.identifier.clone().valid() {
                if valid.contains(&identifier) {
                    output_env.bind_arc(identifier, assignment.expression.resolved_type.clone());
                }
            }
        }

        let output_schema = Arc::new(output_env);

        Ok((
            Self {
                aggregates,
                group_by,
                sort_by,
            },
            output_schema,
        ))
    }
}

// ============================================================================
// TypedWindowCommand - Window Functions with PARTITION BY and ORDER BY
// ============================================================================

/// A type-checked WINDOW command
///
/// WINDOW applies window functions over partitions of the input data.
/// Window functions can be aggregates (sum, avg, etc.) or analytical functions
/// (row_number, rank, lag, lead, etc.).
///
/// Schema transformation:
/// - Extends the schema with new fields from window function expressions
/// - Each window expression produces a field in the output
/// - Original fields are preserved
///
/// Example:
/// ```hamelin
/// FROM events
/// | WINDOW row_num = row_number() BY user_id SORT timestamp
/// | WINDOW running_total = sum(amount) BY user_id SORT timestamp WITHIN -5..0 rows
/// ```
#[derive(Debug, Clone)]
pub struct TypedWindowCommand {
    pub projections: Projections,
    pub group_by: Projections,
    pub sort_by: Vec<TypedSortExpression>,
    pub within: Option<Arc<TypedExpression>>,
}

impl TypedWindowCommand {
    fn from_ast_with_context(
        cmd: &WindowCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context that allows both aggregate and window functions
        let window_fctx = FunctionTranslationContext::default()
            .with_special_allowed(SpecialPosition::Agg)
            .with_special_allowed(SpecialPosition::Window);
        let mut window_expr_ctx = ctx.expression_context(input_schema, window_fctx);

        // Type-check window function expressions (window and aggregate functions allowed)
        let projections =
            Projections::from_assignment_clauses_with_context(&cmd.clauses, &mut window_expr_ctx);

        // Type-check order-by expressions FIRST to determine sort type for frame validation
        // Resolve timestamp field info before borrowing ctx mutably for expression context
        let timestamp_type = input_schema
            .lookup(&ctx.timestamp_field)
            .map(|rc_type| rc_type.as_ref().clone());
        let timestamp_field = ctx.timestamp_field.clone();

        let (order_by, sort_type) = {
            let mut order_by_expr_ctx = ctx.default_expression_context(input_schema);
            let mut order_by: Vec<TypedSortExpression> = cmd
                .sort
                .iter()
                .map(|sort_expr| {
                    TypedSortExpression::from_ast_with_context(
                        sort_expr.clone(),
                        &mut order_by_expr_ctx,
                    )
                })
                .collect();

            // Extract sort type from first sort expression, or default to timestamp field type
            let sort_type = order_by
                .first()
                .map(|sort_expr| sort_expr.expression.resolved_type.as_ref().clone())
                .or(timestamp_type);

            // When no explicit SORT BY is given, inject an implicit ORDER BY on the
            // timestamp field so backends get a valid window frame. This matches
            // legacy behavior where WINDOW always defaults to ORDER BY timestamp.
            if order_by.is_empty() && matches!(sort_type.as_ref(), Some(Type::Timestamp)) {
                let ts_ast = Arc::new(SortExpression {
                    span: Span::NONE,
                    expression: Arc::new(Expression::from(timestamp_field)),
                    order: None,
                });
                order_by.push(TypedSortExpression::from_ast_with_context(
                    ts_ast,
                    &mut order_by_expr_ctx,
                ));
            }

            (order_by, sort_type)
        };

        // Type-check partition-by expressions and frame (regular expressions, no window functions)
        // Frame validation depends on sort type determined above
        let (partition_by, frame) = {
            let mut partition_by_expr_ctx = ctx.default_expression_context(input_schema);
            let partition_by_result = Projections::from_assignment_clauses_with_context(
                &cmd.group_by,
                &mut partition_by_expr_ctx,
            );

            // Type-check frame expression if present, validating against sort type
            let frame = match (cmd.within.as_ref(), sort_type.as_ref()) {
                (Some(within_expr), Some(Type::Timestamp)) => {
                    // Timestamp can use INTERVAL, ROWS, or RANGE of INTERVAL or ROWS
                    let frame_matcher = OrMatcher::default()
                        .with(IntervalMatcher)
                        .with(ExactMatcher::of(ROWS))
                        .with(RangeMatcher::of(
                            OrMatcher::default()
                                .with(IntervalMatcher)
                                .with(ExactMatcher::of(ROWS)),
                        ));
                    let frame_expr = TypedExpression::from_ast_with_expected_type(
                        within_expr.clone(),
                        &mut partition_by_expr_ctx,
                        &frame_matcher,
                    );
                    Some(Arc::new(frame_expr))
                }
                (Some(within_expr), Some(base_type)) => {
                    // Other base types can use that type, ROWS, or RANGE of (that type or ROWS)
                    let frame_matcher = OrMatcher::default()
                        .with(ExactMatcher::of(base_type.clone()))
                        .with(ExactMatcher::of(ROWS))
                        .with(RangeMatcher::of(
                            OrMatcher::default()
                                .with(ExactMatcher::of(base_type.clone()))
                                .with(ExactMatcher::of(ROWS)),
                        ));
                    let frame_expr = TypedExpression::from_ast_with_expected_type(
                        within_expr.clone(),
                        &mut partition_by_expr_ctx,
                        &frame_matcher,
                    );
                    Some(Arc::new(frame_expr))
                }
                (Some(within_expr), None) => {
                    // WITHIN frame specified but no sort field available - this is an error
                    let error = partition_by_expr_ctx
                        .error("WITHIN frame requires an SORT clause or a timestamp field in the schema")
                        .at(within_expr.as_ref())
                        .emit();
                    // Return error expression
                    Some(Arc::new(TypedExpression {
                        ast: within_expr.clone(),
                        resolved_type: Arc::new(UNKNOWN),
                        kind: TypedErrorExpression { error }.into(),
                    }))
                }
                (None, _) => None,
            };

            (partition_by_result, frame)
        };

        // Same semantics as LET
        let valid = projections.check_conflicts(ctx, &[&partition_by]);

        let assignments: Vec<_> = projections
            .assignments
            .iter()
            .chain(partition_by.assignments.iter())
            .filter_map(|a| {
                a.identifier
                    .clone()
                    .valid()
                    .ok()
                    .filter(|id| valid.contains(id))
                    .map(|id| (id, a.expression.resolved_type.clone()))
            })
            .collect();

        let output_schema = Arc::new(TypeEnvironment::extend(&assignments, &input_schema));

        Ok((
            Self {
                projections,
                group_by: partition_by,
                sort_by: order_by,
                within: frame,
            },
            output_schema,
        ))
    }
}

// ============================================================================
// TypedMatchCommand - Pattern Matching for Event Sequences
// ============================================================================

/// A type-checked MATCH command
///
/// MATCH performs ordered pattern matching over event sequences.
/// Patterns represent named datasets that must occur in sequence.
/// Constraints (WHEN clause) specify conditions that must be satisfied.
///
/// Schema transformation:
/// - Adds pattern variables to the output environment
/// - Pattern variables are tracked separately from regular fields
/// - GROUP BY and SORT clauses work with regular fields
///
/// Example:
/// ```hamelin
/// MATCH
///   initial_access+
///   persistence+
///   defense_evasion+
/// WHEN
///   persistence = persistence.timestamp - last(initial_access.timestamp) <= 10m,
///   defense_evasion = defense_evasion.timestamp - last(persistence.timestamp) <= 10m
/// BY host.name
/// SORT BY timestamp
/// ```
#[derive(Debug, Clone)]
pub struct TypedMatchCommand {
    /// Type-checked pattern sequence
    pub patterns: Vec<TypedPattern>,

    /// Type-checked AGG expressions (aggregations over matched rows)
    pub agg: Projections,

    /// Type-checked group-by expressions
    pub group_by: Projections,

    /// Type-checked sort expressions
    pub sort: Vec<TypedSortExpression>,

    /// Type-checked within expression (optional)
    pub within: Option<TypedExpression>,
}

impl TypedMatchCommand {
    fn from_ast_with_context(
        cmd: &MatchCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Type-check all patterns
        let patterns = cmd
            .pattern
            .iter()
            .map(|p| TypedPattern::from_ast_with_context(p.clone(), ctx))
            .collect::<Vec<_>>();

        // Validate no pattern uses {0} quantifier — matching exactly zero times is meaningless
        // This recurses into nested patterns to catch e.g. (a{0} b)+
        for pattern in &patterns {
            if pattern.contains_zero_quantifier() {
                return Err(ctx
                    .error("pattern quantifier {0} is not allowed — matching exactly zero times is meaningless")
                    .emit());
            }
        }

        // Validate that at least one pattern is required (not all optional)
        if patterns.iter().all(|p| p.is_optional()) {
            return Err(ctx
                .error(
                    "MATCH requires at least one required pattern element \
                     (all patterns use * or ? quantifiers)",
                )
                .emit());
        }

        // Build the type environment for pattern bindings using a single Struct.
        // This environment contains:
        // - Flattened fields from each pattern's struct (e.g., timestamp, event_type)
        // - Pattern variable bindings themselves (e.g., e, A, B as struct types)
        //
        // Note: The same pattern variable CAN appear multiple times in a pattern sequence.
        // For example, `A B+ A` is valid. When the same variable appears again, we skip
        // adding it to the struct (it's already there with the same type).
        //
        // We use Struct::merge() to get the proper field ordering (Chris's magical merge order)
        // which matches how TypedFrom handles multiple from clauses.
        let mut merged_struct = Struct::default();

        for pattern in &patterns {
            let bindings = pattern.collect_pattern_bindings();
            for (name, typ) in bindings.into_iter() {
                // Expose fields from each pattern's struct at the top level
                // This allows BY host to work when host is a field in pattern A's struct
                if let Type::Struct(s) = &typ {
                    // Merge fields from this pattern's struct
                    // Using merge() ensures proper field ordering
                    match merged_struct.merge(s) {
                        Ok(new_struct) => merged_struct = new_struct,
                        Err(e) => {
                            return Err(ctx
                                .error(&format!(
                                    "Pattern field '{}' has conflicting types: {}",
                                    e.at.map(|a| a.to_string()).unwrap_or_default(),
                                    e.other
                                ))
                                .emit());
                        }
                    }
                }

                // Add the pattern binding itself (e, A, B, etc.) if not already present
                // Pattern variables ARE included in output schema (matches actual SQL output)
                // Skip if already present (same variable appearing multiple times in pattern)
                if !merged_struct.contains_key(&name) {
                    merged_struct = merged_struct.with(name.clone().into(), typ);
                }
            }
        }

        // Create environment from merged_struct prepended to input schema
        let pattern_env = Arc::new(TypeEnvironment::new(
            input_schema.as_struct().prepend_overwrite(&merged_struct),
        ));

        // Type-check AGG expressions with access to pattern bindings
        let agg_fctx =
            FunctionTranslationContext::default().with_special_allowed(SpecialPosition::Match);
        let mut agg_ctx = ctx.expression_context(&pattern_env, agg_fctx);
        let agg = Projections::from_assignment_clauses_with_context(&cmd.agg, &mut agg_ctx);

        // Type-check group-by expressions with access to pattern bindings
        let mut group_by_ctx = ctx.default_expression_context(&pattern_env);
        let group_by =
            Projections::from_assignment_clauses_with_context(&cmd.group_by, &mut group_by_ctx);

        // Type-check sort expressions with access to pattern bindings
        // WITHIN requires singular SORT (exactly one expression or implicit timestamp)
        let sort: Vec<TypedSortExpression> = {
            let has_within = cmd.within.is_some();
            let user_sort = &cmd.sort;

            if !user_sort.is_empty() {
                // User specified explicit sort expressions
                let mut sort_ctx = ctx.default_expression_context(&pattern_env);
                let typed_sorts: Vec<TypedSortExpression> = user_sort
                    .iter()
                    .map(|sort_expr| {
                        TypedSortExpression::from_ast_with_context(sort_expr.clone(), &mut sort_ctx)
                    })
                    .collect();

                // WITHIN requires singular SORT
                if has_within && typed_sorts.len() != 1 {
                    return Err(ctx
                        .error("MATCH with WITHIN requires exactly one SORT expression")
                        .emit());
                }

                // WITHIN requires SORT field type to be compatible with WITHIN type
                // We'll validate this after we type-check the WITHIN expression below

                typed_sorts
            } else {
                // No explicit SORT - use implicit timestamp if available
                let timestamp_field = ctx.timestamp_field.clone();

                if let Some(timestamp_type) = pattern_env.lookup(&timestamp_field) {
                    if *timestamp_type == TIMESTAMP {
                        let timestamp_simple = match &timestamp_field {
                            Identifier::Simple(s) => s.clone(),
                            Identifier::Compound(_) => {
                                return Err(ctx
                                    .error("timestamp field must be a simple identifier")
                                    .emit());
                            }
                        };

                        let mut sort_ctx = ctx.default_expression_context(&pattern_env);
                        let field_ref = FieldReference {
                            field_name: timestamp_simple.into(),
                        };
                        let ast_expr = Arc::new(Expression {
                            span: Span::NONE,
                            kind: field_ref.into(),
                        });
                        let typed_expr =
                            TypedExpression::from_ast_with_context(ast_expr.clone(), &mut sort_ctx);
                        let sort_expr = Arc::new(SortExpression {
                            span: Span::NONE,
                            expression: ast_expr,
                            order: None,
                        });

                        vec![TypedSortExpression {
                            ast: sort_expr,
                            expression: Arc::new(typed_expr),
                            order: SortOrder::Asc,
                        }]
                    } else {
                        if has_within {
                            return Err(ctx
                                .error("MATCH with WITHIN requires a timestamp field in the schema")
                                .emit());
                        }
                        vec![]
                    }
                } else {
                    if has_within {
                        return Err(ctx
                            .error("MATCH with WITHIN requires a timestamp field in the schema")
                            .emit());
                    }
                    vec![]
                }
            }
        };

        // Type-check within expression if present
        let within = if let Some(within_expr) = cmd.within.as_ref() {
            let mut within_ctx = ctx.default_expression_context(&pattern_env);

            // Build type matcher: accept intervals OR numeric types matching SORT field
            let expected_type = if !sort.is_empty() {
                let sort_type = &sort[0].expression.resolved_type;
                if **sort_type == TIMESTAMP {
                    // For TIMESTAMP, accept INTERVAL or CALENDAR_INTERVAL
                    OrMatcher::default()
                        .with(IntervalMatcher::default())
                        .with(ExactMatcher::of(CALENDAR_INTERVAL))
                } else {
                    // For numeric types, accept the same type as SORT field
                    OrMatcher::default().with(ExactMatcher::of(sort_type.as_ref().clone()))
                }
            } else {
                // No SORT, just accept intervals (for timestamp field case)
                OrMatcher::default()
                    .with(IntervalMatcher::default())
                    .with(ExactMatcher::of(CALENDAR_INTERVAL))
            };

            let typed_within = TypedExpression::from_ast_with_expected_type(
                within_expr.clone(),
                &mut within_ctx,
                &expected_type,
            );

            // Validate WITHIN expression is not negative by checking for semantic patterns
            // We check the original AST structure to detect negative intervals semantically
            if Self::is_negative_interval_expression(&typed_within.ast) {
                return Err(ctx.error("WITHIN interval cannot be negative").emit());
            }

            // Type check: WITHIN constraint is `last(SORT) - first(SORT) <= WITHIN`
            // So we need the subtraction result type to be compatible with WITHIN type
            // For TIMESTAMP: TIMESTAMP - TIMESTAMP = INTERVAL (should match WITHIN interval type)
            // For numeric types: INT - INT = INT (should match WITHIN numeric type)
            if !sort.is_empty() {
                let sort_type = &sort[0].expression.resolved_type;
                let within_type = &typed_within.resolved_type;

                // TIMESTAMP - TIMESTAMP yields INTERVAL or CALENDAR_INTERVAL
                // For other types, subtraction yields the same type (INT - INT = INT)
                let types_compatible = if **sort_type == TIMESTAMP {
                    **within_type == INTERVAL || **within_type == CALENDAR_INTERVAL
                } else {
                    sort_type == within_type
                };

                if !types_compatible {
                    return Err(ctx
                        .error(&format!(
                            "MATCH with WITHIN: SORT field type {:?} is incompatible with WITHIN type {:?}",
                            sort_type, within_type
                        ))
                        .emit());
                }
            }

            Some(typed_within)
        } else {
            None
        };

        // Check for conflicts across agg + group_by assignments
        let valid = agg.check_conflicts(ctx, &[&group_by]);

        // Build output schema for ONE ROW PER MATCH semantics:
        // Order: AGG fields → BY fields → base pattern fields
        // (AGG and BY are prepended to the base pattern fields)
        let mut agg_by_fields = Struct::default();
        for assignment in agg.assignments.iter().chain(group_by.assignments.iter()) {
            if let Ok(identifier) = assignment.identifier.clone().valid() {
                if valid.contains(&identifier) {
                    agg_by_fields
                        .insert_arc(identifier, assignment.expression.resolved_type.clone());
                }
            }
        }
        let output_fields = merged_struct.prepend_overwrite(&agg_by_fields);

        let output_schema = Arc::new(output_fields.into());

        Ok((
            Self {
                patterns,
                agg,
                group_by,
                sort,
                within,
            },
            output_schema,
        ))
    }

    /// Check if an expression represents a negative interval (e.g., -5m)
    /// Check if an expression is a negative interval literal.
    ///
    /// This performs semantic AST analysis to detect patterns like `-5m`.
    ///
    /// **Limitation**: This only detects literal negative intervals (e.g., `-5m` as unary minus
    /// applied to an interval literal). It does NOT detect complex expressions that evaluate to
    /// negative values (e.g., `5m - 10m`), as that would require constant folding and expression
    /// evaluation at compile time.
    ///
    /// The typed AST path has richer semantic information than the old AST path, allowing us to
    /// distinguish between unary minus (`-5m`) and binary minus (`5m - 1m`).
    fn is_negative_interval_expression(expr: &Expression) -> bool {
        match &expr.kind {
            ExpressionKind::UnaryPrefixOperator(UnaryPrefixOperator {
                operator: UnaryPrefixOp::Minus,
                operand,
            }) => {
                // Check if the operand is an interval literal or expression that evaluates to an interval
                match &operand.kind {
                    ExpressionKind::IntervalLiteral(_) => true,
                    // Could also check for other interval-producing expressions here if needed
                    _ => false,
                }
            }
            _ => false,
        }
    }
}

// ============================================================================
// APPEND Command - DML (Data Modification)
// ============================================================================

/// APPEND command - writes pipeline data to a table
///
/// This is a DML (Data Modification Language) command that writes the current
/// pipeline data to a target table. It supports optional deduplication via
/// DISTINCT BY clause.
///
/// Schema behavior:
/// - Preserves input schema (data flows through unchanged until write)
/// - Pipeline classified as DML (not query)
/// - No output data returned (write operation)
///
/// Example:
/// ```hamelin
/// FROM events
/// | WHERE severity > 3
/// | APPEND incidents DISTINCT BY event_id
/// ```
#[derive(Debug, Clone)]
pub struct TypedAppendCommand {
    /// Target table reference (resolved from catalog)
    pub table: TypedTableReference,

    /// Optional DISTINCT BY selections for deduplication
    pub distinct_by: Vec<TypedSelection>,
}

impl TypedAppendCommand {
    fn from_ast_with_context(
        cmd: &AppendCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Resolve the target table from the catalog
        let table = TypedTableReference::try_from_ast_with_context(cmd.table.clone(), ctx)?;

        // Merge input schema with table schema to compute output schema
        // This represents "what the table will look like after the append"
        let output_schema = match &table.resolved {
            ResolvedEnvironment::Resolved(table_env) => {
                // Output schema is the target table's schema
                // (the pipeline data is expanded/aligned to match the table at SQL generation time)
                table_env.clone()
            }
            ResolvedEnvironment::Error(_) => {
                // Table resolution failed. Use input schema as best-effort output.
                input_schema.clone()
            }
        };

        // Type-check DISTINCT BY selections against the input schema (the rows being appended),
        // not the merged output schema. DISTINCT BY references fields in the pipeline output.
        let distinct_by = cmd
            .distinct_by
            .iter()
            .map(|selection| {
                TypedSelection::from_ast_with_context(selection.clone(), input_schema, ctx)
            })
            .collect();

        Ok((Self { table, distinct_by }, output_schema))
    }
}

// ============================================================================
// EXPLODE Command - Array Flattening
// ============================================================================

/// A single explode item within an EXPLODE command.
///
/// Contains the assignment (identifier = array_expression) and the pre-extracted
/// element type for efficient access without re-checking the array type.
#[derive(Debug, Clone)]
pub struct TypedExplodeItem {
    /// The assignment (identifier = array_expression)
    pub assignment: TypedAssignment,

    /// The element type extracted from the array at type-check time
    pub element_type: Arc<Type>,
}

/// EXPLODE command - explodes array fields into multiple rows
///
/// This command takes array fields and creates one output row per array element.
/// When multiple arrays are exploded together, elements at the same index are
/// expanded together (zipped), with shorter arrays padded with NULL.
///
/// Schema behavior:
/// - Preserves all fields except the exploded fields
/// - Replaces array fields with scalar element types
/// - Each array element becomes a separate row
///
/// Type checking:
/// - Each expression must resolve to an ARRAY type
/// - Type checking error if non-array type is provided
///
/// Example (single field):
/// ```hamelin
/// FROM events | EXPLODE tags = tags
/// -- Input:  {id: 1, tags: ["malware", "ransomware"]}
/// -- Output: {id: 1, tags: "malware"}
/// --         {id: 1, tags: "ransomware"}
/// ```
///
/// Example (multi-field, internal use only):
/// ```hamelin
/// | EXPLODE elem = arr, idx = indices
/// -- Arrays are zipped together at each index position
/// ```
#[derive(Debug, Clone)]
pub struct TypedExplodeCommand {
    /// The items to explode. Usually a single item from parsing,
    /// but can be multiple for internal lowering passes (e.g., transform lowering).
    pub items: Vec<TypedExplodeItem>,
}

impl TypedExplodeCommand {
    fn from_ast_with_context(
        cmd: &ExplodeCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        if cmd.clauses.is_empty() {
            return Err(ctx.error("EXPLODE command has no clauses").emit());
        }

        // First pass: type-check all expressions
        let typed_expressions: Vec<_> = {
            let mut expr_ctx = ctx.default_expression_context(input_schema);
            cmd.clauses
                .iter()
                .map(|clause| match clause.as_ref() {
                    AssignmentClause::Assignment(assignment) => {
                        let typed_expr = TypedExpression::from_ast_with_expected_type(
                            assignment.expression.clone(),
                            &mut expr_ctx,
                            &ArrayMatcher::of(AnyMatcher),
                        );
                        Ok((
                            assignment.identifier.clone(),
                            typed_expr,
                            &assignment.expression,
                        ))
                    }
                    AssignmentClause::Error(err) => Err(err.clone()),
                })
                .collect::<Result<Vec<_>, _>>()?
        };

        // Second pass: build items, extract element types, update schema
        let mut items = Vec::with_capacity(cmd.clauses.len());
        let mut output_schema = input_schema.clone();
        let mut new_assignments: Vec<(Identifier, Arc<Type>)> = Vec::new();

        for (parsed_identifier, typed_expression, ast_expression) in typed_expressions {
            // Extract the element type from the array type
            let element_type = match typed_expression.resolved_type.as_ref() {
                Type::Array(array) => array.element_type.clone(),
                _ => {
                    if let TypedExpressionKind::Error(err_expr) = &typed_expression.kind {
                        return Err(err_expr.error.clone());
                    }
                    return Err(ctx
                        .error("EXPLODE requires an ARRAY type expression")
                        .at(ast_expression.as_ref())
                        .emit());
                }
            };

            if let Ok(identifier) = parsed_identifier.clone().valid() {
                if output_schema.lookup(&identifier).is_some() {
                    // Existing field: replace type in-place (preserves position)
                    output_schema = Arc::new(
                        output_schema
                            .replace(&identifier, element_type.as_ref().clone())
                            .map_err(|e| {
                                ctx.error(&format!("Failed to build EXPLODE output schema: {}", e))
                                    .at(ast_expression.as_ref())
                                    .emit()
                            })?,
                    );
                } else {
                    // New field: collect for from_let (prepends to front)
                    new_assignments.push((identifier, element_type.clone()));
                }
            }

            items.push(TypedExplodeItem {
                assignment: TypedAssignment {
                    identifier: parsed_identifier,
                    expression: Arc::new(typed_expression),
                },
                element_type,
            });
        }

        if !new_assignments.is_empty() {
            output_schema = Arc::new(TypeEnvironment::extend(&new_assignments, &output_schema));
        }

        Ok((Self { items }, output_schema))
    }
}

// ============================================================================
// UNNEST Command - Struct/Array Flattening
// ============================================================================

/// UNNEST command - unnests nested struct or array-of-struct fields into columns
///
/// This command flattens a nested structure by promoting its fields to the top level.
/// For STRUCT types, it extracts all nested fields.
/// For ARRAY<STRUCT> types, it creates one row per array element and extracts fields.
///
/// Schema behavior:
/// - Removes the nested field
/// - Adds all nested fields at the top level
/// - Errors if field name conflicts occur
///
/// Type checking:
/// - Expression must resolve to STRUCT or ARRAY<STRUCT>
/// - Type checking error for other types
/// - Schema merge error if flattened fields conflict with existing fields
///
/// Example:
/// ```hamelin
/// FROM events | UNNEST user
/// -- Input:  {id: 1, user: {name: "Alice", email: "alice@example.com"}}
/// -- Output: {id: 1, name: "Alice", email: "alice@example.com"}
/// ```
#[derive(Debug, Clone)]
pub struct TypedUnnestCommand {
    /// The struct/array expression to unnest (type-checked to be STRUCT or ARRAY<STRUCT>)
    pub expression: Arc<TypedExpression>,
}

impl TypedUnnestCommand {
    fn from_expression(
        expression: Arc<Expression>,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Create expression context for type-checking
        let mut expr_ctx = ctx.default_expression_context(input_schema);

        // Type-check the expression
        // We accept either STRUCT or ARRAY<STRUCT>
        let typed_expression = TypedExpression::from_ast_with_context(expression, &mut expr_ctx);

        // Extract the struct type
        let struct_type = match typed_expression.resolved_type.as_ref() {
            Type::Struct(s) => s.clone(),
            Type::Array(array) => {
                // Check if array element is a struct
                match array.element_type.as_ref() {
                    Type::Struct(s) => s.clone(),
                    t => {
                        return Err(ctx
                            .error(format!(
                                "UNNEST requires a STRUCT or ARRAY<STRUCT> type expression. Found: {}",
                                t
                            ))
                            .at(typed_expression.ast.as_ref())
                            .emit());
                    }
                }
            }
            t => {
                return Err(ctx
                    .error(format!(
                        "UNNEST requires a STRUCT or ARRAY<STRUCT> type expression. Found: {}",
                        t
                    ))
                    .at(typed_expression.ast.as_ref())
                    .emit());
            }
        };

        // Prepend unnested struct fields to input
        let output_schema = Arc::new(TypeEnvironment::new(
            input_schema.as_struct().prepend_overwrite(&struct_type),
        ));

        Ok((
            Self {
                expression: Arc::new(typed_expression),
            },
            output_schema,
        ))
    }

    fn from_ast_with_context_unnest(
        cmd: &UnnestCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        Self::from_expression(cmd.expression.clone(), input_schema, ctx)
    }

    fn from_ast_with_context_rows(
        cmd: &RowsCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        Self::from_expression(cmd.expression.clone(), input_schema, ctx)
    }
}

// ============================================================================
// NEST Command - Wraps All Fields into Nested Struct
// ============================================================================

/// NEST command - wraps all input fields into a single nested struct field
///
/// Example:
/// ```hamelin
/// FROM events | NEST user
/// -- Input:  {id: 1, name: "Alice", email: "alice@example.com"}
/// -- Output: {user: {id: 1, name: "Alice", email: "alice@example.com"}}
/// ```
///
/// Multi-level nesting:
/// ```hamelin
/// FROM events | NEST user.address
/// -- Input:  {street: "Main St", city: "NYC"}
/// -- Output: {user: {address: {street: "Main St", city: "NYC"}}}
/// ```
///
/// Key semantics:
/// - Output schema has ONLY the nested struct field (complete replacement)
/// - No name conflicts: if target field name matches an input field, it just gets nested
/// - Empty schema error: cannot nest an empty schema
/// - Supports compound identifiers for multi-level nesting
#[derive(Debug, Clone)]
pub struct TypedNestCommand {
    /// The identifier for the nested struct field (can be compound for multi-level)
    pub identifier: ParsedIdentifier,

    /// The collected fields that will be nested
    pub nested_type: Arc<Struct>,
}

impl TypedNestCommand {
    fn from_ast_with_context(
        cmd: &NestCommand,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Arc<TypeEnvironment>), Arc<TranslationError>> {
        // Collect all fields from input schema (including base chain)
        let flattened_input = Self::flatten_environment(input_schema);

        // Validate that input schema is not empty
        if flattened_input.is_empty() {
            return Err(ctx
                .error("NEST command requires a non-empty input schema")
                .at(cmd.identifier.as_ref())
                .emit());
        }

        // Build the nested structure based on the identifier (only for valid identifiers)
        let (output_struct, nested_type) =
            if let Ok(identifier) = cmd.identifier.as_ref().clone().valid() {
                // NEST user → {user: STRUCT(all input fields)}
                // NEST user.address.location → {user: {address: {location: STRUCT(all input fields)}}}
                let nested_struct = flattened_input.clone();
                let output_struct =
                    Self::build_nested_struct(identifier.segments(), nested_struct.clone());
                (output_struct, nested_struct)
            } else {
                // Parse error in identifier - create empty output (error already emitted)
                (Struct::default(), flattened_input)
            };

        // Output schema: completely replace with the nested structure (no base)
        let output_schema = Arc::new(output_struct.into());

        // Store ParsedIdentifier as-is (including parse errors)
        Ok((
            Self {
                identifier: cmd.identifier.as_ref().clone(),
                nested_type: Arc::new(nested_type),
            },
            output_schema,
        ))
    }

    /// Get the struct representation of an environment
    fn flatten_environment(env: &Arc<TypeEnvironment>) -> Struct {
        env.as_struct().clone()
    }

    /// Build a nested struct from a list of identifiers
    /// Example: [user, address, location] + {street, city}
    /// Result: {user: {address: {location: {street, city}}}}
    fn build_nested_struct(parts: &[SimpleIdentifier], innermost: Struct) -> Struct {
        if parts.is_empty() {
            return innermost;
        }

        // Build from the inside out
        let mut current = innermost;
        for part in parts.iter().rev() {
            let mut outer = Struct::default();
            outer.insert(part.clone().into(), Type::Struct(current));
            current = outer;
        }

        current
    }
}
