hamelin_lib 0.19.0

//! Command builder system for constructing Command AST nodes
//!
//! This module provides a fluent API for building command trees programmatically.
//! Builders follow the same patterns as expression builders but for commands.

use std::sync::Arc;

use crate::tree::ast::clause::{
    Assignment, AssignmentClause, FromClause, Selection, SortExpression, SortOrder, TableAlias,
    TableReference,
};
use crate::tree::ast::command::{
    AggCommand, AppendCommand, Command, DistinctCommand, DropCommand, ExplodeCommand, FromCommand,
    JoinCommand, LimitCommand, LookupCommand, MatchCommand, NestCommand, ParseCommand, RowsCommand,
    SelectCommand, SetCommand, SortCommand, SuppressCommand, UnionCommand, UnnestCommand,
    WhereCommand, WindowCommand, WithinCommand,
};
use crate::tree::ast::dataset_identifier::{DatasetIdentifier, UnqualifiedDatasetIdentifier};
use crate::tree::ast::expression::{Expression, FieldReference};
use crate::tree::ast::identifier::{
    Identifier, ParsedIdentifier, ParsedSimpleIdentifier, SimpleIdentifier,
};
use crate::tree::ast::node::Span;
use crate::tree::ast::pattern::{Pattern, PatternKind, QuantifiedPattern, Quantifier};

use super::IntoExpressionBuilder;

// ============================================================================
// TableReference convenience functions
// ============================================================================

/// Create a TableReference from a dataset path string or [`DatasetIdentifier`].
///
/// Invalid paths produce a [`TableReference`] whose target embeds the parse error.
pub fn table_ref(dataset: impl AsRef<str>) -> TableReference {
    match DatasetIdentifier::from_ref_str(dataset.as_ref()) {
        Ok(parsed) => table_ref_parsed(parsed),
        Err(e) => TableReference {
            span: Span::NONE,
            target: Arc::new(e).into(),
        },
    }
}

pub fn table_ref_parsed(dataset: DatasetIdentifier) -> TableReference {
    TableReference {
        span: Span::NONE,
        target: dataset.into(),
    }
}

impl From<DatasetIdentifier> for TableReference {
    fn from(dataset: DatasetIdentifier) -> Self {
        table_ref_parsed(dataset)
    }
}

impl From<&str> for TableReference {
    fn from(s: &str) -> Self {
        table_ref(s)
    }
}

impl From<String> for TableReference {
    fn from(s: String) -> Self {
        table_ref(s.as_str())
    }
}

impl From<SimpleIdentifier> for TableReference {
    fn from(id: SimpleIdentifier) -> Self {
        table_ref_parsed(DatasetIdentifier::Unqualified(
            UnqualifiedDatasetIdentifier {
                namespace: Vec::new(),
                table: id,
            },
        ))
    }
}

// ============================================================================
// Pattern convenience functions
// ============================================================================

/// Create a quantified pattern from a table reference (e.g., for MATCH commands)
///
/// Examples:
/// - `pattern_quantified(table_ref("events"), None)` - Simple pattern: `events`
/// - `pattern_quantified(table_ref("events"), Some(quantifier_at_least_one()))` - Pattern: `events+`
pub fn pattern_quantified(
    from_clause: impl Into<FromClause>,
    quantifier: Option<Quantifier>,
) -> Pattern {
    let kind: PatternKind = QuantifiedPattern {
        span: Span::NONE,
        from_clause: Arc::new(from_clause.into()),
        quantifier: quantifier.map(Arc::new),
    }
    .into();
    Pattern {
        span: Span::NONE,
        kind,
    }
}

/// Create an "at least one" quantifier (+)
pub fn quantifier_at_least_one() -> Quantifier {
    crate::tree::ast::pattern::QuantifierKind::AtLeastOne.into()
}

/// Create an "any number" quantifier (*)
pub fn quantifier_any_number() -> Quantifier {
    crate::tree::ast::pattern::QuantifierKind::AnyNumber.into()
}

/// Create a "zero or one" quantifier (?)
pub fn quantifier_zero_or_one() -> Quantifier {
    crate::tree::ast::pattern::QuantifierKind::ZeroOrOne.into()
}

/// Create a pattern with a table alias (e.g., e=events)
pub fn pattern_with_alias(
    alias: impl Into<SimpleIdentifier>,
    table: impl Into<TableReference>,
) -> Pattern {
    pattern_quantified(table_alias_from_clause(alias, table), None)
}

