use std::rc::Rc;

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

use crate::{
    err::TranslationError,
    func::def::{FunctionTranslationContext, SpecialPosition},
    sql::expression::identifier::Identifier as SqlIdentifier,
    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::{ColumnReference, Expression, ExpressionKind, 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::{
        Projections, ResolvedEnvironment, 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: Rc<Command>,

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

    /// Schema available as output from this command
    pub output_schema: Rc<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: Rc<Command>,
        input_schema: Rc<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 `Rc<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) -> Rc<Command> {
        Rc::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(Rc<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: &Rc<Command>,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> (Self, Rc<TypeEnvironment>) {
        let result: Result<(TypedCommandKind, Rc<TypeEnvironment>), Rc<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.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| {
                        Rc::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| {
                        Rc::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| {
                        Rc::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: Rc::new(FromClause::TableAlias(cmd.right.ast.clone())),
                on_condition: cmd.condition.as_ref().map(|c| c.cata(alg)),
            }
            .into(),

            TypedCommandKind::Lookup(cmd) => LookupCommand {
                other: Rc::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| {
                        Rc::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| {
                        Rc::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| {
                        Rc::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| {
                        Rc::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| {
                        Rc::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| {
                        Rc::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| {
                        Rc::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| {
                        Rc::new(AssignmentClause::Assignment(Assignment {
                            span: a.span().into(),
                            identifier: a.identifier.clone(),
                            expression: a.expression.cata(alg),
                        }))
                    })
                    .collect(),
                sort: cmd
                    .sort
                    .iter()
                    .map(|s| {
                        Rc::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) => {
                // Compute span covering identifier and expression (like TypedAssignment::span)
                let span: Span = match (cmd.identifier.span(), cmd.expression.ast.span()) {
                    (Some(id_span), Some(expr_span)) => Some(*id_span.start()..=*expr_span.end()),
                    (None, Some(expr_span)) => Some(expr_span),
                    (Some(id_span), None) => Some(id_span),
                    (None, None) => None,
                }
                .into();

                ExplodeCommand {
                    clause: Rc::new(AssignmentClause::Assignment(Assignment {
                        span,
                        identifier: cmd.identifier.clone(),
                        expression: cmd.expression.cata(alg),
                    })),
                }
                .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: Rc<TypedExpression>,
}

impl TypedWhereCommand {
    fn from_ast_with_context(
        cmd: &WhereCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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: Rc::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: Rc<TypedExpression>,
}

impl TypedWithinCommand {
    fn from_ast_with_context(
        cmd: &WithinCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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: Rc::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: Rc<TypedExpression>,
}

impl TypedLimitCommand {
    fn from_ast_with_context(
        cmd: &LimitCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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: Rc::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: Rc<SortExpression>,

    /// The expression to sort by
    pub expression: Rc<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: Rc<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: Rc::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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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);

        // Build output environment from projections
        // Note: Desugaring of struct bindings now happens in Projections layer
        let mut output_env = TypeEnvironment::default();

        // Bind all the (expanded) child assignments
        for typed_assignment in &projections.assignments {
            // Only bind valid identifiers to the environment
            if let Ok(identifier) = typed_assignment.identifier.clone().valid() {
                if let Err(e) =
                    output_env.bind_computed(identifier, typed_assignment.expression.clone())
                {
                    return Err(ctx
                        .error("conflict in field definitions")
                        .at(typed_assignment)
                        .with_source(e)
                        .emit());
                }
            }
            // Skip parse errors - they don't contribute to the environment
        }

        let output_schema = Rc::new(output_env.with_single_parent(input_schema.clone()));

        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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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,
            false, // SELECT replaces schema, doesn't extend
        );

        // Replace schema with only the selected fields (no base)
        // Build environment from projections, checking for conflicts
        let mut output_env = TypeEnvironment::default();
        for typed_assignment in &projections.assignments {
            // Only bind valid identifiers to the environment
            if let Ok(identifier) = typed_assignment.identifier.clone().valid() {
                if let Err(e) =
                    output_env.bind_computed(identifier, typed_assignment.expression.clone())
                {
                    // Conflict detected - emit error
                    return Err(ctx
                        .error("conflict in field definitions")
                        .at(typed_assignment)
                        .with_source(e)
                        .emit());
                }
            }
            // Skip parse errors - they don't contribute to the environment
        }
        let output_schema = Rc::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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<TranslationError>> {
        use std::collections::HashMap;

        let mut dropped_fields = Vec::new();

        // Group fields by parent struct (for batched desugaring)
        let mut groups: HashMap<Option<Identifier>, Vec<Identifier>> = HashMap::new();

        for selection in &cmd.selections {
            // Extract valid identifier from ParsedIdentifier
            // Skip if it's a parse error (already emitted during parsing)
            let identifier = match selection.identifier.clone().valid() {
                Ok(id) => id,
                Err(_) => continue, // Skip parse errors, don't re-emit
            };

            // Check if field has a parent struct binding
            match input_schema.lookup_struct(&identifier) {
                Ok(Some((parent, _))) => {
                    // Non-clean parent - needs desugaring
                    groups.entry(Some(parent)).or_default().push(identifier);
                }
                Ok(None) => {
                    // Clean parent - can tombstone directly
                    groups.entry(None).or_default().push(identifier);
                }
                Err(e) => {
                    // Error (e.g., accessing field on non-struct)
                    ctx.error("Cannot drop field")
                        .at(selection.as_ref())
                        .with_source(e)
                        .emit();
                }
            }
        }

        // Build output environment from input (will be extended with desugared fields)
        let mut output_env = TypeEnvironment::default();

        // Process each group
        for (parent_opt, fields) in groups {
            match parent_opt {
                Some(parent) => {
                    // Desugar: expand all fields EXCEPT dropped ones
                    let parent_struct = match input_schema.lookup(&parent) {
                        Some(typ) => match typ.as_ref() {
                            Type::Struct(s) => s.clone(),
                            _ => {
                                // Should not happen (lookup_struct already validated)
                                continue;
                            }
                        },
                        None => continue,
                    };

                    // Bind all fields EXCEPT dropped ones
                    let mut any_field_bound = false;
                    for (field_id, _field_type) in parent_struct.iter_flattened() {
                        // Convert SQL identifier to AST identifier and prepend parent
                        let field_id: Identifier = field_id.into();
                        let full_id = parent.clone() + field_id;

                        // Skip if this field is being dropped
                        if fields.contains(&full_id) {
                            dropped_fields.push(full_id);
                            continue;
                        }

                        // Skip if this field is a child of a dropped field
                        if fields.iter().any(|dropped| full_id.has_prefix(dropped)) {
                            continue;
                        }

                        // Preserve field with column reference
                        let col_ref_expr =
                            Projections::create_column_reference_expression(&full_id);
                        let typed_expr = Rc::new(TypedExpression::from_ast_with_context(
                            col_ref_expr.into(),
                            &mut ctx.default_expression_context(input_schema),
                        ));

                        if let Err(e) = output_env.bind_computed(full_id, typed_expr) {
                            return Err(ctx
                                .error("conflict in field definitions")
                                .with_source(e)
                                .at(&parent)
                                .emit());
                        }
                        any_field_bound = true;
                    }
                    // If ALL fields were dropped, we need to tombstone the parent
                    // (binding children creates passthrough that shadows, but if no
                    // children are bound, we need an explicit tombstone)
                    if !any_field_bound {
                        if let Err(e) = output_env.drop(&parent) {
                            return Err(ctx
                                .error("failed to tombstone parent struct")
                                .with_source(e)
                                .at(&parent)
                                .emit());
                        }
                    }
                }
                None => {
                    // Clean fields - tombstone directly
                    for field in fields {
                        if input_schema.lookup(&field).is_some() {
                            if let Err(e) = output_env.drop(&field) {
                                ctx.error("conflict in field definitions")
                                    .at(&field)
                                    .with_source(e)
                                    .emit();
                            } else {
                                dropped_fields.push(field);
                            }
                        } else {
                            // Field not found - emit error
                            ctx.error(format!("Field '{:?}' not found", field))
                                .at(&field)
                                .emit();
                        }
                    }
                }
            }
        }

        // Create output schema with input as base chain
        let output_schema = Rc::new(output_env.with_single_parent(input_schema.clone()));

        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<Rc<TypedExpression>>,

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

    /// The pattern AST expression (for reconstructing untyped AST)
    pub pattern_ast: Rc<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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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| {
            Rc::new(TypedExpression::from_ast_with_expected_type(
                src.clone(),
                &mut expr_ctx,
                &ExactMatcher::of(STRING),
            ))
        });

        // Build environment with new string fields from PARSE extraction
        // PARSE outputs have no computable lineage, so we use Base bindings
        let mut env = TypeEnvironment::default();

        for parsed_identifier in &cmd.identifiers {
            // Only bind valid identifiers to environment, but keep all identifiers in TypedParseCommand
            if let Ok(identifier) = parsed_identifier.clone().valid() {
                // Check if this is a throwaway identifier (underscore)
                let is_throwaway = match &identifier {
                    Identifier::Simple(simple) => simple == "_",
                    _ => false,
                };

                if !is_throwaway {
                    // All PARSE-extracted fields are STRING type
                    if let Err(e) = env.bind_base(identifier.clone(), STRING) {
                        return Err(ctx
                            .error("Failed to bind PARSE field")
                            .with_source(e)
                            .at(cmd.pattern.as_ref())
                            .emit());
                    }
                }
            }
            // Parse errors are preserved in cmd.identifiers, just not bound to environment
        }

        // Add input_schema as base chain
        let output_schema = Rc::new(env.with_single_parent(input_schema.clone()));

        // 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, Rc<TypeEnvironment>), Rc<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 - flatten the environment to Struct
                    env.flatten()
                }
                TypedFromClause::Alias(alias_clause) => {
                    // Flatten base schema
                    let base_struct = env.flatten();

                    // 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 SQL identifier for struct field (extract valid identifier)
                    if let Ok(alias) = alias_clause.alias.clone().valid() {
                        let sql_alias = SqlIdentifier::from(alias);
                        // Add prefixed bindings by converting to nested struct
                        // This will create t1.x, t1.y, etc. when flattened
                        aliased_struct = aliased_struct.with(sql_alias, 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.to_hamelin(), 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 = Rc::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, Rc<TypeEnvironment>), Rc<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;
            }

            // Flatten to Struct for merging
            let struct_to_merge = env.flatten();

            // 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.to_hamelin(), 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 = Rc::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<Rc<TypedExpression>>,
}

impl TypedJoinCommand {
    /// Type-check a JOIN command
    fn from_ast_with_context(
        cmd: &JoinCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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<Rc<TypedExpression>>,
}

impl TypedLookupCommand {
    /// Type-check a LOOKUP command
    fn from_ast_with_context(
        cmd: &LookupCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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: &Rc<FromClause>,
    on_condition: Option<&Rc<Expression>>,
    input_schema: &Rc<TypeEnvironment>,
    ctx: &mut StatementTranslationContext,
) -> Result<
    (
        TypedTableAlias,
        Option<Rc<TypedExpression>>,
        Rc<TypeEnvironment>,
    ),
    Rc<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 Identifier to SQL identifier for merging
    let sql_alias = match right_table_alias.alias.clone().valid() {
        Ok(alias) => SqlIdentifier::from(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());
        }
    };

    // Create merged environment: nested right side fields under alias
    // Flatten right environment and nest under alias
    let right_struct = right_env.flatten();
    let nested_struct = Struct::default().with(sql_alias, right_struct.into());

    // Create new TypeEnvironment with nested struct, using input_schema as base
    let output_schema =
        Rc::new(TypeEnvironment::from(nested_struct).with_single_parent(input_schema.clone()));

    // 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),
        );

        Rc::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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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,
            false, // AGG replaces schema, doesn't extend
        );

        // 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,
            false, // GROUP BY doesn't extend either
        );

        // 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()
        };

        // Build output schema: GROUP BY fields + aggregate fields
        // Bind directly to preserve lineage and check for conflicts
        let mut output_env = TypeEnvironment::default();

        // First bind all group_by fields (only valid identifiers)
        for assignment in group_by.assignments.iter() {
            if let Ok(identifier) = assignment.identifier.clone().valid() {
                if let Err(e) = output_env.bind_computed(identifier, assignment.expression.clone())
                {
                    return Err(ctx
                        .error("conflict in field definitions")
                        .with_source(e)
                        .at(assignment)
                        .emit());
                }
            }
        }

        // Then bind all aggregate fields (will detect conflicts)
        for assignment in aggregates.assignments.iter() {
            if let Ok(identifier) = assignment.identifier.clone().valid() {
                if let Err(e) = output_env.bind_computed(identifier, assignment.expression.clone())
                {
                    return Err(ctx
                        .error("conflict in field definitions")
                        .with_source(e)
                        .at(assignment)
                        .emit());
                }
            }
        }

        let output_schema = Rc::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<Rc<TypedExpression>>,
}

impl TypedWindowCommand {
    fn from_ast_with_context(
        cmd: &WindowCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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,
            true, // WINDOW extends schema with new fields
        );

        // Type-check order-by expressions FIRST to determine sort type for frame validation
        let (order_by, sort_type) = {
            let mut order_by_expr_ctx = ctx.default_expression_context(input_schema);
            let 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_else(|| {
                    // No explicit ORDER BY - default to the timestamp field
                    // Try to look up the timestamp field type from the input schema
                    input_schema
                        .lookup(&ctx.timestamp_field)
                        .map(|rc_type| rc_type.as_ref().clone())
                });

            (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 = Projections::from_assignment_clauses_with_context(
                &cmd.group_by,
                &mut partition_by_expr_ctx,
                false, // PARTITION BY is grouping, doesn't extend
            );

            // 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(Rc::new(frame_expr))
                }
                (Some(within_expr), Some(base_type)) => {
                    // Other base types can use that type, ROWS, or RANGE of that type
                    let frame_matcher = OrMatcher::default()
                        .with(ExactMatcher::of(base_type.clone()))
                        .with(ExactMatcher::of(ROWS))
                        .with(RangeMatcher::of(ExactMatcher::of(base_type.clone())));
                    let frame_expr = TypedExpression::from_ast_with_expected_type(
                        within_expr.clone(),
                        &mut partition_by_expr_ctx,
                        &frame_matcher,
                    );
                    Some(Rc::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(Rc::new(TypedExpression {
                        ast: within_expr.clone(),
                        resolved_type: Rc::new(UNKNOWN),
                        kind: TypedErrorExpression { error }.into(),
                    }))
                }
                (None, _) => None,
            };

            (partition_by, frame)
        };

        // Build output schema: input schema + window function fields
        // Build environment from assignment and partition projections, checking for conflicts
        let mut output_env = TypeEnvironment::default();

        // Bind all the (expanded) child assignments (only valid identifiers)
        for typed_assignment in &projections.assignments {
            if let Ok(identifier) = typed_assignment.identifier.clone().valid() {
                if let Err(e) =
                    output_env.bind_computed(identifier, typed_assignment.expression.clone())
                {
                    // Conflict detected - emit error
                    return Err(ctx
                        .error("conflict in field definitions")
                        .at(typed_assignment)
                        .with_source(e)
                        .emit());
                }
            }
        }

        // Bind all of the partition-by projections
        for typed_assignment in &partition_by.assignments {
            if let Ok(identifier) = typed_assignment.identifier.clone().valid() {
                if let Err(e) =
                    output_env.bind_computed(identifier, typed_assignment.expression.clone())
                {
                    return Err(ctx
                        .error("conflict in field definitions")
                        .at(typed_assignment)
                        .with_source(e)
                        .emit());
                }
            }
        }

        let output_schema = Rc::new(output_env.with_single_parent(input_schema.clone()));

        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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<TranslationError>> {
        // Type-check all patterns
        let patterns = cmd
            .pattern
            .iter()
            .map(|p| TypedPattern::from_ast_with_context(p.clone(), ctx))
            .collect::<Vec<_>>();

        // 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.fields.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_hamelin()).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.fields.contains_key(&name) {
                    merged_struct = merged_struct.with(name.clone().into(), typ);
                }
            }
        }

        // Create environment from merged_struct with input schema as parent
        let pattern_env = Rc::new(
            TypeEnvironment::from(merged_struct.clone()).with_single_parent(input_schema.clone()),
        );

        // 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,
            false, // AGG doesn't extend
        );

        // 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,
            false, // GROUP BY doesn't extend
        );

        // 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 column 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 column_ref = ColumnReference {
                            column_name: timestamp_simple.into(),
                        };
                        let ast_expr = Rc::new(Expression {
                            span: Span::NONE,
                            kind: column_ref.into(),
                        });
                        let typed_expr =
                            TypedExpression::from_ast_with_context(ast_expr.clone(), &mut sort_ctx);
                        let sort_expr = Rc::new(SortExpression {
                            span: Span::NONE,
                            expression: ast_expr,
                            order: None,
                        });

                        vec![TypedSortExpression {
                            ast: sort_expr,
                            expression: Rc::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 column
            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 column
                    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 column type {:?} is incompatible with WITHIN type {:?}",
                            sort_type, within_type
                        ))
                        .emit());
                }
            }

            Some(typed_within)
        } else {
            None
        };

        // Build output schema for ONE ROW PER MATCH semantics:
        // - All input fields from ALL patterns (passed through from triggering event)
        // - Pattern variables themselves (e, A, B as struct columns)
        // - GROUP BY fields (if any)
        // - AGG fields (aggregations over matched rows)
        //
        // merged_struct already contains both flattened fields and pattern variables
        let mut output_fields = merged_struct;

        // Add GROUP BY fields to output
        for (ident, typ) in group_by.types().fields.into_iter() {
            output_fields.insert(ident.into(), typ);
        }

        // Add AGG fields to output
        for (ident, typ) in agg.types().fields.into_iter() {
            output_fields.insert(ident.into(), typ);
        }

        let output_schema = Rc::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: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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) => {
                // Merge: table schema can expand to include new fields from pipeline
                let table_struct = table_env.flatten();
                let input_struct = input_schema.flatten();
                match table_struct.merge(&input_struct) {
                    Ok(merged_struct) => Rc::new(merged_struct.into()),
                    Err(e) => {
                        return Err(ctx
                            .error(format!("Cannot append to table: {}", e))
                            .at(cmd.table.as_ref())
                            .emit());
                    }
                }
            }
            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 columns 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
// ============================================================================

/// EXPLODE command - explodes array fields into multiple rows
///
/// This command takes an array field and creates one output row per array element.
/// The exploded field's type becomes the element type of the array.
///
/// Schema behavior:
/// - Preserves all fields except the exploded field
/// - Replaces array field with scalar element type
/// - Each array element becomes a separate row
///
/// Type checking:
/// - The expression must resolve to an ARRAY type
/// - Type checking error if non-array type is provided
///
/// Example:
/// ```hamelin
/// FROM events | EXPLODE tags = tags
/// -- Input:  {id: 1, tags: ["malware", "ransomware"]}
/// -- Output: {id: 1, tags: "malware"}
/// --         {id: 1, tags: "ransomware"}
/// ```
#[derive(Debug, Clone)]
pub struct TypedExplodeCommand {
    /// The field identifier being exploded (left side of assignment)
    pub identifier: ParsedIdentifier,

    /// The array expression to explode (right side of assignment, type-checked to be ARRAY)
    pub expression: Rc<TypedExpression>,

    /// The element type extracted from the array
    pub element_type: Rc<Type>,
}

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

        // Extract identifier and expression from the AssignmentClause
        let (parsed_identifier, ast_expression) = match cmd.clause.as_ref() {
            AssignmentClause::Assignment(assignment) => {
                (assignment.identifier.clone(), &assignment.expression)
            }
            AssignmentClause::Error(err) => {
                // Assignment clause has an error - propagate it
                return Err(err.clone());
            }
        };

        // Type-check the expression with ARRAY type requirement
        // We use ArrayMatcher with AnyMatcher to accept any element type
        let typed_expression = TypedExpression::from_ast_with_expected_type(
            ast_expression.clone(),
            &mut expr_ctx,
            &ArrayMatcher::of(AnyMatcher),
        );

        // Extract the element type from the array type
        let element_type = match typed_expression.resolved_type.as_ref() {
            Type::Array(array) => Rc::from(array.element_type.as_ref().clone()),
            _ => {
                // Type mismatch - expression is not an array
                // The error was already emitted during type checking, extract and return it
                if let TypedExpressionKind::Error(err_expr) = &typed_expression.kind {
                    return Err(err_expr.error.clone());
                }
                // This shouldn't happen (type mismatch should always produce an Error variant)
                // but handle it defensively
                return Err(ctx
                    .error("EXPLODE requires an ARRAY type expression")
                    .at(ast_expression.as_ref())
                    .emit());
            }
        };

        // Build output schema: only bind valid identifiers to environment
        let output_fields = if let Ok(identifier) = parsed_identifier.clone().valid() {
            let sql_ident = SqlIdentifier::from(identifier.clone());
            Struct::default().with(sql_ident, element_type.as_ref().clone())
        } else {
            // Parse error in identifier - create empty output (error already emitted)
            Struct::default()
        };

        let output_schema =
            Rc::new(TypeEnvironment::from(output_fields).with_single_parent(input_schema.clone()));

        // Store ParsedIdentifier as-is (including parse errors)
        Ok((
            Self {
                identifier: parsed_identifier,
                expression: Rc::new(typed_expression),
                element_type,
            },
            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: Rc<TypedExpression>,
}

impl TypedUnnestCommand {
    fn from_expression(
        expression: Rc<Expression>,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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());
            }
        };

        let output_schema = Rc::new(
            TypeEnvironment::from(struct_type.clone()).with_single_parent(input_schema.clone()),
        );

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

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

    fn from_ast_with_context_rows(
        cmd: &RowsCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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: Rc<Struct>,
}

impl TypedNestCommand {
    fn from_ast_with_context(
        cmd: &NestCommand,
        input_schema: &Rc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<(Self, Rc<TypeEnvironment>), Rc<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.fields.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() {
                match identifier {
                    Identifier::Simple(simple) => {
                        // Simple case: NEST user
                        // Output: {user: STRUCT(all input fields)}
                        let nested_struct = flattened_input.clone();
                        let mut output_fields = Struct::default();
                        output_fields
                            .insert(simple.clone().into(), Type::Struct(nested_struct.clone()));
                        (output_fields, nested_struct)
                    }
                    Identifier::Compound(compound) => {
                        // Multi-level case: NEST user.address.location
                        // Output: {user: {address: {location: STRUCT(all input fields)}}}
                        let nested_struct = flattened_input.clone();
                        let output_struct =
                            Self::build_nested_struct(&compound.parts, 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 = Rc::new(output_struct.into());

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

    /// Flatten an environment into a single Struct by recursively merging with base
    fn flatten_environment(env: &Rc<TypeEnvironment>) -> Struct {
        env.flatten()
    }

    /// 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
    }
}
