PostgreSQL Data Dictionary & SQL Catalog¶

-- name: by_schema
-- dialect: postgres
WITH target_schema AS (
    SELECT COALESCE(
        :schema_name::text,
        (
            SELECT rn.nspname::text
            FROM pg_catalog.pg_class rc
            JOIN pg_catalog.pg_namespace rn ON rn.oid = rc.relnamespace
            WHERE rc.oid = pg_catalog.to_regclass(pg_catalog.quote_ident(:table_name::text))
        ),
        current_schema()
    ) AS schema_name
),
table_constraints AS (
    SELECT
        con.conrelid,
        unnest(con.conkey) AS attnum,
        bool_or(con.contype = 'p') AS is_primary,
        bool_or(con.contype = 'u') AS is_unique
    FROM pg_catalog.pg_constraint con
    JOIN pg_catalog.pg_class c ON c.oid = con.conrelid
    JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
    JOIN target_schema ts ON ts.schema_name = n.nspname
    WHERE con.contype IN ('p', 'u')
      AND (:table_name::text IS NULL OR c.relname = :table_name)
    GROUP BY con.conrelid, unnest(con.conkey)
)
SELECT
    current_database()::text AS table_catalog,
    n.nspname::text AS table_schema,
    n.nspname::text AS schema_name,
    c.relname::text AS table_name,
    a.attname::text AS column_name,
    a.attnum::integer AS ordinal_position,
    pg_catalog.format_type(a.atttypid, a.atttypmod)::text AS data_type,
    t.typname::text AS type_name,
    CASE WHEN a.attnotnull THEN 'NO' ELSE 'YES' END AS is_nullable,
    CASE WHEN a.attgenerated = '' THEN pg_catalog.pg_get_expr(d.adbin, d.adrelid)::text ELSE NULL END AS column_default,
    CASE
        WHEN a.attidentity = 'a' THEN 'ALWAYS'
        WHEN a.attidentity = 'd' THEN 'BY DEFAULT'
        ELSE NULL
    END AS identity_generation,
    CASE WHEN a.attidentity IN ('a', 'd') THEN 1 ELSE 0 END AS is_identity,
    CASE WHEN a.attgenerated <> '' THEN 1 ELSE 0 END AS is_generated,
    CASE WHEN a.attgenerated <> '' THEN pg_catalog.pg_get_expr(d.adbin, d.adrelid)::text ELSE NULL END AS generation_expression,
    col_description(c.oid, a.attnum)::text AS comment,
    COALESCE(tc.is_primary, false) AS is_primary,
    COALESCE(tc.is_unique, false) AS is_unique,
    pg_catalog.pg_get_serial_sequence(
        pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname), a.attname
    )::text AS sequence_name
FROM pg_catalog.pg_attribute a
JOIN pg_catalog.pg_class c ON c.oid = a.attrelid
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
JOIN target_schema ts ON ts.schema_name = n.nspname
JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
LEFT JOIN pg_catalog.pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum
LEFT JOIN table_constraints tc ON tc.conrelid = c.oid AND tc.attnum = a.attnum
WHERE c.relkind IN ('r', 'p', 'v', 'm', 'f')
  AND (:table_name::text IS NULL OR c.relname = :table_name)
  AND a.attnum > 0
  AND NOT a.attisdropped
ORDER BY n.nspname, c.relname, a.attnum;

