pgevolve_core/plan/

edges.rs

//! Dependency edge extraction from a [`Catalog`].
//!
//! Edges follow the convention "A depends on B" — i.e., for each edge A → B,
//! B must be created before A. The four edge sources from spec §6.4:
//!
//! - schema ⟵ table ⟵ default-using-sequence
//! - table ⟵ index
//! - FK constraint ⟵ both endpoints (own table + referenced table)
//! - sequence ⟵ owning table (`OWNED BY`)
//!
//! Both `build_create_graph` (over the source catalog) and `build_drop_graph`
//! (over the target catalog) use the same edge logic — drop ordering is
//! produced by reversing the topological sort, not by reversing the edges.

use serde::{Deserialize, Serialize};

use crate::identifier::{Identifier, QualifiedName};
use crate::ir::catalog::Catalog;
use crate::ir::collation::BUILTIN_COLLATIONS;
use crate::ir::column_type::ColumnType;
use crate::ir::constraint::ConstraintKind;
use crate::ir::default_expr::DefaultExpr;
use crate::ir::function::ReturnType;
use crate::ir::user_type::UserTypeKind;
use crate::plan::graph::Graph;

pub use crate::ir::function::NormalizedArgTypes;

/// Identifies any IR object uniquely within a [`Catalog`].
///
/// `Schema` carries an `Identifier` (schemas are not schema-qualified). All
/// other variants carry [`QualifiedName`]. `Constraint` is identified by
/// `(table, name)` because constraint names are scoped to their table.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum NodeId {
    /// A schema (namespace).
    Schema(Identifier),
    /// A table.
    Table(QualifiedName),
    /// An index.
    Index(QualifiedName),
    /// A sequence.
    Sequence(QualifiedName),
    /// A constraint identified by its owning table and constraint name.
    Constraint {
        /// Owning table.
        table: QualifiedName,
        /// Constraint name (the `name` half of the constraint's qname).
        name: Identifier,
    },
    /// A view (`CREATE VIEW`).
    View(QualifiedName),
    /// A materialized view (`CREATE MATERIALIZED VIEW`).
    Mv(QualifiedName),
    /// A user-defined type (enum, domain, or composite).
    Type(QualifiedName),
    /// A user-defined function — disambiguated by argument types (Decision 7).
    Function(QualifiedName, NormalizedArgTypes),
    /// A user-defined procedure — identified by qname only (Decision 2).
    Procedure(QualifiedName),
    /// An installed extension.
    Extension(Identifier),
    /// A trigger (qname unique within schema).
    Trigger(QualifiedName),
    /// A publication (not schema-qualified — publications are a per-database
    /// global namespace).
    Publication(Identifier),
    /// A subscription (not schema-qualified — subscriptions are a per-database
    /// global namespace, like publications).
    Subscription(Identifier),
    /// A database-global event trigger.
    EventTrigger(Identifier),
    /// A statistics object (`CREATE STATISTICS schema.name`).
    Statistic(QualifiedName),
    /// A user-defined collation (`CREATE COLLATION schema.name`).
    ///
    /// The variant lands in v0.3.8 so `ordering::change_node` can route
    /// collation changes correctly; the actual graph *edges* (column /
    /// domain / range / composite-attribute → collation) are added in a
    /// follow-up stage.
    Collation(QualifiedName),
    /// A user-defined aggregate — disambiguated by argument types (aggregates
    /// are overloadable, like functions). The aggregate depends on its state
    /// function (`SFUNC`) and optional final function (`FINALFUNC`), both of
    /// which are managed [`NodeId::Function`] nodes.
    Aggregate(QualifiedName, NormalizedArgTypes),
    /// A `CREATE CAST` — identified by `(source_type, target_type)`. Not
    /// schema-scoped; casts are database-global objects. Depends on the
    /// conversion function (when `WITH FUNCTION`) and on managed source/target
    /// types (when the types are user-defined and in the managed catalog).
    Cast(QualifiedName, QualifiedName),
    /// A `CREATE TEXT SEARCH DICTIONARY` — schema-scoped.
    TsDictionary(QualifiedName),
    /// A `CREATE TEXT SEARCH CONFIGURATION` — schema-scoped. Depends on any
    /// managed [`NodeId::TsDictionary`] objects referenced in its token-type
    /// mappings.
    TsConfiguration(QualifiedName),
}

/// Find the managed function that backs an aggregate's `sfunc`/`finalfunc`.
///
/// An aggregate's `sfunc` has the implied signature `(state_type, arg_types…)`
/// and its `finalfunc` has `(state_type)`. We locate the exact managed function
/// overload so the edge points at the right [`NodeId::Function`]:
///
/// 1. Filter `catalog.functions` to those whose `qname == fn_qname`.
/// 2. If exactly one matches, use it (no signature matching needed).
/// 3. Otherwise narrow to overloads whose IN-arg count equals the implied
///    signature's arg count.
/// 4. If still ambiguous, match by comparing each declared arg type to the
///    implied types.
///
/// Returns `None` if no managed function matches (the closed-world check in a
/// later task surfaces this drift; here we simply skip the edge, mirroring the
/// event-trigger edge's behavior when its function is unmanaged).
fn find_sfunc<'a>(
    catalog: &'a Catalog,
    fn_qname: &QualifiedName,
    implied_arg_types: &[ColumnType],
) -> Option<&'a crate::ir::function::Function> {
    let candidates: Vec<&crate::ir::function::Function> = catalog
        .functions
        .iter()
        .filter(|f| &f.qname == fn_qname)
        .collect();
    match candidates.as_slice() {
        [] => None,
        [only] => Some(only),
        many => {
            let by_arity: Vec<&crate::ir::function::Function> = many
                .iter()
                .copied()
                .filter(|f| positional_arg_types(f).len() == implied_arg_types.len())
                .collect();
            match by_arity.as_slice() {
                [] => None,
                [only] => Some(only),
                still_many => still_many.iter().copied().find(|f| {
                    positional_arg_types(f)
                        .into_iter()
                        .eq(implied_arg_types.iter())
                }),
            }
        }
    }
}

/// Positional (IN/INOUT/VARIADIC) argument types of `f`, in declaration order —
/// the call signature, matching how `NormalizedArgTypes` is built.
fn positional_arg_types(f: &crate::ir::function::Function) -> Vec<&ColumnType> {
    use crate::ir::function::ArgMode;
    f.args
        .iter()
        .filter(|a| matches!(a.mode, ArgMode::In | ArgMode::InOut | ArgMode::Variadic))
        .map(|a| &a.ty)
        .collect()
}

/// Build the `NormalizedArgTypes` identity for an aggregate from its raw
/// argument types. Aggregate args are always positional (`IN`), so we wrap
/// each `ColumnType` in an `IN` [`crate::ir::function::FunctionArg`] and reuse
/// `NormalizedArgTypes::from_args` for an identical hash to the function side.
pub(crate) fn aggregate_arg_types(arg_types: &[ColumnType]) -> NormalizedArgTypes {
    use crate::ir::function::{ArgMode, FunctionArg};
    let args: Vec<FunctionArg> = arg_types
        .iter()
        .map(|ty| FunctionArg {
            name: None,
            mode: ArgMode::In,
            ty: ty.clone(),
            default: None,
        })
        .collect();
    NormalizedArgTypes::from_args(&args)
}

/// Returns `true` iff `qname` refers to a managed collation in `catalog` —
/// i.e., a collation we should emit a dependency edge for. Built-in
/// `pg_catalog` collations (`C`, `POSIX`, `und-x-icu`, …) are skipped.
fn should_add_collation_edge(qname: &QualifiedName, catalog: &Catalog) -> bool {
    if qname.schema.as_str() == "pg_catalog" {
        return false;
    }
    let name = qname.name.as_str();
    if BUILTIN_COLLATIONS.contains(&name) {
        return false;
    }
    catalog.collations.iter().any(|c| &c.qname == qname)
}

