reddb_rql/

core.rs

use std::fmt;

use super::builders::{GraphQueryBuilder, PathQueryBuilder, TableQueryBuilder};
use reddb_types::catalog::CollectionModel;
pub use reddb_types::distance::DistanceMetric;
pub use reddb_types::queue_mode::QueueMode;
use reddb_types::types::{SqlTypeName, Value};
pub use reddb_types::vector_metadata::MetadataFilter;

/// Root query expression
#[derive(Debug, Clone)]
#[allow(clippy::large_enum_variant)]
pub enum QueryExpr {
    /// Persisted function definition, invocation or catalog inspection.
    Function(Box<crate::stored_function::FunctionCommand>),
    /// Pure table query: SELECT ... FROM ...
    Table(TableQuery),
    /// Pure graph query: MATCH ... RETURN ...
    Graph(GraphQuery),
    /// Join between table and graph
    Join(JoinQuery),
    /// Path query: PATH FROM ... TO ...
    Path(PathQuery),
    /// Vector similarity search
    Vector(VectorQuery),
    /// Hybrid query combining structured and vector search
    Hybrid(HybridQuery),
    /// INSERT INTO table (cols) VALUES (vals)
    Insert(InsertQuery),
    /// UPDATE table SET col=val WHERE filter
    Update(UpdateQuery),
    /// DELETE FROM table WHERE filter
    Delete(DeleteQuery),
    /// CREATE TABLE name (columns)
    CreateTable(CreateTableQuery),
    /// CREATE COLLECTION name KIND kind
    CreateCollection(CreateCollectionQuery),
    /// CREATE VECTOR name DIM n [METRIC metric]
    CreateVector(CreateVectorQuery),
    /// DROP TABLE name
    DropTable(DropTableQuery),
    /// DROP GRAPH name
    DropGraph(DropGraphQuery),
    /// DROP VECTOR name
    DropVector(DropVectorQuery),
    /// DROP DOCUMENT name
    DropDocument(DropDocumentQuery),
    /// DROP KV name
    DropKv(DropKvQuery),
    /// DROP COLLECTION name
    DropCollection(DropCollectionQuery),
    /// TRUNCATE [model] name
    Truncate(TruncateQuery),
    /// ALTER TABLE name ADD/DROP/RENAME COLUMN
    AlterTable(AlterTableQuery),
    /// CREATE BRANCH/TAG ref DDL
    CreateVcsRef(CreateVcsRefQuery),
    /// DROP BRANCH/TAG ref DDL
    DropVcsRef(DropVcsRefQuery),
    /// FORK STORE AS name [AT LSN n]
    ForkStore(ForkStoreQuery),
    /// PROMOTE FORK name
    PromoteFork(PromoteForkQuery),
    /// DROP FORK name
    DropFork(DropForkQuery),
    /// GRAPH subcommand (NEIGHBORHOOD, SHORTEST_PATH, etc.)
    GraphCommand(GraphCommand),
    /// SEARCH subcommand (SIMILAR, TEXT, HYBRID)
    SearchCommand(SearchCommand),
    /// ASK 'question' — RAG query with LLM synthesis
    Ask(AskQuery),
    /// CREATE INDEX name ON table (columns) USING type
    CreateIndex(CreateIndexQuery),
    /// DROP INDEX name ON table
    DropIndex(DropIndexQuery),
    /// Probabilistic data structure commands (HLL, SKETCH, FILTER)
    ProbabilisticCommand(ProbabilisticCommand),
    /// CREATE TIMESERIES name [RETENTION duration] [CHUNK_SIZE n]
    CreateTimeSeries(CreateTimeSeriesQuery),
    /// CREATE METRIC path TYPE kind ROLE role
    CreateMetric(CreateMetricQuery),
    /// ALTER METRIC path SET <ROLE|KIND|TYPE|PATH> <value>
    AlterMetric(AlterMetricQuery),
    /// CREATE SLO path ON metric_path TARGET t WINDOW d UNIT
    CreateSlo(CreateSloQuery),
    /// DROP TIMESERIES name
    DropTimeSeries(DropTimeSeriesQuery),
    /// CREATE QUEUE name [WORK|STANDARD|FIFO|FANOUT] [MAX_SIZE n] [PRIORITY] [WITH TTL duration]
    CreateQueue(CreateQueueQuery),
    /// ALTER QUEUE name SET MODE [FANOUT|WORK|STANDARD|FIFO]
    AlterQueue(AlterQueueQuery),
    /// DROP QUEUE name
    DropQueue(DropQueueQuery),
    /// Read-only queue projection: SELECT ... FROM QUEUE name
    QueueSelect(QueueSelectQuery),
    /// QUEUE subcommand (PUSH, POP, PEEK, LEN, PURGE, GROUP, READ, ACK, NACK)
    QueueCommand(QueueCommand),
    /// KV subcommand (PUT, GET, DELETE)
    KvCommand(KvCommand),
    /// CONFIG keyed command (PUT, GET, ROTATE, DELETE, HISTORY)
    ConfigCommand(ConfigCommand),
    /// CREATE TREE name IN collection ROOT ... MAX_CHILDREN n
    CreateTree(CreateTreeQuery),
    /// DROP TREE name IN collection
    DropTree(DropTreeQuery),
    /// TREE subcommand (INSERT, MOVE, DELETE, VALIDATE, REBALANCE)
    TreeCommand(TreeCommand),
    /// SET CONFIG key = value
    SetConfig { key: String, value: Value },
    /// SHOW CONFIG [prefix] [AS JSON|FORMAT JSON]
    ShowConfig {
        prefix: Option<String>,
        as_json: bool,
    },
    /// SCRUB [BACKGROUND BUDGET n]
    Scrub {
        background: bool,
        budget: Option<u64>,
    },
    /// SET SECRET key = value
    SetSecret { key: String, value: Value },
    /// DELETE SECRET key
    DeleteSecret { key: String },
    /// SHOW SECRET[S] [prefix]
    ShowSecrets { prefix: Option<String> },
    /// SET KV key = value
    ///
    /// Stores a plain (non-encrypted) user key-value entry in the KV
    /// store — a sibling to `SET SECRET`, but backed by an independent
    /// flat-map with no vault encryption. Gated by `kv:write` (#1602).
    SetKv { key: String, value: Value },
    /// DELETE KV key — remove a plain KV entry. Gated by `kv:write` (#1602).
    DeleteKv { key: String },
    /// `SET TENANT 'id'` / `SET TENANT = 'id'` / `RESET TENANT`
    ///
    /// Session-scoped multi-tenancy handle. Populates a per-connection
    /// thread-local that `CURRENT_TENANT()` reads and that RLS
    /// policies combine with via `USING (tenant_id = CURRENT_TENANT())`.
    /// `None` clears the current tenant (RESET TENANT or SET TENANT
    /// NULL). Unlike `SetConfig` this is *not* persisted to red_config —
    /// it lives for the connection's lifetime only.
    SetTenant(Option<String>),
    /// `SHOW TENANT` — returns the thread-local tenant id (or NULL).
    ShowTenant,
    /// EXPLAIN ALTER FOR CREATE TABLE name (...) [FORMAT JSON]
    ///
    /// Pure read command that diffs the embedded `CREATE TABLE`
    /// statement against the live `CollectionContract` of the
    /// table with the same name and returns the `ALTER TABLE`
    /// operations that would close the gap. Never executes
    /// anything — output is text (default) or JSON depending on
    /// the optional `FORMAT JSON` suffix. Powers the Purple
    /// framework's migration generator and any other client that
    /// wants reddb to own the schema-diff rules.
    ExplainAlter(ExplainAlterQuery),
    /// CREATE MIGRATION name [DEPENDS ON dep1, dep2] [BATCH n ROWS] [NO ROLLBACK] body
    CreateMigration(CreateMigrationQuery),
    /// APPLY MIGRATION name | APPLY MIGRATION *  [FOR TENANT id]
    ApplyMigration(ApplyMigrationQuery),
    /// ROLLBACK MIGRATION name
    RollbackMigration(RollbackMigrationQuery),
    /// EXPLAIN MIGRATION name
    ExplainMigration(ExplainMigrationQuery),
    /// EXPLAIN <statement>
    ///
    /// Plan-only wrapper: the executor must inspect the wrapped statement
    /// without executing it.
    Explain(ExplainQuery),
    /// `EVENTS BACKFILL collection [WHERE pred] TO queue [LIMIT n]`.
    EventsBackfill(EventsBackfillQuery),
    /// `EVENTS BACKFILL STATUS collection` placeholder for the status slice.
    EventsBackfillStatus { collection: String },
    /// Transaction control: BEGIN, COMMIT, ROLLBACK, SAVEPOINT, RELEASE, ROLLBACK TO.
    ///
    /// Phase 1.1 (PG parity): parser + dispatch are wired so clients (psql, JDBC, etc.)
    /// can issue these statements without errors. Real isolation/atomicity semantics
    /// arrive with Phase 2.3 MVCC. Until then statements behave as autocommit (each
    /// DML is its own transaction); BEGIN/COMMIT/ROLLBACK return success but do NOT
    /// provide rollback-on-failure guarantees across multiple statements.
    TransactionControl(TxnControl),
    /// Maintenance commands: VACUUM [FULL] [table], ANALYZE [table].
    ///
    /// Phase 1.2 (PG parity): `VACUUM` triggers segment/page flush + planner stats
    /// refresh. `ANALYZE` refreshes planner statistics (histograms, null counts,
    /// distinct estimates). Both accept an optional table target; omitting the
    /// target iterates every collection.
    MaintenanceCommand(MaintenanceCommand),
    /// VCS working-set commands: CHECKPOINT, CHECKOUT, RESET, MERGE,
    /// CHERRY PICK, REVERT, RESOLVE CONFLICT.
    VcsCommand(VcsCommand),
    /// `CREATE SCHEMA [IF NOT EXISTS] name`
    ///
    /// Phase 1.3 (PG parity): schemas are logical namespaces stored in
    /// `red_config` under the key `schema.{name}`. Tables created inside a
    /// schema use `schema.table` qualified names (collection name = "schema.table").
    CreateSchema(CreateSchemaQuery),
    /// `DROP SCHEMA [IF EXISTS] name [CASCADE]`
    DropSchema(DropSchemaQuery),
    /// `CREATE SEQUENCE [IF NOT EXISTS] name [START [WITH] n] [INCREMENT [BY] n]`
    ///
    /// Phase 1.3 (PG parity): sequences are 64-bit monotonic counters persisted
    /// in `red_config` under the key `sequence.{name}`. Values are produced by
    /// the scalar functions `nextval('name')` and `currval('name')`.
    CreateSequence(CreateSequenceQuery),
    /// `DROP SEQUENCE [IF EXISTS] name`
    DropSequence(DropSequenceQuery),
    /// `COPY table FROM 'path' [WITH ...]` — CSV import (Phase 1.5 PG parity).
    ///
    /// Supported options: `DELIMITER c`, `HEADER [true|false]`. Rows stream
    /// into the named collection via the `CsvImporter`.
    CopyFrom(CopyFromQuery),
    /// `CREATE [OR REPLACE] [MATERIALIZED] VIEW [IF NOT EXISTS] name AS SELECT ...`
    ///
    /// Phase 2.1 (PG parity): views are stored as `view.{name}` entries in
    /// `red_config`. Materialized views additionally allocate a slot in the
    /// shared `MaterializedViewCache`; `REFRESH MATERIALIZED VIEW` re-runs
    /// the underlying query and repopulates the cache.
    CreateView(CreateViewQuery),
    /// `DROP [MATERIALIZED] VIEW [IF EXISTS] name`
    DropView(DropViewQuery),
    /// `REFRESH MATERIALIZED VIEW name`
    ///
    /// Re-executes the view's query and writes the result into the cache.
    RefreshMaterializedView(RefreshMaterializedViewQuery),
    /// `CREATE POLICY name ON table [FOR action] [TO role] USING (filter)`
    ///
    /// Phase 2.5 (PG parity): row-level security policy definition.
    /// Evaluated at read time — when the table has RLS enabled, all
    /// matching policies for the current role are combined with OR and
    /// AND-ed into the query's WHERE clause.
    CreatePolicy(CreatePolicyQuery),
    /// `DROP POLICY [IF EXISTS] name ON table`
    DropPolicy(DropPolicyQuery),
    /// `CREATE SERVER name FOREIGN DATA WRAPPER kind OPTIONS (...)`
    /// (Phase 3.2 PG parity). Registers a named foreign-data-wrapper
    /// instance in the runtime's `ForeignTableRegistry`.
    CreateServer(CreateServerQuery),
    /// `DROP SERVER [IF EXISTS] name [CASCADE]`
    DropServer(DropServerQuery),
    /// `CREATE FOREIGN TABLE name (cols) SERVER srv OPTIONS (...)`
    /// (Phase 3.2 PG parity). Makes `name` resolvable as a foreign table
    /// via the parent server's `ForeignDataWrapper`.
    CreateForeignTable(CreateForeignTableQuery),
    /// `DROP FOREIGN TABLE [IF EXISTS] name`
    DropForeignTable(DropForeignTableQuery),
    /// `GRANT { actions | ALL [PRIVILEGES] }
    ///   ON { TABLE | SCHEMA | DATABASE | FUNCTION } object_list
    ///   TO grant_principal_list
    ///   [WITH GRANT OPTION]`
    ///
    /// Granular RBAC primitive layered on top of the legacy 3-role model.
    /// See `crate::auth::privileges` for the resolution algorithm.
    Grant(GrantStmt),
    /// `REVOKE [GRANT OPTION FOR] { actions | ALL } ON … FROM …`
    Revoke(RevokeStmt),
    /// `ALTER USER name [VALID UNTIL 'ts'] [CONNECTION LIMIT n]
    ///   [ENABLE | DISABLE] [SET search_path = ...]`
    AlterUser(AlterUserStmt),
    /// `CREATE USER [tenant.]name [WITH] PASSWORD 'plaintext' [ROLE read|write|admin]`
    CreateUser(CreateUserStmt),
    // ----- IAM policy DDL (Agent #28 / IAM kernel integration) -----
    /// `CREATE POLICY '<id>' AS '<json>'` — installs an IAM policy
    /// document in the AuthStore. Distinct from the RLS-flavoured
    /// `CreatePolicy(CreatePolicyQuery)` above (which uses
    /// `CREATE POLICY name ON table ...`); the parser disambiguates
    /// at parse time by inspecting the token after the policy name.
    CreateIamPolicy { id: String, json: String },
    /// `DROP POLICY '<id>'` — removes an IAM policy and its
    /// attachments.
    DropIamPolicy { id: String },
    /// `ATTACH POLICY '<id>' TO USER <name>` /
    /// `ATTACH POLICY '<id>' TO GROUP <name>`.
    AttachPolicy {
        policy_id: String,
        principal: PolicyPrincipalRef,
    },
    /// `DETACH POLICY '<id>' FROM USER <name>` /
    /// `DETACH POLICY '<id>' FROM GROUP <name>`.
    DetachPolicy {
        policy_id: String,
        principal: PolicyPrincipalRef,
    },
    /// `SHOW POLICIES [FOR USER <name> | FOR GROUP <name>]`.
    ShowPolicies { filter: Option<PolicyPrincipalRef> },
    /// `SHOW EFFECTIVE PERMISSIONS FOR <name> [ON <kind>:<name>]`.
    ShowEffectivePermissions {
        user: PolicyUserRef,
        resource: Option<PolicyResourceRef>,
    },
    /// Exact rank of one row in a declared leaderboard ranking.
    RankOf(RankOfQuery),
    /// Approximate tail rank of one row in a declared leaderboard ranking.
    ApproxRankOf(RankOfQuery),
    /// Exact rank-ordered range in a declared leaderboard ranking.
    RankRange(RankRangeQuery),
    /// `SIMULATE <name> ACTION <verb> ON <kind>:<name>`.
    SimulatePolicy {
        user: PolicyUserRef,
        action: String,
        resource: PolicyResourceRef,
    },
    /// `LINT POLICY '<id>'` — fetch a stored policy from the
    /// AuthStore and run the [`PolicyLinter`](crate::auth::policy_linter)
    /// against its serialized JSON.
    ///
    /// `LINT POLICY JSON '<json>'` — lint the supplied JSON document
    /// directly without consulting the AuthStore. Issue #710.
    LintPolicy { source: LintPolicySource },
    /// `MIGRATE POLICY MODE TO '<target>' [DRY RUN]` — switch the
    /// install from the legacy_rbac fallback to strict policy_only
    /// after running the pre-flight delta simulator. With `DRY RUN`,
    /// only the delta is returned. Without it, the migration refuses
    /// if the delta is non-empty and otherwise mutates the
    /// enforcement mode. Issue #714.
    MigratePolicyMode { target: String, dry_run: bool },
}