/// Create a pattern with a table alias and quantifier (e.g., e=events+)
pub fn pattern_with_alias_and_quantifier(
    alias: impl Into<SimpleIdentifier>,
    table: impl Into<TableReference>,
    quantifier: Quantifier,
) -> Pattern {
    pattern_quantified(table_alias_from_clause(alias, table), Some(quantifier))
}

/// Helper to create a TableAlias FromClause
fn table_alias_from_clause(
    alias: impl Into<SimpleIdentifier>,
    table: impl Into<TableReference>,
) -> FromClause {
    let alias_simple: SimpleIdentifier = alias.into();
    FromClause::TableAlias(Arc::new(TableAlias {
        span: Span::NONE,
        alias: alias_simple.into(),
        table: table.into(),
    }))
}

/// Create an "exactly n" quantifier ({n})
pub fn quantifier_exactly(n: impl Into<String>) -> Quantifier {
    crate::tree::ast::pattern::QuantifierKind::Exactly(n.into()).into()
}

// Implement Into<Pattern> for common types
impl From<TableReference> for Pattern {
    fn from(table: TableReference) -> Self {
        pattern_quantified(table, None)
    }
}

impl From<&str> for Pattern {
    fn from(s: &str) -> Self {
        pattern_quantified(table_ref(s), None)
    }
}

impl From<String> for Pattern {
    fn from(s: String) -> Self {
        pattern_quantified(table_ref(s), None)
    }
}

/// Builder for SET command
#[derive(Clone)]
pub struct SetCommandBuilder {
    assignments: Vec<Assignment>,
    span: Span,
}

