rdbms-playground/src/dsl/grammar/sql_create_table.rs

//! SQL `CREATE TABLE` grammar (ADR-0035 §4, sub-phase 4a).
//!
//! Grammar-as-text in the unified tree (ADR-0030 §4), but — unlike
//! the DML `Sql*` commands which execute verbatim — `CREATE TABLE`
//! executes **structurally** (ADR-0035 §1): the builder extracts the
//! columns / types / primary key and the worker drives the existing
//! `do_create_table` machinery, so an advanced-mode-created table is a
//! first-class playground object (metadata, `STRICT`, the ten-type
//! vocabulary). This file holds only the **shape**; the `CommandNode`
//! and `build_sql_create_table` live in `ddl.rs` (mirroring how the
//! DML shapes here pair with `data.rs` builders).
//!
//! Scope (4a): columns + types (the §3 alias map, incl. the two-word
//! `double precision` and discarded length args) + the clean-reuse
//! column constraints `NOT NULL` / `UNIQUE` / column-level
//! `PRIMARY KEY` + single/compound table-level `PRIMARY KEY (…)` +
//! `IF NOT EXISTS`. **No** foreign keys (4b), **no** `DEFAULT` /
//! `CHECK` / table-level `UNIQUE` (the 4a.2 constraint slice) — those
//! shapes are absent here, so typing them is an ordinary parse error
//! until their slice lands.
//!
//! The entry-word dispatch consumes the leading `CREATE` keyword
//! before this shape walks, so it starts at `TABLE` (mirroring
//! `sql_insert::SQL_INSERT_SHAPE`, which starts at `INTO`).

use crate::dsl::grammar::sql_select::reject_internal_table;
use crate::dsl::grammar::{IdentSource, Node, ValidationError, Word, shared, sql_expr};
use crate::dsl::types::Type;

static COMMA: Node = Node::Punct(',');

// --- Type-name slot (advanced-mode aliases, ADR-0035 §3) ----------

/// Reject any type name the SQL alias resolver doesn't recognise.
/// Distinct from `shared::validate_type_name` (the simple-mode
/// validator, which accepts only the ten keywords); this one also
/// admits the standard-SQL aliases via [`Type::from_sql_name`]. The
/// `{expected}` list still names the ten playground keywords — the
/// vocabulary we teach — not the aliases.
fn validate_sql_type_name(value: &str) -> Result<(), ValidationError> {
    if Type::from_sql_name(value).is_some() {
        Ok(())
    } else {
        let expected = Type::all()
            .iter()
            .map(|t| t.keyword())
            .collect::<Vec<_>>()
            .join(", ");
        Err(ValidationError {
            message_key: "parse.custom.unknown_type",
            args: vec![("found", value.to_string()), ("expected", expected)],
        })
    }
}

/// The single-word type name. `double precision` is handled by a
/// separate keyword-pair branch in [`SQL_TYPE`] (ADR-0035 §6.3,
/// implementer call), so the validator only ever sees one word.
const SQL_TYPE_NAME: Node = Node::Ident {
    source: IdentSource::Types,
    role: "col_type",
    validator: Some(validate_sql_type_name),
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};

// Optional `( n [, n] )` length / precision argument — matched and
// **discarded** (the playground's types are unparameterised,
// ADR-0035 §3). `varchar(255)`, `numeric(10, 2)`.
static LENGTH_SECOND_NODES: &[Node] = &[Node::Punct(','), Node::NumberLit { validator: None }];
static LENGTH_NODES: &[Node] = &[
    Node::Punct('('),
    Node::NumberLit { validator: None },
    Node::Optional(&Node::Seq(LENGTH_SECOND_NODES)),
    Node::Punct(')'),
];
const LENGTH_OPT: Node = Node::Optional(&Node::Seq(LENGTH_NODES));

// `double precision` — the lone two-word alias. A dedicated branch so
// the per-word `Ident` validator never has to make sense of `double`
// on its own (ADR-0035 §6.3). The builder maps the pair to
// `Type::Real`.
static DOUBLE_PRECISION_NODES: &[Node] = &[
    Node::Word(Word::keyword("double")),
    Node::Word(Word::keyword("precision")),
];
static TYPE_WITH_LENGTH_NODES: &[Node] = &[SQL_TYPE_NAME, LENGTH_OPT];
static SQL_TYPE_CHOICES: &[Node] = &[
    Node::Seq(DOUBLE_PRECISION_NODES),
    Node::Seq(TYPE_WITH_LENGTH_NODES),
];
/// `double precision | <type-keyword-or-alias> [ '(' n [, n] ')' ]`.
pub(crate) const SQL_TYPE: Node = Node::Choice(SQL_TYPE_CHOICES);

// --- Column-level constraints (4a clean-reuse set only) -----------