/// Source of the policy document being linted.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LintPolicySource {
    /// Fetch the document from the AuthStore by id.
    Id(String),
    /// Use the supplied JSON literal verbatim.
    Json(String),
}

/// Tenant-qualified user reference for IAM policy SQL DDL.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PolicyUserRef {
    pub tenant: Option<String>,
    pub username: String,
}

/// Resource reference (`<kind>:<name>`) used in `SIMULATE` /
/// `SHOW EFFECTIVE PERMISSIONS`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PolicyResourceRef {
    pub kind: String,
    pub name: String,
}

/// Principal target for ATTACH / DETACH / SHOW POLICIES filter.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PolicyPrincipalRef {
    User(PolicyUserRef),
    Group(String),
}

/// `RANK OF <entity_id> IN <ranking>` canonical leaderboard read.
///
/// Redis-flavor `ZRANK <ranking> <entity_id>` desugars to this exact same
/// shape, so it carries no execution semantics that the canonical rank read
/// lacks.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RankOfQuery {
    pub ranking: String,
    pub entity_id: u64,
}

/// `RANK RANGE <lo> TO <hi> IN <ranking>` canonical leaderboard read.
///
/// Redis-flavor `ZRANGE <ranking> <start> <stop> [WITHSCORES]` desugars by
/// translating Redis' zero-based inclusive offsets to the canonical one-based
/// rank bounds.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RankRangeQuery {
    pub ranking: String,
    pub lo: u64,
    pub hi: u64,
}

// ---------------------------------------------------------------------------
// GRANT / REVOKE / ALTER USER AST
// ---------------------------------------------------------------------------

/// Object class targeted by a GRANT/REVOKE.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GrantObjectKind {
    Table,
    Schema,
    Database,
    Function,
}

/// One target object in a `GRANT ... ON ... <object_list>` clause.
///
/// `name` follows the parser's standard `[schema.]object` shape; the
/// optional `schema` is only populated for `Table` / `Function` and
/// stays `None` when the user wrote a bare identifier.
#[derive(Debug, Clone)]
pub struct GrantObject {
    pub schema: Option<String>,
    pub name: String,
}

/// Principal target of a GRANT (i.e. the recipient).
#[derive(Debug, Clone)]
pub enum GrantPrincipalRef {
    /// `TO username` — username may include an `@tenant` suffix.
    User {
        tenant: Option<String>,
        name: String,
    },
    /// `TO PUBLIC`.
    Public,
    /// `TO GROUP groupname` (parsed today, enforcement deferred).
    Group(String),
}

/// `GRANT` statement AST.
#[derive(Debug, Clone)]
pub struct GrantStmt {
    /// Privilege keywords as the user typed them, normalised to upper
    /// case. Matches the `Action` set in `crate::auth::privileges`. An
    /// empty list together with `all = true` represents `ALL [PRIVILEGES]`.
    pub actions: Vec<String>,
    /// Optional column list — populates the AST for column-level
    /// grants but enforcement is deferred (stretch goal).
    pub columns: Option<Vec<String>>,
    pub object_kind: GrantObjectKind,
    pub objects: Vec<GrantObject>,
    pub principals: Vec<GrantPrincipalRef>,
    pub with_grant_option: bool,
    /// `true` when the privilege list was `ALL [PRIVILEGES]`.
    pub all: bool,
}

/// `REVOKE` statement AST.
#[derive(Debug, Clone)]
pub struct RevokeStmt {
    pub actions: Vec<String>,
    pub columns: Option<Vec<String>>,
    pub object_kind: GrantObjectKind,
    pub objects: Vec<GrantObject>,
    pub principals: Vec<GrantPrincipalRef>,
    /// `REVOKE GRANT OPTION FOR ...` — strips just the grant option,
    /// keeping the underlying privilege.
    pub grant_option_for: bool,
    pub all: bool,
}

/// One attribute setting under `ALTER USER`.
#[derive(Debug, Clone)]
pub enum AlterUserAttribute {
    ValidUntil(String),
    ConnectionLimit(i64),
    Enable,
    Disable,
    SetSearchPath(String),
    AddGroup(String),
    DropGroup(String),
    /// Reset password (carry the new plaintext until the runtime
    /// hands it to AuthStore::change_password). Out of scope for the
    /// initial milestone — present so the parser can accept the
    /// keyword without a follow-up grammar change.
    Password(String),
}

/// `ALTER USER` statement AST.
#[derive(Debug, Clone)]
pub struct AlterUserStmt {
    pub tenant: Option<String>,
    pub username: String,
    pub attributes: Vec<AlterUserAttribute>,
}

/// `CREATE USER` statement AST.
#[derive(Debug, Clone)]
pub struct CreateUserStmt {
    pub tenant: Option<String>,
    pub username: String,
    pub password: String,
    pub role: String,
}

#[derive(Debug, Clone)]
pub struct CreateServerQuery {
    pub name: String,
    /// Wrapper kind declared in `FOREIGN DATA WRAPPER <kind>`.
    pub wrapper: String,
    /// Generic `(key 'value', ...)` option bag.
    pub options: Vec<(String, String)>,
    pub if_not_exists: bool,
}

#[derive(Debug, Clone)]
pub struct DropServerQuery {
    pub name: String,
    pub if_exists: bool,
    pub cascade: bool,
}

#[derive(Debug, Clone)]
pub struct CreateForeignTableQuery {
    pub name: String,
    pub server: String,
    pub columns: Vec<ForeignColumnDef>,
    pub options: Vec<(String, String)>,
    pub if_not_exists: bool,
}

#[derive(Debug, Clone)]
pub struct ForeignColumnDef {
    pub name: String,
    pub data_type: String,
    pub not_null: bool,
}

#[derive(Debug, Clone)]
pub struct DropForeignTableQuery {
    pub name: String,
    pub if_exists: bool,
}

/// Row-level security policy definition.
#[derive(Debug, Clone)]
pub struct CreatePolicyQuery {
    pub name: String,
    pub table: String,
    /// Which action this policy gates. `None` = `ALL` (applies to all four).
    pub action: Option<PolicyAction>,
    /// Role the policy applies to. `None` = all roles.
    pub role: Option<String>,
    /// Boolean predicate the row must satisfy.
    pub using: Box<Filter>,
    /// Entity kind this policy targets (Phase 2.5.5 RLS universal).
    /// `CREATE POLICY p ON t ...` defaults to `Table`; writing
    /// `ON NODES OF g` / `ON VECTORS OF v` / `ON MESSAGES OF q` /
    /// `ON POINTS OF ts` / `ON EDGES OF g` targets the matching
    /// non-tabular kind. The evaluator filters polices by kind so
    /// a graph policy only gates graph reads, vector policy only
    /// gates vector reads, etc.
    pub target_kind: PolicyTargetKind,
}

/// Which flavour of entity a policy governs (Phase 2.5.5).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum PolicyTargetKind {
    Table,
    Nodes,
    Edges,
    Vectors,
    Messages,
    Points,
    Documents,
}