-- name: by_table
-- dialect: postgres
WITH target_schema AS (
    SELECT COALESCE(
        :schema_name::text,
        (
            SELECT rn.nspname::text
            FROM pg_catalog.pg_class rc
            JOIN pg_catalog.pg_namespace rn ON rn.oid = rc.relnamespace
            WHERE rc.oid = pg_catalog.to_regclass(pg_catalog.quote_ident(:table_name::text))
        ),
        current_schema()
    ) AS schema_name
),
table_constraints AS (
    SELECT
        con.conrelid,
        unnest(con.conkey) AS attnum,
        bool_or(con.contype = 'p') AS is_primary,
        bool_or(con.contype = 'u') AS is_unique
    FROM pg_catalog.pg_constraint con
    JOIN pg_catalog.pg_class c ON c.oid = con.conrelid
    JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
    JOIN target_schema ts ON ts.schema_name = n.nspname
    WHERE con.contype IN ('p', 'u')
      AND c.relname = :table_name::text
    GROUP BY con.conrelid, unnest(con.conkey)
)
SELECT
    current_database()::text AS table_catalog,
    n.nspname::text AS table_schema,
    n.nspname::text AS schema_name,
    c.relname::text AS table_name,
    a.attname::text AS column_name,
    a.attnum::integer AS ordinal_position,
    pg_catalog.format_type(a.atttypid, a.atttypmod)::text AS data_type,
    t.typname::text AS type_name,
    CASE WHEN a.attnotnull THEN 'NO' ELSE 'YES' END AS is_nullable,
    CASE WHEN a.attgenerated = '' THEN pg_catalog.pg_get_expr(d.adbin, d.adrelid)::text ELSE NULL END AS column_default,
    CASE
        WHEN a.attidentity = 'a' THEN 'ALWAYS'
        WHEN a.attidentity = 'd' THEN 'BY DEFAULT'
        ELSE NULL
    END AS identity_generation,
    CASE WHEN a.attidentity IN ('a', 'd') THEN 1 ELSE 0 END AS is_identity,
    CASE WHEN a.attgenerated <> '' THEN 1 ELSE 0 END AS is_generated,
    CASE WHEN a.attgenerated <> '' THEN pg_catalog.pg_get_expr(d.adbin, d.adrelid)::text ELSE NULL END AS generation_expression,
    col_description(c.oid, a.attnum)::text AS comment,
    COALESCE(tc.is_primary, false) AS is_primary,
    COALESCE(tc.is_unique, false) AS is_unique,
    pg_catalog.pg_get_serial_sequence(
        pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname), a.attname
    )::text AS sequence_name
FROM pg_catalog.pg_attribute a
JOIN pg_catalog.pg_class c ON a.attrelid = c.oid
JOIN pg_catalog.pg_namespace n ON c.relnamespace = n.oid
JOIN target_schema ts ON ts.schema_name = n.nspname
JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
LEFT JOIN pg_catalog.pg_attrdef d ON a.attrelid = d.adrelid AND a.attnum = d.adnum
LEFT JOIN table_constraints tc ON tc.conrelid = c.oid AND tc.attnum = a.attnum
WHERE c.relname = :table_name::text
  AND a.attnum > 0
  AND NOT a.attisdropped
ORDER BY a.attnum;

-- name: by_schema
-- dialect: postgres
SELECT
    n.nspname::text AS schema_name,
    c.relname::text AS object_name,
    'table'::text AS object_type,
    d.description::text AS comment,
    c.oid::text AS object_oid
FROM pg_catalog.pg_description d
JOIN pg_catalog.pg_class c ON c.oid = d.objoid
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = :schema_name
  AND d.objsubid = 0
UNION ALL
SELECT
    n.nspname::text AS schema_name,
    c.relname::text || '.' || a.attname::text AS object_name,
    'column'::text AS object_type,
    d.description::text AS comment,
    c.oid::text AS object_oid
FROM pg_catalog.pg_description d
JOIN pg_catalog.pg_class c ON c.oid = d.objoid
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid AND a.attnum = d.objsubid
WHERE n.nspname = :schema_name
  AND d.objsubid > 0
ORDER BY schema_name, object_type, object_name;

