powdb-query 0.29.1

use crate::ast::{
    AggFunc, AggregateMode, AlterAction, Assignment, Expr, GroupKey, JoinKind, Literal, ViaLink,
    WindowFunc,
};
use powdb_storage::stored_json_path::StoredJsonPathV1;

/// A column definition carried by `PlanNode::CreateTable`. Replaces the
/// old `(name, type_name, required)` tuple so the `unique` modifier can
/// flow from the parser through to the executor's DDL arm.
#[derive(Debug, Clone)]
pub struct CreateField {
    pub name: String,
    pub type_name: String,
    pub required: bool,
    pub unique: bool,
    /// Literal default applied when an insert omits this column.
    pub default: Option<Literal>,
    /// `true` when declared `auto` — integer column auto-assigned from a
    /// per-table sequence when an insert omits it.
    pub auto: bool,
}

/// Physical plan nodes — what the executor actually runs.
#[derive(Debug, Clone)]
pub enum PlanNode {
    SeqScan {
        table: String,
    },
    /// Mission E1.2: sequential scan that renames output columns to
    /// `alias.field`. Used exclusively as the leaves of a join plan so
    /// downstream `NestedLoopJoin` + `Filter` + `Project` nodes can resolve
    /// `Expr::QualifiedField` lookups by direct column-name match. Kept
    /// separate from `SeqScan` so the single-table fast paths (which match
    /// on `PlanNode::SeqScan { .. }` in many places) stay untouched.
    AliasScan {
        table: String,
        alias: String,
    },
    IndexScan {
        table: String,
        column: String,
        key: Expr,
    },
    /// B+tree range scan: returns rows where the indexed column falls within
    /// the given bounds. Generated by the planner when it detects inequality
    /// predicates (>, >=, <, <=, BETWEEN) on an indexed column. The executor
    /// falls back to SeqScan+Filter if no index exists on the column.
    RangeScan {
        table: String,
        column: String,
        /// Lower bound: (expr, inclusive). None = unbounded below.
        start: Option<(Expr, bool)>,
        /// Upper bound: (expr, inclusive). None = unbounded above.
        end: Option<(Expr, bool)>,
    },
    /// Speculative equality lookup through an expression index. The planner is
    /// catalog-pure; the executor resolves `path` at runtime and reconstructs
    /// the equivalent path predicate if no matching index exists.
    ExprIndexScan {
        table: String,
        path: StoredJsonPathV1,
        key: Expr,
    },
    /// Speculative bounded lookup through an expression index. `path` is the
    /// persisted table-local identity, never a runtime catalog identifier.
    ExprRangeScan {
        table: String,
        path: StoredJsonPathV1,
        start: Option<(Expr, bool)>,
        end: Option<(Expr, bool)>,
    },
    /// Exact single-path ORDER BY + LIMIT/OFFSET candidate. Missing or
    /// incompatible runtime indexes fall back to a normal scan/sort/slice.
    OrderedExprIndexScan {
        table: String,
        path: StoredJsonPathV1,
        descending: bool,
        limit: Expr,
        offset: Option<Expr>,
    },
    Filter {
        input: Box<PlanNode>,
        predicate: Expr,
    },
    Project {
        input: Box<PlanNode>,
        fields: Vec<ProjectField>,
    },
    /// Language-lab slice: projection with one or more nested sub-query
    /// fields. `input` is the parent pipeline (AliasScan-based so qualified
    /// references resolve by column name); each nested field is assembled by
    /// a single hash-build pass over its child table, O(parent + child).
    /// Kept separate from `Project` so the single-table fast paths and the
    /// plan cache never see this shape.
    NestedProject {
        input: Box<PlanNode>,
        fields: Vec<NestedProjectField>,
    },
    Sort {
        input: Box<PlanNode>,
        keys: Vec<SortKey>,
    },
    Limit {
        input: Box<PlanNode>,
        count: Expr,
    },
    Offset {
        input: Box<PlanNode>,
        count: Expr,
    },
    Aggregate {
        input: Box<PlanNode>,
        function: AggFunc,
        argument: Option<Expr>,
        mode: AggregateMode,
        /// Source alias whose physical row identity controls symmetric
        /// de-duplication. `None` means raw aggregation semantics.
        provenance_alias: Option<String>,
    },
    /// Mission E1.2: nested-loop join. Correctness-first implementation —
    /// O(L × R) scan for every join. E1.3 will add a hash-join fast path
    /// for equijoins (the common case). The executor handles `Inner`,
    /// `Cross`, and `LeftOuter`; `RightOuter` is rewritten by the planner
    /// into a `LeftOuter` with swapped inputs.
    NestedLoopJoin {
        left: Box<PlanNode>,
        right: Box<PlanNode>,
        /// Join predicate. `None` for `Cross` joins (emit every pair).
        on: Option<Expr>,
        kind: JoinKind,
    },
    Distinct {
        input: Box<PlanNode>,
    },
    /// Mission E2b: grouped aggregation. Output columns are
    /// `keys ++ [agg.output_name for agg in aggregates]`. The optional
    /// `having` predicate is evaluated against each output row *after*
    /// aggregation — it can reference both key columns and aggregate
    /// output names (the planner rewrites `FunctionCall` nodes in the
    /// HAVING expression into `Field("__agg_N")` references).
    GroupBy {
        input: Box<PlanNode>,
        keys: Vec<GroupKey>,
        aggregates: Vec<GroupAgg>,
        having: Option<Expr>,
    },
    AlterTable {
        table: String,
        action: AlterAction,
    },
    DropTable {
        name: String,
        /// `drop if exists` — a missing table is a no-op instead of an error.
        if_exists: bool,
    },
    Insert {
        table: String,
        /// One assignment-block per row to insert. Always at least one.
        rows: Vec<Vec<Assignment>>,
        /// `true` when `returning` was requested — executor returns the
        /// inserted rows instead of a modified-count.
        returning: bool,
    },
    /// UPSERT: probe index on `key_column` — if miss, insert; if hit, update.
    Upsert {
        table: String,
        key_column: String,
        assignments: Vec<Assignment>,
        on_conflict: Vec<Assignment>,
    },
    Update {
        input: Box<PlanNode>,
        table: String,
        assignments: Vec<Assignment>,
        /// `true` when `returning` was requested — executor returns the
        /// post-update rows instead of a modified-count.
        returning: bool,
    },
    Delete {
        input: Box<PlanNode>,
        table: String,
        /// `true` when `returning` was requested — executor returns the
        /// pre-delete rows instead of a modified-count.
        returning: bool,
    },
    CreateTable {
        name: String,
        fields: Vec<CreateField>,
        /// `type X if not exists { ... }` — re-declaring an existing type is
        /// a no-op instead of an error.
        if_not_exists: bool,
    },
    /// `link <Owner>.<name> -> <Target> on <local> = <target>`: declare a
    /// persistent entity link. Lowers to `Catalog::create_link`, which
    /// validates the tables/columns and derives the cardinality.
    CreateLink {
        owner: String,
        name: String,
        target: String,
        local_key: String,
        target_key: String,
    },
    /// `drop link <Owner>.<name>` (also spelled
    /// `alter <Owner> drop link <name>`): remove a declared entity link.
    /// Lowers to `Catalog::drop_link`. The owner's rows and columns are
    /// untouched; only the relationship metadata goes.
    DropLink {
        owner: String,
        name: String,
        /// `drop link if exists` — a missing link is a no-op instead of an
        /// error.
        if_exists: bool,
    },
    /// `schema` — list every type (table) in the catalog. Read-only; reads
    /// live catalog state at execution time, so a cached plan is never stale.
    ListTypes,
    /// `describe <Type>` — the columns and indexes of one type. Read-only.
    Describe {
        table: String,
    },
    /// `schema links`: list every declared entity link. Read-only; reads
    /// live catalog state at execution time, so a cached plan is never stale.
    ListLinks,
    /// Create a materialized view: execute query, store results, register.
    CreateView {
        name: String,
        query_text: String,
    },
    /// Explicitly refresh a materialized view.
    RefreshView {
        name: String,
    },
    /// Drop a materialized view (backing table + registry entry).
    DropView {
        name: String,
        /// `drop view if exists` — a missing view is a no-op instead of an error.
        if_exists: bool,
    },
    /// Window function computation layer.
    Window {
        input: Box<PlanNode>,
        windows: Vec<WindowDef>,
    },
    /// `UNION [ALL]`: execute both sides, concatenate (ALL) or deduplicate.
    Union {
        left: Box<PlanNode>,
        right: Box<PlanNode>,
        all: bool,
    },
    /// EXPLAIN: format the inner plan tree as a text result without executing.
    Explain {
        input: Box<PlanNode>,
    },
    Begin,
    Commit,
    Rollback,
}