impl PolicyTargetKind {
    /// Lowercase identifier for UX — used in messages and the
    /// `red_config.rls.policies.*` persistence key.
    pub fn as_ident(&self) -> &'static str {
        match self {
            Self::Table => "table",
            Self::Nodes => "nodes",
            Self::Edges => "edges",
            Self::Vectors => "vectors",
            Self::Messages => "messages",
            Self::Points => "points",
            Self::Documents => "documents",
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PolicyAction {
    Select,
    Insert,
    Update,
    Delete,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DropPolicyQuery {
    pub name: String,
    pub table: String,
    pub if_exists: bool,
}

#[derive(Debug, Clone)]
pub struct CreateViewQuery {
    pub name: String,
    /// Parsed `SELECT ...` body. Stored as a boxed `QueryExpr` so the
    /// runtime can substitute the tree directly when a query references
    /// this view (no re-parsing per read).
    pub query: Box<QueryExpr>,
    pub materialized: bool,
    pub if_not_exists: bool,
    /// `CREATE OR REPLACE VIEW` — overwrites any existing definition.
    pub or_replace: bool,
    /// `REFRESH EVERY <duration>` clause — only valid on materialized
    /// views. When set, a background scheduler ticks the view at this
    /// cadence. `None` means refresh-on-demand only (slice 9 behaviour).
    /// Issue #583 slice 10.
    pub refresh_every_ms: Option<u64>,
    /// `WITH RETENTION <duration>` clause — only valid on materialized
    /// views (issue #584 slice 12). Opts the view's backing storage
    /// into a retention policy independent of the source. Persisted on
    /// the view definition; the physical sweep is applied to the
    /// MV's backing rows once slice-9's row-storage follow-up lands.
    /// `None` means the view is unaffected by source retention.
    pub retention_duration_ms: Option<u64>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DropViewQuery {
    pub name: String,
    pub materialized: bool,
    pub if_exists: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RefreshMaterializedViewQuery {
    pub name: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CopyFromQuery {
    pub table: String,
    pub path: String,
    pub format: CopyFormat,
    pub delimiter: Option<char>,
    pub has_header: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CopyFormat {
    Csv,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateSchemaQuery {
    pub name: String,
    pub if_not_exists: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DropSchemaQuery {
    pub name: String,
    pub if_exists: bool,
    pub cascade: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreateSequenceQuery {
    pub name: String,
    pub if_not_exists: bool,
    /// First value produced by `nextval`. Default 1.
    pub start: i64,
    /// Added to the current value on each `nextval`. Default 1.
    pub increment: i64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DropSequenceQuery {
    pub name: String,
    pub if_exists: bool,
}

/// Transaction-control statement variants. See [`QueryExpr::TransactionControl`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TxnControl {
    /// `BEGIN [WORK | TRANSACTION]`, `START TRANSACTION`
    Begin(Option<IsolationLevel>),
    /// `COMMIT [WORK | TRANSACTION]`, `END`
    Commit,
    /// `ROLLBACK [WORK | TRANSACTION]`
    Rollback,
    /// `SAVEPOINT name`
    Savepoint(String),
    /// `RELEASE [SAVEPOINT] name`
    ReleaseSavepoint(String),
    /// `ROLLBACK TO [SAVEPOINT] name`
    RollbackToSavepoint(String),
}

/// Transaction isolation level requested by `BEGIN ISOLATION LEVEL ...`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum IsolationLevel {
    /// `READ UNCOMMITTED`
    ReadUncommitted,
    /// `READ COMMITTED`
    ReadCommitted,
    /// `REPEATABLE READ` / `SNAPSHOT`
    #[default]
    SnapshotIsolation,
    /// `SERIALIZABLE`
    Serializable,
}

/// Maintenance command variants. See [`QueryExpr::MaintenanceCommand`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MaintenanceCommand {
    /// `VACUUM [FULL] [table]`
    ///
    /// Triggers segment compaction and planner stats refresh. `FULL` additionally
    /// forces a full pager sync. Target `None` applies to every collection.
    Vacuum { target: Option<String>, full: bool },
    /// `ANALYZE [table]`
    ///
    /// Refreshes planner statistics (histogram, distinct estimates, null counts).
    /// Target `None` re-analyzes every collection.
    Analyze { target: Option<String> },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum VcsCommand {
    Checkpoint {
        message: String,
        author: Option<String>,
    },
    Checkout {
        target: String,
    },
    Reset {
        mode: VcsResetMode,
        target: String,
    },
    Merge {
        branch: String,
    },
    CherryPick {
        commit: String,
    },
    Revert {
        commit: String,
    },
    ResolveConflict {
        key: String,
        resolution: VcsConflictResolution,
    },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum VcsResetMode {
    Hard,
    Soft,
    Mixed,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum VcsConflictResolution {
    Ours,
    Theirs,
}

/// AST node for `EXPLAIN ALTER FOR <CreateTableStmt> [FORMAT JSON]`.
///
/// `target` carries the CREATE TABLE structure exactly as the
/// parser produces it for a regular CREATE — full reuse of
/// `parse_create_table_body`. `format` determines whether the
/// executor emits a `ALTER TABLE …;`-flavored text payload
/// (the default — copy-paste friendly into the REPL) or a
/// structured JSON object (machine-friendly).
#[derive(Debug, Clone)]
pub struct ExplainAlterQuery {
    pub target: CreateTableQuery,
    pub format: ExplainFormat,
}

/// Output format requested for an `EXPLAIN ALTER` command.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ExplainFormat {
    /// Plain SQL text — `ALTER TABLE …;` lines plus header
    /// comments and rename hints. Default; copy-paste friendly.
    #[default]
    Sql,
    /// Structured JSON object with `operations`,
    /// `rename_candidates`, `summary`. Machine-friendly for
    /// driver code (Purple migration generator, dashboards,
    /// CLI tools).
    Json,
}

#[derive(Debug, Clone, PartialEq)]
pub struct CreateMigrationQuery {
    pub name: String,
    pub body: String,
    pub depends_on: Vec<String>,
    pub batch_size: Option<u64>,
    pub no_rollback: bool,
}

#[derive(Debug, Clone, PartialEq)]
pub struct ApplyMigrationQuery {
    pub target: ApplyMigrationTarget,
    pub for_tenant: Option<String>,
}

#[derive(Debug, Clone, PartialEq)]
pub enum ApplyMigrationTarget {
    Named(String),
    All,
}

#[derive(Debug, Clone, PartialEq)]
pub struct RollbackMigrationQuery {
    pub name: String,
}

#[derive(Debug, Clone, PartialEq)]
pub struct ExplainMigrationQuery {
    pub name: String,
}

#[derive(Debug, Clone)]
pub struct ExplainQuery {
    pub inner: Box<QueryExpr>,
}

/// Probabilistic data structure commands
#[derive(Debug, Clone)]
pub enum ProbabilisticCommand {
    // HyperLogLog
    CreateHll {
        name: String,
        precision: u8,
        if_not_exists: bool,
    },
    HllAdd {
        name: String,
        elements: Vec<String>,
    },
    HllCount {
        names: Vec<String>,
    },
    HllMerge {
        dest: String,
        sources: Vec<String>,
    },
    HllInfo {
        name: String,
    },
    DropHll {
        name: String,
        if_exists: bool,
    },

    // Count-Min Sketch (Fase 7)
    CreateSketch {
        name: String,
        width: usize,
        depth: usize,
        if_not_exists: bool,
    },
    SketchAdd {
        name: String,
        element: String,
        count: u64,
    },
    SketchCount {
        name: String,
        element: String,
    },
    SketchMerge {
        dest: String,
        sources: Vec<String>,
    },
    SketchInfo {
        name: String,
    },
    DropSketch {
        name: String,
        if_exists: bool,
    },

    // Cuckoo Filter (Fase 8)
    CreateFilter {
        name: String,
        capacity: usize,
        if_not_exists: bool,
    },
    FilterAdd {
        name: String,
        element: String,
    },
    FilterCheck {
        name: String,
        element: String,
    },
    FilterDelete {
        name: String,
        element: String,
    },
    FilterCount {
        name: String,
    },
    FilterInfo {
        name: String,
    },
    DropFilter {
        name: String,
        if_exists: bool,
    },
}

/// Default H3 resolution for `CREATE INDEX ... USING H3` when no
/// explicit resolution is supplied. Resolution 9 ≈ ~150 m hex edge
/// (PRD #1574 slice 2, #1576).
pub const DEFAULT_H3_RESOLUTION: u8 = 9;

/// Index type for CREATE INDEX ... USING <type>
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IndexMethod {
    BTree,
    Hash,
    Bitmap,
    /// Generic spatial index request (`USING SPATIAL`) — the engine picks
    /// the default spatial backend. As of PRD #1574 slice 4 (#1578) that
    /// default is the disk-resident H3 index (RAM stays O(working set)),
    /// not the retired in-RAM R-tree.
    Spatial,
    /// H3 hexagonal spatial index: the `(lat, lon)` geo column is
    /// encoded to a single H3 cell-id `u64` (at `resolution`) and stored
    /// in the existing disk-paged B-tree (PRD #1574 slice 2, #1576).
    H3 {
        resolution: u8,
    },
}

impl fmt::Display for IndexMethod {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::BTree => write!(f, "BTREE"),
            Self::Hash => write!(f, "HASH"),
            Self::Bitmap => write!(f, "BITMAP"),
            Self::Spatial => write!(f, "SPATIAL"),
            Self::H3 { .. } => write!(f, "H3"),
        }
    }
}

/// CREATE INDEX [UNIQUE] [IF NOT EXISTS] name ON table (col1, col2, ...) [USING method]
#[derive(Debug, Clone)]
pub struct CreateIndexQuery {
    pub name: String,
    pub table: String,
    pub columns: Vec<String>,
    pub method: IndexMethod,
    pub unique: bool,
    pub if_not_exists: bool,
}

/// DROP INDEX [IF EXISTS] name ON table
#[derive(Debug, Clone)]
pub struct DropIndexQuery {
    pub name: String,
    pub table: String,
    pub if_exists: bool,
}

/// ASK 'question' [USING provider] [MODEL 'model'] [DEPTH n] [LIMIT n] [MIN_SCORE x]
///                [COLLECTION col] [TEMPERATURE x] [SEED n] [STRICT ON|OFF] [STREAM]
///                [CACHE TTL '5m' | NOCACHE] [AS RQL]
///
/// `temperature` and `seed` are per-query overrides resolved by the
/// `DeterminismDecider` (issue #400). The parser merely surfaces the
/// requested values; capability-based dropping happens at decide time.
#[derive(Debug, Clone)]
pub struct AskQuery {
    /// `EXPLAIN ASK '...'` returns the retrieval/provider/cost plan
    /// without making the LLM call.
    pub explain: bool,
    pub question: String,
    /// Optional `$N` / `?` parameter slot for the question text.
    pub question_param: Option<usize>,
    pub provider: Option<String>,
    pub model: Option<String>,
    pub depth: Option<usize>,
    pub limit: Option<usize>,
    pub min_score: Option<f32>,
    pub collection: Option<String>,
    /// Per-query temperature override (`ASK '...' TEMPERATURE 0.7`).
    /// `None` means fall back to `ask.default_temperature`.
    pub temperature: Option<f32>,
    /// Per-query seed override (`ASK '...' SEED 42`). `None` means the
    /// decider derives one from `hash(question + sources_fingerprint)`.
    pub seed: Option<u64>,
    /// Strict citation validation is on by default. `STRICT OFF` keeps
    /// citation diagnostics as warnings and skips retry/error handling.
    pub strict: bool,
    /// HTTP-only SSE response requested via `ASK '...' STREAM`.
    pub stream: bool,
    /// Per-query answer-cache override.
    pub cache: AskCacheClause,
    /// `ASK '...' PLAN` returns the typed plan (routed intent + candidate
    /// query) without executing the candidate and without the synthesis
    /// call (ADR 0068, #1751). The `EXECUTE` and `AS RQL` clauses were
    /// removed in the same clean break: read-only candidates auto-execute
    /// by default, and `PLAN` is the inspection-only surface.
    pub plan_only: bool,
    /// Per-query plan-step budget request (`ASK '...' STEPS N`). Clamped to
    /// `red.config.ai.ask.max_plan_steps` at planning time and never exceeds
    /// it; `None` falls back to the configured cap.
    pub steps: Option<usize>,
}

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum AskCacheClause {
    #[default]
    Default,
    CacheTtl(String),
    NoCache,
}

impl QueryExpr {
    /// Create a table query
    pub fn table(name: &str) -> TableQueryBuilder {
        TableQueryBuilder::new(name)
    }

    /// Create a graph query
    pub fn graph() -> GraphQueryBuilder {
        GraphQueryBuilder::new()
    }

    /// Create a path query
    pub fn path(from: NodeSelector, to: NodeSelector) -> PathQueryBuilder {
        PathQueryBuilder::new(from, to)
    }
}

// ============================================================================
// Table Query
// ============================================================================

/// Table query: SELECT columns FROM table WHERE filter ORDER BY ... LIMIT ...
#[derive(Debug, Clone)]
pub struct TableQuery {
    /// Table name. Legacy slot — still populated even when `source`
    /// is set to a subquery so existing call sites that read
    /// `query.table.as_str()` keep compiling. When `source` is
    /// `Some(TableSource::Subquery(…))`, this field holds a synthetic
    /// sentinel name (`"__subq_NNNN"`) that runtime code must never
    /// resolve against the real schema registry.
    pub table: String,
    /// Fase 2 Week 3: structured table source. `None` means the
    /// legacy `table` field is authoritative. `Some(Name)` is the
    /// same information as `table` but in typed form. `Some(Subquery)`
    /// wires a `(SELECT …) AS alias` in a FROM position — the Fase
    /// 1.7 unlock.
    pub source: Option<TableSource>,
    /// Optional table alias
    pub alias: Option<String>,
    /// Canonical SQL select list.
    pub select_items: Vec<SelectItem>,
    /// Columns to select (empty = all)
    pub columns: Vec<Projection>,
    /// Canonical SQL WHERE clause.
    pub where_expr: Option<super::Expr>,
    /// Filter condition
    pub filter: Option<Filter>,
    /// Canonical SQL GROUP BY items.
    pub group_by_exprs: Vec<super::Expr>,
    /// GROUP BY fields
    pub group_by: Vec<String>,
    /// Canonical SQL HAVING clause.
    pub having_expr: Option<super::Expr>,
    /// HAVING filter (applied after grouping)
    pub having: Option<Filter>,
    /// Order by clauses
    pub order_by: Vec<OrderByClause>,
    /// Limit
    pub limit: Option<u64>,
    /// User-supplied-parameter slot for `LIMIT $N`. Set by the parser
    /// when the LIMIT clause references `$N`/`?` instead of a literal;
    /// cleared by the binder (`user_params::bind`) after substituting
    /// the parameter into `limit`. Mirrors the `limit_param` slot on
    /// `SearchCommand` variants — see #361 slice 11.
    pub limit_param: Option<usize>,
    /// Offset
    pub offset: Option<u64>,
    /// User-supplied-parameter slot for `OFFSET $N`. Same lifecycle as
    /// `limit_param`. See #361 slice 11.
    pub offset_param: Option<usize>,
    /// WITH EXPAND options (graph traversal, cross-ref following)
    pub expand: Option<ExpandOptions>,
    /// Time-travel anchor. When present the executor resolves this
    /// to an MVCC xid and evaluates the query against that snapshot
    /// instead of the current one. Mirrors git's `AS OF` semantics.
    pub as_of: Option<AsOfClause>,
    /// `SESSIONIZE BY <actor> GAP <duration> [ORDER BY <ts>]` operator
    /// (issue #585 slice 8). When present, the executor annotates each
    /// result row with a `session_id` column. `actor_col` / `gap_ms`
    /// may be `None` when the source collection's descriptor (slice 1
    /// `SESSION_KEY` / `SESSION_GAP`) supplies the defaults; one
    /// without the other resolved at execution time is the typed
    /// `MissingSessionKey` error.
    pub sessionize: Option<SessionizeClause>,
    /// `SELECT DISTINCT` projection quantifier. When `true` the executor
    /// deduplicates the projected output row-set (over the projected
    /// columns) before ORDER BY / LIMIT. `DISTINCT` inside an aggregate
    /// argument (`COUNT(DISTINCT x)`) is unrelated and lives on the
    /// aggregate call, not here.
    pub distinct: bool,
}

/// `SESSIONIZE BY <actor_col> GAP <duration> [ORDER BY <ts_col>]`.
#[derive(Debug, Clone, Default)]
pub struct SessionizeClause {
    /// Explicit `BY <ident>`. `None` means "default from descriptor's
    /// `SESSION_KEY`" — resolved at execution time.
    pub actor_col: Option<String>,
    /// Explicit `GAP <duration>` in milliseconds. `None` means
    /// "default from descriptor's `SESSION_GAP`".
    pub gap_ms: Option<u64>,
    /// Explicit `ORDER BY <ident>` immediately after `GAP`. When
    /// `None` the executor falls back to the collection's timestamp
    /// column (the same resolution as `retention_filter`).
    pub order_col: Option<String>,
}

/// Source spec for `AS OF` — parsed form sits in `TableQuery`, then
/// `vcs_resolve_as_of` turns it into an MVCC xid at execute time.
#[derive(Debug, Clone)]
pub enum AsOfClause {
    /// Explicit commit hash literal: `AS OF COMMIT '<hex>'`.
    Commit(String),
    /// Branch or ref: `AS OF BRANCH 'main'` or `AS OF 'refs/heads/main'`.
    Branch(String),
    /// Tag: `AS OF TAG 'v1.0'`.
    Tag(String),
    /// Unix epoch milliseconds: `AS OF TIMESTAMP 1710000000000`.
    TimestampMs(i64),
    /// Raw MVCC snapshot xid: `AS OF SNAPSHOT 12345`.
    Snapshot(u64),
}

/// Structured FROM source for a `TableQuery`. Additive alongside the
/// legacy `TableQuery.table: String` slot — callers that understand
/// this type can branch on subqueries; callers that only read `table`
/// fall back to the synthetic sentinel name and, for subqueries,
/// produce an "unknown table" error until they migrate.
#[derive(Debug, Clone)]
pub enum TableSource {
    /// Plain table reference — equivalent to the legacy `String` form.
    Name(String),
    /// A subquery in FROM position: `FROM (SELECT …) AS alias`.
    Subquery(Box<QueryExpr>),
    /// A table-valued function call in FROM position, e.g.
    /// `FROM components(g)` (issue #795). `name` is the function
    /// identifier; `args` are its positional identifier arguments;
    /// `named_args` are `key => <f64>` named arguments such as
    /// `louvain(g, resolution => 0.5)` (issue #796), preserved in source
    /// order. Positional args always precede named args.
    Function {
        name: String,
        args: Vec<String>,
        named_args: Vec<(String, f64)>,
    },
    /// A graph-analytics table-valued function whose graph is supplied
    /// inline as two subqueries instead of a graph-collection reference
    /// (issue #799), e.g.
    /// `components(nodes => (SELECT id FROM hosts), edges => (SELECT src, dst FROM links))`.
    ///
    /// Structurally distinct from `Function` so the executor can tell the
    /// inline form from the graph-collection form. `nodes`/`edges` are the
    /// two materialization subqueries (the first column of `nodes` is the
    /// node id; the first two-or-three columns of `edges` are
    /// `(source, target [, weight])`). `named_args` carries any remaining
    /// numeric named arguments (e.g. `resolution => 0.5`).
    InlineGraphFunction {
        name: String,
        nodes: Box<QueryExpr>,
        edges: Box<QueryExpr>,
        named_args: Vec<(String, f64)>,
    },
}

/// Options for WITH EXPAND clause on SELECT queries.
#[derive(Debug, Clone, Default)]
pub struct ExpandOptions {
    /// Expand via graph edges (WITH EXPAND GRAPH)
    pub graph: bool,
    /// Graph expansion depth (DEPTH n)
    pub graph_depth: usize,
    /// Expand via cross-references (WITH EXPAND CROSS_REFS)
    pub cross_refs: bool,
    /// Index hint from the optimizer (which index to prefer for this query)
    pub index_hint: Option<reddb_types::index_hint::IndexHint>,
}

impl TableQuery {
    /// Create a new table query
    pub fn new(table: &str) -> Self {
        Self {
            table: table.to_string(),
            source: None,
            alias: None,
            select_items: Vec::new(),
            columns: Vec::new(),
            where_expr: None,
            filter: None,
            group_by_exprs: Vec::new(),
            group_by: Vec::new(),
            having_expr: None,
            having: None,
            order_by: Vec::new(),
            limit: None,
            limit_param: None,
            offset: None,
            offset_param: None,
            expand: None,
            as_of: None,
            sessionize: None,
            distinct: false,
        }
    }

    /// Create a TableQuery that wraps a subquery in FROM position.
    /// The legacy `table` slot holds a synthetic sentinel so code that
    /// only reads `table.as_str()` errors loudly with a
    /// recognisable marker instead of silently treating it as a
    /// real collection.
    pub fn from_subquery(subquery: QueryExpr, alias: Option<String>) -> Self {
        let sentinel = match &alias {
            Some(a) => format!("__subq_{a}"),
            None => "__subq_anon".to_string(),
        };
        Self {
            table: sentinel,
            source: Some(TableSource::Subquery(Box::new(subquery))),
            alias,
            select_items: Vec::new(),
            columns: Vec::new(),
            where_expr: None,
            filter: None,
            group_by_exprs: Vec::new(),
            group_by: Vec::new(),
            having_expr: None,
            having: None,
            order_by: Vec::new(),
            limit: None,
            limit_param: None,
            offset: None,
            offset_param: None,
            expand: None,
            as_of: None,
            sessionize: None,
            distinct: false,
        }
    }
}

/// Canonical SQL select item for table queries.
#[derive(Debug, Clone, PartialEq)]
pub enum SelectItem {
    Wildcard,
    Expr {
        expr: super::Expr,
        alias: Option<String>,
    },
}

// ============================================================================
// Graph Query
// ============================================================================

/// Graph query: MATCH pattern WHERE filter RETURN projection
#[derive(Debug, Clone)]
pub struct GraphQuery {
    /// Optional outer alias when used as a join source
    pub alias: Option<String>,
    /// Graph pattern to match
    pub pattern: GraphPattern,
    /// Filter condition
    pub filter: Option<Filter>,
    /// Return projections
    pub return_: Vec<Projection>,
    /// Optional row limit
    pub limit: Option<u64>,
}

impl GraphQuery {
    /// Create a new graph query
    pub fn new(pattern: GraphPattern) -> Self {
        Self {
            alias: None,
            pattern,
            filter: None,
            return_: Vec::new(),
            limit: None,
        }
    }

    /// Set outer alias
    pub fn alias(mut self, alias: &str) -> Self {
        self.alias = Some(alias.to_string());
        self
    }
}

/// Graph pattern: collection of node and edge patterns
#[derive(Debug, Clone, Default)]
pub struct GraphPattern {
    /// Node patterns
    pub nodes: Vec<NodePattern>,
    /// Edge patterns connecting nodes
    pub edges: Vec<EdgePattern>,
}

impl GraphPattern {
    /// Create an empty pattern
    pub fn new() -> Self {
        Self::default()
    }

    /// Add a node pattern
    pub fn node(mut self, pattern: NodePattern) -> Self {
        self.nodes.push(pattern);
        self
    }

    /// Add an edge pattern
    pub fn edge(mut self, pattern: EdgePattern) -> Self {
        self.edges.push(pattern);
        self
    }
}

/// Node pattern: (alias:Type {properties})
#[derive(Debug, Clone)]
pub struct NodePattern {
    /// Variable alias for this node
    pub alias: String,
    /// Optional label filter. Stored as the user-supplied label string so
    /// the parser is registry-free; executors resolve it against the live
    /// [`crate::storage::engine::graph_store::LabelRegistry`].
    pub node_label: Option<String>,
    /// Property filters
    pub properties: Vec<PropertyFilter>,
}

impl NodePattern {
    /// Create a new node pattern
    pub fn new(alias: &str) -> Self {
        Self {
            alias: alias.to_string(),
            node_label: None,
            properties: Vec::new(),
        }
    }

    /// Set the label filter (string form — preferred).
    pub fn of_label(mut self, label: impl Into<String>) -> Self {
        self.node_label = Some(label.into());
        self
    }

    /// Add property filter
    pub fn with_property(mut self, name: &str, op: CompareOp, value: Value) -> Self {
        self.properties.push(PropertyFilter {
            name: name.to_string(),
            op,
            value,
        });
        self
    }
}

/// Edge pattern: -[alias:Type*min..max]->
#[derive(Debug, Clone)]
pub struct EdgePattern {
    /// Optional alias for this edge
    pub alias: Option<String>,
    /// Source node alias
    pub from: String,
    /// Target node alias
    pub to: String,
    /// Optional label filter (user-supplied string).
    pub edge_label: Option<String>,
    /// Edge direction
    pub direction: EdgeDirection,
    /// Minimum hops (for variable-length patterns)
    pub min_hops: u32,
    /// Maximum hops (for variable-length patterns)
    pub max_hops: u32,
}

impl EdgePattern {
    /// Create a new edge pattern
    pub fn new(from: &str, to: &str) -> Self {
        Self {
            alias: None,
            from: from.to_string(),
            to: to.to_string(),
            edge_label: None,
            direction: EdgeDirection::Outgoing,
            min_hops: 1,
            max_hops: 1,
        }
    }

    /// Set label filter (string form — preferred).
    pub fn of_label(mut self, label: impl Into<String>) -> Self {
        self.edge_label = Some(label.into());
        self
    }

    /// Set direction
    pub fn direction(mut self, dir: EdgeDirection) -> Self {
        self.direction = dir;
        self
    }

    /// Set hop range for variable-length patterns
    pub fn hops(mut self, min: u32, max: u32) -> Self {
        self.min_hops = min;
        self.max_hops = max;
        self
    }

    /// Set alias
    pub fn alias(mut self, alias: &str) -> Self {
        self.alias = Some(alias.to_string());
        self
    }
}

/// Edge direction
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EdgeDirection {
    /// Outgoing: (a)-[r]->(b)
    Outgoing,
    /// Incoming: (a)<-[r]-(b)
    Incoming,
    /// Both: (a)-[r]-(b)
    Both,
}

/// Property filter: name op value
#[derive(Debug, Clone)]
pub struct PropertyFilter {
    pub name: String,
    pub op: CompareOp,
    pub value: Value,
}

// ============================================================================
// Join Query
// ============================================================================

/// Join query: combines table and graph queries
#[derive(Debug, Clone)]
pub struct JoinQuery {
    /// Left side (typically table)
    pub left: Box<QueryExpr>,
    /// Right side (typically graph)
    pub right: Box<QueryExpr>,
    /// Join type
    pub join_type: JoinType,
    /// Join condition
    pub on: JoinCondition,
    /// Post-join filter condition
    pub filter: Option<Filter>,
    /// Post-join ordering
    pub order_by: Vec<OrderByClause>,
    /// Post-join limit
    pub limit: Option<u64>,
    /// Post-join offset
    pub offset: Option<u64>,
    /// Canonical SQL RETURN projection.
    pub return_items: Vec<SelectItem>,
    /// Post-join projection
    pub return_: Vec<Projection>,
}

impl JoinQuery {
    /// Create a new join query
    pub fn new(left: QueryExpr, right: QueryExpr, on: JoinCondition) -> Self {
        Self {
            left: Box::new(left),
            right: Box::new(right),
            join_type: JoinType::Inner,
            on,
            filter: None,
            order_by: Vec::new(),
            limit: None,
            offset: None,
            return_items: Vec::new(),
            return_: Vec::new(),
        }
    }

    /// Set join type
    pub fn join_type(mut self, jt: JoinType) -> Self {
        self.join_type = jt;
        self
    }
}

/// Join type
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JoinType {
    /// Inner join — only matching pairs emitted
    Inner,
    /// Left outer join — every left row, matched or padded with nulls on the right
    LeftOuter,
    /// Right outer join — every right row, matched or padded with nulls on the left
    RightOuter,
    /// Full outer join — LeftOuter ∪ RightOuter, each unmatched side padded
    FullOuter,
    /// Cross join — Cartesian product, no predicate
    Cross,
}

/// Join condition: how to match rows with nodes
#[derive(Debug, Clone)]
pub struct JoinCondition {
    /// Left field (table side)
    pub left_field: FieldRef,
    /// Right field (graph side)
    pub right_field: FieldRef,
}

impl JoinCondition {
    /// Create a new join condition
    pub fn new(left: FieldRef, right: FieldRef) -> Self {
        Self {
            left_field: left,
            right_field: right,
        }
    }
}

/// Reference to a field (table column, node property, or edge property)
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum FieldRef {
    /// Table column: table.column
    TableColumn { table: String, column: String },
    /// Node property: alias.property
    NodeProperty { alias: String, property: String },
    /// Edge property: alias.property
    EdgeProperty { alias: String, property: String },
    /// Node ID: alias.id
    NodeId { alias: String },
}

impl FieldRef {
    /// Create a table column reference
    pub fn column(table: &str, column: &str) -> Self {
        Self::TableColumn {
            table: table.to_string(),
            column: column.to_string(),
        }
    }

    /// Create a node property reference
    pub fn node_prop(alias: &str, property: &str) -> Self {
        Self::NodeProperty {
            alias: alias.to_string(),
            property: property.to_string(),
        }
    }

    /// Create a node ID reference
    pub fn node_id(alias: &str) -> Self {
        Self::NodeId {
            alias: alias.to_string(),
        }
    }

    /// Create an edge property reference
    pub fn edge_prop(alias: &str, property: &str) -> Self {
        Self::EdgeProperty {
            alias: alias.to_string(),
            property: property.to_string(),
        }
    }
}

// ============================================================================
// Path Query
// ============================================================================

/// Path query: find paths between nodes
#[derive(Debug, Clone)]
pub struct PathQuery {
    /// Optional outer alias when used as a join source
    pub alias: Option<String>,
    /// Source node selector
    pub from: NodeSelector,
    /// Target node selector
    pub to: NodeSelector,
    /// Edge labels to traverse (empty = any). Strings are resolved against
    /// the runtime registry by the executor.
    pub via: Vec<String>,
    /// Maximum path length
    pub max_length: u32,
    /// Filter on paths
    pub filter: Option<Filter>,
    /// Return projections
    pub return_: Vec<Projection>,
}

impl PathQuery {
    /// Create a new path query
    pub fn new(from: NodeSelector, to: NodeSelector) -> Self {
        Self {
            alias: None,
            from,
            to,
            via: Vec::new(),
            max_length: 10,
            filter: None,
            return_: Vec::new(),
        }
    }

    /// Set outer alias
    pub fn alias(mut self, alias: &str) -> Self {
        self.alias = Some(alias.to_string());
        self
    }

    /// Add an edge label constraint to traverse (string form).
    pub fn via_label(mut self, label: impl Into<String>) -> Self {
        self.via.push(label.into());
        self
    }
}

/// Node selector for path queries
#[derive(Debug, Clone)]
pub enum NodeSelector {
    /// By node ID
    ById(String),
    /// By node label and property
    ByType {
        node_label: String,
        filter: Option<PropertyFilter>,
    },
    /// By table row (linked node)
    ByRow { table: String, row_id: u64 },
}

impl NodeSelector {
    /// Select by node ID
    pub fn by_id(id: &str) -> Self {
        Self::ById(id.to_string())
    }

    /// Select by label string (preferred).
    pub fn by_label(label: impl Into<String>) -> Self {
        Self::ByType {
            node_label: label.into(),
            filter: None,
        }
    }

    /// Select by table row
    pub fn by_row(table: &str, row_id: u64) -> Self {
        Self::ByRow {
            table: table.to_string(),
            row_id,
        }
    }
}

// ============================================================================
// Vector Query
// ============================================================================

/// Accuracy contract for vector retrieval. Approximation requires explicit opt-in.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum VectorSearchMode {
    #[default]
    Exact,
    Approximate,
}

/// Vector similarity search query
///
/// ```text
/// VECTOR SEARCH embeddings
/// SIMILAR TO [0.1, 0.2, ..., 0.5]
/// WHERE metadata.source = 'nmap'
/// LIMIT 10
/// ```
#[derive(Debug, Clone)]
pub struct VectorQuery {
    /// Exact full-precision search by default.
    pub mode: VectorSearchMode,
    /// Optional outer alias when used as a join source
    pub alias: Option<String>,
    /// Collection name to search
    pub collection: String,
    /// Query vector (or reference to get vector from)
    pub query_vector: VectorSource,
    /// Number of results to return
    pub k: usize,
    /// Metadata filter
    pub filter: Option<MetadataFilter>,
    /// Distance metric to use (defaults to collection's metric)
    pub metric: Option<DistanceMetric>,
    /// Include vectors in results
    pub include_vectors: bool,
    /// Include metadata in results
    pub include_metadata: bool,
    /// Minimum similarity threshold (optional)
    pub threshold: Option<f32>,
}

impl VectorQuery {
    /// Create a new vector query
    pub fn new(collection: &str, query: VectorSource) -> Self {
        Self {
            mode: VectorSearchMode::Exact,
            alias: None,
            collection: collection.to_string(),
            query_vector: query,
            k: 10,
            filter: None,
            metric: None,
            include_vectors: false,
            include_metadata: true,
            threshold: None,
        }
    }

    /// Set the number of results
    pub fn limit(mut self, k: usize) -> Self {
        self.k = k;
        self
    }

    /// Set metadata filter
    pub fn with_filter(mut self, filter: MetadataFilter) -> Self {
        self.filter = Some(filter);
        self
    }

    /// Include vectors in results
    pub fn with_vectors(mut self) -> Self {
        self.include_vectors = true;
        self
    }

    /// Set similarity threshold
    pub fn min_similarity(mut self, threshold: f32) -> Self {
        self.threshold = Some(threshold);
        self
    }

    /// Set outer alias
    pub fn alias(mut self, alias: &str) -> Self {
        self.alias = Some(alias.to_string());
        self
    }
}

/// Source of query vector
#[derive(Debug, Clone)]
pub enum VectorSource {
    /// Literal vector values
    Literal(Vec<f32>),
    /// Text to embed (requires embedding function)
    Text(String),
    /// Reference to another vector by ID
    Reference { collection: String, vector_id: u64 },
    /// From a subquery result
    Subquery(Box<QueryExpr>),
}

impl VectorSource {
    /// Create from literal vector
    pub fn literal(values: Vec<f32>) -> Self {
        Self::Literal(values)
    }

    /// Create from text (to be embedded)
    pub fn text(s: &str) -> Self {
        Self::Text(s.to_string())
    }

    /// Reference another vector
    pub fn reference(collection: &str, vector_id: u64) -> Self {
        Self::Reference {
            collection: collection.to_string(),
            vector_id,
        }
    }
}

// ============================================================================
// Hybrid Query
// ============================================================================

/// Hybrid query combining structured (table/graph) and vector search
///
/// ```text
/// FROM hosts h
/// JOIN VECTOR embeddings e ON h.id = e.metadata.host_id
/// SIMILAR TO 'ssh vulnerability'
/// WHERE h.os = 'Linux'
/// RETURN h.*, e.distance
/// ```
#[derive(Debug, Clone)]
pub struct HybridQuery {
    /// Optional outer alias when used as a join source
    pub alias: Option<String>,
    /// Structured query part (table/graph)
    pub structured: Box<QueryExpr>,
    /// Vector search part
    pub vector: VectorQuery,
    /// How to combine results
    pub fusion: FusionStrategy,
    /// Final result limit
    pub limit: Option<usize>,
}

impl HybridQuery {
    /// Create a new hybrid query
    pub fn new(structured: QueryExpr, vector: VectorQuery) -> Self {
        Self {
            alias: None,
            structured: Box::new(structured),
            vector,
            fusion: FusionStrategy::Rerank { weight: 0.5 },
            limit: None,
        }
    }

    /// Set fusion strategy
    pub fn with_fusion(mut self, fusion: FusionStrategy) -> Self {
        self.fusion = fusion;
        self
    }

    /// Set result limit
    pub fn limit(mut self, limit: usize) -> Self {
        self.limit = Some(limit);
        self
    }

    /// Set outer alias
    pub fn alias(mut self, alias: &str) -> Self {
        self.alias = Some(alias.to_string());
        self
    }
}

/// Strategy for combining structured and vector search results
#[derive(Debug, Clone)]
pub enum FusionStrategy {
    /// Vector similarity re-ranks structured results
    /// weight: 0.0 = pure structured, 1.0 = pure vector
    Rerank { weight: f32 },
    /// Filter with structured query, then search vectors among filtered
    FilterThenSearch,
    /// Search vectors first, then filter with structured query
    SearchThenFilter,
    /// Reciprocal Rank Fusion
    /// k: RRF constant (typically 60)
    RRF { k: u32 },
    /// Intersection: only return results that match both
    Intersection,
    /// Union: return results from either (with combined scores)
    Union {
        structured_weight: f32,
        vector_weight: f32,
    },
}

impl Default for FusionStrategy {
    fn default() -> Self {
        Self::Rerank { weight: 0.5 }
    }
}

// ============================================================================
// DML/DDL Query Types
// ============================================================================

/// Entity type qualifier for INSERT statements
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum InsertEntityType {
    /// Default: plain row
    #[default]
    Row,
    /// INSERT INTO t NODE (...)
    Node,
    /// INSERT INTO t EDGE (...)
    Edge,
    /// INSERT INTO t VECTOR (...)
    Vector,
    /// INSERT INTO t DOCUMENT (...)
    Document,
    /// INSERT INTO t KV (...)
    Kv,
}

/// Item-kind qualifier for UPDATE statements.
///
/// The `DOCUMENTS` / `ROWS` / `KV` model markers were removed (ADR 0067,
/// #1711): the catalog knows every existing collection's model, so an unmarked
/// UPDATE parses to `Rows` and the runtime resolves `Documents` / `Kv`
/// semantics from the catalog. Only `NODES` / `EDGES` remain parse-time markers
/// — a graph collection holds both record kinds, so only the user can say which.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum UpdateTarget {
    /// Unmarked UPDATE: parses as row-shaped, catalog-resolved at runtime.
    #[default]
    Rows,
    /// Catalog-inferred document collection (no parse-time marker).
    Documents,
    /// Catalog-inferred KV collection (no parse-time marker).
    Kv,
    /// UPDATE t NODES SET ...
    Nodes,
    /// UPDATE t EDGES SET ...
    Edges,
}

/// An item in a RETURNING clause: either `*` (all columns) or a named column.
#[derive(Debug, Clone, PartialEq)]
pub enum ReturningItem {
    /// RETURNING *
    All,
    /// RETURNING col
    Column(String),
}

/// Conflict handling attached to an INSERT statement.
#[derive(Debug, Clone, PartialEq)]
pub struct OnConflictClause {
    /// Columns used to select the unique constraint. `None` means any
    /// uniqueness violation may be ignored by `DO NOTHING`.
    pub target: Option<Vec<String>>,
    /// Action taken when the selected uniqueness rule conflicts.
    pub action: OnConflictAction,
}

/// Action taken by an INSERT conflict clause.
#[derive(Debug, Clone, PartialEq)]
pub enum OnConflictAction {
    /// Skip the conflicting input row.
    DoNothing,
    /// Update the conflicting row using expressions that may reference
    /// columns from the proposed row through the `EXCLUDED` qualifier.
    DoUpdate {
        assignments: Vec<(String, super::Expr)>,
    },
}

/// INSERT INTO table (columns) VALUES (row1), (row2), ... [WITH TTL duration] [WITH METADATA (k=v)]
#[derive(Debug, Clone)]
pub struct InsertQuery {
    /// Target table name
    pub table: String,
    /// Entity type qualifier
    pub entity_type: InsertEntityType,
    /// Column names
    pub columns: Vec<String>,
    /// Canonical SQL rows of expressions.
    pub value_exprs: Vec<Vec<super::Expr>>,
    /// Rows of values (each inner Vec is one row)
    pub values: Vec<Vec<Value>>,
    /// Optional PostgreSQL-compatible conflict action.
    pub on_conflict: Option<OnConflictClause>,
    /// Optional RETURNING clause items.
    pub returning: Option<Vec<ReturningItem>>,
    /// Optional TTL in milliseconds (from WITH TTL clause)
    pub ttl_ms: Option<u64>,
    /// Optional absolute expiration (from WITH EXPIRES AT clause)
    pub expires_at_ms: Option<u64>,
    /// Optional metadata key-value pairs (from WITH METADATA clause)
    pub with_metadata: Vec<(String, Value)>,
    /// Auto-embed fields on insert (from WITH AUTO EMBED clause)
    pub auto_embed: Option<AutoEmbedConfig>,
    /// Skip event subscription emission for this statement (SUPPRESS EVENTS).
    pub suppress_events: bool,
}

/// Configuration for automatic embedding generation on INSERT.
#[derive(Debug, Clone)]
pub struct AutoEmbedConfig {
    /// Fields to extract text from for embedding
    pub fields: Vec<String>,
    /// AI provider from an explicit `USING <provider>`. Empty when none was
    /// given — the runtime then resolves it via the embeddings task pointer
    /// `red.config.ai.embeddings.provider` (ADR-0068 §5).
    pub provider: String,
    /// Optional model override
    pub model: Option<String>,
}

/// EVENTS BACKFILL collection [WHERE pred] TO queue [LIMIT n]
#[derive(Debug, Clone)]
pub struct EventsBackfillQuery {
    pub collection: String,
    pub where_filter: Option<String>,
    pub target_queue: String,
    pub limit: Option<u64>,
}

/// UPDATE table SET col=val, ... WHERE filter [WITH TTL duration] [WITH METADATA (...)]
#[derive(Debug, Clone)]
pub struct UpdateQuery {
    /// Target table name
    pub table: String,
    /// Explicit item-kind target. Omitted targets default to rows.
    pub target: UpdateTarget,
    /// Canonical SQL assignments.
    pub assignment_exprs: Vec<(String, super::Expr)>,
    /// Per-assignment compound operator for `SET col += expr` forms.
    /// `None` means ordinary `SET col = expr`.
    pub compound_assignment_ops: Vec<Option<super::BinOp>>,
    /// Best-effort literal-only cache of assignments. Non-foldable expressions
    /// are preserved exclusively in `assignment_exprs` and evaluated later
    /// against the row pre-image by the runtime.
    pub assignments: Vec<(String, Value)>,
    /// Canonical SQL WHERE clause.
    pub where_expr: Option<super::Expr>,
    /// Optional WHERE filter
    pub filter: Option<Filter>,
    /// Optional TTL in milliseconds (from WITH TTL clause)
    pub ttl_ms: Option<u64>,
    /// Optional absolute expiration (from WITH EXPIRES AT clause)
    pub expires_at_ms: Option<u64>,
    /// Optional metadata key-value pairs (from WITH METADATA clause)
    pub with_metadata: Vec<(String, Value)>,
    /// Optional RETURNING clause items.
    pub returning: Option<Vec<ReturningItem>>,
    /// Optional Concurrent claim cardinality.
    pub claim_limit: Option<u64>,
    /// Require the claim cardinality to be fully satisfied before mutating.
    pub claim_exact: bool,
    /// Optional deterministic target ordering for limited UPDATE batches.
    pub order_by: Vec<OrderByClause>,
    /// Optional `LIMIT N` cap. Caps the number of targets the executor
    /// will mutate in a single statement. Required by `BATCH N ROWS`
    /// data migrations (#37) which run the same UPDATE body in a
    /// loop, advancing a checkpoint between batches.
    pub limit: Option<u64>,
    /// Skip event subscription emission for this statement (SUPPRESS EVENTS).
    pub suppress_events: bool,
}

/// DELETE FROM table WHERE filter
#[derive(Debug, Clone)]
pub struct DeleteQuery {
    /// Target table name
    pub table: String,
    /// Canonical SQL WHERE clause.
    pub where_expr: Option<super::Expr>,
    /// Optional WHERE filter
    pub filter: Option<Filter>,
    /// Optional RETURNING clause items.
    pub returning: Option<Vec<ReturningItem>>,
    /// Skip event subscription emission for this statement (SUPPRESS EVENTS).
    pub suppress_events: bool,
}

/// A table-level UNIQUE declaration; unnamed constraints receive a stable catalog name.
#[derive(Debug, Clone)]
pub struct CreateUniqueConstraint {
    pub name: Option<String>,
    pub columns: Vec<String>,
}

/// CREATE TABLE name (columns) or CREATE {KV|CONFIG|VAULT} name
#[derive(Debug, Clone)]
pub struct CreateTableQuery {
    /// Declared collection model. Defaults to Table for CREATE TABLE.
    pub collection_model: CollectionModel,
    /// Table name
    pub name: String,
    /// Column definitions
    pub columns: Vec<CreateColumnDef>,
    /// Table-level UNIQUE constraints, including composite keys.
    pub unique_constraints: Vec<CreateUniqueConstraint>,
    /// IF NOT EXISTS flag
    pub if_not_exists: bool,
    /// Optional default TTL applied to newly inserted items in this collection.
    pub default_ttl_ms: Option<u64>,
    /// Metrics rollup tiers declared by `CREATE METRICS ... DOWNSAMPLE`.
    /// Uses the existing time-series policy spelling: target:source:aggregation.
    pub metrics_rollup_policies: Vec<String>,
    /// Fields to prioritize in the context index (WITH CONTEXT INDEX ON (f1, f2))
    pub context_index_fields: Vec<String>,
    /// Enables the global context index for this table
    /// (`WITH context_index = true`). Default false — pure OLTP tables
    /// skip the tokenisation / 3-way RwLock write storm on every insert.
    /// Having `context_index_fields` non-empty also enables it implicitly.
    pub context_index_enabled: bool,
    /// When true, CREATE TABLE implicitly adds two user-visible columns
    /// `created_at` and `updated_at` (BIGINT unix-ms). The runtime
    /// populates them from `UnifiedEntity::created_at/updated_at` on
    /// every write; `created_at` is immutable after insert.
    /// Enabled via `WITH timestamps = true` in the DDL.
    pub timestamps: bool,
    /// Partitioning spec (Phase 2.2 PG parity).
    ///
    /// When present the table is the *parent* of a partition tree — every
    /// child partition is registered via `ALTER TABLE ... ATTACH PARTITION`.
    /// Phase 2.2 stops at registry-only: queries against a partitioned
    /// parent don't auto-rewrite as UNION yet (Phase 4 adds pruning).
    pub partition_by: Option<PartitionSpec>,
    /// Table-scoped multi-tenancy declaration (Phase 2.5.4).
    ///
    /// Syntax: `CREATE TABLE t (...) WITH (tenant_by = 'col_name')` or
    /// the shorthand `CREATE TABLE t (...) TENANT BY (col_name)`. The
    /// runtime treats the named column as the tenant discriminator and
    /// automatically:
    ///
    /// 1. Registers the table → column mapping so INSERTs that omit the
    ///    column get `CURRENT_TENANT()` auto-filled.
    /// 2. Installs an implicit RLS policy equivalent to
    ///    `USING (col = CURRENT_TENANT())` for SELECT/UPDATE/DELETE/INSERT.
    /// 3. Flips `rls_enabled_tables` on so the policy actually applies.
    ///
    /// None leaves the table non-tenant-scoped — callers manage tenancy
    /// manually via explicit CREATE POLICY if they want it.
    pub tenant_by: Option<String>,
    /// When true, UPDATE and DELETE on this table are rejected at
    /// parse time. Corresponds to `CREATE TABLE ... APPEND ONLY` or
    /// `WITH (append_only = true)`. Default false (mutable).
    pub append_only: bool,
    /// Declarative event subscriptions for this table. #291 stores
    /// metadata only; event emission is intentionally out of scope.
    pub subscriptions: Vec<reddb_types::catalog::SubscriptionDescriptor>,
    /// Analytics views declared by `CREATE GRAPH ... WITH ANALYTICS (...)`
    /// (issue #800). Empty for every collection model except graphs that
    /// opt in. Threaded into the persisted `CollectionContract` at execution
    /// time so each `<graph>.<output>` view is durable.
    pub analytics_config: Vec<reddb_types::catalog::AnalyticsViewDescriptor>,
    /// `CREATE VAULT ... WITH OWN MASTER KEY`: provision per-vault
    /// key material instead of using the cluster vault key.
    pub vault_own_master_key: bool,
    /// Per-collection AI policy declared by `WITH (EMBED (...) | MODERATE
    /// (...) | VISION (...))` (PRD #1267, issue #1271). `None` when no AI
    /// clause is present. The parser validates the grammar; the runtime
    /// validates each modality's provider/model against the capability
    /// matrix (#1269) at DDL execution time and persists the policy in
    /// the `CollectionContract`.
    pub ai_policy: Option<reddb_types::catalog::AiPolicy>,
}

/// CREATE METRIC path TYPE kind ROLE role
///   [SOURCE <ident>] [QUERY '<text>'] [WINDOW <duration>] [TIME_FIELD <ident>]
///
/// Issue #790 — when any of the derived-metric clauses are present the
/// descriptor is a *derived* metric: it names the inputs that a future
/// execution layer would consume. v0 stores the metadata only; reads of
/// the metric's *output* (not its descriptor) return a structured
/// "not yet implemented" error.
#[derive(Debug, Clone)]
pub struct CreateMetricQuery {
    pub path: String,
    pub kind: String,
    pub role: String,
    pub source: Option<String>,
    pub query: Option<String>,
    pub window_ms: Option<u64>,
    pub time_field: Option<String>,
}

/// ALTER METRIC path SET <field> <value>
///
/// v0 mutability:
/// - `set_role`: mutable — role is a semantic label (operational/kpi/sli).
/// - `attempted_kind`: parser captured a `SET KIND`/`SET TYPE` clause; the
///   runtime rejects with a clear error because kind changes alter the
///   metric's mathematical meaning (counter vs gauge vs histogram, etc.).
/// - `attempted_path`: parser captured a `SET PATH` clause; the runtime
///   rejects because path is the descriptor's identity.
#[derive(Debug, Clone)]
pub struct AlterMetricQuery {
    pub path: String,
    pub set_role: Option<String>,
    pub attempted_kind: Option<String>,
    pub attempted_path: Option<String>,
}

/// CREATE SLO path ON metric_path TARGET t WINDOW d UNIT
///
/// Issue #791 — declared over an existing SLI-role metric descriptor.
/// `target` is the objective (0 < target <= 1, e.g. 0.999); `window_ms`
/// is the rolling window the objective is evaluated over. Burn-rate /
/// error-budget evaluation is deferred to later slices — v0 stores
/// catalog state only.
#[derive(Debug, Clone)]
pub struct CreateSloQuery {
    pub path: String,
    pub metric_path: String,
    pub target: f64,
    pub window_ms: u64,
}

/// CREATE COLLECTION name KIND kind [SIGNED_BY ('pubkey_hex', ...)]
#[derive(Debug, Clone)]
pub struct CreateCollectionQuery {
    pub name: String,
    pub kind: String,
    pub if_not_exists: bool,
    pub vector_dimension: Option<usize>,
    pub vector_metric: Option<DistanceMetric>,
    /// Initial Ed25519 allowed-signer registry. Empty = unsigned collection.
    /// Each entry is a 32-byte Ed25519 public key. Mutable post-create via
    /// `ALTER COLLECTION ... ADD|REVOKE SIGNER` (see issue #520).
    pub allowed_signers: Vec<[u8; 32]>,
}

/// CREATE VECTOR name DIM n [METRIC metric]
#[derive(Debug, Clone)]
pub struct CreateVectorQuery {
    pub name: String,
    pub dimension: usize,
    pub metric: DistanceMetric,
    pub if_not_exists: bool,
}

/// `PARTITION BY RANGE|LIST|HASH (column)` clause.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionSpec {
    pub kind: PartitionKind,
    /// Partition key column(s). Simple single-column for Phase 2.2.
    pub column: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PartitionKind {
    /// `PARTITION BY RANGE(col)` — children bind `FOR VALUES FROM (a) TO (b)`.
    Range,
    /// `PARTITION BY LIST(col)` — children bind `FOR VALUES IN (v1, v2, ...)`.
    List,
    /// `PARTITION BY HASH(col)` — children bind `FOR VALUES WITH (MODULUS m, REMAINDER r)`.
    Hash,
}

/// Column definition for CREATE TABLE
#[derive(Debug, Clone)]
pub struct CreateColumnDef {
    /// Column name
    pub name: String,
    /// Legacy declared type string preserved for the runtime/storage pipeline.
    pub data_type: String,
    /// Structured SQL type used by the semantic layer.
    pub sql_type: SqlTypeName,
    /// NOT NULL constraint
    pub not_null: bool,
    /// DEFAULT value expression
    pub default: Option<String>,
    /// Persisted, deterministic expression for a stored generated field.
    pub generated: Option<crate::schema_expression::SchemaExpression>,
    /// SQL CHECK: FALSE rejects, TRUE or NULL accepts.
    pub check: Option<crate::schema_expression::SchemaExpression>,
    /// Compression level (COMPRESS:N)
    pub compress: Option<u8>,
    /// UNIQUE constraint
    pub unique: bool,
    /// PRIMARY KEY constraint
    pub primary_key: bool,
    /// Enum variant names (for ENUM type)
    pub enum_variants: Vec<String>,
    /// Array element type (for ARRAY type)
    pub array_element: Option<String>,
    /// Decimal precision (for DECIMAL type)
    pub decimal_precision: Option<u8>,
}

/// DROP TABLE name
#[derive(Debug, Clone)]
pub struct DropTableQuery {
    /// Table name
    pub name: String,
    /// IF EXISTS flag
    pub if_exists: bool,
}

/// DROP GRAPH [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct DropGraphQuery {
    pub name: String,
    pub if_exists: bool,
}

/// DROP VECTOR [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct DropVectorQuery {
    pub name: String,
    pub if_exists: bool,
}

/// DROP DOCUMENT [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct DropDocumentQuery {
    pub name: String,
    pub if_exists: bool,
}

/// DROP {KV|CONFIG|VAULT} [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct DropKvQuery {
    pub name: String,
    pub if_exists: bool,
    pub model: CollectionModel,
}

/// DROP COLLECTION [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct DropCollectionQuery {
    pub name: String,
    pub if_exists: bool,
    pub model: Option<CollectionModel>,
}

/// TRUNCATE {TABLE|GRAPH|VECTOR|DOCUMENT|TIMESERIES|KV|QUEUE|COLLECTION} [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct TruncateQuery {
    pub name: String,
    pub model: Option<CollectionModel>,
    pub if_exists: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum VcsRefKind {
    Branch,
    Tag,
}

/// CREATE BRANCH 'name' [FROM commitish] / CREATE TAG 'name' [AT commitish]
#[derive(Debug, Clone)]
pub struct CreateVcsRefQuery {
    pub kind: VcsRefKind,
    pub name: String,
    pub target: Option<String>,
}

/// DROP BRANCH 'name' / DROP TAG 'name'
#[derive(Debug, Clone)]
pub struct DropVcsRefQuery {
    pub kind: VcsRefKind,
    pub name: String,
}

/// FORK STORE AS name [AT LSN n]
#[derive(Debug, Clone)]
pub struct ForkStoreQuery {
    pub name: String,
    pub at_lsn: Option<u64>,
}

/// PROMOTE FORK name
#[derive(Debug, Clone)]
pub struct PromoteForkQuery {
    pub name: String,
}

/// DROP FORK name
#[derive(Debug, Clone)]
pub struct DropForkQuery {
    pub name: String,
    pub if_exists: bool,
}

/// ALTER TABLE name operations
#[derive(Debug, Clone)]
pub struct AlterTableQuery {
    /// Table name
    pub name: String,
    /// Alter operations
    pub operations: Vec<AlterOperation>,
}

/// Single ALTER TABLE operation
#[derive(Debug, Clone)]
pub enum AlterOperation {
    /// ADD COLUMN definition
    AddColumn(CreateColumnDef),
    /// DROP COLUMN name
    DropColumn(String),
    /// RENAME COLUMN from TO to
    RenameColumn { from: String, to: String },
    /// `ATTACH PARTITION child FOR VALUES ...` (Phase 2.2 PG parity).
    ///
    /// Binds an existing child table to the parent partitioned table.
    /// The `bound` string captures the raw bound expression so the
    /// runtime can round-trip it back into `red_config` without a
    /// dedicated per-kind AST.
    AttachPartition {
        child: String,
        /// Human-readable bound string, e.g. `FROM (2024-01-01) TO (2025-01-01)`
        /// or `IN (1, 2, 3)` or `WITH (MODULUS 4, REMAINDER 0)`.
        bound: String,
    },
    /// `DETACH PARTITION child`
    DetachPartition { child: String },
    /// `ENABLE ROW LEVEL SECURITY` (Phase 2.5 PG parity).
    ///
    /// Flips the table into RLS-enforced mode. Reads against the table
    /// will be filtered by every matching `CREATE POLICY` (for the
    /// current role) combined with `AND`.
    EnableRowLevelSecurity,
    /// `DISABLE ROW LEVEL SECURITY` — disables enforcement; policies
    /// remain defined but are ignored until re-enabled.
    DisableRowLevelSecurity,
    /// `ENABLE TENANCY ON (col)` (Phase 2.5.4 PG parity-ish).
    ///
    /// Retrofit a tenant-scoped declaration onto an existing table —
    /// registers the column, installs the auto `__tenant_iso` RLS
    /// policy, and flips RLS on. Equivalent to re-running
    /// `CREATE TABLE ... TENANT BY (col)` minus the schema creation.
    EnableTenancy { column: String },
    /// `DISABLE TENANCY` — tears down the auto-policy and clears the
    /// tenancy registration. User-defined policies on the table are
    /// untouched; RLS stays enabled if any survive.
    DisableTenancy,
    /// `SET APPEND_ONLY = true|false` — flips the catalog flag.
    /// Setting `true` rejects all future UPDATE/DELETE at parse-time
    /// guard; setting `false` re-enables them. Existing rows are
    /// untouched either way — this is a purely declarative switch.
    SetAppendOnly(bool),
    /// `SET VERSIONED = true|false` — opt the table into (or out of)
    /// Git-for-Data. Enables merge / diff / AS OF semantics against
    /// this collection. Works retroactively: previously-created
    /// rows become part of the history accessible via AS OF as long
    /// as their xmin is still pinned by an existing commit.
    SetVersioned(bool),
    /// `ENABLE EVENTS ...` — install or re-enable table event subscription metadata.
    EnableEvents(reddb_types::catalog::SubscriptionDescriptor),
    /// `DISABLE EVENTS` — mark all table event subscriptions disabled.
    DisableEvents,
    /// `ADD SUBSCRIPTION name TO queue [REDACT (...)] [WHERE ...]` — add a named subscription.
    AddSubscription {
        name: String,
        descriptor: reddb_types::catalog::SubscriptionDescriptor,
    },
    /// `DROP SUBSCRIPTION name` — remove a named subscription by name.
    DropSubscription { name: String },
    /// Issue #522 — `ALTER COLLECTION name ADD SIGNER 'hex_pubkey'`.
    /// Appends the key to the per-collection signer registry and
    /// records an `Add` entry on the `signer_history` audit log.
    AddSigner { pubkey: [u8; 32] },
    /// Issue #522 — `ALTER COLLECTION name REVOKE SIGNER 'hex_pubkey'`.
    /// Removes the key from the *currently allowed* set and records a
    /// `Revoke` entry. Past rows signed by `pubkey` remain readable
    /// and re-verifiable — only future inserts are rejected.
    RevokeSigner { pubkey: [u8; 32] },
    /// Issue #580 — `ALTER COLLECTION name SET RETENTION <duration>`.
    /// Stores a declarative retention policy on the collection contract.
    /// Enforcement is lazy-on-scan: reads silently filter out rows older
    /// than `now - duration_ms` by the collection's timestamp column.
    SetRetention { duration_ms: u64 },
    /// Issue #580 — `ALTER COLLECTION name UNSET RETENTION`.
    /// Removes the policy. Previously-hidden expired rows become
    /// readable again — the slice never physically dropped them.
    UnsetRetention,
    /// Issue #801 — `ALTER GRAPH name ADD ANALYTICS (<output> [opts] [, ...])`.
    /// Idempotently enables analytics outputs on an existing graph's
    /// `analytics_config` without recreating the collection. Adding an
    /// already-enabled output is a no-op (no error, no duplicate state);
    /// the next read of `<graph>.<output>` materializes on demand.
    AddAnalytics(Vec<reddb_types::catalog::AnalyticsViewDescriptor>),
    /// Issue #801 — `ALTER GRAPH name DROP ANALYTICS <output>`.
    /// Removes the output from `analytics_config`; the next read of
    /// `<graph>.<output>` no longer resolves. Dropping an output that is
    /// not currently enabled is a clean error (handled in the executor).
    DropAnalytics(reddb_types::catalog::AnalyticsOutput),
}

// ============================================================================
// Shared Types
// ============================================================================

/// Column/field projection
#[derive(Debug, Clone, PartialEq)]
pub enum Projection {
    /// Select all columns (*)
    All,
    /// Single column by name
    Column(String),
    /// Column with alias
    Alias(String, String),
    /// Function call (name, args)
    Function(String, Vec<Projection>),
    /// Expression with optional alias
    Expression(Box<Filter>, Option<String>),
    /// Field reference (for graph properties)
    Field(FieldRef, Option<String>),
    /// Window function call: `fn(args) OVER (PARTITION BY ... ORDER BY ...
    /// [frame])`. Carries the window specification as a sibling so the
    /// planner can lower it without re-parsing. No runtime in slice 7a —
    /// the analytics executor lands in a subsequent slice (issue #589
    /// follow-ups). See `super::WindowSpec`.
    Window {
        name: String,
        args: Vec<Projection>,
        window: Box<super::WindowSpec>,
        alias: Option<String>,
    },
}

impl Projection {
    /// Create a projection from a field reference
    pub fn from_field(field: FieldRef) -> Self {
        Projection::Field(field, None)
    }