-- name: by_schema
-- dialect: postgres
SELECT
    current_database()::text AS constraint_catalog,
    n.nspname::text AS constraint_schema,
    n.nspname::text AS schema_name,
    c.relname::text AS table_name,
    con.conname::text AS constraint_name,
    CASE con.contype
        WHEN 'c' THEN 'CHECK'
        WHEN 'f' THEN 'FOREIGN KEY'
        WHEN 'p' THEN 'PRIMARY KEY'
        WHEN 'u' THEN 'UNIQUE'
        WHEN 'x' THEN 'EXCLUDE'
        ELSE con.contype::text
    END AS constraint_type,
    pg_catalog.pg_get_constraintdef(con.oid, true)::text AS definition,
    ARRAY(
        SELECT att.attname::text
        FROM unnest(con.conkey) WITH ORDINALITY AS keys(attnum, ord)
        JOIN pg_catalog.pg_attribute att ON att.attrelid = con.conrelid AND att.attnum = keys.attnum
        ORDER BY keys.ord
    )::text[] AS columns,
    ref_ns.nspname::text AS referenced_schema,
    ref.relname::text AS referenced_table,
    ARRAY(
        SELECT att.attname::text
        FROM unnest(con.confkey) WITH ORDINALITY AS keys(attnum, ord)
        JOIN pg_catalog.pg_attribute att ON att.attrelid = con.confrelid AND att.attnum = keys.attnum
        ORDER BY keys.ord
    )::text[] AS referenced_columns,
    con.condeferrable AS is_deferrable,
    con.condeferred AS initially_deferred,
    dep.refobjid IS NOT NULL AS is_extension_owned
FROM pg_catalog.pg_constraint con
JOIN pg_catalog.pg_namespace n ON n.oid = con.connamespace
LEFT JOIN pg_catalog.pg_class c ON c.oid = con.conrelid
LEFT JOIN pg_catalog.pg_class ref ON ref.oid = con.confrelid
LEFT JOIN pg_catalog.pg_namespace ref_ns ON ref_ns.oid = ref.relnamespace
LEFT JOIN pg_catalog.pg_depend dep ON dep.objid = con.oid AND dep.deptype = 'e'
WHERE n.nspname = :schema_name
  AND (:table_name::text IS NULL OR c.relname = :table_name)
ORDER BY n.nspname, c.relname, con.conname;

-- name: by_object
-- dialect: postgres
WITH extension_deps AS (
    SELECT DISTINCT
        classid,
        objid
    FROM pg_catalog.pg_depend
    WHERE deptype = 'e'
)
SELECT
    n.nspname::text AS schema_name,
    c.relname::text AS object_name,
    CASE c.relkind
        WHEN 'v' THEN pg_catalog.pg_get_viewdef(c.oid, true)::text
        WHEN 'm' THEN pg_catalog.pg_get_viewdef(c.oid, true)::text
        WHEN 'S' THEN 'CREATE SEQUENCE ' || pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname)
        ELSE NULL::text
    END AS ddl,
    CASE c.relkind
        WHEN 'v' THEN 'native'
        WHEN 'm' THEN 'native'
        WHEN 'S' THEN 'generated'
        ELSE 'unsupported'
    END AS fidelity,
    dep.objid IS NOT NULL AS is_extension_owned
FROM pg_catalog.pg_class c
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
LEFT JOIN extension_deps dep
    ON dep.classid = 'pg_catalog.pg_class'::regclass
   AND dep.objid = c.oid
WHERE n.nspname = :schema_name
  AND c.relname = :object_name
  AND (:object_type::text IS NULL OR c.relkind = CASE :object_type
      WHEN 'view' THEN 'v'
      WHEN 'materialized_view' THEN 'm'
      WHEN 'sequence' THEN 'S'
      WHEN 'table' THEN 'r'
      ELSE c.relkind
  END)
UNION ALL
SELECT
    n.nspname::text AS schema_name,
    p.proname::text AS object_name,
    pg_catalog.pg_get_functiondef(p.oid)::text AS ddl,
    'native'::text AS fidelity,
    dep.objid IS NOT NULL AS is_extension_owned
FROM pg_catalog.pg_proc p
JOIN pg_catalog.pg_namespace n ON n.oid = p.pronamespace
LEFT JOIN extension_deps dep
    ON dep.classid = 'pg_catalog.pg_proc'::regclass
   AND dep.objid = p.oid
WHERE n.nspname = :schema_name
  AND p.proname = :object_name
  AND (:object_type::text IS NULL OR :object_type IN ('routine', 'function', 'procedure'))
ORDER BY schema_name, object_name;