powdb_query/

parser.rs

use crate::ast::*;
use crate::lexer::lex;
use crate::token::Token;

/// Maximum nesting depth for the AST the parser produces.
///
/// Recursive descent (parentheses, subqueries, CASE) bumps `Parser::depth` on
/// the way down, so the limit falls out of the call stack there. Binary,
/// arithmetic and `having` chains are parsed ITERATIVELY, and a loop that
/// re-wraps its accumulator (`left = BinaryOp(left, op, right)`) grows the AST
/// one level per iteration without ever touching the call stack. Those loops
/// therefore count their own iterations against this same limit (see
/// [`Parser::check_chain_depth`]): what has to stay bounded is the SHAPE OF THE
/// PRODUCED TREE, not just parser recursion. Every later walk of that tree
/// (planner, canonicalizer, executor, and the recursive `Drop` of the boxed
/// expression itself) is recursive, so an unbounded chain overflows the stack
/// well after the parse returned, aborting the process under `panic = "abort"`.
const MAX_NESTING_DEPTH: usize = 64;

/// Discriminated parse error; callers can match on category.
///
/// Display strings are wire-visible behavior (the server's egress
/// sanitization prefix-matches them). Every message is pinned byte-exact by
/// `tests/error_display.rs`; do not reword one without updating that suite
/// deliberately.
#[derive(Debug, thiserror::Error)]
pub enum ParseError {
    /// Lexer failed to tokenize the input.
    #[error("at position {position}: {message}")]
    Lex { message: String, position: usize },
    /// Expected one token but found another.
    #[error("expected {expected}, got {got}")]
    UnexpectedToken { expected: String, got: String },
    /// Recursive nesting exceeded the safety limit.
    #[error("query nesting depth exceeds maximum of {max}")]
    NestingDepthExceeded { max: usize },
    /// Syntactically valid construct that the engine doesn't support yet.
    #[error("{feature}")]
    Unsupported { feature: String },
    /// Catch-all for other syntax errors.
    #[error("{message}")]
    Syntax { message: String },
}

impl ParseError {
    /// Convenience: human-readable message for any variant.
    pub fn message(&self) -> String {
        self.to_string()
    }
}

fn token_to_scalar_fn(tok: &Token) -> ScalarFn {
    match tok {
        Token::Upper => ScalarFn::Upper,
        Token::Lower => ScalarFn::Lower,
        Token::Length => ScalarFn::Length,
        Token::Trim => ScalarFn::Trim,
        Token::Substring => ScalarFn::Substring,
        Token::Concat => ScalarFn::Concat,
        Token::Abs => ScalarFn::Abs,
        Token::Round => ScalarFn::Round,
        Token::Ceil => ScalarFn::Ceil,
        Token::Floor => ScalarFn::Floor,
        Token::Sqrt => ScalarFn::Sqrt,
        Token::Pow => ScalarFn::Pow,
        Token::Now => ScalarFn::Now,
        Token::Extract => ScalarFn::Extract,
        Token::DateAdd => ScalarFn::DateAdd,
        Token::DateDiff => ScalarFn::DateDiff,
        Token::JsonType => ScalarFn::JsonType,
        Token::JsonText => ScalarFn::JsonText,
        _ => unreachable!(),
    }
}

struct Parser {
    tokens: Vec<Token>,
    pos: usize,
    depth: usize,
}

/// Parse a PowQL query string into an AST [`Statement`].
///
/// # Examples
///
/// ```
/// use powdb_query::parser::parse;
/// use powdb_query::ast::Statement;
///
/// // A bare type name is a query (select all rows).
/// let stmt = parse("User").unwrap();
/// assert!(matches!(stmt, Statement::Query(_)));
/// ```
///
/// ```
/// use powdb_query::parser::parse;
/// use powdb_query::ast::Statement;
///
/// // DDL: define a new type (table).
/// let stmt = parse("type User { required name: str, age: int }").unwrap();
/// assert!(matches!(stmt, Statement::CreateType(_)));
/// ```
pub fn parse(input: &str) -> Result<Statement, ParseError> {
    let tokens = lex(input).map_err(|e| ParseError::Lex {
        message: e.message,
        position: e.position,
    })?;
    parse_tokens(tokens)
}

/// Parse PowQL with `$N` placeholders bound to positional `params`.
///
/// Binding happens at the **token level**: the input is lexed, each
/// `$N` placeholder token is replaced in place with the literal token
/// for `params[N-1]` (a string param becomes a `StringLit` byte-for-byte,
/// `null` becomes `Token::Null`), and the resulting token stream is parsed
/// normally. Values are never re-lexed or string-interpolated, so an
/// injection-shaped string is inert data — it can never change the query's
/// shape.
///
/// Placeholders are 1-based (`$1`, `$2`, …). A reference to a placeholder
/// with no corresponding parameter is a clean [`ParseError::Syntax`], as is
/// a non-numeric `$name` (the named-parameter form belongs to the in-process
/// prepared API, not the positional wire-binding path).
pub fn parse_with_params(input: &str, params: &[ParamValue]) -> Result<Statement, ParseError> {
    let mut tokens = lex(input).map_err(|e| ParseError::Lex {
        message: e.message,
        position: e.position,
    })?;
    for tok in tokens.iter_mut() {
        if let Token::Param(name) = tok {
            let n: usize = name.parse().map_err(|_| ParseError::Syntax {
                message: format!(
                    "positional parameters must be numeric (`$1`, `$2`, …); got `${name}`"
                ),
            })?;
            if n == 0 {
                return Err(ParseError::Syntax {
                    message: "parameter placeholders are 1-based; `$0` is invalid".into(),
                });
            }
            let p = params.get(n - 1).ok_or_else(|| ParseError::Syntax {
                message: format!(
                    "query references ${n} but only {} parameter(s) were supplied",
                    params.len()
                ),
            })?;
            *tok = match p {
                ParamValue::Null => Token::Null,
                ParamValue::Int(v) => Token::IntLit(*v),
                ParamValue::Float(v) => Token::FloatLit(*v),
                ParamValue::Bool(v) => Token::BoolLit(*v),
                ParamValue::Str(s) => Token::StringLit(s.clone()),
            };
        }
    }
    parse_tokens(tokens)
}

fn edit_distance(a: &str, b: &str) -> usize {
    let mut prev: Vec<usize> = (0..=b.len()).collect();
    let mut curr = vec![0; b.len() + 1];
    for (i, ca) in a.bytes().enumerate() {
        curr[0] = i + 1;
        for (j, cb) in b.bytes().enumerate() {
            let cost = usize::from(ca != cb);
            curr[j + 1] = (prev[j + 1] + 1).min(curr[j] + 1).min(prev[j] + cost);
        }
        std::mem::swap(&mut prev, &mut curr);
    }
    prev[b.len()]
}

fn keyword_suggestion(word: &str) -> Option<&'static str> {
    const STATEMENT_KEYWORDS: &[&str] = &[
        "alter", "begin", "commit", "delete", "drop", "explain", "insert", "refresh", "rollback",
        "select", "type", "update", "upsert",
    ];
    let lower = word.to_ascii_lowercase();
    STATEMENT_KEYWORDS
        .iter()
        .copied()
        .filter(|kw| edit_distance(&lower, kw) <= 2)
        .min_by_key(|kw| edit_distance(&lower, kw))
}

/// Shared tail of [`parse`] / [`parse_with_params`]: run the recursive
/// descent over an already-lexed (and possibly param-substituted) token
/// stream and reject any trailing tokens.
fn parse_tokens(tokens: Vec<Token>) -> Result<Statement, ParseError> {
    let mut parser = Parser {
        tokens,
        pos: 0,
        depth: 0,
    };
    let stmt = parser.parse_statement()?;
    // Reject trailing tokens. Without this, unrecognized tails like
    // `User create_index .email` silently succeed as `User` — which
    // misled the TS client into thinking those non-existent DDL forms
    // returned rows. A parse error here tells users that the syntax
    // they wrote isn't recognized.
    if !matches!(parser.peek(), Token::Eof) {
        let mut message = format!(
            "unexpected trailing token near token {}: {}",
            parser.pos,
            parser.peek().display_name()
        );
        if let Some(Token::Ident(first)) = parser.tokens.first() {
            if let Some(suggestion) = keyword_suggestion(first) {
                message.push_str(&format!("; did you mean `{suggestion}`?"));
            }
        }
        return Err(ParseError::Syntax { message });
    }
    Ok(stmt)
}

/// Rewrite `Field(alias)` references inside `expr` to the underlying
/// projection expression they alias. Used to desugar post-projection HAVING
/// (`{ ..., cnt: count(.name) } having cnt >= 2`) into a form the planner's
/// aggregate extraction can handle.
fn substitute_projection_aliases(expr: Expr, fields: &[ProjectionField]) -> Expr {
    match expr {
        Expr::Field(ref name) => {
            for f in fields {
                if f.alias.as_deref() == Some(name.as_str()) {
                    return f.expr.clone();
                }
            }
            expr
        }
        Expr::BinaryOp(l, op, r) => Expr::BinaryOp(
            Box::new(substitute_projection_aliases(*l, fields)),
            op,
            Box::new(substitute_projection_aliases(*r, fields)),
        ),
        Expr::UnaryOp(op, inner) => {
            Expr::UnaryOp(op, Box::new(substitute_projection_aliases(*inner, fields)))
        }
        Expr::Coalesce(l, r) => Expr::Coalesce(
            Box::new(substitute_projection_aliases(*l, fields)),
            Box::new(substitute_projection_aliases(*r, fields)),
        ),
        Expr::InList {
            expr: e,
            list,
            negated,
        } => Expr::InList {
            expr: Box::new(substitute_projection_aliases(*e, fields)),
            list: list
                .into_iter()
                .map(|i| substitute_projection_aliases(i, fields))
                .collect(),
            negated,
        },
        Expr::ScalarFunc(f, args) => Expr::ScalarFunc(
            f,
            args.into_iter()
                .map(|a| substitute_projection_aliases(a, fields))
                .collect(),
        ),
        other => other,
    }
}

impl Parser {
    fn peek(&self) -> &Token {
        &self.tokens[self.pos]
    }

    fn advance(&mut self) -> Token {
        let t = self.tokens[self.pos].clone();
        self.pos += 1;
        t
    }

    fn expect(&mut self, expected: &Token) -> Result<(), ParseError> {
        let t = self.advance();
        if &t == expected {
            Ok(())
        } else {
            Err(ParseError::UnexpectedToken {
                expected: expected.display_name(),
                got: t.display_name(),
            })
        }
    }

    /// Convenience: create an UnexpectedToken error.
    fn unexpected(&self, expected: &str, got: &Token) -> ParseError {
        ParseError::UnexpectedToken {
            expected: expected.into(),
            got: got.display_name(),
        }
    }

    /// Consume an identifier in a position that names something (a field,
    /// column, …). When the caller wrote a reserved word instead, the error
    /// says so and points at the backtick-quoting escape hatch, rather than
    /// the opaque `expected field name, got 'type'`.
    fn expect_named_ident(&mut self, context: &str) -> Result<String, ParseError> {
        match self.advance() {
            Token::Ident(n) => Ok(n),
            t => Err(self.named_ident_error(context, &t)),
        }
    }

    /// Build the error for a reserved word (or other token) appearing where an
    /// identifier was required. `context` is the noun ("field name", "column
    /// name") spliced into the message.
    fn named_ident_error(&self, context: &str, got: &Token) -> ParseError {
        if let Some(kw) = got.keyword_str() {
            ParseError::Syntax {
                // "syntax error" prefix keeps the message on the server's
                // safe-to-forward allowlist (SAFE_ERROR_PREFIXES) so wire
                // clients see the guidance instead of the generic mask.
                message: format!(
                    "syntax error: '{kw}' is a reserved word and cannot be used as a {context}; \
                     rename it or quote it as `{kw}`"
                ),
            }
        } else {
            ParseError::UnexpectedToken {
                expected: context.into(),
                got: got.display_name(),
            }
        }
    }

    /// Consume an optional `if not exists` clause. `if` is not a keyword token
    /// (it lexes as an identifier), so match it by spelling.
    fn parse_optional_if_not_exists(&mut self) -> bool {
        if matches!(self.peek(), Token::Ident(w) if w == "if")
            && matches!(self.tokens.get(self.pos + 1), Some(Token::Not))
            && matches!(self.tokens.get(self.pos + 2), Some(Token::Exists))
        {
            self.pos += 3;
            true
        } else {
            false
        }
    }

    /// Consume an optional `if exists` clause.
    fn parse_optional_if_exists(&mut self) -> bool {
        if matches!(self.peek(), Token::Ident(w) if w == "if")
            && matches!(self.tokens.get(self.pos + 1), Some(Token::Exists))
        {
            self.pos += 2;
            true
        } else {
            false
        }
    }

    fn parse_statement(&mut self) -> Result<Statement, ParseError> {
        self.depth += 1;
        if self.depth > MAX_NESTING_DEPTH {
            self.depth -= 1;
            return Err(ParseError::NestingDepthExceeded {
                max: MAX_NESTING_DEPTH,
            });
        }
        if matches!(self.peek(), Token::Explain) {
            self.advance();
            let inner = self.parse_statement()?;
            self.depth -= 1;
            return Ok(Statement::Explain(Box::new(inner)));
        }
        let stmt = match self.peek() {
            Token::Insert => self.parse_insert(),
            Token::Upsert => self.parse_upsert(),
            Token::Type => self.parse_create_type(),
            Token::Link => self.parse_create_link(),
            Token::Alter => self.parse_alter_table(),
            Token::Drop => self.parse_drop_or_drop_view(),
            Token::Materialized => self.parse_create_view(),
            Token::Refresh => self.parse_refresh_view(),
            Token::Begin => {
                self.advance();
                // Optional `transaction` keyword after `begin`.
                if *self.peek() == Token::Transaction {
                    self.advance();
                }
                return Ok(Statement::Begin);
            }
            Token::Commit => {
                self.advance();
                return Ok(Statement::Commit);
            }
            Token::Rollback => {
                self.advance();
                return Ok(Statement::Rollback);
            }
            Token::Schema => self.parse_schema(),
            Token::Describe => self.parse_describe(),
            Token::Count | Token::Avg | Token::Sum | Token::Min | Token::Max => {
                self.parse_aggregate_query()
            }
            Token::Ident(_) => self.parse_query_or_mutation(),
            Token::Update => Err(ParseError::Syntax {
                message: "'update' cannot start a statement — in PowQL, use pipeline syntax: \
                    TableName filter ... update { ... }"
                    .into(),
            }),
            Token::Delete => Err(ParseError::Syntax {
                message: "'delete' cannot start a statement — in PowQL, use pipeline syntax: \
                    TableName filter ... delete"
                    .into(),
            }),
            _ => Err(self.unexpected("statement", self.peek())),
        }?;
        // Check for UNION chaining after any query-producing statement.
        let result = self.maybe_parse_union(stmt);
        self.depth -= 1;
        result
    }