    /// Create a column projection
    pub fn column(name: &str) -> Self {
        Projection::Column(name.to_string())
    }

    /// Create an aliased projection
    pub fn with_alias(column: &str, alias: &str) -> Self {
        Projection::Alias(column.to_string(), alias.to_string())
    }
}

/// Filter condition
#[derive(Debug, Clone, PartialEq)]
pub enum Filter {
    /// Comparison: field op value
    Compare {
        field: FieldRef,
        op: CompareOp,
        value: Value,
    },
    /// Field-to-field comparison: left.field op right.field. Used when
    /// WHERE / BETWEEN operands reference another column instead of a
    /// literal — the pre-Fase-2-parser-v2 shim for column-to-column
    /// predicates. Once the Expr-rewrite lands, this collapses into
    /// `Compare { left: Expr, op, right: Expr }`.
    CompareFields {
        left: FieldRef,
        op: CompareOp,
        right: FieldRef,
    },
    /// Expression-to-expression comparison: `lhs op rhs` where either
    /// side may be an arbitrary `Expr` tree (function call, CAST,
    /// arithmetic, nested CASE). This is the most general compare
    /// variant — `Compare` and `CompareFields` stay as fast-path
    /// specialisations because the planner / cost model / index
    /// selector all pattern-match on the simpler shapes. The parser
    /// only emits this variant when a simpler one cannot express the
    /// predicate.
    CompareExpr {
        lhs: super::Expr,
        op: CompareOp,
        rhs: super::Expr,
    },
    /// Logical AND
    And(Box<Filter>, Box<Filter>),
    /// Logical OR
    Or(Box<Filter>, Box<Filter>),
    /// Logical NOT
    Not(Box<Filter>),
    /// IS NULL
    IsNull(FieldRef),
    /// IS NOT NULL
    IsNotNull(FieldRef),
    /// IN (value1, value2, ...)
    In { field: FieldRef, values: Vec<Value> },
    /// BETWEEN low AND high
    Between {
        field: FieldRef,
        low: Value,
        high: Value,
    },
    /// LIKE pattern
    Like { field: FieldRef, pattern: String },
    /// STARTS WITH prefix
    StartsWith { field: FieldRef, prefix: String },
    /// ENDS WITH suffix
    EndsWith { field: FieldRef, suffix: String },
    /// CONTAINS substring
    Contains { field: FieldRef, substring: String },
}

impl Filter {
    /// Create a comparison filter
    pub fn compare(field: FieldRef, op: CompareOp, value: Value) -> Self {
        Self::Compare { field, op, value }
    }