/// Build the dependency graph for `catalog`, used for create/modify ordering.
///
/// Topologically sorting this graph yields **dependencies first**: schemas
/// before tables, tables before indexes, referenced tables before FKs, etc.
#[allow(clippy::too_many_lines)] // two-phase walk of every catalog object family adding nodes + edges; one place per object.
pub fn build_create_graph(catalog: &Catalog) -> Graph<NodeId> {
    let mut g = Graph::new();

    // Phase 1: every IR object gets a node, even if it has no edges.
    for s in &catalog.schemas {
        g.add_node(NodeId::Schema(s.name.clone()));
    }
    for t in &catalog.tables {
        g.add_node(NodeId::Table(t.qname.clone()));
    }
    for i in &catalog.indexes {
        g.add_node(NodeId::Index(i.qname.clone()));
    }
    for s in &catalog.sequences {
        g.add_node(NodeId::Sequence(s.qname.clone()));
    }
    for t in &catalog.tables {
        for c in &t.constraints {
            g.add_node(NodeId::Constraint {
                table: t.qname.clone(),
                name: c.qname.name.clone(),
            });
        }
    }
    // Register view and MV nodes so they participate in topological ordering
    // and body-dependency edges are rooted correctly.
    for v in &catalog.views {
        g.add_node(NodeId::View(v.qname.clone()));
    }
    for mv in &catalog.materialized_views {
        g.add_node(NodeId::Mv(mv.qname.clone()));
    }
    // Register user-defined type nodes.
    for t in &catalog.types {
        g.add_node(NodeId::Type(t.qname.clone()));
    }
    // Register function and procedure nodes.
    for f in &catalog.functions {
        g.add_node(NodeId::Function(
            f.qname.clone(),
            f.arg_types_normalized.clone(),
        ));
    }
    for p in &catalog.procedures {
        g.add_node(NodeId::Procedure(p.qname.clone()));
    }
    // Register triggers; trigger depends on its target relation and function.
    for t in &catalog.triggers {
        g.add_node(NodeId::Trigger(t.qname.clone()));
        let target_node = if catalog.tables.iter().any(|x| x.qname == t.table) {
            NodeId::Table(t.table.clone())
        } else if catalog.views.iter().any(|x| x.qname == t.table) {
            NodeId::View(t.table.clone())
        } else if catalog
            .materialized_views
            .iter()
            .any(|x| x.qname == t.table)
        {
            NodeId::Mv(t.table.clone())
        } else {
            // Unresolved target — the lint rule trigger-references-unmanaged-table
            // catches this in T9. Skip the edge so the graph builder doesn't
            // panic on a missing target node.
            continue;
        };
        g.add_edge(NodeId::Trigger(t.qname.clone()), target_node);
        if let Some(func) = catalog
            .functions
            .iter()
            .find(|f| f.qname == t.function_qname)
        {
            g.add_edge(
                NodeId::Trigger(t.qname.clone()),
                NodeId::Function(t.function_qname.clone(), func.arg_types_normalized.clone()),
            );
        }
    }
    // Register extensions; an extension with WITH SCHEMA s depends on the schema.
    for e in &catalog.extensions {
        g.add_node(NodeId::Extension(e.name.clone()));
        if let Some(schema) = &e.schema {
            g.add_edge(
                NodeId::Extension(e.name.clone()),
                NodeId::Schema(schema.clone()),
            );
        }
    }
    // Register publications. For Selective publications, add edges from each
    // referenced table and schema to the publication node so publications are
    // ordered after their dependencies. AllTables publications have no explicit
    // edges; they are ordered by tier rule.
    for p in &catalog.publications {
        let pub_node = NodeId::Publication(p.name.clone());
        g.add_node(pub_node.clone());
        if let crate::ir::publication::PublicationScope::Selective { schemas, tables } = &p.scope {
            for t in tables {
                g.add_edge(pub_node.clone(), NodeId::Table(t.qname.clone()));
            }
            for s in schemas {
                g.add_edge(pub_node.clone(), NodeId::Schema(s.clone()));
            }
        }
    }
    // Register subscriptions. Subscriptions cross-reference publications in
    // a *different* cluster — no local dep edges anchor them. They are
    // registered as isolated nodes; the tier rule in ordering.rs schedules
    // them create-last, drop-first via sort_key.
    for s in &catalog.subscriptions {
        g.add_node(NodeId::Subscription(s.name.clone()));
    }
    // Event triggers: global nodes; edge to the function they execute so the
    // function is created before the event trigger (and dropped after).
    for et in &catalog.event_triggers {
        g.add_node(NodeId::EventTrigger(et.name.clone()));
        if let Some(func) = catalog.functions.iter().find(|f| f.qname == et.function) {
            g.add_edge(
                NodeId::EventTrigger(et.name.clone()),
                NodeId::Function(et.function.clone(), func.arg_types_normalized.clone()),
            );
        }
    }
    // Register statistics; each depends on its target table (must be created
    // after the table exists and dropped before the table is dropped).
    for s in &catalog.statistics {
        let stat_node = NodeId::Statistic(s.qname.clone());
        g.add_node(stat_node.clone());
        g.add_edge(stat_node, NodeId::Table(s.target.clone()));
    }
    // Register collations and wire each to its schema (collation depends on
    // schema — CREATE COLLATION must follow CREATE SCHEMA; DROP COLLATION must
    // precede DROP SCHEMA). Without this edge the drop ordering is only
    // accidentally correct (via enum discriminant tie-breaking).
    // Edges from table columns / domain / range / composite attributes
    // → collation are added below in the relevant loops.
    for c in &catalog.collations {
        let coll_node = NodeId::Collation(c.qname.clone());
        g.add_node(coll_node.clone());
        g.add_edge(coll_node, NodeId::Schema(c.qname.schema.clone()));
    }
    // Register aggregates and wire each to the managed function(s) it relies on:
    // the state function (`SFUNC`, implied signature `(state_type, arg_types…)`)
    // and the optional final function (`FINALFUNC`, implied signature
    // `(state_type)`). The edge direction is aggregate (dependent) → function
    // (dependency), so the function is created before the aggregate and dropped
    // after it — mirroring the event-trigger → function edge.
    for agg in &catalog.aggregates {
        let agg_node = NodeId::Aggregate(agg.qname.clone(), aggregate_arg_types(&agg.arg_types));
        g.add_node(agg_node.clone());
        // sfunc implied signature: (state_type, arg_types…)
        let mut sfunc_sig = Vec::with_capacity(1 + agg.arg_types.len());
        sfunc_sig.push(agg.state_type.clone());
        sfunc_sig.extend(agg.arg_types.iter().cloned());
        if let Some(sfunc) = find_sfunc(catalog, &agg.sfunc, &sfunc_sig) {
            g.add_edge(
                agg_node.clone(),
                NodeId::Function(sfunc.qname.clone(), sfunc.arg_types_normalized.clone()),
            );
        }
        // finalfunc implied signature: (state_type)
        if let Some(finalfunc_qname) = &agg.finalfunc
            && let Some(finalfunc) = find_sfunc(
                catalog,
                finalfunc_qname,
                std::slice::from_ref(&agg.state_type),
            )
        {
            g.add_edge(
                agg_node,
                NodeId::Function(
                    finalfunc.qname.clone(),
                    finalfunc.arg_types_normalized.clone(),
                ),
            );
        }
    }
    // Register casts and wire each to:
    //   1. The conversion function (only `WITH FUNCTION`): find the managed
    //      function overload whose qname and positional arg types match the
    //      cast's recorded `arg_types`. Skip the edge if no managed overload
    //      matches (closed-world drift is surfaced later, not here).
    //   2. Managed source / target types: if either type is a user-defined type
    //      present in `catalog.types`, add a dependency edge so the type is
    //      created before the cast and dropped after it. Built-in pg_catalog
    //      types are absent from `catalog.types` → no edge for them.
    for cast in &catalog.casts {
        let cast_node = NodeId::Cast(cast.source.clone(), cast.target.clone());
        g.add_node(cast_node.clone());
        // Edge to conversion function (WITH FUNCTION only).
        if let crate::ir::cast::CastMethod::Function { name, arg_types } = &cast.method
            && let Some(func) = find_sfunc(catalog, name, arg_types)
        {
            g.add_edge(
                cast_node.clone(),
                NodeId::Function(func.qname.clone(), func.arg_types_normalized.clone()),
            );
        }
        // Edge to source type if it is a managed user-defined type.
        if catalog.types.iter().any(|t| t.qname == cast.source) {
            g.add_edge(cast_node.clone(), NodeId::Type(cast.source.clone()));
        }
        // Edge to target type if it is a managed user-defined type.
        if catalog.types.iter().any(|t| t.qname == cast.target) {
            g.add_edge(cast_node.clone(), NodeId::Type(cast.target.clone()));
        }
    }

    // Register text-search dictionaries; each depends on its schema (same
    // pattern as collations). No edges to the template — templates are always
    // unmanaged (e.g. pg_catalog.snowball).
    for d in &catalog.ts_dictionaries {
        let dict_node = NodeId::TsDictionary(d.qname.clone());
        g.add_node(dict_node.clone());
        g.add_edge(dict_node, NodeId::Schema(d.qname.schema.clone()));
    }
    // Register text-search configurations; each depends on its schema and on
    // every managed dictionary referenced in its token-type mappings.
    // A mapping may list built-in/extension dicts (e.g. pg_catalog.simple)
    // that are absent from catalog.ts_dictionaries — skip those edges.
    // Deduplicate: a dict referenced by multiple token-type mappings gets
    // only one edge (use a HashSet to track seen qnames).
    for cfg in &catalog.ts_configurations {
        let cfg_node = NodeId::TsConfiguration(cfg.qname.clone());
        g.add_node(cfg_node.clone());
        g.add_edge(cfg_node.clone(), NodeId::Schema(cfg.qname.schema.clone()));
        let mut seen: std::collections::HashSet<&QualifiedName> = std::collections::HashSet::new();
        for mapping in &cfg.mappings {
            for dict_qname in &mapping.dictionaries {
                if seen.contains(dict_qname) {
                    continue;
                }
                if catalog
                    .ts_dictionaries
                    .iter()
                    .any(|d| &d.qname == dict_qname)
                {
                    g.add_edge(cfg_node.clone(), NodeId::TsDictionary(dict_qname.clone()));
                    seen.insert(dict_qname);
                }
            }
        }
    }

    // Phase 1b.0: type → schema edges. Every user-defined type lives inside
    // a schema and must be created after CREATE SCHEMA emits.
    for t in &catalog.types {
        g.add_edge(
            NodeId::Type(t.qname.clone()),
            NodeId::Schema(t.qname.schema.clone()),
        );
    }
    // Phase 1b.0 (routines): every function/procedure depends on its schema.
    for f in &catalog.functions {
        let node = NodeId::Function(f.qname.clone(), f.arg_types_normalized.clone());
        g.add_edge(node, NodeId::Schema(f.qname.schema.clone()));
    }
    for p in &catalog.procedures {
        g.add_edge(
            NodeId::Procedure(p.qname.clone()),
            NodeId::Schema(p.qname.schema.clone()),
        );
    }

    // Phase 1b: type → type edges from composite attributes and domain bases.
    // These edges ensure composites/domains that reference other user-defined
    // types are created after those types. Range types additionally depend on
    // their subtype (when managed) and on their canonical / subtype_diff
    // functions (when managed).
    for ut in &catalog.types {
        match &ut.kind {
            UserTypeKind::Composite { attributes } => {
                for attr in attributes {
                    if let ColumnType::UserDefined(dep_qname) = &attr.ty {
                        g.add_edge(
                            NodeId::Type(ut.qname.clone()),
                            NodeId::Type(dep_qname.clone()),
                        );
                    }
                    if let Some(coll_qname) = &attr.collation
                        && should_add_collation_edge(coll_qname, catalog)
                    {
                        g.add_edge(
                            NodeId::Type(ut.qname.clone()),
                            NodeId::Collation(coll_qname.clone()),
                        );
                    }
                }
            }
            UserTypeKind::Domain {
                base, collation, ..
            } => {
                if let ColumnType::UserDefined(base_qname) = base {
                    g.add_edge(
                        NodeId::Type(ut.qname.clone()),
                        NodeId::Type(base_qname.clone()),
                    );
                }
                if let Some(coll_qname) = collation
                    && should_add_collation_edge(coll_qname, catalog)
                {
                    g.add_edge(
                        NodeId::Type(ut.qname.clone()),
                        NodeId::Collation(coll_qname.clone()),
                    );
                }
            }
            UserTypeKind::Range {
                subtype,
                canonical,
                subtype_diff,
                collation,
                ..
            } => {
                // Skip built-in subtypes (pg_catalog.* are not user-managed).
                if subtype.schema.as_str() != "pg_catalog"
                    && catalog.types.iter().any(|t| &t.qname == subtype)
                {
                    g.add_edge(
                        NodeId::Type(ut.qname.clone()),
                        NodeId::Type(subtype.clone()),
                    );
                }
                if let Some(fn_qname) = canonical
                    && let Some(f) = catalog.functions.iter().find(|f| &f.qname == fn_qname)
                {
                    g.add_edge(
                        NodeId::Type(ut.qname.clone()),
                        NodeId::Function(f.qname.clone(), f.arg_types_normalized.clone()),
                    );
                }
                if let Some(fn_qname) = subtype_diff
                    && let Some(f) = catalog.functions.iter().find(|f| &f.qname == fn_qname)
                {
                    g.add_edge(
                        NodeId::Type(ut.qname.clone()),
                        NodeId::Function(f.qname.clone(), f.arg_types_normalized.clone()),
                    );
                }
                if let Some(coll_qname) = collation
                    && should_add_collation_edge(coll_qname, catalog)
                {
                    g.add_edge(
                        NodeId::Type(ut.qname.clone()),
                        NodeId::Collation(coll_qname.clone()),
                    );
                }
            }
            UserTypeKind::Enum { .. } => {}
        }
    }

    // Phase 1c: table → type edges from columns with user-defined types.
    // Tables that reference a user-defined type must be created after the type.
    // The same loop emits table → collation edges for columns with an explicit
    // `COLLATE` clause that points at a managed collation.
    for t in &catalog.tables {
        for col in &t.columns {
            if let ColumnType::UserDefined(type_qname) = &col.ty {
                g.add_edge(
                    NodeId::Table(t.qname.clone()),
                    NodeId::Type(type_qname.clone()),
                );
            }
            if let Some(coll_qname) = &col.collation
                && should_add_collation_edge(coll_qname, catalog)
            {
                g.add_edge(
                    NodeId::Table(t.qname.clone()),
                    NodeId::Collation(coll_qname.clone()),
                );
            }
        }
    }
    // Phase 1c (routines): function/procedure → types referenced in args and
    // return types. These ensure routines are created after their type deps.
    for f in &catalog.functions {
        let node = NodeId::Function(f.qname.clone(), f.arg_types_normalized.clone());
        for arg in &f.args {
            if let ColumnType::UserDefined(t_qname) = &arg.ty {
                g.add_edge(node.clone(), NodeId::Type(t_qname.clone()));
            }
        }
        match &f.return_type {
            ReturnType::Scalar {
                ty: ColumnType::UserDefined(t),
            }
            | ReturnType::SetOf {
                ty: ColumnType::UserDefined(t),
            } => {
                g.add_edge(node.clone(), NodeId::Type(t.clone()));
            }
            ReturnType::Table { columns } => {
                for col in columns {
                    if let ColumnType::UserDefined(t) = &col.ty {
                        g.add_edge(node.clone(), NodeId::Type(t.clone()));
                    }
                }
            }
            _ => {}
        }
    }
    for p in &catalog.procedures {
        let node = NodeId::Procedure(p.qname.clone());
        for arg in &p.args {
            if let ColumnType::UserDefined(t_qname) = &arg.ty {
                g.add_edge(node.clone(), NodeId::Type(t_qname.clone()));
            }
        }
    }

    // Phase 2: tables depend on their schema and on any sequence used as a
    // column default. We add the schema node implicitly via add_edge in case
    // the caller did not declare it (defensive: source-side parsing typically
    // does declare every referenced schema, but a hand-built Catalog might not).
    // Partition children depend on their parent table.
    for t in &catalog.tables {
        g.add_edge(
            NodeId::Table(t.qname.clone()),
            NodeId::Schema(t.qname.schema.clone()),
        );
        for col in &t.columns {
            if let Some(DefaultExpr::Sequence(seq_qname)) = &col.default {
                g.add_edge(
                    NodeId::Table(t.qname.clone()),
                    NodeId::Sequence(seq_qname.clone()),
                );
            }
        }
        if let Some(po) = &t.partition_of {
            // Partition child depends on its parent existing first.
            g.add_edge(
                NodeId::Table(t.qname.clone()),
                NodeId::Table(po.parent.clone()),
            );
        }
    }

    // Phase 2b: views and MVs depend on objects in their body_dependencies.
    // `body_dependencies` edges use NodeId directly (already the correct
    // variant); we just re-register each edge into the graph.
    for v in &catalog.views {
        for dep in &v.body_dependencies {
            g.add_edge(dep.from.clone(), dep.to.clone());
        }
    }
    for mv in &catalog.materialized_views {
        for dep in &mv.body_dependencies {
            g.add_edge(dep.from.clone(), dep.to.clone());
        }
    }

    // Phase 2c: functions and procedures body_dependencies.
    for f in &catalog.functions {
        for dep in &f.body_dependencies {
            g.add_edge(dep.from.clone(), dep.to.clone());
        }
    }
    for p in &catalog.procedures {
        for dep in &p.body_dependencies {
            g.add_edge(dep.from.clone(), dep.to.clone());
        }
    }

    // Phase 3: indexes depend on their parent (table or MV).
    // For `IndexParent::Mv`, we use `NodeId::Mv` so the graph correctly
    // orders CREATE INDEX after CREATE MATERIALIZED VIEW.
    for i in &catalog.indexes {
        use crate::ir::index::IndexParent;
        let parent_node = match &i.on {
            IndexParent::Table(q) => NodeId::Table(q.clone()),
            IndexParent::Mv(q) => NodeId::Mv(q.clone()),
        };
        g.add_edge(NodeId::Index(i.qname.clone()), parent_node);
    }

    // Phase 4: constraints depend on their owning table; FKs additionally
    // depend on the referenced table.
    //
    // For FKs we ALSO add a direct table → referenced_table edge. Inline FKs
    // are emitted as part of `CREATE TABLE`, so the owning table's create
    // statement requires the referenced table to exist first. Without this
    // edge, two-table FK cycles never produce a cycle in the graph and the
    // planner's FK-extraction post-pass would have nothing to detect.
    for t in &catalog.tables {
        for c in &t.constraints {
            let constraint_node = NodeId::Constraint {
                table: t.qname.clone(),
                name: c.qname.name.clone(),
            };
            g.add_edge(constraint_node.clone(), NodeId::Table(t.qname.clone()));
            if let ConstraintKind::ForeignKey(fk) = &c.kind {
                g.add_edge(constraint_node, NodeId::Table(fk.referenced_table.clone()));
                // Self-referential FKs don't induce a real table-level cycle
                // (table can be created first, FK satisfied at row time).
                if fk.referenced_table != t.qname {
                    g.add_edge(
                        NodeId::Table(t.qname.clone()),
                        NodeId::Table(fk.referenced_table.clone()),
                    );
                }
            }
        }
    }

    // Phase 5: an `OWNED BY` sequence depends on its owner table.
    for s in &catalog.sequences {
        if let Some(owner) = &s.owned_by {
            g.add_edge(
                NodeId::Sequence(s.qname.clone()),
                NodeId::Table(owner.table.clone()),
            );
        }
    }

    g
}