pub(crate) static NOT_NULL_NODES: &[Node] = &[
    Node::Word(Word::keyword("not")),
    Node::Word(Word::keyword("null")),
];
static PRIMARY_KEY_NODES: &[Node] = &[
    Node::Word(Word::keyword("primary")),
    Node::Word(Word::keyword("key")),
];
// `DEFAULT <value>` / `CHECK (<expr>)` reuse the full ADR-0031
// `sql_expr` surface (the same fragment `WHERE`/projections use). The
// fragment is validate-only (no AST), so the builder captures the
// matched text's **raw SQL** by byte span (ADR-0035 §4a.2).
//
// A bare `DEFAULT` value is a **literal** (or a *parenthesised*
// expression) — matching standard SQL, where a complex default must be
// `DEFAULT (expr)`. This is not just spec fidelity: a bare unbounded
// `sql_expr` greedily consumes a following `NOT` (as the start of
// `NOT IN`/`NOT LIKE`/`NOT BETWEEN`), which would break the common
// `DEFAULT 0 NOT NULL`. The parens give the expression a clean end.
static DEFAULT_PAREN_EXPR_NODES: &[Node] = &[
    Node::Punct('('),
    Node::Subgrammar(&sql_expr::SQL_OR_EXPR),
    Node::Punct(')'),
];
static DEFAULT_VALUE_CHOICES: &[Node] = &[
    Node::Seq(DEFAULT_PAREN_EXPR_NODES),
    Node::NumberLit { validator: None },
    Node::StringLit,
    Node::Word(Word::keyword("null")),
    Node::Word(Word::keyword("true")),
    Node::Word(Word::keyword("false")),
];
const DEFAULT_VALUE: Node = Node::Choice(DEFAULT_VALUE_CHOICES);
pub(crate) static DEFAULT_NODES: &[Node] = &[Node::Word(Word::keyword("default")), DEFAULT_VALUE];
pub(crate) static CHECK_NODES: &[Node] = &[
    Node::Word(Word::keyword("check")),
    Node::Punct('('),
    Node::Subgrammar(&sql_expr::SQL_OR_EXPR),
    Node::Punct(')'),
];

// --- Foreign keys (ADR-0035 §5, sub-phase 4b) ---------------------
//
// Inline `REFERENCES <parent>[(<col>)] [ON DELETE/UPDATE …]` and
// table-level `[CONSTRAINT <name>] FOREIGN KEY (<col>) REFERENCES …`.
// Each is the SQL spelling of an ADR-0013 named relationship. The
// referenced parent table/column use the `Tables`/`Columns` sources
// (completion + existence hints), matching the `add relationship`
// endpoints; the `( <col> )` is optional (the bare `REFERENCES
// <parent>` form resolves to the parent's PK at execution).

// `IdentSource::Tables` existence-checks the parent, so a typo'd parent
// shows a pre-submit hint. A self-referencing FK (`references <self>`
// while creating `<self>`) is NOT flagged: the schema-existence
// diagnostic exempts a `Tables` reference matching the `CREATE TABLE`
// target — the table being created in the same statement (ADR-0035 §4i c,
// `schema_existence_diagnostics`'s `created_tables`).
const FK_PARENT_TABLE: Node = Node::Ident {
    source: IdentSource::Tables,
    role: "fk_parent_table",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};
const FK_PARENT_COLUMN: Node = Node::Ident {
    source: IdentSource::Columns,
    role: "fk_parent_column",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};
static FK_PARENT_COL_NODES: &[Node] = &[Node::Punct('('), FK_PARENT_COLUMN, Node::Punct(')')];
const FK_PARENT_COL_OPT: Node = Node::Optional(&Node::Seq(FK_PARENT_COL_NODES));

// `REFERENCES <parent> [ ( <col> ) ] [on delete/update …]` — the inline
// column-FK constraint. The referential clauses reuse the shared
// `on <delete|update> <action>` grammar (the DSL `add relationship`
// keywords are the SQL keywords).
static REFERENCES_NODES: &[Node] = &[
    Node::Word(Word::keyword("references")),
    FK_PARENT_TABLE,
    FK_PARENT_COL_OPT,
    shared::REFERENTIAL_CLAUSES,
];
const REFERENCES_CLAUSE: Node = Node::Seq(REFERENCES_NODES);
// `NOT NULL` | `UNIQUE` | `PRIMARY KEY` | `DEFAULT <expr>` |
// `CHECK (<expr>)`. Each branch starts on a distinct keyword, so the
// `Choice` never ambiguously commits.
static COL_CONSTRAINT_CHOICES: &[Node] = &[
    Node::Seq(NOT_NULL_NODES),
    Node::Word(Word::keyword("unique")),
    Node::Seq(PRIMARY_KEY_NODES),
    Node::Seq(DEFAULT_NODES),
    Node::Seq(CHECK_NODES),
    REFERENCES_CLAUSE,
];
const COL_CONSTRAINT: Node = Node::Choice(COL_CONSTRAINT_CHOICES);
/// Zero-or-more column constraints after the type (`min: 0`).
const COL_CONSTRAINT_SUFFIX: Node = Node::Repeated {
    inner: &COL_CONSTRAINT,
    separator: None,
    min: 0,
};

// --- Column definition: `<name> <type> [constraints…]` ------------

pub(crate) const COL_NAME: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "col_name",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};

static COLUMN_DEF_NODES: &[Node] = &[COL_NAME, SQL_TYPE, COL_CONSTRAINT_SUFFIX];
const COLUMN_DEF: Node = Node::Seq(COLUMN_DEF_NODES);

// --- Table-level `PRIMARY KEY ( col, … )` (single + compound) -----

// A column reference inside the table-level PK list. The columns are
// defined in this same statement (not in the schema yet), so
// `NewName` (no schema completion); the builder checks each name
// against the defined columns.
const PK_COLUMN_REF: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "pk_column",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};