    /// Combine with AND
    pub fn and(self, other: Filter) -> Self {
        Self::And(Box::new(self), Box::new(other))
    }

    /// Combine with OR
    pub fn or(self, other: Filter) -> Self {
        Self::Or(Box::new(self), Box::new(other))
    }

    /// Negate
    // Filter combinator wrapping `self` in `Filter::Not`; unrelated to
    // `std::ops::Not`, so that trait is intentionally not implemented.
    #[allow(clippy::should_implement_trait)]
    pub fn not(self) -> Self {
        Self::Not(Box::new(self))
    }

    /// Bottom-up AST rewrites: OR-of-equalities → IN, AND/OR flatten.
    /// Inspired by MongoDB's `MatchExpression::optimize()`.
    /// Call on the result of `effective_table_filter()` before evaluation.
    pub fn optimize(self) -> Self {
        crate::filter_optimizer::optimize(self)
    }
}

/// Comparison operator
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompareOp {
    /// Equal (=)
    Eq,
    /// Not equal (<> or !=)
    Ne,
    /// Less than (<)
    Lt,
    /// Less than or equal (<=)
    Le,
    /// Greater than (>)
    Gt,
    /// Greater than or equal (>=)
    Ge,
}

impl fmt::Display for CompareOp {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            CompareOp::Eq => write!(f, "="),
            CompareOp::Ne => write!(f, "<>"),
            CompareOp::Lt => write!(f, "<"),
            CompareOp::Le => write!(f, "<="),
            CompareOp::Gt => write!(f, ">"),
            CompareOp::Ge => write!(f, ">="),
        }
    }
}

