pgmt 0.6.4

//! The identity snapshot query, composed from one branch definition per object
//! kind.
//!
//! `CatalogIdentity` needs the identity of every object in the database and
//! nothing else, in one round trip — the converters' N fetches would cost far
//! more than the answer is worth. So the snapshot is a single `UNION ALL`, one
//! branch per object kind, and each branch is built here out of the SQL
//! spellings in [`super::exclusion::sql`]: the branch selects the columns a
//! [`crate::catalog::id::DbObjectId`] is built from, and filters with the same
//! rules the matching converter applies in Rust.
//!
//! The invariant is set equality: for any database, the identities this query
//! reports and the identities a full catalog load yields are the same set
//! (`tests/catalog/identity_consistency.rs`). A branch whose selection drifts
//! from its converter's is a drift in that test.
//!
//! Three asymmetries are deliberate and not drift:
//!
//! - **Grants** have no branch. They are state attached to an object, not
//!   objects with identities of their own.
//! - **`public`** is never reported. Every database has it from initdb onward,
//!   so no schema file creates it, and attributing it to one would make the
//!   first file that touches the database appear to own it.
//! - **Materialized views** are reported by neither the snapshot nor the
//!   catalog. An identity no fetcher can yield would be attributed to a file and
//!   then never found in the catalog it is diffed against.

use super::exclusion::sql;
use anyhow::Result;
use sqlx::PgPool;

/// One row of the snapshot: a kind tag, the coordinates a
/// [`crate::catalog::id::DbObjectId`] of that kind is built from, and the OID
/// the object was allocated. Every coordinate is text, and the ones a kind does
/// not have are null.
///
/// Turning a row back into an identity is [`crate::catalog::identity`]'s job —
/// the OID stops here, as it does for every other raw fetch.
#[derive(sqlx::FromRow)]
pub struct Row {
    #[sqlx(rename = "type")]
    pub kind: String,
    pub schema: Option<String>,
    pub name: String,
    #[sqlx(rename = "tbl")]
    pub table: Option<String>,
    pub args: Option<String>,
    pub oid: i64,
}

/// Every object identity in the database, in one round trip.
pub async fn fetch(pool: &PgPool) -> Result<Vec<Row>> {
    // The query is composed entirely from the branch definitions below —
    // catalog table names and predicates written in this repository, with no
    // value from the database or the user reaching it.
    Ok(sqlx::query_as(sqlx::AssertSqlSafe(query()))
        .fetch_all(pool)
        .await?)
}

/// The current OID boundary mark (see [`marks_query`]).
pub async fn current_mark(pool: &PgPool) -> Result<i64> {
    let mark: Option<i64> = sqlx::query_scalar(sqlx::AssertSqlSafe(marks_query()))
        .fetch_one(pool)
        .await?;
    // Every catalog empty is not a state a live database can be in; treat the
    // absence as "nothing allocated yet" rather than failing.
    Ok(mark.unwrap_or(0))
}

/// The columns every branch projects, in order. They are what a
/// [`crate::catalog::id::DbObjectId`] is built from: the kind tag, then the
/// object's schema, name, parent table and argument string, each `NULL` for a
/// kind that has no such coordinate.
const COLUMNS: [&str; 5] = ["type", "schema", "name", "tbl", "args"];

/// One object kind's branch of the snapshot query.
pub struct Branch {
    kind: &'static str,
    from: String,
    oid_catalog: &'static str,
    oid: String,
    schema: String,
    name: String,
    table: String,
    args: String,
    predicates: Vec<String>,
}

impl Branch {
    /// A branch for `kind`, reading the catalog tables in `from` (a `FROM` body,
    /// joins included). Every identity column defaults to `NULL`.
    ///
    /// `oid_catalog` is the catalog table this kind's own OID is allocated
    /// from, and `oid` the expression selecting it. They are one declaration
    /// because the boundary marks ([`marks_query`]) and the snapshot's `oid`
    /// column must range over exactly the same catalogs: a mark that ignores a
    /// catalog some branch reports from would place that kind's objects beyond
    /// the last boundary and attribute them to no file at all.
    fn new(kind: &'static str, from: &str, oid_catalog: &'static str, oid: &str) -> Self {
        Self {
            kind,
            from: from.to_string(),
            oid_catalog,
            oid: oid.to_string(),
            schema: "NULL".to_string(),
            name: "NULL".to_string(),
            table: "NULL".to_string(),
            args: "NULL".to_string(),
            predicates: Vec::new(),
        }
    }