static TABLE_PK_NODES: &[Node] = &[
    Node::Word(Word::keyword("primary")),
    Node::Word(Word::keyword("key")),
    Node::Punct('('),
    Node::Repeated {
        inner: &PK_COLUMN_REF,
        separator: Some(&COMMA),
        min: 1,
    },
    Node::Punct(')'),
];
// `pub(crate)` so `ALTER TABLE … ADD PRIMARY KEY (…)` (ADR-0035 §4g)
// reuses the body — it parses, then the ALTER builder refuses it with a
// specific message (adding a PK to an existing table is unsupported).
pub(crate) const TABLE_PK: Node = Node::Seq(TABLE_PK_NODES);

// Table-level `UNIQUE ( col, … )`. A single column normalises into
// that column's `unique` flag (round-trips via the existing
// single-column path); two or more become a composite UNIQUE
// constraint (ADR-0035 §4a.2). Distinct ident role from `pk_column`
// so the builder routes them separately.
const UNIQUE_COLUMN_REF: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "unique_column",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};
static TABLE_UNIQUE_NODES: &[Node] = &[
    Node::Word(Word::keyword("unique")),
    Node::Punct('('),
    Node::Repeated {
        inner: &UNIQUE_COLUMN_REF,
        separator: Some(&COMMA),
        min: 1,
    },
    Node::Punct(')'),
];
// `pub(crate)` so `ALTER TABLE … ADD UNIQUE (…)` (ADR-0035 §4g) reuses
// the same composite-UNIQUE body.
pub(crate) const TABLE_UNIQUE: Node = Node::Seq(TABLE_UNIQUE_NODES);

// Table-level `CHECK ( <expr> )` (ADR-0035 §4a.3) — a multi-column
// CHECK referencing several columns. Same paren-bounded shape as the
// column-level CHECK; the builder tells them apart by position (a
// CHECK at element start, with no column definition open, is
// table-level). The engine reports no CHECK constraints, so a
// table-level CHECK round-trips via a dedicated metadata table.
static TABLE_CHECK_NODES: &[Node] = &[
    Node::Word(Word::keyword("check")),
    Node::Punct('('),
    Node::Subgrammar(&sql_expr::SQL_OR_EXPR),
    Node::Punct(')'),
];
// `pub(crate)` so `ALTER TABLE … ADD [CONSTRAINT <name>] CHECK (…)`
// (ADR-0035 §4g) reuses the same table-CHECK body.
pub(crate) const TABLE_CHECK: Node = Node::Seq(TABLE_CHECK_NODES);

// Table-level foreign key (ADR-0035 §5, sub-phase 4b):
// `[CONSTRAINT <name>] FOREIGN KEY ( <child col> ) REFERENCES
// <parent> [ ( <col> ) ] [on delete/update …]`. The child column is
// being defined in this statement (`NewName`); the optional
// `CONSTRAINT <name>` names the relationship (an inline `REFERENCES`
// is always auto-named instead).
const FK_CHILD_COLUMN: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "fk_child_column",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};
const FK_NAME: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "fk_name",
    validator: None,
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};
// The `FOREIGN KEY (col) REFERENCES …` body, shared by the named and
// unnamed table-FK element branches. Each branch starts with a concrete
// keyword (`foreign` / `constraint`) — never a leading `Optional`,
// which would advance the Seq index and turn a later mismatch into a
// hard failure that aborts the enclosing element `Choice`.
static FOREIGN_KEY_BODY_NODES: &[Node] = &[
    Node::Word(Word::keyword("foreign")),
    Node::Word(Word::keyword("key")),
    Node::Punct('('),
    FK_CHILD_COLUMN,
    Node::Punct(')'),
    Node::Word(Word::keyword("references")),
    FK_PARENT_TABLE,
    FK_PARENT_COL_OPT,
    shared::REFERENTIAL_CLAUSES,
];
const FOREIGN_KEY_BODY: Node = Node::Seq(FOREIGN_KEY_BODY_NODES);
// `FOREIGN KEY (…) …` — the unnamed table-level FK (auto-named).
// `pub(crate)` so `ALTER TABLE … ADD [CONSTRAINT <name>] FOREIGN KEY
// (…)` (ADR-0035 §4g) reuses the FK body (the §4g `CONSTRAINT <name>`
// prefix is supplied by the ALTER grammar, not this body).
pub(crate) const TABLE_FK: Node = FOREIGN_KEY_BODY;
// `CONSTRAINT <name> FOREIGN KEY (…) …` — the named table-level FK.
static TABLE_FK_NAMED_NODES: &[Node] = &[
    Node::Word(Word::keyword("constraint")),
    FK_NAME,
    Node::Word(Word::keyword("foreign")),
    Node::Word(Word::keyword("key")),
    Node::Punct('('),
    FK_CHILD_COLUMN,
    Node::Punct(')'),
    Node::Word(Word::keyword("references")),
    FK_PARENT_TABLE,
    FK_PARENT_COL_OPT,
    shared::REFERENTIAL_CLAUSES,
];
const TABLE_FK_NAMED: Node = Node::Seq(TABLE_FK_NAMED_NODES);

// One element of the column list: a table-level `PRIMARY KEY (…)` /
// `UNIQUE (…)` / `CHECK (…)` / `[CONSTRAINT <name>] FOREIGN KEY (…)`,
// or a column definition. The table-level forms are tried first — each
// starts with a keyword (`primary` / `unique` / `check` / `constraint`
// / `foreign`) that disambiguates it from a column name. (A column
// literally named with one of those keywords is therefore unavailable,
// the same trade real SQL makes with its reserved words.)
static ELEMENT_CHOICES: &[Node] =
    &[TABLE_PK, TABLE_UNIQUE, TABLE_CHECK, TABLE_FK_NAMED, TABLE_FK, COLUMN_DEF];