#[derive(Debug, Clone)]
pub struct ProjectField {
    pub alias: Option<String>,
    pub expr: Expr,
}

/// One output column of a `PlanNode::NestedProject`.
#[derive(Debug, Clone)]
pub enum NestedProjectField {
    /// A scalar field, evaluated against the parent row like `Project`.
    Plain(ProjectField),
    /// A nested sub-query field, emitted as a PJ1 JSON array of objects.
    Nested(Box<NestedProjection>),
    /// A scalar link traversal field (`o.user.name`), emitted as the
    /// traversed target column's value (or empty when the FK is NULL or
    /// dangling at any hop).
    Link(Box<ScalarLinkField>),
}

/// A scalar link traversal projection field. Like `via_link` on
/// [`NestedProjection`], the planner cannot resolve link names (it never
/// touches the catalog), so `resolved` is `None` out of the planner and is
/// filled in by the executor before assembly.
#[derive(Debug, Clone)]
pub struct ScalarLinkField {
    /// Output column name (the field alias, or the dotted source spelling).
    pub name: String,
    /// The outer scan alias the path starts from (`o`).
    pub outer_alias: String,
    /// Declared to-one link names in traversal order (`["user", "company"]`).
    pub links: Vec<String>,
    /// The column read on the final target (`name`).
    pub column: String,
    /// Filled by the executor's catalog resolution; `None` out of the planner.
    pub resolved: Option<ScalarLinkResolved>,
}