    fn parse_query_or_mutation(&mut self) -> Result<Statement, ParseError> {
        let source = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "type name".into(),
                    got: t.display_name(),
                })
            }
        };
        let alias = self.try_parse_alias();
        let joins = self.parse_joins()?;

        // Walk filter/order/limit/offset/projection, peeling off update/delete
        // mutations as we hit them. Anything else terminates the read pipeline
        // and we return a Query.
        let mut filter = None;
        let mut order = None;
        let mut limit = None;
        let mut offset = None;
        let mut projection = None;
        let mut distinct = false;
        let mut group_by = None;
        // Repeated `having` clauses stack `And` nodes onto one predicate, so
        // the pipeline loop grows the AST the same way a binary chain does and
        // needs the same bound.
        let mut having_chain = 0usize;

        loop {
            match self.peek() {
                Token::Distinct => {
                    self.advance();
                    distinct = true;
                }
                Token::Group => {
                    self.advance();
                    group_by = Some(self.parse_group_by()?);
                }
                Token::Filter => {
                    self.advance();
                    filter = Some(self.parse_expr()?);
                }
                Token::Order => {
                    self.advance();
                    order = Some(self.parse_order()?);
                }
                Token::Limit => {
                    self.advance();
                    limit = Some(self.parse_expr()?);
                }
                Token::Offset => {
                    self.advance();
                    offset = Some(self.parse_expr()?);
                }
                Token::LBrace => {
                    projection = Some(self.parse_projection()?);
                }
                Token::Having => {
                    // Post-projection HAVING — see parse_query_tail for details.
                    self.advance();
                    let having_expr = self.parse_expr()?;
                    let group = group_by.as_mut().ok_or_else(|| ParseError::Syntax {
                        message: "having without group by".into(),
                    })?;
                    let rewritten = match projection.as_ref() {
                        Some(fields) => substitute_projection_aliases(having_expr, fields),
                        None => having_expr,
                    };
                    group.having = Some(match group.having.take() {
                        Some(existing) => {
                            having_chain += 1;
                            self.check_chain_depth(having_chain)?;
                            Expr::BinaryOp(Box::new(existing), BinOp::And, Box::new(rewritten))
                        }
                        None => rewritten,
                    });
                }
                Token::Update => {
                    if !joins.is_empty() {
                        return Err(ParseError::Unsupported {
                            feature: "update on a joined query is not supported".into(),
                        });
                    }
                    self.advance();
                    let assignments = self.parse_assignments()?;
                    // Optional trailing `returning` — return the post-update rows.
                    let returning = *self.peek() == Token::Returning;
                    if returning {
                        self.advance();
                    }
                    return Ok(Statement::UpdateQuery(UpdateExpr {
                        source,
                        alias,
                        filter,
                        assignments,
                        returning,
                    }));
                }
                Token::Delete => {
                    if !joins.is_empty() {
                        return Err(ParseError::Unsupported {
                            feature: "delete on a joined query is not supported".into(),
                        });
                    }
                    self.advance();
                    // Optional trailing `returning` — return the pre-delete rows.
                    let returning = *self.peek() == Token::Returning;
                    if returning {
                        self.advance();
                    }
                    return Ok(Statement::DeleteQuery(DeleteExpr {
                        source,
                        alias,
                        filter,
                        returning,
                    }));
                }
                _ => break,
            }
        }

        Ok(Statement::Query(QueryExpr {
            source,
            alias,
            joins,
            filter,
            order,
            limit,
            offset,
            projection,
            aggregation: None,
            distinct,
            group_by,
        }))
    }

    /// Parse the read-only tail of a query (filter/order/limit/offset/projection)
    /// after `source` has already been consumed. Stops at the first token that
    /// isn't part of a read pipeline — the caller decides whether that's a
    /// terminator (RParen for an aggregate, EOF for a top-level query, etc.).
    /// Always returns `aggregation: None`; the caller layers that on.
    fn parse_query_tail(&mut self, source: String) -> Result<QueryExpr, ParseError> {
        let alias = self.try_parse_alias();
        let joins = self.parse_joins()?;
        let mut filter = None;
        let mut order = None;
        let mut limit = None;
        let mut offset = None;
        let mut projection = None;
        let mut distinct = false;
        let mut group_by = None;
        // Repeated `having` clauses stack `And` nodes onto one predicate, so
        // the pipeline loop grows the AST the same way a binary chain does and
        // needs the same bound.
        let mut having_chain = 0usize;

        loop {
            match self.peek() {
                Token::Distinct => {
                    self.advance();
                    distinct = true;
                }
                Token::Group => {
                    self.advance();
                    group_by = Some(self.parse_group_by()?);
                }
                Token::Filter => {
                    self.advance();
                    filter = Some(self.parse_expr()?);
                }
                Token::Order => {
                    self.advance();
                    order = Some(self.parse_order()?);
                }
                Token::Limit => {
                    self.advance();
                    limit = Some(self.parse_expr()?);
                }
                Token::Offset => {
                    self.advance();
                    offset = Some(self.parse_expr()?);
                }
                Token::LBrace => {
                    projection = Some(self.parse_projection()?);
                }
                Token::Having => {
                    // Post-projection HAVING — `... group .k { .k, cnt: count(.name) } having cnt >= 2`.
                    // Only meaningful when a GROUP BY is present. We desugar
                    // to a regular HAVING on the GroupByClause, rewriting
                    // projection aliases back into their underlying expressions
                    // so the planner's extract_aggregates can dedup them.
                    self.advance();
                    let having_expr = self.parse_expr()?;
                    let group = group_by.as_mut().ok_or_else(|| ParseError::Syntax {
                        message: "having without group by".into(),
                    })?;
                    let rewritten = match projection.as_ref() {
                        Some(fields) => substitute_projection_aliases(having_expr, fields),
                        None => having_expr,
                    };
                    group.having = Some(match group.having.take() {
                        Some(existing) => {
                            having_chain += 1;
                            self.check_chain_depth(having_chain)?;
                            Expr::BinaryOp(Box::new(existing), BinOp::And, Box::new(rewritten))
                        }
                        None => rewritten,
                    });
                }
                _ => break,
            }
        }

        Ok(QueryExpr {
            source,
            alias,
            joins,
            filter,
            order,
            limit,
            offset,
            projection,
            aggregation: None,
            distinct,
            group_by,
        })
    }

    /// Consume an optional `as <ident>` suffix on a source. Returns `None`
    /// if the next token isn't `as`. Used by both the primary source and each
    /// join source so queries can disambiguate columns via `alias.field`.
    fn try_parse_alias(&mut self) -> Option<String> {
        if *self.peek() == Token::As {
            self.advance();
            if let Token::Ident(name) = self.peek().clone() {
                self.advance();
                return Some(name);
            }
        }
        None
    }

    /// Parse zero or more join clauses. Each clause is:
    ///   (`inner` | `left` [`outer`] | `right` [`outer`] | `cross`)? `join`
    ///   <Ident> [`as` <ident>] [`on` <expr>]
    ///
    /// `on` is required for every kind except `cross`. The default kind is
    /// `inner` when the caller wrote bare `join` without a preceding modifier.
    fn parse_joins(&mut self) -> Result<Vec<JoinClause>, ParseError> {
        let mut joins = Vec::new();
        loop {
            let kind = match self.peek() {
                Token::Join => {
                    self.advance();
                    JoinKind::Inner
                }
                Token::Inner => {
                    self.advance();
                    self.expect(&Token::Join)?;
                    JoinKind::Inner
                }
                Token::LeftKw => {
                    self.advance();
                    if *self.peek() == Token::Outer {
                        self.advance();
                    }
                    self.expect(&Token::Join)?;
                    JoinKind::LeftOuter
                }
                Token::RightKw => {
                    self.advance();
                    if *self.peek() == Token::Outer {
                        self.advance();
                    }
                    self.expect(&Token::Join)?;
                    JoinKind::RightOuter
                }
                Token::Cross => {
                    self.advance();
                    self.expect(&Token::Join)?;
                    JoinKind::Cross
                }
                _ => break,
            };

            let source = match self.advance() {
                Token::Ident(name) => name,
                t => {
                    return Err(ParseError::UnexpectedToken {
                        expected: "type name after join".into(),
                        got: t.display_name(),
                    });
                }
            };
            let alias = self.try_parse_alias();
            let on = if kind == JoinKind::Cross {
                None
            } else if *self.peek() == Token::On {
                self.advance();
                Some(self.parse_expr()?)
            } else {
                return Err(ParseError::Syntax {
                    message: format!("expected `on <expr>` after join {source}"),
                });
            };

            joins.push(JoinClause {
                kind,
                source,
                alias,
                on,
            });
        }
        Ok(joins)
    }

    fn parse_insert(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Insert)?;
        let target = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "type name".into(),
                    got: t.display_name(),
                })
            }
        };
        // One or more comma-separated assignment blocks:
        //   insert T { a := 1 }
        //   insert T { a := 1 }, { a := 2 }, { a := 3 }
        let mut rows = vec![self.parse_assignments()?];
        while *self.peek() == Token::Comma {
            self.advance(); // consume the comma between row blocks
            rows.push(self.parse_assignments()?);
        }
        // Optional trailing `returning` — return the inserted rows.
        let returning = *self.peek() == Token::Returning;
        if returning {
            self.advance();
        }
        Ok(Statement::Insert(InsertExpr {
            target,
            rows,
            returning,
        }))
    }

    /// Parse: `upsert Table on .key_col { assignments } [on conflict { update_assignments }]`
    fn parse_upsert(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Upsert)?;
        let target = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "type name".into(),
                    got: t.display_name(),
                })
            }
        };
        self.expect(&Token::On)?;
        let key_column = match self.advance() {
            Token::DotIdent(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: ".key_column".into(),
                    got: t.display_name(),
                })
            }
        };
        let assignments = self.parse_assignments()?;
        let on_conflict = if *self.peek() == Token::On {
            self.advance(); // consume `on`
            self.expect(&Token::Conflict)?;
            self.parse_assignments()?
        } else {
            Vec::new()
        };
        Ok(Statement::Upsert(UpsertExpr {
            target,
            key_column,
            assignments,
            on_conflict,
        }))
    }

    fn parse_assignments(&mut self) -> Result<Vec<Assignment>, ParseError> {
        self.expect(&Token::LBrace)?;
        let mut assignments = Vec::new();
        while !matches!(self.peek(), Token::RBrace | Token::Eof) {
            // A JSON path target (`.data->x := ...` or `data->x := ...`) is a
            // field/dot-field immediately followed by `->`. Detect it before
            // the generic ident/`:=` parse so the error names the unsupported
            // position and the whole-column alternative, instead of a bare
            // "expected field name" (the `.data` DotIdent case) or "expected
            // ':='" (the `data` Ident case).
            if matches!(self.peek(), Token::DotIdent(_) | Token::Ident(_))
                && matches!(self.tokens.get(self.pos + 1), Some(Token::Arrow))
            {
                let field = match self.peek() {
                    Token::DotIdent(n) | Token::Ident(n) => n.clone(),
                    _ => unreachable!("guarded by the matches! above"),
                };
                return Err(ParseError::Unsupported {
                    feature: format!(
                        "cannot assign to a JSON path target `.{field}->...` (at token {pos}): \
                         JSON path assignment targets are not supported; write the whole JSON \
                         column instead (path mutation such as json_set is not yet available)",
                        pos = self.pos
                    ),
                });
            }
            let field = self.expect_named_ident("field name")?;
            self.expect(&Token::Assign)?;
            let value = self.parse_expr()?;
            assignments.push(Assignment { field, value });
            if *self.peek() == Token::Comma {
                self.advance();
            }
        }
        self.expect(&Token::RBrace)?;
        Ok(assignments)
    }

    fn parse_projection(&mut self) -> Result<Vec<ProjectionField>, ParseError> {
        self.expect(&Token::LBrace)?;
        let mut fields = Vec::new();
        while !matches!(self.peek(), Token::RBrace | Token::Eof) {
            // `alias: expr` is detected with two-token lookahead so the bare
            // form can parse a full expression from its first token. Every
            // projection slot — aliased or bare — flows through the shared
            // expression parser, so fields, qualified refs, aggregates, window
            // functions, scalar calls, CASE/CAST, and arithmetic like `.a - 1`
            // are all accepted. Previously the bare slot used a restricted
            // dispatch that rejected any binary operator with "expected field".
            if matches!(self.peek(), Token::Ident(_))
                && matches!(self.tokens.get(self.pos + 1), Some(Token::Colon))
            {
                let alias = match self.advance() {
                    Token::Ident(name) => name,
                    _ => unreachable!("guarded by the matches! above"),
                };
                self.advance(); // consume ':'
                self.reject_bare_dotted_path()?;
                // `alias: Ident as ...` is a nested sub-query projection
                // (language-lab slice): `orders: Order as o filter ... { ... }`.
                let expr = if matches!(self.peek(), Token::Ident(_))
                    && matches!(self.tokens.get(self.pos + 1), Some(Token::As))
                {
                    Expr::NestedQuery(Box::new(self.parse_nested_query()?))
                } else if self.at_link_traversal() {
                    // `orders: u.orders [filter ...] { ... }`: block link
                    // traversal, sugar over the nested-projection machinery.
                    // Distinguished from an ordinary `x: u.name` projection by
                    // the `{`/clause that follows the `alias.ident` (a plain
                    // qualified field is followed by `,` or `}`).
                    Expr::NestedQuery(Box::new(self.parse_link_traversal()?))
                } else if self.at_scalar_link_path() {
                    // `buyer: o.user.name`: scalar link traversal.
                    self.parse_scalar_link_path()?
                } else {
                    self.parse_expr()?
                };
                fields.push(ProjectionField {
                    alias: Some(alias),
                    expr,
                });
            } else {
                if matches!(self.peek(), Token::Ident(_))
                    && matches!(self.tokens.get(self.pos + 1), Some(Token::As))
                {
                    return Err(ParseError::Syntax {
                        message: "a nested projection needs a field name: \
                                  `<name>: <Table> as <alias> filter ... { ... }`"
                            .into(),
                    });
                }
                self.reject_bare_dotted_path()?;
                let expr = if self.at_scalar_link_path() {
                    // `o.user.name`: scalar link traversal, named by its
                    // dotted spelling when no alias is written.
                    self.parse_scalar_link_path()?
                } else {
                    self.parse_expr()?
                };
                fields.push(ProjectionField { alias: None, expr });
            }
            if *self.peek() == Token::Comma {
                self.advance();
            }
        }
        self.expect(&Token::RBrace)?;
        Ok(fields)
    }

    /// Reject a projection slot that starts with two adjacent dotted parts
    /// (`.user.name`). The token stream cannot distinguish an intended link
    /// path from two comma-less bare fields (`.user .name` lexes identically),
    /// and a link path needs the aliased form to name its outer scan, so this
    /// used to silently parse as TWO separate fields and project Empty
    /// columns. A hard error with guidance beats the silent wrong shape.
    fn reject_bare_dotted_path(&self) -> Result<(), ParseError> {
        if let (Token::DotIdent(first), Some(Token::DotIdent(second))) =
            (self.peek(), self.tokens.get(self.pos + 1))
        {
            return Err(ParseError::Syntax {
                message: format!(
                    "`.{first}.{second}` is ambiguous in a projection: for a link \
                     path, alias the table and qualify the path \
                     (`Order as o {{ o.{first}.{second} }}`); for separate fields, \
                     separate them with commas (`.{first}, .{second}`)"
                ),
            });
        }
        Ok(())
    }

    /// Conservative lookahead for a block link-traversal projection value:
    /// `<Ident>.<DotIdent>` immediately followed by `{` or a
    /// `filter`/`order`/`limit`/`offset` clause. An ordinary qualified-field
    /// projection (`x: u.name`) is instead followed by `,` or `}`, so it never
    /// matches here.
    fn at_link_traversal(&self) -> bool {
        matches!(self.peek(), Token::Ident(_))
            && matches!(self.tokens.get(self.pos + 1), Some(Token::DotIdent(_)))
            && matches!(
                self.tokens.get(self.pos + 2),
                Some(Token::LBrace | Token::Filter | Token::Order | Token::Limit | Token::Offset)
            )
    }

    /// Lookahead for a scalar link-traversal projection value:
    /// `<Ident>.<DotIdent>.<DotIdent>...`, i.e. three or more dotted parts.
    /// A plain qualified field (`o.total`) has exactly two parts and never
    /// matches; a block traversal (`u.orders { ... }`) is caught first by
    /// `at_link_traversal` (its second part is followed by `{` or a clause,
    /// never by another `.ident`).
    fn at_scalar_link_path(&self) -> bool {
        matches!(self.peek(), Token::Ident(_))
            && matches!(self.tokens.get(self.pos + 1), Some(Token::DotIdent(_)))
            && matches!(self.tokens.get(self.pos + 2), Some(Token::DotIdent(_)))
    }

    /// Parse a scalar link traversal path: `o.user.name` or
    /// `o.user.company.name`. All parts but the last are declared to-one link
    /// names (resolved from the persistent catalog at execution time); the
    /// last is the target column to read. Only valid as a projection field
    /// value.
    fn parse_scalar_link_path(&mut self) -> Result<Expr, ParseError> {
        let outer_alias = match self.advance() {
            Token::Ident(n) => n,
            t => return Err(self.named_ident_error("link path outer alias", &t)),
        };
        let mut parts = Vec::new();
        while let Some(Token::DotIdent(_)) = self.tokens.get(self.pos) {
            match self.advance() {
                Token::DotIdent(n) => parts.push(n),
                _ => unreachable!("peeked DotIdent"),
            }
        }
        debug_assert!(parts.len() >= 2, "guarded by at_scalar_link_path");
        let column = parts.pop().expect("at least two parts");
        Ok(Expr::LinkPath {
            outer_alias,
            links: parts,
            column,
        })
    }

    /// Parse a block link traversal projection value:
    /// `<outer_alias>.<link_name> [filter <conds>] [order ...] [limit N]
    /// [offset M] { <bare child fields> }`. Desugars to a [`NestedQuery`]
    /// carrying a [`ViaLink`]: the child source table and correlation columns
    /// are unknown here (they live in the persistent catalog), so `source` is a
    /// placeholder, the synthetic child alias is the link name itself, and only
    /// the user's residual conditions land in `filter`. Bare child columns in
    /// the block, filter, and order are qualified with the synthetic alias so
    /// downstream planning matches the explicit correlated spelling.
    fn parse_link_traversal(&mut self) -> Result<NestedQuery, ParseError> {
        self.depth += 1;
        if self.depth > MAX_NESTING_DEPTH {
            self.depth -= 1;
            return Err(ParseError::NestingDepthExceeded {
                max: MAX_NESTING_DEPTH,
            });
        }
        let result = self.parse_link_traversal_inner();
        self.depth -= 1;
        result
    }

    fn parse_link_traversal_inner(&mut self) -> Result<NestedQuery, ParseError> {
        let outer_alias = match self.advance() {
            Token::Ident(n) => n,
            t => return Err(self.named_ident_error("link outer alias", &t)),
        };
        let link_name = match self.advance() {
            Token::DotIdent(n) => n,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "link name".into(),
                    got: t.display_name(),
                })
            }
        };
        // The synthetic child alias is the link name itself; bare child
        // references are qualified with it below.
        let child_alias = link_name.clone();

        // Optional residual filter, then order/limit/offset, matching the
        // explicit nested-query grammar. No correlation predicate is written
        // by the user: the whole filter is residual.
        let mut residual = None;
        if *self.peek() == Token::Filter {
            self.advance();
            residual = Some(qualify_bare_fields(self.parse_expr()?, &child_alias));
        }
        let mut order = None;
        let mut limit = None;
        let mut offset = None;
        let mut offset_before_limit = false;
        loop {
            match self.peek() {
                Token::Order => {
                    self.advance();
                    let mut clause = self.parse_order()?;
                    for key in &mut clause.keys {
                        key.expr = qualify_bare_fields(
                            std::mem::replace(&mut key.expr, Expr::Null),
                            &child_alias,
                        );
                    }
                    order = Some(clause);
                }
                Token::Limit => {
                    self.advance();
                    if offset.is_some() {
                        offset_before_limit = true;
                    }
                    limit = Some(self.parse_expr()?);
                }
                Token::Offset => {
                    self.advance();
                    offset = Some(self.parse_expr()?);
                }
                _ => break,
            }
        }
        self.expect(&Token::LBrace)?;
        let mut fields = Vec::new();
        while !matches!(self.peek(), Token::RBrace | Token::Eof) {
            let alias = if matches!(self.peek(), Token::Ident(_))
                && matches!(self.tokens.get(self.pos + 1), Some(Token::Colon))
            {
                let alias = self.expect_named_ident("field alias")?;
                self.advance(); // consume ':'
                Some(alias)
            } else {
                None
            };
            self.reject_bare_dotted_path()?;
            let expr = qualify_bare_fields(self.parse_expr()?, &child_alias);
            fields.push(ProjectionField { alias, expr });
            if *self.peek() == Token::Comma {
                self.advance();
            }
        }
        self.expect(&Token::RBrace)?;
        if fields.is_empty() {
            return Err(ParseError::Syntax {
                message: "link traversal requires at least one field".into(),
            });
        }
        Ok(NestedQuery {
            // Placeholder source: resolved from the catalog at execution.
            source: String::new(),
            alias: child_alias,
            via_link: Some(ViaLink {
                outer_alias,
                link_name,
            }),
            // A `true` residual is equivalent to no residual; supply `true`
            // when the user wrote no filter to keep the planner simple.
            filter: residual.unwrap_or(Expr::Literal(Literal::Bool(true))),
            order,
            limit,
            offset,
            offset_before_limit,
            fields,
        })
    }

    /// Parse a nested sub-query projection value:
    /// `<ChildTable> as <alias> filter <predicate> [order ...] [limit N]
    /// [offset M] { <fields> }`. The block accepts plain and aliased scalar
    /// fields plus further `name: Table as alias ...` nesting.
    fn parse_nested_query(&mut self) -> Result<NestedQuery, ParseError> {
        // Nested blocks recurse; share the expression nesting-depth guard so
        // pathological inputs fail cleanly instead of overflowing the stack.
        self.depth += 1;
        if self.depth > MAX_NESTING_DEPTH {
            self.depth -= 1;
            return Err(ParseError::NestingDepthExceeded {
                max: MAX_NESTING_DEPTH,
            });
        }
        let result = self.parse_nested_query_inner();
        self.depth -= 1;
        result
    }

    fn parse_nested_query_inner(&mut self) -> Result<NestedQuery, ParseError> {
        let source = self.expect_named_ident("nested source type")?;
        self.expect(&Token::As)?;
        let alias = self.expect_named_ident("nested source alias")?;
        self.expect(&Token::Filter)?;
        let filter = self.parse_expr()?;
        let mut order = None;
        let mut limit = None;
        let mut offset = None;
        let mut offset_before_limit = false;
        loop {
            match self.peek() {
                Token::Order => {
                    self.advance();
                    order = Some(self.parse_order()?);
                }
                Token::Limit => {
                    self.advance();
                    if offset.is_some() {
                        offset_before_limit = true;
                    }
                    limit = Some(self.parse_expr()?);
                }
                Token::Offset => {
                    self.advance();
                    offset = Some(self.parse_expr()?);
                }
                _ => break,
            }
        }
        self.expect(&Token::LBrace)?;
        let mut fields = Vec::new();
        while !matches!(self.peek(), Token::RBrace | Token::Eof) {
            let alias = if matches!(self.peek(), Token::Ident(_))
                && matches!(self.tokens.get(self.pos + 1), Some(Token::Colon))
            {
                let alias = self.expect_named_ident("field alias")?;
                self.advance(); // consume ':'
                Some(alias)
            } else {
                None
            };
            let expr = if matches!(self.peek(), Token::Ident(_))
                && matches!(self.tokens.get(self.pos + 1), Some(Token::As))
            {
                if alias.is_none() {
                    return Err(ParseError::Syntax {
                        message: "a nested projection needs a field name: \
                                  `<name>: <Table> as <alias> filter ... { ... }`"
                            .into(),
                    });
                }
                Expr::NestedQuery(Box::new(self.parse_nested_query()?))
            } else {
                self.reject_bare_dotted_path()?;
                self.parse_expr()?
            };
            fields.push(ProjectionField { alias, expr });
            if *self.peek() == Token::Comma {
                self.advance();
            }
        }
        self.expect(&Token::RBrace)?;
        if fields.is_empty() {
            return Err(ParseError::Syntax {
                message: "nested projection requires at least one field".into(),
            });
        }
        Ok(NestedQuery {
            source,
            alias,
            via_link: None,
            filter,
            order,
            limit,
            offset,
            offset_before_limit,
            fields,
        })
    }

    /// Parse the OVER clause for a window function:
    /// `over (partition .col1, .col2 order .col3 asc, .col4 desc)`
    fn parse_over_clause(&mut self) -> Result<(Vec<Expr>, Vec<OrderKey>), ParseError> {
        self.expect(&Token::Over)?;
        self.expect(&Token::LParen)?;
        let mut partition_by = Vec::new();
        let mut order_by = Vec::new();
        if *self.peek() == Token::Partition {
            self.advance();
            loop {
                partition_by.push(self.parse_expr()?);
                if *self.peek() == Token::Comma {
                    if !matches!(
                        self.tokens.get(self.pos + 1),
                        Some(Token::Order | Token::RParen)
                    ) {
                        self.advance();
                    } else {
                        break;
                    }
                } else {
                    break;
                }
            }
        }
        if *self.peek() == Token::Order {
            self.advance();
            loop {
                let expr = self.parse_expr()?;
                let descending = match self.peek() {
                    Token::Desc => {
                        self.advance();
                        true
                    }
                    Token::Asc => {
                        self.advance();
                        false
                    }
                    _ => false,
                };
                order_by.push(OrderKey { expr, descending });
                if *self.peek() == Token::Comma {
                    self.advance();
                } else {
                    break;
                }
            }
        }
        self.expect(&Token::RParen)?;
        Ok((partition_by, order_by))
    }

    /// Parse a cast target type from a string literal: `"int"`, `"float"`, `"str"`, `"bool"`, `"datetime"`.
    fn parse_cast_type(&mut self) -> Result<CastType, ParseError> {
        match self.advance() {
            Token::StringLit(s) => match s.as_str() {
                "int" | "Int" | "INT" => Ok(CastType::Int),
                "float" | "Float" | "FLOAT" => Ok(CastType::Float),
                "str" | "Str" | "STR" | "string" | "String" => Ok(CastType::Str),
                "bool" | "Bool" | "BOOL" | "boolean" => Ok(CastType::Bool),
                "datetime" | "DateTime" | "DATETIME" => Ok(CastType::DateTime),
                "uuid" | "Uuid" | "UUID" => Ok(CastType::Uuid),
                "bytes" | "Bytes" | "BYTES" | "bytea" => Ok(CastType::Bytes),
                other => Err(ParseError::Syntax {
                    message: format!("invalid cast type: \"{other}\""),
                }),
            },
            t => Err(ParseError::UnexpectedToken {
                expected: "string literal for cast type".into(),
                got: t.display_name(),
            }),
        }
    }

    fn parse_order(&mut self) -> Result<OrderClause, ParseError> {
        let mut keys = Vec::new();
        loop {
            let expr = self.parse_expr()?;
            let descending = match self.peek() {
                Token::Desc => {
                    self.advance();
                    true
                }
                Token::Asc => {
                    self.advance();
                    false
                }
                _ => false,
            };
            keys.push(OrderKey { expr, descending });
            if *self.peek() == Token::Comma {
                self.advance();
            } else {
                break;
            }
        }
        Ok(OrderClause { keys })
    }

    fn parse_aggregate_query(&mut self) -> Result<Statement, ParseError> {
        let mut func = match self.advance() {
            Token::Count => AggFunc::Count,
            Token::Avg => AggFunc::Avg,
            Token::Sum => AggFunc::Sum,
            Token::Min => AggFunc::Min,
            Token::Max => AggFunc::Max,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "aggregate function".into(),
                    got: t.display_name(),
                })
            }
        };
        self.expect(&Token::LParen)?;
        let mode = if *self.peek() == Token::Raw {
            self.advance();
            AggregateMode::Raw
        } else {
            AggregateMode::Symmetric
        };
        // count(distinct User ...) → CountDistinct
        if func == AggFunc::Count && *self.peek() == Token::Distinct {
            self.advance();
            func = AggFunc::CountDistinct;
        }
        let source = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "type name".into(),
                    got: t.display_name(),
                })
            }
        };
        // Allow a full read-pipeline tail inside the parens, e.g.
        // `count(User filter .age > 27 limit 100)`. parse_query_tail stops at
        // the first non-pipeline token, which here must be RParen.
        let mut query = self.parse_query_tail(source)?;
        self.expect(&Token::RParen)?;

        // The caller writes the aggregate's target column via the trailing
        // projection form:
        //     sum(User filter .age > 30 { .age })
        //     count(distinct User { .name })
        //     count(User { .nickname })
        // We lift that single unaliased `.field` into AggregateExpr.field so
        // the executor's aggregate fast paths can see it. Plain `count` is
        // included: an argument makes it a non-null count of that column
        // (matching the grouped `count(.col)` path and SQL's `COUNT(col)`),
        // while the argument-less `count(User)` stays a row count. Ignoring
        // the projection here used to make `count(User { .nickname })` silently
        // return the row count.
        let mut argument: Option<Expr> = None;
        if let Some(proj) = &query.projection {
            if proj.len() == 1 && proj[0].alias.is_none() {
                argument = Some(proj[0].expr.clone());
            }
        }
        if argument.is_some() {
            query.projection = None;
        }
        query.aggregation = Some(AggregateExpr {
            function: func,
            argument,
            mode,
        });
        Ok(Statement::Query(query))
    }

    fn parse_expr(&mut self) -> Result<Expr, ParseError> {
        self.depth += 1;
        if self.depth > MAX_NESTING_DEPTH {
            self.depth -= 1;
            return Err(ParseError::NestingDepthExceeded {
                max: MAX_NESTING_DEPTH,
            });
        }
        let result = self.parse_or_expr();
        self.depth -= 1;
        result
    }

    /// Guard for a loop that stacks `chain` levels of AST on top of the
    /// current recursion depth. `chain` is the number of levels this loop has
    /// already added; the caller counts it locally rather than mutating
    /// `self.depth` so sibling expressions each get the full budget.
    fn check_chain_depth(&self, chain: usize) -> Result<(), ParseError> {
        if self.depth + chain > MAX_NESTING_DEPTH {
            return Err(ParseError::NestingDepthExceeded {
                max: MAX_NESTING_DEPTH,
            });
        }
        Ok(())
    }

    fn parse_or_expr(&mut self) -> Result<Expr, ParseError> {
        let mut left = self.parse_and_expr()?;
        let mut chain = 0usize;
        while *self.peek() == Token::Or {
            chain += 1;
            self.check_chain_depth(chain)?;
            self.advance();
            let right = self.parse_and_expr()?;
            left = Expr::BinaryOp(Box::new(left), BinOp::Or, Box::new(right));
        }
        Ok(left)
    }

    fn parse_and_expr(&mut self) -> Result<Expr, ParseError> {
        let mut left = self.parse_comparison()?;
        let mut chain = 0usize;
        while *self.peek() == Token::And {
            chain += 1;
            self.check_chain_depth(chain)?;
            self.advance();
            let right = self.parse_comparison()?;
            left = Expr::BinaryOp(Box::new(left), BinOp::And, Box::new(right));
        }
        Ok(left)
    }

    fn parse_comparison(&mut self) -> Result<Expr, ParseError> {
        // Prefix `not` lives at precedence level 4 (docs/POWQL.md): looser
        // than the comparisons parsed below, tighter than `and`/`or`. Consume
        // the whole prefix chain here so `not .v > 0` means `not (.v > 0)`,
        // matching the SQL frontend. `not exists` keeps its dedicated
        // primary-level parse (ExistsSubquery / NotExists), and an explicit
        // `(not .v) > 0` still reaches parse_primary's `not` through the
        // parentheses. The chain is counted iteratively (no recursion) and
        // capped like parse_primary's guard so `not not … .x` cannot
        // overflow the stack in the wrapping loop below or in later walks.
        let mut negations = 0usize;
        while *self.peek() == Token::Not
            && !matches!(self.tokens.get(self.pos + 1), Some(Token::Exists))
        {
            self.advance();
            negations += 1;
            self.check_chain_depth(negations)?;
        }
        let mut expr = self.parse_comparison_body()?;
        for _ in 0..negations {
            expr = Expr::UnaryOp(UnaryOp::Not, Box::new(expr));
        }
        Ok(expr)
    }

    fn parse_comparison_body(&mut self) -> Result<Expr, ParseError> {
        let left = self.parse_additive()?;

        // IS NULL / IS NOT NULL (postfix)
        if *self.peek() == Token::Is {
            self.advance();
            if *self.peek() == Token::Not {
                self.advance();
                self.expect(&Token::Null)?;
                return Ok(Expr::UnaryOp(UnaryOp::IsNotNull, Box::new(left)));
            } else {
                self.expect(&Token::Null)?;
                return Ok(Expr::UnaryOp(UnaryOp::IsNull, Box::new(left)));
            }
        }

        // Postfix: `in (...)`, `like "..."`, `between X and Y`
        // and their negated forms: `not in`, `not like`, `not between`.
        match self.peek() {
            Token::In => {
                self.advance();
                return self.parse_in_list(left, false);
            }
            Token::Like => {
                self.advance();
                let pattern = self.parse_additive()?;
                return Ok(Expr::BinaryOp(
                    Box::new(left),
                    BinOp::Like,
                    Box::new(pattern),
                ));
            }
            Token::Between => {
                self.advance();
                return self.parse_between(left, false);
            }
            Token::Not => {
                // Peek ahead: `not in`, `not like`, `not between`.
                // If the token after `not` isn't one of these, don't consume
                // `not` — let the caller handle it.
                let next = self.tokens.get(self.pos + 1);
                match next {
                    Some(Token::In) => {
                        self.advance(); // not
                        self.advance(); // in
                        return self.parse_in_list(left, true);
                    }
                    Some(Token::Like) => {
                        self.advance(); // not
                        self.advance(); // like
                        let pattern = self.parse_additive()?;
                        let like = Expr::BinaryOp(Box::new(left), BinOp::Like, Box::new(pattern));
                        return Ok(Expr::UnaryOp(UnaryOp::Not, Box::new(like)));
                    }
                    Some(Token::Between) => {
                        self.advance(); // not
                        self.advance(); // between
                        return self.parse_between(left, true);
                    }
                    _ => {}
                }
            }
            _ => {}
        }

        let op = match self.peek() {
            Token::Eq => BinOp::Eq,
            Token::Neq => BinOp::Neq,
            Token::Lt => BinOp::Lt,
            Token::Gt => BinOp::Gt,
            Token::Lte => BinOp::Lte,
            Token::Gte => BinOp::Gte,
            _ => return Ok(left),
        };
        self.advance();
        // `expr = null` / `expr != null` desugar to the same UnaryOp as
        // `expr is null` / `expr is not null`. Ordering comparisons against
        // null (`< null`, `>= null`, etc.) remain parse errors.
        if *self.peek() == Token::Null {
            match op {
                BinOp::Eq => {
                    self.advance();
                    return Ok(Expr::UnaryOp(UnaryOp::IsNull, Box::new(left)));
                }
                BinOp::Neq => {
                    self.advance();
                    return Ok(Expr::UnaryOp(UnaryOp::IsNotNull, Box::new(left)));
                }
                _ => {}
            }
        }
        let right = self.parse_additive()?;
        Ok(Expr::BinaryOp(Box::new(left), op, Box::new(right)))
    }

    /// Parse `(val1, val2, ...)` or `(subquery)` after `in` / `not in`.
    /// A subquery is detected by `(` followed by an `Ident` that is NOT
    /// followed by `,` or `)` — in PowQL, bare identifiers in value lists
    /// don't appear (field refs start with `.`).
    fn parse_in_list(&mut self, expr: Expr, negated: bool) -> Result<Expr, ParseError> {
        self.expect(&Token::LParen)?;
        // Detect subquery: `( Ident ...` where the Ident is a table name.
        if let Token::Ident(_) = self.peek() {
            // Peek further: if the next token after the Ident is NOT `,` or
            // `)`, it's a subquery source name.
            let after = self.tokens.get(self.pos + 1);
            let is_subquery = !matches!(after, Some(Token::Comma) | Some(Token::RParen));
            if is_subquery {
                let source = match self.advance() {
                    Token::Ident(name) => name,
                    _ => unreachable!(),
                };
                let subquery = self.parse_query_tail(source)?;
                self.expect(&Token::RParen)?;
                return Ok(Expr::InSubquery {
                    expr: Box::new(expr),
                    subquery: Box::new(subquery),
                    negated,
                });
            }
        }
        let mut list = Vec::new();
        while !matches!(self.peek(), Token::RParen | Token::Eof) {
            list.push(self.parse_expr()?);
            if *self.peek() == Token::Comma {
                self.advance();
            }
        }
        self.expect(&Token::RParen)?;
        Ok(Expr::InList {
            expr: Box::new(expr),
            list,
            negated,
        })
    }

    /// Try to parse a `(subquery)` tail for `exists` / `not exists`.
    /// A subquery is detected when the next tokens are `( Ident ...` —
    /// bare identifiers inside parens are always table/view names in
    /// PowQL (column refs start with `.`). Returns `Ok(Some(query))` if
    /// consumed, `Ok(None)` if the shape doesn't match (so the caller
    /// falls back to parsing a scalar primary for the legacy
    /// `exists <expr>` form).
    fn try_parse_exists_subquery(&mut self) -> Result<Option<QueryExpr>, ParseError> {
        if *self.peek() != Token::LParen {
            return Ok(None);
        }
        // Peek one token inside the paren. Anything starting with `Ident`
        // is a source name — PowQL column references use `DotIdent`, so
        // an `exists (X ...)` with a bare `X` is unambiguously a subquery.
        let after_lparen = self.tokens.get(self.pos + 1);
        if !matches!(after_lparen, Some(Token::Ident(_))) {
            return Ok(None);
        }
        self.expect(&Token::LParen)?;
        let source = match self.advance() {
            Token::Ident(name) => name,
            _ => unreachable!(),
        };
        let subquery = self.parse_query_tail(source)?;
        self.expect(&Token::RParen)?;
        Ok(Some(subquery))
    }

    /// Parse `low and high` after `between` / `not between`.
    /// Desugars into `expr >= low AND expr <= high` (or negated:
    /// `expr < low OR expr > high`).
    fn parse_between(&mut self, expr: Expr, negated: bool) -> Result<Expr, ParseError> {
        let low = self.parse_additive()?;
        self.expect(&Token::And)?;
        let high = self.parse_additive()?;
        if negated {
            // NOT BETWEEN: expr < low OR expr > high
            Ok(Expr::BinaryOp(
                Box::new(Expr::BinaryOp(
                    Box::new(expr.clone()),
                    BinOp::Lt,
                    Box::new(low),
                )),
                BinOp::Or,
                Box::new(Expr::BinaryOp(Box::new(expr), BinOp::Gt, Box::new(high))),
            ))
        } else {
            // BETWEEN: expr >= low AND expr <= high
            Ok(Expr::BinaryOp(
                Box::new(Expr::BinaryOp(
                    Box::new(expr.clone()),
                    BinOp::Gte,
                    Box::new(low),
                )),
                BinOp::And,
                Box::new(Expr::BinaryOp(Box::new(expr), BinOp::Lte, Box::new(high))),
            ))
        }
    }

    /// Parse expression-valued group keys.
    fn parse_group_by(&mut self) -> Result<GroupByClause, ParseError> {
        let mut keys = Vec::new();
        loop {
            let expr = self.parse_expr()?;
            let output_name = match &expr {
                Expr::Field(_) | Expr::QualifiedField { .. } | Expr::JsonPath { .. } => {
                    expression_output_name(&expr)
                }
                _ => format!("__group_{}", keys.len()),
            };
            keys.push(GroupKey { expr, output_name });
            if *self.peek() == Token::Comma {
                self.advance();
            } else {
                break;
            }
        }
        if keys.is_empty() {
            return Err(ParseError::Syntax {
                message: "expected at least one group key after group".into(),
            });
        }
        let having = if *self.peek() == Token::Having {
            self.advance();
            Some(self.parse_expr()?)
        } else {
            None
        };
        Ok(GroupByClause { keys, having })
    }

    fn parse_additive(&mut self) -> Result<Expr, ParseError> {
        let mut left = self.parse_multiplicative()?;
        let mut chain = 0usize;
        loop {
            let op = match self.peek() {
                Token::Plus => BinOp::Add,
                Token::Minus => BinOp::Sub,
                Token::Coalesce => {
                    chain += 1;
                    self.check_chain_depth(chain)?;
                    self.advance();
                    let right = self.parse_multiplicative()?;
                    left = Expr::Coalesce(Box::new(left), Box::new(right));
                    continue;
                }
                _ => break,
            };
            chain += 1;
            self.check_chain_depth(chain)?;
            self.advance();
            let right = self.parse_multiplicative()?;
            left = Expr::BinaryOp(Box::new(left), op, Box::new(right));
        }
        Ok(left)
    }

    fn parse_multiplicative(&mut self) -> Result<Expr, ParseError> {
        let mut left = self.parse_primary()?;
        let mut chain = 0usize;
        loop {
            let op = match self.peek() {
                Token::Star => BinOp::Mul,
                Token::Slash => BinOp::Div,
                _ => break,
            };
            chain += 1;
            self.check_chain_depth(chain)?;
            self.advance();
            let right = self.parse_primary()?;
            left = Expr::BinaryOp(Box::new(left), op, Box::new(right));
        }
        Ok(left)
    }

    fn parse_primary(&mut self) -> Result<Expr, ParseError> {
        // Guard recursion here too: unary prefixes (`not`, `exists`, `not exists`)
        // recurse straight back into parse_primary without going through parse_expr,
        // so a chain like `not not … .x` would otherwise overflow the stack (process
        // abort under panic=abort). See test_unary_prefix_nesting_depth_limit.
        self.depth += 1;
        if self.depth > MAX_NESTING_DEPTH {
            self.depth -= 1;
            return Err(ParseError::NestingDepthExceeded {
                max: MAX_NESTING_DEPTH,
            });
        }
        let result = self.parse_primary_inner();
        self.depth -= 1;
        // JSON `->` path access is the tightest-binding postfix level: it binds
        // above every binary operator, so `.data->age > 21` parses as
        // `(.data->age) > 21`. Applying it here (inside `parse_primary`) means
        // every caller — multiplicative, unary prefixes, function args — gets
        // path access for free without threading a new precedence level.
        self.parse_json_path_postfix(result?)
    }

    /// If the current token is `->`, consume a chain of path segments and wrap
    /// `base` in an `Expr::JsonPath`. `base` must be a `Field`, `QualifiedField`,
    /// or (nested) `JsonPath`; any other base is a parse error. Each segment is
    /// an object key (bareword `Ident` or double-quoted string) or an array
    /// index (non-negative integer). Segments are STRUCTURAL — the plan cache
    /// hashes them into the query shape and never treats them as literal slots.
    fn parse_json_path_postfix(&mut self, base: Expr) -> Result<Expr, ParseError> {
        if *self.peek() != Token::Arrow {
            return Ok(base);
        }
        match &base {
            Expr::Field(_) | Expr::QualifiedField { .. } | Expr::JsonPath { .. } => {}
            _ => {
                return Err(ParseError::Syntax {
                    message: "'->' JSON path access requires a field base \
                              (e.g. .data->key or posts.data->author)"
                        .into(),
                })
            }
        }
        let mut segments = Vec::new();
        while *self.peek() == Token::Arrow {
            self.advance(); // consume `->`
            let seg = match self.advance() {
                // Bareword key: `->author`.
                Token::Ident(name) => PathSeg::Key(name),
                // String-form key: `->"weird key!"` (PowQL strings are
                // double-quoted; the design's single-quote spelling maps here).
                Token::StringLit(s) => PathSeg::Key(s),
                // Array index: `->0`. Must be a non-negative integer that fits
                // a u32; the lexer produces a signed `IntLit`, so reject `< 0`
                // and overflow explicitly.
                Token::IntLit(v) => {
                    let idx = u32::try_from(v).map_err(|_| ParseError::Syntax {
                        message: format!(
                            "invalid JSON path array index {v}: expected a non-negative integer that fits in 32 bits"
                        ),
                    })?;
                    PathSeg::Index(idx)
                }
                other => {
                    return Err(ParseError::Syntax {
                        message: format!(
                            "expected a JSON path segment (object key or array index) after '->', found {}",
                            other.display_name()
                        ),
                    })
                }
            };
            segments.push(seg);
        }
        // Flatten a nested-JsonPath base into a single segment list so the AST
        // for `a->b->c` is one node regardless of how it was assembled.
        if let Expr::JsonPath {
            base: inner_base,
            segments: mut inner_segments,
        } = base
        {
            inner_segments.extend(segments);
            return Ok(Expr::JsonPath {
                base: inner_base,
                segments: inner_segments,
            });
        }
        Ok(Expr::JsonPath {
            base: Box::new(base),
            segments,
        })
    }

    fn parse_primary_inner(&mut self) -> Result<Expr, ParseError> {
        match self.peek().clone() {
            Token::DotIdent(name) => {
                self.advance();
                Ok(Expr::Field(name))
            }
            Token::IntLit(v) => {
                self.advance();
                Ok(Expr::Literal(Literal::Int(v)))
            }
            Token::FloatLit(v) => {
                self.advance();
                Ok(Expr::Literal(Literal::Float(v)))
            }
            Token::StringLit(v) => {
                self.advance();
                Ok(Expr::Literal(Literal::String(v)))
            }
            Token::BoolLit(v) => {
                self.advance();
                Ok(Expr::Literal(Literal::Bool(v)))
            }
            // `$N` placeholders are only valid through
            // `parse_with_params`, which substitutes them for literal
            // tokens before this expression parser ever runs. Reaching a
            // raw `Token::Param` here means the caller used the plain
            // (no-params) path with a placeholder — surface the standard
            // unexpected-token error so the message names the parameter.
            Token::Null => {
                self.advance();
                Ok(Expr::Null)
            }
            Token::Not => {
                self.advance();
                if *self.peek() == Token::Exists {
                    self.advance();
                    // `not exists (Q)` → ExistsSubquery{ negated: true } when
                    // followed by `( Ident ...` (subquery form). Otherwise
                    // fall back to the scalar `is not null` unary op.
                    if let Some(sub) = self.try_parse_exists_subquery()? {
                        return Ok(Expr::ExistsSubquery {
                            subquery: Box::new(sub),
                            negated: true,
                        });
                    }
                    let expr = self.parse_primary()?;
                    Ok(Expr::UnaryOp(UnaryOp::NotExists, Box::new(expr)))
                } else {
                    let expr = self.parse_primary()?;
                    Ok(Expr::UnaryOp(UnaryOp::Not, Box::new(expr)))
                }
            }
            Token::Exists => {
                self.advance();
                // `exists (Q)` → ExistsSubquery when followed by a
                // parenthesised query. Scalar `exists .field` still parses
                // as UnaryOp::Exists for backwards compatibility.
                if let Some(sub) = self.try_parse_exists_subquery()? {
                    return Ok(Expr::ExistsSubquery {
                        subquery: Box::new(sub),
                        negated: false,
                    });
                }
                let expr = self.parse_primary()?;
                Ok(Expr::UnaryOp(UnaryOp::Exists, Box::new(expr)))
            }
            Token::LParen => {
                self.advance();
                let expr = self.parse_expr()?;
                self.expect(&Token::RParen)?;
                Ok(expr)
            }
            Token::Ident(name) => {
                self.advance();
                // `uuid("…")` / `bytes("…")` cast sugar. `uuid`/`bytes` are not
                // lexer keywords (so `type T { id: uuid }` and identifiers named
                // `uuid` are untouched); an `Ident` immediately followed by `(`
                // with a matching name is single-argument cast sugar.
                if *self.peek() == Token::LParen {
                    let cast_type = match name.as_str() {
                        "uuid" => Some(CastType::Uuid),
                        "bytes" => Some(CastType::Bytes),
                        _ => None,
                    };
                    if let Some(cast_type) = cast_type {
                        self.advance(); // consume `(`
                        let inner = self.parse_expr()?;
                        self.expect(&Token::RParen)?;
                        return Ok(Expr::Cast(Box::new(inner), cast_type));
                    }
                }
                // `alias.field` → QualifiedField. The lexer emits `t1.name` as
                // `Ident("t1")` + `DotIdent("name")` (see lexer.rs line 30),
                // so a trailing DotIdent here means a qualified reference.
                if let Token::DotIdent(field) = self.peek().clone() {
                    self.advance();
                    return Ok(Expr::QualifiedField {
                        qualifier: name,
                        field,
                    });
                }
                Ok(Expr::Field(name))
            }
            // Window-only functions: row_number(), rank(), dense_rank()
            Token::RowNumber | Token::Rank | Token::DenseRank => {
                let wfunc = match self.advance() {
                    Token::RowNumber => WindowFunc::RowNumber,
                    Token::Rank => WindowFunc::Rank,
                    Token::DenseRank => WindowFunc::DenseRank,
                    _ => {
                        return Err(ParseError::Syntax {
                            message: "unexpected window function token".into(),
                        })
                    }
                };
                self.expect(&Token::LParen)?;
                self.expect(&Token::RParen)?;
                let (partition_by, order_by) = self.parse_over_clause()?;
                Ok(Expr::Window {
                    function: wfunc,
                    args: vec![],
                    mode: AggregateMode::Symmetric,
                    partition_by,
                    order_by,
                })
            }
            // Aggregate function calls inside expressions (projections, HAVING).
            // Top-level `count(User)` still routes through parse_aggregate_query
            // in parse_statement; this arm handles `count(.id)`, `sum(.age)`, etc.
            Token::Count | Token::Avg | Token::Sum | Token::Min | Token::Max => {
                let mut func = match self.advance() {
                    Token::Count => AggFunc::Count,
                    Token::Avg => AggFunc::Avg,
                    Token::Sum => AggFunc::Sum,
                    Token::Min => AggFunc::Min,
                    Token::Max => AggFunc::Max,
                    _ => {
                        return Err(ParseError::Syntax {
                            message: "unexpected aggregate token".into(),
                        })
                    }
                };
                self.expect(&Token::LParen)?;
                let mode = if *self.peek() == Token::Raw {
                    self.advance();
                    AggregateMode::Raw
                } else {
                    AggregateMode::Symmetric
                };
                // count(*) — count all rows including nulls
                if func == AggFunc::Count && *self.peek() == Token::Star {
                    self.advance();
                    self.expect(&Token::RParen)?;
                    // Check for OVER — count(*) over (...)
                    if *self.peek() == Token::Over {
                        let (partition_by, order_by) = self.parse_over_clause()?;
                        return Ok(Expr::Window {
                            function: WindowFunc::Count,
                            args: vec![Expr::Field("*".into())],
                            mode,
                            partition_by,
                            order_by,
                        });
                    }
                    return Ok(Expr::FunctionCall(
                        AggFunc::Count,
                        Box::new(Expr::Field("*".into())),
                        mode,
                    ));
                }
                // count(distinct .field) → CountDistinct
                if func == AggFunc::Count && *self.peek() == Token::Distinct {
                    self.advance();
                    func = AggFunc::CountDistinct;
                }
                let inner = self.parse_expr()?;
                self.expect(&Token::RParen)?;
                // Check for OVER — e.g. sum(.salary) over (...)
                if *self.peek() == Token::Over {
                    let wfunc = match func {
                        AggFunc::Count => WindowFunc::Count,
                        AggFunc::Avg => WindowFunc::Avg,
                        AggFunc::Sum => WindowFunc::Sum,
                        AggFunc::Min => WindowFunc::Min,
                        AggFunc::Max => WindowFunc::Max,
                        _ => {
                            return Err(ParseError::Unsupported {
                                feature: "count(distinct ...) over (...) is not supported".into(),
                            })
                        }
                    };
                    let (partition_by, order_by) = self.parse_over_clause()?;
                    return Ok(Expr::Window {
                        function: wfunc,
                        args: vec![inner],
                        mode,
                        partition_by,
                        order_by,
                    });
                }
                Ok(Expr::FunctionCall(func, Box::new(inner), mode))
            }
            Token::Upper
            | Token::Lower
            | Token::Length
            | Token::Trim
            | Token::Substring
            | Token::Concat
            | Token::Abs
            | Token::Round
            | Token::Ceil
            | Token::Floor
            | Token::Sqrt
            | Token::Pow
            | Token::Now
            | Token::Extract
            | Token::DateAdd
            | Token::DateDiff
            | Token::JsonType
            | Token::JsonText => {
                let tok = self.advance();
                let func = token_to_scalar_fn(&tok);
                self.expect(&Token::LParen)?;
                let mut args = Vec::new();
                while !matches!(self.peek(), Token::RParen | Token::Eof) {
                    args.push(self.parse_expr()?);
                    if *self.peek() == Token::Comma {
                        self.advance();
                    }
                }
                self.expect(&Token::RParen)?;
                Ok(Expr::ScalarFunc(func, args))
            }
            Token::Cast => {
                self.advance();
                self.expect(&Token::LParen)?;
                let inner = self.parse_expr()?;
                self.expect(&Token::Comma)?;
                let cast_type = self.parse_cast_type()?;
                self.expect(&Token::RParen)?;
                Ok(Expr::Cast(Box::new(inner), cast_type))
            }
            Token::Case => {
                self.advance();
                let mut whens = Vec::new();
                while *self.peek() == Token::When {
                    self.advance();
                    let condition = self.parse_expr()?;
                    self.expect(&Token::Then)?;
                    let result = self.parse_expr()?;
                    whens.push((Box::new(condition), Box::new(result)));
                }
                let else_expr = if *self.peek() == Token::Else {
                    self.advance();
                    Some(Box::new(self.parse_expr()?))
                } else {
                    None
                };
                self.expect(&Token::End)?;
                Ok(Expr::Case { whens, else_expr })
            }
            t => Err(ParseError::Syntax {
                message: format!("unexpected token in expression: {}", t.display_name()),
            }),
        }
    }

    /// `alter <Table> add [column] [required] <name>: <type>`
    /// `alter <Table> drop [column] <name>`
    /// Parse a bare link declaration:
    /// `link <Owner>.<name> -> <Target> on <local> = <target>`. Called with
    /// the cursor on the leading `link` token. Lowers to
    /// `Statement::CreateLink`, which the executor routes to
    /// `Catalog::create_link`.
    fn parse_create_link(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Link)?;
        let owner = self.expect_named_ident("link owner type")?;
        let name = match self.advance() {
            Token::DotIdent(n) => n,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "`.<name>` after the owner type (link <Owner>.<name> -> ...)".into(),
                    got: t.display_name(),
                })
            }
        };
        let (target, local_key, target_key) = self.parse_link_tail()?;
        Ok(Statement::CreateLink(CreateLinkExpr {
            owner,
            name,
            target,
            local_key,
            target_key,
        }))
    }

    /// Parse the shared tail of a link declaration after its name:
    /// `-> <Target> on <local> = <target>`. Returns `(target, local, target)`.
    fn parse_link_tail(&mut self) -> Result<(String, String, String), ParseError> {
        self.expect(&Token::Arrow)?;
        let target = self.expect_named_ident("link target type")?;
        self.expect(&Token::On)?;
        let local_key = self.parse_link_column("link local key")?;
        self.expect(&Token::Eq)?;
        let target_key = self.parse_link_column("link target key")?;
        Ok((target, local_key, target_key))
    }

    /// Read a column name written either bare (`user_id`) or dot-prefixed
    /// (`.user_id`), as used in a link's correlation clause.
    fn parse_link_column(&mut self, context: &str) -> Result<String, ParseError> {
        match self.advance() {
            Token::Ident(n) | Token::DotIdent(n) => Ok(n),
            t => Err(self.named_ident_error(context, &t)),
        }
    }

    fn parse_alter_table(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Alter)?;
        let table = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "table name after alter".into(),
                    got: t.display_name(),
                })
            }
        };
        match self.peek() {
            Token::Add => {
                self.advance();
                // `alter <Table> add index [if not exists] <target>`
                if *self.peek() == Token::Index {
                    self.advance();
                    let if_not_exists = self.parse_optional_if_not_exists();
                    let target = self.parse_index_target("add index")?;
                    return Ok(Statement::AlterTable(AlterTableExpr {
                        table,
                        action: AlterAction::AddIndex {
                            target,
                            if_not_exists,
                        },
                    }));
                }
                // `alter <Table> add unique [if not exists] <target>`
                if *self.peek() == Token::Unique {
                    self.advance();
                    let if_not_exists = self.parse_optional_if_not_exists();
                    let target = self.parse_index_target("add unique")?;
                    return Ok(Statement::AlterTable(AlterTableExpr {
                        table,
                        action: AlterAction::AddUnique {
                            target,
                            if_not_exists,
                        },
                    }));
                }
                // `alter <Owner> add link <name> -> <Target> on <local> = <target>`
                if *self.peek() == Token::Link {
                    self.advance();
                    let name = self.expect_named_ident("link name")?;
                    let (target, local_key, target_key) = self.parse_link_tail()?;
                    return Ok(Statement::AlterTable(AlterTableExpr {
                        table,
                        action: AlterAction::AddLink {
                            name,
                            target,
                            local_key,
                            target_key,
                        },
                    }));
                }
                // optional `column` keyword
                if *self.peek() == Token::Column {
                    self.advance();
                }
                let required = if *self.peek() == Token::Required {
                    self.advance();
                    true
                } else {
                    false
                };
                let name = self.expect_named_ident("column name")?;
                self.expect(&Token::Colon)?;
                let type_name = match self.advance() {
                    Token::Ident(n) => n,
                    t => {
                        return Err(ParseError::UnexpectedToken {
                            expected: "type name".into(),
                            got: t.display_name(),
                        })
                    }
                };
                Ok(Statement::AlterTable(AlterTableExpr {
                    table,
                    action: AlterAction::AddColumn {
                        name,
                        type_name,
                        required,
                    },
                }))
            }
            Token::Drop => {
                self.advance();
                if *self.peek() == Token::Index {
                    self.advance();
                    let if_exists = self.parse_optional_if_exists();
                    let target = self.parse_index_target("drop index")?;
                    return Ok(Statement::AlterTable(AlterTableExpr {
                        table,
                        action: AlterAction::DropIndex { target, if_exists },
                    }));
                }
                // optional `column` keyword
                if *self.peek() == Token::Column {
                    self.advance();
                }
                let if_exists = self.parse_optional_if_exists();
                let name = self.expect_named_ident("column name")?;
                Ok(Statement::AlterTable(AlterTableExpr {
                    table,
                    action: AlterAction::DropColumn { name, if_exists },
                }))
            }
            t => Err(ParseError::UnexpectedToken {
                expected: "add or drop after alter <table>".into(),
                got: t.display_name(),
            }),
        }
    }

    /// Parse a column target (`.slug`) or a parenthesized, unqualified JSON
    /// path target (`(.data->slug)`) for ALTER INDEX actions. Parentheses are
    /// deliberately the syntax boundary between stored-column and expression
    /// indexes so future expression forms cannot silently change old DDL.
    fn parse_index_target(&mut self, action: &str) -> Result<IndexTarget, ParseError> {
        match self.peek() {
            Token::DotIdent(_) => {
                if matches!(self.tokens.get(self.pos + 1), Some(Token::Arrow)) {
                    return Err(ParseError::Syntax {
                        message: format!(
                            "JSON path index targets must be parenthesized after {action}; use `(.data->key)`"
                        ),
                    });
                }
                let Token::DotIdent(column) = self.advance() else {
                    unreachable!("guarded by DotIdent match")
                };
                Ok(IndexTarget::Column(column))
            }
            Token::LParen => {
                self.advance();
                let expr = self.parse_expr().map_err(|error| match error {
                    ParseError::NestingDepthExceeded { .. } => error,
                    _ => ParseError::Syntax {
                        message: format!(
                            "invalid expression index target after {action}: expected an unqualified JSON path like `(.data->key)`"
                        ),
                    },
                })?;
                if *self.peek() != Token::RParen {
                    return Err(ParseError::Syntax {
                        message: format!(
                            "invalid expression index target after {action}: only a direct JSON path is supported"
                        ),
                    });
                }
                self.advance();

                match JsonPathIdentityV1::from_expr(&expr) {
                    Some(identity) => identity.bind_table_local(None).map(IndexTarget::JsonPath).ok_or_else(|| {
                        ParseError::Syntax {
                            message: format!(
                                "qualified JSON paths are not valid index targets after {action}; use an unqualified table-local path like `(.data->key)`"
                            ),
                        }
                    }),
                    None => match expr {
                        Expr::Field(_) => Err(ParseError::Syntax {
                            message: format!(
                                "invalid expression index target after {action}: parentheses are reserved for a direct JSON path like `(.data->key)`; use `.column` for a stored column"
                            ),
                        }),
                        Expr::QualifiedField { .. } => Err(ParseError::Syntax {
                            message: format!(
                                "qualified references are not valid index targets after {action}; use a table-local `.column` or `(.data->key)`"
                            ),
                        }),
                        _ => Err(ParseError::Syntax {
                            message: format!(
                                "invalid expression index target after {action}: only a direct JSON path is supported"
                            ),
                        }),
                    },
                }
            }
            token => Err(ParseError::UnexpectedToken {
                expected: format!(".<column> or parenthesized JSON path after {action}"),
                got: token.display_name(),
            }),
        }
    }

    /// `drop [if exists] <Table>` or `drop view [if exists] <ViewName>`
    fn parse_drop_or_drop_view(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Drop)?;
        if *self.peek() == Token::View {
            self.advance(); // consume `view`
            let if_exists = self.parse_optional_if_exists();
            let name = match self.advance() {
                Token::Ident(name) => name,
                t => {
                    return Err(ParseError::UnexpectedToken {
                        expected: "view name after drop view".into(),
                        got: t.display_name(),
                    })
                }
            };
            return Ok(Statement::DropView(DropViewExpr { name, if_exists }));
        }
        let if_exists = self.parse_optional_if_exists();
        let table = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "table name after drop".into(),
                    got: t.display_name(),
                })
            }
        };
        Ok(Statement::DropTable(DropTableExpr { table, if_exists }))
    }

    /// `materialize <ViewName> as <Query>`
    ///
    /// The source query text is captured by slicing the original token stream
    /// from the position after `as` to the end.
    fn parse_create_view(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Materialized)?;
        let name = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "view name after materialize".into(),
                    got: t.display_name(),
                })
            }
        };
        self.expect(&Token::As)?;
        // Record position so we can reconstruct the query text for storage.
        let query_start = self.pos;
        let source = match self.advance() {
            Token::Ident(s) => s,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "source table name".into(),
                    got: t.display_name(),
                })
            }
        };
        let query = self.parse_query_tail(source)?;
        // Reconstruct query text from tokens for storage and re-execution.
        let query_text = tokens_to_text(&self.tokens[query_start..self.pos])?;
        Ok(Statement::CreateView(CreateViewExpr {
            name,
            query,
            query_text,
        }))
    }

    /// Check for `union [all]` after a query and build a left-associative
    /// chain if present.
    fn maybe_parse_union(&mut self, left: Statement) -> Result<Statement, ParseError> {
        if *self.peek() != Token::Union {
            return Ok(left);
        }
        if !matches!(left, Statement::Query(_) | Statement::Union(_)) {
            return Err(ParseError::Syntax {
                message: "UNION requires a query on the left side".into(),
            });
        }
        self.advance(); // consume `union`
        let all = if let Token::Ident(s) = self.peek() {
            if s == "all" {
                self.advance();
                true
            } else {
                false
            }
        } else {
            false
        };
        // Parse the RHS as a single query (not chained — we'll chain ourselves).
        let right = self.parse_single_query()?;
        let union = Statement::Union(UnionExpr {
            left: Box::new(left),
            right: Box::new(right),
            all,
        });
        // Recursively check for further chaining: `A union B union C`
        self.maybe_parse_union(union)
    }

    /// Parse a single query statement (no UNION chaining). Used for UNION RHS.
    fn parse_single_query(&mut self) -> Result<Statement, ParseError> {
        match self.peek() {
            Token::Count | Token::Avg | Token::Sum | Token::Min | Token::Max => {
                self.parse_aggregate_query()
            }
            Token::Ident(_) => self.parse_query_or_mutation(),
            _ => Err(ParseError::Syntax {
                message: format!(
                    "expected query after UNION, got {}",
                    self.peek().display_name()
                ),
            }),
        }
    }

    /// `refresh <ViewName>`
    fn parse_refresh_view(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Refresh)?;
        let name = match self.advance() {
            Token::Ident(name) => name,
            t => {
                return Err(ParseError::UnexpectedToken {
                    expected: "view name after refresh".into(),
                    got: t.display_name(),
                })
            }
        };
        Ok(Statement::RefreshView(RefreshViewExpr { name }))
    }

    fn parse_create_type(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Type)?;
        let name = self.expect_named_ident("type name")?;
        let if_not_exists = self.parse_optional_if_not_exists();
        self.expect(&Token::LBrace)?;
        let mut fields = Vec::new();
        while !matches!(self.peek(), Token::RBrace | Token::Eof) {
            // Accept `required`, `unique`, and `auto` modifiers in any order.
            // A modifier keyword immediately followed by `:` is instead the
            // field's *name* (e.g. `required: int`) — leave it for
            // `expect_named_ident`, which emits the reserved-word guidance.
            let (mut required, mut unique, mut auto) = (false, false, false);
            loop {
                let is_modifier =
                    matches!(self.peek(), Token::Required | Token::Unique | Token::Auto)
                        && !matches!(self.tokens.get(self.pos + 1), Some(Token::Colon));
                if !is_modifier {
                    break;
                }
                match self.advance() {
                    Token::Required => required = true,
                    Token::Unique => unique = true,
                    Token::Auto => auto = true,
                    _ => unreachable!("guarded by is_modifier"),
                }
            }
            let field_name = self.expect_named_ident("field name")?;
            self.expect(&Token::Colon)?;
            let type_name = match self.advance() {
                Token::Ident(n) => n,
                t => {
                    return Err(ParseError::UnexpectedToken {
                        expected: "type name".into(),
                        got: t.display_name(),
                    })
                }
            };
            // Optional `default <literal>` — value applied when an insert
            // omits this column.
            let default = if *self.peek() == Token::Default {
                self.advance();
                Some(self.parse_default_literal()?)
            } else {
                None
            };
            fields.push(FieldDef {
                name: field_name,
                type_name,
                required,
                unique,
                default,
                auto,
            });
            if *self.peek() == Token::Comma {
                self.advance();
            }
        }
        self.expect(&Token::RBrace)?;
        Ok(Statement::CreateType(CreateTypeExpr {
            name,
            fields,
            if_not_exists,
        }))
    }

    /// `schema`: list all types. `schema links` lists every declared entity
    /// link. Any other `schema <Type>` is an alias for `describe <Type>`.
    fn parse_schema(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Schema)?;
        if let Token::Ident(name) = self.peek() {
            if name == "links" {
                self.advance();
                return Ok(Statement::ListLinks);
            }
            let table = self.expect_named_ident("type name")?;
            return Ok(Statement::Describe(table));
        }
        Ok(Statement::ListTypes)
    }

    /// `describe <Type>` — the columns and indexes of one type.
    fn parse_describe(&mut self) -> Result<Statement, ParseError> {
        self.expect(&Token::Describe)?;
        let table = self.expect_named_ident("type name")?;
        Ok(Statement::Describe(table))
    }

    /// Parse the literal following a `default` column modifier. Only scalar
    /// literals are allowed — expression defaults (e.g. `now()`) are not yet
    /// supported.
    fn parse_default_literal(&mut self) -> Result<Literal, ParseError> {
        match self.advance() {
            Token::IntLit(v) => Ok(Literal::Int(v)),
            Token::FloatLit(v) => Ok(Literal::Float(v)),
            Token::StringLit(v) => Ok(Literal::String(v)),
            Token::BoolLit(v) => Ok(Literal::Bool(v)),
            t => Err(ParseError::UnexpectedToken {
                expected: "literal default value".into(),
                got: t.display_name(),
            }),
        }
    }
}