const ELEMENT_INNER: Node = Node::Choice(ELEMENT_CHOICES);
// Issue #4: wrap the element slot in `IntroProse` so a fresh element
// position (`create table T (` and after every `,`) surfaces a prose
// hint that names the column-name role *and* the table-level
// constraint keywords. The bare candidate render shows only the
// constraint keywords because the `COLUMN_DEF` branch starts with a
// `NewName` ident that has no concrete candidate to offer; the prose
// makes the dominant first move visible without suppressing Tab.
const ELEMENT: Node = Node::Hinted {
    mode: crate::dsl::grammar::HintMode::IntroProse("hint.create_table_element"),
    inner: &ELEMENT_INNER,
};

static COLUMN_LIST_NODES: &[Node] = &[
    Node::Punct('('),
    Node::Repeated {
        inner: &ELEMENT,
        separator: Some(&COMMA),
        min: 1,
    },
    Node::Punct(')'),
];
const COLUMN_LIST: Node = Node::Seq(COLUMN_LIST_NODES);

// --- `IF NOT EXISTS` (ADR-0035 §4, no-op-with-note at execution) ---

static IF_NOT_EXISTS_NODES: &[Node] = &[
    Node::Word(Word::keyword("if")),
    Node::Word(Word::keyword("not")),
    Node::Word(Word::keyword("exists")),
];
const IF_NOT_EXISTS_OPT: Node = Node::Optional(&Node::Seq(IF_NOT_EXISTS_NODES));

// --- The full post-`CREATE` shape ---------------------------------

/// The table name. `NewName` (the user invents it); `__rdbms_*`
/// rejected (ADR-0030 §6) so the walker's `[ERR]` indicator flags an
/// internal-table target before submit, mirroring the DML shapes.
const TABLE_NAME: Node = Node::Ident {
    source: IdentSource::NewName,
    role: "table_name",
    validator: Some(reject_internal_table),
    highlight_override: None,
    writes_table: false,
    writes_column: false,
    writes_user_listed_column: false,
    writes_table_alias: false,
    writes_cte_name: false,
    writes_projection_alias: false,
};

static SQL_CREATE_TABLE_TAIL_NODES: &[Node] = &[
    Node::Word(Word::keyword("table")),
    IF_NOT_EXISTS_OPT,
    TABLE_NAME,
    COLUMN_LIST,
    Node::Optional(&Node::Punct(';')),
];

/// The post-`CREATE` portion of a SQL `CREATE TABLE` statement.
///
/// `TABLE [IF NOT EXISTS] <name> ( <element> (',' <element>)* ) [';']`,
/// where an element is a column definition or a table-level
/// `PRIMARY KEY (…)` (ADR-0035 §4). The entry-word dispatch consumes
/// the leading `CREATE` before this shape walks, so a `CommandNode`
/// references it via `Subgrammar` (the `ddl::SQL_CREATE_TABLE` node).
pub static SQL_CREATE_TABLE_SHAPE: Node = Node::Seq(SQL_CREATE_TABLE_TAIL_NODES);

// =================================================================
// Tests — grammar accept/reject for the post-`CREATE` tail.
// =================================================================

#[cfg(test)]
mod tests {
    use super::SQL_CREATE_TABLE_SHAPE;
    use crate::dsl::walker::context::WalkContext;
    use crate::dsl::walker::driver::{NodeWalkResult, walk_node};
    use crate::dsl::walker::outcome::MatchedPath;

    /// Walk `input` against the CREATE TABLE tail. `true` only when the
    /// walk matches *and* consumes all of `input` (trailing whitespace
    /// allowed). Schemaless context: the shape is structural, so the
    /// table/column idents match by shape and `reject_internal_table`
    /// still fires on `__rdbms_*`.
    fn walks(input: &str) -> bool {
        let mut ctx = WalkContext::new();
        let mut path = MatchedPath::new();
        let mut per_byte = Vec::new();
        match walk_node(input, 0, &SQL_CREATE_TABLE_SHAPE, &mut ctx, &mut path, &mut per_byte) {
            NodeWalkResult::Matched { end, .. } => input[end..].trim().is_empty(),
            _ => false,
        }
    }

    fn good(input: &str) {
        assert!(walks(input), "{input:?} should be a valid CREATE TABLE tail");
    }

    fn bad(input: &str) {
        assert!(!walks(input), "{input:?} should NOT walk as a complete CREATE TABLE tail");
    }

    #[test]
    fn minimal_single_column() {
        good("table t (id int)");
        good("table t (id int);");
        good("table widgets (sku text)");
    }

    #[test]
    fn multiple_columns() {
        good("table t (id int, name text)");
        good("table orders (id int, total real, note text)");
    }

    #[test]
    fn column_level_constraints() {
        good("table t (id int primary key)");
        good("table t (id int primary key, name text not null)");
        good("table t (id serial primary key, email text unique)");
        good("table t (a int not null unique, b text)");
    }

    #[test]
    fn integer_primary_key_parses() {
        // INTEGER PRIMARY KEY is accepted (it maps to plain int at the
        // builder, ADR-0035 §3 — verified there, not here).
        good("table t (id integer primary key)");
    }

    #[test]
    fn table_level_primary_key_single_and_compound() {
        good("table t (id int, primary key (id))");
        good("table t (a int, b int, primary key (a, b))");
        good("table t (a int, b int, c text, primary key (a, b, c))");
    }