/// Build the dependency graph for drop-ordering. Same edges as the create
/// graph; the ordering is reversed at sort time.
pub fn build_drop_graph(catalog: &Catalog) -> Graph<NodeId> {
    build_create_graph(catalog)
}

/// Where a dependency edge came from.
///
/// `Structural` edges are derived from the IR shape itself (schema←table,
/// table←index, FK references, sequence ownership). They exist in v0.1.
///
/// `AstExtracted` edges are derived by walking the parsed AST of an object
/// body (view body, function body, expression-index predicate, etc.).
/// First produced in v0.2 view sub-spec.
///
/// `AstDeclared` edges come from explicit `-- @pgevolve dep:` directives
/// that close the PL/pgSQL-dynamic-SQL gap (Decision 11). First produced
/// in v0.2 function sub-spec.
///
/// Ordering: `Structural < AstExtracted < AstDeclared` — structural edges
/// are tie-broken first in the Kahn min-heap to preserve v0.1 ordering.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum DepSource {
    /// Derived from the IR shape; v0.1 default.
    Structural,
    /// Walked out of a parsed body AST.
    AstExtracted,
    /// Declared by a `-- @pgevolve dep:` directive in source SQL.
    AstDeclared,
}

/// An edge in the dependency graph, with provenance metadata.
///
/// Convention matches the existing graph: `from` depends on `to`, so `to`
/// must be created before `from`.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct DepEdge {
    /// Dependent node (the one that needs `to` to exist first).
    pub from: NodeId,
    /// Dependency target.
    pub to: NodeId,
    /// Provenance of this edge.
    pub source: DepSource,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ir::column::Column;
    use crate::ir::column_type::ColumnType;
    use crate::ir::constraint::{
        Constraint, ConstraintKind, Deferrable, FkMatchType, ForeignKey, ReferentialAction,
    };
    use crate::ir::index::{
        Index, IndexColumn, IndexColumnExpr, IndexMethod, IndexParent, NullsOrder, SortOrder,
    };
    use crate::ir::schema::Schema;
    use crate::ir::sequence::{Sequence, SequenceOwner};
    use crate::ir::table::Table;

    fn id(s: &str) -> Identifier {
        Identifier::from_unquoted(s).unwrap()
    }

    fn qn(schema: &str, name: &str) -> QualifiedName {
        QualifiedName::new(id(schema), id(name))
    }

    fn col_id_bigint() -> Column {
        Column {
            name: id("id"),
            ty: ColumnType::BigInt,
            nullable: false,
            default: None,
            identity: None,
            generated: None,
            collation: None,
            storage: None,
            compression: None,
            comment: None,
        }
    }

    fn col_text_notnull(name: &str) -> Column {
        Column {
            name: id(name),
            ty: ColumnType::Text,
            nullable: false,
            default: None,
            identity: None,
            generated: None,
            collation: None,
            storage: None,
            compression: None,
            comment: None,
        }
    }

    fn pk(name: &str, cols: &[&str]) -> Constraint {
        Constraint {
            qname: qn("app", name),
            kind: ConstraintKind::PrimaryKey {
                columns: cols.iter().map(|c| id(c)).collect(),
                include: vec![],
            },
            deferrable: Deferrable::NotDeferrable,
            comment: None,
        }
    }

    fn fk(name: &str, ref_table: QualifiedName) -> Constraint {
        Constraint {
            qname: qn("app", name),
            kind: ConstraintKind::ForeignKey(ForeignKey {
                columns: vec![id("ref_id")],
                referenced_table: ref_table,
                referenced_columns: vec![id("id")],
                on_update: ReferentialAction::NoAction,
                on_delete: ReferentialAction::NoAction,
                match_type: FkMatchType::Simple,
            }),
            deferrable: Deferrable::NotDeferrable,
            comment: None,
        }
    }

    fn has_edge(g: &Graph<NodeId>, from: &NodeId, to: &NodeId) -> bool {
        g.dependencies_of(from).any(|n| n == to)
    }

    #[test]
    fn empty_catalog_yields_empty_graph() {
        let g = build_create_graph(&Catalog::empty());
        assert_eq!(g.node_count(), 0);
    }

    #[test]
    fn every_object_appears_as_a_node() {
        let mut c = Catalog::empty();
        c.schemas.push(Schema::new(id("app")));
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![col_id_bigint()],
            constraints: vec![pk("users_pkey", &["id"])],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        c.indexes.push(Index {
            qname: qn("app", "users_idx"),
            on: IndexParent::Table(qn("app", "users")),
            method: IndexMethod::BTree,
            columns: vec![IndexColumn {
                expr: IndexColumnExpr::Column(id("id")),
                collation: None,
                opclass: None,
                sort_order: SortOrder::Asc,
                nulls_order: NullsOrder::NullsLast,
            }],
            include: vec![],
            unique: false,
            nulls_not_distinct: false,
            predicate: None,
            tablespace: None,
            comment: None,
            storage: crate::ir::reloptions::IndexStorageOptions::default(),
        });
        c.sequences.push(Sequence {
            qname: qn("app", "seq1"),
            data_type: ColumnType::BigInt,
            start: 1,
            increment: 1,
            min_value: None,
            max_value: None,
            cache: 1,
            cycle: false,
            owned_by: None,
            comment: None,
            owner: None,
            grants: vec![],
        });

        let g = build_create_graph(&c);
        // schema + table + index + sequence + constraint = 5
        assert_eq!(g.node_count(), 5);
    }

    #[test]
    fn table_depends_on_its_schema() {
        let mut c = Catalog::empty();
        c.schemas.push(Schema::new(id("app")));
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![col_id_bigint()],
            constraints: vec![],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        let g = build_create_graph(&c);
        assert!(has_edge(
            &g,
            &NodeId::Table(qn("app", "users")),
            &NodeId::Schema(id("app")),
        ));
    }

    #[test]
    fn index_depends_on_its_table() {
        let mut c = Catalog::empty();
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![col_id_bigint()],
            constraints: vec![],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        c.indexes.push(Index {
            qname: qn("app", "users_idx"),
            on: IndexParent::Table(qn("app", "users")),
            method: IndexMethod::BTree,
            columns: vec![IndexColumn {
                expr: IndexColumnExpr::Column(id("id")),
                collation: None,
                opclass: None,
                sort_order: SortOrder::Asc,
                nulls_order: NullsOrder::NullsLast,
            }],
            include: vec![],
            unique: false,
            nulls_not_distinct: false,
            predicate: None,
            tablespace: None,
            comment: None,
            storage: crate::ir::reloptions::IndexStorageOptions::default(),
        });
        let g = build_create_graph(&c);
        assert!(has_edge(
            &g,
            &NodeId::Index(qn("app", "users_idx")),
            &NodeId::Table(qn("app", "users")),
        ));
    }

    #[test]
    fn fk_constraint_depends_on_both_endpoints() {
        let mut c = Catalog::empty();
        c.tables.push(Table {
            qname: qn("app", "orgs"),
            columns: vec![col_id_bigint()],
            constraints: vec![pk("orgs_pkey", &["id"])],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![
                col_id_bigint(),
                Column {
                    name: id("ref_id"),
                    ty: ColumnType::BigInt,
                    nullable: false,
                    default: None,
                    identity: None,
                    generated: None,
                    collation: None,
                    storage: None,
                    compression: None,
                    comment: None,
                },
            ],
            constraints: vec![fk("users_orgs_fk", qn("app", "orgs"))],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        let g = build_create_graph(&c);
        let fk_node = NodeId::Constraint {
            table: qn("app", "users"),
            name: id("users_orgs_fk"),
        };
        // Owning-table edge.
        assert!(has_edge(&g, &fk_node, &NodeId::Table(qn("app", "users"))));
        // Referenced-table edge.
        assert!(has_edge(&g, &fk_node, &NodeId::Table(qn("app", "orgs"))));
    }

    #[test]
    fn table_depends_on_default_sequence() {
        let mut c = Catalog::empty();
        c.sequences.push(Sequence {
            qname: qn("app", "id_seq"),
            data_type: ColumnType::BigInt,
            start: 1,
            increment: 1,
            min_value: None,
            max_value: None,
            cache: 1,
            cycle: false,
            owned_by: None,
            comment: None,
            owner: None,
            grants: vec![],
        });
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![Column {
                name: id("id"),
                ty: ColumnType::BigInt,
                nullable: false,
                default: Some(DefaultExpr::Sequence(qn("app", "id_seq"))),
                identity: None,
                generated: None,
                collation: None,
                storage: None,
                compression: None,
                comment: None,
            }],
            constraints: vec![],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        let g = build_create_graph(&c);
        assert!(has_edge(
            &g,
            &NodeId::Table(qn("app", "users")),
            &NodeId::Sequence(qn("app", "id_seq")),
        ));
    }

    #[test]
    fn owned_sequence_depends_on_owner_table() {
        let mut c = Catalog::empty();
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![col_id_bigint()],
            constraints: vec![],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        c.sequences.push(Sequence {
            qname: qn("app", "users_id_seq"),
            data_type: ColumnType::BigInt,
            start: 1,
            increment: 1,
            min_value: None,
            max_value: None,
            cache: 1,
            cycle: false,
            owned_by: Some(SequenceOwner {
                table: qn("app", "users"),
                column: id("id"),
            }),
            comment: None,
            owner: None,
            grants: vec![],
        });
        let g = build_create_graph(&c);
        assert!(has_edge(
            &g,
            &NodeId::Sequence(qn("app", "users_id_seq")),
            &NodeId::Table(qn("app", "users")),
        ));
    }

    #[test]
    fn non_fk_constraint_depends_only_on_its_table() {
        let mut c = Catalog::empty();
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![col_id_bigint()],
            constraints: vec![pk("users_pkey", &["id"])],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        let g = build_create_graph(&c);
        let pk_node = NodeId::Constraint {
            table: qn("app", "users"),
            name: id("users_pkey"),
        };
        let deps: Vec<&NodeId> = g.dependencies_of(&pk_node).collect();
        assert_eq!(deps, vec![&NodeId::Table(qn("app", "users"))]);
    }

    #[test]
    fn drop_graph_matches_create_graph() {
        let mut c = Catalog::empty();
        c.tables.push(Table {
            qname: qn("app", "users"),
            columns: vec![col_id_bigint()],
            constraints: vec![pk("users_pkey", &["id"])],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        // Same edges; equality is structural via topological output.
        let cg = build_create_graph(&c);
        let dg = build_drop_graph(&c);
        assert_eq!(cg.topological_sort(), dg.topological_sort());
    }

    #[test]
    fn fk_cycle_produces_table_level_cycle() {
        // Two tables with FKs to each other; each FK induces an inline-create
        // edge table → referenced_table, so the table subgraph cycles.
        let mut c = Catalog::empty();
        c.tables.push(Table {
            qname: qn("app", "a"),
            columns: vec![
                col_id_bigint(),
                Column {
                    name: id("ref_id"),
                    ty: ColumnType::BigInt,
                    nullable: false,
                    default: None,
                    identity: None,
                    generated: None,
                    collation: None,
                    storage: None,
                    compression: None,
                    comment: None,
                },
            ],
            constraints: vec![pk("a_pk", &["id"]), fk("a_to_b", qn("app", "b"))],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        c.tables.push(Table {
            qname: qn("app", "b"),
            columns: vec![
                col_id_bigint(),
                Column {
                    name: id("ref_id"),
                    ty: ColumnType::BigInt,
                    nullable: false,
                    default: None,
                    identity: None,
                    generated: None,
                    collation: None,
                    storage: None,
                    compression: None,
                    comment: None,
                },
            ],
            constraints: vec![pk("b_pk", &["id"]), fk("b_to_a", qn("app", "a"))],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        let g = build_create_graph(&c);
        let err = g.topological_sort().unwrap_err();
        assert!(err.nodes.contains(&NodeId::Table(qn("app", "a"))));
        assert!(err.nodes.contains(&NodeId::Table(qn("app", "b"))));
    }

    #[test]
    fn self_referential_fk_does_not_cycle() {
        // A self-referential FK doesn't force the table to depend on itself —
        // the rows are inserted after the table exists.
        let mut c = Catalog::empty();
        c.tables.push(Table {
            qname: qn("app", "tree"),
            columns: vec![
                col_id_bigint(),
                Column {
                    name: id("ref_id"),
                    ty: ColumnType::BigInt,
                    nullable: true,
                    default: None,
                    identity: None,
                    generated: None,
                    collation: None,
                    storage: None,
                    compression: None,
                    comment: None,
                },
            ],
            constraints: vec![
                pk("tree_pk", &["id"]),
                fk("tree_parent_fk", qn("app", "tree")),
            ],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        let g = build_create_graph(&c);
        assert!(g.topological_sort().is_ok());
    }

    #[test]
    fn partition_child_depends_on_parent() {
        // A partition child table depends on its parent table being created first.
        use crate::ir::partition::{PartitionBounds, PartitionBy, PartitionOf};
        let mut c = Catalog::empty();

        // Parent table with PARTITION BY LIST.
        let parent = Table {
            qname: qn("app", "parent"),
            columns: vec![col_id_bigint(), col_text_notnull("status")],
            constraints: vec![pk("parent_pkey", &["id"])],
            partition_by: Some(PartitionBy {
                strategy: crate::ir::partition::PartitionStrategy::List,
                columns: vec![crate::ir::partition::PartitionColumn {
                    kind: crate::ir::partition::PartitionColumnKind::Column(id("status")),
                    collation: None,
                    opclass: None,
                }],
            }),
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        };
        c.tables.push(parent);

        // Child partition table.
        let child = Table {
            qname: qn("app", "child"),
            columns: vec![col_id_bigint(), col_text_notnull("status")],
            constraints: vec![],
            partition_by: None,
            partition_of: Some(PartitionOf {
                parent: qn("app", "parent"),
                bounds: PartitionBounds::List { values: vec![] },
            }),
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        };
        c.tables.push(child);

        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::Table(qn("app", "child")),
                &NodeId::Table(qn("app", "parent")),
            ),
            "expected child partition → parent table edge"
        );
    }

    // ── User-defined type edge tests ──────────────────────────────────────────

    use crate::ir::user_type::{CompositeAttribute, UserType, UserTypeKind};

    fn make_enum(schema: &str, name: &str) -> UserType {
        UserType {
            qname: qn(schema, name),
            kind: UserTypeKind::Enum { values: vec![] },
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    fn make_composite_with_attr(schema: &str, name: &str, attr_type: ColumnType) -> UserType {
        UserType {
            qname: qn(schema, name),
            kind: UserTypeKind::Composite {
                attributes: vec![CompositeAttribute {
                    name: id("val"),
                    ty: attr_type,
                    collation: None,
                }],
            },
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    fn make_domain_over(schema: &str, name: &str, base: ColumnType) -> UserType {
        UserType {
            qname: qn(schema, name),
            kind: UserTypeKind::Domain {
                base,
                nullable: true,
                default: None,
                check_constraints: vec![],
                collation: None,
            },
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    #[test]
    fn type_nodes_registered() {
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.types.push(make_enum("app", "status"));
        let g = build_create_graph(&c);
        // Type depends ONLY on its schema (no other edges for a bare enum).
        let deps: Vec<_> = g
            .dependencies_of(&NodeId::Type(qn("app", "status")))
            .collect();
        assert_eq!(deps, vec![&NodeId::Schema(id("app"))]);
        // Both the schema node and the type node are registered.
        assert_eq!(g.node_count(), 2);
    }

    #[test]
    fn table_depends_on_user_defined_column_type() {
        let mut c = Catalog::empty();
        c.types.push(make_enum("app", "status"));
        c.tables.push(Table {
            qname: qn("app", "orders"),
            columns: vec![Column {
                name: id("status"),
                ty: ColumnType::UserDefined(qn("app", "status")),
                nullable: false,
                default: None,
                identity: None,
                generated: None,
                collation: None,
                storage: None,
                compression: None,
                comment: None,
            }],
            constraints: vec![],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        });
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::Table(qn("app", "orders")),
                &NodeId::Type(qn("app", "status"))
            ),
            "table must depend on its user-defined column type"
        );
    }

    #[test]
    fn composite_depends_on_user_defined_attribute_type() {
        let mut c = Catalog::empty();
        c.types.push(make_enum("app", "inner_t"));
        c.types.push(make_composite_with_attr(
            "app",
            "outer_t",
            ColumnType::UserDefined(qn("app", "inner_t")),
        ));
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "outer_t")),
                &NodeId::Type(qn("app", "inner_t"))
            ),
            "composite must depend on the type of its user-defined attribute"
        );
    }

    #[test]
    fn domain_depends_on_user_defined_base_type() {
        let mut c = Catalog::empty();
        c.types.push(make_enum("app", "base_t"));
        c.types.push(make_domain_over(
            "app",
            "derived_t",
            ColumnType::UserDefined(qn("app", "base_t")),
        ));
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "derived_t")),
                &NodeId::Type(qn("app", "base_t"))
            ),
            "domain must depend on its user-defined base type"
        );
    }

    // ── Range type edge tests ────────────────────────────────────────────────

    fn make_range(
        schema: &str,
        name: &str,
        subtype: QualifiedName,
        canonical: Option<QualifiedName>,
        subtype_diff: Option<QualifiedName>,
    ) -> UserType {
        UserType {
            qname: qn(schema, name),
            kind: UserTypeKind::Range {
                subtype,
                subtype_opclass: None,
                collation: None,
                canonical,
                subtype_diff,
                multirange_type_name: None,
            },
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    fn make_function(schema: &str, name: &str) -> crate::ir::function::Function {
        use crate::ir::function::{
            FunctionLanguage, NormalizedArgTypes, ParallelSafety, ReturnType, SecurityMode,
            Volatility,
        };
        use crate::parse::normalize_body::NormalizedBody;
        crate::ir::function::Function {
            qname: qn(schema, name),
            args: vec![],
            arg_types_normalized: NormalizedArgTypes::from_args(&[]),
            return_type: ReturnType::Scalar {
                ty: ColumnType::Integer,
            },
            language: FunctionLanguage::Sql,
            body: NormalizedBody::empty(),
            body_dependencies: vec![],
            volatility: Volatility::Volatile,
            strict: false,
            security: SecurityMode::Invoker,
            parallel: ParallelSafety::Unsafe,
            leakproof: false,
            cost: None,
            rows: None,
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    #[test]
    fn range_with_builtin_subtype_adds_no_subtype_edge() {
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.types.push(make_range(
            "app",
            "ir",
            qn("pg_catalog", "int4"),
            None,
            None,
        ));
        let g = build_create_graph(&c);
        // Type edges: only schema; no edge to pg_catalog.int4.
        let deps: Vec<&NodeId> = g.dependencies_of(&NodeId::Type(qn("app", "ir"))).collect();
        assert_eq!(deps, vec![&NodeId::Schema(id("app"))]);
    }

    #[test]
    fn range_with_managed_subtype_adds_type_edge() {
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.types.push(make_enum("app", "base_t"));
        c.types.push(make_range(
            "app",
            "myrange",
            qn("app", "base_t"),
            None,
            None,
        ));
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "myrange")),
                &NodeId::Type(qn("app", "base_t"))
            ),
            "range type must depend on its managed subtype"
        );
    }

    #[test]
    fn range_canonical_fn_adds_function_edge() {
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        let f = make_function("app", "canon_fn");
        let arg_types = f.arg_types_normalized.clone();
        c.functions.push(f);
        c.types.push(make_range(
            "app",
            "myrange",
            qn("pg_catalog", "int4"),
            Some(qn("app", "canon_fn")),
            None,
        ));
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "myrange")),
                &NodeId::Function(qn("app", "canon_fn"), arg_types),
            ),
            "range type must depend on its canonical function"
        );
    }

    #[test]
    fn range_subtype_diff_fn_adds_function_edge() {
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        let f = make_function("app", "diff_fn");
        let arg_types = f.arg_types_normalized.clone();
        c.functions.push(f);
        c.types.push(make_range(
            "app",
            "myrange",
            qn("pg_catalog", "int4"),
            None,
            Some(qn("app", "diff_fn")),
        ));
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "myrange")),
                &NodeId::Function(qn("app", "diff_fn"), arg_types),
            ),
            "range type must depend on its subtype_diff function"
        );
    }

    #[test]
    fn range_unmanaged_canonical_fn_adds_no_edge() {
        // canonical references a function that is NOT in the source catalog.
        // No edge should be added (the lint rule would surface this later).
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.types.push(make_range(
            "app",
            "myrange",
            qn("pg_catalog", "int4"),
            Some(qn("app", "unmanaged_fn")),
            None,
        ));
        let g = build_create_graph(&c);
        // Only the schema edge.
        let deps: Vec<&NodeId> = g
            .dependencies_of(&NodeId::Type(qn("app", "myrange")))
            .collect();
        assert_eq!(deps, vec![&NodeId::Schema(id("app"))]);
    }

    // ── Collation edge tests ─────────────────────────────────────────────────

    use crate::ir::collation::{Collation, CollationProvider};

    fn make_collation(schema: &str, name: &str) -> Collation {
        Collation {
            qname: qn(schema, name),
            provider: CollationProvider::Libc,
            lc_collate: "C".into(),
            lc_ctype: "C".into(),
            deterministic: true,
            version: None,
            owner: None,
            comment: None,
        }
    }

    fn make_table_empty(schema: &str, name: &str) -> Table {
        Table {
            qname: qn(schema, name),
            columns: vec![],
            constraints: vec![],
            partition_by: None,
            partition_of: None,
            comment: None,
            owner: None,
            grants: vec![],
            rls_enabled: false,
            rls_forced: false,
            policies: vec![],
            storage: crate::ir::reloptions::TableStorageOptions::default(),
            access_method: None,
            tablespace: None,
        }
    }

    fn col_text_with_collation(name: &str, collation: QualifiedName) -> Column {
        Column {
            name: id(name),
            ty: ColumnType::Text,
            nullable: true,
            default: None,
            identity: None,
            generated: None,
            collation: Some(collation),
            storage: None,
            compression: None,
            comment: None,
        }
    }

    fn make_domain_with_collation(schema: &str, name: &str, collation: QualifiedName) -> UserType {
        UserType {
            qname: qn(schema, name),
            kind: UserTypeKind::Domain {
                base: ColumnType::Text,
                nullable: true,
                default: None,
                check_constraints: vec![],
                collation: Some(collation),
            },
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    fn make_range_with_collation(
        schema: &str,
        name: &str,
        subtype: QualifiedName,
        collation: QualifiedName,
    ) -> UserType {
        UserType {
            qname: qn(schema, name),
            kind: UserTypeKind::Range {
                subtype,
                subtype_opclass: None,
                collation: Some(collation),
                canonical: None,
                subtype_diff: None,
                multirange_type_name: None,
            },
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    fn make_composite_with_collated_attr(
        schema: &str,
        name: &str,
        collation: QualifiedName,
    ) -> UserType {
        UserType {
            qname: qn(schema, name),
            kind: UserTypeKind::Composite {
                attributes: vec![CompositeAttribute {
                    name: id("val"),
                    ty: ColumnType::Text,
                    collation: Some(collation),
                }],
            },
            comment: None,
            owner: None,
            grants: vec![],
        }
    }

    #[test]
    fn collation_node_registered_with_schema_edge() {
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.collations.push(make_collation("app", "ci"));
        let g = build_create_graph(&c);
        // schema + collation nodes; one structural edge collation → schema.
        assert_eq!(g.node_count(), 2);
        assert!(
            has_edge(
                &g,
                &NodeId::Collation(qn("app", "ci")),
                &NodeId::Schema(id("app")),
            ),
            "collation must depend on its schema (ensures DROP COLLATION before DROP SCHEMA)"
        );
    }

    #[test]
    fn column_with_managed_collation_adds_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.collations.push(make_collation("app", "ci"));
        let mut t = make_table_empty("app", "users");
        t.columns
            .push(col_text_with_collation("email", qn("app", "ci")));
        cat.tables.push(t);
        let g = build_create_graph(&cat);
        assert!(
            has_edge(
                &g,
                &NodeId::Table(qn("app", "users")),
                &NodeId::Collation(qn("app", "ci")),
            ),
            "table must depend on its column's managed collation"
        );
    }

    #[test]
    fn column_with_pg_catalog_collation_no_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        let mut t = make_table_empty("app", "users");
        t.columns
            .push(col_text_with_collation("email", qn("pg_catalog", "C")));
        cat.tables.push(t);
        let g = build_create_graph(&cat);
        assert!(
            !g.edges().any(|(_, to)| matches!(to, NodeId::Collation(_))),
            "no collation edge should be added for pg_catalog.C"
        );
    }

    #[test]
    fn column_with_builtin_shortname_collation_no_edge() {
        // Even with no schema (or any schema), the builtin shortname `POSIX`
        // should be skipped because it's in BUILTIN_COLLATIONS.
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        let mut t = make_table_empty("app", "users");
        t.columns
            .push(col_text_with_collation("email", qn("app", "POSIX")));
        cat.tables.push(t);
        let g = build_create_graph(&cat);
        assert!(!g.edges().any(|(_, to)| matches!(to, NodeId::Collation(_))),);
    }

    #[test]
    fn column_with_unmanaged_collation_no_edge() {
        // Source references a collation that isn't in catalog.collations →
        // no edge (the lint rule surfaces the drift; here we just don't add
        // a phantom edge to a node that doesn't exist).
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        let mut t = make_table_empty("app", "users");
        t.columns
            .push(col_text_with_collation("email", qn("app", "unmanaged")));
        cat.tables.push(t);
        let g = build_create_graph(&cat);
        assert!(!g.edges().any(|(_, to)| matches!(to, NodeId::Collation(_))),);
    }

    #[test]
    fn domain_with_managed_collation_adds_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.collations.push(make_collation("app", "ci"));
        cat.types.push(make_domain_with_collation(
            "app",
            "email_t",
            qn("app", "ci"),
        ));
        let g = build_create_graph(&cat);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "email_t")),
                &NodeId::Collation(qn("app", "ci")),
            ),
            "domain must depend on its managed collation"
        );
    }

    #[test]
    fn domain_with_pg_catalog_collation_no_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.types.push(make_domain_with_collation(
            "app",
            "email_t",
            qn("pg_catalog", "C"),
        ));
        let g = build_create_graph(&cat);
        assert!(!g.edges().any(|(_, to)| matches!(to, NodeId::Collation(_))),);
    }

    #[test]
    fn range_with_managed_collation_adds_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.collations.push(make_collation("app", "ci"));
        cat.types.push(make_range_with_collation(
            "app",
            "textrange",
            qn("pg_catalog", "text"),
            qn("app", "ci"),
        ));
        let g = build_create_graph(&cat);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "textrange")),
                &NodeId::Collation(qn("app", "ci")),
            ),
            "range type must depend on its managed collation"
        );
    }

    #[test]
    fn range_with_pg_catalog_collation_no_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.types.push(make_range_with_collation(
            "app",
            "textrange",
            qn("pg_catalog", "text"),
            qn("pg_catalog", "C"),
        ));
        let g = build_create_graph(&cat);
        assert!(!g.edges().any(|(_, to)| matches!(to, NodeId::Collation(_))),);
    }

    #[test]
    fn composite_attribute_with_managed_collation_adds_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.collations.push(make_collation("app", "ci"));
        cat.types.push(make_composite_with_collated_attr(
            "app",
            "addr_t",
            qn("app", "ci"),
        ));
        let g = build_create_graph(&cat);
        assert!(
            has_edge(
                &g,
                &NodeId::Type(qn("app", "addr_t")),
                &NodeId::Collation(qn("app", "ci")),
            ),
            "composite type must depend on collation of a collated attribute"
        );
    }

    #[test]
    fn composite_attribute_with_pg_catalog_collation_no_edge() {
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.types.push(make_composite_with_collated_attr(
            "app",
            "addr_t",
            qn("pg_catalog", "C"),
        ));
        let g = build_create_graph(&cat);
        assert!(!g.edges().any(|(_, to)| matches!(to, NodeId::Collation(_))),);
    }

    #[test]
    fn collation_ordered_before_table_using_it() {
        // Topological sort must place the collation before the table that
        // references it (since the edge is Table → Collation).
        let mut cat = Catalog::empty();
        cat.schemas.push(crate::ir::schema::Schema::new(id("app")));
        cat.collations.push(make_collation("app", "ci"));
        let mut t = make_table_empty("app", "users");
        t.columns
            .push(col_text_with_collation("email", qn("app", "ci")));
        cat.tables.push(t);
        let g = build_create_graph(&cat);
        let order = g.topological_sort().expect("no cycle");
        let coll_pos = order
            .iter()
            .position(|n| n == &NodeId::Collation(qn("app", "ci")))
            .expect("collation in order");
        let table_pos = order
            .iter()
            .position(|n| n == &NodeId::Table(qn("app", "users")))
            .expect("table in order");
        assert!(
            coll_pos < table_pos,
            "collation must be created before the table that uses it"
        );
    }

    #[test]
    fn type_create_ordering_respects_edges() {
        // derived_t depends on base_t; topological sort must put base_t first.
        let mut c = Catalog::empty();
        c.types.push(make_enum("app", "base_t"));
        c.types.push(make_domain_over(
            "app",
            "derived_t",
            ColumnType::UserDefined(qn("app", "base_t")),
        ));
        let g = build_create_graph(&c);
        let order = g.topological_sort().expect("no cycle expected");
        let base_pos = order
            .iter()
            .position(|n| n == &NodeId::Type(qn("app", "base_t")))
            .expect("base_t in order");
        let derived_pos = order
            .iter()
            .position(|n| n == &NodeId::Type(qn("app", "derived_t")))
            .expect("derived_t in order");
        assert!(base_pos < derived_pos, "base_t must come before derived_t");
    }

    // ── Event trigger edge tests ─────────────────────────────────────────────

    use crate::ir::event_trigger::{EventTrigger, EventTriggerEnabled, EventTriggerEvent};

    fn make_event_trigger(name: &str, function: QualifiedName) -> EventTrigger {
        EventTrigger {
            name: id(name),
            event: EventTriggerEvent::DdlCommandEnd,
            tag_filter: vec![],
            function,
            enabled: EventTriggerEnabled::Enabled,
            owner: None,
            comment: None,
        }
    }

    #[test]
    fn event_trigger_node_registered() {
        let mut c = Catalog::empty();
        let et = make_event_trigger("audit_et", qn("app", "audit_fn"));
        c.event_triggers.push(et);
        let g = build_create_graph(&c);
        assert!(
            g.nodes()
                .any(|n| n == &NodeId::EventTrigger(id("audit_et"))),
            "event trigger node must be registered in the graph"
        );
    }

    #[test]
    fn event_trigger_depends_on_managed_function() {
        let mut c = Catalog::empty();
        let f = make_function("app", "audit_fn");
        let arg_types = f.arg_types_normalized.clone();
        c.functions.push(f);
        let et = make_event_trigger("audit_et", qn("app", "audit_fn"));
        c.event_triggers.push(et);
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::EventTrigger(id("audit_et")),
                &NodeId::Function(qn("app", "audit_fn"), arg_types),
            ),
            "event trigger must depend on the function it executes"
        );
    }

    #[test]
    fn event_trigger_no_edge_for_unmanaged_function() {
        // If the function is not in the catalog, no edge is added.
        let mut c = Catalog::empty();
        let et = make_event_trigger("audit_et", qn("app", "unmanaged_fn"));
        c.event_triggers.push(et);
        let g = build_create_graph(&c);
        // Node exists but has no dependencies.
        assert!(
            g.nodes()
                .any(|n| n == &NodeId::EventTrigger(id("audit_et"))),
            "event trigger node must still be registered even if function is unmanaged"
        );
        assert!(
            g.dependencies_of(&NodeId::EventTrigger(id("audit_et")))
                .next()
                .is_none(),
            "no edge expected for unmanaged function reference"
        );
    }

    // ── Aggregate edge tests ─────────────────────────────────────────────────

    use crate::ir::aggregate::Aggregate;
    use crate::ir::function::{ArgMode, Function, FunctionArg};

    /// Build a managed function with the given positional (IN) argument types.
    fn make_function_with_args(schema: &str, name: &str, arg_types: &[ColumnType]) -> Function {
        let mut f = make_function(schema, name);
        f.args = arg_types
            .iter()
            .map(|ty| FunctionArg {
                name: None,
                mode: ArgMode::In,
                ty: ty.clone(),
                default: None,
            })
            .collect();
        f.arg_types_normalized = NormalizedArgTypes::from_args(&f.args);
        f
    }

    fn make_aggregate(
        schema: &str,
        name: &str,
        arg_types: Vec<ColumnType>,
        state_type: ColumnType,
        sfunc: QualifiedName,
        finalfunc: Option<QualifiedName>,
    ) -> Aggregate {
        Aggregate {
            qname: qn(schema, name),
            arg_types,
            state_type,
            sfunc,
            finalfunc,
            initcond: None,
            owner: None,
            comment: None,
        }
    }

    #[test]
    fn aggregate_node_registered() {
        let mut c = Catalog::empty();
        c.aggregates.push(make_aggregate(
            "app",
            "my_sum",
            vec![ColumnType::Integer],
            ColumnType::BigInt,
            qn("app", "my_sfunc"),
            None,
        ));
        let g = build_create_graph(&c);
        let agg_node = NodeId::Aggregate(
            qn("app", "my_sum"),
            aggregate_arg_types(&[ColumnType::Integer]),
        );
        assert!(
            g.nodes().any(|n| n == &agg_node),
            "aggregate node must be registered in the graph"
        );
    }

    #[test]
    fn aggregate_depends_on_managed_sfunc() {
        let mut c = Catalog::empty();
        // sfunc implied signature: (state_type, arg_types…) = (bigint, integer)
        let sfunc = make_function_with_args(
            "app",
            "my_sfunc",
            &[ColumnType::BigInt, ColumnType::Integer],
        );
        let sfunc_norm = sfunc.arg_types_normalized.clone();
        c.functions.push(sfunc);
        c.aggregates.push(make_aggregate(
            "app",
            "my_sum",
            vec![ColumnType::Integer],
            ColumnType::BigInt,
            qn("app", "my_sfunc"),
            None,
        ));
        let g = build_create_graph(&c);
        let agg_node = NodeId::Aggregate(
            qn("app", "my_sum"),
            aggregate_arg_types(&[ColumnType::Integer]),
        );
        assert!(
            has_edge(
                &g,
                &agg_node,
                &NodeId::Function(qn("app", "my_sfunc"), sfunc_norm),
            ),
            "aggregate must depend on its managed state function"
        );
    }

    #[test]
    fn aggregate_depends_on_managed_finalfunc() {
        let mut c = Catalog::empty();
        let sfunc = make_function_with_args(
            "app",
            "my_sfunc",
            &[ColumnType::BigInt, ColumnType::Integer],
        );
        c.functions.push(sfunc);
        // finalfunc implied signature: (state_type) = (bigint)
        let finalfunc = make_function_with_args("app", "my_final", &[ColumnType::BigInt]);
        let final_norm = finalfunc.arg_types_normalized.clone();
        c.functions.push(finalfunc);
        c.aggregates.push(make_aggregate(
            "app",
            "my_sum",
            vec![ColumnType::Integer],
            ColumnType::BigInt,
            qn("app", "my_sfunc"),
            Some(qn("app", "my_final")),
        ));
        let g = build_create_graph(&c);
        let agg_node = NodeId::Aggregate(
            qn("app", "my_sum"),
            aggregate_arg_types(&[ColumnType::Integer]),
        );
        assert!(
            has_edge(
                &g,
                &agg_node,
                &NodeId::Function(qn("app", "my_final"), final_norm),
            ),
            "aggregate must depend on its managed final function"
        );
    }

    #[test]
    fn aggregate_sfunc_overload_resolved_by_signature() {
        // Two overloads of `agg_sf`; only the (bigint, integer) one matches the
        // aggregate's implied sfunc signature.
        let mut c = Catalog::empty();
        let wrong = make_function_with_args("app", "agg_sf", &[ColumnType::Text, ColumnType::Text]);
        let right =
            make_function_with_args("app", "agg_sf", &[ColumnType::BigInt, ColumnType::Integer]);
        let right_norm = right.arg_types_normalized.clone();
        let wrong_norm = wrong.arg_types_normalized.clone();
        c.functions.push(wrong);
        c.functions.push(right);
        c.aggregates.push(make_aggregate(
            "app",
            "my_sum",
            vec![ColumnType::Integer],
            ColumnType::BigInt,
            qn("app", "agg_sf"),
            None,
        ));
        let g = build_create_graph(&c);
        let agg_node = NodeId::Aggregate(
            qn("app", "my_sum"),
            aggregate_arg_types(&[ColumnType::Integer]),
        );
        assert!(
            has_edge(
                &g,
                &agg_node,
                &NodeId::Function(qn("app", "agg_sf"), right_norm),
            ),
            "edge must point at the (bigint, integer) overload"
        );
        assert!(
            !has_edge(
                &g,
                &agg_node,
                &NodeId::Function(qn("app", "agg_sf"), wrong_norm),
            ),
            "edge must NOT point at the (text, text) overload"
        );
    }

    #[test]
    fn aggregate_no_edge_for_unmanaged_sfunc() {
        let mut c = Catalog::empty();
        c.aggregates.push(make_aggregate(
            "app",
            "my_sum",
            vec![ColumnType::Integer],
            ColumnType::BigInt,
            qn("app", "unmanaged_sfunc"),
            None,
        ));
        let g = build_create_graph(&c);
        let agg_node = NodeId::Aggregate(
            qn("app", "my_sum"),
            aggregate_arg_types(&[ColumnType::Integer]),
        );
        assert!(
            g.nodes().any(|n| n == &agg_node),
            "aggregate node must still be registered even if sfunc is unmanaged"
        );
        assert!(
            g.dependencies_of(&agg_node).next().is_none(),
            "no edge expected for unmanaged sfunc reference"
        );
    }

    // ── Cast edge tests ──────────────────────────────────────────────────────

    use crate::ir::cast::{Cast, CastContext, CastMethod};

    fn make_cast_with_function(
        src_schema: &str,
        src_name: &str,
        tgt_schema: &str,
        tgt_name: &str,
        fn_schema: &str,
        fn_name: &str,
        fn_arg_types: Vec<ColumnType>,
    ) -> Cast {
        Cast {
            source: qn(src_schema, src_name),
            target: qn(tgt_schema, tgt_name),
            method: CastMethod::Function {
                name: qn(fn_schema, fn_name),
                arg_types: fn_arg_types,
            },
            context: CastContext::Explicit,
            comment: None,
        }
    }

    fn make_cast_binary(
        src_schema: &str,
        src_name: &str,
        tgt_schema: &str,
        tgt_name: &str,
    ) -> Cast {
        Cast {
            source: qn(src_schema, src_name),
            target: qn(tgt_schema, tgt_name),
            method: CastMethod::Binary,
            context: CastContext::Implicit,
            comment: None,
        }
    }

    #[test]
    fn cast_node_registered() {
        let mut c = Catalog::empty();
        c.casts
            .push(make_cast_binary("app", "src_t", "app", "tgt_t"));
        let g = build_create_graph(&c);
        let cast_node = NodeId::Cast(qn("app", "src_t"), qn("app", "tgt_t"));
        assert!(
            g.nodes().any(|n| n == &cast_node),
            "cast node must be registered in the graph"
        );
    }

    #[test]
    fn cast_depends_on_managed_function() {
        let mut c = Catalog::empty();
        // The conversion function takes (app.src_t) as a single arg.
        let func = make_function_with_args(
            "app",
            "src_to_tgt",
            &[ColumnType::UserDefined(qn("app", "src_t"))],
        );
        let func_norm = func.arg_types_normalized.clone();
        c.functions.push(func);
        c.casts.push(make_cast_with_function(
            "app",
            "src_t",
            "app",
            "tgt_t",
            "app",
            "src_to_tgt",
            vec![ColumnType::UserDefined(qn("app", "src_t"))],
        ));
        let g = build_create_graph(&c);
        let cast_node = NodeId::Cast(qn("app", "src_t"), qn("app", "tgt_t"));
        assert!(
            has_edge(
                &g,
                &cast_node,
                &NodeId::Function(qn("app", "src_to_tgt"), func_norm),
            ),
            "cast must depend on its managed conversion function"
        );
    }

    #[test]
    fn cast_depends_on_managed_source_and_target_types() {
        let mut c = Catalog::empty();
        c.types.push(make_enum("app", "src_t"));
        c.types.push(make_enum("app", "tgt_t"));
        c.casts
            .push(make_cast_binary("app", "src_t", "app", "tgt_t"));
        let g = build_create_graph(&c);
        let cast_node = NodeId::Cast(qn("app", "src_t"), qn("app", "tgt_t"));
        assert!(
            has_edge(&g, &cast_node, &NodeId::Type(qn("app", "src_t"))),
            "cast must depend on the managed source type"
        );
        assert!(
            has_edge(&g, &cast_node, &NodeId::Type(qn("app", "tgt_t"))),
            "cast must depend on the managed target type"
        );
    }

    #[test]
    fn cast_no_type_edge_for_builtin_target() {
        // Source is managed; target is a pg_catalog built-in — no edge for target.
        let mut c = Catalog::empty();
        c.types.push(make_enum("app", "src_t"));
        // Target pg_catalog.text is not in catalog.types → no type edge.
        let cast = Cast {
            source: qn("app", "src_t"),
            target: qn("pg_catalog", "text"),
            method: CastMethod::Inout,
            context: CastContext::Assignment,
            comment: None,
        };
        c.casts.push(cast);
        let g = build_create_graph(&c);
        let cast_node = NodeId::Cast(qn("app", "src_t"), qn("pg_catalog", "text"));
        // Source type edge present.
        assert!(
            has_edge(&g, &cast_node, &NodeId::Type(qn("app", "src_t"))),
            "cast must depend on the managed source type"
        );
        // No edge for the built-in target.
        assert!(
            !has_edge(&g, &cast_node, &NodeId::Type(qn("pg_catalog", "text"))),
            "no type edge expected for a built-in (pg_catalog) target type"
        );
    }

    #[test]
    fn cast_no_edge_for_unmanaged_function() {
        // Conversion function not in catalog → cast node registered but no
        // function edge (identical behavior to aggregate unmanaged sfunc).
        let mut c = Catalog::empty();
        c.casts.push(make_cast_with_function(
            "app",
            "src_t",
            "pg_catalog",
            "text",
            "app",
            "unmanaged_fn",
            vec![ColumnType::UserDefined(qn("app", "src_t"))],
        ));
        let g = build_create_graph(&c);
        let cast_node = NodeId::Cast(qn("app", "src_t"), qn("pg_catalog", "text"));
        assert!(
            g.nodes().any(|n| n == &cast_node),
            "cast node must be registered even when function is unmanaged"
        );
        // The only edge would be to the function; no types are managed here.
        assert!(
            g.dependencies_of(&cast_node).next().is_none(),
            "no edge expected when conversion function is unmanaged"
        );
    }

    // ── Text-search dictionary / configuration edge tests ────────────────────

    use crate::ir::text_search::{TsConfiguration, TsDictionary, TsMapping};

    fn make_ts_dictionary(schema: &str, name: &str) -> TsDictionary {
        TsDictionary {
            qname: qn(schema, name),
            template: qn("pg_catalog", "snowball"),
            options: vec![],
            owner: None,
            comment: None,
        }
    }

    fn make_ts_configuration(
        schema: &str,
        name: &str,
        mappings: Vec<TsMapping>,
    ) -> TsConfiguration {
        TsConfiguration {
            qname: qn(schema, name),
            parser: qn("pg_catalog", "default"),
            mappings,
            owner: None,
            comment: None,
        }
    }

    #[test]
    fn ts_dictionary_node_registered_with_schema_edge() {
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.ts_dictionaries
            .push(make_ts_dictionary("app", "english_stem"));
        let g = build_create_graph(&c);
        // schema + dictionary nodes; one structural edge dictionary -> schema.
        assert_eq!(g.node_count(), 2);
        assert!(
            has_edge(
                &g,
                &NodeId::TsDictionary(qn("app", "english_stem")),
                &NodeId::Schema(id("app")),
            ),
            "ts dictionary must depend on its schema"
        );
    }

    #[test]
    fn ts_configuration_depends_on_managed_dictionary() {
        // A configuration whose mapping references a managed dictionary gets an
        // edge to that dictionary.
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.ts_dictionaries
            .push(make_ts_dictionary("app", "english_stem"));
        c.ts_configurations.push(make_ts_configuration(
            "app",
            "english_cfg",
            vec![TsMapping {
                token_type: "word".to_string(),
                dictionaries: vec![qn("app", "english_stem")],
            }],
        ));
        let g = build_create_graph(&c);
        assert!(
            has_edge(
                &g,
                &NodeId::TsConfiguration(qn("app", "english_cfg")),
                &NodeId::TsDictionary(qn("app", "english_stem")),
            ),
            "ts configuration must depend on its managed dictionary"
        );
    }

    #[test]
    fn ts_configuration_no_edge_for_builtin_dictionary() {
        // A configuration referencing only a pg_catalog (built-in) dictionary
        // must not add an edge — and must not panic.
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        // No managed dictionaries in catalog.
        c.ts_configurations.push(make_ts_configuration(
            "app",
            "simple_cfg",
            vec![TsMapping {
                token_type: "word".to_string(),
                dictionaries: vec![qn("pg_catalog", "simple")],
            }],
        ));
        let g = build_create_graph(&c);
        let cfg_node = NodeId::TsConfiguration(qn("app", "simple_cfg"));
        assert!(
            g.nodes().any(|n| n == &cfg_node),
            "ts configuration node must be registered even when all dicts are unmanaged"
        );
        // The only dependency should be the schema edge.
        let deps: Vec<&NodeId> = g.dependencies_of(&cfg_node).collect();
        assert_eq!(
            deps,
            vec![&NodeId::Schema(id("app"))],
            "only a schema edge expected; no edge to a built-in dictionary"
        );
    }

    #[test]
    fn ts_configuration_deduplicates_dictionary_edges() {
        // A dictionary referenced by multiple token types must produce only one
        // edge from the configuration to that dictionary.
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.ts_dictionaries
            .push(make_ts_dictionary("app", "english_stem"));
        c.ts_configurations.push(make_ts_configuration(
            "app",
            "english_cfg",
            vec![
                TsMapping {
                    token_type: "word".to_string(),
                    dictionaries: vec![qn("app", "english_stem")],
                },
                TsMapping {
                    token_type: "asciiword".to_string(),
                    dictionaries: vec![qn("app", "english_stem")],
                },
            ],
        ));
        let g = build_create_graph(&c);
        let cfg_node = NodeId::TsConfiguration(qn("app", "english_cfg"));
        let dict_node = NodeId::TsDictionary(qn("app", "english_stem"));
        // Exactly one edge from config to dictionary (deduplicated).
        assert_eq!(
            g.dependencies_of(&cfg_node)
                .filter(|n| *n == &dict_node)
                .count(),
            1,
            "same dictionary referenced by two token types should produce exactly one edge"
        );
    }

    #[test]
    fn ts_configuration_ordered_after_its_dictionary() {
        // Topological sort must place the dictionary before the configuration
        // that references it.
        let mut c = Catalog::empty();
        c.schemas.push(crate::ir::schema::Schema::new(id("app")));
        c.ts_dictionaries
            .push(make_ts_dictionary("app", "english_stem"));
        c.ts_configurations.push(make_ts_configuration(
            "app",
            "english_cfg",
            vec![TsMapping {
                token_type: "word".to_string(),
                dictionaries: vec![qn("app", "english_stem")],
            }],
        ));
        let g = build_create_graph(&c);
        let order = g.topological_sort().expect("no cycle");
        let dict_pos = order
            .iter()
            .position(|n| n == &NodeId::TsDictionary(qn("app", "english_stem")))
            .expect("dictionary in order");
        let cfg_pos = order
            .iter()
            .position(|n| n == &NodeId::TsConfiguration(qn("app", "english_cfg")))
            .expect("configuration in order");
        assert!(
            dict_pos < cfg_pos,
            "dictionary must be created before the configuration that references it"
        );
    }
}