/// Order by clause.
///
/// Fase 2 migration: `field` is the legacy bare column reference and
/// remains populated for back-compat with existing callers (SPARQL /
/// Gremlin / Cypher translators, the planner cost model, etc.). The
/// new `expr` slot carries an arbitrary `Expr` tree — when present,
/// runtime comparators prefer it over `field`, so the parser can
/// emit `ORDER BY CAST(a AS INT)`, `ORDER BY a + b * 2`, etc. without
/// breaking the rest of the codebase.
///
/// When `expr` is `None`, the clause behaves exactly like before.
/// When `expr` is `Some(Expr::Column(f))`, runtime code may still use
/// the legacy path — it's equivalent. Constructors default `expr` to
/// `None` so all existing call sites stay source-compatible.
#[derive(Debug, Clone)]
pub struct OrderByClause {
    /// Field to order by. Left populated even when `expr` is set so
    /// legacy consumers (planner cardinality estimate, cost model,
    /// mode translators) that still pattern-match on `field` keep
    /// working during the Fase 2 migration.
    pub field: FieldRef,
    /// Fase 2 expression-aware sort key. When `Some`, runtime order
    /// comparators evaluate this expression per row and sort on the
    /// resulting values — unlocks `ORDER BY expr` (Fase 1.6).
    pub expr: Option<super::Expr>,
    /// Ascending or descending
    pub ascending: bool,
    /// Nulls first or last
    pub nulls_first: bool,
}

impl OrderByClause {
    /// Create ascending order
    pub fn asc(field: FieldRef) -> Self {
        Self {
            field,
            expr: None,
            ascending: true,
            nulls_first: false,
        }
    }

    /// Create descending order
    pub fn desc(field: FieldRef) -> Self {
        Self {
            field,
            expr: None,
            ascending: false,
            nulls_first: true,
        }
    }

    /// Attach an `Expr` sort key to an existing clause. Leaves `field`
    /// untouched so back-compat match sites keep their pattern.
    pub fn with_expr(mut self, expr: super::Expr) -> Self {
        self.expr = Some(expr);
        self
    }
}

// ============================================================================
// Window OVER clause (issue #589 slice 7a)
// ============================================================================
//
// Syntactic representation of `OVER (PARTITION BY ... ORDER BY ... [frame])`.
// Slice 7a: AST + parser only. No runtime / executor wiring.

/// Frame unit: `ROWS` (physical row offset) or `RANGE` (logical value
/// offset). Slice 7a stores the choice but does not yet differentiate
/// at runtime — semantics arrive with the analytics executor.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WindowFrameUnit {
    Rows,
    Range,
}

/// One endpoint of a frame: UNBOUNDED PRECEDING / CURRENT ROW /
/// PRECEDING(expr) / FOLLOWING(expr) / UNBOUNDED FOLLOWING.
#[derive(Debug, Clone, PartialEq)]
pub enum WindowFrameBound {
    UnboundedPreceding,
    UnboundedFollowing,
    CurrentRow,
    Preceding(Box<super::Expr>),
    Following(Box<super::Expr>),
}

/// `ROWS|RANGE BETWEEN start AND end` — or the single-bound shorthand
/// `ROWS start` (end implied as CURRENT ROW per SQL standard). Slice 7a
/// represents both shapes uniformly with `end: Option<...>` so downstream
/// code can normalise.
#[derive(Debug, Clone, PartialEq)]
pub struct WindowFrame {
    pub unit: WindowFrameUnit,
    pub start: WindowFrameBound,
    pub end: Option<WindowFrameBound>,
}

/// One ORDER BY item inside a window spec. Window order keys are
/// expression-based by SQL standard, so we carry an `Expr` directly
/// rather than reusing the top-level `OrderByClause` (which still has
/// a legacy `FieldRef` slot for the Fase 2 migration).
#[derive(Debug, Clone, PartialEq)]
pub struct WindowOrderItem {
    pub expr: super::Expr,
    pub ascending: bool,
    pub nulls_first: bool,
}

/// Full window specification — the AST node behind `OVER (...)`.
/// `frame` is `None` when the user did not specify a frame clause; the
/// analytics executor will materialise the SQL default (RANGE UNBOUNDED
/// PRECEDING AND CURRENT ROW when ORDER BY is present, the full
/// partition otherwise) once it lands.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct WindowSpec {
    pub partition_by: Vec<super::Expr>,
    pub order_by: Vec<WindowOrderItem>,
    pub frame: Option<WindowFrame>,
}

// ============================================================================
// Graph Commands
// ============================================================================

/// Graph analytics command issued via SQL-like syntax
#[derive(Debug, Clone)]
pub struct GraphCommandOrderBy {
    pub metric: String,
    pub ascending: bool,
}