/// Rewrite every bare `Expr::Field(f)` in `expr` into
/// `Expr::QualifiedField { qualifier: alias, field: f }`, so a block link
/// traversal's bare child columns (`total`) match the qualified spelling
/// (`o.total`) the planner expects. Descends through operators but stops at a
/// nested query (its own scope owns its qualification) and leaves already
/// qualified references untouched.
fn qualify_bare_fields(expr: Expr, alias: &str) -> Expr {
    let recur = |e: Expr| Box::new(qualify_bare_fields(e, alias));
    match expr {
        Expr::Field(field) => Expr::QualifiedField {
            qualifier: alias.to_string(),
            field,
        },
        Expr::BinaryOp(l, op, r) => Expr::BinaryOp(recur(*l), op, recur(*r)),
        Expr::UnaryOp(op, inner) => Expr::UnaryOp(op, recur(*inner)),
        Expr::Coalesce(l, r) => Expr::Coalesce(recur(*l), recur(*r)),
        Expr::Cast(inner, ty) => Expr::Cast(recur(*inner), ty),
        Expr::ScalarFunc(func, args) => Expr::ScalarFunc(
            func,
            args.into_iter()
                .map(|a| qualify_bare_fields(a, alias))
                .collect(),
        ),
        Expr::InList {
            expr,
            list,
            negated,
        } => Expr::InList {
            expr: recur(*expr),
            list: list
                .into_iter()
                .map(|a| qualify_bare_fields(a, alias))
                .collect(),
            negated,
        },
        Expr::Case { whens, else_expr } => Expr::Case {
            whens: whens
                .into_iter()
                .map(|(c, r)| (recur(*c), recur(*r)))
                .collect(),
            else_expr: else_expr.map(|e| recur(*e)),
        },
        Expr::JsonPath { base, segments } => Expr::JsonPath {
            base: recur(*base),
            segments,
        },
        // Leaves and cross-scope nodes are left as-is.
        other => other,
    }
}