    fn schema(mut self, expr: &str) -> Self {
        self.schema = expr.to_string();
        self
    }

    fn name(mut self, expr: &str) -> Self {
        self.name = expr.to_string();
        self
    }

    fn table(mut self, expr: &str) -> Self {
        self.table = expr.to_string();
        self
    }

    fn args(mut self, expr: &str) -> Self {
        self.args = expr.to_string();
        self
    }

    /// Add a predicate. All predicates of a branch are `AND`ed; each is
    /// parenthesised, so a predicate may itself be a disjunction.
    fn filter(mut self, predicate: impl Into<String>) -> Self {
        self.predicates.push(predicate.into());
        self
    }

    /// The branch as a `SELECT`. Every projected column is cast to `text`: the
    /// catalog columns are a mix of `name`, `text` and untyped `NULL`, and the
    /// union's column types must not depend on which branch happens to come
    /// first.
    fn sql(&self) -> String {
        let mut projections: Vec<String> = [
            format!("'{}'", self.kind),
            self.schema.clone(),
            self.name.clone(),
            self.table.clone(),
            self.args.clone(),
        ]
        .iter()
        .zip(COLUMNS)
        .map(|(expr, column)| format!("    ({expr})::text AS \"{column}\""))
        .collect();
        // Widened to int8: OIDs are unsigned 32-bit, so the whole range is
        // representable and comparisons keep their order.
        projections.push(format!("    ({})::int8 AS \"oid\"", self.oid));

        let mut sql = format!("SELECT\n{}\nFROM {}", projections.join(",\n"), self.from);
        if !self.predicates.is_empty() {
            let conditions: Vec<String> = self
                .predicates
                .iter()
                .map(|predicate| format!("({predicate})"))
                .collect();
            sql.push_str(&format!("\nWHERE {}", conditions.join("\n  AND ")));
        }
        sql
    }
}

/// The full snapshot query: every branch, `UNION ALL`ed into one round trip.
pub fn query() -> String {
    branches()
        .iter()
        .map(Branch::sql)
        .collect::<Vec<_>>()
        .join("\n\nUNION ALL\n\n")
}

/// The current OID boundary mark: the highest OID allocated in any catalog a
/// branch reports from.
///
/// Taken between schema files, consecutive marks bracket the objects a file
/// created — `(previous, current]` — which is what attributes objects to files
/// without re-reading the whole catalog after every one. Each `max(oid)` is a
/// backward scan of that catalog's OID index, so the whole query is a handful
/// of index lookups regardless of how large the schema has grown.
///
/// Reading it as a mark relies on OIDs being handed out in increasing order,
/// which holds for a database that has not wrapped the 32-bit counter. The
/// callers run against a shadow branch created moments earlier, so the counter
/// they observe is the source database's, untouched by pgmt.
pub fn marks_query() -> String {
    let mut catalogs: Vec<&'static str> = branches().iter().map(|b| b.oid_catalog).collect();
    catalogs.sort_unstable();
    catalogs.dedup();

    let maxima: Vec<String> = catalogs
        .iter()
        .map(|catalog| format!("    SELECT max(oid) AS m FROM {catalog}"))
        .collect();
    format!(
        "SELECT max(m)::int8 AS mark FROM (\n{}\n) s",
        maxima.join("\n    UNION ALL\n")
    )
}

