use super::ddl::{
CheckConstraint, Column, Enum, ForeignKey, Index, IndexColumn, Policy, PostgresEntity,
PrimaryKey, Role, Schema, Sequence, Table, UniqueConstraint, View,
};
use super::grammar::{is_system_namespace, is_system_role};
use super::snapshot::PostgresSnapshot;
#[derive(Debug, Clone)]
pub struct IntrospectError {
pub message: String,
pub table: Option<String>,
pub schema: Option<String>,
}
impl std::fmt::Display for IntrospectError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match (&self.schema, &self.table) {
(Some(s), Some(t)) => {
write!(f, "Introspection error for '{}.{}': {}", s, t, self.message)
}
(Some(s), None) => write!(f, "Introspection error in schema '{}': {}", s, self.message),
(None, Some(t)) => write!(f, "Introspection error for '{}': {}", t, self.message),
(None, None) => write!(f, "Introspection error: {}", self.message),
}
}
}
impl std::error::Error for IntrospectError {}
pub type IntrospectResult<T> = Result<T, IntrospectError>;
#[derive(Debug, Clone)]
pub struct RawTableInfo {
pub schema: String,
pub name: String,
pub is_rls_enabled: bool,
pub is_unlogged: bool,
pub is_temporary: bool,
pub tablespace: Option<String>,
pub comment: Option<String>,
}
#[derive(Debug, Clone)]
pub struct RawColumnInfo {
pub schema: String,
pub table: String,
pub name: String,
pub column_type: String,
pub type_schema: Option<String>,
pub not_null: bool,
pub default_value: Option<String>,
pub is_identity: bool,
pub identity_type: Option<String>,
pub is_generated: bool,
pub generated_expression: Option<String>,
pub generated_stored: bool,
pub dimensions: Option<i32>,
pub comment: Option<String>,
pub ordinal_position: i32,
}
#[derive(Debug, Clone)]
pub struct RawEnumInfo {
pub schema: String,
pub name: String,
pub values: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct RawSequenceInfo {
pub schema: String,
pub name: String,
pub data_type: Option<String>,
pub start_value: Option<String>,
pub min_value: Option<String>,
pub max_value: Option<String>,
pub increment: Option<String>,
pub cycle: Option<bool>,
pub cache_value: Option<String>,
pub owned_by: Option<String>,
}
#[derive(Debug, Clone)]
pub struct RawIndexInfo {
pub schema: String,
pub table: String,
pub name: String,
pub is_unique: bool,
pub is_primary: bool,
pub method: String,
pub columns: Vec<RawIndexColumnInfo>,
pub where_clause: Option<String>,
pub concurrent: bool,
}
#[derive(Debug, Clone)]
pub struct RawIndexColumnInfo {
pub name: String,
pub is_expression: bool,
pub asc: bool,
pub nulls_first: bool,
pub opclass: Option<String>,
}
#[derive(Debug, Clone)]
pub struct RawForeignKeyInfo {
pub schema: String,
pub table: String,
pub name: String,
pub columns: Vec<String>,
pub schema_to: String,
pub table_to: String,
pub columns_to: Vec<String>,
pub on_update: String,
pub on_delete: String,
pub deferrable: bool,
pub initially_deferred: bool,
}
#[derive(Debug, Clone)]
pub struct RawPrimaryKeyInfo {
pub schema: String,
pub table: String,
pub name: String,
pub columns: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct RawUniqueInfo {
pub schema: String,
pub table: String,
pub name: String,
pub columns: Vec<String>,
pub nulls_not_distinct: bool,
pub deferrable: bool,
pub initially_deferred: bool,
}
#[derive(Debug, Clone)]
pub struct RawCheckInfo {
pub schema: String,
pub table: String,
pub name: String,
pub expression: String,
}
#[derive(Debug, Clone)]
pub struct RawViewInfo {
pub schema: String,
pub name: String,
pub definition: String,
pub is_materialized: bool,
}
#[derive(Debug, Clone)]
pub struct RawPolicyInfo {
pub schema: String,
pub table: String,
pub name: String,
pub as_clause: String,
pub for_clause: String,
pub to: Vec<String>,
pub using: Option<String>,
pub with_check: Option<String>,
}
#[derive(Debug, Clone)]
pub struct RawRoleInfo {
pub name: String,
pub create_db: bool,
pub create_role: bool,
pub inherit: bool,
}
#[derive(Debug, Clone, Default)]
pub struct RawIntrospection {
pub schemas: Vec<Schema>,
pub tables: Vec<RawTableInfo>,
pub columns: Vec<RawColumnInfo>,
pub enums: Vec<RawEnumInfo>,
pub sequences: Vec<RawSequenceInfo>,
pub views: Vec<RawViewInfo>,
pub indexes: Vec<RawIndexInfo>,
pub foreign_keys: Vec<RawForeignKeyInfo>,
pub primary_keys: Vec<RawPrimaryKeyInfo>,
pub unique_constraints: Vec<RawUniqueInfo>,
pub check_constraints: Vec<RawCheckInfo>,
pub roles: Vec<RawRoleInfo>,
pub policies: Vec<RawPolicyInfo>,
}
#[must_use]
pub fn assemble_ddl(raw: RawIntrospection) -> super::PostgresDDL {
let mut ddl = super::PostgresDDL::new();
for schema in raw.schemas {
ddl.schemas.push(schema);
}
for value in process_enums(&raw.enums) {
ddl.enums.push(value);
}
for sequence in process_sequences(&raw.sequences) {
ddl.sequences.push(sequence);
}
for role in process_roles(&raw.roles) {
ddl.roles.push(role);
}
for policy in process_policies(&raw.policies) {
ddl.policies.push(policy);
}
for table in process_tables(&raw.tables) {
ddl.tables.push(table);
}
for column in process_columns(&raw.columns) {
ddl.columns.push(column);
}
for index in process_indexes(&raw.indexes) {
ddl.indexes.push(index);
}
for foreign_key in process_foreign_keys(&raw.foreign_keys) {
ddl.fks.push(foreign_key);
}
for primary_key in process_primary_keys(&raw.primary_keys) {
ddl.pks.push(primary_key);
}
for unique in process_unique_constraints(&raw.unique_constraints) {
ddl.uniques.push(unique);
}
for check in process_check_constraints(&raw.check_constraints) {
ddl.checks.push(check);
}
for view in process_views(&raw.views) {
ddl.views.push(view);
}
ddl
}
#[derive(Debug, Clone, Default)]
pub struct IntrospectionResult {
pub schemas: Vec<Schema>,
pub enums: Vec<Enum>,
pub sequences: Vec<Sequence>,
pub roles: Vec<Role>,
pub tables: Vec<Table>,
pub columns: Vec<Column>,
pub indexes: Vec<Index>,
pub foreign_keys: Vec<ForeignKey>,
pub primary_keys: Vec<PrimaryKey>,
pub unique_constraints: Vec<UniqueConstraint>,
pub check_constraints: Vec<CheckConstraint>,
pub views: Vec<View>,
pub policies: Vec<Policy>,
pub errors: Vec<IntrospectError>,
}
impl IntrospectionResult {
#[must_use]
pub fn to_snapshot(&self) -> PostgresSnapshot {
let mut snapshot = PostgresSnapshot::new();
for schema in &self.schemas {
snapshot.add_entity(PostgresEntity::Schema(schema.clone()));
}
for e in &self.enums {
snapshot.add_entity(PostgresEntity::Enum(e.clone()));
}
for seq in &self.sequences {
snapshot.add_entity(PostgresEntity::Sequence(seq.clone()));
}
for role in &self.roles {
snapshot.add_entity(PostgresEntity::Role(role.clone()));
}
for table in &self.tables {
snapshot.add_entity(PostgresEntity::Table(table.clone()));
}
for column in &self.columns {
snapshot.add_entity(PostgresEntity::Column(column.clone()));
}
for index in &self.indexes {
snapshot.add_entity(PostgresEntity::Index(index.clone()));
}
for fk in &self.foreign_keys {
snapshot.add_entity(PostgresEntity::ForeignKey(fk.clone()));
}
for pk in &self.primary_keys {
snapshot.add_entity(PostgresEntity::PrimaryKey(pk.clone()));
}
for unique in &self.unique_constraints {
snapshot.add_entity(PostgresEntity::UniqueConstraint(unique.clone()));
}
for check in &self.check_constraints {
snapshot.add_entity(PostgresEntity::CheckConstraint(check.clone()));
}
for view in &self.views {
snapshot.add_entity(PostgresEntity::View(view.clone()));
}
for policy in &self.policies {
snapshot.add_entity(PostgresEntity::Policy(policy.clone()));
}
snapshot
}
#[must_use]
pub const fn has_errors(&self) -> bool {
!self.errors.is_empty()
}
#[must_use]
pub fn to_entities(&self) -> Vec<PostgresEntity> {
let mut entities = Vec::new();
for s in &self.schemas {
entities.push(PostgresEntity::Schema(s.clone()));
}
for e in &self.enums {
entities.push(PostgresEntity::Enum(e.clone()));
}
for s in &self.sequences {
entities.push(PostgresEntity::Sequence(s.clone()));
}
for r in &self.roles {
entities.push(PostgresEntity::Role(r.clone()));
}
for t in &self.tables {
entities.push(PostgresEntity::Table(t.clone()));
}
for c in &self.columns {
entities.push(PostgresEntity::Column(c.clone()));
}
for i in &self.indexes {
entities.push(PostgresEntity::Index(i.clone()));
}
for f in &self.foreign_keys {
entities.push(PostgresEntity::ForeignKey(f.clone()));
}
for p in &self.primary_keys {
entities.push(PostgresEntity::PrimaryKey(p.clone()));
}
for u in &self.unique_constraints {
entities.push(PostgresEntity::UniqueConstraint(u.clone()));
}
for c in &self.check_constraints {
entities.push(PostgresEntity::CheckConstraint(c.clone()));
}
for v in &self.views {
entities.push(PostgresEntity::View(v.clone()));
}
for p in &self.policies {
entities.push(PostgresEntity::Policy(p.clone()));
}
entities
}
}
#[must_use]
pub fn process_tables(raw_tables: &[RawTableInfo]) -> Vec<Table> {
raw_tables
.iter()
.filter(|t| !is_system_namespace(&t.schema))
.map(|t| Table {
schema: t.schema.clone().into(),
name: t.name.clone().into(),
is_unlogged: if t.is_unlogged { Some(true) } else { None },
is_temporary: if t.is_temporary { Some(true) } else { None },
inherits: None,
tablespace: t.tablespace.clone().map(Into::into),
is_rls_enabled: Some(t.is_rls_enabled),
comment: t.comment.clone().map(Into::into),
})
.collect()
}
#[derive(Debug, Clone, Default)]
struct IdentityOptions {
start: Option<String>,
increment: Option<String>,
min: Option<String>,
max: Option<String>,
cycle: Option<bool>,
}
fn parse_identity_type(raw: &str) -> (String, IdentityOptions) {
let trimmed = raw.trim();
if trimmed.starts_with('{')
&& let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed)
{
let get = |key: &str| {
value
.get(key)
.and_then(serde_json::Value::as_str)
.map(ToString::to_string)
};
let type_str = get("type").unwrap_or_else(|| "ALWAYS".to_string());
let options = IdentityOptions {
start: get("start"),
increment: get("increment"),
min: get("min"),
max: get("max"),
cycle: value.get("cycle").and_then(serde_json::Value::as_bool),
};
return (type_str, options);
}
(trimmed.to_string(), IdentityOptions::default())
}
#[must_use]
pub fn process_columns(raw_columns: &[RawColumnInfo]) -> Vec<Column> {
use super::ddl::{GeneratedType, IdentityType};
raw_columns
.iter()
.filter(|c| !is_system_namespace(&c.schema))
.map(|c| {
let generated = if c.is_generated {
c.generated_expression
.as_ref()
.map(|expr| super::ddl::Generated {
expression: expr.clone().into(),
gen_type: if c.generated_stored {
GeneratedType::Stored
} else {
GeneratedType::Virtual
},
})
} else {
None
};
let identity = if c.is_identity {
c.identity_type.as_ref().map(|t| {
let (type_str, options) = parse_identity_type(t);
let identity_type = if type_str.eq_ignore_ascii_case("always") {
IdentityType::Always
} else {
IdentityType::ByDefault
};
super::ddl::Identity {
name: format!("{}_{}_seq", c.table, c.name).into(),
schema: Some(c.schema.clone().into()),
type_: identity_type,
increment: options.increment.map(Into::into),
min_value: options.min.map(Into::into),
max_value: options.max.map(Into::into),
start_with: options.start.map(Into::into),
cache: None,
cycle: options.cycle,
}
})
} else {
None
};
let dimensions = c.dimensions.filter(|dims| *dims > 0).or_else(|| {
if c.column_type.starts_with('_') {
Some(1)
} else {
None
}
});
let column_type = if dimensions.is_some() {
c.column_type
.strip_prefix('_')
.unwrap_or(&c.column_type)
.to_string()
} else {
c.column_type.clone()
};
Column {
schema: c.schema.clone().into(),
table: c.table.clone().into(),
name: c.name.clone().into(),
sql_type: column_type.into(),
type_schema: c.type_schema.clone().map(std::convert::Into::into),
not_null: c.not_null,
default: c.default_value.clone().map(std::convert::Into::into),
generated,
identity,
dimensions,
comment: c.comment.clone().map(std::convert::Into::into),
collate: None,
ordinal_position: Some(c.ordinal_position),
}
})
.collect()
}
#[must_use]
pub fn process_enums(raw_enums: &[RawEnumInfo]) -> Vec<Enum> {
raw_enums
.iter()
.filter(|e| !is_system_namespace(&e.schema))
.map(|e| Enum {
schema: e.schema.clone().into(),
name: e.name.clone().into(),
values: e.values.iter().map(|v| v.clone().into()).collect(),
})
.collect()
}
#[must_use]
pub fn process_sequences(raw_sequences: &[RawSequenceInfo]) -> Vec<Sequence> {
raw_sequences
.iter()
.filter(|s| !is_system_namespace(&s.schema) && s.owned_by.is_none())
.map(|s| Sequence {
schema: s.schema.clone().into(),
name: s.name.clone().into(),
increment_by: s.increment.clone().map(Into::into),
min_value: s.min_value.clone().map(Into::into),
max_value: s.max_value.clone().map(Into::into),
start_with: s.start_value.clone().map(Into::into),
cache_size: s.cache_value.as_deref().and_then(|v| v.parse().ok()),
cycle: s.cycle,
})
.collect()
}
#[must_use]
pub fn process_indexes(raw_indexes: &[RawIndexInfo]) -> Vec<Index> {
use super::ddl::Opclass;
raw_indexes
.iter()
.filter(|i| !is_system_namespace(&i.schema) && !i.is_primary)
.map(|i| {
let columns: Vec<IndexColumn> = i
.columns
.iter()
.map(|c| IndexColumn {
value: c.name.clone().into(),
is_expression: c.is_expression,
asc: c.asc,
nulls_first: c.nulls_first,
opclass: c.opclass.clone().map(Opclass::new),
})
.collect();
Index {
schema: i.schema.clone().into(),
table: i.table.clone().into(),
name: i.name.clone().into(),
name_explicit: true,
columns,
is_unique: i.is_unique,
where_clause: i.where_clause.clone().map(std::convert::Into::into),
method: Some(i.method.clone().into()),
concurrently: i.concurrent,
r#with: None,
}
})
.collect()
}
#[must_use]
pub fn process_foreign_keys(raw_fks: &[RawForeignKeyInfo]) -> Vec<ForeignKey> {
raw_fks
.iter()
.filter(|f| !is_system_namespace(&f.schema))
.map(|f| ForeignKey {
schema: f.schema.clone().into(),
table: f.table.clone().into(),
name: f.name.clone().into(),
name_explicit: true,
columns: f.columns.iter().map(|c| c.clone().into()).collect(),
schema_to: f.schema_to.clone().into(),
table_to: f.table_to.clone().into(),
columns_to: f.columns_to.iter().map(|c| c.clone().into()).collect(),
on_update: Some(f.on_update.clone().into()),
on_delete: Some(f.on_delete.clone().into()),
deferrable: f.deferrable,
initially_deferred: f.initially_deferred,
})
.collect()
}
#[must_use]
pub fn process_primary_keys(raw_pks: &[RawPrimaryKeyInfo]) -> Vec<PrimaryKey> {
raw_pks
.iter()
.filter(|p| !is_system_namespace(&p.schema))
.map(|p| PrimaryKey {
schema: p.schema.clone().into(),
table: p.table.clone().into(),
name: p.name.clone().into(),
name_explicit: true,
columns: p.columns.iter().map(|c| c.clone().into()).collect(),
})
.collect()
}
#[must_use]
pub fn process_unique_constraints(raw_uniques: &[RawUniqueInfo]) -> Vec<UniqueConstraint> {
raw_uniques
.iter()
.filter(|u| !is_system_namespace(&u.schema))
.map(|u| UniqueConstraint {
schema: u.schema.clone().into(),
table: u.table.clone().into(),
name: u.name.clone().into(),
name_explicit: true,
columns: u.columns.iter().map(|c| c.clone().into()).collect(),
nulls_not_distinct: u.nulls_not_distinct,
deferrable: u.deferrable,
initially_deferred: u.initially_deferred,
})
.collect()
}
#[must_use]
pub fn process_check_constraints(raw_checks: &[RawCheckInfo]) -> Vec<CheckConstraint> {
raw_checks
.iter()
.filter(|c| !is_system_namespace(&c.schema))
.map(|c| CheckConstraint {
schema: c.schema.clone().into(),
table: c.table.clone().into(),
name: c.name.clone().into(),
value: c.expression.clone().into(),
})
.collect()
}
#[must_use]
pub fn process_views(raw_views: &[RawViewInfo]) -> Vec<View> {
raw_views
.iter()
.filter(|v| !is_system_namespace(&v.schema))
.map(|v| View {
schema: v.schema.clone().into(),
name: v.name.clone().into(),
definition: Some(v.definition.clone().into()),
materialized: v.is_materialized,
r#with: None,
is_existing: false,
with_no_data: None,
using: None,
tablespace: None,
})
.collect()
}
#[must_use]
pub fn process_policies(raw_policies: &[RawPolicyInfo]) -> Vec<Policy> {
use std::borrow::Cow;
raw_policies
.iter()
.filter(|p| !is_system_namespace(&p.schema))
.map(|p| {
let roles = p.to.iter().cloned().map(Cow::Owned).collect();
Policy {
schema: p.schema.clone().into(),
table: p.table.clone().into(),
name: p.name.clone().into(),
as_clause: Some(p.as_clause.clone().into()),
for_clause: Some(p.for_clause.clone().into()),
to: Some(roles),
using: p.using.clone().map(std::convert::Into::into),
with_check: p.with_check.clone().map(std::convert::Into::into),
}
})
.collect()
}
#[must_use]
pub fn process_roles(raw_roles: &[RawRoleInfo]) -> Vec<Role> {
raw_roles
.iter()
.filter(|r| !is_system_role(&r.name))
.map(|r| Role {
name: r.name.clone().into(),
superuser: None,
create_db: Some(r.create_db),
create_role: Some(r.create_role),
inherit: Some(r.inherit),
can_login: None,
replication: None,
bypass_rls: None,
conn_limit: None,
password: None,
valid_until: None,
})
.collect()
}
pub mod queries {
pub const SCHEMAS_QUERY: &str = r"
SELECT n.nspname AS name
FROM pg_namespace n
WHERE n.nspname NOT LIKE 'pg_%'
AND n.nspname != 'information_schema'
AND has_schema_privilege(current_user, n.oid, 'USAGE')
ORDER BY n.nspname
";
pub const TABLES_QUERY: &str = r"
SELECT
n.nspname AS schema,
c.relname AS name,
c.relrowsecurity AS is_rls_enabled,
c.relpersistence = 'u' AS is_unlogged,
c.relpersistence = 't' AS is_temporary,
tsp.spcname AS tablespace,
obj_description(c.oid, 'pg_class') AS comment
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
LEFT JOIN pg_tablespace tsp ON tsp.oid = c.reltablespace
WHERE c.relkind IN ('r', 'p')
AND n.nspname NOT LIKE 'pg_%'
AND n.nspname != 'information_schema'
AND has_schema_privilege(current_user, n.oid, 'USAGE')
AND has_table_privilege(current_user, c.oid, 'SELECT')
ORDER BY n.nspname, c.relname
";
pub const COLUMNS_QUERY: &str = r"
SELECT
c.table_schema AS schema,
c.table_name AS table,
c.column_name AS name,
c.udt_name || CASE
WHEN c.data_type != 'ARRAY' AND c.character_maximum_length IS NOT NULL
THEN '(' || c.character_maximum_length || ')'
WHEN c.udt_name IN ('numeric', 'decimal')
AND c.numeric_precision IS NOT NULL
AND c.numeric_scale IS NOT NULL
THEN '(' || c.numeric_precision || ',' || c.numeric_scale || ')'
ELSE ''
END AS column_type,
c.udt_schema AS type_schema,
c.is_nullable = 'NO' AS not_null,
c.column_default AS default_value,
c.is_identity = 'YES' AS is_identity,
CASE
WHEN c.is_identity = 'YES' THEN json_build_object(
'type', c.identity_generation,
'start', c.identity_start,
'increment', c.identity_increment,
'min', c.identity_minimum,
'max', c.identity_maximum,
'cycle', c.identity_cycle = 'YES'
)::text
ELSE NULL
END AS identity_type,
c.is_generated = 'ALWAYS' AS is_generated,
c.generation_expression AS generated_expression,
COALESCE(a.attgenerated = 's', false) AS generated_stored,
NULLIF(a.attndims::int4, 0) AS dimensions,
col_description(cls.oid, a.attnum) AS comment,
c.ordinal_position
FROM information_schema.columns c
LEFT JOIN pg_namespace n
ON n.nspname = c.table_schema
LEFT JOIN pg_class cls
ON cls.relnamespace = n.oid
AND cls.relname = c.table_name
LEFT JOIN pg_attribute a
ON a.attrelid = cls.oid
AND a.attname = c.column_name
AND a.attnum > 0
AND NOT a.attisdropped
WHERE c.table_schema NOT LIKE 'pg_%'
AND c.table_schema != 'information_schema'
AND n.oid IS NOT NULL
AND has_schema_privilege(current_user, n.oid, 'USAGE')
UNION ALL
-- Materialized-view columns: information_schema.columns excludes
-- matviews entirely, so read them straight from pg_attribute.
SELECT
mn.nspname AS schema,
mc.relname AS table,
ma.attname AS name,
mt.typname || COALESCE(
substring(format_type(ma.atttypid, ma.atttypmod) from '\(.*\)'),
''
) AS column_type,
mtn.nspname AS type_schema,
ma.attnotnull AS not_null,
NULL::text AS default_value,
FALSE AS is_identity,
NULL::text AS identity_type,
FALSE AS is_generated,
NULL::text AS generated_expression,
FALSE AS generated_stored,
NULLIF(ma.attndims::int4, 0) AS dimensions,
col_description(mc.oid, ma.attnum) AS comment,
ma.attnum::int4 AS ordinal_position
FROM pg_class mc
JOIN pg_namespace mn ON mn.oid = mc.relnamespace
JOIN pg_attribute ma ON ma.attrelid = mc.oid
JOIN pg_type mt ON mt.oid = ma.atttypid
JOIN pg_namespace mtn ON mtn.oid = mt.typnamespace
WHERE mc.relkind = 'm'
AND ma.attnum > 0
AND NOT ma.attisdropped
AND mn.nspname NOT LIKE 'pg_%'
AND mn.nspname != 'information_schema'
AND has_schema_privilege(current_user, mn.oid, 'USAGE')
AND has_table_privilege(current_user, mc.oid, 'SELECT')
ORDER BY 1, 2, 15
";
pub const ENUMS_QUERY: &str = r"
SELECT
n.nspname AS schema,
t.typname AS name,
array_agg(e.enumlabel ORDER BY e.enumsortorder) AS values
FROM pg_type t
JOIN pg_enum e ON t.oid = e.enumtypid
JOIN pg_namespace n ON n.oid = t.typnamespace
WHERE n.nspname NOT LIKE 'pg_%'
AND n.nspname != 'information_schema'
AND has_schema_privilege(current_user, n.oid, 'USAGE')
GROUP BY n.nspname, t.typname
ORDER BY n.nspname, t.typname
";
pub const SEQUENCES_QUERY: &str = r"
SELECT
n.nspname AS schema,
c.relname AS name,
format_type(s.seqtypid, NULL)::text AS data_type,
s.seqstart::text AS start_value,
s.seqmin::text AS min_value,
s.seqmax::text AS max_value,
s.seqincrement::text AS increment,
s.seqcycle AS cycle,
s.seqcache::text AS cache_value,
-- Owning column's schema.table when the sequence is auto-owned by
-- a serial (deptype 'a') or identity (deptype 'i') column; NULL
-- for standalone, hand-managed sequences.
(
SELECT format('%s.%s', dn.nspname, dc.relname)
FROM pg_depend d
JOIN pg_class dc ON dc.oid = d.refobjid
JOIN pg_namespace dn ON dn.oid = dc.relnamespace
WHERE d.objid = s.seqrelid
AND d.classid = 'pg_class'::regclass
AND d.refclassid = 'pg_class'::regclass
AND d.refobjsubid > 0
AND d.deptype IN ('a', 'i')
LIMIT 1
)::text AS owned_by
FROM pg_sequence s
JOIN pg_class c ON c.oid = s.seqrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname NOT LIKE 'pg_%'
AND n.nspname != 'information_schema'
AND has_schema_privilege(current_user, n.oid, 'USAGE')
AND (
-- Reference s.seqrelid (not c.oid) so this qual only depends on
-- pg_sequence: PostgreSQL 18's planner can push it down to the
-- pg_class scan before the join filters to sequences, and
-- has_sequence_privilege errors on non-sequence relations.
has_sequence_privilege(current_user, s.seqrelid, 'USAGE')
OR has_sequence_privilege(current_user, s.seqrelid, 'SELECT')
)
ORDER BY n.nspname, c.relname
";
pub const VIEWS_QUERY: &str = r"
SELECT
n.nspname AS schema,
c.relname AS name,
pg_get_viewdef(c.oid) AS definition,
FALSE AS is_materialized
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE (
($1::text[] IS NOT NULL AND n.nspname = ANY($1::text[]))
OR ($1::text[] IS NULL AND n.nspname NOT LIKE 'pg_%'
AND n.nspname != 'information_schema')
)
AND c.relkind = 'v'
AND has_schema_privilege(current_user, n.oid, 'USAGE')
AND has_table_privilege(current_user, c.oid, 'SELECT')
AND pg_get_viewdef(c.oid) IS NOT NULL
UNION ALL
SELECT
n.nspname AS schema,
c.relname AS name,
pg_get_viewdef(c.oid) AS definition,
TRUE AS is_materialized
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE (
($1::text[] IS NOT NULL AND n.nspname = ANY($1::text[]))
OR ($1::text[] IS NULL AND n.nspname NOT LIKE 'pg_%'
AND n.nspname != 'information_schema')
)
AND c.relkind = 'm'
AND has_schema_privilege(current_user, n.oid, 'USAGE')
AND has_table_privilege(current_user, c.oid, 'SELECT')
AND pg_get_viewdef(c.oid) IS NOT NULL
ORDER BY schema, name
";
pub const INDEXES_QUERY: &str = r"
SELECT
ns.nspname AS schema,
tbl.relname AS table,
idx.relname AS name,
ix.indisunique AS is_unique,
ix.indisprimary AS is_primary,
am.amname AS method,
array_agg(pg_get_indexdef(ix.indexrelid, s.n, true) ORDER BY s.n) AS columns,
pg_get_expr(ix.indpred, ix.indrelid) AS where_clause
FROM pg_index ix
JOIN pg_class idx ON idx.oid = ix.indexrelid
JOIN pg_class tbl ON tbl.oid = ix.indrelid
JOIN pg_namespace ns ON ns.oid = tbl.relnamespace
JOIN pg_am am ON am.oid = idx.relam
JOIN generate_series(1, ix.indnkeyatts) AS s(n) ON TRUE
WHERE ns.nspname NOT LIKE 'pg_%'
AND ns.nspname <> 'information_schema'
AND has_schema_privilege(current_user, ns.oid, 'USAGE')
AND has_table_privilege(current_user, tbl.oid, 'SELECT')
GROUP BY ns.nspname, tbl.relname, idx.relname, ix.indisunique, ix.indisprimary, am.amname, ix.indpred, ix.indrelid
ORDER BY ns.nspname, tbl.relname, idx.relname
";
pub const INDEXES_QUERY_FILTERED: &str = r"
SELECT
ns.nspname AS schema,
tbl.relname AS table,
idx.relname AS name,
ix.indisunique AS is_unique,
ix.indisprimary AS is_primary,
am.amname AS method,
array_agg(pg_get_indexdef(ix.indexrelid, s.n, true) ORDER BY s.n) AS columns,
pg_get_expr(ix.indpred, ix.indrelid) AS where_clause
FROM pg_index ix
JOIN pg_class idx ON idx.oid = ix.indexrelid
JOIN pg_class tbl ON tbl.oid = ix.indrelid
JOIN pg_namespace ns ON ns.oid = tbl.relnamespace
JOIN pg_am am ON am.oid = idx.relam
JOIN generate_series(1, ix.indnkeyatts) AS s(n) ON TRUE
WHERE ns.nspname = ANY($1::text[])
AND has_schema_privilege(current_user, ns.oid, 'USAGE')
AND has_table_privilege(current_user, tbl.oid, 'SELECT')
GROUP BY ns.nspname, tbl.relname, idx.relname, ix.indisunique, ix.indisprimary, am.amname, ix.indpred, ix.indrelid
ORDER BY ns.nspname, tbl.relname, idx.relname
";
pub const FOREIGN_KEYS_QUERY: &str = r"
SELECT
ns.nspname AS schema,
tbl.relname AS table,
con.conname AS name,
array_agg(src.attname ORDER BY s.ord) AS columns,
ns_to.nspname AS schema_to,
tbl_to.relname AS table_to,
array_agg(dst.attname ORDER BY s.ord) AS columns_to,
con.confupdtype::text AS on_update,
con.confdeltype::text AS on_delete,
con.condeferrable AS deferrable,
con.condeferred AS initially_deferred
FROM pg_constraint con
JOIN pg_class tbl ON tbl.oid = con.conrelid
JOIN pg_namespace ns ON ns.oid = tbl.relnamespace
JOIN pg_class tbl_to ON tbl_to.oid = con.confrelid
JOIN pg_namespace ns_to ON ns_to.oid = tbl_to.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS s(attnum, ord) ON TRUE
JOIN pg_attribute src ON src.attrelid = tbl.oid AND src.attnum = s.attnum
JOIN unnest(con.confkey) WITH ORDINALITY AS r(attnum, ord) ON r.ord = s.ord
JOIN pg_attribute dst ON dst.attrelid = tbl_to.oid AND dst.attnum = r.attnum
WHERE con.contype = 'f'
AND ns.nspname NOT LIKE 'pg_%'
AND ns.nspname <> 'information_schema'
AND has_schema_privilege(current_user, ns.oid, 'USAGE')
AND has_table_privilege(current_user, tbl.oid, 'SELECT')
GROUP BY ns.nspname, tbl.relname, con.conname, ns_to.nspname, tbl_to.relname, con.confupdtype, con.confdeltype, con.condeferrable, con.condeferred
ORDER BY ns.nspname, tbl.relname, con.conname
";
pub const PRIMARY_KEYS_QUERY: &str = r"
SELECT
ns.nspname AS schema,
tbl.relname AS table,
con.conname AS name,
array_agg(att.attname ORDER BY s.ord) AS columns
FROM pg_constraint con
JOIN pg_class tbl ON tbl.oid = con.conrelid
JOIN pg_namespace ns ON ns.oid = tbl.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS s(attnum, ord) ON TRUE
JOIN pg_attribute att ON att.attrelid = tbl.oid AND att.attnum = s.attnum
WHERE con.contype = 'p'
AND ns.nspname NOT LIKE 'pg_%'
AND ns.nspname <> 'information_schema'
AND has_schema_privilege(current_user, ns.oid, 'USAGE')
AND has_table_privilege(current_user, tbl.oid, 'SELECT')
GROUP BY ns.nspname, tbl.relname, con.conname
ORDER BY ns.nspname, tbl.relname, con.conname
";
pub const UNIQUES_QUERY: &str = r"
SELECT
ns.nspname AS schema,
tbl.relname AS table,
con.conname AS name,
array_agg(att.attname ORDER BY s.ord) AS columns,
COALESCE((
SELECT (to_jsonb(ix) ->> 'indnullsnotdistinct')::bool
FROM pg_index ix
WHERE ix.indexrelid = con.conindid
), FALSE) AS nulls_not_distinct,
con.condeferrable AS deferrable,
con.condeferred AS initially_deferred
FROM pg_constraint con
JOIN pg_class tbl ON tbl.oid = con.conrelid
JOIN pg_namespace ns ON ns.oid = tbl.relnamespace
JOIN unnest(con.conkey) WITH ORDINALITY AS s(attnum, ord) ON TRUE
JOIN pg_attribute att ON att.attrelid = tbl.oid AND att.attnum = s.attnum
WHERE con.contype = 'u'
AND ns.nspname NOT LIKE 'pg_%'
AND ns.nspname <> 'information_schema'
AND has_schema_privilege(current_user, ns.oid, 'USAGE')
AND has_table_privilege(current_user, tbl.oid, 'SELECT')
GROUP BY ns.nspname, tbl.relname, con.conname, con.conindid, con.condeferrable, con.condeferred
ORDER BY ns.nspname, tbl.relname, con.conname
";
pub const CHECKS_QUERY: &str = r"
SELECT
ns.nspname AS schema,
tbl.relname AS table,
con.conname AS name,
pg_get_expr(con.conbin, con.conrelid) AS expression
FROM pg_constraint con
JOIN pg_class tbl ON tbl.oid = con.conrelid
JOIN pg_namespace ns ON ns.oid = tbl.relnamespace
WHERE con.contype = 'c'
AND ns.nspname NOT LIKE 'pg_%'
AND ns.nspname <> 'information_schema'
AND has_schema_privilege(current_user, ns.oid, 'USAGE')
AND has_table_privilege(current_user, tbl.oid, 'SELECT')
ORDER BY ns.nspname, tbl.relname, con.conname
";
pub const CHECKS_QUERY_FILTERED: &str = r"
SELECT
ns.nspname AS schema,
tbl.relname AS table,
con.conname AS name,
pg_get_expr(con.conbin, con.conrelid) AS expression
FROM pg_constraint con
JOIN pg_class tbl ON tbl.oid = con.conrelid
JOIN pg_namespace ns ON ns.oid = tbl.relnamespace
WHERE con.contype = 'c'
AND ns.nspname = ANY($1::text[])
AND has_schema_privilege(current_user, ns.oid, 'USAGE')
AND has_table_privilege(current_user, tbl.oid, 'SELECT')
ORDER BY ns.nspname, tbl.relname, con.conname
";
pub const ROLES_QUERY: &str = r"
SELECT
rolname AS name,
rolcreatedb AS create_db,
rolcreaterole AS create_role,
rolinherit AS inherit
FROM pg_roles
ORDER BY rolname
";
pub const POLICIES_QUERY: &str = r#"
SELECT
n.nspname AS schema,
c.relname AS table,
p.polname AS name,
CASE
WHEN p.polpermissive THEN 'PERMISSIVE'::text
ELSE 'RESTRICTIVE'::text
END AS as_clause,
CASE p.polcmd
WHEN 'r'::"char" THEN 'SELECT'::text
WHEN 'a'::"char" THEN 'INSERT'::text
WHEN 'w'::"char" THEN 'UPDATE'::text
WHEN 'd'::"char" THEN 'DELETE'::text
WHEN '*'::"char" THEN 'ALL'::text
ELSE NULL::text
END AS for_clause,
CASE
WHEN p.polroles = '{0}'::oid[] THEN (string_to_array('public'::text, ''::text))::name[]
ELSE ARRAY(
SELECT pg_authid.rolname
FROM pg_authid
WHERE pg_authid.oid = ANY(p.polroles)
ORDER BY pg_authid.rolname
)
END AS to,
pg_get_expr(p.polqual, p.polrelid) AS using,
pg_get_expr(p.polwithcheck, p.polrelid) AS with_check
FROM pg_policy p
JOIN pg_class c ON c.oid = p.polrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname NOT LIKE 'pg_%'
AND n.nspname <> 'information_schema'
AND has_schema_privilege(current_user, n.oid, 'USAGE')
AND has_table_privilege(current_user, c.oid, 'SELECT')
ORDER BY n.nspname, c.relname, p.polname
"#;
}
#[must_use]
pub fn action_code_to_string(code: &str) -> String {
match code {
"r" => "RESTRICT",
"c" => "CASCADE",
"n" => "SET NULL",
"d" => "SET DEFAULT",
_ => "NO ACTION",
}
.to_string()
}
fn strip_trailing_directive<'a>(value: &'a str, directive: &str) -> Option<&'a str> {
let value = value.trim_end();
if value.len() <= directive.len() {
return None;
}
let split = value.len() - directive.len();
if !value.is_char_boundary(split) {
return None;
}
let (head, tail) = value.split_at(split);
if tail.eq_ignore_ascii_case(directive) && head.ends_with(char::is_whitespace) {
Some(head.trim_end())
} else {
None
}
}
fn is_plain_identifier(value: &str) -> bool {
!value.is_empty()
&& value
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '$')
&& value
.chars()
.next()
.is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
}
fn parens_balanced(value: &str) -> bool {
let mut depth = 0_i32;
for ch in value.chars() {
match ch {
'(' => depth += 1,
')' => {
depth -= 1;
if depth < 0 {
return false;
}
}
_ => {}
}
}
depth == 0
}
fn unquote_identifier(value: &str) -> Option<String> {
let inner = value.strip_prefix('"')?.strip_suffix('"')?;
Some(inner.replace("\"\"", "\""))
}
#[must_use]
pub fn parse_index_columns(cols: Vec<String>) -> Vec<RawIndexColumnInfo> {
cols.into_iter()
.map(|c| {
let mut core = c.trim().to_string();
let mut nulls_first: Option<bool> = None;
if let Some(rest) = strip_trailing_directive(&core, "NULLS FIRST") {
nulls_first = Some(true);
core = rest.to_string();
} else if let Some(rest) = strip_trailing_directive(&core, "NULLS LAST") {
nulls_first = Some(false);
core = rest.to_string();
}
let asc = if let Some(rest) = strip_trailing_directive(&core, "DESC") {
core = rest.to_string();
false
} else {
if let Some(rest) = strip_trailing_directive(&core, "ASC") {
core = rest.to_string();
}
true
};
let nulls_first = nulls_first.unwrap_or(!asc);
let mut opclass: Option<String> = None;
if let Some(idx) = core.rfind(char::is_whitespace) {
let candidate = core[idx..].trim();
let rest = core[..idx].trim_end();
if is_plain_identifier(candidate)
&& (candidate.ends_with("_ops")
|| super::grammar::VECTOR_OPS.contains(&candidate))
&& !rest.is_empty()
&& parens_balanced(rest)
{
opclass = Some(candidate.to_string());
core = rest.to_string();
}
}
let (name, is_expression) = if let Some(unquoted) = unquote_identifier(&core) {
(unquoted, false)
} else if is_plain_identifier(&core) {
(core, false)
} else {
(core, true)
};
RawIndexColumnInfo {
name,
is_expression,
asc,
nulls_first,
opclass,
}
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_process_tables() {
let raw = vec![
RawTableInfo {
schema: "public".to_string(),
name: "users".to_string(),
is_unlogged: false,
is_temporary: false,
tablespace: None,
comment: None,
is_rls_enabled: false,
},
RawTableInfo {
schema: "pg_catalog".to_string(),
name: "pg_class".to_string(),
is_unlogged: false,
is_temporary: false,
tablespace: None,
comment: None,
is_rls_enabled: false,
},
];
let tables = process_tables(&raw);
assert_eq!(tables.len(), 1);
assert_eq!(tables[0].name, "users");
}
#[test]
fn test_introspection_result_to_snapshot() {
let mut result = IntrospectionResult::default();
result.schemas.push(Schema::new("public"));
result.tables.push(Table {
schema: "public".into(),
name: "users".into(),
is_unlogged: None,
is_temporary: None,
inherits: None,
tablespace: None,
is_rls_enabled: None,
comment: None,
});
let snapshot = result.to_snapshot();
assert_eq!(snapshot.ddl.len(), 2);
}
#[test]
fn parse_index_columns_handles_realistic_indexdef_output() {
let cols = parse_index_columns(vec![
"(price * quantity)".to_string(),
"\"userName\" DESC".to_string(),
"lower(email) varchar_pattern_ops".to_string(),
"col DESC NULLS LAST".to_string(),
"\"say\"\"hi\"\"\"".to_string(),
"plain_col".to_string(),
"name text_pattern_ops".to_string(),
]);
assert_eq!(cols[0].name, "(price * quantity)");
assert!(cols[0].is_expression);
assert_eq!(cols[0].opclass, None);
assert!(cols[0].asc);
assert!(!cols[0].nulls_first);
assert_eq!(cols[1].name, "userName");
assert!(!cols[1].is_expression);
assert!(!cols[1].asc);
assert!(cols[1].nulls_first);
assert_eq!(cols[2].name, "lower(email)");
assert!(cols[2].is_expression);
assert_eq!(cols[2].opclass.as_deref(), Some("varchar_pattern_ops"));
assert_eq!(cols[3].name, "col");
assert!(!cols[3].asc);
assert!(!cols[3].nulls_first);
assert_eq!(cols[4].name, "say\"hi\"");
assert!(!cols[4].is_expression);
assert_eq!(cols[5].name, "plain_col");
assert!(!cols[5].is_expression);
assert!(cols[5].asc);
assert!(!cols[5].nulls_first);
assert_eq!(cols[6].name, "name");
assert!(!cols[6].is_expression);
assert_eq!(cols[6].opclass.as_deref(), Some("text_pattern_ops"));
}
#[test]
fn process_columns_populates_identity_options_from_packed_json() {
let raw = RawColumnInfo {
schema: "public".to_string(),
table: "users".to_string(),
name: "id".to_string(),
column_type: "int4".to_string(),
type_schema: Some("pg_catalog".to_string()),
not_null: true,
default_value: None,
is_identity: true,
identity_type: Some(
r#"{"type":"ALWAYS","start":"100","increment":"5","min":"1","max":"1000","cycle":true}"#
.to_string(),
),
is_generated: false,
generated_expression: None,
generated_stored: false,
dimensions: None,
comment: None,
ordinal_position: 1,
};
let columns = process_columns(&[raw]);
let identity = columns[0].identity.as_ref().expect("identity");
assert_eq!(identity.start_with.as_deref(), Some("100"));
assert_eq!(identity.increment.as_deref(), Some("5"));
assert_eq!(identity.min_value.as_deref(), Some("1"));
assert_eq!(identity.max_value.as_deref(), Some("1000"));
assert_eq!(identity.cycle, Some(true));
}
#[test]
fn process_columns_accepts_legacy_plain_identity_type() {
let raw = RawColumnInfo {
schema: "public".to_string(),
table: "users".to_string(),
name: "id".to_string(),
column_type: "int4".to_string(),
type_schema: Some("pg_catalog".to_string()),
not_null: true,
default_value: None,
is_identity: true,
identity_type: Some("BY DEFAULT".to_string()),
is_generated: false,
generated_expression: None,
generated_stored: false,
dimensions: None,
comment: None,
ordinal_position: 1,
};
let columns = process_columns(&[raw]);
let identity = columns[0].identity.as_ref().expect("identity");
assert_eq!(identity.type_, super::super::ddl::IdentityType::ByDefault);
assert_eq!(identity.increment, None);
}
#[test]
fn postgres_catalog_queries_are_privilege_scoped() {
use queries::{
CHECKS_QUERY, COLUMNS_QUERY, ENUMS_QUERY, FOREIGN_KEYS_QUERY, INDEXES_QUERY,
POLICIES_QUERY, PRIMARY_KEYS_QUERY, SCHEMAS_QUERY, SEQUENCES_QUERY, TABLES_QUERY,
UNIQUES_QUERY, VIEWS_QUERY,
};
for query in [
SCHEMAS_QUERY,
TABLES_QUERY,
COLUMNS_QUERY,
ENUMS_QUERY,
SEQUENCES_QUERY,
VIEWS_QUERY,
INDEXES_QUERY,
FOREIGN_KEYS_QUERY,
PRIMARY_KEYS_QUERY,
UNIQUES_QUERY,
CHECKS_QUERY,
POLICIES_QUERY,
] {
assert!(
query.contains("has_schema_privilege"),
"query is not schema-privilege scoped: {query}"
);
}
for query in [
TABLES_QUERY,
VIEWS_QUERY,
INDEXES_QUERY,
FOREIGN_KEYS_QUERY,
PRIMARY_KEYS_QUERY,
UNIQUES_QUERY,
CHECKS_QUERY,
POLICIES_QUERY,
] {
assert!(
query.contains("has_table_privilege"),
"query is not table-privilege scoped: {query}"
);
}
}
}