/// True when `text` lexes to exactly the one token `tok` (plus EOF).
///
/// Lets the writers below *ask the lexer* which spelling round-trips instead
/// of re-deriving its rules, so the two can never drift apart.
fn relexes_to(text: &str, tok: &Token) -> bool {
    match lex(text) {
        Ok(toks) => matches!(toks.as_slice(), [t, Token::Eof] if t == tok),
        Err(_) => false,
    }
}

/// Write a string literal in source form, escaping what the lexer decodes.
///
/// Exact inverse of the lexer's string rule: it turns `\"`, `\\`, `\n` and
/// `\t` into `"`, `\`, LF and TAB, copies every other character through
/// verbatim, and swallows the backslash of any other escape (`\r` decodes to
/// `r`). So those four are the only escapes that may be emitted, and every
/// other character (CR included) must be written raw.
fn push_string_literal(out: &mut String, s: &str) {
    out.push('"');
    for c in s.chars() {
        match c {
            '"' => out.push_str("\\\""),
            '\\' => out.push_str("\\\\"),
            '\n' => out.push_str("\\n"),
            '\t' => out.push_str("\\t"),
            other => out.push(other),
        }
    }
    out.push('"');
}

/// Write an identifier (`prefix` is `""` for a bare name, `"."` for a field
/// reference) in whichever spelling re-lexes to `tok` itself: bare when the
/// lexer reads it back as this exact token, backtick-quoted otherwise.
///
/// Quoting is not cosmetic. `` `order` `` lexes to `Ident("order")`; written
/// back bare it re-lexes to the *keyword* `order`, and the stored view runs a
/// different query. Same for names with spaces, leading digits, or symbols.
fn push_ident(out: &mut String, prefix: &str, name: &str, tok: &Token) -> Result<(), ParseError> {
    let bare = format!("{prefix}{name}");
    if relexes_to(&bare, tok) {
        out.push_str(&bare);
        return Ok(());
    }
    let quoted = format!("{prefix}`{name}`");
    if relexes_to(&quoted, tok) {
        out.push_str(&quoted);
        return Ok(());
    }
    Err(ParseError::Unsupported {
        feature: format!(
            "cannot store view source: identifier '{name}' has no PowQL spelling that reads back unchanged"
        ),
    })
}