#[derive(Debug, Clone)]
pub enum GraphCommand {
    /// GRAPH NEIGHBORHOOD 'source' [DEPTH n] [DIRECTION dir] [EDGES IN ('label', ...)]
    Neighborhood {
        source: String,
        depth: u32,
        direction: String,
        edge_labels: Option<Vec<String>>,
    },
    /// GRAPH SHORTEST_PATH 'source' TO 'target' [ALGORITHM alg] [DIRECTION dir] [ORDER BY metric [ASC|DESC]] [LIMIT n]
    ShortestPath {
        source: String,
        target: String,
        algorithm: String,
        direction: String,
        limit: Option<u32>,
        order_by: Option<GraphCommandOrderBy>,
    },
    /// GRAPH TRAVERSE 'source' [STRATEGY bfs|dfs] [DEPTH n] [DIRECTION dir] [EDGES IN ('label', ...)]
    Traverse {
        source: String,
        strategy: String,
        depth: u32,
        direction: String,
        edge_labels: Option<Vec<String>>,
    },
    /// GRAPH CENTRALITY [ALGORITHM alg] [ORDER BY metric [ASC|DESC]] [LIMIT n]
    ///
    /// `limit = None` keeps the historical implicit top-100 cap. `Some(n)`
    /// caps the returned rows at `n`.
    Centrality {
        algorithm: String,
        limit: Option<u32>,
        order_by: Option<GraphCommandOrderBy>,
    },
    /// GRAPH COMMUNITY [ALGORITHM alg] [MAX_ITERATIONS n] [ORDER BY metric [ASC|DESC]] [LIMIT n] [RETURN ASSIGNMENTS]
    ///
    /// `return_assignments = false` (default) keeps the historical per-community
    /// aggregate shape (`community_id`, `size`). `true` emits one row per node
    /// (`node_id`, `community_id`) — the node→community map (#660).
    Community {
        algorithm: String,
        max_iterations: u32,
        limit: Option<u32>,
        order_by: Option<GraphCommandOrderBy>,
        return_assignments: bool,
    },
    /// GRAPH COMPONENTS [MODE connected|weak|strong] [ORDER BY metric [ASC|DESC]] [LIMIT n]
    Components {
        mode: String,
        limit: Option<u32>,
        order_by: Option<GraphCommandOrderBy>,
    },
    /// GRAPH CYCLES [MAX_LENGTH n]
    Cycles { max_length: u32 },
    /// GRAPH CLUSTERING
    Clustering,
    /// GRAPH TOPOLOGICAL_SORT
    TopologicalSort,
    /// GRAPH PROPERTIES ['<id-or-label>']
    ///
    /// `source = None` returns graph-wide stats. `source = Some("...")` returns
    /// the full property bag of a specific node, resolved via the same label
    /// index as `GRAPH NEIGHBORHOOD` / `GRAPH TRAVERSE` (issue #416).
    Properties { source: Option<String> },
}

// ============================================================================
// Search Commands
// ============================================================================