    #[test]
    fn standard_sql_type_aliases() {
        good("table t (a integer, b varchar, c boolean, d timestamp)");
        good("table t (e bigint, f smallint, g char, h numeric)");
        good("table t (i binary, j varbinary, k float)");
    }

    #[test]
    fn double_precision_two_word_type() {
        good("table t (x double precision)");
        good("table t (id int, x double precision, y real)");
    }

    #[test]
    fn length_precision_args_accepted_and_ignored() {
        good("table t (name varchar(255))");
        good("table t (price numeric(10, 2))");
        good("table t (code char(8), amount decimal(12, 4))");
    }

    #[test]
    fn if_not_exists_admitted() {
        good("table if not exists t (id int)");
        good("table if not exists widgets (sku text, qty int);");
    }

    #[test]
    fn internal_target_table_rejected() {
        bad("table __rdbms_playground_columns (id int)");
        bad("table if not exists __rdbms_playground_relationships (id int)");
    }

    #[test]
    fn unknown_type_rejected() {
        bad("table t (id money)");
        bad("table t (id json)");
        // Bare `double` (no `precision`) is not a supported type.
        bad("table t (x double)");
    }

    #[test]
    fn structurally_incomplete_or_wrong_rejected() {
        // Empty column list — at least one element required.
        bad("table t ()");
        // No column list at all.
        bad("table t");
        // Missing table name.
        bad("table (id int)");
        // Column with no type.
        bad("table t (id)");
        // Trailing comma with no following element.
        bad("table t (id int,)");
        // Missing TABLE keyword (entry dispatch would have eaten it).
        bad("t (id int)");
        // Unclosed column list.
        bad("table t (id int");
    }

    #[test]
    fn column_default_and_check_accepted() {
        // 4a.2: DEFAULT / CHECK reuse the full sql_expr surface.
        good("table t (id int, n int default 0)");
        good("table t (id int, name text default 'x')");
        good("table t (id int check (id > 0))");
        good("table t (id int check (id > 0 and id < 100))");
        good("table t (price real default 0.0 check (price >= 0.0))");
    }

    #[test]
    fn table_level_unique_accepted() {
        // 4a.2: composite + single-column table-level UNIQUE.
        good("table t (a int, b int, unique (a, b))");
        good("table t (a int, b text, unique (b))");
        good("table t (id int primary key, email text, unique (email))");
    }

    #[test]
    fn table_level_check_accepted() {
        // 4a.3: a table-level (multi-column) CHECK is now admitted, in
        // any position among the elements and alongside other forms.
        good("table t (a int, b int, check (a < b))");
        good("table t (a int, b int, c int, check (a < b), check (b < c))");
        good("table t (a int, b int, primary key (a), check (a < b))");
        good("table t (a int, b int, unique (a, b), check (a <> b))");
        good("table t (price real check (price >= 0), total real, check (total >= price))");
    }

    #[test]
    fn foreign_keys_accepted() {
        // 4b: inline `REFERENCES` and table-level `FOREIGN KEY`, with
        // optional `(col)`, `ON DELETE`/`ON UPDATE`, and `CONSTRAINT`.
        good("table t (id int, ref int references other(id))");
        good("table t (id int, ref int references other)"); // bare ref
        good("table t (id int, ref int references other(id) on delete cascade)");
        good("table t (id int, ref int references other(id) on update set null on delete restrict)");
        good("table t (id int, ref int, foreign key (ref) references other(id))");
        good("table t (id int, ref int, constraint fk_x foreign key (ref) references other(id))");
        good(
            "table t (id int, a int, b int, foreign key (a) references p(id), \
             foreign key (b) references q(id))",
        );
        // FK alongside the other table elements (coexistence).
        good("table t (id int primary key, ref int references other(id), check (id > 0))");
        // self-reference (parent is the table being created).
        good("table emp (id int primary key, mgr int references emp(id))");
    }
}

// =================================================================
// Builder tests — `parse_command` (advanced mode) lowers the SQL
// `CREATE TABLE` to `Command::SqlCreateTable` with the right fields
// (ADR-0035 §1/§3, sub-phase 4a).
// =================================================================

#[cfg(test)]
mod builder_tests {
    use crate::dsl::action::ReferentialAction;
    use crate::dsl::command::{ColumnSpec, Command, SqlForeignKey};
    use crate::dsl::parser::{parse_command, parse_command_in_mode};
    use crate::dsl::types::Type;
    use crate::mode::Mode;

    /// Parse in advanced mode and unwrap the `SqlCreateTable` fields.
    fn sct(input: &str) -> (String, Vec<(String, Type)>, Vec<String>, bool) {
        match parse_command(input).expect("should parse as SQL CREATE TABLE") {
            Command::SqlCreateTable {
                name,
                columns,
                primary_key,
                if_not_exists,
                ..
            } => (
                name,
                columns.into_iter().map(|c| (c.name, c.ty)).collect(),
                primary_key,
                if_not_exists,
            ),
            other => panic!("expected SqlCreateTable, got {other:?}"),
        }
    }

    #[test]
    fn minimal_columns_and_types() {
        let (name, cols, pk, ine) = sct("create table t (id int, name text)");
        assert_eq!(name, "t");
        assert_eq!(
            cols,
            vec![("id".to_string(), Type::Int), ("name".to_string(), Type::Text)]
        );
        assert!(pk.is_empty(), "no PK declared");
        assert!(!ine);
    }