/// Reconstruct PowQL source text from a slice of tokens. Used to store the
/// view's source query for re-execution on refresh.
///
/// Whitespace is normalised, but the token stream is not: re-lexing the result
/// yields exactly the tokens passed in. That is the whole contract. A
/// reconstruction that re-lexes to *different* tokens makes the stored view run
/// a different query than the user wrote, with no error anywhere, so the result
/// is verified against the lexer before it is returned and a token with no
/// faithful spelling is a typed error rather than a quiet mismatch.
fn tokens_to_text(tokens: &[Token]) -> Result<String, ParseError> {
    let mut out = String::with_capacity(64);
    for tok in tokens {
        if !out.is_empty() && !matches!(tok, Token::Eof) {
            out.push(' ');
        }
        match tok {
            Token::Ident(s) => push_ident(&mut out, "", s, tok)?,
            Token::DotIdent(s) => push_ident(&mut out, ".", s, tok)?,
            Token::IntLit(v) => out.push_str(&v.to_string()),
            Token::FloatLit(v) => {
                if !v.is_finite() {
                    return Err(ParseError::Unsupported {
                        feature: "cannot store view source: non-finite number literal".into(),
                    });
                }
                let rendered = v.to_string();
                out.push_str(&rendered);
                // `Display` drops a redundant fraction (`3.0` prints as `3`,
                // `-0.0` as `-0`), and the lexer only reads a float when a
                // digit follows a dot, so without this the literal comes back
                // as an INTEGER token.
                if !rendered.contains('.') {
                    out.push_str(".0");
                }
            }
            Token::StringLit(s) => push_string_literal(&mut out, s),
            Token::BoolLit(v) => out.push_str(if *v { "true" } else { "false" }),
            Token::Param(s) => {
                out.push('$');
                out.push_str(s);
            }
            Token::Type => out.push_str("type"),
            Token::Filter => out.push_str("filter"),
            Token::Order => out.push_str("order"),
            Token::Limit => out.push_str("limit"),
            Token::Offset => out.push_str("offset"),
            Token::Insert => out.push_str("insert"),
            Token::Update => out.push_str("update"),
            Token::Delete => out.push_str("delete"),
            Token::Upsert => out.push_str("upsert"),
            Token::Returning => out.push_str("returning"),
            Token::Conflict => out.push_str("conflict"),
            Token::Select => out.push_str("select"),
            Token::Required => out.push_str("required"),
            Token::Default => out.push_str("default"),
            Token::Auto => out.push_str("auto"),
            Token::Multi => out.push_str("multi"),
            Token::Link => out.push_str("link"),
            Token::Index => out.push_str("index"),
            Token::Unique => out.push_str("unique"),
            Token::On => out.push_str("on"),
            Token::Asc => out.push_str("asc"),
            Token::Desc => out.push_str("desc"),
            Token::And => out.push_str("and"),
            Token::Or => out.push_str("or"),
            Token::Not => out.push_str("not"),
            Token::Exists => out.push_str("exists"),
            Token::Let => out.push_str("let"),
            Token::As => out.push_str("as"),
            Token::Match => out.push_str("match"),
            Token::Group => out.push_str("group"),
            Token::Join => out.push_str("join"),
            Token::Inner => out.push_str("inner"),
            Token::LeftKw => out.push_str("left"),
            Token::RightKw => out.push_str("right"),
            Token::Outer => out.push_str("outer"),
            Token::Cross => out.push_str("cross"),
            Token::Transaction => out.push_str("transaction"),
            Token::Begin => out.push_str("begin"),
            Token::Commit => out.push_str("commit"),
            Token::Rollback => out.push_str("rollback"),
            Token::View => out.push_str("view"),
            Token::Materialized => out.push_str("materialized"),
            Token::Refresh => out.push_str("refresh"),
            Token::Union => out.push_str("union"),
            Token::Having => out.push_str("having"),
            Token::Distinct => out.push_str("distinct"),
            Token::In => out.push_str("in"),
            Token::Between => out.push_str("between"),
            Token::Like => out.push_str("like"),
            Token::Count => out.push_str("count"),
            Token::Avg => out.push_str("avg"),
            Token::Sum => out.push_str("sum"),
            Token::Raw => out.push_str("raw"),
            Token::Min => out.push_str("min"),
            Token::Max => out.push_str("max"),
            Token::Is => out.push_str("is"),
            Token::Null => out.push_str("null"),
            Token::Upper => out.push_str("upper"),
            Token::Lower => out.push_str("lower"),
            Token::Length => out.push_str("length"),
            Token::Trim => out.push_str("trim"),
            Token::Substring => out.push_str("substring"),
            Token::Concat => out.push_str("concat"),
            Token::Abs => out.push_str("abs"),
            Token::Round => out.push_str("round"),
            Token::Ceil => out.push_str("ceil"),
            Token::Floor => out.push_str("floor"),
            Token::Sqrt => out.push_str("sqrt"),
            Token::Pow => out.push_str("pow"),
            Token::Now => out.push_str("now"),
            Token::Extract => out.push_str("extract"),
            Token::DateAdd => out.push_str("date_add"),
            Token::DateDiff => out.push_str("date_diff"),
            Token::JsonType => out.push_str("json_type"),
            Token::JsonText => out.push_str("json_text"),
            Token::Cast => out.push_str("cast"),
            Token::Case => out.push_str("case"),
            Token::When => out.push_str("when"),
            Token::Then => out.push_str("then"),
            Token::Else => out.push_str("else"),
            Token::End => out.push_str("end"),
            Token::Over => out.push_str("over"),
            Token::Partition => out.push_str("partition"),
            Token::RowNumber => out.push_str("row_number"),
            Token::Rank => out.push_str("rank"),
            Token::DenseRank => out.push_str("dense_rank"),
            Token::Alter => out.push_str("alter"),
            Token::Drop => out.push_str("drop"),
            Token::Add => out.push_str("add"),
            Token::Column => out.push_str("column"),
            Token::Eq => out.push('='),
            Token::Neq => out.push_str("!="),
            Token::Lt => out.push('<'),
            Token::Gt => out.push('>'),
            Token::Lte => out.push_str("<="),
            Token::Gte => out.push_str(">="),
            Token::Assign => out.push_str(":="),
            Token::Arrow => out.push_str("->"),
            Token::Pipe => out.push('|'),
            Token::Coalesce => out.push_str("??"),
            Token::Plus => out.push('+'),
            Token::Minus => out.push('-'),
            Token::Star => out.push('*'),
            Token::Slash => out.push('/'),
            Token::LBrace => out.push('{'),
            Token::RBrace => out.push('}'),
            Token::LParen => out.push('('),
            Token::RParen => out.push(')'),
            Token::Comma => out.push(','),
            Token::Colon => out.push(':'),
            Token::Dot => out.push('.'),
            Token::Explain => out.push_str("explain"),
            Token::Schema => out.push_str("schema"),
            Token::Describe => out.push_str("describe"),
            Token::Eof => {}
        }
    }
    // Verify rather than trust. Every arm above is meant to be the lexer's
    // inverse, but a wrong one is invisible: it stores a query that merely
    // *looks* like the user's. Re-lex the result and require the same stream
    // back, so any arm that is (or later becomes) wrong fails loudly here
    // instead of silently redefining someone's view. EOF is dropped, since a
    // slice of the middle of a stream never carries one.
    let mut expected: Vec<Token> = tokens
        .iter()
        .filter(|t| **t != Token::Eof)
        .cloned()
        .collect();
    expected.push(Token::Eof);
    match lex(&out) {
        Ok(round_tripped) if round_tripped == expected => Ok(out),
        _ => Err(ParseError::Unsupported {
            feature: "cannot store view source: query text does not read back unchanged".into(),
        }),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn test_parse_simple_query() {
        let stmt = parse("User").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.source, "User");
                assert!(q.filter.is_none());
                assert!(q.projection.is_none());
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_filter() {
        let stmt = parse("User filter .age > 30").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.source, "User");
                assert!(q.filter.is_some());
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_projection() {
        let stmt = parse("User { name, email }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 2);
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_bare_dotted_path_projection_is_a_parse_error() {
        // `.user.name` without an outer alias is token-identical to two
        // comma-less fields, so it is rejected with alias guidance instead of
        // silently parsing as two fields (2026-07-23 plan-quality audit P3).
        for q in [
            "Order { .user.name }",
            "Order { .id, uname: .user.name }",
            "Order { uname: .user.company.name }",
        ] {
            let err = parse(q).unwrap_err();
            let msg = err.to_string();
            assert!(
                msg.contains("alias the table"),
                "`{q}` should error with alias guidance, got: {msg}"
            );
        }
    }

    #[test]
    fn test_aliased_scalar_link_path_parses_as_one_field() {
        let stmt = parse("Order as o { uname: o.user.name }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 1);
                match &proj[0].expr {
                    Expr::LinkPath {
                        outer_alias,
                        links,
                        column,
                    } => {
                        assert_eq!(outer_alias, "o");
                        assert_eq!(links, &["user".to_string()]);
                        assert_eq!(column, "name");
                    }
                    other => panic!("expected LinkPath, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_filter_order_limit() {
        let stmt = parse("User filter .age > 30 order .name desc limit 10").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert!(q.filter.is_some());
                let order = q.order.unwrap();
                assert_eq!(order.keys.len(), 1);
                assert_eq!(order.keys[0].expr, Expr::Field("name".into()));
                assert!(order.keys[0].descending);
                assert!(q.limit.is_some());
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_insert() {
        let stmt = parse(r#"insert User { name := "Alice", age := 30 }"#).unwrap();
        match stmt {
            Statement::Insert(ins) => {
                assert_eq!(ins.target, "User");
                assert_eq!(ins.rows.len(), 1);
                assert_eq!(ins.rows[0].len(), 2);
                assert_eq!(ins.rows[0][0].field, "name");
                assert_eq!(ins.rows[0][1].field, "age");
            }
            _ => panic!("expected insert"),
        }
    }

    #[test]
    fn test_parse_insert_multi_row() {
        let stmt =
            parse(r#"insert User { name := "Alice", age := 30 }, { name := "Bob", age := 25 }, { name := "Cy" }"#)
                .unwrap();
        match stmt {
            Statement::Insert(ins) => {
                assert_eq!(ins.target, "User");
                assert_eq!(ins.rows.len(), 3);
                assert_eq!(ins.rows[0].len(), 2);
                assert_eq!(ins.rows[1][0].field, "name");
                assert_eq!(ins.rows[2].len(), 1);
                assert_eq!(ins.rows[2][0].field, "name");
            }
            _ => panic!("expected insert"),
        }
    }

    #[test]
    fn test_parse_update() {
        let stmt = parse(r#"User filter .email = "alice@ex.com" update { age := 31 }"#).unwrap();
        match stmt {
            Statement::UpdateQuery(upd) => {
                assert_eq!(upd.source, "User");
                assert!(upd.filter.is_some());
                assert_eq!(upd.assignments.len(), 1);
                assert!(!upd.returning);
            }
            _ => panic!("expected update"),
        }
    }

    #[test]
    fn test_parse_update_returning() {
        let stmt = parse(r#"User filter .name = "Alice" update { age := 31 } returning"#).unwrap();
        match stmt {
            Statement::UpdateQuery(upd) => assert!(upd.returning),
            _ => panic!("expected update"),
        }
    }

    #[test]
    fn test_parse_delete() {
        let stmt = parse("User filter .age < 18 delete").unwrap();
        match stmt {
            Statement::DeleteQuery(del) => {
                assert_eq!(del.source, "User");
                assert!(del.filter.is_some());
                assert!(!del.returning);
            }
            _ => panic!("expected delete"),
        }
    }

    #[test]
    fn test_parse_delete_returning() {
        let stmt = parse("User filter .age < 18 delete returning").unwrap();
        match stmt {
            Statement::DeleteQuery(del) => assert!(del.returning),
            _ => panic!("expected delete"),
        }
    }

    #[test]
    fn test_parse_count() {
        let stmt = parse("count(User)").unwrap();
        match stmt {
            Statement::Query(q) => {
                let agg = q.aggregation.unwrap();
                assert_eq!(agg.function, AggFunc::Count);
                assert!(q.filter.is_none());
            }
            _ => panic!("expected query with aggregation"),
        }
    }

    #[test]
    fn test_parse_count_with_filter() {
        // Regression: previously returned "expected RParen, got Filter".
        // count(<query>) must accept a full read-pipeline tail.
        let stmt = parse("count(User filter .age > 30)").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.source, "User");
                let agg = q.aggregation.unwrap();
                assert_eq!(agg.function, AggFunc::Count);
                assert!(q.filter.is_some(), "filter should have been parsed");
            }
            _ => panic!("expected query with aggregation"),
        }
    }

    #[test]
    fn test_parse_count_with_filter_and_limit() {
        let stmt = parse("count(User filter .age > 30 limit 100)").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.source, "User");
                assert!(q.filter.is_some());
                assert!(q.limit.is_some());
                assert_eq!(q.aggregation.unwrap().function, AggFunc::Count);
            }
            _ => panic!("expected query with aggregation"),
        }
    }

    #[test]
    fn test_parse_create_type() {
        let stmt = parse("type User { required name: str, age: int }").unwrap();
        match stmt {
            Statement::CreateType(ct) => {
                assert_eq!(ct.name, "User");
                assert_eq!(ct.fields.len(), 2);
                assert!(ct.fields[0].required);
                assert!(!ct.fields[1].required);
            }
            _ => panic!("expected create type"),
        }
    }

    #[test]
    fn test_parse_sum_with_field_projection() {
        // `sum(... { .age })` should lift `.age` into AggregateExpr.argument and
        // clear the projection so the executor's aggregate fast path fires.
        let stmt = parse("sum(User filter .age > 30 { .age })").unwrap();
        match stmt {
            Statement::Query(q) => {
                let agg = q.aggregation.expect("aggregate");
                assert_eq!(agg.function, AggFunc::Sum);
                assert_eq!(agg.argument, Some(Expr::Field("age".into())));
                assert!(
                    q.projection.is_none(),
                    "projection should be lifted into agg.field"
                );
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_raw_aggregate_modes() {
        let Statement::Query(top_level) = parse("sum(raw User { .age })").unwrap() else {
            panic!("expected query");
        };
        assert_eq!(top_level.aggregation.unwrap().mode, AggregateMode::Raw);

        let Statement::Query(grouped) = parse("User group .dept { total: sum(raw .age) }").unwrap()
        else {
            panic!("expected query");
        };
        assert!(matches!(
            grouped.projection.unwrap()[0].expr,
            Expr::FunctionCall(AggFunc::Sum, _, AggregateMode::Raw)
        ));
    }

    #[test]
    fn test_parse_avg_min_max_with_field() {
        for (src, expected) in [
            ("avg(User { .age })", AggFunc::Avg),
            ("min(User { .age })", AggFunc::Min),
            ("max(User { .age })", AggFunc::Max),
        ] {
            let stmt = parse(src).unwrap();
            match stmt {
                Statement::Query(q) => {
                    let agg = q.aggregation.unwrap();
                    assert_eq!(agg.function, expected, "func mismatch for {src}");
                    assert_eq!(
                        agg.argument,
                        Some(Expr::Field("age".into())),
                        "field mismatch for {src}"
                    );
                    assert!(
                        q.projection.is_none(),
                        "projection should be cleared for {src}"
                    );
                }
                _ => panic!("expected query for {src}"),
            }
        }
    }

    #[test]
    fn test_parse_count_lifts_projection_into_argument() {
        // A projected column names what to count: `count(User { .age })` is a
        // non-null count of `.age`, matching the grouped `count(.age)` path and
        // SQL's `COUNT(age)`. Leaving the projection in place (the previous
        // behavior) silently made it a row count.
        let stmt = parse("count(User { .age })").unwrap();
        match stmt {
            Statement::Query(q) => {
                let agg = q.aggregation.unwrap();
                assert_eq!(agg.function, AggFunc::Count);
                assert_eq!(agg.argument, Some(Expr::Field("age".into())));
                assert!(q.projection.is_none(), "projection should be lifted");
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_count_without_projection_has_no_argument() {
        // `count(User)` stays a row count.
        let stmt = parse("count(User)").unwrap();
        match stmt {
            Statement::Query(q) => {
                let agg = q.aggregation.unwrap();
                assert_eq!(agg.function, AggFunc::Count);
                assert!(agg.argument.is_none());
            }
            _ => panic!("expected query"),
        }
    }

    // ---- Mission E1.1: JOIN parser tests ----------------------------------
    // Parser-level only. The planner rejects joins with a clean error until
    // E1.2 wires up execution.

    #[test]
    fn test_parse_source_alias() {
        let stmt = parse("User as u filter u.age > 30").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.source, "User");
                assert_eq!(q.alias.as_deref(), Some("u"));
                assert!(q.joins.is_empty());
                match q.filter.unwrap() {
                    Expr::BinaryOp(l, BinOp::Gt, _) => match *l {
                        Expr::QualifiedField { qualifier, field } => {
                            assert_eq!(qualifier, "u");
                            assert_eq!(field, "age");
                        }
                        other => panic!("expected qualified field, got {other:?}"),
                    },
                    other => panic!("expected >, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_inner_join_on() {
        let stmt = parse("User as u inner join Order as o on u.id = o.user_id").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.source, "User");
                assert_eq!(q.alias.as_deref(), Some("u"));
                assert_eq!(q.joins.len(), 1);
                let j = &q.joins[0];
                assert_eq!(j.kind, JoinKind::Inner);
                assert_eq!(j.source, "Order");
                assert_eq!(j.alias.as_deref(), Some("o"));
                let on = j.on.as_ref().expect("on clause");
                match on {
                    Expr::BinaryOp(l, BinOp::Eq, r) => {
                        assert!(matches!(**l, Expr::QualifiedField { .. }));
                        assert!(matches!(**r, Expr::QualifiedField { .. }));
                    }
                    other => panic!("expected eq, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_bare_join_defaults_to_inner() {
        let stmt = parse("User join Order on User.id = Order.user_id").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.joins.len(), 1);
                assert_eq!(q.joins[0].kind, JoinKind::Inner);
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_left_outer_join() {
        let stmt = parse("User as u left outer join Order as o on u.id = o.user_id").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.joins.len(), 1);
                assert_eq!(q.joins[0].kind, JoinKind::LeftOuter);
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_left_join_without_outer_keyword() {
        // `left join` is shorthand for `left outer join` in SQL — we accept it.
        let stmt = parse("User as u left join Order as o on u.id = o.user_id").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.joins[0].kind, JoinKind::LeftOuter);
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_right_join() {
        let stmt = parse("User as u right join Order as o on u.id = o.user_id").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.joins[0].kind, JoinKind::RightOuter);
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_cross_join_has_no_on() {
        let stmt = parse("User cross join Order").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.joins[0].kind, JoinKind::Cross);
                assert!(q.joins[0].on.is_none());
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_multi_join_chain() {
        let stmt = parse(
            "User as u join Order as o on u.id = o.user_id \
             join Product as p on o.product_id = p.id",
        )
        .unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.joins.len(), 2);
                assert_eq!(q.joins[0].source, "Order");
                assert_eq!(q.joins[1].source, "Product");
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_join_with_filter_tail() {
        // Filter/order/limit still work after a join clause.
        let stmt = parse(
            "User as u join Order as o on u.id = o.user_id \
             filter o.total > 100 order .name limit 10",
        )
        .unwrap();
        match stmt {
            Statement::Query(q) => {
                assert_eq!(q.joins.len(), 1);
                assert!(q.filter.is_some());
                assert!(q.order.is_some());
                assert!(q.limit.is_some());
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_join_requires_on_for_inner() {
        // Non-cross joins require `on <expr>`. Missing `on` is a parse error.
        let err = parse("User join Order").unwrap_err();
        assert!(
            err.message().contains("on"),
            "expected on-clause error, got {:?}",
            err.message()
        );
    }

    #[test]
    fn test_parse_update_on_joined_query_errors() {
        // E1.1 explicitly rejects update/delete on joined queries — SQL
        // semantics here are messy and we're not implementing them yet.
        let err =
            parse("User as u join Order as o on u.id = o.user_id update { age := 1 }").unwrap_err();
        assert!(err.message().contains("update"));
    }

    #[test]
    fn test_parse_delete_on_joined_query_errors() {
        let err = parse("User as u join Order as o on u.id = o.user_id delete").unwrap_err();
        assert!(err.message().contains("delete"));
    }

    // ---- Mission E2a: DISTINCT + IN-list + BETWEEN + LIKE -----------------

    #[test]
    fn test_parse_distinct() {
        let stmt = parse("User distinct { .name }").unwrap();
        match stmt {
            Statement::Query(q) => {
                assert!(q.distinct);
                assert!(q.projection.is_some());
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_in_list() {
        let stmt = parse(r#"User filter .name in ("Alice", "Bob")"#).unwrap();
        match stmt {
            Statement::Query(q) => match q.filter.unwrap() {
                Expr::InList {
                    expr,
                    list,
                    negated,
                } => {
                    assert!(!negated);
                    assert!(matches!(*expr, Expr::Field(f) if f == "name"));
                    assert_eq!(list.len(), 2);
                }
                other => panic!("expected InList, got {other:?}"),
            },
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_not_in_list() {
        let stmt = parse("User filter .age not in (1, 2, 3)").unwrap();
        match stmt {
            Statement::Query(q) => match q.filter.unwrap() {
                Expr::InList { negated, list, .. } => {
                    assert!(negated);
                    assert_eq!(list.len(), 3);
                }
                other => panic!("expected InList, got {other:?}"),
            },
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_between() {
        // BETWEEN desugars into >= AND <=.
        let stmt = parse("User filter .age between 10 and 20").unwrap();
        match stmt {
            Statement::Query(q) => {
                match q.filter.unwrap() {
                    Expr::BinaryOp(_, BinOp::And, _) => {} // desugared
                    other => panic!("expected And (desugared between), got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_not_between() {
        // NOT BETWEEN desugars into < OR >.
        let stmt = parse("User filter .age not between 10 and 20").unwrap();
        match stmt {
            Statement::Query(q) => {
                match q.filter.unwrap() {
                    Expr::BinaryOp(_, BinOp::Or, _) => {} // desugared
                    other => panic!("expected Or (desugared not between), got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_like() {
        let stmt = parse(r#"User filter .name like "A%""#).unwrap();
        match stmt {
            Statement::Query(q) => match q.filter.unwrap() {
                Expr::BinaryOp(l, BinOp::Like, r) => {
                    assert!(matches!(*l, Expr::Field(f) if f == "name"));
                    assert!(matches!(*r, Expr::Literal(Literal::String(s)) if s == "A%"));
                }
                other => panic!("expected Like, got {other:?}"),
            },
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_not_like() {
        let stmt = parse(r#"User filter .name not like "A%""#).unwrap();
        match stmt {
            Statement::Query(q) => match q.filter.unwrap() {
                Expr::UnaryOp(UnaryOp::Not, inner) => {
                    assert!(matches!(*inner, Expr::BinaryOp(_, BinOp::Like, _)));
                }
                other => panic!("expected Not(Like), got {other:?}"),
            },
            _ => panic!("expected query"),
        }
    }

    // ---- Mission E2b: GROUP BY + HAVING ------------------------------------

    #[test]
    fn test_parse_group_by_single_key() {
        let stmt = parse("User group .status { .status, n: count(.name) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let gb = q.group_by.unwrap();
                assert_eq!(
                    gb.keys,
                    vec![GroupKey {
                        expr: Expr::Field("status".into()),
                        output_name: "status".into(),
                    }]
                );
                assert!(gb.having.is_none());
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 2);
                assert!(matches!(
                    &proj[1].expr,
                    Expr::FunctionCall(AggFunc::Count, _, _)
                ));
                assert_eq!(proj[1].alias.as_deref(), Some("n"));
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_group_by_multi_key() {
        let stmt = parse("User group .status, .age { .status, .age }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let gb = q.group_by.unwrap();
                assert_eq!(
                    gb.keys,
                    vec![
                        GroupKey {
                            expr: Expr::Field("status".into()),
                            output_name: "status".into(),
                        },
                        GroupKey {
                            expr: Expr::Field("age".into()),
                            output_name: "age".into(),
                        }
                    ]
                );
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_group_by_having() {
        let stmt = parse("User group .status having count(.name) > 1 { .status }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let gb = q.group_by.unwrap();
                assert_eq!(
                    gb.keys,
                    vec![GroupKey {
                        expr: Expr::Field("status".into()),
                        output_name: "status".into(),
                    }]
                );
                assert!(gb.having.is_some());
                // HAVING is `count(.name) > 1` — BinaryOp(FunctionCall, Gt, Literal)
                match gb.having.unwrap() {
                    Expr::BinaryOp(l, BinOp::Gt, _) => {
                        assert!(matches!(*l, Expr::FunctionCall(AggFunc::Count, _, _)));
                    }
                    other => panic!("expected BinaryOp, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_aggregate_in_projection() {
        // Unaliased aggregate function calls in projection.
        let stmt = parse("User group .status { .status, count(.name), sum(.age) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 3);
                assert!(matches!(
                    &proj[1].expr,
                    Expr::FunctionCall(AggFunc::Count, _, _)
                ));
                assert!(matches!(
                    &proj[2].expr,
                    Expr::FunctionCall(AggFunc::Sum, _, _)
                ));
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_aggregate_in_aliased_projection() {
        let stmt = parse("User group .status { .status, total: count(.name), average: avg(.age) }")
            .unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj[1].alias.as_deref(), Some("total"));
                assert!(matches!(
                    &proj[1].expr,
                    Expr::FunctionCall(AggFunc::Count, _, _)
                ));
                assert_eq!(proj[2].alias.as_deref(), Some("average"));
                assert!(matches!(
                    &proj[2].expr,
                    Expr::FunctionCall(AggFunc::Avg, _, _)
                ));
            }
            _ => panic!("expected query"),
        }
    }

    // ─── IS NULL / IS NOT NULL parser tests ────────────────────────────

    #[test]
    fn test_parse_is_null() {
        let stmt = parse("User filter .age is null").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                assert_eq!(
                    filter,
                    Expr::UnaryOp(UnaryOp::IsNull, Box::new(Expr::Field("age".into())))
                );
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_is_not_null() {
        let stmt = parse("User filter .age is not null").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                assert_eq!(
                    filter,
                    Expr::UnaryOp(UnaryOp::IsNotNull, Box::new(Expr::Field("age".into())))
                );
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_eq_null_desugars_to_is_null() {
        let stmt = parse("User filter .age = null").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                assert_eq!(
                    filter,
                    Expr::UnaryOp(UnaryOp::IsNull, Box::new(Expr::Field("age".into())))
                );
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_neq_null_desugars_to_is_not_null() {
        let stmt = parse("User filter .age != null").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                assert_eq!(
                    filter,
                    Expr::UnaryOp(UnaryOp::IsNotNull, Box::new(Expr::Field("age".into())))
                );
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_null_comparisons_parse_ok() {
        // `< null`, `>= null` etc. parse successfully now that `null` is a
        // valid expression. At runtime they evaluate to Empty (no match),
        // which is correct null-propagation semantics.
        assert!(parse("User filter .age < null").is_ok());
        assert!(parse("User filter .age >= null").is_ok());
    }

    #[test]
    fn test_parse_count_star_expr() {
        let stmt = parse("User filter count(*) > 0").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                match filter {
                    Expr::BinaryOp(left, BinOp::Gt, _) => {
                        assert_eq!(
                            *left,
                            Expr::FunctionCall(
                                AggFunc::Count,
                                Box::new(Expr::Field("*".into())),
                                AggregateMode::Symmetric,
                            )
                        );
                    }
                    _ => panic!("expected comparison"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    // ─── String function parser tests ──────────────────────────────────

    #[test]
    fn test_parse_upper_in_filter() {
        let stmt = parse(r#"User filter upper(.name) = "ALICE""#).unwrap();
        match stmt {
            Statement::Query(q) => {
                let f = q.filter.unwrap();
                match f {
                    Expr::BinaryOp(left, BinOp::Eq, _right) => {
                        assert!(matches!(*left, Expr::ScalarFunc(ScalarFn::Upper, _)));
                    }
                    _ => panic!("expected binary op with upper"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_substring() {
        let stmt = parse("User { sub: substring(.name, 1, 3) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                match &proj[0].expr {
                    Expr::ScalarFunc(ScalarFn::Substring, args) => {
                        assert_eq!(args.len(), 3);
                    }
                    other => panic!("expected ScalarFunc Substring, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_concat() {
        let stmt = parse(r#"User { full: concat(.name, " - ", .email) }"#).unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                match &proj[0].expr {
                    Expr::ScalarFunc(ScalarFn::Concat, args) => {
                        assert_eq!(args.len(), 3);
                    }
                    other => panic!("expected ScalarFunc Concat, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    // ─── CASE WHEN parser tests ────────────────────────────────────────

    #[test]
    fn test_parse_case_single_when() {
        let stmt = parse(r#"User filter case when .age > 30 then true else false end"#).unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                match filter {
                    Expr::Case { whens, else_expr } => {
                        assert_eq!(whens.len(), 1);
                        assert!(else_expr.is_some());
                    }
                    other => panic!("expected Case expr, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_case_multiple_whens() {
        let stmt = parse(
            r#"User { label: case when .age > 30 then "senior" when .age > 20 then "adult" else "young" end }"#
        ).unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                match &proj[0].expr {
                    Expr::Case { whens, else_expr } => {
                        assert_eq!(whens.len(), 2);
                        assert!(else_expr.is_some());
                    }
                    other => panic!("expected Case expr, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_case_without_else() {
        let stmt = parse(r#"User filter case when .age > 30 then true end"#).unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                match filter {
                    Expr::Case { whens, else_expr } => {
                        assert_eq!(whens.len(), 1);
                        assert!(else_expr.is_none());
                    }
                    other => panic!("expected Case expr, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    // ─── Mul/Div expression tests (E2f) ───────────────────────────────

    #[test]
    fn test_parse_mul_expr() {
        let stmt = parse("User filter .price * .quantity > 100").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                match filter {
                    Expr::BinaryOp(left, BinOp::Gt, _) => match *left {
                        Expr::BinaryOp(_, BinOp::Mul, _) => {}
                        other => panic!("expected Mul, got {other:?}"),
                    },
                    other => panic!("expected BinaryOp Gt, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_div_expr() {
        let stmt = parse("User { ratio: .total / .count }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj[0].alias.as_deref(), Some("ratio"));
                match &proj[0].expr {
                    Expr::BinaryOp(_, BinOp::Div, _) => {}
                    other => panic!("expected Div, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_mul_div_precedence() {
        // .a + .b * .c should parse as .a + (.b * .c)
        let stmt = parse("User filter .a + .b * .c > 0").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                match filter {
                    Expr::BinaryOp(left, BinOp::Gt, _) => match *left {
                        Expr::BinaryOp(_, BinOp::Add, right) => {
                            assert!(matches!(*right, Expr::BinaryOp(_, BinOp::Mul, _)));
                        }
                        other => panic!("expected Add, got {other:?}"),
                    },
                    other => panic!("expected Gt, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    // ─── Multi-column ORDER BY tests (E2f) ────────────────────────────

    #[test]
    fn test_parse_multi_order() {
        let stmt = parse("User order .name asc, .age desc").unwrap();
        match stmt {
            Statement::Query(q) => {
                let order = q.order.unwrap();
                assert_eq!(order.keys.len(), 2);
                assert_eq!(order.keys[0].expr, Expr::Field("name".into()));
                assert!(!order.keys[0].descending);
                assert_eq!(order.keys[1].expr, Expr::Field("age".into()));
                assert!(order.keys[1].descending);
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_order_default_asc() {
        let stmt = parse("User order .name").unwrap();
        match stmt {
            Statement::Query(q) => {
                let order = q.order.unwrap();
                assert_eq!(order.keys.len(), 1);
                assert!(!order.keys[0].descending);
            }
            _ => panic!("expected query"),
        }
    }

    // ─── ALTER TABLE / DROP TABLE parser tests (E2g) ──────────────────

    #[test]
    fn test_parse_alter_add_column() {
        let stmt = parse("alter User add column status: str").unwrap();
        match stmt {
            Statement::AlterTable(at) => {
                assert_eq!(at.table, "User");
                match at.action {
                    AlterAction::AddColumn {
                        name,
                        type_name,
                        required,
                    } => {
                        assert_eq!(name, "status");
                        assert_eq!(type_name, "str");
                        assert!(!required);
                    }
                    other => panic!("expected AddColumn, got {other:?}"),
                }
            }
            other => panic!("expected AlterTable, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_alter_add_required_column() {
        let stmt = parse("alter User add required status: str").unwrap();
        match stmt {
            Statement::AlterTable(at) => match at.action {
                AlterAction::AddColumn { required, .. } => assert!(required),
                other => panic!("expected AddColumn, got {other:?}"),
            },
            other => panic!("expected AlterTable, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_type_with_unique_modifier() {
        let stmt = parse("type User { required unique email: str, age: int }").unwrap();
        match stmt {
            Statement::CreateType(ct) => {
                assert!(ct.fields[0].required && ct.fields[0].unique);
                assert!(!ct.fields[1].unique);
            }
            other => panic!("expected CreateType, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_type_unique_before_required() {
        // Modifiers accepted in either order.
        let stmt = parse("type User { unique required email: str }").unwrap();
        match stmt {
            Statement::CreateType(ct) => {
                assert!(ct.fields[0].required && ct.fields[0].unique);
            }
            other => panic!("expected CreateType, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_alter_add_unique() {
        let stmt = parse("alter User add unique .email").unwrap();
        match stmt {
            Statement::AlterTable(at) => assert!(matches!(
                at.action,
                AlterAction::AddUnique {
                    target: IndexTarget::Column(ref column),
                    ..
                } if column == "email"
            )),
            other => panic!("expected AlterTable, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_alter_drop_column() {
        let stmt = parse("alter User drop column status").unwrap();
        match stmt {
            Statement::AlterTable(at) => {
                assert_eq!(at.table, "User");
                match at.action {
                    AlterAction::DropColumn { name, .. } => assert_eq!(name, "status"),
                    other => panic!("expected DropColumn, got {other:?}"),
                }
            }
            other => panic!("expected AlterTable, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_alter_drop_without_column_keyword() {
        let stmt = parse("alter User drop status").unwrap();
        match stmt {
            Statement::AlterTable(at) => match at.action {
                AlterAction::DropColumn { name, .. } => assert_eq!(name, "status"),
                other => panic!("expected DropColumn, got {other:?}"),
            },
            other => panic!("expected AlterTable, got {other:?}"),
        }
    }

    #[test]
    fn test_parse_drop_table() {
        let stmt = parse("drop User").unwrap();
        match stmt {
            Statement::DropTable(dt) => assert_eq!(dt.table, "User"),
            other => panic!("expected DropTable, got {other:?}"),
        }
    }

    // ─── IN subquery parser tests (E2h) ───────────────────────────────

    #[test]
    fn test_parse_in_subquery() {
        let stmt = parse("User filter .name in (VIP { .name })").unwrap();
        match stmt {
            Statement::Query(q) => {
                let filter = q.filter.unwrap();
                match filter {
                    Expr::InSubquery {
                        expr,
                        subquery,
                        negated,
                    } => {
                        assert!(!negated);
                        assert!(matches!(*expr, Expr::Field(ref f) if f == "name"));
                        assert_eq!(subquery.source, "VIP");
                    }
                    other => panic!("expected InSubquery, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_not_in_subquery() {
        let stmt = parse("User filter .id not in (Order { .user_id })").unwrap();
        match stmt {
            Statement::Query(q) => match q.filter.unwrap() {
                Expr::InSubquery { negated, .. } => assert!(negated),
                other => panic!("expected InSubquery, got {other:?}"),
            },
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_in_literal_list_still_works() {
        // Ensure existing IN (literal) parsing isn't broken
        let stmt = parse("User filter .age in (25, 30, 35)").unwrap();
        match stmt {
            Statement::Query(q) => match q.filter.unwrap() {
                Expr::InList { list, negated, .. } => {
                    assert!(!negated);
                    assert_eq!(list.len(), 3);
                }
                other => panic!("expected InList, got {other:?}"),
            },
            _ => panic!("expected query"),
        }
    }

    // ---- Materialized view parser tests ------------------------------------

    #[test]
    fn test_parse_create_view() {
        let stmt = parse("materialize OldUsers as User filter .age > 28").unwrap();
        match stmt {
            Statement::CreateView(cv) => {
                assert_eq!(cv.name, "OldUsers");
                assert_eq!(cv.query.source, "User");
                assert!(cv.query.filter.is_some());
                assert!(!cv.query_text.is_empty());
            }
            _ => panic!("expected CreateView"),
        }
    }

    #[test]
    fn test_parse_create_view_with_projection() {
        let stmt = parse("materialize UserNames as User { .name }").unwrap();
        match stmt {
            Statement::CreateView(cv) => {
                assert_eq!(cv.name, "UserNames");
                assert!(cv.query.projection.is_some());
            }
            _ => panic!("expected CreateView"),
        }
    }

    #[test]
    fn test_parse_refresh_view() {
        let stmt = parse("refresh OldUsers").unwrap();
        match stmt {
            Statement::RefreshView(rv) => {
                assert_eq!(rv.name, "OldUsers");
            }
            _ => panic!("expected RefreshView"),
        }
    }

    #[test]
    fn test_parse_drop_view() {
        let stmt = parse("drop view OldUsers").unwrap();
        match stmt {
            Statement::DropView(dv) => {
                assert_eq!(dv.name, "OldUsers");
            }
            _ => panic!("expected DropView"),
        }
    }

    #[test]
    fn test_parse_drop_table_still_works() {
        let stmt = parse("drop Users").unwrap();
        match stmt {
            Statement::DropTable(dt) => {
                assert_eq!(dt.table, "Users");
            }
            _ => panic!("expected DropTable"),
        }
    }

    #[test]
    fn test_parse_union() {
        let stmt = parse("User union Order").unwrap();
        match stmt {
            Statement::Union(u) => {
                assert!(!u.all);
                match *u.left {
                    Statement::Query(_) => {}
                    _ => panic!("expected Query on left"),
                }
                match *u.right {
                    Statement::Query(_) => {}
                    _ => panic!("expected Query on right"),
                }
            }
            _ => panic!("expected Union"),
        }
    }

    #[test]
    fn test_parse_union_all() {
        let stmt = parse("User union all Order").unwrap();
        match stmt {
            Statement::Union(u) => {
                assert!(u.all, "expected UNION ALL");
                match *u.left {
                    Statement::Query(_) => {}
                    _ => panic!("expected Query on left"),
                }
                match *u.right {
                    Statement::Query(_) => {}
                    _ => panic!("expected Query on right"),
                }
            }
            _ => panic!("expected Union"),
        }
    }

    #[test]
    fn test_parse_union_chain() {
        // Left-associative: A union B union C => Union(Union(A, B), C)
        let stmt = parse("User union Order union Product").unwrap();
        match stmt {
            Statement::Union(outer) => {
                assert!(!outer.all);
                // Right side is Product
                match *outer.right {
                    Statement::Query(q) => assert_eq!(q.source, "Product"),
                    _ => panic!("expected Query(Product) on right"),
                }
                // Left side is Union(User, Order)
                match *outer.left {
                    Statement::Union(inner) => {
                        assert!(!inner.all);
                        match *inner.left {
                            Statement::Query(q) => assert_eq!(q.source, "User"),
                            _ => panic!("expected Query(User)"),
                        }
                        match *inner.right {
                            Statement::Query(q) => assert_eq!(q.source, "Order"),
                            _ => panic!("expected Query(Order)"),
                        }
                    }
                    _ => panic!("expected inner Union"),
                }
            }
            _ => panic!("expected Union"),
        }
    }

    #[test]
    fn test_parse_union_with_filter() {
        let stmt = parse("User filter .age > 10 union Order filter .total > 50").unwrap();
        match stmt {
            Statement::Union(u) => {
                assert!(!u.all);
                // Both sides should be queries (the filter is part of each query)
                match *u.left {
                    Statement::Query(q) => {
                        assert_eq!(q.source, "User");
                        assert!(q.filter.is_some());
                    }
                    _ => panic!("expected Query on left"),
                }
                match *u.right {
                    Statement::Query(q) => {
                        assert_eq!(q.source, "Order");
                        assert!(q.filter.is_some());
                    }
                    _ => panic!("expected Query on right"),
                }
            }
            _ => panic!("expected Union"),
        }
    }

    #[test]
    fn test_parse_count_distinct_standalone() {
        let stmt = parse("count(distinct User { .name })").unwrap();
        match stmt {
            Statement::Query(q) => {
                let agg = q.aggregation.unwrap();
                assert_eq!(agg.function, AggFunc::CountDistinct);
                assert_eq!(agg.argument, Some(Expr::Field("name".into())));
            }
            _ => panic!("expected Query"),
        }
    }

    #[test]
    fn test_parse_count_distinct_in_projection() {
        let stmt = parse("User group .dept { .dept, count(distinct .name) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 2);
                match &proj[1].expr {
                    Expr::FunctionCall(func, _, _) => {
                        assert_eq!(*func, AggFunc::CountDistinct);
                    }
                    _ => panic!("expected FunctionCall"),
                }
            }
            _ => panic!("expected Query"),
        }
    }

    // ---- Window function parser tests ----------------------------------------

    #[test]
    fn test_parse_window_row_number_order() {
        let stmt = parse("User { .name, rn: row_number() over (order .age) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 2);
                assert_eq!(proj[1].alias.as_deref(), Some("rn"));
                match &proj[1].expr {
                    Expr::Window {
                        function,
                        args,
                        partition_by,
                        order_by,
                        ..
                    } => {
                        assert_eq!(*function, WindowFunc::RowNumber);
                        assert!(args.is_empty());
                        assert!(partition_by.is_empty());
                        assert_eq!(order_by.len(), 1);
                        assert_eq!(order_by[0].expr, Expr::Field("age".into()));
                        assert!(!order_by[0].descending);
                    }
                    other => panic!("expected Window, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_window_sum_partition_order() {
        let stmt =
            parse("User { .name, s: sum(.salary) over (partition .dept order .salary) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 2);
                assert_eq!(proj[1].alias.as_deref(), Some("s"));
                match &proj[1].expr {
                    Expr::Window {
                        function,
                        args,
                        partition_by,
                        order_by,
                        ..
                    } => {
                        assert_eq!(*function, WindowFunc::Sum);
                        assert_eq!(args.len(), 1);
                        assert!(matches!(&args[0], Expr::Field(f) if f == "salary"));
                        assert_eq!(partition_by, &[Expr::Field("dept".into())]);
                        assert_eq!(order_by.len(), 1);
                        assert_eq!(order_by[0].expr, Expr::Field("salary".into()));
                        assert!(!order_by[0].descending);
                    }
                    other => panic!("expected Window, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_window_rank_desc() {
        let stmt =
            parse("User { .dept, .salary, r: rank() over (partition .dept order .salary desc) }")
                .unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 3);
                match &proj[2].expr {
                    Expr::Window {
                        function,
                        partition_by,
                        order_by,
                        ..
                    } => {
                        assert_eq!(*function, WindowFunc::Rank);
                        assert_eq!(partition_by, &[Expr::Field("dept".into())]);
                        assert_eq!(order_by.len(), 1);
                        assert!(order_by[0].descending);
                    }
                    other => panic!("expected Window, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_window_dense_rank() {
        let stmt = parse("User { .name, dr: dense_rank() over (order .score desc) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 2);
                match &proj[1].expr {
                    Expr::Window { function, .. } => {
                        assert_eq!(*function, WindowFunc::DenseRank);
                    }
                    other => panic!("expected Window, got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_parse_sum_without_over_is_aggregate() {
        // sum(.salary) alone (no `over`) stays as FunctionCall, not Window.
        let stmt = parse("User group .dept { .dept, total: sum(.salary) }").unwrap();
        match stmt {
            Statement::Query(q) => {
                let proj = q.projection.unwrap();
                assert_eq!(proj.len(), 2);
                match &proj[1].expr {
                    Expr::FunctionCall(AggFunc::Sum, _, _) => {} // correct
                    other => panic!("expected FunctionCall(Sum), got {other:?}"),
                }
            }
            _ => panic!("expected query"),
        }
    }

    #[test]
    fn test_nesting_depth_limit() {
        // Build a deeply nested parenthesized expression that exceeds MAX_NESTING_DEPTH.
        let mut query = String::from("User filter ");
        for _ in 0..70 {
            query.push('(');
        }
        query.push_str(".age > 1");
        for _ in 0..70 {
            query.push(')');
        }
        let result = parse(&query);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(
            err.message().contains("nesting depth"),
            "expected nesting depth error, got: {}",
            err.message()
        );
    }

    #[test]
    fn test_unary_prefix_nesting_depth_limit() {
        // A long chain of `not` prefixes recurses through parse_primary
        // without passing through parse_expr's guard. It must error cleanly
        // at the depth limit instead of overflowing the stack.
        let query = String::from("User filter ") + &"not ".repeat(5000) + ".active";
        let result = parse(&query);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(
            err.message().contains("nesting depth"),
            "expected nesting depth error, got: {}",
            err.message()
        );
    }

    #[test]
    fn test_moderate_nesting_succeeds() {
        // 10 levels of nesting should be fine.
        let mut query = String::from("User filter ");
        for _ in 0..10 {
            query.push('(');
        }
        query.push_str(".age > 1");
        for _ in 0..10 {
            query.push(')');
        }
        assert!(parse(&query).is_ok());
    }

    /// Regression for issue #26: `fuzz_parser` crashed on the 3-byte input
    /// `nn{` — the projection loop consumed the Eof token and then indexed
    /// past the end of `tokens`. Must return an error instead.
    #[test]
    fn test_parse_fuzz_repro_projection_eof() {
        let err = parse("nn{").expect_err("unterminated projection must error, not panic");
        let _ = err.message();
    }

    /// Regression for issue #26: `fuzz_roundtrip` tripped the same bug with
    /// the 2-byte input `z{`.
    #[test]
    fn test_parse_fuzz_repro_short_projection_eof() {
        let err = parse("z{").expect_err("unterminated projection must error, not panic");
        let _ = err.message();
    }

    #[test]
    fn test_update_at_statement_start_gives_helpful_error() {
        let err =
            parse(r#"update User filter .name = "Alice" { age := 31 }"#).expect_err("should fail");
        let msg = err.message();
        assert!(
            msg.contains("pipeline syntax"),
            "error should mention pipeline syntax, got: {msg}"
        );
        assert!(
            msg.contains("update"),
            "error should mention 'update', got: {msg}"
        );
    }

    #[test]
    fn test_delete_at_statement_start_gives_helpful_error() {
        let err = parse("delete User filter .age < 18").expect_err("should fail");
        let msg = err.message();
        assert!(
            msg.contains("pipeline syntax"),
            "error should mention pipeline syntax, got: {msg}"
        );
        assert!(
            msg.contains("delete"),
            "error should mention 'delete', got: {msg}"
        );
    }
}

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

    #[test]
    fn typoed_statement_keyword_gets_suggestion() {
        let err = parse("updat User set age = 1").unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("near token"), "{msg}");
        assert!(msg.contains("did you mean `update`"), "{msg}");
    }
}

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

    // ── P-6: reserved words as column names ────────────────────────────

    #[test]
    fn reserved_word_field_name_gives_actionable_error() {
        let err = parse("type Post { type: str }").unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("'type' is a reserved word")
                && msg.contains("field name")
                && msg.contains("quote it as `type`"),
            "unhelpful message: {msg}"
        );
    }

    #[test]
    fn reserved_modifier_word_as_field_name_gives_actionable_error() {
        // `required` is a modifier keyword; followed directly by `:` it is
        // instead the field's (reserved) name — the old error was the opaque
        // "expected field name, got ':'".
        let err = parse("type Post { required: bool }").unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("'required' is a reserved word") && msg.contains("quote it as `required`"),
            "unhelpful message: {msg}"
        );
    }

    #[test]
    fn reserved_word_in_insert_assignment_gives_actionable_error() {
        let err = parse(r#"insert Post { type := "x" }"#).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("'type' is a reserved word"), "{msg}");
    }

    #[test]
    fn reserved_word_in_alter_column_gives_actionable_error() {
        let err = parse("alter Post add column order: int").unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("'order' is a reserved word"), "{msg}");
    }

    #[test]
    fn backtick_field_name_parses_as_identifier() {
        let stmt = parse("type Post { `type`: str, `order`: int }").unwrap();
        match stmt {
            Statement::CreateType(ct) => {
                assert_eq!(ct.fields[0].name, "type");
                assert_eq!(ct.fields[1].name, "order");
            }
            other => panic!("expected CreateType, got {other:?}"),
        }
    }

    #[test]
    fn backtick_field_still_honors_modifiers() {
        let stmt = parse("type Post { required `type`: str }").unwrap();
        match stmt {
            Statement::CreateType(ct) => {
                assert_eq!(ct.fields[0].name, "type");
                assert!(ct.fields[0].required);
            }
            other => panic!("expected CreateType, got {other:?}"),
        }
    }

    // ── P-7: DDL idempotency ───────────────────────────────────────────

    #[test]
    fn create_type_if_not_exists_parses() {
        let stmt = parse("type Post if not exists { id: int }").unwrap();
        match stmt {
            Statement::CreateType(ct) => assert!(ct.if_not_exists),
            other => panic!("expected CreateType, got {other:?}"),
        }
    }

    #[test]
    fn create_type_without_clause_defaults_false() {
        let stmt = parse("type Post { id: int }").unwrap();
        match stmt {
            Statement::CreateType(ct) => assert!(!ct.if_not_exists),
            other => panic!("expected CreateType, got {other:?}"),
        }
    }

    #[test]
    fn drop_if_exists_parses() {
        match parse("drop if exists Post").unwrap() {
            Statement::DropTable(dt) => assert!(dt.if_exists),
            other => panic!("expected DropTable, got {other:?}"),
        }
        match parse("drop Post").unwrap() {
            Statement::DropTable(dt) => assert!(!dt.if_exists),
            other => panic!("expected DropTable, got {other:?}"),
        }
    }

    #[test]
    fn drop_view_if_exists_parses() {
        match parse("drop view if exists ActiveUsers").unwrap() {
            Statement::DropView(dv) => {
                assert!(dv.if_exists);
                assert_eq!(dv.name, "ActiveUsers");
            }
            other => panic!("expected DropView, got {other:?}"),
        }
    }

    #[test]
    fn add_index_and_unique_if_not_exists_parse() {
        match parse("alter Post add index if not exists .slug").unwrap() {
            Statement::AlterTable(at) => {
                assert!(matches!(
                    at.action,
                    AlterAction::AddIndex {
                        if_not_exists: true,
                        ..
                    }
                ));
            }
            other => panic!("expected AlterTable, got {other:?}"),
        }
        match parse("alter Post add unique if not exists .slug").unwrap() {
            Statement::AlterTable(at) => {
                assert!(matches!(
                    at.action,
                    AlterAction::AddUnique {
                        if_not_exists: true,
                        ..
                    }
                ));
            }
            other => panic!("expected AlterTable, got {other:?}"),
        }
    }

    #[test]
    fn expression_index_targets_parse_with_stable_table_local_identity() {
        use powdb_storage::stored_json_path::{
            StoredJsonPathSegmentV1 as Segment, StoredJsonPathV1,
        };

        let expected = StoredJsonPathV1::new(
            "data",
            vec![Segment::Key("author".into()), Segment::Index(0)],
        );
        for query in [
            "alter Post add index (.data->author->0)",
            "alter Post add unique if not exists (.data->\"author\"->0)",
            "alter Post drop index if exists (.data->author->0)",
        ] {
            let Statement::AlterTable(alter) = parse(query).unwrap() else {
                panic!("expected alter table for {query}");
            };
            let (target, flag) = match alter.action {
                AlterAction::AddIndex {
                    target,
                    if_not_exists,
                }
                | AlterAction::AddUnique {
                    target,
                    if_not_exists,
                } => (target, if_not_exists),
                AlterAction::DropIndex { target, if_exists } => (target, if_exists),
                other => panic!("expected index action, got {other:?}"),
            };
            assert_eq!(target, IndexTarget::JsonPath(expected.clone()));
            assert_eq!(
                flag,
                query.contains("if not exists") || query.contains("if exists")
            );
        }
    }

    #[test]
    fn expression_index_target_rejects_ambiguous_or_non_path_forms() {
        let cases = [
            (
                "alter Post add index .data->author",
                "must be parenthesized",
            ),
            (
                "alter Post add index (p.data->author)",
                "qualified JSON paths",
            ),
            (
                "alter Post add index (.data)",
                "use `.column` for a stored column",
            ),
            (
                "alter Post add index (.data->age + 1)",
                "only a direct JSON path",
            ),
            (
                "alter Post drop index ({ value := 1 })",
                "expected an unqualified JSON path",
            ),
        ];
        for (query, expected) in cases {
            let error = parse(query).expect_err(query).to_string();
            assert!(
                error.contains(expected),
                "`{query}` should mention `{expected}`, got `{error}`"
            );
        }
    }

    #[test]
    fn alter_drop_column_if_exists_parses() {
        match parse("alter Post drop column if exists status").unwrap() {
            Statement::AlterTable(at) => {
                assert!(matches!(
                    at.action,
                    AlterAction::DropColumn {
                        if_exists: true,
                        ..
                    }
                ));
            }
            other => panic!("expected AlterTable, got {other:?}"),
        }
    }

    // ── P-8: introspection ─────────────────────────────────────────────

    #[test]
    fn schema_parses_to_list_types() {
        assert_eq!(parse("schema").unwrap(), Statement::ListTypes);
    }

    #[test]
    fn describe_parses_to_describe() {
        assert_eq!(
            parse("describe Post").unwrap(),
            Statement::Describe("Post".to_string())
        );
    }

    #[test]
    fn schema_with_type_aliases_describe() {
        assert_eq!(
            parse("schema Post").unwrap(),
            Statement::Describe("Post".to_string())
        );
    }

    #[test]
    fn schema_links_parses_to_list_links() {
        assert_eq!(parse("schema links").unwrap(), Statement::ListLinks);
    }

    #[test]
    fn describe_links_still_names_a_table() {
        // Only the `schema links` spelling is the link listing; `describe`
        // keeps treating `links` as an ordinary type name.
        assert_eq!(
            parse("describe links").unwrap(),
            Statement::Describe("links".to_string())
        );
    }
}

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

    /// Pull the filter expression out of a single-table query.
    fn filter_of(src: &str) -> Expr {
        match parse(src).unwrap() {
            Statement::Query(q) => q.filter.expect("expected a filter"),
            other => panic!("expected a query, got {other:?}"),
        }
    }

    #[test]
    fn ident_key_path() {
        // .data->author->name
        let e = filter_of(r#"Post filter .data->author->name = "x""#);
        let Expr::BinaryOp(lhs, BinOp::Eq, _) = e else {
            panic!("expected an equality, got {e:?}");
        };
        assert_eq!(
            *lhs,
            Expr::JsonPath {
                base: Box::new(Expr::Field("data".into())),
                segments: vec![PathSeg::Key("author".into()), PathSeg::Key("name".into())],
            }
        );
    }

    #[test]
    fn string_form_key_path() {
        // .data->"weird key!" (PowQL strings are double-quoted)
        let e = filter_of(r#"Post filter .data->"weird key!" = 1"#);
        let Expr::BinaryOp(lhs, _, _) = e else {
            panic!("expected binop");
        };
        assert_eq!(
            *lhs,
            Expr::JsonPath {
                base: Box::new(Expr::Field("data".into())),
                segments: vec![PathSeg::Key("weird key!".into())],
            }
        );
    }

    #[test]
    fn array_index_path() {
        // .data->tags->0
        let e = filter_of(r#"Post filter .data->tags->0 = "rust""#);
        let Expr::BinaryOp(lhs, _, _) = e else {
            panic!("expected binop");
        };
        assert_eq!(
            *lhs,
            Expr::JsonPath {
                base: Box::new(Expr::Field("data".into())),
                segments: vec![PathSeg::Key("tags".into()), PathSeg::Index(0)],
            }
        );
    }

    #[test]
    fn qualified_base_path() {
        // posts.data->author  (join-qualified base)
        let e = filter_of(r#"Post as posts filter posts.data->author = "a""#);
        let Expr::BinaryOp(lhs, _, _) = e else {
            panic!("expected binop");
        };
        assert_eq!(
            *lhs,
            Expr::JsonPath {
                base: Box::new(Expr::QualifiedField {
                    qualifier: "posts".into(),
                    field: "data".into(),
                }),
                segments: vec![PathSeg::Key("author".into())],
            }
        );
    }

    #[test]
    fn path_binds_tighter_than_comparison_and_arithmetic() {
        // `.data->age > 21` must be `(.data->age) > 21`, not `.data->(age > 21)`.
        let e = filter_of("Post filter .data->age > 21");
        let Expr::BinaryOp(lhs, BinOp::Gt, rhs) = e else {
            panic!("expected a top-level `>`, got {e:?}");
        };
        assert!(matches!(*lhs, Expr::JsonPath { .. }));
        assert_eq!(*rhs, Expr::Literal(Literal::Int(21)));

        // `.a->b + 1` must be `(.a->b) + 1`.
        let e = filter_of("Post filter .a->b + 1 = 3");
        let Expr::BinaryOp(add, BinOp::Eq, _) = e else {
            panic!("expected eq");
        };
        let Expr::BinaryOp(lhs, BinOp::Add, _) = *add else {
            panic!("expected `+` under `=`, got {add:?}");
        };
        assert!(matches!(*lhs, Expr::JsonPath { .. }));
    }

    #[test]
    fn dash_vs_arrow_lexing() {
        // `.a->1` is a path index: `->` lexes as one token because the chars
        // are adjacent, ahead of the single-char `-`.
        let idx = filter_of("Post filter .a->1 = 0");
        let Expr::BinaryOp(lhs, BinOp::Eq, _) = idx else {
            panic!("expected eq");
        };
        assert_eq!(
            *lhs,
            Expr::JsonPath {
                base: Box::new(Expr::Field("a".into())),
                segments: vec![PathSeg::Index(1)],
            }
        );

        // `.a - 1` (spaced) is subtraction — the `-` is not glued to a digit,
        // so it lexes as the minus operator.
        let sub = filter_of("Post filter .a - 1 = 0");
        let Expr::BinaryOp(lhs, BinOp::Eq, _) = sub else {
            panic!("expected eq");
        };
        assert!(
            matches!(*lhs, Expr::BinaryOp(_, BinOp::Sub, _)),
            "`.a - 1` should be subtraction, got {lhs:?}"
        );

        // `.a-1` (no spaces) is the lexer gotcha: `-1` is a NEGATIVE INTEGER
        // literal (the number rule fires when `-` is glued to a digit), so the
        // stream is `.a` then `-1` with no operator between — a parse error,
        // NOT subtraction and NOT a path.
        assert!(
            parse("Post filter .a-1 = 0").is_err(),
            "`.a-1` should fail to parse (negative-literal gotcha)"
        );

        // `.a - >` is `.a` `-` `>` — a dangling `>`, a parse error.
        assert!(
            parse("Post filter .a - > 0").is_err(),
            "`.a - >` should fail to parse"
        );
    }

    #[test]
    fn negative_index_rejected() {
        let err = parse("Post filter .data->-1 = 0").unwrap_err();
        assert!(
            err.to_string().contains("array index"),
            "expected an index error, got: {err}"
        );
    }

    #[test]
    fn path_on_literal_base_rejected() {
        // A `->` after a non-field base is a parse error.
        let err = parse("Post filter 5->x = 1").unwrap_err();
        assert!(
            err.to_string().to_lowercase().contains("field base"),
            "expected a field-base error, got: {err}"
        );
    }

    #[test]
    fn json_path_assignment_target_is_targeted_unsupported() {
        // `Doc update { .data->x := 5 }` must not die with a generic
        // "expected field name": it must name the unsupported position and
        // the whole-column alternative. Both the leading-dot path-target form
        // and the bare `data->x` form take the targeted branch.
        for stmt in [
            "Doc update { .data->x := 5 }",
            "Doc update { data->x := 5 }",
        ] {
            let err = parse(stmt).unwrap_err();
            assert!(
                matches!(err, ParseError::Unsupported { .. }),
                "{stmt}: expected Unsupported, got {err:?}"
            );
            let msg = err.to_string();
            assert!(
                msg.contains("JSON path assignment targets are not supported"),
                "{stmt}: message must state the unsupported feature: {msg}"
            );
            assert!(
                msg.contains("json_set"),
                "{stmt}: message must point at the whole-column alternative: {msg}"
            );
        }
        // A normal whole-column update still parses.
        assert!(parse(r#"Doc update { data := "{}" }"#).is_ok());
    }

    #[test]
    fn json_type_scalar_parses() {
        let e = filter_of(r#"Post filter json_type(.data->x) = "string""#);
        let Expr::BinaryOp(lhs, _, _) = e else {
            panic!("expected binop");
        };
        let Expr::ScalarFunc(ScalarFn::JsonType, args) = *lhs else {
            panic!("expected json_type call, got {lhs:?}");
        };
        assert_eq!(args.len(), 1);
        assert!(matches!(args[0], Expr::JsonPath { .. }));
    }

    #[test]
    fn path_in_projection() {
        // Projection, ordering, and grouping all retain the same structural
        // JsonPath expression rather than lowering it to an alias string.
        let stmt = parse("Post { author: .data->author }").unwrap();
        let Statement::Query(q) = stmt else {
            panic!("expected query");
        };
        let proj = q.projection.unwrap();
        assert_eq!(proj[0].alias.as_deref(), Some("author"));
        assert!(matches!(proj[0].expr, Expr::JsonPath { .. }));

        let Statement::Query(ordered) = parse("Post order .data->author { .id }").unwrap() else {
            panic!("expected query");
        };
        assert!(matches!(
            ordered.order.unwrap().keys[0].expr,
            Expr::JsonPath { .. }
        ));

        let Statement::Query(grouped) =
            parse("Post group .data->author { .data->author }").unwrap()
        else {
            panic!("expected query");
        };
        assert!(matches!(
            grouped.group_by.unwrap().keys[0].expr,
            Expr::JsonPath { .. }
        ));
    }
}

/// `tokens_to_text` is the inverse of the lexer: whatever it writes must read
/// back as the very tokens it was given. A view stores its defining query as
/// that text and re-lexes it on every refresh, so any disagreement silently
/// redefines the view.
#[cfg(test)]
mod token_text_roundtrip {
    use super::*;
    use proptest::prelude::*;

    /// One value of every payload-carrying `Token` variant (with payloads
    /// chosen to break a naive reconstruction: reserved words that exist
    /// only as quoted identifiers, names needing quotes, escapes the lexer
    /// decodes, numbers whose `Display` changes their type) plus every
    /// payload-free variant except EOF.
    ///
    /// A variant added to `Token` and not added here still cannot slip
    /// through unnoticed: `tokens_to_text` re-lexes what it built and
    /// refuses to return a mismatch.
    fn every_token() -> Vec<Token> {
        let mut toks = vec![
            // Identifiers: plain, reserved words, and spellings that only
            // exist inside backticks.
            Token::Ident("User".into()),
            Token::Ident("order".into()),
            Token::Ident("true".into()),
            Token::Ident("null".into()),
            Token::Ident("column name".into()),
            Token::Ident("1st".into()),
            Token::Ident("a-b".into()),
            Token::Ident("has#hash".into()),
            Token::Ident("has.dot".into()),
            Token::Ident("héllo".into()),
            Token::DotIdent("name".into()),
            Token::DotIdent("order".into()),
            Token::DotIdent("field name".into()),
            Token::DotIdent("1st".into()),
            Token::DotIdent("a-b".into()),
            // Numbers, including the ones whose `Display` loses the
            // fraction or overflows to a very long expansion.
            Token::IntLit(0),
            Token::IntLit(-1),
            Token::IntLit(i64::MIN),
            Token::IntLit(i64::MAX),
            Token::FloatLit(0.0),
            Token::FloatLit(-0.0),
            Token::FloatLit(3.0),
            Token::FloatLit(-2.0),
            Token::FloatLit(1.5),
            Token::FloatLit(1e300),
            Token::FloatLit(1e-300),
            Token::FloatLit(f64::MIN_POSITIVE),
            // Strings: every escape the lexer decodes, plus characters
            // that would otherwise leak into the grammar.
            Token::StringLit(String::new()),
            Token::StringLit("plain".into()),
            Token::StringLit("back\\slash".into()),
            Token::StringLit("he said \"hi\"".into()),
            Token::StringLit("line\nbreak".into()),
            Token::StringLit("tab\there".into()),
            Token::StringLit("carriage\rreturn".into()),
            Token::StringLit("`backtick`".into()),
            Token::StringLit("# not a comment".into()),
            Token::StringLit("} filter .x = 1".into()),
            Token::StringLit("\\\"".into()),
            Token::BoolLit(true),
            Token::BoolLit(false),
            Token::Param("1".into()),
            Token::Param("name".into()),
            Token::Param(String::new()),
        ];
        toks.extend([
            Token::Type,
            Token::Filter,
            Token::Order,
            Token::Limit,
            Token::Offset,
            Token::Insert,
            Token::Update,
            Token::Delete,
            Token::Upsert,
            Token::Returning,
            Token::Select,
            Token::Required,
            Token::Default,
            Token::Auto,
            Token::Multi,
            Token::Link,
            Token::Index,
            Token::Unique,
            Token::On,
            Token::Conflict,
            Token::Asc,
            Token::Desc,
            Token::And,
            Token::Or,
            Token::Not,
            Token::Exists,
            Token::Let,
            Token::As,
            Token::Match,
            Token::Group,
            Token::Join,
            Token::Inner,
            Token::LeftKw,
            Token::RightKw,
            Token::Outer,
            Token::Cross,
            Token::Transaction,
            Token::Begin,
            Token::Commit,
            Token::Rollback,
            Token::View,
            Token::Materialized,
            Token::Refresh,
            Token::Union,
            Token::Having,
            Token::Distinct,
            Token::In,
            Token::Between,
            Token::Like,
            Token::Count,
            Token::Avg,
            Token::Sum,
            Token::Min,
            Token::Max,
            Token::Raw,
            Token::Is,
            Token::Null,
        ]);
        toks.extend([
            Token::Upper,
            Token::Lower,
            Token::Length,
            Token::Trim,
            Token::Substring,
            Token::Concat,
            Token::Abs,
            Token::Round,
            Token::Ceil,
            Token::Floor,
            Token::Sqrt,
            Token::Pow,
            Token::Now,
            Token::Extract,
            Token::DateAdd,
            Token::DateDiff,
            Token::JsonType,
            Token::JsonText,
            Token::Cast,
            Token::Case,
            Token::When,
            Token::Then,
            Token::Else,
            Token::End,
            Token::Over,
            Token::Partition,
            Token::RowNumber,
            Token::Rank,
            Token::DenseRank,
            Token::Alter,
            Token::Drop,
            Token::Add,
            Token::Column,
            Token::Explain,
            Token::Schema,
            Token::Describe,
        ]);
        toks.extend([
            Token::Eq,
            Token::Neq,
            Token::Lt,
            Token::Gt,
            Token::Lte,
            Token::Gte,
            Token::Assign,
            Token::Arrow,
            Token::Pipe,
            Token::Coalesce,
            Token::Plus,
            Token::Minus,
            Token::Star,
            Token::Slash,
            Token::LBrace,
            Token::RBrace,
            Token::LParen,
            Token::RParen,
            Token::Comma,
            Token::Colon,
            Token::Dot,
        ]);
        toks
    }

    /// Reconstruct, re-lex, and compare against the input stream.
    fn assert_round_trips(tokens: &[Token]) {
        let text =
            tokens_to_text(tokens).unwrap_or_else(|e| panic!("no source text for {tokens:?}: {e}"));
        let relexed = lex(&text)
            .unwrap_or_else(|e| panic!("`{text}` from {tokens:?} does not lex: {}", e.message));
        let mut expected = tokens.to_vec();
        expected.push(Token::Eof);
        assert_eq!(relexed, expected, "`{text}` re-lexes to different tokens");
    }

    #[test]
    fn every_token_round_trips_on_its_own() {
        for tok in every_token() {
            assert_round_trips(std::slice::from_ref(&tok));
        }
    }

    #[test]
    fn eof_contributes_no_text() {
        assert_eq!(tokens_to_text(&[Token::Eof]).unwrap(), "");
        assert_eq!(tokens_to_text(&[]).unwrap(), "");
    }

    /// A token with no faithful spelling is refused, not written wrong.
    /// A backtick inside an identifier is the case that cannot be quoted
    /// (the lexer has no escape inside backticks), and a non-finite float has
    /// no literal form at all. The parameter name is caught by nothing but the
    /// closing re-lex check, which is the point of having one.
    #[test]
    fn unspellable_tokens_are_typed_errors() {
        for tok in [
            Token::Ident("has`tick".into()),
            Token::DotIdent("has`tick".into()),
            Token::Ident(String::new()),
            Token::FloatLit(f64::INFINITY),
            Token::FloatLit(f64::NAN),
            Token::Param("two words".into()),
        ] {
            let err = tokens_to_text(std::slice::from_ref(&tok))
                .expect_err("{tok:?} must not be written back wrong");
            assert!(
                matches!(err, ParseError::Unsupported { .. }),
                "expected a typed Unsupported error for {tok:?}, got {err:?}"
            );
        }
    }

    /// A view whose source text cannot be stored faithfully is refused at
    /// creation instead of quietly becoming a different query.
    #[test]
    fn unspellable_view_source_is_refused() {
        let huge = format!("1{}.0", "0".repeat(400)); // overflows f64 to inf
        let err = parse(&format!("materialize V as U filter .x = {huge}"))
            .expect_err("a view source that cannot round-trip must be refused");
        assert!(
            matches!(err, ParseError::Unsupported { .. }),
            "expected a typed Unsupported error, got {err:?}"
        );
    }

    proptest! {
        /// The property the concrete cases are instances of: for ANY token
        /// stream, `lex(tokens_to_text(tokens)) == tokens`.
        #[test]
        fn any_token_stream_round_trips(
            tokens in proptest::collection::vec(
                proptest::sample::select(every_token()),
                0..12usize,
            )
        ) {
            let text = tokens_to_text(&tokens)
                .map_err(|e| TestCaseError::fail(format!("no source text: {e}")))?;
            let relexed = lex(&text)
                .map_err(|e| TestCaseError::fail(format!("`{text}` does not lex: {}", e.message)))?;
            let mut expected = tokens.clone();
            expected.push(Token::Eof);
            prop_assert_eq!(relexed, expected, "`{}` re-lexes to different tokens", text);
        }
    }
}