/// Search command issued via SQL-like syntax
#[derive(Debug, Clone)]
pub enum SearchCommand {
    /// SEARCH SIMILAR [v1, v2, ...] | $N | TEXT 'query' [COLLECTION col] [LIMIT n] [MIN_SCORE f] [USING provider]
    Similar {
        vector: Vec<f32>,
        text: Option<String>,
        provider: Option<String>,
        collection: String,
        limit: usize,
        min_score: f32,
        /// `$N` placeholder for the vector slot. `Some(idx)` when the SQL
        /// used `SEARCH SIMILAR $N ...`; the binder substitutes the
        /// user-supplied `Value::Vector` and clears this back to `None`.
        /// Runtime executors assert this is `None` post-bind.
        vector_param: Option<usize>,
        /// `$N` placeholder for the `LIMIT` slot (issue #361). The binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
        /// `$N` placeholder for the `MIN_SCORE` slot (issue #361). The
        /// binder substitutes the user-supplied float into `min_score`
        /// and clears this back to `None`.
        min_score_param: Option<usize>,
        /// `$N` placeholder for `SEARCH SIMILAR TEXT $N` (issue #361).
        /// Binder substitutes the user-supplied text into `text` and
        /// clears this back to `None`.
        text_param: Option<usize>,
    },
    /// SEARCH TEXT 'query' [COLLECTION col] [LIMIT n] [FUZZY]
    Text {
        query: String,
        collection: Option<String>,
        limit: usize,
        fuzzy: bool,
        /// `$N` placeholder for the `LIMIT` slot (issue #361). Same
        /// shape as `SearchCommand::Hybrid::limit_param`; the binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH HYBRID [vector] [TEXT 'query'] COLLECTION col [LIMIT n]
    Hybrid {
        vector: Option<Vec<f32>>,
        query: Option<String>,
        collection: String,
        limit: usize,
        /// `$N` placeholder for the `LIMIT` / `K` slot (issue #361).
        /// Same shape as `SearchCommand::Similar::limit_param`; the
        /// binder substitutes the user-supplied positive integer and
        /// clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH MULTIMODAL 'key_or_query' [COLLECTION col] [LIMIT n]
    Multimodal {
        query: String,
        collection: Option<String>,
        limit: usize,
        /// `$N` placeholder for the `LIMIT` slot (issue #361). Same
        /// shape as `SearchCommand::Hybrid::limit_param`; the binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH INDEX index VALUE 'value' [COLLECTION col] [LIMIT n] [EXACT]
    Index {
        index: String,
        value: String,
        collection: Option<String>,
        limit: usize,
        exact: bool,
        /// `$N` placeholder for the `LIMIT` slot (issue #361). Same
        /// shape as `SearchCommand::Hybrid::limit_param`; the binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH CONTEXT 'query' [FIELD field] [COLLECTION col] [LIMIT n] [DEPTH n]
    Context {
        query: String,
        field: Option<String>,
        collection: Option<String>,
        limit: usize,
        depth: usize,
        /// `$N` placeholder for the `LIMIT` slot (issue #361). Same
        /// shape as `SearchCommand::Hybrid::limit_param`; the binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH SPATIAL RADIUS lat lon radius_km COLLECTION col COLUMN col [LIMIT n]
    SpatialRadius {
        center_lat: f64,
        center_lon: f64,
        radius_km: f64,
        collection: String,
        column: String,
        limit: usize,
        /// `$N` placeholder for the `LIMIT` slot (issue #361). Same
        /// shape as `SearchCommand::Hybrid::limit_param`; the binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH SPATIAL BBOX min_lat min_lon max_lat max_lon COLLECTION col COLUMN col [LIMIT n]
    SpatialBbox {
        min_lat: f64,
        min_lon: f64,
        max_lat: f64,
        max_lon: f64,
        collection: String,
        column: String,
        limit: usize,
        /// `$N` placeholder for the `LIMIT` slot (issue #361). Same
        /// shape as `SearchCommand::Hybrid::limit_param`; the binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH SPATIAL WITHIN POLYGON ((lat lon), ...) COLLECTION col COLUMN col [LIMIT n]
    SpatialWithinPolygon {
        vertices: Vec<(f64, f64)>,
        collection: String,
        column: String,
        limit: usize,
        /// `$N` placeholder for the `LIMIT` slot. The binder
        /// substitutes the user-supplied positive integer into `limit`
        /// and clears this back to `None`.
        limit_param: Option<usize>,
    },
    /// SEARCH SPATIAL NEAREST lat lon K n COLLECTION col COLUMN col
    SpatialNearest {
        lat: f64,
        lon: f64,
        k: usize,
        collection: String,
        column: String,
        /// `$N` placeholder for the `K` slot (issue #361). Same shape
        /// as `SearchCommand::Hybrid::limit_param`; the binder
        /// substitutes the user-supplied positive integer into `k`
        /// and clears this back to `None`.
        k_param: Option<usize>,
    },
}

// ============================================================================
// Time-Series DDL
// ============================================================================

/// CREATE TIMESERIES name [RETENTION duration] [CHUNK_SIZE n] [DOWNSAMPLE spec[, spec...]]
///
/// `CREATE HYPERTABLE` lands on the same AST with `hypertable` populated.
/// The TimescaleDB-style syntax (time column + chunk_interval) gives the
/// runtime enough to register a `HypertableSpec` alongside the
/// underlying collection contract, so chunk routing and TTL sweeps can
/// address the table without a separate DDL.
#[derive(Debug, Clone)]
pub struct CreateTimeSeriesQuery {
    pub name: String,
    pub retention_ms: Option<u64>,
    pub chunk_size: Option<usize>,
    pub downsample_policies: Vec<String>,
    pub if_not_exists: bool,
    /// When `Some`, the DDL was spelled `CREATE HYPERTABLE` and the
    /// runtime must register the spec with the hypertable registry.
    pub hypertable: Option<HypertableDdl>,
    /// `WITH SESSION_KEY <col>` — default partition column for the
    /// `SESSIONIZE` operator. Persisted on the collection contract so
    /// queries that omit `BY <col>` pick it up. Issue #576 slice 1.
    pub session_key: Option<String>,
    /// `SESSION_GAP <duration>` — default inactivity gap (ms) for the
    /// `SESSIONIZE` operator. Issue #576 slice 1.
    pub session_gap_ms: Option<u64>,
    /// Columnar analytical storage policy (PRD #850, #911, #1772).
    /// Defaults to true for in-scope collections; `NO COLUMNAR` sets it
    /// false as the per-collection opt-out. When true the collection
    /// contract is built with
    /// `analytical_storage.columnar = true`, so sealing a chunk routes
    /// through `seal_chunk_with_config`'s columnar arm and emits an
    /// RDCC `ColumnBlock` instead of the row seal.
    pub columnar: bool,
}

/// Hypertable-specific DDL fields — set only when the caller used
/// `CREATE HYPERTABLE`.
#[derive(Debug, Clone)]
pub struct HypertableDdl {
    /// Column that carries the nanosecond timestamp axis.
    pub time_column: String,
    /// Chunk width in nanoseconds.
    pub chunk_interval_ns: u64,
    /// Per-chunk default TTL in nanoseconds (`None` = no TTL).
    pub default_ttl_ns: Option<u64>,
}

/// DROP TIMESERIES [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct DropTimeSeriesQuery {
    pub name: String,
    pub if_exists: bool,
}

// ============================================================================
// Queue DDL & Commands
// ============================================================================

/// Default `MAX_ATTEMPTS` for `CREATE QUEUE` when omitted.
pub const DEFAULT_QUEUE_MAX_ATTEMPTS: u32 = 3;
/// Default `LOCK_DEADLINE_MS` for `CREATE QUEUE` when omitted.
pub const DEFAULT_QUEUE_LOCK_DEADLINE_MS: u64 = 30_000;
/// Default `IN_FLIGHT_CAP_PER_GROUP` for `CREATE QUEUE` when omitted.
pub const DEFAULT_QUEUE_IN_FLIGHT_CAP_PER_GROUP: u32 = 10_000;
/// Default push idempotency window for `QUEUE PUSH ... DEDUP` when omitted.
pub const DEFAULT_QUEUE_DEDUP_WINDOW_MS: u64 = 300_000;

/// CREATE QUEUE name [MAX_SIZE n] [PRIORITY] [WITH TTL duration] [WITH DLQ name]
/// [MAX_ATTEMPTS n] [LOCK_DEADLINE_MS n] [IN_FLIGHT_CAP_PER_GROUP n]
/// [RETRY_DELAY duration] [DEDUP_WINDOW duration]
#[derive(Debug, Clone)]
pub struct CreateQueueQuery {
    pub name: String,
    pub mode: QueueMode,
    pub priority: bool,
    pub max_size: Option<usize>,
    pub ttl_ms: Option<u64>,
    pub dlq: Option<String>,
    pub max_attempts: u32,
    pub lock_deadline_ms: u64,
    pub in_flight_cap_per_group: u32,
    pub if_not_exists: bool,
    /// Default retry delay applied to NACK-requeued messages before they
    /// become re-deliverable. `None` means no delay — the released
    /// message is immediately available again (pre-#723 behaviour).
    /// `Some(ms)` reuses the per-message availability machinery from
    /// issue #722 to defer the next delivery attempt. An authorized
    /// `NACK ... WITH DELAY <duration>` overrides this per-failure.
    pub retry_delay_ms: Option<u64>,
    /// Queue-scoped push idempotency window. `None` means the engine
    /// default applies at runtime.
    pub dedup_window_ms: Option<u64>,
}

/// ALTER QUEUE name SET <clause>
///   MODE [FANOUT|WORK|STANDARD|FIFO]
///   MAX_ATTEMPTS n
///   LOCK_DEADLINE_MS n
///   IN_FLIGHT_CAP_PER_GROUP n
///   DLQ name
///   RETRY_DELAY duration
///   DEDUP_WINDOW duration
#[derive(Debug, Clone, Default)]
pub struct AlterQueueQuery {
    pub name: String,
    pub mode: Option<QueueMode>,
    pub max_attempts: Option<u32>,
    pub lock_deadline_ms: Option<u64>,
    pub in_flight_cap_per_group: Option<u32>,
    pub dlq: Option<String>,
    /// Update the queue's default retry delay (issue #723). `Some(0)`
    /// clears the delay back to immediate requeue.
    pub retry_delay_ms: Option<u64>,
    /// Update the queue-scoped push idempotency window.
    pub dedup_window_ms: Option<u64>,
}

/// DROP QUEUE [IF EXISTS] name
#[derive(Debug, Clone)]
pub struct DropQueueQuery {
    pub name: String,
    pub if_exists: bool,
}

/// SELECT <columns> FROM QUEUE name [WHERE filter] [LIMIT n]
#[derive(Debug, Clone)]
pub struct QueueSelectQuery {
    pub queue: String,
    pub columns: Vec<String>,
    pub filter: Option<Filter>,
    pub limit: Option<u64>,
}

/// Which end of the queue
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueueSide {
    Left,
    Right,
}

/// Per-message delayed availability for `QUEUE PUSH` (PRD #718 / #722).
///
/// `DelayMs` is relative — the runtime resolves it against the push-time
/// wall clock. `AtUnixMs` is absolute — the runtime promotes it to
/// nanoseconds unchanged. Both ultimately surface to consumers as an
/// `available_at_ns` metadata field that delivery paths filter on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueueAvailability {
    /// Delay the first delivery by this many milliseconds from push time.
    DelayMs(u64),
    /// Make the message first-deliverable at this absolute unix-ms instant.
    AtUnixMs(u64),
}

/// Queue operation commands
// The largest variant carries an inline `Filter`; boxing it would ripple
// to every construction and match site for a marginal stack-size win, so
// the size difference is accepted.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone)]
pub enum QueueCommand {
    Push {
        queue: String,
        value: Value,
        side: QueueSide,
        priority: Option<i32>,
        /// Optional per-message ordering key. Grouped delivery serializes
        /// messages sharing this key per consumer group.
        key: Option<String>,
        /// Per-message delayed availability (issue #722). `None` means the
        /// message is deliverable immediately. `Some(_)` resolves to an
        /// `available_at_ns` metadata field at push time; delivery paths
        /// (`QUEUE READ`, `QUEUE POP`, `QUEUE READ … WAIT`) refuse to
        /// deliver the message until that instant.
        available: Option<QueueAvailability>,
        /// Optional producer push idempotency key. Duplicate pushes with
        /// the same key on the same queue return the first message id
        /// while the queue's dedup window is active.
        dedup_key: Option<String>,
    },
    Pop {
        queue: String,
        side: QueueSide,
        count: usize,
    },
    Peek {
        queue: String,
        count: usize,
    },
    Len {
        queue: String,
    },
    Purge {
        queue: String,
    },
    GroupCreate {
        queue: String,
        group: String,
    },
    GroupRead {
        queue: String,
        group: Option<String>,
        consumer: String,
        count: usize,
        /// Optional blocking-read deadline in milliseconds (PRD #718 slice
        /// A: `QUEUE READ … WAIT <duration>`). `None` means classic
        /// non-blocking semantics. The runtime currently honors the field
        /// synchronously — the actual wait registry lands in slice C.
        wait_ms: Option<u64>,
    },
    Pending {
        queue: String,
        group: String,
    },
    Claim {
        queue: String,
        group: String,
        consumer: String,
        min_idle_ms: u64,
    },
    Ack {
        queue: String,
        // Legacy tuple handle. Empty `group` / `message_id` strings mean
        // the request relies solely on `delivery_id`. ADR 0026: when both
        // `delivery_id` and the tuple are supplied, `delivery_id` wins.
        group: String,
        message_id: String,
        /// Server-issued opaque base32 delivery handle (ADR 0026). When
        /// present, takes precedence over the legacy tuple; the tuple is
        /// kept for one minor release as a wire-compat bridge.
        delivery_id: Option<String>,
    },
    Nack {
        queue: String,
        group: String,
        message_id: String,
        delivery_id: Option<String>,
        /// Per-failure retry delay override (issue #723). `Some(ms)`
        /// requests that the failed message become re-deliverable only
        /// after `ms` milliseconds; takes precedence over the queue's
        /// default `retry_delay_ms`. Authorization is enforced at the
        /// runtime layer: requests from a read-only identity are
        /// rejected.
        delay_ms: Option<u64>,
    },
    Move {
        source: String,
        destination: String,
        filter: Option<Filter>,
        limit: usize,
    },
}

// ============================================================================
// Tree DDL & Commands
// ============================================================================

#[derive(Debug, Clone)]
pub struct TreeNodeSpec {
    pub label: String,
    pub node_type: Option<String>,
    pub properties: Vec<(String, Value)>,
    pub metadata: Vec<(String, Value)>,
    pub max_children: Option<usize>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TreePosition {
    First,
    Last,
    Index(usize),
}

#[derive(Debug, Clone)]
pub struct CreateTreeQuery {
    pub collection: String,
    pub name: String,
    pub root: TreeNodeSpec,
    pub default_max_children: usize,
    pub if_not_exists: bool,
}

#[derive(Debug, Clone)]
pub struct DropTreeQuery {
    pub collection: String,
    pub name: String,
    pub if_exists: bool,
}

#[derive(Debug, Clone)]
pub enum TreeCommand {
    Insert {
        collection: String,
        tree_name: String,
        parent_id: u64,
        node: TreeNodeSpec,
        position: TreePosition,
    },
    Move {
        collection: String,
        tree_name: String,
        node_id: u64,
        parent_id: u64,
        position: TreePosition,
    },
    Delete {
        collection: String,
        tree_name: String,
        node_id: u64,
    },
    Validate {
        collection: String,
        tree_name: String,
    },
    Rebalance {
        collection: String,
        tree_name: String,
        dry_run: bool,
    },
}

// ============================================================================
// KV DSL Commands
// ============================================================================

/// KV verb commands: `KV PUT key = value [EXPIRE n] [IF NOT EXISTS]`, `KV GET key`, `KV DELETE key`
#[derive(Debug, Clone)]
pub enum KvCommand {
    Put {
        model: CollectionModel,
        collection: String,
        key: String,
        value: Value,
        /// TTL in milliseconds (from EXPIRE clause)
        ttl_ms: Option<u64>,
        tags: Vec<String>,
        if_not_exists: bool,
    },
    InvalidateTags {
        collection: String,
        tags: Vec<String>,
    },
    Get {
        model: CollectionModel,
        collection: String,
        key: String,
    },
    Reveal {
        collection: String,
        key: String,
        version: Option<i64>,
    },
    Rotate {
        collection: String,
        key: String,
        value: Value,
        tags: Vec<String>,
    },
    History {
        collection: String,
        key: String,
    },
    List {
        model: CollectionModel,
        collection: String,
        prefix: Option<String>,
        limit: Option<usize>,
        offset: usize,
        as_json: bool,
    },
    Purge {
        collection: String,
        key: String,
    },
    Watch {
        model: CollectionModel,
        collection: String,
        key: String,
        prefix: bool,
        from_lsn: Option<u64>,
    },
    Delete {
        model: CollectionModel,
        collection: String,
        key: String,
    },
    /// `KV INCR key [BY n] [EXPIRE dur]` — atomic increment; negative `by` = decrement.
    Incr {
        model: CollectionModel,
        collection: String,
        key: String,
        /// Step value; negative for DECR. Defaults to 1.
        by: i64,
        ttl_ms: Option<u64>,
    },
    /// `KV CAS key EXPECT <expected|NULL> SET <new> [EXPIRE dur]` — compare-and-set.
    ///
    /// `expected = None` means `EXPECT NULL` (key must be absent).
    Cas {
        model: CollectionModel,
        collection: String,
        key: String,
        /// The value the caller expects to be current; `None` = key must be absent.
        expected: Option<Value>,
        new_value: Value,
        ttl_ms: Option<u64>,
    },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConfigValueType {
    Bool,
    Int,
    String,
    Url,
    Object,
    Array,
}

impl ConfigValueType {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Bool => "bool",
            Self::Int => "int",
            Self::String => "string",
            Self::Url => "url",
            Self::Object => "object",
            Self::Array => "array",
        }
    }

    pub fn parse(input: &str) -> Option<Self> {
        match input.to_ascii_lowercase().as_str() {
            "bool" | "boolean" => Some(Self::Bool),
            "int" | "integer" => Some(Self::Int),
            "string" | "str" | "text" => Some(Self::String),
            "url" => Some(Self::Url),
            "object" | "json_object" => Some(Self::Object),
            "array" | "list" => Some(Self::Array),
            _ => None,
        }
    }
}

#[derive(Debug, Clone)]
pub enum ConfigCommand {
    Put {
        collection: String,
        key: String,
        value: Value,
        value_type: Option<ConfigValueType>,
        tags: Vec<String>,
    },
    Get {
        collection: String,
        key: String,
    },
    Resolve {
        collection: String,
        key: String,
    },
    Rotate {
        collection: String,
        key: String,
        value: Value,
        value_type: Option<ConfigValueType>,
        tags: Vec<String>,
    },
    Delete {
        collection: String,
        key: String,
    },
    History {
        collection: String,
        key: String,
    },
    List {
        collection: String,
        prefix: Option<String>,
        limit: Option<usize>,
        offset: usize,
    },
    Watch {
        collection: String,
        key: String,
        prefix: bool,
        from_lsn: Option<u64>,
    },
    InvalidVolatileOperation {
        operation: String,
        collection: String,
        key: Option<String>,
    },
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ast::{Expr, Span};

    fn table_expr(name: &str) -> QueryExpr {
        QueryExpr::Table(TableQuery::new(name))
    }

    #[test]
    fn policy_target_kind_identifiers_cover_all_variants() {
        assert_eq!(PolicyTargetKind::Table.as_ident(), "table");
        assert_eq!(PolicyTargetKind::Nodes.as_ident(), "nodes");
        assert_eq!(PolicyTargetKind::Edges.as_ident(), "edges");
        assert_eq!(PolicyTargetKind::Vectors.as_ident(), "vectors");
        assert_eq!(PolicyTargetKind::Messages.as_ident(), "messages");
        assert_eq!(PolicyTargetKind::Points.as_ident(), "points");
        assert_eq!(PolicyTargetKind::Documents.as_ident(), "documents");
    }

    #[test]
    fn index_method_display_names_are_sql_keywords() {
        assert_eq!(IndexMethod::BTree.to_string(), "BTREE");
        assert_eq!(IndexMethod::Hash.to_string(), "HASH");
        assert_eq!(IndexMethod::Bitmap.to_string(), "BITMAP");
        assert_eq!(IndexMethod::Spatial.to_string(), "SPATIAL");
        assert_eq!(IndexMethod::H3 { resolution: 9 }.to_string(), "H3");
    }

    #[test]
    fn table_query_defaults_and_subquery_source() {
        let query = TableQuery::new("hosts");
        assert_eq!(query.table, "hosts");
        assert!(query.source.is_none());
        assert!(query.alias.is_none());
        assert!(query.select_items.is_empty());
        assert!(!query.distinct);

        let subquery = table_expr("inner");
        let wrapped = TableQuery::from_subquery(subquery, Some("h".to_string()));
        assert_eq!(wrapped.table, "__subq_h");
        assert_eq!(wrapped.alias.as_deref(), Some("h"));
        assert!(matches!(wrapped.source, Some(TableSource::Subquery(_))));

        let anonymous = TableQuery::from_subquery(table_expr("inner"), None);
        assert_eq!(anonymous.table, "__subq_anon");
        assert!(anonymous.alias.is_none());
    }

    #[test]
    fn graph_pattern_query_and_components_builders_set_fields() {
        let node = NodePattern::new("h").of_label("Host").with_property(
            "os",
            CompareOp::Eq,
            Value::text("linux"),
        );
        assert_eq!(node.alias, "h");
        assert_eq!(node.node_label.as_deref(), Some("Host"));
        assert_eq!(node.properties.len(), 1);

        let edge = EdgePattern::new("h", "s")
            .alias("r")
            .of_label("HAS_SERVICE")
            .direction(EdgeDirection::Incoming)
            .hops(2, 4);
        assert_eq!(edge.alias.as_deref(), Some("r"));
        assert_eq!(edge.edge_label.as_deref(), Some("HAS_SERVICE"));
        assert_eq!(edge.direction, EdgeDirection::Incoming);
        assert_eq!((edge.min_hops, edge.max_hops), (2, 4));

        let pattern = GraphPattern::new().node(node).edge(edge);
        assert_eq!(pattern.nodes.len(), 1);
        assert_eq!(pattern.edges.len(), 1);

        let graph = GraphQuery::new(pattern).alias("g");
        assert_eq!(graph.alias.as_deref(), Some("g"));
        assert!(graph.filter.is_none());
        assert!(graph.return_.is_empty());
    }

    #[test]
    fn join_condition_and_join_query_defaults() {
        let left = FieldRef::column("hosts", "id");
        let right = FieldRef::node_id("h");
        let condition = JoinCondition::new(left.clone(), right.clone());
        assert_eq!(condition.left_field, left);
        assert_eq!(condition.right_field, right);

        let join = JoinQuery::new(
            table_expr("hosts"),
            table_expr("services"),
            condition.clone(),
        )
        .join_type(JoinType::LeftOuter);
        assert!(matches!(*join.left, QueryExpr::Table(_)));
        assert!(matches!(*join.right, QueryExpr::Table(_)));
        assert_eq!(join.join_type, JoinType::LeftOuter);
        assert_eq!(join.on.left_field, condition.left_field);
        assert!(join.return_items.is_empty());
    }

    #[test]
    fn field_ref_projection_filter_and_order_builders() {
        let column = FieldRef::column("hosts", "ip");
        let node_prop = FieldRef::node_prop("h", "os");
        let edge_prop = FieldRef::edge_prop("r", "weight");
        assert!(matches!(column, FieldRef::TableColumn { .. }));
        assert!(matches!(node_prop, FieldRef::NodeProperty { .. }));
        assert!(matches!(edge_prop, FieldRef::EdgeProperty { .. }));

        assert!(matches!(
            Projection::from_field(column.clone()),
            Projection::Field(_, None)
        ));
        assert!(matches!(
            Projection::column("ip"),
            Projection::Column(ref name) if name == "ip"
        ));
        assert!(matches!(
            Projection::with_alias("ip", "addr"),
            Projection::Alias(ref column, ref alias) if column == "ip" && alias == "addr"
        ));

        let eq = Filter::compare(column.clone(), CompareOp::Eq, Value::text("127.0.0.1"));
        let gt = Filter::compare(node_prop, CompareOp::Gt, Value::Integer(7));
        let combined = eq.clone().and(gt.clone()).or(eq.clone().not());
        assert!(matches!(combined, Filter::Or(_, _)));
        assert!(matches!(gt.optimize(), Filter::Compare { .. }));

        assert_eq!(CompareOp::Eq.to_string(), "=");
        assert_eq!(CompareOp::Ne.to_string(), "<>");
        assert_eq!(CompareOp::Lt.to_string(), "<");
        assert_eq!(CompareOp::Le.to_string(), "<=");
        assert_eq!(CompareOp::Gt.to_string(), ">");
        assert_eq!(CompareOp::Ge.to_string(), ">=");

        let asc = OrderByClause::asc(column.clone());
        assert!(asc.ascending);
        assert!(!asc.nulls_first);
        assert!(asc.expr.is_none());

        let desc = OrderByClause::desc(column).with_expr(Expr::lit(Value::Integer(1)));
        assert!(!desc.ascending);
        assert!(desc.nulls_first);
        assert!(desc.expr.is_some());
    }

    #[test]
    fn path_and_node_selector_builders_set_expected_defaults() {
        let from = NodeSelector::by_id("a");
        let to = NodeSelector::by_row("hosts", 42);
        let path = PathQuery::new(from, to).alias("p").via_label("CONNECTS_TO");
        assert_eq!(path.alias.as_deref(), Some("p"));
        assert_eq!(path.via, vec!["CONNECTS_TO"]);
        assert_eq!(path.max_length, 10);
        assert!(path.filter.is_none());

        assert!(matches!(NodeSelector::by_id("n1"), NodeSelector::ById(id) if id == "n1"));
        assert!(matches!(
            NodeSelector::by_label("Host"),
            NodeSelector::ByType { node_label, filter: None } if node_label == "Host"
        ));
        assert!(matches!(
            NodeSelector::by_row("hosts", 7),
            NodeSelector::ByRow { table, row_id } if table == "hosts" && row_id == 7
        ));
    }

    #[test]
    fn vector_and_hybrid_builders_set_options() {
        let literal = VectorSource::literal(vec![0.1, 0.2]);
        assert!(matches!(literal, VectorSource::Literal(ref values) if values == &[0.1, 0.2]));
        assert!(matches!(VectorSource::text("ssh"), VectorSource::Text(ref text) if text == "ssh"));
        assert!(matches!(
            VectorSource::reference("embeddings", 9),
            VectorSource::Reference { collection, vector_id }
                if collection == "embeddings" && vector_id == 9
        ));

        let vector = VectorQuery::new("embeddings", VectorSource::text("ssh"))
            .limit(3)
            .with_filter(MetadataFilter::eq("source", "nmap"))
            .with_vectors()
            .min_similarity(0.8)
            .alias("sim");
        assert_eq!(vector.collection, "embeddings");
        assert_eq!(vector.k, 3);
        assert!(vector.filter.is_some());
        assert!(vector.include_vectors);
        assert!(vector.include_metadata);
        assert_eq!(vector.threshold, Some(0.8));
        assert_eq!(vector.alias.as_deref(), Some("sim"));

        let hybrid = HybridQuery::new(table_expr("hosts"), vector)
            .with_fusion(FusionStrategy::RRF { k: 60 })
            .limit(5)
            .alias("hy");
        assert!(matches!(*hybrid.structured, QueryExpr::Table(_)));
        assert_eq!(hybrid.limit, Some(5));
        assert_eq!(hybrid.alias.as_deref(), Some("hy"));
        assert!(matches!(hybrid.fusion, FusionStrategy::RRF { k: 60 }));
    }

    #[test]
    fn window_spec_structs_are_constructible() {
        let order = WindowOrderItem {
            expr: Expr::lit(Value::Integer(1)),
            ascending: false,
            nulls_first: true,
        };
        let frame = WindowFrame {
            unit: WindowFrameUnit::Rows,
            start: WindowFrameBound::UnboundedPreceding,
            end: Some(WindowFrameBound::CurrentRow),
        };
        let spec = WindowSpec {
            partition_by: vec![Expr::lit(Value::text("tenant"))],
            order_by: vec![order],
            frame: Some(frame),
        };
        assert_eq!(spec.partition_by.len(), 1);
        assert_eq!(spec.order_by.len(), 1);
        assert!(matches!(
            spec.frame,
            Some(WindowFrame {
                unit: WindowFrameUnit::Rows,
                ..
            })
        ));

        let default = WindowSpec::default();
        assert!(default.partition_by.is_empty());
        assert!(default.order_by.is_empty());
        assert!(default.frame.is_none());

        let _ = Span::synthetic();
    }

    #[test]
    fn config_value_type_aliases_and_display_names() {
        for (input, expected, as_str) in [
            ("bool", ConfigValueType::Bool, "bool"),
            ("boolean", ConfigValueType::Bool, "bool"),
            ("int", ConfigValueType::Int, "int"),
            ("integer", ConfigValueType::Int, "int"),
            ("str", ConfigValueType::String, "string"),
            ("text", ConfigValueType::String, "string"),
            ("url", ConfigValueType::Url, "url"),
            ("json_object", ConfigValueType::Object, "object"),
            ("list", ConfigValueType::Array, "array"),
        ] {
            let parsed = ConfigValueType::parse(input).expect("known type");
            assert_eq!(parsed, expected);
            assert_eq!(parsed.as_str(), as_str);
        }
        assert_eq!(ConfigValueType::parse("bogus"), None);
    }
}

// ============================================================================
// Builders (Fluent API)
// ============================================================================