    #[test]
    fn integer_primary_key_is_plain_int_not_serial() {
        // ADR-0035 §3: the type map is lexical; INTEGER PRIMARY KEY is
        // a plain int PK, NOT auto-increment (that is `serial`).
        let (_, cols, pk, _) = sct("create table t (id integer primary key)");
        assert_eq!(cols, vec![("id".to_string(), Type::Int)]);
        assert_eq!(pk, vec!["id".to_string()]);
    }

    #[test]
    fn standard_sql_aliases_map_to_playground_types() {
        let (_, cols, _, _) = sct(
            "create table t (a bigint, b varchar, c boolean, d timestamp, \
             e numeric, f float, g binary)",
        );
        assert_eq!(
            cols,
            vec![
                ("a".to_string(), Type::Int),
                ("b".to_string(), Type::Text),
                ("c".to_string(), Type::Bool),
                ("d".to_string(), Type::DateTime),
                ("e".to_string(), Type::Decimal),
                ("f".to_string(), Type::Real),
                ("g".to_string(), Type::Blob),
            ]
        );
    }

    #[test]
    fn double_precision_maps_to_real() {
        let (_, cols, _, _) = sct("create table t (id int, x double precision)");
        assert_eq!(
            cols,
            vec![("id".to_string(), Type::Int), ("x".to_string(), Type::Real)]
        );
    }

    #[test]
    fn length_args_are_ignored() {
        let (_, cols, _, _) = sct("create table t (a varchar(255), b numeric(10, 2))");
        assert_eq!(
            cols,
            vec![("a".to_string(), Type::Text), ("b".to_string(), Type::Decimal)]
        );
    }

    #[test]
    fn column_level_primary_key_populates_pk() {
        let (_, _, pk, _) = sct("create table t (id serial primary key, name text)");
        assert_eq!(pk, vec!["id".to_string()]);
    }

    #[test]
    fn table_level_compound_primary_key() {
        let (_, _, pk, _) = sct("create table t (a int, b int, primary key (a, b))");
        assert_eq!(pk, vec!["a".to_string(), "b".to_string()]);
    }

    #[test]
    fn if_not_exists_sets_the_flag() {
        let (name, _, _, ine) = sct("create table if not exists t (id int)");
        assert_eq!(name, "t");
        assert!(ine);
    }

    #[test]
    fn not_null_and_unique_attach_to_their_column() {
        match parse_command("create table t (id int primary key, code text not null unique)")
            .expect("parses")
        {
            Command::SqlCreateTable { columns, .. } => {
                let code = columns.iter().find(|c| c.name == "code").expect("code col");
                assert!(code.not_null && code.unique);
            }
            other => panic!("expected SqlCreateTable, got {other:?}"),
        }
    }

    #[test]
    fn redundant_constraints_deduped_off_sole_pk_column() {
        // ADR-0035 §6.5: advanced mode accepts the redundant spelling
        // and silently drops the flags off the sole PK column.
        match parse_command("create table t (id int primary key not null unique)")
            .expect("parses")
        {
            Command::SqlCreateTable {
                columns,
                primary_key,
                ..
            } => {
                assert_eq!(primary_key, vec!["id".to_string()]);
                let id = &columns[0];
                assert!(!id.not_null && !id.unique, "redundant flags deduped off PK");
            }
            other => panic!("expected SqlCreateTable, got {other:?}"),
        }
    }

    #[test]
    fn simple_create_still_parses_as_dsl_in_both_modes() {
        // The shared `create` entry word: the DSL `with pk` form falls
        // back to `Command::CreateTable` even in advanced mode, and is
        // the only shape in simple mode (ADR-0035 §2 dispatch).
        for mode in [Mode::Simple, Mode::Advanced] {
            let cmd = parse_command_in_mode("create table T with pk id(serial)", mode)
                .unwrap_or_else(|e| panic!("{mode:?} should parse the DSL form: {e:?}"));
            assert!(
                matches!(cmd, Command::CreateTable { .. }),
                "{mode:?}: expected DSL CreateTable, got {cmd:?}"
            );
        }
    }

    #[test]
    fn sql_create_is_advanced_only() {
        // The SQL `( … )` form is not available in simple mode.
        assert!(
            parse_command_in_mode("create table t (id int)", Mode::Simple).is_err(),
            "SQL CREATE TABLE must not parse in simple mode"
        );
    }

    // --- 4a.2: CHECK / DEFAULT raw text + composite UNIQUE ---

    /// Parse and return the full `SqlCreateTable` columns +
    /// composite-unique constraints.
    fn parse_sct(input: &str) -> (Vec<ColumnSpec>, Vec<Vec<String>>) {
        match parse_command(input).expect("should parse") {
            Command::SqlCreateTable {
                columns,
                unique_constraints,
                ..
            } => (columns, unique_constraints),
            other => panic!("expected SqlCreateTable, got {other:?}"),
        }
    }