/// One branch per object kind the catalog models, each mirroring the converter
/// of the same name in this module's siblings.
pub fn branches() -> Vec<Branch> {
    vec![
        // raw::schema — the namespace map minus PostgreSQL's own namespaces.
        // `public` is dropped on top of that, for the snapshot only.
        Branch::new("schema", "pg_namespace n", "pg_namespace", "n.oid")
            .name("n.nspname")
            .filter(sql::not_a_system_namespace("n.nspname"))
            .filter("n.nspname <> 'public'"),
        // raw::table
        Branch::new(
            "table",
            "pg_class c\n     JOIN pg_namespace n ON c.relnamespace = n.oid",
            "pg_class",
            "c.oid",
        )
        .schema("n.nspname")
        .name("c.relname")
        .filter("c.relkind = 'r'")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_class", "c.oid")),
        // raw::view
        Branch::new(
            "view",
            "pg_class c\n     JOIN pg_namespace n ON c.relnamespace = n.oid",
            "pg_class",
            "c.oid",
        )
        .schema("n.nspname")
        .name("c.relname")
        .filter("c.relkind = 'v'")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_class", "c.oid")),
        // raw::sequence
        Branch::new(
            "sequence",
            "pg_class c\n     JOIN pg_namespace n ON c.relnamespace = n.oid",
            "pg_class",
            "c.oid",
        )
        .schema("n.nspname")
        .name("c.relname")
        .filter("c.relkind = 'S'")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_class", "c.oid"))
        .filter(sql::not_an_identity_sequence("c.oid")),
        // raw::index. The enumeration is `pg_index`, as the raw fetch's is: it
        // is the set of indexes whatever the index relation's own relkind says.
        // An index belongs to an extension through its own OID (a standalone
        // extension index) or through the table it is on (an index an extension
        // script created records membership only on the parent), and either
        // schema being a system one puts it out of scope.
        Branch::new(
            "index",
            "pg_index idx\n     \
             JOIN pg_class i ON idx.indexrelid = i.oid\n     \
             JOIN pg_namespace n ON i.relnamespace = n.oid\n     \
             JOIN pg_class t ON idx.indrelid = t.oid\n     \
             JOIN pg_namespace tn ON t.relnamespace = tn.oid",
            "pg_class",
            "i.oid",
        )
        .schema("n.nspname")
        .name("i.relname")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_a_system_namespace("tn.nspname"))
        .filter(sql::not_extension_owned("pg_class", "i.oid"))
        .filter(sql::parent_relation_not_extension_owned("t.oid"))
        .filter(sql::not_a_constraint_backing_index("i.oid")),
        // raw::function, whose converter calls everything that is not a
        // procedure a function — a window function included.
        Branch::new(
            "function",
            "pg_proc p\n     JOIN pg_namespace n ON p.pronamespace = n.oid",
            "pg_proc",
            "p.oid",
        )
        .schema("n.nspname")
        .name("p.proname")
        .args("pg_catalog.pg_get_function_identity_arguments(p.oid)")
        .filter("p.prokind NOT IN ('a', 'p')")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_proc", "p.oid")),
        // raw::function, procedure half.
        Branch::new(
            "procedure",
            "pg_proc p\n     JOIN pg_namespace n ON p.pronamespace = n.oid",
            "pg_proc",
            "p.oid",
        )
        .schema("n.nspname")
        .name("p.proname")
        .args("pg_catalog.pg_get_function_identity_arguments(p.oid)")
        .filter("p.prokind = 'p'")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_proc", "p.oid")),
        // raw::aggregate, which enumerates `pg_aggregate` joined to its
        // `pg_proc` row — exactly the routines of `prokind = 'a'`.
        Branch::new(
            "aggregate",
            "pg_proc p\n     JOIN pg_namespace n ON p.pronamespace = n.oid",
            "pg_proc",
            "p.oid",
        )
        .schema("n.nspname")
        .name("p.proname")
        .args("pg_catalog.pg_get_function_identity_arguments(p.oid)")
        .filter("p.prokind = 'a'")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_proc", "p.oid")),
        // raw::custom_type — enums, composites and ranges. The row-type test
        // mirrors the one in that fetch's own query: a relation's row type is
        // the relation, not a type a schema file wrote, and only a standalone
        // composite's own backing entry (relkind 'c') is not such a relation.
        Branch::new(
            "type",
            "pg_type t\n     JOIN pg_namespace n ON t.typnamespace = n.oid",
            "pg_type",
            "t.oid",
        )
        .schema("n.nspname")
        .name("t.typname")
        .filter("t.typtype IN ('e', 'c', 'r')")
        .filter(
            "NOT EXISTS (\n    SELECT 1 FROM pg_class c\n    \
             WHERE c.reltype = t.oid\n      \
             AND c.relkind != 'c'\n)",
        )
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_type", "t.oid")),
        // raw::domain
        Branch::new(
            "domain",
            "pg_type t\n     JOIN pg_namespace n ON t.typnamespace = n.oid",
            "pg_type",
            "t.oid",
        )
        .schema("n.nspname")
        .name("t.typname")
        .filter("t.typtype = 'd'")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_type", "t.oid")),
        // raw::collation. `pg_catalog` holds the built-in collations and one per
        // libc locale found at initdb, so the system-namespace filter is what
        // separates a user collation from the hundreds the server ships.
        Branch::new(
            "collation",
            "pg_collation c\n     JOIN pg_namespace n ON c.collnamespace = n.oid",
            "pg_collation",
            "c.oid",
        )
        .schema("n.nspname")
        .name("c.collname")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_collation", "c.oid")),
        // raw::constraint. Primary keys are carried by their table, so they are
        // not constraints of their own here.
        Branch::new(
            "constraint",
            "pg_constraint co\n     \
             JOIN pg_class cl ON co.conrelid = cl.oid\n     \
             JOIN pg_namespace n ON cl.relnamespace = n.oid",
            "pg_constraint",
            "co.oid",
        )
        .schema("n.nspname")
        .name("co.conname")
        .table("cl.relname")
        .filter("cl.relkind = 'r'")
        .filter("co.contype IN ('u', 'f', 'c', 'x')")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::parent_relation_not_extension_owned("cl.oid")),
        // raw::trigger
        Branch::new(
            "trigger",
            "pg_trigger tg\n     \
             JOIN pg_class c ON tg.tgrelid = c.oid\n     \
             JOIN pg_namespace n ON c.relnamespace = n.oid",
            "pg_trigger",
            "tg.oid",
        )
        .schema("n.nspname")
        .name("tg.tgname")
        .table("c.relname")
        .filter("c.relkind IN ('r', 'v', 'm')")
        .filter("NOT tg.tgisinternal")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::parent_relation_not_extension_owned("c.oid")),
        // raw::policy
        Branch::new(
            "policy",
            "pg_policy pol\n     \
             JOIN pg_class c ON pol.polrelid = c.oid\n     \
             JOIN pg_namespace n ON c.relnamespace = n.oid",
            "pg_policy",
            "pol.oid",
        )
        .schema("n.nspname")
        .name("pol.polname")
        .table("c.relname")
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::parent_relation_not_extension_owned("c.oid")),
        // raw::operator. The identity args are the canonical "left, right"
        // operand string `DROP`/`COMMENT ON OPERATOR` require, with NONE for an
        // absent operand.
        Branch::new(
            "operator",
            "pg_operator o\n     JOIN pg_namespace n ON o.oprnamespace = n.oid",
            "pg_operator",
            "o.oid",
        )
        .schema("n.nspname")
        .name("o.oprname")
        .args(
            "CASE WHEN o.oprleft = 0 THEN 'NONE' \
             ELSE pg_catalog.format_type(o.oprleft, NULL) END\n       \
             || ', '\n       \
             || CASE WHEN o.oprright = 0 THEN 'NONE' \
             ELSE pg_catalog.format_type(o.oprright, NULL) END",
        )
        .filter(sql::not_a_system_namespace("n.nspname"))
        .filter(sql::not_extension_owned("pg_operator", "o.oid")),
        // raw::cast. A cast is not schema-scoped: its identity is the (source,
        // target) type pair, carried in the "name" and "tbl" columns. Creating
        // one requires owning the source or the target type, so every user cast
        // has at least one side outside the system schemas; each side is
        // resolved through the array indirection first, as the converter's
        // `resolve_type` does.
        Branch::new(
            "cast",
            "pg_cast ca\n     \
             JOIN pg_type st ON ca.castsource = st.oid\n     \
             JOIN pg_type ste ON ste.oid = COALESCE(NULLIF(st.typelem, 0), st.oid)\n     \
             JOIN pg_namespace stn ON ste.typnamespace = stn.oid\n     \
             JOIN pg_type tt ON ca.casttarget = tt.oid\n     \
             JOIN pg_type tte ON tte.oid = COALESCE(NULLIF(tt.typelem, 0), tt.oid)\n     \
             JOIN pg_namespace ttn ON tte.typnamespace = ttn.oid",
            "pg_cast",
            "ca.oid",
        )
        .name("pg_catalog.format_type(ca.castsource, NULL)")
        .table("pg_catalog.format_type(ca.casttarget, NULL)")
        .filter(format!(
            "{} OR {}",
            sql::not_a_system_namespace("stn.nspname"),
            sql::not_a_system_namespace("ttn.nspname")
        ))
        .filter(sql::not_extension_owned("pg_cast", "ca.oid")),
        // raw::extension
        Branch::new("extension", "pg_extension e", "pg_extension", "e.oid")
            .name("e.extname")
            .filter(sql::not_a_built_in_extension("e.extname")),
    ]
}

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

    #[test]
    fn test_every_branch_projects_the_identity_columns() {
        for branch in branches() {
            let sql = branch.sql();
            for column in COLUMNS {
                assert!(
                    sql.contains(&format!("AS \"{column}\"")),
                    "branch {} does not project {column}:\n{sql}",
                    branch.kind
                );
            }
        }
    }

    #[test]
    fn test_every_branch_projects_its_oid() {
        for branch in branches() {
            assert!(
                branch.sql().contains("AS \"oid\""),
                "branch {} does not project its oid",
                branch.kind
            );
        }
    }

    /// A branch's declared OID catalog must be the one its OID expression
    /// actually reads from.
    ///
    /// The boundary marks are derived from those declarations, so a branch that
    /// names the wrong catalog leaves its kind's OIDs outside every mark: the
    /// file that created such an object closes its boundary beneath it and the
    /// object is attributed to whichever file comes next — or to none, if it was
    /// the last one. Whether an end-to-end test notices depends on which kind a
    /// file happens to create last, so the binding is checked here directly by
    /// resolving the expression's alias against the branch's own FROM body.
    #[test]
    fn test_declared_oid_catalog_matches_the_expression() {
        /// Every `<catalog> <alias>` binding in a FROM body, joins included.
        fn aliases(from: &str) -> std::collections::BTreeMap<&str, &str> {
            let tokens: Vec<&str> = from.split_whitespace().collect();
            let mut bound = std::collections::BTreeMap::new();
            if tokens.len() >= 2 {
                bound.insert(tokens[1], tokens[0]);
            }
            for (i, token) in tokens.iter().enumerate() {
                if token.eq_ignore_ascii_case("JOIN") && i + 2 < tokens.len() {
                    bound.insert(tokens[i + 2], tokens[i + 1]);
                }
            }
            bound
        }

        for branch in branches() {
            let alias = branch
                .oid
                .split_once('.')
                .unwrap_or_else(|| panic!("branch {}: oid is not <alias>.oid", branch.kind))
                .0;
            let bound = aliases(&branch.from);
            let reads = bound.get(alias).unwrap_or_else(|| {
                panic!(
                    "branch {}: oid expression {} names alias {alias}, which its FROM does not bind",
                    branch.kind, branch.oid
                )
            });
            assert_eq!(
                *reads, branch.oid_catalog,
                "branch {} selects {} (from {reads}) but declares {}",
                branch.kind, branch.oid, branch.oid_catalog
            );
        }
    }

    #[test]
    fn test_query_unions_every_branch() {
        let query = query();
        assert_eq!(query.matches("UNION ALL").count(), branches().len() - 1);
        for branch in branches() {
            assert!(
                query.contains(&format!("('{}')::text", branch.kind)),
                "branch {} is missing from the query",
                branch.kind
            );
        }
    }

    #[test]
    fn test_exclusion_predicates_are_class_qualified() {
        // An OID identifies a row within one catalog table, so every
        // extension-ownership test names the catalog its OID belongs to.
        let query = query();
        for fragment in query.split("dep.objid = ").skip(1) {
            assert!(
                fragment.contains("dep.classid = '"),
                "an extension-ownership test is not class-qualified:\n{fragment}"
            );
        }
    }
}