use std::collections::{HashMap, HashSet};
use serde::{Deserialize, Serialize};
use crate::schema::{SchemaDescriptor, SearchBackend, TypeDescriptor};
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct DbState {
#[serde(default)]
pub schemas: Vec<String>,
#[serde(default)]
pub tables: Vec<DbTable>,
#[serde(default)]
pub enums: Vec<DbEnum>,
#[serde(default)]
pub domains: Vec<DbDomain>,
#[serde(default)]
pub sequences: Vec<DbSequence>,
#[serde(default)]
pub views: Vec<DbView>,
#[serde(default)]
pub functions: Vec<DbFunction>,
#[serde(default)]
pub extensions: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbTable {
pub schema: String,
pub name: String,
pub columns: Vec<DbColumn>,
pub foreign_keys: Vec<DbForeignKey>,
pub indexes: Vec<DbIndex>,
pub checks: Vec<DbCheck>,
#[serde(default)]
pub triggers: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbColumn {
pub name: String,
pub pg_type: String,
pub nullable: bool,
pub is_generated: bool,
#[serde(default)]
pub column_default: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbForeignKey {
pub constraint_name: String,
pub local_column: String,
pub ref_schema: String,
pub ref_table: String,
}
fn capped_index_name(name: String) -> String {
const MAX: usize = 63;
if name.len() <= MAX {
return name;
}
let hash = crate::export::fnv(&[&name]);
let head: String = name.chars().take(MAX - 9).collect();
format!("{head}_{}", &hash[..8])
}
fn pointer_column(td: &TypeDescriptor, pointer: &str) -> String {
if td.links.iter().any(|l| l.name == pointer && !l.is_junction_backed()) {
format!("{pointer}_id")
} else {
pointer.to_string()
}
}
fn inherited_from_an_interface(schema: &SchemaDescriptor, td: &TypeDescriptor, pointer: &str) -> bool {
td.interfaces.iter().any(|iface| {
schema
.types
.iter()
.filter(|t| format!("{}::{}", t.module, t.name) == *iface)
.any(|t| {
t.properties.iter().any(|p| p.name == pointer)
|| t.links.iter().any(|l| l.name == pointer)
|| t.multilinks.iter().any(|ml| ml.name == pointer)
})
})
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbIndex {
pub name: String,
pub is_unique: bool,
pub method: String,
#[serde(default)]
pub columns: Vec<String>,
#[serde(default)]
pub predicate: Option<String>,
#[serde(default)]
pub key: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbCheck {
pub constraint_name: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbEnum {
pub schema: String,
pub name: String,
pub members: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbDomain {
pub schema: String,
pub name: String,
#[serde(default)]
pub checks: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbSequence {
pub schema: String,
pub name: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbView {
pub schema: String,
pub name: String,
pub body_hash: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DbFunction {
pub schema: String,
pub name: String,
pub body_hash: String,
}
pub fn schema_to_db_state(schema: &SchemaDescriptor) -> DbState {
use std::collections::BTreeSet;
let type_map: HashMap<String, (&str, &str)> = schema
.types
.iter()
.map(|t| {
(
format!("{}::{}", t.module, t.name),
(t.module.as_str(), t.table.as_str()),
)
})
.collect();
let mut schema_set: BTreeSet<String> = BTreeSet::new();
for t in &schema.types {
schema_set.insert(t.module.clone());
}
for e in &schema.enums {
schema_set.insert(e.module.clone());
}
for s in &schema.scalars {
schema_set.insert(s.module.clone());
}
let schemas: Vec<String> = schema_set.into_iter().collect();
let enums: Vec<DbEnum> = schema
.enums
.iter()
.map(|e| DbEnum {
schema: e.module.clone(),
name: e.name.clone(),
members: e.members.clone(),
})
.collect();
let domains: Vec<DbDomain> = schema
.scalars
.iter()
.map(|s| DbDomain {
schema: s.module.clone(),
name: s.name.clone(),
checks: crate::export::scalar_check_constraints(schema)
.into_iter()
.filter(|(module, name, _, _)| *module == s.module && *name == s.name)
.map(|(_, _, cname, _)| cname)
.collect(),
})
.collect();
let sequences: Vec<DbSequence> = schema
.scalars
.iter()
.filter(|s| s.is_sequence)
.map(|s| DbSequence {
schema: s.module.clone(),
name: format!("{}_seq", s.name),
})
.collect();
let expected_trigger_names = expected_triggers(schema, &type_map);
let mut expected_checks: HashMap<(String, String), Vec<String>> = HashMap::new();
for (module, table, cname, _) in crate::export::check_constraints(schema).unwrap_or_default() {
expected_checks.entry((module, table)).or_default().push(cname);
}
let mut tables: Vec<DbTable> = Vec::new();
for td in &schema.types {
if td.abstract_ || td.junction {
continue;
}
let mut columns: Vec<DbColumn> = Vec::new();
for p in &td.properties {
columns.push(DbColumn {
name: p.name.clone(),
pg_type: col_type_str(p).to_string(),
nullable: p.nullable,
is_generated: false,
column_default: resolve_default(p, schema),
});
}
for l in &td.links {
if l.is_junction_backed() {
continue;
}
columns.push(DbColumn {
name: format!("{}_id", l.name),
pg_type: "uuid".to_string(),
nullable: l.nullable,
is_generated: false,
column_default: resolve_link_default(l, schema),
});
}
for vi in &td.vector_indexes {
let col = vi.column_name();
if !columns.iter().any(|c| c.name == col) {
columns.push(DbColumn {
name: col,
pg_type: format!("vector({})", vi.dimensions),
nullable: true,
is_generated: false,
column_default: None,
});
}
}
for si in &td.search_indexes {
if si.backend != SearchBackend::Postgres {
continue;
}
let col = si.column_name();
if !columns.iter().any(|c| c.name == col) {
columns.push(DbColumn {
name: col,
pg_type: "tsvector".to_string(),
nullable: true,
is_generated: true,
column_default: None,
});
}
}
let mut foreign_keys: Vec<DbForeignKey> = Vec::new();
for l in &td.links {
if l.is_junction_backed() {
continue;
}
let cname = format!("{}_{}_fkey", td.table, l.name);
if let Some((tgt_schema, tgt_table)) = type_map.get(&l.target) {
foreign_keys.push(DbForeignKey {
constraint_name: cname,
local_column: format!("{}_id", l.name),
ref_schema: tgt_schema.to_string(),
ref_table: tgt_table.to_string(),
});
}
}
let mut indexes: Vec<DbIndex> = Vec::new();
for p in &td.properties {
if p.is_exclusive && !p.is_pk && !inherited_from_an_interface(schema, td, &p.name) {
indexes.push(DbIndex {
name: format!("{}_{}_key", td.table, p.name),
is_unique: true,
method: "btree".to_string(),
columns: vec![p.name.clone()],
predicate: None,
key: None,
});
}
}
for l in &td.links {
if l.is_exclusive && !l.is_junction_backed() && !inherited_from_an_interface(schema, td, &l.name) {
indexes.push(DbIndex {
name: format!("{}_{}_id_key", td.table, l.name),
is_unique: true,
method: "btree".to_string(),
columns: vec![format!("{}_id", l.name)],
predicate: None,
key: None,
});
}
}
for (i, constraint) in td.constraints.iter().enumerate() {
use crate::schema::TypeConstraint;
if let TypeConstraint::Exclusive {
pointers: fields,
unless,
} = constraint
{
let idx_name = format!("{}_{}_{}_key", td.table, fields.join("_"), i);
indexes.push(DbIndex {
name: idx_name,
is_unique: true,
method: "btree".to_string(),
columns: fields.iter().map(|f| pointer_column(td, f)).collect(),
predicate: unless.as_deref().and_then(|u| {
let qualified = format!("{}::{}", td.module, td.name);
crate::ir::compile_constraint_expr(u, &qualified, schema)
.ok()
.map(|c| format!(" WHERE NOT ({c})"))
}),
key: None,
});
}
}
for (i, idx) in td.indexes.iter().enumerate() {
let name = if idx.expression.is_some() {
format!("{}__expr{}_idx", td.table, i)
} else {
format!("{}__{}_idx", td.table, idx.pointers.join("_"))
};
let Ok((body, predicate)) = crate::export::index_body_and_predicate(
td,
&idx.pointers,
idx.expression.as_deref(),
idx.unless.as_deref(),
schema,
) else {
continue;
};
indexes.push(DbIndex {
name,
is_unique: idx.unique,
method: "btree".to_string(),
columns: vec![],
key: Some(body),
predicate: (!predicate.is_empty()).then_some(predicate),
});
}
for vi in &td.vector_indexes {
let idx_name = match &vi.index_name {
None => format!("{}__vector__", td.table),
Some(n) => format!("{}__vector_{}__", td.table, n),
};
indexes.push(DbIndex {
name: idx_name,
is_unique: false,
method: "hnsw".to_string(),
columns: vec![],
predicate: None,
key: None,
});
}
for si in &td.search_indexes {
if si.backend != SearchBackend::Postgres {
continue;
}
let idx_name = match &si.index_name {
None => format!("{}__search__", td.table),
Some(n) => format!("{}__search_{}__", td.table, n),
};
indexes.push(DbIndex {
name: idx_name,
is_unique: false,
method: "gin".to_string(),
columns: vec![],
predicate: None,
key: None,
});
}
let checks: Vec<DbCheck> = expected_checks
.get(&(td.module.clone(), td.table.clone()))
.cloned()
.unwrap_or_default()
.into_iter()
.map(|constraint_name| DbCheck { constraint_name })
.collect();
let triggers: Vec<String> = expected_trigger_names
.get(&(td.module.clone(), td.table.clone()))
.cloned()
.unwrap_or_default()
.into_iter()
.collect();
tables.push(DbTable {
schema: td.module.clone(),
name: td.table.clone(),
columns,
foreign_keys,
indexes,
checks,
triggers,
});
for ml in &td.multilinks {
tables.push(build_junction_db_table(
schema,
&type_map,
td,
&ml.name,
&ml.target,
ml.through.as_deref(),
ml.is_exclusive,
&expected_trigger_names,
));
}
for l in &td.links {
if !l.is_junction_backed() {
continue;
}
tables.push(build_junction_db_table(
schema,
&type_map,
td,
&l.name,
&l.target,
l.through.as_deref(),
l.is_exclusive,
&expected_trigger_names,
));
}
}
let views: Vec<DbView> = crate::export::interface_view_ddl_with_names(schema)
.into_iter()
.chain(crate::export::interface_junction_view_ddl_with_names(schema))
.map(|(module, name, ddl)| DbView {
schema: module,
name,
body_hash: ddl_hash(&ddl),
})
.collect();
let functions: Vec<DbFunction> = crate::export::function_ddl_with_names(schema)
.unwrap_or_default()
.into_iter()
.map(|(module, name, ddl)| DbFunction {
schema: module,
name,
body_hash: ddl_hash(&ddl),
})
.collect();
let extensions: Vec<String> = required_extensions(schema).iter().map(|s| s.to_string()).collect();
let polymorphic_tables: HashSet<(&str, &str)> = {
let polymorphic = crate::export::polymorphic_types(schema);
schema
.types
.iter()
.filter(|t| polymorphic.contains(&format!("{}::{}", t.module, t.name)))
.map(|t| (t.module.as_str(), t.table.as_str()))
.collect()
};
for table in &mut tables {
table.foreign_keys.retain(|fk| {
fk.local_column == "source"
|| !polymorphic_tables.contains(&(fk.ref_schema.as_str(), fk.ref_table.as_str()))
});
}
DbState {
schemas,
tables,
enums,
domains,
sequences,
views,
functions,
extensions,
}
}
pub fn required_extensions(target: &SchemaDescriptor) -> Vec<&'static str> {
let mut out = Vec::new();
if target.types.iter().any(|t| !t.vector_indexes.is_empty()) {
out.push("vector");
}
if target.types.iter().any(|t| t.partition.is_some()) {
out.push("pg_partman");
}
out
}
pub fn missing_extension_ddl(target: &SchemaDescriptor, current: &DbState) -> Vec<String> {
required_extensions(target)
.into_iter()
.filter(|ext| !current.extensions.iter().any(|e| e == ext))
.map(|ext| format!("CREATE EXTENSION IF NOT EXISTS \"{ext}\";"))
.collect()
}
#[allow(clippy::too_many_arguments)]
fn build_junction_db_table(
schema: &SchemaDescriptor,
type_map: &HashMap<String, (&str, &str)>,
td: &TypeDescriptor,
name: &str,
target: &str,
through: Option<&str>,
exclusive: bool,
expected_trigger_names: &HashMap<(String, String), HashSet<String>>,
) -> DbTable {
let jt_name = format!("{}.{}", td.table, name);
let mut jt_columns = vec![
DbColumn {
name: "source".to_string(),
pg_type: "uuid".to_string(),
nullable: false,
is_generated: false,
column_default: None,
},
DbColumn {
name: "target".to_string(),
pg_type: "uuid".to_string(),
nullable: false,
is_generated: false,
column_default: None,
},
];
if let Some(through_qname) = through
&& let Some(through_td) = schema
.types
.iter()
.find(|t| format!("{}::{}", t.module, t.name) == *through_qname && t.junction)
{
for p in &through_td.properties {
if p.name == "id" {
continue;
}
let pg_type = col_type_str(p).to_string();
jt_columns.push(DbColumn {
name: p.name.clone(),
pg_type,
nullable: p.nullable,
is_generated: false,
column_default: p.default_sql.clone(),
});
}
}
let mut jt_fks = Vec::new();
let src_fk_name = format!("{}_{}_source_fkey", td.table, name);
jt_fks.push(DbForeignKey {
constraint_name: src_fk_name,
local_column: "source".to_string(),
ref_schema: td.module.clone(),
ref_table: td.table.clone(),
});
if let Some((tgt_schema, tgt_table)) = type_map.get(target) {
let tgt_fk_name = format!("{}_{}_target_fkey", td.table, name);
jt_fks.push(DbForeignKey {
constraint_name: tgt_fk_name,
local_column: "target".to_string(),
ref_schema: tgt_schema.to_string(),
ref_table: tgt_table.to_string(),
});
}
let mut indexes = Vec::new();
if exclusive && !inherited_from_an_interface(schema, td, name) {
indexes.push(DbIndex {
name: format!("{jt_name}_target_key"),
is_unique: true,
method: "btree".to_string(),
columns: vec!["target".to_string()],
predicate: None,
key: None,
});
}
let triggers: Vec<String> = expected_trigger_names
.get(&(td.module.clone(), jt_name.clone()))
.cloned()
.unwrap_or_default()
.into_iter()
.collect();
DbTable {
schema: td.module.clone(),
name: jt_name,
columns: jt_columns,
foreign_keys: jt_fks,
indexes,
checks: vec![],
triggers,
}
}
fn ddl_hash(ddl: &str) -> String {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(ddl.as_bytes());
hex::encode(&digest[..8])
}
pub fn db_state_to_json(state: &DbState) -> String {
serde_json::to_string(state).expect("DbState serialization is infallible")
}
pub fn db_state_from_json(json: &str) -> Result<DbState, String> {
serde_json::from_str(json).map_err(|e| e.to_string())
}
impl DbState {
pub fn add_trigger(&mut self, module: &str, table: &str, trigger_name: &str) {
if let Some(t) = self.tables.iter_mut().find(|t| t.schema == module && t.name == table) {
t.triggers.push(trigger_name.to_string());
}
}
}
fn junction_table_belongs_to_an_interface(schema: &SchemaDescriptor, td: &TypeDescriptor) -> bool {
let Some((owner_table, _)) = td.table.rsplit_once('.') else {
return false;
};
schema
.types
.iter()
.any(|t| t.abstract_ && t.module == td.module && t.table == owner_table)
}
pub fn expected_triggers(
schema: &SchemaDescriptor,
type_map: &HashMap<String, (&str, &str)>,
) -> HashMap<(String, String), HashSet<String>> {
let mut expected: HashMap<(String, String), HashSet<String>> = HashMap::new();
for info in crate::export::interface_exclusive_trigger_infos(schema) {
expected
.entry((info.impl_module.clone(), info.impl_table.clone()))
.or_default()
.extend([info.ins_trigger_name, info.upd_trigger_name]);
}
for info in crate::export::deletion_policy_trigger_infos(schema, type_map) {
expected
.entry((info.table_module.clone(), info.table_name.clone()))
.or_default()
.insert(info.trigger_name);
}
for info in crate::export::signal_trigger_infos(schema) {
expected
.entry((info.table_module.clone(), info.table_name.clone()))
.or_default()
.insert(info.trigger_name);
}
for (module, table, name) in crate::export::user_trigger_names(schema) {
expected.entry((module, table)).or_default().insert(name);
}
let mut cache_trigger_tables: HashSet<(String, String)> = HashSet::new();
for td in &schema.types {
if td.abstract_ || junction_table_belongs_to_an_interface(schema, td) {
continue;
}
cache_trigger_tables.insert((td.module.clone(), td.table.clone()));
if !td.junction {
for ml in &td.multilinks {
cache_trigger_tables.insert((td.module.clone(), format!("{}.{}", td.table, ml.name)));
}
for l in &td.links {
if !l.is_junction_backed() {
continue;
}
cache_trigger_tables.insert((td.module.clone(), format!("{}.{}", td.table, l.name)));
}
}
}
for key in cache_trigger_tables {
expected
.entry(key)
.or_default()
.insert("pylon_cache_invalidate".to_string());
}
expected
}
#[derive(Debug)]
pub struct TypeRenameCandidate {
pub old_module: String,
pub old_table: String,
pub new_module: String,
pub new_table: String,
pub new_type_name: String,
pub confidence: f64,
}
#[derive(Debug)]
pub struct ColRenameCandidate {
pub module: String,
pub table: String,
pub old_col: String,
pub new_col: String,
pub pg_type: String,
}
#[derive(Debug)]
pub struct FillRequired {
pub module: String,
pub table: String,
pub column: String,
pub pg_type: String,
pub type_name: String,
pub is_new_column: bool,
pub default_sql: Option<String>,
}
#[derive(Debug, Clone)]
pub struct DiffOp {
pub sql: String,
pub non_transactional: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Verb {
Create,
Alter,
Drop,
Rename,
}
impl Verb {
pub fn as_str(&self) -> &'static str {
match self {
Verb::Create => "create",
Verb::Alter => "alter",
Verb::Drop => "drop",
Verb::Rename => "rename",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum OpKey {
Module(String),
Scalar(String, String),
Table(String, String),
Function(String, String),
View(String, String),
ForeignKey(String, String),
}
#[derive(Debug)]
pub struct MigrationStep {
pub prompt: String,
pub verb: Verb,
pub object_desc: String,
pub ddl: Vec<DiffOp>,
pub op_key: OpKey,
pub required_input: Vec<RequiredInput>,
}
impl MigrationStep {
pub fn resolved_ddl(&self, overrides: &HashMap<String, String>) -> Vec<DiffOp> {
self.ddl
.iter()
.map(|op| {
let mut sql = op.sql.clone();
for input in &self.required_input {
let value = overrides.get(&input.placeholder).unwrap_or(&input.default_expr);
sql = sql.replace(&format!("\\({})", input.placeholder), value);
}
DiffOp {
sql,
non_transactional: op.non_transactional,
}
})
.collect()
}
}
#[derive(Debug, Clone)]
pub struct RequiredInput {
pub placeholder: String,
pub prompt: String,
pub default_expr: String,
pub type_name: String,
}
fn verbosename_module(name: &str) -> String {
format!("module '{name}'")
}
fn is_generated_check_name(name: &str) -> bool {
let Some(rest) = name.strip_suffix("_check") else {
return false;
};
match rest.rsplit_once('_') {
Some((prefix, hash)) => !prefix.is_empty() && hash.len() == 8 && hash.chars().all(|c| c.is_ascii_hexdigit()),
None => false,
}
}
fn verbosename_scalar(module: &str, name: &str) -> String {
format!("scalar type '{module}::{name}'")
}
fn verbosename_type(module: &str, name: &str) -> String {
format!("object type '{module}::{name}'")
}
fn verbosename_interface(module: &str, name: &str) -> String {
format!("interface type '{module}::{name}'")
}
fn verbosename_function(module: &str, name: &str) -> String {
format!("function '{module}::{name}'")
}
#[derive(Default)]
struct StepBuilder {
order: Vec<OpKey>,
drafts: HashMap<OpKey, (Verb, String, Vec<DiffOp>, Vec<RequiredInput>)>,
}
impl StepBuilder {
fn new() -> Self {
Self::default()
}
fn push(&mut self, key: OpKey, verb: Verb, object_desc: impl Into<String>, op: DiffOp) {
self.extend(key, verb, object_desc, vec![op]);
}
fn extend(&mut self, key: OpKey, verb: Verb, object_desc: impl Into<String>, ops: Vec<DiffOp>) {
self.extend_with_input(key, verb, object_desc, ops, vec![]);
}
fn extend_with_input(
&mut self,
key: OpKey,
verb: Verb,
object_desc: impl Into<String>,
ops: Vec<DiffOp>,
inputs: Vec<RequiredInput>,
) {
if ops.is_empty() && inputs.is_empty() {
return;
}
use std::collections::hash_map::Entry;
match self.drafts.entry(key.clone()) {
Entry::Occupied(mut e) => {
e.get_mut().2.extend(ops);
e.get_mut().3.extend(inputs);
}
Entry::Vacant(e) => {
e.insert((verb, object_desc.into(), ops, inputs));
self.order.push(key);
}
}
}
fn position(&self, key: &OpKey) -> Option<usize> {
self.order.iter().position(|existing| existing == key)
}
fn finish(self) -> Vec<MigrationStep> {
let Self { order, mut drafts } = self;
order
.into_iter()
.map(|key| {
let (verb, object_desc, ddl, required_input) = drafts.remove(&key).unwrap();
let prompt = format!("did you {} {}?", verb.as_str(), object_desc);
MigrationStep {
prompt,
verb,
object_desc,
ddl,
op_key: key,
required_input,
}
})
.collect()
}
}
#[derive(Debug, Default, Clone)]
pub struct Guidance {
pub banned_type_renames: HashSet<(String, String, String, String)>,
pub banned_col_renames: HashSet<(String, String, String, String)>,
}
pub fn diff_schema(target: &SchemaDescriptor, current: &DbState) -> Result<Vec<String>, String> {
Ok(flatten_ops(diff_inner(target, current, false, &HashMap::new())?)
.into_iter()
.map(|op| op.sql)
.collect())
}
pub fn diff_schema_ops(target: &SchemaDescriptor, current: &DbState) -> Result<Vec<DiffOp>, String> {
Ok(flatten_ops(diff_inner(target, current, true, &HashMap::new())?))
}
pub fn schema_content_changed(target: &SchemaDescriptor, previous: Option<&SchemaDescriptor>) -> bool {
let previous = previous.cloned().unwrap_or_default();
serde_json::to_value(target).ok() != serde_json::to_value(&previous).ok()
}
pub fn diff_schema_steps(
target: &SchemaDescriptor,
current: &DbState,
fill_index: &HashMap<(String, String), HashSet<String>>,
) -> Result<Vec<MigrationStep>, String> {
diff_inner(target, current, true, fill_index)
}
fn flatten_ops(steps: Vec<MigrationStep>) -> Vec<DiffOp> {
let no_overrides = HashMap::new();
steps.iter().flat_map(|s| s.resolved_ddl(&no_overrides)).collect()
}
pub fn diff_states(before: &DbState, after: &DbState) -> Vec<DiffOp> {
diff_states_inner(before, after)
}
pub fn detect_type_renames(
target: &SchemaDescriptor,
current: &DbState,
guidance: &Guidance,
) -> Vec<TypeRenameCandidate> {
let target_keys: HashSet<(&str, &str)> = target
.types
.iter()
.filter(|t| !t.abstract_ && !t.junction)
.map(|t| (t.module.as_str(), t.table.as_str()))
.collect();
let current_keys: HashSet<(&str, &str)> = current
.tables
.iter()
.map(|t| (t.schema.as_str(), t.name.as_str()))
.collect();
let dropped: Vec<&DbTable> = current
.tables
.iter()
.filter(|t| !target_keys.contains(&(t.schema.as_str(), t.name.as_str())))
.collect();
let created: Vec<&TypeDescriptor> = target
.types
.iter()
.filter(|t| !t.abstract_ && !t.junction)
.filter(|t| !current_keys.contains(&(t.module.as_str(), t.table.as_str())))
.collect();
if dropped.is_empty() || created.is_empty() {
return vec![];
}
let mut candidates: Vec<TypeRenameCandidate> = Vec::new();
for dropped_t in &dropped {
let old_cols: HashSet<&str> = dropped_t
.columns
.iter()
.map(|c| c.name.as_str())
.filter(|n| !n.starts_with("__"))
.collect();
for new_type in &created {
let new_cols: HashSet<&str> = new_type.properties.iter().map(|p| p.name.as_str()).collect();
let intersection = old_cols.intersection(&new_cols).count();
let union_size = old_cols.union(&new_cols).count();
if union_size == 0 {
continue;
}
let confidence = intersection as f64 / union_size as f64;
let banned = guidance.banned_type_renames.contains(&(
dropped_t.schema.clone(),
dropped_t.name.clone(),
new_type.module.clone(),
new_type.table.clone(),
));
if confidence >= 0.4 && !banned {
candidates.push(TypeRenameCandidate {
old_module: dropped_t.schema.clone(),
old_table: dropped_t.name.clone(),
new_module: new_type.module.clone(),
new_table: new_type.table.clone(),
new_type_name: new_type.name.clone(),
confidence,
});
}
}
}
candidates.sort_by(|a, b| {
b.confidence
.partial_cmp(&a.confidence)
.unwrap_or(std::cmp::Ordering::Equal)
});
candidates
}
pub fn detect_col_renames(
target: &SchemaDescriptor,
current: &DbState,
guidance: &Guidance,
) -> Vec<ColRenameCandidate> {
let cur_tables: HashMap<(&str, &str), &DbTable> = current
.tables
.iter()
.map(|t| ((t.schema.as_str(), t.name.as_str()), t))
.collect();
let mut candidates: Vec<ColRenameCandidate> = Vec::new();
for td in &target.types {
if td.abstract_ || td.junction {
continue;
}
let Some(cur) = cur_tables.get(&(td.module.as_str(), td.table.as_str())) else {
continue;
};
let target_cols: Vec<(String, String)> = td
.properties
.iter()
.map(|p| (p.name.clone(), col_type_str(p).to_string()))
.chain(
td.links
.iter()
.filter(|l| !l.is_junction_backed())
.map(|l| (format!("{}_id", l.name), "uuid".to_string())),
)
.collect();
let cur_cols: Vec<(&str, &str)> = cur
.columns
.iter()
.filter(|c| !c.name.starts_with("__"))
.map(|c| (c.name.as_str(), c.pg_type.as_str()))
.collect();
let dropped: Vec<(&str, &str)> = cur_cols
.iter()
.copied()
.filter(|(name, _)| !target_cols.iter().any(|(t, _)| t.as_str() == *name))
.collect();
let added: Vec<(&str, &str)> = target_cols
.iter()
.filter(|(name, _)| !cur_cols.iter().any(|&(c, _)| c == name.as_str()))
.map(|(n, t)| (n.as_str(), t.as_str()))
.collect();
if dropped.is_empty() || added.is_empty() {
continue;
}
let mut dropped_by_type: HashMap<&str, Vec<&str>> = HashMap::new();
for (name, pg_type) in &dropped {
dropped_by_type.entry(pg_type).or_default().push(name);
}
let mut added_by_type: HashMap<&str, Vec<&str>> = HashMap::new();
for (name, pg_type) in &added {
added_by_type.entry(pg_type).or_default().push(name);
}
for (pg_type, dropped_names) in &dropped_by_type {
if let Some(added_names) = added_by_type.get(pg_type)
&& dropped_names.len() == 1
&& added_names.len() == 1
{
let banned = guidance.banned_col_renames.contains(&(
td.module.clone(),
td.table.clone(),
dropped_names[0].to_string(),
added_names[0].to_string(),
));
if !banned {
candidates.push(ColRenameCandidate {
module: td.module.clone(),
table: td.table.clone(),
old_col: dropped_names[0].to_string(),
new_col: added_names[0].to_string(),
pg_type: pg_type.to_string(),
});
}
}
}
}
candidates
}
pub fn detect_fill_required(target: &SchemaDescriptor, current: &DbState) -> Vec<FillRequired> {
let cur_tables: HashMap<(&str, &str), &DbTable> = current
.tables
.iter()
.map(|t| ((t.schema.as_str(), t.name.as_str()), t))
.collect();
let mut result: Vec<FillRequired> = Vec::new();
for td in &target.types {
if td.abstract_ || td.junction {
continue;
}
let Some(cur) = cur_tables.get(&(td.module.as_str(), td.table.as_str())) else {
continue;
};
let cur_col_map: HashMap<&str, &DbColumn> = cur.columns.iter().map(|c| (c.name.as_str(), c)).collect();
for p in &td.properties {
if p.nullable || p.is_pk {
continue;
}
match cur_col_map.get(p.name.as_str()) {
None => {
if p.default_sql.is_none() {
result.push(FillRequired {
module: td.module.clone(),
table: td.table.clone(),
column: p.name.clone(),
pg_type: col_type_str(p).to_string(),
type_name: td.name.clone(),
is_new_column: true,
default_sql: None,
});
}
}
Some(cur_col) if cur_col.nullable => {
result.push(FillRequired {
module: td.module.clone(),
table: td.table.clone(),
column: p.name.clone(),
pg_type: col_type_str(p).to_string(),
type_name: td.name.clone(),
is_new_column: false,
default_sql: p.default_sql.clone(),
});
}
_ => {}
}
}
for l in &td.links {
if l.nullable || l.is_junction_backed() {
continue;
}
let col = format!("{}_id", l.name);
match cur_col_map.get(col.as_str()) {
None => {
result.push(FillRequired {
module: td.module.clone(),
table: td.table.clone(),
column: col,
pg_type: "uuid".to_string(),
type_name: td.name.clone(),
is_new_column: true,
default_sql: None,
});
}
Some(cur_col) if cur_col.nullable => {
result.push(FillRequired {
module: td.module.clone(),
table: td.table.clone(),
column: col,
pg_type: "uuid".to_string(),
type_name: td.name.clone(),
is_new_column: false,
default_sql: None,
});
}
_ => {}
}
}
}
result
}
fn apply_renames(
state: &mut DbState,
type_renames: &[(String, String, String, String)],
col_renames: &[(String, String, String, String)],
) {
for (old_mod, old_table, new_mod, new_table) in type_renames {
if let Some(t) = state
.tables
.iter_mut()
.find(|t| &t.schema == old_mod && &t.name == old_table)
{
t.schema = new_mod.clone();
t.name = new_table.clone();
}
}
for (module, table, old_col, new_col) in col_renames {
if let Some(t) = state
.tables
.iter_mut()
.find(|t| &t.schema == module && &t.name == table)
&& let Some(col) = t.columns.iter_mut().find(|c| &c.name == old_col)
{
col.name = new_col.clone();
}
}
}
pub fn diff_schema_ops_with_renames_and_fills(
target: &SchemaDescriptor,
current: &DbState,
type_renames: &[(String, String, String, String)],
col_renames: &[(String, String, String, String)],
fills: &[(String, String, String, String)],
) -> Result<Vec<DiffOp>, String> {
let mut ops: Vec<DiffOp> = Vec::new();
let mut modified = current.clone();
for (old_mod, old_table, new_mod, new_table) in type_renames {
if old_mod == new_mod {
push_tx(
&mut ops,
format!("ALTER TABLE {} RENAME TO {};", qn(old_mod, old_table), qi(new_table)),
);
} else {
push_tx(
&mut ops,
format!("ALTER TABLE {} SET SCHEMA {};", qn(old_mod, old_table), qi(new_mod)),
);
push_tx(
&mut ops,
format!("ALTER TABLE {} RENAME TO {};", qn(new_mod, old_table), qi(new_table)),
);
}
}
for (module, table, old_col, new_col) in col_renames {
push_tx(
&mut ops,
format!(
"ALTER TABLE {} RENAME COLUMN {} TO {};",
qn(module, table),
qi(old_col),
qi(new_col)
),
);
}
apply_renames(&mut modified, type_renames, col_renames);
let mut fill_index: HashMap<(String, String), HashSet<String>> = HashMap::new();
for (module, table, col, _) in fills {
fill_index
.entry((module.clone(), table.clone()))
.or_default()
.insert(col.clone());
}
let mut diff_ops = flatten_ops(diff_inner(target, &modified, true, &fill_index)?);
ops.append(&mut diff_ops);
for (module, table, col, fill_expr) in fills {
push_tx(
&mut ops,
format!(
"UPDATE {} SET {} = {} WHERE {} IS NULL;",
qn(module, table),
qi(col),
fill_expr,
qi(col)
),
);
push_tx(
&mut ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} SET NOT NULL;",
qn(module, table),
qi(col)
),
);
}
Ok(ops)
}
pub fn diff_schema_steps_with_renames_and_fills(
target: &SchemaDescriptor,
current: &DbState,
type_renames: &[(String, String, String, String)],
col_renames: &[(String, String, String, String)],
fills: &[(String, String, String, String)],
) -> Result<Vec<MigrationStep>, String> {
let mut modified = current.clone();
apply_renames(&mut modified, type_renames, col_renames);
let mut fill_index: HashMap<(String, String), HashSet<String>> = HashMap::new();
for (module, table, col, _) in fills {
fill_index
.entry((module.clone(), table.clone()))
.or_default()
.insert(col.clone());
}
let mut steps = diff_inner(target, &modified, true, &fill_index)?;
for (module, table, col, fill_expr) in fills {
let fill_ops = vec![
DiffOp {
sql: format!(
"UPDATE {} SET {} = {} WHERE {} IS NULL;",
qn(module, table),
qi(col),
fill_expr,
qi(col)
),
non_transactional: false,
},
DiffOp {
sql: format!(
"ALTER TABLE {} ALTER COLUMN {} SET NOT NULL;",
qn(module, table),
qi(col)
),
non_transactional: false,
},
];
match steps
.iter_mut()
.find(|s| matches!(&s.op_key, OpKey::Table(m, t) if m == module && t == table))
{
Some(step) => step.ddl.extend(fill_ops),
None => steps.push(MigrationStep {
prompt: format!("did you {} {}?", Verb::Alter.as_str(), verbosename_type(module, table)),
verb: Verb::Alter,
object_desc: verbosename_type(module, table),
ddl: fill_ops,
required_input: vec![],
op_key: OpKey::Table(module.clone(), table.clone()),
}),
}
}
Ok(steps)
}
fn qi(s: &str) -> String {
format!("\"{}\"", s.replace('"', "\"\""))
}
fn pg_schema(module: &str) -> String {
if module == "default" {
"\"public\"".into()
} else {
qi(module)
}
}
fn enum_target(schema: &SchemaDescriptor, pg_type: &str) -> bool {
let bare = pg_type.strip_suffix("[]").unwrap_or(pg_type);
schema
.enums
.iter()
.any(|e| bare == qn(&e.module, &e.name) || bare == qi(&e.name))
}
fn qn(schema: &str, name: &str) -> String {
format!("{}.{}", pg_schema(schema), qi(name))
}
fn topo_sort_types(types: &[TypeDescriptor], polymorphic: &HashSet<String>) -> Vec<usize> {
let idx_of: HashMap<String, usize> = types
.iter()
.enumerate()
.map(|(i, t)| (format!("{}::{}", t.module, t.name), i))
.collect();
let mut colour = vec![0u8; types.len()];
let mut order: Vec<usize> = Vec::with_capacity(types.len());
fn visit(
i: usize,
types: &[TypeDescriptor],
idx_of: &HashMap<String, usize>,
polymorphic: &HashSet<String>,
colour: &mut Vec<u8>,
order: &mut Vec<usize>,
) {
if colour[i] != 0 {
return; }
colour[i] = 1;
for l in &types[i].links {
if l.is_junction_backed() || polymorphic.contains(&l.target) {
continue;
}
if let Some(&dep) = idx_of.get(&l.target) {
visit(dep, types, idx_of, polymorphic, colour, order);
}
}
colour[i] = 2;
order.push(i);
}
for i in 0..types.len() {
visit(i, types, &idx_of, polymorphic, &mut colour, &mut order);
}
order
}
fn col_type_str(p: &crate::schema::PropertyDescriptor) -> &str {
p.column_type
.as_deref()
.unwrap_or_else(|| p.pg_type.strip_prefix("__nt__:").map(|_| "jsonb").unwrap_or(&p.pg_type))
}
fn canonical_pg_type(pg_type: &str) -> String {
if let Some(elem) = pg_type.strip_suffix("[]") {
return format!("{}[]", canonical_pg_type(elem));
}
match pg_type {
"int2" => "smallint",
"int4" => "integer",
"int8" => "bigint",
"float4" => "real",
"float8" => "double precision",
"timestamptz" => "timestamp with time zone",
"timestamp" => "timestamp without time zone",
"time" => "time without time zone",
other => other,
}
.to_string()
}
fn bare_type_name(pg_type: &str) -> &str {
pg_type.rsplit('.').next().unwrap_or(pg_type).trim_matches('"')
}
fn pg_type_changed(target: &str, current: &str) -> bool {
if target.starts_with('"') {
bare_type_name(target) != bare_type_name(current)
} else {
canonical_pg_type(target) != canonical_pg_type(current)
}
}
fn diff_inner(
target: &SchemaDescriptor,
current: &DbState,
for_migration: bool,
fill_index: &HashMap<(String, String), HashSet<String>>,
) -> Result<Vec<MigrationStep>, String> {
let mut steps = StepBuilder::new();
let cur_schemas: HashSet<&str> = current.schemas.iter().map(|s| s.as_str()).collect();
let cur_tables: HashMap<(&str, &str), &DbTable> = current
.tables
.iter()
.map(|t| ((t.schema.as_str(), t.name.as_str()), t))
.collect();
let cur_enums: HashMap<(&str, &str), &DbEnum> = current
.enums
.iter()
.map(|e| ((e.schema.as_str(), e.name.as_str()), e))
.collect();
let cur_domains: HashSet<(&str, &str)> = current
.domains
.iter()
.map(|d| (d.schema.as_str(), d.name.as_str()))
.collect();
let cur_sequences: HashSet<(&str, &str)> = current
.sequences
.iter()
.map(|s| (s.schema.as_str(), s.name.as_str()))
.collect();
let cur_views: HashMap<(&str, &str), &str> = current
.views
.iter()
.map(|v| ((v.schema.as_str(), v.name.as_str()), v.body_hash.as_str()))
.collect();
let cur_functions: HashMap<(&str, &str), &str> = current
.functions
.iter()
.map(|f| ((f.schema.as_str(), f.name.as_str()), f.body_hash.as_str()))
.collect();
let type_map: HashMap<String, (&str, &str)> = target
.types
.iter()
.map(|t| {
(
format!("{}::{}", t.module, t.name),
(t.module.as_str(), t.table.as_str()),
)
})
.collect();
let polymorphic = crate::export::polymorphic_types(target);
let mut target_schemas: HashSet<String> = HashSet::new();
for t in &target.types {
target_schemas.insert(t.module.clone());
}
for e in &target.enums {
target_schemas.insert(e.module.clone());
}
for s in &target.scalars {
target_schemas.insert(s.module.clone());
}
for f in &target.functions {
target_schemas.insert(f.module.clone());
}
for g in &target.globals {
target_schemas.insert(g.module.clone());
}
for a in &target.aliases {
target_schemas.insert(a.module.clone());
}
for module in &target_schemas {
if module == "default" {
continue;
} if !cur_schemas.contains(module.as_str()) {
steps.push(
OpKey::Module(module.clone()),
Verb::Create,
verbosename_module(module),
DiffOp {
sql: format!("CREATE SCHEMA IF NOT EXISTS {};", pg_schema(module)),
non_transactional: false,
},
);
}
}
for e in &target.enums {
match cur_enums.get(&(e.module.as_str(), e.name.as_str())) {
None => {
let members: Vec<String> = e
.members
.iter()
.map(|m| format!("'{}'", m.replace('\'', "''")))
.collect();
steps.push(
OpKey::Scalar(e.module.clone(), e.name.clone()),
Verb::Create,
verbosename_scalar(&e.module, &e.name),
DiffOp {
sql: format!(
"DO $$ BEGIN CREATE TYPE {}.{} AS ENUM ({}); \
EXCEPTION WHEN duplicate_object THEN NULL; END $$;",
pg_schema(&e.module),
qi(&e.name),
members.join(", ")
),
non_transactional: false,
},
);
}
Some(existing) => {
let existing_set: HashSet<&str> = existing.members.iter().map(|m| m.as_str()).collect();
for member in &e.members {
if !existing_set.contains(member.as_str()) {
steps.push(
OpKey::Scalar(e.module.clone(), e.name.clone()),
Verb::Alter,
verbosename_scalar(&e.module, &e.name),
DiffOp {
sql: format!(
"ALTER TYPE {}.{} ADD VALUE IF NOT EXISTS '{}';",
pg_schema(&e.module),
qi(&e.name),
member.replace('\'', "''")
),
non_transactional: false,
},
);
}
}
}
}
}
for s in &target.scalars {
if s.is_sequence {
let seq_name = format!("{}_seq", s.name);
if !cur_sequences.contains(&(s.module.as_str(), seq_name.as_str())) {
let verb = if cur_domains.contains(&(s.module.as_str(), s.name.as_str())) {
Verb::Alter
} else {
Verb::Create
};
steps.push(
OpKey::Scalar(s.module.clone(), s.name.clone()),
verb,
verbosename_scalar(&s.module, &s.name),
DiffOp {
sql: format!(
"CREATE SEQUENCE IF NOT EXISTS {}.{};",
pg_schema(&s.module),
qi(&seq_name)
),
non_transactional: false,
},
);
}
}
}
for s in &target.scalars {
if !cur_domains.contains(&(s.module.as_str(), s.name.as_str())) {
let check_clause = crate::export::scalar_check_clauses(target, &s.module, &s.name);
steps.push(
OpKey::Scalar(s.module.clone(), s.name.clone()),
Verb::Create,
verbosename_scalar(&s.module, &s.name),
DiffOp {
sql: format!(
"DO $do$ BEGIN CREATE DOMAIN {}.{} AS {}{}; \
EXCEPTION WHEN duplicate_object THEN NULL; END $do$;",
pg_schema(&s.module),
qi(&s.name),
s.pg_type,
check_clause
),
non_transactional: false,
},
);
}
}
let expected_domain_checks = crate::export::scalar_check_constraints(target);
for cur_domain in ¤t.domains {
if !target
.scalars
.iter()
.any(|s| s.module == cur_domain.schema && s.name == cur_domain.name)
{
continue; }
let wanted: Vec<&(String, String, String, String)> = expected_domain_checks
.iter()
.filter(|(module, name, _, _)| *module == cur_domain.schema && *name == cur_domain.name)
.collect();
let domain = format!("{}.{}", pg_schema(&cur_domain.schema), qi(&cur_domain.name));
let unchanged = wanted.len() == cur_domain.checks.len()
&& wanted
.iter()
.all(|(_, _, cname, _)| cur_domain.checks.iter().any(|name| name == cname));
if unchanged {
continue;
}
let mut local: Vec<DiffOp> = Vec::new();
push_tx(
&mut local,
format!(
"DO $do$ DECLARE existing record; BEGIN FOR existing IN SELECT conname FROM pg_constraint WHERE contypid = '{domain}'::regtype AND contype = 'c' LOOP EXECUTE format('ALTER DOMAIN {domain} DROP CONSTRAINT %I', existing.conname); END LOOP; END $do$;"
),
);
for (_, _, cname, expr) in wanted {
push_tx(
&mut local,
format!("ALTER DOMAIN {} ADD CONSTRAINT {} CHECK ({});", domain, qi(cname), expr),
);
}
if !local.is_empty() {
steps.extend(
OpKey::Scalar(cur_domain.schema.clone(), cur_domain.name.clone()),
Verb::Alter,
verbosename_scalar(&cur_domain.schema, &cur_domain.name),
local,
);
}
}
let scalar_fn_ddls = crate::export::scalar_function_ddl_with_names(target).map_err(|e| e.to_string())?;
for (module, name, ddl) in scalar_fn_ddls {
let emit = if for_migration {
let hash = ddl_hash(&ddl);
cur_functions
.get(&(module.as_str(), name.as_str()))
.map(|&h| h != hash)
.unwrap_or(true)
} else {
true
};
if emit {
let verb = if cur_functions.contains_key(&(module.as_str(), name.as_str())) {
Verb::Alter
} else {
Verb::Create
};
steps.push(
OpKey::Function(module.clone(), name.clone()),
verb,
verbosename_function(&module, &name),
DiffOp {
sql: ddl,
non_transactional: false,
},
);
}
}
let sort_order = topo_sort_types(&target.types, &polymorphic);
let mut new_tables: HashSet<(String, String)> = HashSet::new();
for &i in &sort_order {
let td = &target.types[i];
if td.abstract_ || td.junction {
continue;
}
let key = (td.module.as_str(), td.table.as_str());
match cur_tables.get(&key) {
None => {
let mut local: Vec<DiffOp> = Vec::new();
emit_create_table(td, target, &mut local);
steps.extend(
OpKey::Table(td.module.clone(), td.table.clone()),
Verb::Create,
verbosename_type(&td.module, &td.name),
local,
);
new_tables.insert((td.module.clone(), td.table.clone()));
}
Some(existing) => {
let fill_cols = fill_index
.get(&(td.module.clone(), td.table.clone()))
.cloned()
.unwrap_or_default();
let mut local: Vec<DiffOp> = Vec::new();
let mut inputs: Vec<RequiredInput> = Vec::new();
emit_column_diff(td, existing, &mut local, for_migration, &fill_cols, target, &mut inputs);
steps.extend_with_input(
OpKey::Table(td.module.clone(), td.table.clone()),
Verb::Alter,
verbosename_type(&td.module, &td.name),
local,
inputs,
);
}
}
}
for &i in &sort_order {
let td = &target.types[i];
if td.abstract_ || td.junction {
continue;
}
let existing = cur_tables.get(&(td.module.as_str(), td.table.as_str())).copied();
let verb = if new_tables.contains(&(td.module.clone(), td.table.clone())) {
Verb::Create
} else {
Verb::Alter
};
let mut local: Vec<DiffOp> = Vec::new();
emit_fk_diff(td, existing, &type_map, &polymorphic, &mut local);
steps.extend(
OpKey::ForeignKey(td.module.clone(), td.table.clone()),
verb,
verbosename_type(&td.module, &td.name),
local,
);
}
for (jt_module, jt_name, cname, ddl) in crate::export::junction_fk_constraints(target, &type_map) {
let already_there = cur_tables
.get(&(jt_module.as_str(), jt_name.as_str()))
.map(|t| t.foreign_keys.iter().any(|fk| fk.constraint_name == cname))
.unwrap_or(false);
if already_there {
continue;
}
let verb = if cur_tables.contains_key(&(jt_module.as_str(), jt_name.as_str())) {
Verb::Alter
} else {
Verb::Create
};
let mut local: Vec<DiffOp> = Vec::new();
push_tx(&mut local, ddl);
steps.extend(
OpKey::ForeignKey(jt_module.clone(), jt_name.clone()),
verb,
format!("link table '{}.{}'", jt_module, jt_name),
local,
);
}
for (module, table, cname, ddl) in crate::export::check_constraints(target).map_err(|e| e.to_string())? {
let already_there = cur_tables
.get(&(module.as_str(), table.as_str()))
.map(|t| t.checks.iter().any(|c| c.constraint_name == cname))
.unwrap_or(false);
if already_there {
continue;
}
let verb = if cur_tables.contains_key(&(module.as_str(), table.as_str())) {
Verb::Alter
} else {
Verb::Create
};
let mut local: Vec<DiffOp> = Vec::new();
push_tx(&mut local, ddl);
steps.extend(
OpKey::ForeignKey(module.clone(), format!("{table}#checks")),
verb,
verbosename_type(&module, &table),
local,
);
}
for &i in &sort_order {
let td = &target.types[i];
if td.abstract_ || td.junction {
continue;
}
let owner_key = OpKey::Table(td.module.clone(), td.table.clone());
let owner_verb = if new_tables.contains(&(td.module.clone(), td.table.clone())) {
Verb::Create
} else {
Verb::Alter
};
let owner_desc = verbosename_type(&td.module, &td.name);
for ml in &td.multilinks {
let jt = format!("{}.{}", td.table, ml.name);
if !cur_tables.contains_key(&(td.module.as_str(), jt.as_str())) {
let mut local: Vec<DiffOp> = Vec::new();
emit_junction_table(td, &ml.name, ml.through.as_deref(), target, false, &mut local);
steps.extend(owner_key.clone(), owner_verb, owner_desc.clone(), local);
new_tables.insert((td.module.clone(), jt));
}
}
for l in &td.links {
if !l.is_junction_backed() {
continue;
}
let jt = format!("{}.{}", td.table, l.name);
if !cur_tables.contains_key(&(td.module.as_str(), jt.as_str())) {
let mut local: Vec<DiffOp> = Vec::new();
emit_junction_table(td, &l.name, l.through.as_deref(), target, true, &mut local);
steps.extend(owner_key.clone(), owner_verb, owner_desc.clone(), local);
new_tables.insert((td.module.clone(), jt));
}
}
}
for &i in &sort_order {
let td = &target.types[i];
if td.abstract_ || td.vector_indexes.is_empty() {
continue;
}
let existing = cur_tables.get(&(td.module.as_str(), td.table.as_str()));
let table_is_new = new_tables.contains(&(td.module.clone(), td.table.clone()));
let owner_key = OpKey::Table(td.module.clone(), td.table.clone());
let owner_verb = if table_is_new { Verb::Create } else { Verb::Alter };
let owner_desc = verbosename_type(&td.module, &td.name);
for vi in &td.vector_indexes {
let col = vi.column_name();
if existing
.map(|t| t.columns.iter().any(|c| c.name == col))
.unwrap_or(false)
{
continue;
}
let mut local: Vec<DiffOp> = Vec::new();
push_tx(
&mut local,
format!(
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} vector({});",
qn(&td.module, &td.table),
qi(&col),
vi.dimensions
),
);
let idx_name = match &vi.index_name {
None => format!("{}__vector__", td.table),
Some(n) => format!("{}__vector_{}__", td.table, n),
};
let use_concurrently = for_migration && !table_is_new;
let idx_sql = if use_concurrently {
format!(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS {} ON {} USING hnsw ({} {});",
qi(&idx_name),
qn(&td.module, &td.table),
qi(&col),
vi.ops_class()
)
} else {
format!(
"CREATE INDEX IF NOT EXISTS {} ON {} USING hnsw ({} {});",
qi(&idx_name),
qn(&td.module, &td.table),
qi(&col),
vi.ops_class()
)
};
local.push(DiffOp {
sql: idx_sql,
non_transactional: use_concurrently,
});
steps.extend(owner_key.clone(), owner_verb, owner_desc.clone(), local);
}
}
for &i in &sort_order {
let td = &target.types[i];
if td.abstract_ || td.search_indexes.is_empty() {
continue;
}
let existing = cur_tables.get(&(td.module.as_str(), td.table.as_str()));
let table_is_new = new_tables.contains(&(td.module.clone(), td.table.clone()));
let owner_key = OpKey::Table(td.module.clone(), td.table.clone());
let owner_verb = if table_is_new { Verb::Create } else { Verb::Alter };
let owner_desc = verbosename_type(&td.module, &td.name);
for si in &td.search_indexes {
if si.backend != SearchBackend::Postgres {
continue;
}
let col = si.column_name();
if existing
.map(|t| t.columns.iter().any(|c| c.name == col))
.unwrap_or(false)
{
continue;
}
let mut local: Vec<DiffOp> = Vec::new();
let parts: Vec<String> = si
.pointers
.iter()
.map(|sf| {
format!(
"setweight(to_tsvector('english', coalesce({}, '')), '{}')",
qi(&sf.name),
sf.weight.as_str()
)
})
.collect();
let expr = if parts.len() == 1 {
parts.into_iter().next().unwrap()
} else {
parts.join(" || ")
};
push_tx(
&mut local,
format!(
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} tsvector GENERATED ALWAYS AS ({}) STORED;",
qn(&td.module, &td.table),
qi(&col),
expr
),
);
let idx_name = match &si.index_name {
None => format!("{}__search__", td.table),
Some(n) => format!("{}__search_{}__", td.table, n),
};
let use_concurrently = for_migration && !table_is_new;
let idx_sql = if use_concurrently {
format!(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS {} ON {} USING gin ({});",
qi(&idx_name),
qn(&td.module, &td.table),
qi(&col)
)
} else {
format!(
"CREATE INDEX IF NOT EXISTS {} ON {} USING gin ({});",
qi(&idx_name),
qn(&td.module, &td.table),
qi(&col)
)
};
local.push(DiffOp {
sql: idx_sql,
non_transactional: use_concurrently,
});
steps.extend(owner_key.clone(), owner_verb, owner_desc.clone(), local);
}
}
let expected_state = schema_to_db_state(target);
let current_indexes: HashMap<(&str, &str), &[DbIndex]> = current
.tables
.iter()
.map(|t| ((t.schema.as_str(), t.name.as_str()), t.indexes.as_slice()))
.collect();
for table in &expected_state.tables {
let existing = current_indexes
.get(&(table.schema.as_str(), table.name.as_str()))
.copied()
.unwrap_or_default();
let table_is_new = !current_indexes.contains_key(&(table.schema.as_str(), table.name.as_str()));
for idx in &table.indexes {
if idx.columns.is_empty() && idx.key.is_none() {
continue;
}
let already_there = match &idx.key {
Some(_) => existing.iter().any(|b| b.name == idx.name),
None => existing.iter().any(|b| {
b.columns == idx.columns
&& b.is_unique == idx.is_unique
&& b.predicate.is_some() == idx.predicate.is_some()
}),
};
if already_there {
continue;
}
let use_concurrently = for_migration && !table_is_new;
let concurrently = if use_concurrently { "CONCURRENTLY " } else { "" };
let unique = if idx.is_unique { "UNIQUE " } else { "" };
let body = match &idx.key {
Some(key) => key.clone(),
None => format!("({})", idx.columns.iter().map(|c| qi(c)).collect::<Vec<_>>().join(", ")),
};
let sql = format!(
"CREATE {unique}INDEX {concurrently}IF NOT EXISTS {} ON {} {}{};",
qi(&capped_index_name(idx.name.clone())),
qn(&table.schema, &table.name),
body,
idx.predicate.clone().unwrap_or_default(),
);
let owner_key = OpKey::Table(table.schema.clone(), table.name.clone());
let owner_verb = if table_is_new { Verb::Create } else { Verb::Alter };
let owner_desc = verbosename_type(&table.schema, &table.name);
steps.extend(
owner_key,
owner_verb,
owner_desc,
vec![DiffOp {
sql,
non_transactional: use_concurrently,
}],
);
}
}
for (module, name, ddl) in crate::export::interface_view_ddl_with_names(target) {
let emit = if for_migration {
let hash = ddl_hash(&ddl);
cur_views
.get(&(module.as_str(), name.as_str()))
.map(|&h| h != hash)
.unwrap_or(true)
} else {
true
};
if emit {
let verb = if cur_views.contains_key(&(module.as_str(), name.as_str())) {
Verb::Alter
} else {
Verb::Create
};
steps.push(
OpKey::View(module.clone(), name.clone()),
verb,
verbosename_interface(&module, &name),
DiffOp {
sql: ddl,
non_transactional: false,
},
);
}
}
for (module, name, ddl) in crate::export::interface_junction_view_ddl_with_names(target) {
let emit = if for_migration {
let hash = ddl_hash(&ddl);
cur_views
.get(&(module.as_str(), name.as_str()))
.map(|&h| h != hash)
.unwrap_or(true)
} else {
true
};
if emit {
let verb = if cur_views.contains_key(&(module.as_str(), name.as_str())) {
Verb::Alter
} else {
Verb::Create
};
steps.push(
OpKey::View(module.clone(), name.clone()),
verb,
verbosename_interface(&module, &name),
DiffOp {
sql: ddl,
non_transactional: false,
},
);
}
}
let target_views: HashSet<(String, String)> = crate::export::interface_view_ddl_with_names(target)
.into_iter()
.chain(crate::export::interface_junction_view_ddl_with_names(target))
.map(|(module, name, _)| (module, name))
.collect();
for view in ¤t.views {
if !target_views.contains(&(view.schema.clone(), view.name.clone())) {
steps.push(
OpKey::View(view.schema.clone(), view.name.clone()),
Verb::Drop,
verbosename_interface(&view.schema, &view.name),
DiffOp {
sql: format!("DROP VIEW IF EXISTS {};", qn(&view.schema, &view.name)),
non_transactional: false,
},
);
}
}
let obj_fn_ddls = crate::export::object_function_ddl_with_names(target).map_err(|e| e.to_string())?;
for (module, name, ddl) in obj_fn_ddls {
let emit = if for_migration {
let hash = ddl_hash(&ddl);
cur_functions
.get(&(module.as_str(), name.as_str()))
.map(|&h| h != hash)
.unwrap_or(true)
} else {
true
};
if emit {
let verb = if cur_functions.contains_key(&(module.as_str(), name.as_str())) {
Verb::Alter
} else {
Verb::Create
};
steps.push(
OpKey::Function(module.clone(), name.clone()),
verb,
verbosename_function(&module, &name),
DiffOp {
sql: ddl,
non_transactional: false,
},
);
}
}
let mut target_tables: HashSet<(String, String)> = HashSet::new();
for td in &target.types {
if !td.abstract_ {
target_tables.insert((td.module.clone(), td.table.clone()));
}
if !td.abstract_ && !td.junction {
for ml in &td.multilinks {
target_tables.insert((td.module.clone(), format!("{}.{}", td.table, ml.name)));
}
for l in &td.links {
if !l.is_junction_backed() {
continue;
}
target_tables.insert((td.module.clone(), format!("{}.{}", td.table, l.name)));
}
}
}
{
let infos = crate::export::interface_exclusive_trigger_infos(target);
let cur_trigger_map: HashMap<(&str, &str), HashSet<&str>> = current
.tables
.iter()
.map(|t| {
(
(t.schema.as_str(), t.name.as_str()),
t.triggers.iter().map(|n| n.as_str()).collect::<HashSet<_>>(),
)
})
.collect();
let expected_trigger_map = expected_triggers(target, &type_map);
let owner_of = |module: &str, table: &str| -> (OpKey, Verb, String) {
for &i in &sort_order {
let td = &target.types[i];
if td.abstract_ || td.junction {
continue;
}
let is_owner = td.module == module
&& (td.table == table
|| td
.multilinks
.iter()
.any(|ml| format!("{}.{}", td.table, ml.name) == table)
|| td
.links
.iter()
.any(|l| l.is_junction_backed() && format!("{}.{}", td.table, l.name) == table));
if is_owner {
let verb = if new_tables.contains(&(td.module.clone(), td.table.clone())) {
Verb::Create
} else {
Verb::Alter
};
return (
OpKey::Table(td.module.clone(), td.table.clone()),
verb,
verbosename_type(&td.module, &td.name),
);
}
}
(
OpKey::Table(module.to_string(), table.to_string()),
Verb::Alter,
verbosename_type(module, table),
)
};
struct PendingExclTrigger<'a> {
info: &'a crate::export::ExclTriggerInfo,
need_ins: bool,
need_upd: bool,
owner: (OpKey, Verb, String),
}
let pending: Vec<PendingExclTrigger<'_>> = infos
.iter()
.filter_map(|info| {
let cur = cur_trigger_map
.get(&(info.impl_module.as_str(), info.impl_table.as_str()))
.cloned()
.unwrap_or_default();
let need_ins = !cur.contains(info.ins_trigger_name.as_str());
let need_upd = !cur.contains(info.upd_trigger_name.as_str());
(need_ins || need_upd).then(|| PendingExclTrigger {
info,
need_ins,
need_upd,
owner: owner_of(&info.impl_module, &info.impl_table),
})
})
.collect();
let mut fn_carrier: HashMap<String, usize> = HashMap::new();
for (index, entry) in pending.iter().enumerate() {
let position = steps.position(&entry.owner.0).unwrap_or(usize::MAX);
let beats_carrier = match fn_carrier.get(&entry.info.fn_name) {
Some(&carrier) => position < steps.position(&pending[carrier].owner.0).unwrap_or(usize::MAX),
None => true,
};
if beats_carrier {
fn_carrier.insert(entry.info.fn_name.clone(), index);
}
}
for (index, entry) in pending.into_iter().enumerate() {
let mut local: Vec<DiffOp> = Vec::new();
if fn_carrier.get(&entry.info.fn_name) == Some(&index) {
push_tx(&mut local, entry.info.fn_ddl.clone());
}
if entry.need_ins {
push_tx(&mut local, entry.info.ins_ddl.clone());
}
if entry.need_upd {
push_tx(&mut local, entry.info.upd_ddl.clone());
}
let (key, verb, desc) = entry.owner;
steps.extend(key, verb, desc, local);
}
for info in crate::export::deletion_policy_trigger_infos(target, &type_map) {
let cur = cur_trigger_map
.get(&(info.table_module.as_str(), info.table_name.as_str()))
.cloned()
.unwrap_or_default();
if !cur.contains(info.trigger_name.as_str()) {
let (key, verb, desc) = owner_of(&info.table_module, &info.table_name);
steps.extend(
key,
verb,
desc,
vec![DiffOp {
sql: info.ddl.clone(),
non_transactional: false,
}],
);
}
}
for info in crate::export::signal_trigger_infos(target) {
let cur = cur_trigger_map
.get(&(info.table_module.as_str(), info.table_name.as_str()))
.cloned()
.unwrap_or_default();
if !cur.contains(info.trigger_name.as_str()) {
let (key, verb, desc) = owner_of(&info.table_module, &info.table_name);
steps.extend(
key,
verb,
desc,
vec![DiffOp {
sql: info.ddl.clone(),
non_transactional: false,
}],
);
}
}
for info in crate::export::user_trigger_infos(target).map_err(|e| e.to_string())? {
let cur = cur_trigger_map
.get(&(info.table_module.as_str(), info.table_name.as_str()))
.cloned()
.unwrap_or_default();
if !cur.contains(info.trigger_name.as_str()) {
let (key, verb, desc) = owner_of(&info.table_module, &info.table_name);
steps.extend(
key,
verb,
desc,
vec![DiffOp {
sql: info.ddl.clone(),
non_transactional: false,
}],
);
}
}
let mut cache_trigger_tables: HashSet<(String, String)> = HashSet::new();
for td in &target.types {
if td.abstract_ || junction_table_belongs_to_an_interface(target, td) {
continue;
}
cache_trigger_tables.insert((td.module.clone(), td.table.clone()));
if !td.junction {
for ml in &td.multilinks {
cache_trigger_tables.insert((td.module.clone(), format!("{}.{}", td.table, ml.name)));
}
for l in &td.links {
if !l.is_junction_backed() {
continue;
}
cache_trigger_tables.insert((td.module.clone(), format!("{}.{}", td.table, l.name)));
}
}
}
for (module, table) in &cache_trigger_tables {
let already_present = cur_trigger_map
.get(&(module.as_str(), table.as_str()))
.map(|t| t.contains("pylon_cache_invalidate"))
.unwrap_or(false);
if !already_present {
let (key, verb, desc) = owner_of(module, table);
steps.extend(
key,
verb,
desc,
vec![DiffOp {
sql: cache_invalidate_trigger_sql(&qn(module, table)),
non_transactional: false,
}],
);
}
}
for cur_table in ¤t.tables {
let key = (cur_table.schema.clone(), cur_table.name.clone());
if !target_tables.contains(&key) {
continue;
}
let expected = expected_trigger_map.get(&key).cloned().unwrap_or_default();
for trigger_name in &cur_table.triggers {
if !expected.contains(trigger_name) {
let (owner_key, verb, desc) = owner_of(&cur_table.schema, &cur_table.name);
steps.extend(
owner_key,
verb,
desc,
vec![
DiffOp {
sql: format!(
"DROP TRIGGER IF EXISTS {} ON {};",
qi(trigger_name),
qn(&cur_table.schema, &cur_table.name)
),
non_transactional: false,
},
DiffOp {
sql: format!("DROP FUNCTION IF EXISTS {}();", qn(&cur_table.schema, trigger_name)),
non_transactional: false,
},
],
);
}
}
}
}
let expected_checks: HashSet<(String, String, String)> = crate::export::check_constraints(target)
.map_err(|e| e.to_string())?
.into_iter()
.map(|(module, table, name, _)| (module, table, name))
.collect();
for cur_table in ¤t.tables {
if !target_tables.contains(&(cur_table.schema.clone(), cur_table.name.clone())) {
continue; }
for check in &cur_table.checks {
if !is_generated_check_name(&check.constraint_name) {
continue;
}
let key = (
cur_table.schema.clone(),
cur_table.name.clone(),
check.constraint_name.clone(),
);
if expected_checks.contains(&key) {
continue;
}
steps.extend(
OpKey::ForeignKey(cur_table.schema.clone(), format!("{}#checks", cur_table.name)),
Verb::Alter,
verbosename_type(&cur_table.schema, &cur_table.name),
vec![DiffOp {
sql: format!(
"ALTER TABLE {} DROP CONSTRAINT IF EXISTS {};",
qn(&cur_table.schema, &cur_table.name),
qi(&check.constraint_name)
),
non_transactional: false,
}],
);
}
}
let polymorphic_targets = crate::export::polymorphic_types(target);
let mut unwanted_fks: HashSet<(String, String, String)> = HashSet::new();
for t in target.types.iter().filter(|t| !t.abstract_ && !t.junction) {
for l in &t.links {
if !polymorphic_targets.contains(&l.target) {
continue;
}
unwanted_fks.insert(if l.is_junction_backed() {
(
t.module.clone(),
format!("{}.{}", t.table, l.name),
format!("{}_{}_target_fkey", t.table, l.name),
)
} else {
(
t.module.clone(),
t.table.clone(),
format!("{}_{}_fkey", t.table, l.name),
)
});
}
for ml in t
.multilinks
.iter()
.filter(|ml| polymorphic_targets.contains(&ml.target))
{
unwanted_fks.insert((
t.module.clone(),
format!("{}.{}", t.table, ml.name),
format!("{}_{}_target_fkey", t.table, ml.name),
));
}
}
for cur_table in ¤t.tables {
for fk in &cur_table.foreign_keys {
let key = (
cur_table.schema.clone(),
cur_table.name.clone(),
fk.constraint_name.clone(),
);
let references_a_table = target
.types
.iter()
.any(|t| !t.abstract_ && t.module == fk.ref_schema && t.table == fk.ref_table);
if !unwanted_fks.contains(&key) || !references_a_table {
continue;
}
steps.extend(
OpKey::ForeignKey(cur_table.schema.clone(), cur_table.name.clone()),
Verb::Alter,
verbosename_type(&cur_table.schema, &cur_table.name),
vec![DiffOp {
sql: format!(
"ALTER TABLE {} DROP CONSTRAINT IF EXISTS {};",
qn(&cur_table.schema, &cur_table.name),
qi(&fk.constraint_name)
),
non_transactional: false,
}],
);
}
}
for cur_table in ¤t.tables {
let key = (cur_table.schema.clone(), cur_table.name.clone());
if !target_tables.contains(&key) {
steps.push(
OpKey::Table(cur_table.schema.clone(), cur_table.name.clone()),
Verb::Drop,
verbosename_type(&cur_table.schema, &cur_table.name),
DiffOp {
sql: format!(
"DROP TABLE IF EXISTS {} CASCADE;",
qn(&cur_table.schema, &cur_table.name)
),
non_transactional: false,
},
);
}
}
let target_enum_set: HashSet<(String, String)> = target
.enums
.iter()
.map(|e| (e.module.clone(), e.name.clone()))
.collect();
for cur_enum in ¤t.enums {
if !target_enum_set.contains(&(cur_enum.schema.clone(), cur_enum.name.clone())) {
steps.push(
OpKey::Scalar(cur_enum.schema.clone(), cur_enum.name.clone()),
Verb::Drop,
verbosename_scalar(&cur_enum.schema, &cur_enum.name),
DiffOp {
sql: format!(
"DROP TYPE IF EXISTS {}.{} CASCADE;",
pg_schema(&cur_enum.schema),
qi(&cur_enum.name)
),
non_transactional: false,
},
);
}
}
let target_domain_set: HashSet<(String, String)> = target
.scalars
.iter()
.map(|s| (s.module.clone(), s.name.clone()))
.collect();
for cur_domain in ¤t.domains {
if !target_domain_set.contains(&(cur_domain.schema.clone(), cur_domain.name.clone())) {
steps.push(
OpKey::Scalar(cur_domain.schema.clone(), cur_domain.name.clone()),
Verb::Drop,
verbosename_scalar(&cur_domain.schema, &cur_domain.name),
DiffOp {
sql: format!(
"DROP DOMAIN IF EXISTS {}.{} CASCADE;",
pg_schema(&cur_domain.schema),
qi(&cur_domain.name)
),
non_transactional: false,
},
);
}
}
let target_sequence_set: HashSet<(String, String)> = target
.scalars
.iter()
.filter(|s| s.is_sequence)
.map(|s| (s.module.clone(), format!("{}_seq", s.name)))
.collect();
for cur_seq in ¤t.sequences {
if !target_sequence_set.contains(&(cur_seq.schema.clone(), cur_seq.name.clone())) {
let scalar_name = cur_seq.name.strip_suffix("_seq").unwrap_or(&cur_seq.name).to_string();
steps.push(
OpKey::Scalar(cur_seq.schema.clone(), scalar_name.clone()),
Verb::Drop,
verbosename_scalar(&cur_seq.schema, &scalar_name),
DiffOp {
sql: format!(
"DROP SEQUENCE IF EXISTS {}.{};",
pg_schema(&cur_seq.schema),
qi(&cur_seq.name)
),
non_transactional: false,
},
);
}
}
for module in ¤t.schemas {
if module == "default" {
continue;
} if !target_schemas.contains(module) {
steps.push(
OpKey::Module(module.clone()),
Verb::Drop,
verbosename_module(module),
DiffOp {
sql: format!("DROP SCHEMA IF EXISTS {} CASCADE;", pg_schema(module)),
non_transactional: false,
},
);
}
}
Ok(steps.finish())
}
fn push_tx(ops: &mut Vec<DiffOp>, sql: String) {
ops.push(DiffOp {
sql,
non_transactional: false,
});
}
fn resolve_default(p: &crate::schema::PropertyDescriptor, schema: &SchemaDescriptor) -> Option<String> {
if let Some(sql) = &p.default_sql {
return Some(sql.clone());
}
if let Some(pyql) = &p.default_pyql {
return crate::ir::column_default_sql(pyql, schema);
}
None
}
#[cfg(test)]
pub(crate) fn resolve_default_for_test(
p: &crate::schema::PropertyDescriptor,
schema: &SchemaDescriptor,
) -> Option<String> {
resolve_default(p, schema)
}
fn resolve_link_default(l: &crate::schema::LinkDescriptor, schema: &SchemaDescriptor) -> Option<String> {
if let Some(pyql) = &l.default_pyql {
return crate::ir::column_default_sql(pyql, schema);
}
None
}
fn emit_create_table(td: &TypeDescriptor, schema: &SchemaDescriptor, ops: &mut Vec<DiffOp>) {
let mut lines: Vec<String> = Vec::new();
for p in &td.properties {
let not_null = if p.nullable { "" } else { " NOT NULL" };
let default = resolve_default(p, schema)
.map(|d| format!(" DEFAULT {}", d))
.unwrap_or_default();
lines.push(format!(
" {} {}{}{}",
qi(&p.name),
col_type_str(p),
not_null,
default
));
}
for l in &td.links {
if l.is_junction_backed() {
continue;
}
let not_null = if l.nullable { "" } else { " NOT NULL" };
let default = resolve_link_default(l, schema)
.map(|d| format!(" DEFAULT {}", d))
.unwrap_or_default();
lines.push(format!(
" {} uuid{}{}",
qi(&format!("{}_id", l.name)),
not_null,
default
));
}
let pk_cols: Vec<String> = td.properties.iter().filter(|p| p.is_pk).map(|p| qi(&p.name)).collect();
if !pk_cols.is_empty() {
lines.push(format!(" PRIMARY KEY ({})", pk_cols.join(", ")));
}
push_tx(
ops,
format!(
"CREATE TABLE IF NOT EXISTS {} (\n{}\n);",
qn(&td.module, &td.table),
lines.join(",\n")
),
);
}
fn cache_invalidate_trigger_sql(qualified_table: &str) -> String {
format!(
"CREATE OR REPLACE TRIGGER pylon_cache_invalidate\n AFTER INSERT OR UPDATE OR DELETE ON {}\n FOR EACH STATEMENT EXECUTE FUNCTION _pylon.notify_cache_invalidate();",
qualified_table
)
}
fn emit_column_diff(
td: &TypeDescriptor,
existing: &DbTable,
ops: &mut Vec<DiffOp>,
for_migration: bool,
fill_cols: &HashSet<String>,
schema: &SchemaDescriptor,
required_input: &mut Vec<RequiredInput>,
) {
let existing_col_map: HashMap<&str, &DbColumn> = existing.columns.iter().map(|c| (c.name.as_str(), c)).collect();
for p in &td.properties {
if existing_col_map.contains_key(p.name.as_str()) {
continue;
}
let eff_default = resolve_default(p, schema);
let needs_fill = for_migration && !p.nullable && eff_default.is_none() && fill_cols.contains(&p.name);
let not_null = if p.nullable || needs_fill { "" } else { " NOT NULL" };
let default = eff_default.map(|d| format!(" DEFAULT {}", d)).unwrap_or_default();
push_tx(
ops,
format!(
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {}{}{};",
qn(&td.module, &td.table),
qi(&p.name),
col_type_str(p),
not_null,
default
),
);
}
for l in &td.links {
if l.is_junction_backed() {
continue;
}
let col = format!("{}_id", l.name);
if existing_col_map.contains_key(col.as_str()) {
continue;
}
let eff_default = resolve_link_default(l, schema);
let needs_fill = for_migration && !l.nullable && eff_default.is_none() && fill_cols.contains(&col);
let not_null = if l.nullable || needs_fill { "" } else { " NOT NULL" };
let default = eff_default.map(|d| format!(" DEFAULT {}", d)).unwrap_or_default();
push_tx(
ops,
format!(
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} uuid{}{};",
qn(&td.module, &td.table),
qi(&col),
not_null,
default
),
);
}
let type_changes: Vec<(&str, &str, bool)> = td
.properties
.iter()
.filter_map(|p| {
let cur = existing_col_map.get(p.name.as_str())?;
if cur.is_generated {
return None;
}
let target_type = col_type_str(p);
pg_type_changed(target_type, &cur.pg_type).then_some((
p.name.as_str(),
target_type,
cur.column_default.is_some(),
))
})
.collect();
if !type_changes.is_empty() {
let affected_views: Vec<(String, String, String)> = crate::export::interface_view_ddl_with_names(schema)
.into_iter()
.filter(|(m, n, _)| td.interfaces.contains(&format!("{}::{}", m, n)))
.collect();
for (m, n, _) in &affected_views {
push_tx(ops, format!("DROP VIEW IF EXISTS {};", qn(m, n)));
}
for (col, target_type, has_default) in &type_changes {
if *has_default {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} DROP DEFAULT;",
qn(&td.module, &td.table),
qi(col)
),
);
}
let placeholder = format!("cast_expr__{col}");
let default_expr = if enum_target(schema, target_type) {
let text_type = if target_type.ends_with("[]") { "text[]" } else { "text" };
format!("{}::{text_type}::{target_type}", qi(col))
} else {
format!("{}::{target_type}", qi(col))
};
required_input.push(RequiredInput {
placeholder: placeholder.clone(),
prompt: format!(
"Please specify a conversion expression to alter the type of property '{col}' of {}",
verbosename_type(&td.module, &td.name),
),
default_expr,
type_name: format!("{}::{}", td.module, td.name),
});
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} TYPE {} USING \\({});",
qn(&td.module, &td.table),
qi(col),
target_type,
placeholder
),
);
}
for (_, _, ddl) in &affected_views {
push_tx(ops, ddl.clone());
}
}
for p in &td.properties {
let Some(cur) = existing_col_map.get(p.name.as_str()) else {
continue;
};
if cur.is_generated {
continue;
}
if !cur.nullable && p.nullable {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} DROP NOT NULL;",
qn(&td.module, &td.table),
qi(&p.name)
),
);
} else if cur.nullable && !p.nullable && !for_migration {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} SET NOT NULL;",
qn(&td.module, &td.table),
qi(&p.name)
),
);
}
let target_default = resolve_default(p, schema);
let db_default = cur.column_default.as_deref();
match (&target_default, db_default) {
(Some(want), Some(have)) if want != have => {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} SET DEFAULT {};",
qn(&td.module, &td.table),
qi(&p.name),
want
),
);
}
(Some(want), None) => {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} SET DEFAULT {};",
qn(&td.module, &td.table),
qi(&p.name),
want
),
);
}
(None, Some(_)) => {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} DROP DEFAULT;",
qn(&td.module, &td.table),
qi(&p.name)
),
);
}
_ => {}
}
}
for l in &td.links {
if l.is_junction_backed() {
continue;
}
let col = format!("{}_id", l.name);
let Some(cur) = existing_col_map.get(col.as_str()) else {
continue;
};
if !cur.nullable && l.nullable {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} DROP NOT NULL;",
qn(&td.module, &td.table),
qi(&col)
),
);
} else if cur.nullable && !l.nullable && !for_migration {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} SET NOT NULL;",
qn(&td.module, &td.table),
qi(&col)
),
);
}
let target_default = resolve_link_default(l, schema);
let db_default = cur.column_default.as_deref();
match (&target_default, db_default) {
(Some(want), Some(have)) if want != have => {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} SET DEFAULT {};",
qn(&td.module, &td.table),
qi(&col),
want
),
);
}
(Some(want), None) => {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} SET DEFAULT {};",
qn(&td.module, &td.table),
qi(&col),
want
),
);
}
(None, Some(_)) => {
push_tx(
ops,
format!(
"ALTER TABLE {} ALTER COLUMN {} DROP DEFAULT;",
qn(&td.module, &td.table),
qi(&col)
),
);
}
_ => {}
}
}
let target_cols: HashSet<String> = td
.properties
.iter()
.map(|p| p.name.clone())
.chain(
td.links
.iter()
.filter(|l| !l.is_junction_backed())
.map(|l| format!("{}_id", l.name)),
)
.collect();
for col in &existing.columns {
let n = col.name.as_str();
if target_cols.contains(n) {
continue;
}
if n.starts_with("__") && n.ends_with("__") {
continue;
}
push_tx(
ops,
format!(
"ALTER TABLE {} DROP COLUMN IF EXISTS {};",
qn(&td.module, &td.table),
qi(n)
),
);
}
}
fn emit_fk_diff(
td: &TypeDescriptor,
existing: Option<&DbTable>,
type_map: &HashMap<String, (&str, &str)>,
polymorphic: &HashSet<String>,
ops: &mut Vec<DiffOp>,
) {
use crate::schema::{DeleteAction, DeleteSide};
let existing_fk_names: HashSet<&str> = existing
.map(|e| e.foreign_keys.iter().map(|fk| fk.constraint_name.as_str()).collect())
.unwrap_or_default();
for l in &td.links {
if l.is_junction_backed() {
continue;
}
if polymorphic.contains(&l.target) {
continue;
}
let cname = format!("{}_{}_fkey", td.table, l.name);
if existing_fk_names.contains(cname.as_str()) {
continue;
}
let Some((tgt_module, tgt_table)) = type_map.get(&l.target) else {
continue;
};
let needs_deferred = crate::export::needs_deferred_target_fk(&l.on_delete);
let on_delete = l
.on_delete
.iter()
.find(|p| p.side == DeleteSide::Target)
.map(|p| match &p.action {
DeleteAction::Restrict if needs_deferred => " DEFERRABLE INITIALLY DEFERRED",
DeleteAction::Restrict => " ON DELETE RESTRICT",
DeleteAction::DeferredRestrict => " DEFERRABLE INITIALLY DEFERRED",
DeleteAction::DeleteSource => " ON DELETE CASCADE",
DeleteAction::Allow => " ON DELETE SET NULL",
_ => " ON DELETE RESTRICT",
})
.unwrap_or(if needs_deferred {
" DEFERRABLE INITIALLY DEFERRED"
} else {
" ON DELETE RESTRICT"
});
push_tx(
ops,
format!(
"ALTER TABLE {} ADD CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {}(id){};",
qn(&td.module, &td.table),
qi(&cname),
qi(&format!("{}_id", l.name)),
qn(tgt_module, tgt_table),
on_delete
),
);
}
}
#[allow(clippy::too_many_arguments)]
fn emit_junction_table(
td: &TypeDescriptor,
ml_name: &str,
through: Option<&str>,
schema: &SchemaDescriptor,
single: bool,
ops: &mut Vec<DiffOp>,
) {
let jt_name = format!("{}.{}", td.table, ml_name);
let src_on_delete = " ON DELETE CASCADE";
let mut col_lines = format!(
" source uuid NOT NULL REFERENCES {}(id){},\n target uuid NOT NULL",
qn(&td.module, &td.table),
src_on_delete,
);
if let Some(through_qname) = through
&& let Some(through_td) = schema
.types
.iter()
.find(|t| format!("{}::{}", t.module, t.name) == through_qname && t.junction)
{
for p in &through_td.properties {
if p.name == "id" {
continue;
}
let not_null = if p.nullable { "" } else { " NOT NULL" };
col_lines.push_str(&format!(",\n {} {}{}", qi(&p.name), col_type_str(p), not_null));
}
}
let pk_clause = if single {
"PRIMARY KEY (source)"
} else {
"PRIMARY KEY (source, target)"
};
push_tx(
ops,
format!(
"CREATE TABLE IF NOT EXISTS {} (\n{},\n {}\n);",
qn(&td.module, &jt_name),
col_lines,
pk_clause,
),
);
}
fn diff_states_inner(before: &DbState, after: &DbState) -> Vec<DiffOp> {
let mut ops: Vec<DiffOp> = Vec::new();
let before_schemas: HashSet<&str> = before.schemas.iter().map(|s| s.as_str()).collect();
let before_tables: HashMap<(&str, &str), &DbTable> = before
.tables
.iter()
.map(|t| ((t.schema.as_str(), t.name.as_str()), t))
.collect();
let before_enums: HashMap<(&str, &str), &DbEnum> = before
.enums
.iter()
.map(|e| ((e.schema.as_str(), e.name.as_str()), e))
.collect();
let before_domains: HashSet<(&str, &str)> = before
.domains
.iter()
.map(|d| (d.schema.as_str(), d.name.as_str()))
.collect();
for schema in &after.schemas {
if schema == "default" {
continue;
} if !before_schemas.contains(schema.as_str()) {
push_tx(&mut ops, format!("CREATE SCHEMA IF NOT EXISTS {};", pg_schema(schema)));
}
}
for e in &after.enums {
match before_enums.get(&(e.schema.as_str(), e.name.as_str())) {
None => {
let members: Vec<String> = e
.members
.iter()
.map(|m| format!("'{}'", m.replace('\'', "''")))
.collect();
push_tx(
&mut ops,
format!(
"DO $$ BEGIN CREATE TYPE {}.{} AS ENUM ({}); \
EXCEPTION WHEN duplicate_object THEN NULL; END $$;",
pg_schema(&e.schema),
qi(&e.name),
members.join(", ")
),
);
}
Some(existing) => {
let existing_set: HashSet<&str> = existing.members.iter().map(|m| m.as_str()).collect();
for member in &e.members {
if !existing_set.contains(member.as_str()) {
push_tx(
&mut ops,
format!(
"ALTER TYPE {}.{} ADD VALUE IF NOT EXISTS '{}';",
pg_schema(&e.schema),
qi(&e.name),
member.replace('\'', "''")
),
);
}
}
}
}
}
for d in &after.domains {
if !before_domains.contains(&(d.schema.as_str(), d.name.as_str())) {
push_tx(
&mut ops,
format!(
"-- TODO: recreate domain {}.{} (reconstruct DDL from source migrations)",
pg_schema(&d.schema),
qi(&d.name)
),
);
}
}
let mut new_tables: HashSet<(String, String)> = HashSet::new();
for t in &after.tables {
let key = (t.schema.as_str(), t.name.as_str());
match before_tables.get(&key) {
None => {
emit_create_table_from_db(t, &mut ops);
new_tables.insert((t.schema.clone(), t.name.clone()));
}
Some(before_t) => {
emit_column_diff_from_db(t, before_t, &mut ops);
}
}
}
for t in &after.tables {
if let Some(before_t) = before_tables.get(&(t.schema.as_str(), t.name.as_str())) {
let before_fk_names: HashSet<&str> = before_t
.foreign_keys
.iter()
.map(|fk| fk.constraint_name.as_str())
.collect();
for fk in &t.foreign_keys {
if !before_fk_names.contains(fk.constraint_name.as_str()) {
push_tx(
&mut ops,
format!(
"ALTER TABLE {}.{} ADD CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {}.{}(id);",
pg_schema(&t.schema),
qi(&t.name),
qi(&fk.constraint_name),
qi(&fk.local_column),
pg_schema(&fk.ref_schema),
qi(&fk.ref_table)
),
);
}
}
}
}
for t in &after.tables {
let table_is_new = new_tables.contains(&(t.schema.clone(), t.name.clone()));
let before_indexes = before_tables
.get(&(t.schema.as_str(), t.name.as_str()))
.map(|bt| bt.indexes.as_slice())
.unwrap_or_default();
for idx in &t.indexes {
let already_there = if idx.columns.is_empty() {
before_indexes.iter().any(|b| b.name == idx.name)
} else {
before_indexes
.iter()
.any(|b| b.columns == idx.columns && b.is_unique == idx.is_unique)
};
if already_there {
continue;
}
if idx.columns.is_empty() {
continue;
}
let use_concurrently = !table_is_new;
let concurrently = if use_concurrently { "CONCURRENTLY " } else { "" };
let unique = if idx.is_unique { "UNIQUE " } else { "" };
let idx_sql = format!(
"CREATE {unique}INDEX {concurrently}IF NOT EXISTS {} ON {}.{} ({});",
qi(&idx.name),
pg_schema(&t.schema),
qi(&t.name),
idx.columns.iter().map(|c| qi(c)).collect::<Vec<_>>().join(", "),
);
ops.push(DiffOp {
sql: idx_sql,
non_transactional: use_concurrently,
});
}
}
let after_tables: HashSet<(&str, &str)> = after
.tables
.iter()
.map(|t| (t.schema.as_str(), t.name.as_str()))
.collect();
for t in &before.tables {
if !after_tables.contains(&(t.schema.as_str(), t.name.as_str())) {
push_tx(
&mut ops,
format!("DROP TABLE IF EXISTS {}.{} CASCADE;", pg_schema(&t.schema), qi(&t.name)),
);
}
}
let after_enum_set: HashSet<(&str, &str)> = after
.enums
.iter()
.map(|e| (e.schema.as_str(), e.name.as_str()))
.collect();
for e in &before.enums {
if !after_enum_set.contains(&(e.schema.as_str(), e.name.as_str())) {
push_tx(
&mut ops,
format!("DROP TYPE IF EXISTS {}.{} CASCADE;", pg_schema(&e.schema), qi(&e.name)),
);
}
}
let after_schema_set: HashSet<&str> = after.schemas.iter().map(|s| s.as_str()).collect();
for schema in &before.schemas {
if schema == "default" {
continue;
} if !after_schema_set.contains(schema.as_str()) {
push_tx(
&mut ops,
format!("DROP SCHEMA IF EXISTS {} CASCADE;", pg_schema(schema)),
);
}
}
ops
}
fn emit_create_table_from_db(t: &DbTable, ops: &mut Vec<DiffOp>) {
let mut lines: Vec<String> = Vec::new();
for col in &t.columns {
let not_null = if col.nullable { "" } else { " NOT NULL" };
if col.is_generated {
lines.push(format!(
" {} {} GENERATED ALWAYS AS (/* see source */) STORED",
qi(&col.name),
col.pg_type
));
} else {
lines.push(format!(" {} {}{}", qi(&col.name), col.pg_type, not_null));
}
}
push_tx(
ops,
format!(
"CREATE TABLE IF NOT EXISTS {}.{} (\n{}\n);",
pg_schema(&t.schema),
qi(&t.name),
lines.join(",\n")
),
);
}
fn emit_column_diff_from_db(after: &DbTable, before: &DbTable, ops: &mut Vec<DiffOp>) {
let before_cols: HashSet<&str> = before.columns.iter().map(|c| c.name.as_str()).collect();
let after_cols: HashSet<&str> = after.columns.iter().map(|c| c.name.as_str()).collect();
for col in &after.columns {
if !before_cols.contains(col.name.as_str()) {
let not_null = if col.nullable { "" } else { " NOT NULL" };
push_tx(
ops,
format!(
"ALTER TABLE {}.{} ADD COLUMN IF NOT EXISTS {} {}{};",
pg_schema(&after.schema),
qi(&after.name),
qi(&col.name),
col.pg_type,
not_null
),
);
}
}
for col in &before.columns {
if !after_cols.contains(col.name.as_str()) {
push_tx(
ops,
format!(
"ALTER TABLE {}.{} DROP COLUMN IF EXISTS {};",
pg_schema(&after.schema),
qi(&after.name),
qi(&col.name)
),
);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::schema::{EnumDescriptor, LinkDescriptor, PropertyDescriptor, SchemaDescriptor, TypeDescriptor};
fn empty_state() -> DbState {
DbState::default()
}
fn prop(name: &str, pg_type: &str, nullable: bool) -> PropertyDescriptor {
PropertyDescriptor {
name: name.into(),
pg_type: pg_type.into(),
nullable,
default_sql: if name == "id" { Some("uuidv7()".into()) } else { None },
default_pyql: None,
description: None,
check_constraints: vec![],
is_exclusive: name == "id",
is_pk: name == "id",
is_readonly: name == "id",
rewrites: vec![],
tuple_members: None,
column_type: None,
}
}
fn simple_type(module: &str, name: &str, table: &str) -> TypeDescriptor {
TypeDescriptor {
name: name.into(),
module: module.into(),
table: table.into(),
abstract_: false,
materialized: false,
description: None,
parents: vec![],
interfaces: vec![],
bases: vec![],
properties: vec![prop("id", "uuid", false), prop("name", "text", true)],
links: vec![],
multilinks: vec![],
computed: vec![],
constraints: vec![],
indexes: vec![],
partition: None,
vector_indexes: vec![],
search_indexes: vec![],
triggers: vec![],
junction: false,
signals: vec![],
}
}
#[test]
fn test_schema_content_changed_detects_a_readonly_only_flip() {
let before = simple_type("default", "Person", "Person");
let mut after = before.clone();
after.properties[1].is_readonly = true; assert_ne!(before.properties[1].is_readonly, after.properties[1].is_readonly);
let schema_before = SchemaDescriptor {
types: vec![before],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let schema_after = SchemaDescriptor {
types: vec![after],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
assert!(schema_content_changed(&schema_after, Some(&schema_before)));
}
#[test]
fn test_schema_content_changed_detects_a_new_rewrite() {
let before = simple_type("default", "Person", "Person");
let mut after = before.clone();
after.properties[1].rewrites.push(crate::schema::RewriteEntry {
on: 1,
handler: "str_upper(.name)".into(),
});
let schema_before = SchemaDescriptor {
types: vec![before],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let schema_after = SchemaDescriptor {
types: vec![after],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
assert!(schema_content_changed(&schema_after, Some(&schema_before)));
}
#[test]
fn test_schema_content_changed_is_false_for_identical_schemas() {
let t = simple_type("default", "Person", "Person");
let schema = SchemaDescriptor {
types: vec![t],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let other = schema.clone();
assert!(!schema_content_changed(&schema, Some(&other)));
}
#[test]
fn test_schema_content_changed_true_against_none_when_target_is_non_empty() {
let t = simple_type("default", "Person", "Person");
let schema = SchemaDescriptor {
types: vec![t],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
assert!(
schema_content_changed(&schema, None),
"no prior snapshot at all must count as changed"
);
}
#[test]
fn test_schema_content_changed_false_against_none_when_target_is_also_empty() {
let schema = SchemaDescriptor::default();
assert!(!schema_content_changed(&schema, None));
}
#[test]
fn test_schema_content_changed_still_true_when_ddl_visible_things_also_changed() {
let schema_before = SchemaDescriptor::default();
let schema_after = SchemaDescriptor {
types: vec![simple_type("default", "Person", "Person")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
assert!(schema_content_changed(&schema_after, Some(&schema_before)));
}
#[test]
fn test_schema_content_changed_detects_a_new_channel() {
let schema_before = SchemaDescriptor::default();
let schema_after = SchemaDescriptor {
channels: vec![crate::schema::ChannelDescriptor {
name: "UserUpdates".into(),
module: "default".into(),
wire_name: "default__user_updates".into(),
payload: crate::schema::ChannelPayload::Scalar("text".into()),
description: None,
}],
..Default::default()
};
assert!(schema_content_changed(&schema_after, Some(&schema_before)));
}
#[test]
fn test_new_schema_and_table() {
let schema = SchemaDescriptor {
types: vec![simple_type("catalog", "Product", "Product")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let ops = diff_schema(&schema, &empty_state()).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains("CREATE SCHEMA IF NOT EXISTS \"catalog\""),
"got:\n{joined}"
);
assert!(
joined.contains("CREATE TABLE IF NOT EXISTS \"catalog\".\"Product\""),
"got:\n{joined}"
);
assert!(
joined.contains("CREATE OR REPLACE TRIGGER pylon_cache_invalidate\n AFTER INSERT OR UPDATE OR DELETE ON \"catalog\".\"Product\""),
"new table must get the cache-invalidation trigger; got:\n{joined}"
);
}
fn widget_with_trigger(on: u8, timing: &str, handler: &str) -> TypeDescriptor {
let mut t = simple_type("default", "Widget", "Widget");
t.triggers = vec![crate::schema::TriggerDescriptor {
on,
timing: timing.into(),
handler: handler.into(),
}];
t
}
#[test]
fn test_new_table_with_user_trigger_emits_the_compiled_trigger_ddl() {
let schema = SchemaDescriptor {
types: vec![widget_with_trigger(
1,
"After",
"update Widget set { name := __new__.name }",
)],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let ops = diff_schema(&schema, &empty_state()).unwrap();
let joined = ops.join("\n");
assert!(joined.contains("NEW.\"name\""), "got:\n{joined}");
}
#[test]
fn a_trigger_emitted_by_an_older_build_is_replaced() {
let schema = SchemaDescriptor {
types: vec![widget_with_trigger(
1,
"After",
"update Widget set { name := __new__.name }",
)],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let mut stale = schema_to_db_state(&schema);
let current_name = stale
.tables
.iter()
.flat_map(|t| t.triggers.iter().cloned())
.find(|n| n.starts_with("Widget_"))
.expect("the fixture should project a Widget trigger");
let stale_name = "Widget_trg_0badc0de".to_string();
for table in &mut stale.tables {
for trigger in &mut table.triggers {
if *trigger == current_name {
*trigger = stale_name.clone();
}
}
}
let joined = diff_schema(&schema, &stale).unwrap().join("\n");
assert!(
joined.contains(&format!("DROP TRIGGER IF EXISTS \"{stale_name}\"")),
"the stale trigger should be dropped, got:\n{joined}"
);
assert!(
joined.contains(¤t_name),
"the current trigger should be created, got:\n{joined}"
);
assert!(
joined.contains(&format!("DROP FUNCTION IF EXISTS \"public\".\"{stale_name}\"()")),
"the orphaned function should be dropped, got:\n{joined}"
);
}
#[test]
fn test_user_trigger_already_present_in_offline_baseline_produces_no_further_steps() {
let schema = SchemaDescriptor {
types: vec![widget_with_trigger(
1,
"After",
"update Widget set { name := __new__.name }",
)],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let baseline = schema_to_db_state(&schema);
let steps = diff_schema_steps(&schema, &baseline, &HashMap::new()).unwrap();
assert!(
steps.is_empty(),
"expected zero further migration steps, got: {steps:?}"
);
}
#[test]
fn a_link_target_gaining_a_subtype_loses_its_foreign_key() {
let mut order = simple_type("default", "Order", "Order");
order.links.push(LinkDescriptor {
name: "customer".into(),
target: "default::Person".into(),
nullable: false,
through: None,
description: None,
default_pyql: None,
is_exclusive: false,
is_readonly: false,
rewrites: vec![],
on_delete: vec![],
});
let before = SchemaDescriptor {
types: vec![order, simple_type("default", "Person", "Person")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let mut after = before.clone();
let mut vip = simple_type("default", "Vip", "Vip");
vip.bases = vec!["default::Person".into()];
after.types.push(vip);
let joined = diff_schema(&after, &schema_to_db_state(&before)).unwrap().join("\n");
assert!(
joined.contains("ALTER TABLE \"public\".\"Order\" DROP CONSTRAINT IF EXISTS \"Order_customer_fkey\""),
"got:\n{joined}"
);
let settled = diff_schema_steps(&after, &schema_to_db_state(&after), &HashMap::new()).unwrap();
assert!(settled.is_empty(), "nothing further once applied, got: {settled:?}");
}
#[test]
fn test_new_table_with_plain_link_gets_its_fk_constraint() {
let mut order = simple_type("default", "Order", "Order");
order.links.push(LinkDescriptor {
name: "customer".into(),
target: "default::Person".into(),
nullable: false,
through: None,
description: None,
default_pyql: None,
is_exclusive: false,
is_readonly: false,
rewrites: vec![],
on_delete: vec![],
});
let schema = SchemaDescriptor {
types: vec![order, simple_type("default", "Person", "Person")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let ops = diff_schema(&schema, &empty_state()).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains("ADD CONSTRAINT \"Order_customer_fkey\" FOREIGN KEY (\"customer_id\") REFERENCES \"public\".\"Person\"(id)"),
"new table's plain link must get its FK constraint in the same diff; got:\n{joined}"
);
}
#[test]
fn test_an_existing_exclusive_multilink_junction_gains_its_unique_target() {
use crate::schema::MultiLinkDescriptor;
let mut person = simple_type("default", "Person", "Person");
person.multilinks.push(MultiLinkDescriptor {
name: "keys".into(),
target: "default::Key".into(),
through: None,
nullable: false,
description: None,
default_pyql: None,
on_delete: vec![],
is_exclusive: true,
});
let schema = SchemaDescriptor {
types: vec![person, simple_type("default", "Key", "Key")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let mut state = schema_to_db_state(&schema);
assert!(diff_schema(&schema, &state).unwrap().is_empty());
for table in state.tables.iter_mut().filter(|t| t.name == "Person.keys") {
table.indexes.clear();
}
let joined = diff_schema(&schema, &state).unwrap().join("\n");
assert!(
joined.contains("CREATE UNIQUE INDEX IF NOT EXISTS \"Person.keys_target_key\" ON \"public\".\"Person.keys\" (\"target\")"),
"got:\n{joined}"
);
}
#[test]
fn test_cache_invalidate_trigger_not_duplicated_for_junction_through_type() {
use crate::schema::MultiLinkDescriptor;
let mut product = simple_type("default", "Product", "Product");
product.multilinks.push(MultiLinkDescriptor {
name: "tags".into(),
target: "default::Tag".into(),
through: Some("default::ProductTag".into()),
nullable: false,
description: None,
default_pyql: None,
on_delete: vec![],
is_exclusive: false,
});
let mut junction = simple_type("default", "ProductTag", "Product.tags");
junction.junction = true;
let schema = SchemaDescriptor {
types: vec![product, junction, simple_type("default", "Tag", "Tag")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let ops = diff_schema(&schema, &empty_state()).unwrap();
let trigger_count = ops
.iter()
.filter(|op| op.contains("AFTER INSERT OR UPDATE OR DELETE ON \"public\".\"Product.tags\""))
.count();
assert_eq!(
trigger_count, 1,
"junction table's own td.table and the owning type's multilink both resolve to \
the same physical table — must be deduped to one trigger, got {trigger_count} in: {ops:?}"
);
}
fn person_with_junction_backed_spouse() -> SchemaDescriptor {
let mut person = simple_type("default", "Person", "Person");
person.links.push(LinkDescriptor {
name: "spouse".into(),
target: "default::Person".into(),
nullable: true,
through: Some("default::Marriage".into()),
description: None,
default_pyql: None,
is_exclusive: true,
is_readonly: false,
rewrites: vec![],
on_delete: vec![],
});
let mut junction = simple_type("default", "Marriage", "Person.spouse");
junction.junction = true;
SchemaDescriptor {
types: vec![person, junction],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
}
}
#[test]
fn test_junction_backed_single_link_creates_junction_table_from_scratch() {
let schema = person_with_junction_backed_spouse();
let ops = diff_schema(&schema, &empty_state()).unwrap();
let joined = ops.join("\n");
assert!(
!joined.contains("spouse_id"),
"no {{name}}_id column/FK for a junction-backed link, got:\n{joined}"
);
assert!(
joined.contains("CREATE TABLE IF NOT EXISTS \"public\".\"Person.spouse\""),
"got:\n{joined}"
);
assert!(joined.contains("PRIMARY KEY (source)"), "got:\n{joined}");
assert!(
joined.contains("CREATE UNIQUE INDEX IF NOT EXISTS \"Person.spouse_target_key\" ON \"public\".\"Person.spouse\" (\"target\")"),
"got:\n{joined}"
);
}
#[test]
fn test_junction_backed_single_link_diff_is_idempotent_once_applied() {
let schema = person_with_junction_backed_spouse();
let state = DbState {
schemas: vec![],
tables: vec![
DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec!["pylon_cache_invalidate".into()],
},
DbTable {
schema: "default".into(),
name: "Person.spouse".into(),
columns: vec![
DbColumn {
name: "source".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: None,
},
DbColumn {
name: "target".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: None,
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![
DbForeignKey {
constraint_name: "Person_spouse_source_fkey".into(),
local_column: "source".into(),
ref_schema: "default".into(),
ref_table: "Person".into(),
},
DbForeignKey {
constraint_name: "Person_spouse_target_fkey".into(),
local_column: "target".into(),
ref_schema: "default".into(),
ref_table: "Person".into(),
},
],
indexes: vec![DbIndex {
name: "Person.spouse_target_key".into(),
is_unique: true,
method: "btree".into(),
columns: vec!["target".into()],
predicate: None,
key: None,
}],
checks: vec![],
triggers: vec!["pylon_cache_invalidate".into()],
},
],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
assert!(
ops.is_empty(),
"already-migrated junction-backed single link must diff to no ops, got: {:?}",
ops
);
}
#[test]
fn test_cache_invalidate_trigger_backfilled_on_pre_existing_table() {
let schema = SchemaDescriptor {
types: vec![simple_type("default", "Person", "Person")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![], }],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains("CREATE OR REPLACE TRIGGER pylon_cache_invalidate\n AFTER INSERT OR UPDATE OR DELETE ON \"public\".\"Person\""),
"pre-existing table missing the trigger must get it backfilled; got:\n{joined}"
);
}
#[test]
fn test_cache_invalidate_trigger_not_dropped_when_already_present() {
let schema = SchemaDescriptor {
types: vec![simple_type("default", "Person", "Person")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec!["pylon_cache_invalidate".into()],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
assert!(
ops.iter()
.all(|op| !op.contains("DROP TRIGGER") && !op.contains("pylon_cache_invalidate")),
"already-present trigger must not be re-created or dropped; got: {:?}",
ops
);
}
#[test]
fn test_no_ops_when_in_sync() {
let schema = SchemaDescriptor {
types: vec![simple_type("default", "Person", "Person")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec!["pylon_cache_invalidate".into()],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
assert!(ops.is_empty(), "expected no ops, got: {:?}", ops);
}
#[test]
fn test_add_column() {
let mut td = simple_type("default", "Person", "Person");
td.properties.push(prop("email", "text", true));
let schema = SchemaDescriptor {
types: vec![td],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
let joined = ops.join("\n");
assert!(joined.contains("ADD COLUMN IF NOT EXISTS \"email\""), "got:\n{joined}");
}
#[test]
fn test_property_type_change_emits_alter_column_type() {
let mut td = simple_type("default", "Person", "Person");
td.properties.push(prop("rating", "int8", true));
let schema = SchemaDescriptor {
types: vec![td],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
DbColumn {
name: "rating".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains(
"ALTER TABLE \"public\".\"Person\" ALTER COLUMN \"rating\" TYPE int8 USING \"rating\"::int8;"
),
"got:\n{joined}"
);
}
#[test]
fn test_a_column_moving_between_enums_converts_through_text() {
let mut td = simple_type("default", "Person", "Person");
td.properties.push(prop("verb", "\"public\".\"HttpMethod\"", true));
let schema = SchemaDescriptor {
types: vec![td],
enums: vec![crate::schema::EnumDescriptor {
name: "HttpMethod".into(),
module: "default".into(),
members: vec!["GET".into(), "POST".into()],
}],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "verb".into(),
pg_type: "\"public\".\"Method\"".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let joined = diff_schema(&schema, &state).unwrap().join("\n");
assert!(
joined.contains(r#"USING "verb"::text::"public"."HttpMethod";"#),
"the conversion should go through text:\n{joined}"
);
}
#[test]
fn test_a_column_default_is_dropped_before_its_type_changes() {
let mut td = simple_type("default", "Person", "Person");
let mut rating = prop("rating", "int8", true);
rating.default_sql = Some("'1'".into());
td.properties.push(rating);
let schema = SchemaDescriptor {
types: vec![td],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "rating".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: Some("'x'::text".into()),
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let joined = diff_schema(&schema, &state).unwrap().join("\n");
let Some(drop) = joined.find(r#"ALTER COLUMN "rating" DROP DEFAULT;"#) else {
panic!("the default should be dropped:\n{joined}");
};
let Some(retype) = joined.find(r#"ALTER COLUMN "rating" TYPE int8"#) else {
panic!("the type should change:\n{joined}");
};
assert!(drop < retype, "the drop must come first:\n{joined}");
}
#[test]
fn test_property_type_change_surfaces_a_required_cast_expression_step() {
let mut td = simple_type("default", "Person", "Person");
td.properties.push(prop("rating", "int8", true));
let schema = SchemaDescriptor {
types: vec![td],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
DbColumn {
name: "rating".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let steps = diff_schema_steps(&schema, &state, &HashMap::new()).unwrap();
let step = steps
.iter()
.find(|s| matches!(&s.op_key, OpKey::Table(m, t) if m == "default" && t == "Person"))
.expect("expected an alter step for Person");
assert_eq!(step.required_input.len(), 1, "got: {:?}", step.required_input);
let input = &step.required_input[0];
assert_eq!(input.placeholder, "cast_expr__rating");
assert_eq!(input.default_expr, "\"rating\"::int8");
assert_eq!(input.type_name, "default::Person");
let placeholder_token = format!("\\({})", input.placeholder);
assert!(
step.ddl.iter().any(|op| op.sql.contains(&placeholder_token)),
"expected the placeholder token in the step's DDL, got: {:?}",
step.ddl.iter().map(|op| &op.sql).collect::<Vec<_>>()
);
}
#[test]
fn test_equivalent_base_type_spelling_is_not_a_diff() {
let mut td = simple_type("default", "Person", "Person");
td.properties.push(prop("age", "int8", true));
let schema = SchemaDescriptor {
types: vec![td],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
DbColumn {
name: "age".into(),
pg_type: "bigint".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec!["pylon_cache_invalidate".into()],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
assert!(
ops.iter().all(|op| !op.contains("ALTER COLUMN")),
"expected no ALTER COLUMN ops, got: {:?}",
ops
);
}
#[test]
fn test_check_the_schema_no_longer_declares_is_dropped() {
let mut person = simple_type("default", "Person", "Person");
person.properties = vec![prop("id", "uuid", false)];
let schema = SchemaDescriptor {
types: vec![person],
..SchemaDescriptor::default()
};
let state = DbState {
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: None,
}],
foreign_keys: vec![],
indexes: vec![],
checks: vec![
DbCheck {
constraint_name: "Person_name_deadbeef_check".into(),
},
DbCheck {
constraint_name: "a_hand_written_check".into(),
},
],
triggers: vec![],
}],
..DbState::default()
};
let ops = diff_schema_ops(&schema, &state).unwrap();
let joined = ops.iter().map(|op| op.sql.as_str()).collect::<Vec<_>>().join("\n");
assert!(
joined.contains("DROP CONSTRAINT IF EXISTS \"Person_name_deadbeef_check\""),
"got:\n{joined}"
);
assert!(
!joined.contains("a_hand_written_check"),
"a constraint Pylon did not name is left alone:\n{joined}"
);
}
#[test]
fn test_changed_scalar_domain_checks_are_reconciled() {
use crate::schema::ScalarDescriptor;
let schema = SchemaDescriptor {
scalars: vec![ScalarDescriptor {
name: "EmailStr".into(),
module: "default".into(),
base: "Str".into(),
pg_type: "text".into(),
check_constraints: vec!["char_length(VALUE) <= 320".into()],
is_sequence: false,
}],
..SchemaDescriptor::default()
};
let state = DbState {
domains: vec![DbDomain {
schema: "default".into(),
name: "EmailStr".into(),
checks: vec!["EmailStr_stale000_check".into()],
}],
..DbState::default()
};
let ops = diff_schema_ops(&schema, &state).unwrap();
let joined = ops.iter().map(|op| op.sql.as_str()).collect::<Vec<_>>().join("\n");
assert!(joined.contains("DROP CONSTRAINT"), "the old set is cleared:\n{joined}");
assert!(
joined.contains("ADD CONSTRAINT") && joined.contains("char_length(VALUE) <= 320"),
"the new set is added:\n{joined}"
);
}
#[test]
fn test_demoting_an_interface_to_a_mixin_drops_its_view() {
let mut individual = simple_type("default", "Individual", "Individual");
individual.properties = vec![prop("id", "uuid", false)];
let schema = SchemaDescriptor {
types: vec![individual],
..SchemaDescriptor::default()
};
let state = DbState {
tables: vec![],
views: vec![DbView {
schema: "public".into(),
name: "Account".into(),
body_hash: "whatever".into(),
}],
..DbState::default()
};
let ops = diff_schema_ops(&schema, &state).unwrap();
let joined = ops.iter().map(|op| op.sql.as_str()).collect::<Vec<_>>().join("\n");
assert!(
joined.contains("DROP VIEW IF EXISTS \"public\".\"Account\""),
"got:\n{joined}"
);
}
#[test]
fn test_registered_scalar_domain_adoption_drops_and_recreates_dependent_interface_view() {
use crate::schema::ScalarDescriptor;
let mut account = simple_type("default", "Account", "Account");
account.abstract_ = true;
account.materialized = true;
account.properties = vec![prop("id", "uuid", false), prop("email", "text", false)];
account.properties[1].column_type = Some("\"public\".\"Email\"".into());
let mut individual = simple_type("default", "Individual", "Individual");
individual.interfaces = vec!["default::Account".into()];
individual.properties = vec![prop("id", "uuid", false), prop("email", "text", false)];
individual.properties[1].column_type = Some("\"public\".\"Email\"".into());
let schema = SchemaDescriptor {
types: vec![account, individual],
scalars: vec![ScalarDescriptor {
name: "Email".into(),
module: "default".into(),
base: "Str".into(),
pg_type: "text".into(),
check_constraints: vec!["value ~ '@'".into()],
is_sequence: false,
}],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let view_ddl = crate::export::interface_view_ddl_with_names(&schema)
.into_iter()
.find(|(_, n, _)| n == "Account")
.unwrap()
.2;
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Individual".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "email".into(),
pg_type: "text".into(),
nullable: false,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec!["pylon_cache_invalidate".into()],
}],
views: vec![DbView {
schema: "default".into(),
name: "Account".into(),
body_hash: ddl_hash(&view_ddl),
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema_ops(&schema, &state).unwrap();
let joined = ops.iter().map(|op| op.sql.as_str()).collect::<Vec<_>>().join("\n");
let drop_pos = joined
.find("DROP VIEW IF EXISTS \"public\".\"Account\"")
.unwrap_or_else(|| panic!("missing DROP VIEW; got:\n{joined}"));
let alter_pos = joined
.find("ALTER TABLE \"public\".\"Individual\" ALTER COLUMN \"email\" TYPE \"public\".\"Email\"")
.unwrap_or_else(|| panic!("missing ALTER COLUMN TYPE; got:\n{joined}"));
let create_pos = joined
.rfind("CREATE VIEW \"public\".\"Account\"")
.unwrap_or_else(|| panic!("missing CREATE VIEW; got:\n{joined}"));
assert!(drop_pos < alter_pos, "DROP VIEW must precede the ALTER; got:\n{joined}");
assert!(
alter_pos < create_pos,
"CREATE VIEW must follow the ALTER; got:\n{joined}"
);
assert_eq!(
joined.matches("CREATE VIEW \"public\".\"Account\"").count(),
1,
"view must be recreated exactly once, not duplicated by Phase 10; got:\n{joined}"
);
}
fn exclusive_email_account_schema(implementor_names: &[&str]) -> SchemaDescriptor {
let mut account = simple_type("default", "Account", "Account");
account.abstract_ = true;
account.materialized = true;
account.properties = vec![prop("id", "uuid", false), prop("email", "text", false)];
account.properties[1].is_exclusive = true;
let mut types = vec![account];
for name in implementor_names {
let mut t = simple_type("default", name, name);
t.interfaces = vec!["default::Account".into()];
t.properties = vec![prop("id", "uuid", false), prop("email", "text", false)];
t.properties[1].is_exclusive = true;
types.push(t);
}
SchemaDescriptor {
types,
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
}
}
#[test]
fn test_new_implementor_added_to_existing_interface_gets_exclusive_triggers_retroactively() {
let schema = exclusive_email_account_schema(&["Individual", "Organization"]);
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Individual".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "email".into(),
pg_type: "text".into(),
nullable: false,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![
"pylon_cache_invalidate".into(),
"_excl_Account_email_ins".into(),
"_excl_Account_email_upd".into(),
],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema_ops(&schema, &state).unwrap();
let joined = ops.iter().map(|op| op.sql.as_str()).collect::<Vec<_>>().join("\n");
assert!(
joined.contains(
"CREATE CONSTRAINT TRIGGER \"_excl_Account_email_ins\"\nAFTER INSERT ON \"public\".\"Organization\""
),
"the new implementor must get the exclusive trigger; got:\n{joined}"
);
assert!(
!joined.contains("ON \"public\".\"Individual\""),
"the already-migrated implementor's existing triggers must not be re-emitted; got:\n{joined}"
);
}
#[test]
fn test_the_shared_exclusive_guard_function_precedes_every_implementors_trigger() {
let mut schema = exclusive_email_account_schema(&["Individual", "Organization"]);
let individual = schema
.types
.iter_mut()
.find(|t| t.name == "Individual")
.expect("Individual is in the schema");
individual.links.push(LinkDescriptor {
name: "employer".into(),
target: "default::Organization".into(),
nullable: false,
through: None,
description: None,
default_pyql: None,
is_exclusive: false,
is_readonly: false,
rewrites: vec![],
on_delete: vec![],
});
let ops = diff_schema_ops(&schema, &DbState::default()).unwrap();
let joined = ops.iter().map(|op| op.sql.as_str()).collect::<Vec<_>>().join("\n");
let function_pos = joined
.find("CREATE OR REPLACE FUNCTION \"public\".\"_excl_Account_email\"")
.unwrap_or_else(|| panic!("missing the guard function; got:\n{joined}"));
let first_trigger_pos = joined
.find("CREATE CONSTRAINT TRIGGER \"_excl_Account_email_ins\"")
.unwrap_or_else(|| panic!("missing the exclusive triggers; got:\n{joined}"));
assert!(
function_pos < first_trigger_pos,
"the guard function must be defined before the first trigger naming it; got:\n{joined}"
);
assert_eq!(
joined
.matches("CREATE OR REPLACE FUNCTION \"public\".\"_excl_Account_email\"")
.count(),
1,
"the guard function must be emitted exactly once; got:\n{joined}"
);
}
#[test]
fn test_removing_exclusivity_drops_the_cross_table_triggers() {
let mut schema = exclusive_email_account_schema(&["Individual"]);
for t in &mut schema.types {
for p in &mut t.properties {
if p.name == "email" {
p.is_exclusive = false;
}
}
}
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Individual".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "email".into(),
pg_type: "text".into(),
nullable: false,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![
"pylon_cache_invalidate".into(),
"_excl_Account_email_ins".into(),
"_excl_Account_email_upd".into(),
],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema_ops(&schema, &state).unwrap();
let joined = ops.iter().map(|op| op.sql.as_str()).collect::<Vec<_>>().join("\n");
assert!(
joined.contains("DROP TRIGGER IF EXISTS \"_excl_Account_email_ins\" ON \"public\".\"Individual\""),
"got:\n{joined}"
);
assert!(
joined.contains("DROP TRIGGER IF EXISTS \"_excl_Account_email_upd\" ON \"public\".\"Individual\""),
"got:\n{joined}"
);
assert!(
!joined.contains("DROP TRIGGER IF EXISTS \"pylon_cache_invalidate\""),
"unrelated triggers must not be touched; got:\n{joined}"
);
}
#[test]
fn test_new_enum() {
let schema = SchemaDescriptor {
types: vec![],
scalars: vec![],
enums: vec![EnumDescriptor {
name: "Status".into(),
module: "default".into(),
members: vec!["Active".into(), "Inactive".into()],
}],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let ops = diff_schema(&schema, &empty_state()).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains("CREATE TYPE \"public\".\"Status\" AS ENUM"),
"got:\n{joined}"
);
}
#[test]
fn test_drop_table() {
let schema = SchemaDescriptor {
types: vec![],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "OldType".into(),
columns: vec![],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains("DROP TABLE IF EXISTS \"public\".\"OldType\" CASCADE"),
"got:\n{joined}"
);
}
#[test]
fn test_index_on_existing_table_is_concurrently() {
use crate::schema::VectorIndexDescriptor;
let mut td = simple_type("default", "Post", "Post");
td.vector_indexes.push(VectorIndexDescriptor {
index_name: None,
pointers: vec!["name".into()],
model: "test".into(),
metric: "cosine".into(),
dimensions: 1536,
});
let schema = SchemaDescriptor {
types: vec![td],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Post".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let ops = diff_schema_ops(&schema, &state).unwrap();
let idx_op = ops.iter().find(|op| op.sql.contains("hnsw")).unwrap();
assert!(
idx_op.non_transactional,
"index on pre-existing table should be non-transactional"
);
assert!(
idx_op.sql.contains("CONCURRENTLY"),
"should use CONCURRENTLY: {}",
idx_op.sql
);
}
#[test]
fn test_required_extensions_empty_without_vector_indexes() {
let schema = SchemaDescriptor {
types: vec![simple_type("default", "Post", "Post")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
assert!(required_extensions(&schema).is_empty());
}
#[test]
fn test_missing_extension_ddl_when_vector_index_present_and_not_yet_installed() {
use crate::schema::VectorIndexDescriptor;
let mut td = simple_type("default", "Post", "Post");
td.vector_indexes.push(VectorIndexDescriptor {
index_name: None,
pointers: vec!["name".into()],
model: "test".into(),
metric: "cosine".into(),
dimensions: 1536,
});
let schema = SchemaDescriptor {
types: vec![td],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
assert_eq!(required_extensions(&schema), vec!["vector"]);
let ddl = missing_extension_ddl(&schema, &DbState::default());
assert_eq!(ddl, vec!["CREATE EXTENSION IF NOT EXISTS \"vector\";".to_string()]);
let already_installed = DbState {
extensions: vec!["vector".into()],
..DbState::default()
};
assert!(missing_extension_ddl(&schema, &already_installed).is_empty());
}
#[test]
fn test_index_on_new_table_is_transactional() {
use crate::schema::VectorIndexDescriptor;
let mut td = simple_type("default", "Post", "Post");
td.vector_indexes.push(VectorIndexDescriptor {
index_name: None,
pointers: vec!["name".into()],
model: "test".into(),
metric: "cosine".into(),
dimensions: 1536,
});
let schema = SchemaDescriptor {
types: vec![td],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let ops = diff_schema_ops(&schema, &empty_state()).unwrap();
let idx_op = ops.iter().find(|op| op.sql.contains("hnsw")).unwrap();
assert!(!idx_op.non_transactional, "index on new table should be transactional");
assert!(
!idx_op.sql.contains("CONCURRENTLY"),
"should NOT use CONCURRENTLY: {}",
idx_op.sql
);
}
fn sequence_scalar(module: &str, name: &str) -> crate::schema::ScalarDescriptor {
crate::schema::ScalarDescriptor {
name: name.into(),
module: module.into(),
base: "Sequence".into(),
pg_type: "int8".into(),
check_constraints: vec![],
is_sequence: true,
}
}
#[test]
fn test_new_sequence_creates_sequence_and_domain() {
let schema = SchemaDescriptor {
types: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
scalars: vec![sequence_scalar("default", "OrderNumber")],
..Default::default()
};
let ops = diff_schema(&schema, &empty_state()).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains("CREATE SEQUENCE IF NOT EXISTS \"public\".\"OrderNumber_seq\""),
"got:\n{joined}"
);
assert!(
joined.contains("CREATE DOMAIN \"public\".\"OrderNumber\" AS int8"),
"got:\n{joined}"
);
let seq_pos = joined.find("CREATE SEQUENCE").unwrap();
let dom_pos = joined.find("CREATE DOMAIN").unwrap();
assert!(seq_pos < dom_pos, "sequence must be created before domain");
}
#[test]
fn test_no_ops_sequence_already_exists() {
let schema = SchemaDescriptor {
types: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
scalars: vec![sequence_scalar("default", "OrderNumber")],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
domains: vec![DbDomain {
schema: "default".into(),
name: "OrderNumber".into(),
checks: vec![],
}],
sequences: vec![DbSequence {
schema: "default".into(),
name: "OrderNumber_seq".into(),
}],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
assert!(
ops.is_empty(),
"expected no ops when sequence and domain exist, got: {:?}",
ops
);
}
#[test]
fn test_drop_removed_sequence() {
let schema = SchemaDescriptor {
types: vec![],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
domains: vec![DbDomain {
schema: "default".into(),
name: "OrderNumber".into(),
checks: vec![],
}],
sequences: vec![DbSequence {
schema: "default".into(),
name: "OrderNumber_seq".into(),
}],
..DbState::default()
};
let ops = diff_schema(&schema, &state).unwrap();
let joined = ops.join("\n");
assert!(
joined.contains("DROP DOMAIN IF EXISTS \"public\".\"OrderNumber\""),
"got:\n{joined}"
);
assert!(
joined.contains("DROP SEQUENCE IF EXISTS \"public\".\"OrderNumber_seq\""),
"got:\n{joined}"
);
}
#[test]
fn test_diff_schema_steps_groups_multiple_column_changes_into_one_alter_step() {
let mut person = simple_type("default", "Person", "Person");
person.properties.push(prop("nickname", "text", true));
person.properties.push(prop("age", "int8", true));
let schema = SchemaDescriptor {
types: vec![person],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec!["pylon_cache_invalidate".into()],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let steps = diff_schema_steps(&schema, &state, &HashMap::new()).unwrap();
let table_steps: Vec<&MigrationStep> = steps
.iter()
.filter(|s| matches!(&s.op_key, OpKey::Table(m, t) if m == "default" && t == "Person"))
.collect();
assert_eq!(
table_steps.len(),
1,
"two new columns on the same table must produce one step, got: {:?}",
steps.iter().map(|s| &s.prompt).collect::<Vec<_>>()
);
assert_eq!(table_steps[0].verb, Verb::Alter);
assert_eq!(table_steps[0].prompt, "did you alter object type 'default::Person'?");
assert_eq!(
table_steps[0].ddl.len(),
2,
"expected one ADD COLUMN per new property, got: {:?}",
table_steps[0].ddl.iter().map(|d| &d.sql).collect::<Vec<_>>()
);
}
#[test]
fn test_diff_schema_steps_new_table_is_one_create_step_including_its_trigger() {
let schema = SchemaDescriptor {
types: vec![simple_type("catalog", "Product", "Product")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let steps = diff_schema_steps(&schema, &empty_state(), &HashMap::new()).unwrap();
let table_steps: Vec<&MigrationStep> = steps
.iter()
.filter(|s| matches!(&s.op_key, OpKey::Table(m, t) if m == "catalog" && t == "Product"))
.collect();
assert_eq!(
table_steps.len(),
1,
"got steps: {:?}",
steps.iter().map(|s| &s.prompt).collect::<Vec<_>>()
);
assert_eq!(table_steps[0].verb, Verb::Create);
assert_eq!(table_steps[0].prompt, "did you create object type 'catalog::Product'?");
let joined: String = table_steps[0]
.ddl
.iter()
.map(|d| d.sql.as_str())
.collect::<Vec<_>>()
.join("\n");
assert!(
joined.contains("CREATE TABLE IF NOT EXISTS \"catalog\".\"Product\""),
"got:\n{joined}"
);
assert!(joined.contains("pylon_cache_invalidate"), "got:\n{joined}");
}
#[test]
fn test_guidance_bans_a_rejected_type_rename_candidate() {
let schema = SchemaDescriptor {
types: vec![simple_type("default", "Customer", "Customer")],
scalars: vec![],
enums: vec![],
named_tuples: vec![],
globals: vec![],
functions: vec![],
aliases: vec![],
channels: vec![],
..Default::default()
};
let state = DbState {
schemas: vec!["default".into()],
tables: vec![DbTable {
schema: "default".into(),
name: "Person".into(),
columns: vec![
DbColumn {
name: "id".into(),
pg_type: "uuid".into(),
nullable: false,
is_generated: false,
column_default: Some("uuidv7()".into()),
},
DbColumn {
name: "name".into(),
pg_type: "text".into(),
nullable: true,
is_generated: false,
column_default: None,
},
],
foreign_keys: vec![],
indexes: vec![],
checks: vec![],
triggers: vec![],
}],
enums: vec![],
domains: vec![],
..DbState::default()
};
let candidates = detect_type_renames(&schema, &state, &Guidance::default());
assert_eq!(
candidates.len(),
1,
"expected Person -> Customer to be proposed as a rename"
);
let mut guidance = Guidance::default();
guidance.banned_type_renames.insert((
"default".to_string(),
"Person".to_string(),
"default".to_string(),
"Customer".to_string(),
));
let candidates = detect_type_renames(&schema, &state, &guidance);
assert!(
candidates.is_empty(),
"a banned rename candidate must not be re-proposed"
);
}
}