    fn col<'a>(cols: &'a [ColumnSpec], name: &str) -> &'a ColumnSpec {
        cols.iter().find(|c| c.name == name).expect("column")
    }

    #[test]
    fn check_captures_raw_inner_sql_text() {
        let (cols, _) = parse_sct("create table t (id int check (id > 0))");
        assert_eq!(col(&cols, "id").check_sql.as_deref(), Some("id > 0"));
    }

    #[test]
    fn check_with_nested_parens_captures_balanced_text() {
        let (cols, _) = parse_sct("create table t (a int, b int check ((a + b) > 0))");
        assert_eq!(col(&cols, "b").check_sql.as_deref(), Some("(a + b) > 0"));
    }

    #[test]
    fn default_captures_raw_sql_text() {
        let (cols, _) =
            parse_sct("create table t (id int primary key, n int default 42, s text default 'x')");
        assert_eq!(col(&cols, "n").default_sql.as_deref(), Some("42"));
        assert_eq!(col(&cols, "s").default_sql.as_deref(), Some("'x'"));
    }

    #[test]
    fn default_expression_stops_before_following_constraint() {
        // The boundary case: `default 0 not null` — the expr is just
        // `0`; `not null` is the next constraint, not part of it.
        let (cols, _) = parse_sct("create table t (id int, n int default 0 not null)");
        let n = col(&cols, "n");
        assert_eq!(n.default_sql.as_deref(), Some("0"));
        assert!(n.not_null, "NOT NULL still recognised after the default");
    }

    #[test]
    fn parenthesised_default_captures_expression_with_parens() {
        // A complex (non-literal) default must be parenthesised
        // (standard SQL); the captured text keeps the parens so it
        // re-emits as valid `DEFAULT (…)`.
        let (cols, _) = parse_sct("create table t (id int, n int default (1 + 2) not null)");
        let n = col(&cols, "n");
        assert_eq!(n.default_sql.as_deref(), Some("(1 + 2)"));
        assert!(n.not_null);
    }

    #[test]
    fn composite_unique_collected_as_constraint() {
        let (cols, uniq) = parse_sct("create table t (a int, b int, unique (a, b))");
        assert_eq!(uniq, vec![vec!["a".to_string(), "b".to_string()]]);
        // The columns themselves are not individually unique.
        assert!(!col(&cols, "a").unique && !col(&cols, "b").unique);
    }

    #[test]
    fn single_column_table_unique_folds_into_the_column() {
        let (cols, uniq) = parse_sct("create table t (a int, b text, unique (b))");
        assert!(uniq.is_empty(), "single-column UNIQUE is not a composite");
        assert!(col(&cols, "b").unique, "it folds into the column's flag");
        assert!(!col(&cols, "a").unique);
    }

    // --- 4a.3: table-level / multi-column CHECK ---

    /// Parse and return the columns + the table-level CHECK constraints.
    fn parse_sct_checks(input: &str) -> (Vec<ColumnSpec>, Vec<String>) {
        match parse_command(input).expect("should parse") {
            Command::SqlCreateTable {
                columns,
                check_constraints,
                ..
            } => (columns, check_constraints),
            other => panic!("expected SqlCreateTable, got {other:?}"),
        }
    }

    #[test]
    fn table_level_check_captured_as_raw_text() {
        let (cols, checks) = parse_sct_checks("create table t (a int, b int, check (a < b))");
        assert_eq!(checks, vec!["a < b".to_string()]);
        // The CHECK belongs to no column — it stays table-level.
        assert!(col(&cols, "a").check_sql.is_none() && col(&cols, "b").check_sql.is_none());
    }

    #[test]
    fn multiple_table_checks_preserve_declaration_order() {
        let (_, checks) =
            parse_sct_checks("create table t (a int, b int, c int, check (a < b), check (b < c))");
        assert_eq!(checks, vec!["a < b".to_string(), "b < c".to_string()]);
    }

    #[test]
    fn column_check_and_table_check_route_separately() {
        // A column-level CHECK (after a column's type) and a table-level
        // CHECK (its own element) in the same statement must not be
        // conflated — the load-bearing distinction of 4a.3.
        let (cols, checks) = parse_sct_checks(
            "create table t (price real check (price >= 0), total real, check (total >= price))",
        );
        assert_eq!(col(&cols, "price").check_sql.as_deref(), Some("price >= 0"));
        assert!(col(&cols, "total").check_sql.is_none());
        assert_eq!(checks, vec!["total >= price".to_string()]);
    }

    #[test]
    fn column_check_after_length_arg_stays_column_level() {
        // The depth trap: the `,` inside `numeric(10, 2)` is at paren
        // depth 2, not an element boundary, so the following `check`
        // is still column-level. A naive "reset on any comma" would
        // misclassify it as table-level (the §4.2 probe).
        let (cols, checks) =
            parse_sct_checks("create table t (n numeric(10, 2) check (n > 0))");
        assert_eq!(col(&cols, "n").check_sql.as_deref(), Some("n > 0"));
        assert!(checks.is_empty(), "no table-level CHECK was produced");
    }

    #[test]
    fn table_check_after_table_primary_key() {
        // A table-PK `(a, b)` injects its own parens/comma into the
        // item stream; the following table CHECK must still be detected.
        let (_, checks) =
            parse_sct_checks("create table t (a int, b int, primary key (a, b), check (a < b))");
        assert_eq!(checks, vec!["a < b".to_string()]);
    }

    #[test]
    fn table_check_after_table_unique() {
        let (_, checks) =
            parse_sct_checks("create table t (a int, b int, unique (a, b), check (a <> b))");
        assert_eq!(checks, vec!["a <> b".to_string()]);
    }

    #[test]
    fn table_check_captures_balanced_nested_parens() {
        let (_, checks) =
            parse_sct_checks("create table t (a int, b int, check ((a + b) > (a - b)))");
        assert_eq!(checks, vec!["(a + b) > (a - b)".to_string()]);
    }

    #[test]
    fn table_check_before_a_later_column_is_table_level() {
        // A CHECK element that appears between columns (not after a
        // column's type) is table-level even though more columns follow.
        let (cols, checks) =
            parse_sct_checks("create table t (a int, check (a > 0), b int)");
        assert_eq!(checks, vec!["a > 0".to_string()]);
        assert!(col(&cols, "a").check_sql.is_none() && col(&cols, "b").check_sql.is_none());
    }

    // --- 4b: foreign keys (inline + table-level) ---

    /// Parse and return the foreign keys.
    fn parse_sct_fks(input: &str) -> Vec<SqlForeignKey> {
        match parse_command(input).expect("should parse") {
            Command::SqlCreateTable { foreign_keys, .. } => foreign_keys,
            other => panic!("expected SqlCreateTable, got {other:?}"),
        }
    }

    #[test]
    fn inline_reference_captured() {
        let fks = parse_sct_fks("create table t (id int, pid int references parent(id))");
        assert_eq!(fks.len(), 1);
        let fk = &fks[0];
        assert_eq!(fk.name, None, "inline FK is auto-named at execution");
        assert_eq!(fk.child_column, "pid");
        assert_eq!(fk.parent_table, "parent");
        assert_eq!(fk.parent_column.as_deref(), Some("id"));
        assert_eq!(fk.on_delete, ReferentialAction::NoAction);
        assert_eq!(fk.on_update, ReferentialAction::NoAction);
    }

    #[test]
    fn bare_inline_reference_has_no_parent_column() {
        let fks = parse_sct_fks("create table t (id int, pid int references parent)");
        assert_eq!(fks[0].parent_column, None, "bare REFERENCES — resolved at execution");
        assert_eq!(fks[0].parent_table, "parent");
        assert_eq!(fks[0].child_column, "pid");
    }

    #[test]
    fn inline_reference_with_referential_actions() {
        let fks = parse_sct_fks(
            "create table t (id int, pid int references parent(id) \
             on delete cascade on update set null)",
        );
        assert_eq!(fks[0].on_delete, ReferentialAction::Cascade);
        assert_eq!(fks[0].on_update, ReferentialAction::SetNull);
    }

    #[test]
    fn referential_action_order_is_flexible() {
        // `on update` before `on delete` — either order is accepted.
        let fks = parse_sct_fks(
            "create table t (id int, pid int references parent(id) \
             on update restrict on delete no action)",
        );
        assert_eq!(fks[0].on_update, ReferentialAction::Restrict);
        assert_eq!(fks[0].on_delete, ReferentialAction::NoAction);
    }

    #[test]
    fn table_level_foreign_key_captured() {
        let fks =
            parse_sct_fks("create table t (id int, pid int, foreign key (pid) references parent(id))");
        assert_eq!(fks.len(), 1);
        assert_eq!(fks[0].name, None);
        assert_eq!(fks[0].child_column, "pid");
        assert_eq!(fks[0].parent_table, "parent");
        assert_eq!(fks[0].parent_column.as_deref(), Some("id"));
    }

    #[test]
    fn table_level_foreign_key_with_constraint_name() {
        let fks = parse_sct_fks(
            "create table t (id int, pid int, \
             constraint fk_parent foreign key (pid) references parent(id))",
        );
        assert_eq!(fks[0].name.as_deref(), Some("fk_parent"));
        assert_eq!(fks[0].child_column, "pid");
    }

    #[test]
    fn multiple_foreign_keys_collected_in_order() {
        let fks = parse_sct_fks(
            "create table t (id int, a int, b int, \
             foreign key (a) references p(id), foreign key (b) references q(id))",
        );
        assert_eq!(fks.len(), 2);
        assert_eq!((fks[0].child_column.as_str(), fks[0].parent_table.as_str()), ("a", "p"));
        assert_eq!((fks[1].child_column.as_str(), fks[1].parent_table.as_str()), ("b", "q"));
    }

    #[test]
    fn self_referencing_foreign_key_captured() {
        let fks =
            parse_sct_fks("create table emp (id int primary key, mgr int references emp(id))");
        assert_eq!(fks[0].parent_table, "emp", "self-reference");
        assert_eq!(fks[0].child_column, "mgr");
        assert_eq!(fks[0].parent_column.as_deref(), Some("id"));
    }

    #[test]
    fn inline_fk_coexists_with_check_and_pk() {
        // FK clause must not be confused with the column CHECK that
        // follows, nor disturb the table-level PK / CHECK detection.
        match parse_command(
            "create table t (id int primary key, pid int references parent(id) check (pid > 0), \
             check (id <> pid))",
        )
        .expect("parses")
        {
            Command::SqlCreateTable {
                primary_key,
                foreign_keys,
                check_constraints,
                columns,
                ..
            } => {
                assert_eq!(primary_key, vec!["id".to_string()]);
                assert_eq!(foreign_keys.len(), 1);
                assert_eq!(foreign_keys[0].child_column, "pid");
                // the column-level CHECK still attaches to `pid`
                assert_eq!(
                    columns.iter().find(|c| c.name == "pid").unwrap().check_sql.as_deref(),
                    Some("pid > 0")
                );
                // the table-level CHECK is captured separately
                assert_eq!(check_constraints, vec!["id <> pid".to_string()]);
            }
            other => panic!("expected SqlCreateTable, got {other:?}"),
        }
    }
}