/// The catalog-resolved form of a scalar link path.
#[derive(Debug, Clone)]
pub struct ScalarLinkResolved {
    /// The FK column on the parent scan, alias-qualified as the parent's
    /// `AliasScan` emits it (`o.user_id`).
    pub first_fk: String,
    /// One hop per link name, in traversal order.
    pub hops: Vec<ScalarLinkHop>,
}

/// One resolved hop of a scalar link path: look the incoming FK value up by
/// `key_col` in `table`, then read `out_col` (the next hop's FK column, or the
/// final target column on the last hop).
#[derive(Debug, Clone)]
pub struct ScalarLinkHop {
    pub table: String,
    pub key_col: String,
    pub out_col: String,
}

/// The resolved form of one nested sub-query projection field. Correlation
/// columns are pre-split by the planner: at the top level `parent_key` is
/// the qualified parent column name (`u.id`) as produced by an `AliasScan`;
/// for a nested-in-nested field it is the bare column name of the enclosing
/// child table. `child_key` is always the bare child column name.
#[derive(Debug, Clone)]
pub struct NestedProjection {
    /// Output column name (the projection field's alias).
    pub name: String,
    pub table: String,
    /// Set when this level came from a block link traversal
    /// (`orders: u.orders { ... }`). While set, `table`/`child_key`/
    /// `parent_key` are placeholders; the executor resolves them from the
    /// persistent catalog before assembly (like `RangeScan` late lowering)
    /// and clears this. The planner never touches the catalog, so it cannot
    /// resolve it.
    pub via_link: Option<ViaLink>,
    /// The child alias from the source text, kept for EXPLAIN and errors.
    pub alias: String,
    /// The enclosing scope's alias, kept for EXPLAIN (the executor
    /// correlates through `parent_key` alone).
    pub parent_alias: String,
    pub child_key: String,
    pub parent_key: String,
    /// Residual filter conditions beyond the correlation predicate,
    /// rewritten by the planner to reference bare child columns.
    pub residual: Option<Expr>,
    /// Per-parent ordering: `(bare child column, descending)` keys applied
    /// to each parent's child bucket before truncation.
    pub order: Vec<(String, bool)>,
    /// Per-parent truncation bounds, applied after ordering. Must evaluate
    /// to non-negative integer literals (like top-level limit/offset).
    pub limit: Option<Expr>,
    pub offset: Option<Expr>,
    /// Source wrote `offset` before `limit`; the plan cache refuses this
    /// form (see `plan_cache::nested_projection_defeats_cache`).
    pub offset_before_limit: bool,
    pub fields: Vec<NestedField>,
}

/// One entry in a nested projection's output object.
#[derive(Debug, Clone)]
pub enum NestedField {
    /// A scalar child column emitted under `key`.
    Scalar { key: String, column: String },
    /// A deeper nested sub-query, correlated to this child table by its
    /// (bare) `parent_key` column.
    Nested(Box<NestedProjection>),
}

impl NestedProjection {
    /// Visit this projection's child table and every deeper nested child
    /// table. Used for dirty-view escalation and auto-refresh.
    pub fn visit_tables(&self, visit: &mut dyn FnMut(&str)) {
        visit(&self.table);
        for field in &self.fields {
            if let NestedField::Nested(inner) = field {
                inner.visit_tables(visit);
            }
        }
    }
}

#[derive(Debug, Clone)]
pub struct SortKey {
    pub expr: Expr,
    pub descending: bool,
}

/// One aggregate computation inside a `PlanNode::GroupBy`.
#[derive(Debug, Clone)]
pub struct GroupAgg {
    pub function: AggFunc,
    pub argument: Expr,
    pub mode: AggregateMode,
    /// Source alias whose physical row identity controls symmetric
    /// de-duplication. `None` means raw aggregation semantics.
    pub provenance_alias: Option<String>,
    /// Synthetic output column name (`__agg_0`, `__agg_1`, …).
    pub output_name: String,
}

/// One window function definition inside a `PlanNode::Window`.
#[derive(Debug, Clone)]
pub struct WindowDef {
    pub function: WindowFunc,
    pub args: Vec<Expr>,
    pub mode: AggregateMode,
    pub partition_by: Vec<Expr>,
    pub order_by: Vec<SortKey>,
    pub output_name: String,
}