impl SetCommandBuilder {
    pub fn new() -> Self {
        Self {
            assignments: Vec::new(),
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    /// Add a named field to the SET command
    pub fn named_field(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.assignments.push(Assignment {
            span: Span::NONE,
            identifier: ParsedIdentifier::Valid(identifier.into()),
            expression: Arc::new(expr),
        });
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: SetCommand {
                assignments: self.assignments.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for SetCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for WHERE command
#[derive(Clone)]
pub struct WhereCommandBuilder {
    condition: Expression,
    span: Span,
}

impl WhereCommandBuilder {
    pub fn new(condition: impl IntoExpressionBuilder) -> Self {
        Self {
            condition: condition.into_expression_builder().build(),
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: WhereCommand {
                condition: Arc::new(self.condition.clone()),
            }
            .into(),
        }
    }
}

/// Builder for SELECT command
#[derive(Clone)]
pub struct SelectCommandBuilder {
    clauses: Vec<AssignmentClause>,
    span: Span,
}

impl SelectCommandBuilder {
    pub fn new() -> Self {
        Self {
            clauses: Vec::new(),
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    /// Add a named field (identifier = expression)
    pub fn named_field(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Add a field reference (identifier inferred from field name)
    pub fn field(mut self, column: impl Into<FieldReference>) -> Self {
        let col_ref = column.into();
        let identifier = col_ref.field_name.clone().into();
        let expr = Expression::from_kind(col_ref);
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier,
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: SelectCommand {
                clauses: self.clauses.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for SelectCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for DROP command
#[derive(Clone)]
pub struct DropCommandBuilder {
    selections: Vec<Selection>,
    span: Span,
}

impl DropCommandBuilder {
    pub fn new() -> Self {
        Self {
            selections: Vec::new(),
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    /// Add a field to drop
    pub fn field(mut self, identifier: impl Into<Identifier>) -> Self {
        self.selections.push(Selection {
            span: Span::NONE,
            identifier: ParsedIdentifier::Valid(identifier.into()),
        });
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: DropCommand {
                selections: self.selections.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for DropCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for FROM command
#[derive(Clone)]
pub struct FromCommandBuilder {
    clauses: Vec<FromClause>,
    span: Span,
}

impl FromCommandBuilder {
    pub fn new() -> Self {
        Self {
            clauses: Vec::new(),
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    /// Add a table alias (name = table_reference)
    pub fn table_alias(
        mut self,
        alias: impl Into<SimpleIdentifier>,
        table_reference: impl Into<TableReference>,
    ) -> Self {
        self.clauses.push(
            TableAlias {
                span: Span::NONE,
                alias: ParsedSimpleIdentifier::Valid(alias.into()),
                table: table_reference.into(),
            }
            .into(),
        );
        self
    }

    /// Add a direct table reference from a qualified dataset identifier.
    pub fn dataset_reference(mut self, dataset: DatasetIdentifier) -> Self {
        self.clauses.push(table_ref_parsed(dataset).into());
        self
    }

    /// Add a direct table reference (string path, [`DatasetIdentifier`], or CTE name).
    pub fn table_reference(mut self, table_reference: impl Into<TableReference>) -> Self {
        self.clauses.push(table_reference.into().into());
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: FromCommand {
                clauses: self.clauses.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for FromCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for UNION command
#[derive(Clone)]
pub struct UnionCommandBuilder {
    clauses: Vec<FromClause>,
    span: Span,
}

impl UnionCommandBuilder {
    pub fn new() -> Self {
        Self {
            clauses: Vec::new(),
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    /// Add a table alias
    pub fn table_alias(
        mut self,
        alias: impl Into<String>,
        table_reference: impl Into<TableReference>,
    ) -> Self {
        self.clauses.push(
            TableAlias {
                span: Span::NONE,
                alias: ParsedSimpleIdentifier::Valid(SimpleIdentifier::new(alias)),
                table: table_reference.into(),
            }
            .into(),
        );
        self
    }

    /// Add a direct table reference from a qualified dataset identifier.
    pub fn dataset_reference(mut self, dataset: DatasetIdentifier) -> Self {
        self.clauses.push(table_ref_parsed(dataset).into());
        self
    }

    /// Add a direct table reference (string path, [`DatasetIdentifier`], or CTE name).
    pub fn table_reference(mut self, table_reference: impl Into<TableReference>) -> Self {
        self.clauses.push(table_reference.into().into());
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: UnionCommand {
                clauses: self.clauses.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for UnionCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for LIMIT command
#[derive(Clone)]
pub struct LimitCommandBuilder {
    count: Expression,
}

impl LimitCommandBuilder {
    pub fn new(count: impl IntoExpressionBuilder) -> Self {
        Self {
            count: count.into_expression_builder().build(),
        }
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: LimitCommand {
                count: Arc::new(self.count.clone()),
            }
            .into(),
        }
    }
}

/// Builder for WITHIN command
#[derive(Clone)]
pub struct WithinCommandBuilder {
    duration: Expression,
}

impl WithinCommandBuilder {
    pub fn new(duration: impl IntoExpressionBuilder) -> Self {
        Self {
            duration: duration.into_expression_builder().build(),
        }
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: WithinCommand {
                duration: Arc::new(self.duration.clone()),
            }
            .into(),
        }
    }
}

/// Builder for SORT command
#[derive(Clone)]
pub struct SortCommandBuilder {
    pub(crate) expressions: Vec<SortExpression>,
}

impl SortCommandBuilder {
    pub fn new() -> Self {
        Self {
            expressions: Vec::new(),
        }
    }

    /// Add a sort expression with no explicit order (database default)
    pub fn by(mut self, expression: impl IntoExpressionBuilder) -> Self {
        self.expressions.push(SortExpression {
            span: Span::NONE,
            expression: Arc::new(expression.into_expression_builder().build()),
            order: None,
        });
        self
    }

    /// Add an ascending sort expression
    pub fn asc(mut self, expression: impl IntoExpressionBuilder) -> Self {
        self.expressions.push(SortExpression {
            span: Span::NONE,
            expression: Arc::new(expression.into_expression_builder().build()),
            order: Some(SortOrder::Asc),
        });
        self
    }

    /// Add a descending sort expression
    pub fn desc(mut self, expression: impl IntoExpressionBuilder) -> Self {
        self.expressions.push(SortExpression {
            span: Span::NONE,
            expression: Arc::new(expression.into_expression_builder().build()),
            order: Some(SortOrder::Desc),
        });
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: SortCommand {
                expressions: self.expressions.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for SortCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for PARSE command (initial state, no pattern set yet)
#[derive(Clone)]
pub struct ParseCommandBuilder {
    source: Option<Expression>,
    nodrop: bool,
}

impl ParseCommandBuilder {
    pub fn new() -> Self {
        Self {
            source: None,
            nodrop: false,
        }
    }

    /// Set the pattern to parse (transitions to ParseCommandBuilderWithPattern)
    pub fn pattern(self, pattern: impl IntoExpressionBuilder) -> ParseCommandBuilderWithPattern {
        ParseCommandBuilderWithPattern {
            source: self.source,
            pattern: pattern.into_expression_builder().build(),
            identifiers: Vec::new(),
            nodrop: self.nodrop,
        }
    }
}

impl Default for ParseCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Parse command builder after pattern is set but before identifiers
#[derive(Clone)]
pub struct ParseCommandBuilderWithPattern {
    source: Option<Expression>,
    pattern: Expression,
    identifiers: Vec<ParsedIdentifier>,
    nodrop: bool,
}

impl ParseCommandBuilderWithPattern {
    /// Add an identifier to extract (transitions to ValidParseCommandBuilder)
    pub fn identifier(mut self, identifier: impl Into<Identifier>) -> ValidParseCommandBuilder {
        self.identifiers
            .push(ParsedIdentifier::Valid(identifier.into()));
        ValidParseCommandBuilder {
            source: self.source,
            pattern: self.pattern,
            identifiers: self.identifiers,
            nodrop: self.nodrop,
        }
    }
}

/// A fully-configured ParseCommandBuilder ready to build.
///
/// Pattern and at least one identifier are guaranteed present.
#[derive(Clone)]
pub struct ValidParseCommandBuilder {
    source: Option<Expression>,
    pattern: Expression,
    identifiers: Vec<ParsedIdentifier>,
    nodrop: bool,
}

impl ValidParseCommandBuilder {
    /// Add another identifier to extract
    pub fn identifier(mut self, identifier: impl Into<Identifier>) -> Self {
        self.identifiers
            .push(ParsedIdentifier::Valid(identifier.into()));
        self
    }

    /// Set the source expression
    pub fn source(mut self, source: impl IntoExpressionBuilder) -> Self {
        self.source = Some(source.into_expression_builder().build());
        self
    }

    /// Set nodrop flag
    pub fn nodrop(mut self, nodrop: bool) -> Self {
        self.nodrop = nodrop;
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: ParseCommand {
                source: self.source.clone().map(Arc::new),
                pattern: Arc::new(self.pattern.clone()),
                identifiers: self.identifiers.clone(),
                nodrop: self.nodrop,
            }
            .into(),
        }
    }
}

/// Builder for AGG command
#[derive(Clone)]
pub struct AggCommandBuilder {
    clauses: Vec<AssignmentClause>,
    group_by: Vec<AssignmentClause>,
    sort: Vec<SortExpression>,
    span: Span,
}

impl AggCommandBuilder {
    pub fn new() -> Self {
        Self {
            clauses: Vec::new(),
            group_by: Vec::new(),
            sort: Vec::new(),
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    /// Add a named aggregation field
    pub fn named_aggregate(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Add a named group by field
    pub fn named_group(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.group_by.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Add a field reference to group by (identifier inferred from field name)
    pub fn group_by(mut self, column: impl Into<FieldReference>) -> Self {
        let col_ref = column.into();
        let identifier = col_ref.field_name.clone().into();
        let expr = Expression::from_kind(col_ref);
        self.group_by.push(
            Assignment {
                span: Span::NONE,
                identifier,
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Set sort expressions using a SortCommandBuilder
    pub fn sort(mut self, sort_builder: SortCommandBuilder) -> Self {
        self.sort = sort_builder.expressions;
        self
    }

    /// Add a single sort expression
    pub fn sort_expr(mut self, sort_expr: SortExpression) -> Self {
        self.sort.push(sort_expr);
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: AggCommand {
                clauses: self.clauses.iter().cloned().map(Arc::new).collect(),
                group_by: self.group_by.iter().cloned().map(Arc::new).collect(),
                sort: self.sort.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for AggCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for DISTINCT command
#[derive(Clone)]
pub struct DistinctCommandBuilder {
    clauses: Vec<AssignmentClause>,
    span: Span,
}

impl DistinctCommandBuilder {
    pub fn new() -> Self {
        Self {
            clauses: Vec::new(),
            span: Span::NONE,
        }
    }

    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    pub fn named_group(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    pub fn group_by(mut self, column: impl Into<FieldReference>) -> Self {
        let col_ref = column.into();
        let identifier = col_ref.field_name.clone().into();
        let expr = Expression::from_kind(col_ref);
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier,
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: DistinctCommand {
                clauses: self.clauses.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

impl Default for DistinctCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for SUPPRESS command
#[derive(Clone)]
pub struct SuppressCommandBuilder {
    interval: Expression,
    clauses: Vec<AssignmentClause>,
    span: Span,
}

impl SuppressCommandBuilder {
    pub fn new(interval: impl IntoExpressionBuilder) -> Self {
        Self {
            interval: interval.into_expression_builder().build(),
            clauses: Vec::new(),
            span: Span::NONE,
        }
    }

    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    pub fn named_group(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    pub fn group_by(mut self, column: impl Into<FieldReference>) -> Self {
        let col_ref = column.into();
        let identifier = col_ref.field_name.clone().into();
        let expr = Expression::from_kind(col_ref);
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier,
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: SuppressCommand {
                interval: Arc::new(self.interval.clone()),
                group_by: self.clauses.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

/// Builder for WINDOW command
#[derive(Clone)]
pub struct WindowCommandBuilder {
    clauses: Vec<AssignmentClause>,
    group_by: Vec<AssignmentClause>,
    sort: Vec<SortExpression>,
    within: Option<Expression>,
    span: Span,
}

impl WindowCommandBuilder {
    pub fn new() -> Self {
        Self {
            clauses: Vec::new(),
            group_by: Vec::new(),
            sort: Vec::new(),
            within: None,
            span: Span::NONE,
        }
    }

    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }

    /// Add a named window field
    pub fn named_field(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.clauses.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Add a named group by field
    pub fn group_by(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.group_by.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Set sort expressions using a SortCommandBuilder
    pub fn sort(mut self, sort_builder: SortCommandBuilder) -> Self {
        self.sort = sort_builder.expressions;
        self
    }

    /// Add a single sort expression
    pub fn sort_expr(mut self, sort_expr: SortExpression) -> Self {
        self.sort.push(sort_expr);
        self
    }

    /// Set within duration
    pub fn within(mut self, duration: impl IntoExpressionBuilder) -> Self {
        self.within = Some(duration.into_expression_builder().build());
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: WindowCommand {
                clauses: self.clauses.iter().cloned().map(Arc::new).collect(),
                group_by: self.group_by.iter().cloned().map(Arc::new).collect(),
                sort: self.sort.iter().cloned().map(Arc::new).collect(),
                within: self.within.clone().map(Arc::new),
            }
            .into(),
        }
    }
}

impl Default for WindowCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Builder for APPEND command
#[derive(Clone)]
pub struct AppendCommandBuilder {
    table: TableReference,
    distinct_by: Vec<Selection>,
}

impl AppendCommandBuilder {
    pub fn new(table: impl Into<TableReference>) -> Self {
        Self {
            table: table.into(),
            distinct_by: Vec::new(),
        }
    }

    /// Add a distinct_by identifier
    pub fn distinct_by(mut self, identifier: impl Into<Identifier>) -> Self {
        self.distinct_by.push(Selection {
            span: Span::NONE,
            identifier: ParsedIdentifier::Valid(identifier.into()),
        });
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: AppendCommand {
                table: Arc::new(self.table.clone()),
                distinct_by: self.distinct_by.iter().cloned().map(Arc::new).collect(),
            }
            .into(),
        }
    }
}

/// Builder for JOIN command (call `.on()` before `build()`)
#[derive(Clone)]
pub struct JoinCommandBuilder {
    other: FromClause,
}

/// JOIN command builder with a required ON condition
#[derive(Clone)]
pub struct JoinCommandBuilderWithOn {
    other: FromClause,
    on: Expression,
}

impl JoinCommandBuilder {
    pub fn table_alias(
        alias: impl Into<SimpleIdentifier>,
        table_reference: impl Into<TableReference>,
    ) -> Self {
        Self {
            other: TableAlias {
                span: Span::NONE,
                alias: ParsedSimpleIdentifier::Valid(alias.into()),
                table: table_reference.into(),
            }
            .into(),
        }
    }

    pub fn table_reference(table_reference: impl Into<TableReference>) -> Self {
        Self {
            other: table_reference.into().into(),
        }
    }

    pub fn dataset_reference(dataset: DatasetIdentifier) -> Self {
        Self {
            other: table_ref_parsed(dataset).into(),
        }
    }

    pub fn table_alias_dataset(
        alias: impl Into<SimpleIdentifier>,
        dataset: DatasetIdentifier,
    ) -> Self {
        Self {
            other: TableAlias {
                span: Span::NONE,
                alias: ParsedSimpleIdentifier::Valid(alias.into()),
                table: table_ref_parsed(dataset),
            }
            .into(),
        }
    }

    pub fn on(self, condition: impl IntoExpressionBuilder) -> JoinCommandBuilderWithOn {
        JoinCommandBuilderWithOn {
            other: self.other,
            on: condition.into_expression_builder().build(),
        }
    }
}

impl JoinCommandBuilderWithOn {
    pub fn build(self) -> Command {
        Command {
            span: Span::NONE,
            kind: JoinCommand {
                other: Arc::new(self.other),
                on_condition: Arc::new(self.on),
            }
            .into(),
        }
    }
}

/// Builder for LOOKUP command (call `.on()` before `build()`)
#[derive(Clone)]
pub struct LookupCommandBuilder {
    other: FromClause,
}

/// LOOKUP command builder with a required ON condition
#[derive(Clone)]
pub struct LookupCommandBuilderWithOn {
    other: FromClause,
    on: Expression,
}

impl LookupCommandBuilder {
    pub fn table_alias(
        alias: impl Into<SimpleIdentifier>,
        table_reference: impl Into<TableReference>,
    ) -> Self {
        Self {
            other: TableAlias {
                span: Span::NONE,
                alias: ParsedSimpleIdentifier::Valid(alias.into()),
                table: table_reference.into(),
            }
            .into(),
        }
    }

    pub fn table_reference(table_reference: impl Into<TableReference>) -> Self {
        Self {
            other: table_reference.into().into(),
        }
    }

    pub fn dataset_reference(dataset: DatasetIdentifier) -> Self {
        Self {
            other: table_ref_parsed(dataset).into(),
        }
    }

    pub fn on(self, condition: impl IntoExpressionBuilder) -> LookupCommandBuilderWithOn {
        LookupCommandBuilderWithOn {
            other: self.other,
            on: condition.into_expression_builder().build(),
        }
    }
}

impl LookupCommandBuilderWithOn {
    pub fn build(self) -> Command {
        Command {
            span: Span::NONE,
            kind: LookupCommand {
                other: Arc::new(self.other),
                on_condition: Arc::new(self.on),
            }
            .into(),
        }
    }
}

// Typestate markers for ExplodeCommandBuilder
pub struct NoClause;
pub struct WithClause;

/// Builder for EXPLODE command with compile-time safety
///
/// Uses typestate pattern to ensure at least one clause is added.
/// Supports multiple clauses for multi-field EXPLODE (internal use).
#[derive(Clone)]
pub struct ExplodeCommandBuilder<State = NoClause> {
    clauses: Vec<Arc<AssignmentClause>>,
    span: Span,
    _state: std::marker::PhantomData<State>,
}

impl ExplodeCommandBuilder<NoClause> {
    pub fn new() -> Self {
        Self {
            clauses: Vec::new(),
            span: Span::NONE,
            _state: std::marker::PhantomData,
        }
    }

    /// Set an assignment to explode (identifier = expression)
    pub fn named_field(
        self,
        identifier: impl Into<Identifier>,
        expr: impl IntoExpressionBuilder,
    ) -> ExplodeCommandBuilder<WithClause> {
        let expr = expr.into_expression_builder().build();
        let clause = Arc::new(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        ExplodeCommandBuilder {
            clauses: vec![clause],
            span: self.span,
            _state: std::marker::PhantomData,
        }
    }

    /// Set a field reference to explode (identifier inferred from field name)
    pub fn field(self, column: impl Into<FieldReference>) -> ExplodeCommandBuilder<WithClause> {
        let col_ref = column.into();
        let identifier = col_ref.field_name.clone().into();
        let expr = Expression::from_kind(col_ref);
        let clause = Arc::new(
            Assignment {
                span: Span::NONE,
                identifier,
                expression: Arc::new(expr),
            }
            .into(),
        );
        ExplodeCommandBuilder {
            clauses: vec![clause],
            span: self.span,
            _state: std::marker::PhantomData,
        }
    }
}

impl<State> ExplodeCommandBuilder<State> {
    /// Set the span for this command
    pub fn at(mut self, span: Span) -> Self {
        self.span = span;
        self
    }
}

impl Default for ExplodeCommandBuilder<NoClause> {
    fn default() -> Self {
        Self::new()
    }
}

impl ExplodeCommandBuilder<WithClause> {
    /// Add another assignment to explode (for multi-field EXPLODE)
    pub fn named_field(
        mut self,
        identifier: impl Into<Identifier>,
        expr: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expr.into_expression_builder().build();
        self.clauses.push(Arc::new(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        ));
        self
    }

    /// Add another field reference to explode (for multi-field EXPLODE)
    pub fn field(mut self, column: impl Into<FieldReference>) -> Self {
        let col_ref = column.into();
        let identifier = col_ref.field_name.clone().into();
        let expr = Expression::from_kind(col_ref);
        self.clauses.push(Arc::new(
            Assignment {
                span: Span::NONE,
                identifier,
                expression: Arc::new(expr),
            }
            .into(),
        ));
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: self.span,
            kind: ExplodeCommand {
                clauses: self.clauses.clone(),
            }
            .into(),
        }
    }
}

// Only implement From<Command> for the WithClause state
impl From<ExplodeCommandBuilder<WithClause>> for Command {
    fn from(value: ExplodeCommandBuilder<WithClause>) -> Self {
        value.build()
    }
}

/// Builder for UNNEST command
#[derive(Clone)]
pub struct UnnestCommandBuilder {
    expression: Expression,
}

impl UnnestCommandBuilder {
    pub fn new(expression: impl IntoExpressionBuilder) -> Self {
        Self {
            expression: expression.into_expression_builder().build(),
        }
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: UnnestCommand {
                expression: Arc::new(self.expression.clone()),
            }
            .into(),
        }
    }
}

/// Builder for ROWS command
#[derive(Clone)]
pub struct RowsCommandBuilder {
    expression: Expression,
}

impl RowsCommandBuilder {
    pub fn new(expression: impl IntoExpressionBuilder) -> Self {
        Self {
            expression: expression.into_expression_builder().build(),
        }
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: RowsCommand {
                expression: Arc::new(self.expression.clone()),
            }
            .into(),
        }
    }
}

/// Builder for NEST command
#[derive(Clone)]
pub struct NestCommandBuilder {
    identifier: ParsedIdentifier,
}

impl NestCommandBuilder {
    pub fn new(identifier: impl Into<Identifier>) -> Self {
        Self {
            identifier: ParsedIdentifier::Valid(identifier.into()),
        }
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: NestCommand {
                identifier: Arc::new(self.identifier.clone()),
            }
            .into(),
        }
    }
}

/// Builder for MATCH command
#[derive(Clone)]
pub struct MatchCommandBuilder {
    pattern: Vec<Pattern>,
    agg: Vec<AssignmentClause>,
    group_by: Vec<AssignmentClause>,
    sort: Vec<SortExpression>,
    within: Option<Expression>,
}

impl MatchCommandBuilder {
    pub fn new() -> Self {
        Self {
            pattern: Vec::new(),
            agg: Vec::new(),
            group_by: Vec::new(),
            sort: Vec::new(),
            within: None,
        }
    }

    /// Add a pattern
    pub fn pattern(mut self, pattern: impl Into<Pattern>) -> Self {
        self.pattern.push(pattern.into());
        self
    }

    /// Add a pattern with a table alias (e.g., e=events)
    pub fn named_pattern(
        mut self,
        alias: impl Into<SimpleIdentifier>,
        table: impl Into<TableReference>,
    ) -> Self {
        self.pattern.push(pattern_with_alias(alias, table));
        self
    }

    /// Add a pattern with a table alias and quantifier (e.g., e=events+)
    pub fn named_quantified_pattern(
        mut self,
        alias: impl Into<SimpleIdentifier>,
        table: impl Into<TableReference>,
        quantifier: Quantifier,
    ) -> Self {
        self.pattern
            .push(pattern_with_alias_and_quantifier(alias, table, quantifier));
        self
    }

    /// Add an agg clause (aggregation over matched rows)
    pub fn agg(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.agg.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Add a named group by field
    pub fn named_group(
        mut self,
        identifier: impl Into<Identifier>,
        expression: impl IntoExpressionBuilder,
    ) -> Self {
        let expr = expression.into_expression_builder().build();
        self.group_by.push(
            Assignment {
                span: Span::NONE,
                identifier: ParsedIdentifier::Valid(identifier.into()),
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Add a field reference to group by (identifier inferred from field name)
    pub fn group_by(mut self, column: impl Into<FieldReference>) -> Self {
        let col_ref = column.into();
        let identifier = col_ref.field_name.clone().into();
        let expr = Expression::from_kind(col_ref);
        self.group_by.push(
            Assignment {
                span: Span::NONE,
                identifier,
                expression: Arc::new(expr),
            }
            .into(),
        );
        self
    }

    /// Set sort expressions using a SortCommandBuilder
    pub fn sort(mut self, sort_builder: SortCommandBuilder) -> Self {
        self.sort = sort_builder.expressions;
        self
    }

    /// Set the within expression
    pub fn within(mut self, expr: impl IntoExpressionBuilder) -> Self {
        self.within = Some(expr.into_expression_builder().build());
        self
    }

    pub fn build(&self) -> Command {
        Command {
            span: Span::NONE,
            kind: MatchCommand {
                pattern: self
                    .pattern
                    .iter()
                    .cloned()
                    .map(|pattern| Arc::new(pattern))
                    .collect(),
                agg: self.agg.iter().cloned().map(|s| Arc::new(s)).collect(),
                group_by: self
                    .group_by
                    .iter()
                    .cloned()
                    .map(|group_by| Arc::new(group_by))
                    .collect(),
                sort: self
                    .sort
                    .iter()
                    .map(|sort| Arc::new(sort.clone()))
                    .collect(),
                within: self.within.clone().map(Arc::new),
            }
            .into(),
        }
    }
}

impl Default for MatchCommandBuilder {
    fn default() -> Self {
        Self::new()
    }
}

// Convenience functions for creating command builders

/// Create a SET command builder
pub fn set_command() -> SetCommandBuilder {
    SetCommandBuilder::new()
}

/// Create a WHERE command builder
pub fn where_command(condition: impl IntoExpressionBuilder) -> WhereCommandBuilder {
    WhereCommandBuilder::new(condition)
}

/// Create a SELECT command builder
pub fn select_command() -> SelectCommandBuilder {
    SelectCommandBuilder::new()
}

/// Create a DROP command builder
pub fn drop_command() -> DropCommandBuilder {
    DropCommandBuilder::new()
}

/// Create a FROM command builder
pub fn from_command() -> FromCommandBuilder {
    FromCommandBuilder::new()
}

/// Create a UNION command builder
pub fn union_command() -> UnionCommandBuilder {
    UnionCommandBuilder::new()
}

/// Create a LIMIT command builder
pub fn limit_command(count: impl IntoExpressionBuilder) -> LimitCommandBuilder {
    LimitCommandBuilder::new(count)
}

/// Create a WITHIN command builder
pub fn within_command(duration: impl IntoExpressionBuilder) -> WithinCommandBuilder {
    WithinCommandBuilder::new(duration)
}

/// Create a SORT command builder
pub fn sort_command() -> SortCommandBuilder {
    SortCommandBuilder::new()
}

/// Create a PARSE command builder
pub fn parse_command() -> ParseCommandBuilder {
    ParseCommandBuilder::new()
}

/// Create an AGG command builder
pub fn agg_command() -> AggCommandBuilder {
    AggCommandBuilder::new()
}

/// Create a DISTINCT command builder
pub fn distinct_command() -> DistinctCommandBuilder {
    DistinctCommandBuilder::new()
}

/// Create a SUPPRESS command builder (`interval` is required)
pub fn suppress_command(interval: impl IntoExpressionBuilder) -> SuppressCommandBuilder {
    SuppressCommandBuilder::new(interval)
}

/// Create a WINDOW command builder
pub fn window_command() -> WindowCommandBuilder {
    WindowCommandBuilder::new()
}

/// Create an APPEND command builder
pub fn append_command(table: impl Into<TableReference>) -> AppendCommandBuilder {
    AppendCommandBuilder::new(table)
}

/// Create a JOIN command builder with table alias
pub fn join_table_alias(
    alias: impl Into<SimpleIdentifier>,
    table_reference: impl Into<TableReference>,
) -> JoinCommandBuilder {
    JoinCommandBuilder::table_alias(alias, table_reference)
}

/// Create a JOIN command builder with table reference
pub fn join_table_reference(table_reference: impl Into<TableReference>) -> JoinCommandBuilder {
    JoinCommandBuilder::table_reference(table_reference)
}

/// Create a JOIN command builder with a space-qualified dataset reference
pub fn join_dataset_reference(dataset: DatasetIdentifier) -> JoinCommandBuilder {
    JoinCommandBuilder::dataset_reference(dataset)
}

/// Create a LOOKUP command builder with table alias
pub fn lookup_table_alias(
    alias: impl Into<SimpleIdentifier>,
    table_reference: impl Into<TableReference>,
) -> LookupCommandBuilder {
    LookupCommandBuilder::table_alias(alias, table_reference)
}

/// Create a LOOKUP command builder with table reference
pub fn lookup_table_reference(table_reference: impl Into<TableReference>) -> LookupCommandBuilder {
    LookupCommandBuilder::table_reference(table_reference)
}

/// Create a LOOKUP command builder with a space-qualified dataset reference
pub fn lookup_dataset_reference(dataset: DatasetIdentifier) -> LookupCommandBuilder {
    LookupCommandBuilder::dataset_reference(dataset)
}

/// Create an EXPLODE command builder
pub fn explode_command() -> ExplodeCommandBuilder<NoClause> {
    ExplodeCommandBuilder::new()
}

/// Create an UNNEST command builder
pub fn unnest_command(expression: impl IntoExpressionBuilder) -> UnnestCommandBuilder {
    UnnestCommandBuilder::new(expression)
}

/// Create a ROWS command builder
pub fn rows_command(expression: impl IntoExpressionBuilder) -> RowsCommandBuilder {
    RowsCommandBuilder::new(expression)
}

/// Create a NEST command builder
pub fn nest_command(identifier: impl Into<Identifier>) -> NestCommandBuilder {
    NestCommandBuilder::new(identifier)
}

/// Create a MATCH command builder
pub fn match_command() -> MatchCommandBuilder {
    MatchCommandBuilder::new()
}