use std::collections::{BTreeMap, BTreeSet};
use super::projection::BucketKey;
use super::schema::{
AppliedSchema, ColumnSchema, EnumSchema, ExclusionConstraintSchema, ForeignKeySchema,
GeneratedColumnSchema, IndexSchema, PkKindSchema, TableSchema,
};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SchemaDelta {
pub bucket: BucketKey,
pub operations: Vec<SchemaOperation>,
pub classification: Classification,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SchemaOperation {
AddTable(TableSchema),
DropTable(String),
RenameTable { from: String, to: String },
AddColumn { table: String, column: ColumnSchema },
DropColumn { table: String, column: String },
RenameColumn {
table: String,
from: String,
to: String,
},
AlterColumn {
table: String,
column: String,
change: ColumnChange,
},
AddForeignKey {
table: String,
column: String,
fk: ForeignKeySchema,
},
DropForeignKey {
table: String,
column: String,
fk: ForeignKeySchema,
},
AddIndex(IndexSchema),
DropIndex(IndexSchema),
AddExclusionConstraint {
table: String,
exclusion: ExclusionConstraintSchema,
},
DropExclusionConstraint {
table: String,
name: String,
exclusion: ExclusionConstraintSchema,
},
AddEnum(EnumSchema),
DropEnum(String),
AddEnumVariant {
enum_name: String,
variant: String,
anchor: Option<EnumVariantAnchor>,
},
PkTypeFlip {
table: String,
from: PkKindSchema,
to: PkKindSchema,
},
PkTypeFlipGroup(PkTypeFlipGroup),
PkTypeFlipMultiGroup(Vec<PkTypeFlipGroup>),
RenameApp { from: String, to: String },
MoveModelBetweenApps {
model: String,
from_app: String,
to_app: String,
},
SetTableComment {
table: String,
from: Option<String>,
to: Option<String>,
},
SetStorageParams {
table: String,
from: Option<String>,
to: Option<String>,
},
SetTablespace {
table: String,
from: Option<String>,
to: Option<String>,
},
Unsupported { reason: String },
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ColumnChange {
SetNullable(bool),
SetDefault(Option<String>),
ChangeType {
from: String,
to: String,
using: Option<String>,
},
CodecChange {
from_codec: Option<String>,
to_codec: Option<String>,
},
SetCheck {
from: Option<String>,
to: Option<String>,
},
SetUnique(bool),
SetIndexed(bool),
SetGenerated {
from: Option<GeneratedColumnSchema>,
to: Option<GeneratedColumnSchema>,
},
SetComment {
from: Option<String>,
to: Option<String>,
},
SetIdentity {
from: Option<crate::migrate::schema::IdentityKindSchema>,
to: Option<crate::migrate::schema::IdentityKindSchema>,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnumVariantAnchor {
pub variant: String,
pub kind: EnumVariantAnchorKind,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EnumVariantAnchorKind {
Before,
After,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PkTypeFlipGroup {
pub parent_table: String,
pub parent_from: PkKindSchema,
pub parent_to: PkKindSchema,
pub direction: PkFlipDirection,
pub children: Vec<PkFlipChild>,
pub self_fk: Option<PkFlipSelfFk>,
pub join_tables: Vec<PkFlipJoinTable>,
pub cycles: Vec<PkFlipCycle>,
pub partitioned_parent: Option<PkFlipPartitionedMeta>,
pub co_destructive: bool,
pub co_lossy: bool,
pub join_table_option: PkFlipJoinTableOption,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PkFlipJoinTableOption {
OptionA,
OptionB,
}
impl PkFlipJoinTableOption {
pub fn from_config_char(c: char) -> Self {
match c {
'B' | 'b' => Self::OptionB,
_ => Self::OptionA,
}
}
}
pub fn apply_pk_flip_join_table_option(deltas: &mut [SchemaDelta], option: PkFlipJoinTableOption) {
for delta in deltas.iter_mut() {
for op in &mut delta.operations {
if let SchemaOperation::PkTypeFlipGroup(group) = op {
group.join_table_option = option;
} else if let SchemaOperation::PkTypeFlipMultiGroup(groups) = op {
for g in groups.iter_mut() {
g.join_table_option = option;
}
}
}
}
if option != PkFlipJoinTableOption::OptionA {
return;
}
for delta in deltas.iter_mut() {
merge_cross_flipping_groups_into_multi(delta).expect(
"partitioned multi-parent clusters must already be rejected by diff_bucket_maps",
);
}
}
pub(crate) fn partitioned_multi_parent_cluster_error(
groups: &[PkTypeFlipGroup],
) -> Option<DiffError> {
if groups.len() < 2 {
return None;
}
let mut partitioned_parents: Vec<String> = groups
.iter()
.filter(|g| g.partitioned_parent.is_some())
.map(|g| g.parent_table.clone())
.collect();
if partitioned_parents.is_empty() {
return None;
}
partitioned_parents.sort();
partitioned_parents.dedup();
let mut cross_flipping_partners: Vec<String> =
groups.iter().map(|g| g.parent_table.clone()).collect();
cross_flipping_partners.sort();
cross_flipping_partners.dedup();
Some(DiffError::PartitionedMultiParentClusterUnsupported {
partitioned_parents,
cross_flipping_partners,
})
}
fn cluster_pk_flip_groups(ops: &[SchemaOperation]) -> BTreeMap<String, Vec<&PkTypeFlipGroup>> {
let groups: Vec<&PkTypeFlipGroup> = ops
.iter()
.filter_map(|op| match op {
SchemaOperation::PkTypeFlipGroup(group) => Some(group),
_ => None,
})
.collect();
let mut peer_edges: BTreeSet<(String, String)> = BTreeSet::new();
let mut rep: BTreeMap<String, String> = BTreeMap::new();
for group in &groups {
rep.insert(group.parent_table.clone(), group.parent_table.clone());
for jt in &group.join_tables {
if let Some(partner) = jt.fk_to_partner_table.as_deref() {
let a = group.parent_table.clone();
let b = partner.to_string();
let edge = if a <= b { (a, b) } else { (b, a) };
peer_edges.insert(edge);
}
}
}
if peer_edges.is_empty() {
return BTreeMap::new();
}
fn find(rep: &mut BTreeMap<String, String>, x: &str) -> String {
let mut cur = x.to_string();
loop {
let parent = rep.get(&cur).cloned().unwrap_or_else(|| cur.clone());
if parent == cur {
return cur;
}
let grand = rep.get(&parent).cloned().unwrap_or_else(|| parent.clone());
rep.insert(cur.clone(), grand.clone());
cur = grand;
}
}
for (a, b) in &peer_edges {
let ra = find(&mut rep, a);
let rb = find(&mut rep, b);
if ra != rb {
let (winner, loser) = if ra <= rb { (ra, rb) } else { (rb, ra) };
rep.insert(loser, winner);
}
}
let mut clusters: BTreeMap<String, Vec<&PkTypeFlipGroup>> = BTreeMap::new();
for group in groups {
let root = find(&mut rep, &group.parent_table);
clusters.entry(root).or_default().push(group);
}
clusters
}
fn reject_partitioned_multi_parent_clusters(ops: &[SchemaOperation]) -> Result<(), DiffError> {
let clusters = cluster_pk_flip_groups(ops);
for (_root, mut cluster) in clusters {
if cluster.len() < 2 {
continue;
}
cluster.sort_by(|a, b| a.parent_table.cmp(&b.parent_table));
let member_set: BTreeSet<String> = cluster.iter().map(|g| g.parent_table.clone()).collect();
let has_internal_cross_flip = cluster.iter().any(|group| {
group.join_tables.iter().any(|jt| {
jt.fk_to_partner_table
.as_ref()
.is_some_and(|partner| member_set.contains(partner))
})
});
if !has_internal_cross_flip {
continue;
}
let cluster_groups: Vec<PkTypeFlipGroup> = cluster.into_iter().cloned().collect();
if let Some(err) = partitioned_multi_parent_cluster_error(&cluster_groups) {
return Err(err);
}
}
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DiffError {
PkFlipCascadeDepthExceeded {
parent_table: String,
chain: Vec<String>,
max_depth: u32,
},
PartitionedMultiParentClusterUnsupported {
partitioned_parents: Vec<String>,
cross_flipping_partners: Vec<String>,
},
PkFlipMalformedSelfFkMetadata(super::pk_flip::PkFlipError),
}
impl From<super::pk_flip::PkFlipError> for DiffError {
fn from(err: super::pk_flip::PkFlipError) -> Self {
DiffError::PkFlipMalformedSelfFkMetadata(err)
}
}
impl std::fmt::Display for DiffError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
DiffError::PkFlipCascadeDepthExceeded {
parent_table,
chain,
max_depth,
} => write!(
f,
"PK-flip transitive FK closure exceeded {max_depth} levels rooted at \
{parent_table}; table_chain={chain:?}; graph likely has a pathological \
cycle or unbounded fan-out — refusing to compose the migration",
),
DiffError::PkFlipMalformedSelfFkMetadata(err) => {
write!(f, "diff lowering rejected a malformed PK-flip group: {err}")
}
DiffError::PartitionedMultiParentClusterUnsupported {
partitioned_parents,
cross_flipping_partners,
} => write!(
f,
"PK-flip cross-flipping cluster mixes partitioned parents \
{partitioned_parents:?} with partners {cross_flipping_partners:?}; \
this unsupported multi-parent partitioned shape is rejected",
),
}
}
}
impl std::error::Error for DiffError {}
fn merge_cross_flipping_groups_into_multi(delta: &mut SchemaDelta) -> Result<(), DiffError> {
let clusters: BTreeMap<String, Vec<String>> = cluster_pk_flip_groups(&delta.operations)
.into_iter()
.map(|(root, members)| {
(
root,
members
.into_iter()
.map(|group| group.parent_table.clone())
.collect(),
)
})
.collect();
let mut multi_groups: Vec<Vec<PkTypeFlipGroup>> = Vec::new();
for (_rep, members) in clusters {
if members.len() < 2 {
continue;
}
let mem_set: std::collections::BTreeSet<String> = members.iter().cloned().collect();
let mut cluster_groups: Vec<PkTypeFlipGroup> = delta
.operations
.extract_if(.., |op| {
matches!(
op,
SchemaOperation::PkTypeFlipGroup(g) if mem_set.contains(&g.parent_table)
)
})
.filter_map(|op| match op {
SchemaOperation::PkTypeFlipGroup(g) => Some(g),
_ => None,
})
.collect();
cluster_groups.sort_by(|a, b| a.parent_table.cmp(&b.parent_table));
if let Some(err) = partitioned_multi_parent_cluster_error(&cluster_groups) {
return Err(err);
}
let cross_flipping_jt_names: std::collections::BTreeSet<String> = cluster_groups
.iter()
.flat_map(|g| {
g.join_tables
.iter()
.filter(|jt| jt.fk_to_partner_column.is_some())
.map(|jt| jt.table.clone())
})
.collect();
for (idx, g) in cluster_groups.iter_mut().enumerate() {
if idx == 0 {
continue;
}
g.join_tables.retain(|jt| {
!(jt.fk_to_partner_column.is_some() && cross_flipping_jt_names.contains(&jt.table))
});
}
let winner_existing: std::collections::BTreeSet<String> = cluster_groups[0]
.join_tables
.iter()
.filter(|jt| jt.fk_to_partner_column.is_some())
.map(|jt| jt.table.clone())
.collect();
for jt_name in &cross_flipping_jt_names {
if !winner_existing.contains(jt_name) {
continue;
}
}
multi_groups.push(cluster_groups);
}
for groups in multi_groups {
delta
.operations
.push(SchemaOperation::PkTypeFlipMultiGroup(groups));
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PkFlipDirection {
AscToDesc,
DescToAsc,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PkFlipFamily {
Heer,
Ranj,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PkFlipChild {
pub table: String,
pub fk_column: String,
pub fk_constraint_name: String,
pub on_delete: super::schema::OnDeleteSchema,
pub fk_deferrable: bool,
pub fk_initially_deferred: bool,
pub fk_nullable: bool,
pub fk_unique: bool,
pub family: PkFlipFamily,
pub cycle_flag: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PkFlipSelfFk {
pub fk_columns: Vec<String>,
pub fk_constraint_names: Vec<String>,
pub fk_deferrable: Vec<bool>,
pub fk_initially_deferred: Vec<bool>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PkFlipJoinTable {
pub table: String,
pub fk_to_parent_column: String,
pub fk_to_parent_constraint: String,
pub fk_to_parent_deferrable: bool,
pub fk_to_parent_initially_deferred: bool,
pub fk_to_partner_column: Option<String>,
pub fk_to_partner_constraint: Option<String>,
pub fk_to_partner_table: Option<String>,
pub fk_to_partner_deferrable: bool,
pub fk_to_partner_initially_deferred: bool,
pub family: PkFlipFamily,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PkFlipCycle {
pub peer_table: String,
pub peer_fk_column: String,
pub self_fk_column: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PkFlipPartitionedMeta {
pub partition: super::schema::PartitionSchema,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Classification {
NoOp,
Additive,
Reversible,
Destructive,
Lossy,
Unsupported { reason: String },
PkTypeFlip {
co_destructive: bool,
co_lossy: bool,
},
}
pub(crate) fn diff_schemas(
before: &AppliedSchema,
after: &AppliedSchema,
bucket: BucketKey,
) -> SchemaDelta {
if before == after {
return SchemaDelta {
bucket,
operations: Vec::new(),
classification: Classification::NoOp,
};
}
let mut ops: Vec<SchemaOperation> = Vec::new();
let table_rename_targets: BTreeMap<&str, &str> = after
.models
.values()
.filter_map(|t| {
t.renamed_from
.as_deref()
.filter(|prev| before.models.contains_key(*prev))
.map(|prev| (prev, t.table.as_str()))
})
.collect();
let renamed_table_destinations: BTreeSet<&str> =
table_rename_targets.values().copied().collect();
diff_tables(
before,
after,
&table_rename_targets,
&renamed_table_destinations,
&mut ops,
);
diff_indexes(before, after, &table_rename_targets, &mut ops);
diff_enums(before, after, &mut ops);
let classification = classify(&ops);
SchemaDelta {
bucket,
operations: ops,
classification,
}
}
pub fn diff_bucket_maps(
before: &BTreeMap<BucketKey, AppliedSchema>,
after: &BTreeMap<BucketKey, AppliedSchema>,
) -> Result<Vec<SchemaDelta>, DiffError> {
let mut moves: Vec<(BucketKey, BucketKey, String)> = Vec::new();
let mut suppressed_drops: BTreeSet<(BucketKey, String)> = BTreeSet::new();
let mut suppressed_adds: BTreeSet<(BucketKey, String)> = BTreeSet::new();
for (dest_bucket, dest_schema) in after {
for (table_name, table) in &dest_schema.models {
let Some(from_app) = table.moved_from_app.as_deref() else {
continue;
};
let src_bucket = BucketKey {
database: dest_bucket.database.clone(),
app: from_app.to_string(),
};
let Some(src_schema) = before.get(&src_bucket) else {
continue;
};
if !src_schema.models.contains_key(table_name) {
continue;
}
moves.push((src_bucket.clone(), dest_bucket.clone(), table_name.clone()));
suppressed_drops.insert((src_bucket, table_name.clone()));
suppressed_adds.insert((dest_bucket.clone(), table_name.clone()));
}
}
let mut buckets: BTreeSet<BucketKey> = BTreeSet::new();
buckets.extend(before.keys().cloned());
buckets.extend(after.keys().cloned());
let mut out = Vec::with_capacity(buckets.len());
let empty = AppliedSchema {
djogi_version: String::new(),
enums: BTreeMap::new(),
format_version: super::schema::SNAPSHOT_FORMAT_VERSION.to_string(),
generated_at: String::new(),
indexes: Vec::new(),
models: BTreeMap::new(),
registered_apps: Vec::new(),
};
for bucket in buckets {
let b = before.get(&bucket).unwrap_or(&empty);
let a = after.get(&bucket).unwrap_or(&empty);
let mut delta = diff_schemas(b, a, bucket.clone());
delta.operations.retain(|op| match op {
SchemaOperation::DropTable(name) => {
!suppressed_drops.contains(&(bucket.clone(), name.clone()))
}
SchemaOperation::AddTable(t) => {
!suppressed_adds.contains(&(bucket.clone(), t.table.clone()))
}
_ => true,
});
for (src_bucket, dest_bucket, model) in &moves {
if dest_bucket == &bucket {
delta
.operations
.push(SchemaOperation::MoveModelBetweenApps {
model: model.clone(),
from_app: src_bucket.app.clone(),
to_app: dest_bucket.app.clone(),
});
}
}
promote_pk_flips_to_groups(a, &mut delta.operations)?;
reject_partitioned_multi_parent_clusters(&delta.operations)?;
delta.classification = classify(&delta.operations);
out.push(delta);
}
Ok(out)
}
fn promote_pk_flips_to_groups(
after: &AppliedSchema,
ops: &mut Vec<SchemaOperation>,
) -> Result<(), DiffError> {
let mut parents: Vec<(String, PkKindSchema, PkKindSchema)> = Vec::new();
ops.retain(|op| match op {
SchemaOperation::PkTypeFlip { table, from, to } => {
parents.push((table.clone(), from.clone(), to.clone()));
false
}
_ => true,
});
if parents.is_empty() {
return Ok(());
}
let migrating_parents: BTreeSet<String> = parents.iter().map(|(t, _, _)| t.clone()).collect();
fn family_for_kind(k: &PkKindSchema) -> Option<PkFlipFamily> {
match k {
PkKindSchema::HeerId | PkKindSchema::HeerIdRecencyBiased => Some(PkFlipFamily::Heer),
PkKindSchema::RanjId | PkKindSchema::RanjIdRecencyBiased => Some(PkFlipFamily::Ranj),
_ => None,
}
}
for (parent_table, parent_from, parent_to) in parents {
let direction = match (&parent_from, &parent_to) {
(PkKindSchema::HeerId, PkKindSchema::HeerIdRecencyBiased)
| (PkKindSchema::RanjId, PkKindSchema::RanjIdRecencyBiased) => {
PkFlipDirection::AscToDesc
}
(PkKindSchema::HeerIdRecencyBiased, PkKindSchema::HeerId)
| (PkKindSchema::RanjIdRecencyBiased, PkKindSchema::RanjId) => {
PkFlipDirection::DescToAsc
}
_ => PkFlipDirection::AscToDesc,
};
let family = family_for_kind(&parent_from).unwrap_or(PkFlipFamily::Heer);
let mut children: Vec<PkFlipChild> = Vec::new();
let mut self_fk_cols: Vec<String> = Vec::new();
let mut self_fk_constraints: Vec<String> = Vec::new();
let mut self_fk_deferrable: Vec<bool> = Vec::new();
let mut self_fk_initially_deferred: Vec<bool> = Vec::new();
let mut join_tables: Vec<PkFlipJoinTable> = Vec::new();
let mut cycles: Vec<PkFlipCycle> = Vec::new();
for (other_table_name, other_table) in &after.models {
for col in &other_table.columns {
let Some(fk) = col.foreign_key.as_ref() else {
continue;
};
if fk.ref_table != parent_table {
continue;
}
let constraint_name = format!("{}_{}_fkey", other_table_name, col.name);
if other_table_name == &parent_table {
self_fk_cols.push(col.name.clone());
self_fk_constraints.push(constraint_name);
self_fk_deferrable.push(fk.deferrable);
self_fk_initially_deferred.push(fk.initially_deferred);
continue;
}
let cycle_back = if let Some(parent_schema) = after.models.get(&parent_table) {
parent_schema.columns.iter().find_map(|pc| {
let pfk = pc.foreign_key.as_ref()?;
if pfk.ref_table == *other_table_name {
Some(pc.name.clone())
} else {
None
}
})
} else {
None
};
if let Some(self_fk_col) = cycle_back {
cycles.push(PkFlipCycle {
peer_table: other_table_name.clone(),
peer_fk_column: col.name.clone(),
self_fk_column: self_fk_col,
});
children.push(PkFlipChild {
table: other_table_name.clone(),
fk_column: col.name.clone(),
fk_constraint_name: constraint_name,
on_delete: fk.on_delete,
fk_deferrable: true,
fk_initially_deferred: true,
fk_nullable: col.nullable,
fk_unique: col.unique,
family,
cycle_flag: true,
});
continue;
}
if other_table.is_through {
let partner = other_table.columns.iter().find_map(|pc| {
if pc.name == col.name {
return None;
}
let pfk = pc.foreign_key.as_ref()?;
if pfk.ref_table == parent_table {
return None;
}
Some((pc.name.clone(), pfk.ref_table.clone()))
});
let (partner_col, partner_constraint, partner_table) = match partner {
Some((pcol, partner_target))
if migrating_parents.contains(&partner_target) =>
{
let pcons = format!("{}_{}_fkey", other_table_name, pcol);
(Some(pcol), Some(pcons), Some(partner_target))
}
_ => (None, None, None),
};
let (partner_def, partner_init_def) =
if let Some(partner_col_name) = partner_col.as_deref() {
other_table
.columns
.iter()
.find(|c| c.name == partner_col_name)
.and_then(|c| c.foreign_key.as_ref())
.map(|fk| (fk.deferrable, fk.initially_deferred))
.unwrap_or((false, false))
} else {
(false, false)
};
join_tables.push(PkFlipJoinTable {
table: other_table_name.clone(),
fk_to_parent_column: col.name.clone(),
fk_to_parent_constraint: constraint_name,
fk_to_parent_deferrable: fk.deferrable,
fk_to_parent_initially_deferred: fk.initially_deferred,
fk_to_partner_column: partner_col,
fk_to_partner_constraint: partner_constraint,
fk_to_partner_table: partner_table,
fk_to_partner_deferrable: partner_def,
fk_to_partner_initially_deferred: partner_init_def,
family,
});
continue;
}
children.push(PkFlipChild {
table: other_table_name.clone(),
fk_column: col.name.clone(),
fk_constraint_name: constraint_name,
on_delete: fk.on_delete,
fk_deferrable: fk.deferrable,
fk_initially_deferred: fk.initially_deferred,
fk_nullable: col.nullable,
fk_unique: col.unique,
family,
cycle_flag: false,
});
}
}
let mut visited_tables: BTreeSet<String> = BTreeSet::new();
visited_tables.insert(parent_table.clone());
for c in &children {
visited_tables.insert(c.table.clone());
}
for jt in &join_tables {
visited_tables.insert(jt.table.clone());
}
const MAX_CLOSURE_DEPTH: u32 = 65;
let mut depth_chain: Vec<String> = vec![parent_table.clone()];
let mut frontier: BTreeSet<String> = visited_tables.clone();
for depth in 1u32..=MAX_CLOSURE_DEPTH {
if frontier.is_empty() {
break;
}
let mut next_frontier: BTreeSet<String> = BTreeSet::new();
for (other_name, other_schema) in &after.models {
if visited_tables.contains(other_name) {
continue;
}
for col in &other_schema.columns {
let Some(fk) = col.foreign_key.as_ref() else {
continue;
};
if frontier.contains(&fk.ref_table) {
next_frontier.insert(other_name.clone());
break;
}
}
}
if next_frontier.is_empty() {
break;
}
if let Some(first) = next_frontier.iter().next() {
depth_chain.push(first.clone());
}
for t in &next_frontier {
visited_tables.insert(t.clone());
}
frontier = next_frontier;
if depth == MAX_CLOSURE_DEPTH {
return Err(DiffError::PkFlipCascadeDepthExceeded {
parent_table: parent_table.clone(),
chain: depth_chain.clone(),
max_depth: MAX_CLOSURE_DEPTH,
});
}
}
children.sort_by(|a, b| a.table.cmp(&b.table).then(a.fk_column.cmp(&b.fk_column)));
join_tables.sort_by(|a, b| {
a.table
.cmp(&b.table)
.then(a.fk_to_parent_column.cmp(&b.fk_to_parent_column))
});
cycles.sort_by(|a, b| a.peer_table.cmp(&b.peer_table));
let mut self_fk_zipped: Vec<(String, String, bool, bool)> = self_fk_cols
.into_iter()
.zip(self_fk_constraints)
.zip(self_fk_deferrable)
.zip(self_fk_initially_deferred)
.map(|(((c, n), d), id)| (c, n, d, id))
.collect();
self_fk_zipped.sort_by(|a, b| a.0.cmp(&b.0));
let self_fk = if self_fk_zipped.is_empty() {
None
} else {
let mut cols: Vec<String> = Vec::with_capacity(self_fk_zipped.len());
let mut cons: Vec<String> = Vec::with_capacity(self_fk_zipped.len());
let mut deferr: Vec<bool> = Vec::with_capacity(self_fk_zipped.len());
let mut init_def: Vec<bool> = Vec::with_capacity(self_fk_zipped.len());
for (c, n, d, id) in self_fk_zipped {
cols.push(c);
cons.push(n);
deferr.push(d);
init_def.push(id);
}
if !cycles.is_empty() {
for d in deferr.iter_mut() {
*d = true;
}
for d in init_def.iter_mut() {
*d = true;
}
}
Some(PkFlipSelfFk {
fk_columns: cols,
fk_constraint_names: cons,
fk_deferrable: deferr,
fk_initially_deferred: init_def,
})
};
let partitioned_parent = after
.models
.get(&parent_table)
.and_then(|t| t.partition.as_ref())
.map(|p| PkFlipPartitionedMeta {
partition: p.clone(),
});
let co_destructive = ops.iter().any(|op| {
matches!(
op,
SchemaOperation::DropTable(_)
| SchemaOperation::DropColumn { .. }
| SchemaOperation::DropEnum(_)
| SchemaOperation::DropIndex(_)
| SchemaOperation::DropForeignKey { .. }
)
});
let co_lossy = ops.iter().any(|op| match op {
SchemaOperation::AlterColumn { change, .. } => {
matches!(change, ColumnChange::SetNullable(false))
}
SchemaOperation::AddColumn { column, .. } => {
!column.nullable && column.default_sql.is_none()
}
_ => false,
});
ops.push(SchemaOperation::PkTypeFlipGroup(PkTypeFlipGroup {
parent_table,
parent_from,
parent_to,
direction,
children,
self_fk,
join_tables,
cycles,
partitioned_parent,
co_destructive,
co_lossy,
join_table_option: PkFlipJoinTableOption::OptionA,
}));
}
Ok(())
}
fn diff_tables(
before: &AppliedSchema,
after: &AppliedSchema,
table_rename_targets: &BTreeMap<&str, &str>,
renamed_table_destinations: &BTreeSet<&str>,
ops: &mut Vec<SchemaOperation>,
) {
for old_name in before.models.keys() {
if after.models.contains_key(old_name) {
continue;
}
if table_rename_targets.contains_key(old_name.as_str()) {
continue;
}
ops.push(SchemaOperation::DropTable(old_name.clone()));
}
for new_table in after.models.values() {
if before.models.contains_key(&new_table.table) {
continue;
}
if renamed_table_destinations.contains(new_table.table.as_str()) {
continue;
}
ops.push(SchemaOperation::AddTable(new_table.clone()));
}
for (from, to) in table_rename_targets {
let before_table = &before.models[*from];
let after_table = &after.models[*to];
ops.push(SchemaOperation::RenameTable {
from: (*from).to_string(),
to: (*to).to_string(),
});
let column_renames = diff_columns_in_table(before_table, after_table, ops);
diff_pk_in_table(before_table, after_table, &column_renames, ops);
diff_exclusion_constraints_in_table(before_table, after_table, ops);
diff_table_metadata_in_table(before_table, after_table, ops);
}
for (name, after_table) in &after.models {
let Some(before_table) = before.models.get(name) else {
continue;
};
if before_table == after_table {
continue;
}
let column_renames = diff_columns_in_table(before_table, after_table, ops);
diff_pk_in_table(before_table, after_table, &column_renames, ops);
diff_app_move_in_table(before_table, after_table, ops);
diff_exclusion_constraints_in_table(before_table, after_table, ops);
diff_table_metadata_in_table(before_table, after_table, ops);
}
}
fn diff_table_metadata_in_table(
before: &TableSchema,
after: &TableSchema,
ops: &mut Vec<SchemaOperation>,
) {
if before.table_comment != after.table_comment {
ops.push(SchemaOperation::SetTableComment {
table: after.table.clone(),
from: before.table_comment.clone(),
to: after.table_comment.clone(),
});
}
if before.storage_params != after.storage_params {
ops.push(SchemaOperation::SetStorageParams {
table: after.table.clone(),
from: before.storage_params.clone(),
to: after.storage_params.clone(),
});
}
if before.tablespace != after.tablespace {
ops.push(SchemaOperation::SetTablespace {
table: after.table.clone(),
from: before.tablespace.clone(),
to: after.tablespace.clone(),
});
}
}
fn diff_exclusion_constraints_in_table(
before: &TableSchema,
after: &TableSchema,
ops: &mut Vec<SchemaOperation>,
) {
let before_by_name: BTreeMap<&str, &ExclusionConstraintSchema> = before
.exclusion_constraints
.iter()
.map(|e| (e.name.as_str(), e))
.collect();
let after_by_name: BTreeMap<&str, &ExclusionConstraintSchema> = after
.exclusion_constraints
.iter()
.map(|e| (e.name.as_str(), e))
.collect();
for (name, before_excl) in &before_by_name {
if after_by_name.contains_key(name) {
continue;
}
ops.push(SchemaOperation::DropExclusionConstraint {
table: after.table.clone(),
name: (*name).to_string(),
exclusion: (*before_excl).clone(),
});
}
for (name, after_excl) in &after_by_name {
match before_by_name.get(name) {
Some(before_excl) if before_excl == after_excl => {}
Some(before_excl) => {
ops.push(SchemaOperation::DropExclusionConstraint {
table: after.table.clone(),
name: (*name).to_string(),
exclusion: (*before_excl).clone(),
});
ops.push(SchemaOperation::AddExclusionConstraint {
table: after.table.clone(),
exclusion: (*after_excl).clone(),
});
}
None => {
ops.push(SchemaOperation::AddExclusionConstraint {
table: after.table.clone(),
exclusion: (*after_excl).clone(),
});
}
}
}
}
fn diff_columns_in_table<'a>(
before: &TableSchema,
after: &'a TableSchema,
ops: &mut Vec<SchemaOperation>,
) -> BTreeMap<&'a str, &'a str> {
let before_cols: BTreeMap<&str, &ColumnSchema> = before
.columns
.iter()
.map(|c| (c.name.as_str(), c))
.collect();
let after_cols: BTreeMap<&str, &ColumnSchema> =
after.columns.iter().map(|c| (c.name.as_str(), c)).collect();
let column_rename_targets: BTreeMap<&str, &str> = after_cols
.values()
.filter_map(|c| {
c.renamed_from
.as_deref()
.filter(|prev| before_cols.contains_key(prev))
.map(|prev| (prev, c.name.as_str()))
})
.collect();
let renamed_destinations: BTreeSet<&str> = column_rename_targets.values().copied().collect();
for old_name in before_cols.keys() {
if after_cols.contains_key(old_name) {
continue;
}
if column_rename_targets.contains_key(old_name) {
continue;
}
ops.push(SchemaOperation::DropColumn {
table: after.table.clone(),
column: (*old_name).to_string(),
});
}
for new_col in after.columns.iter() {
if before_cols.contains_key(new_col.name.as_str()) {
continue;
}
if renamed_destinations.contains(new_col.name.as_str()) {
continue;
}
ops.push(SchemaOperation::AddColumn {
table: after.table.clone(),
column: new_col.clone(),
});
}
for (from, to) in &column_rename_targets {
let bc = before_cols[from];
let ac = after_cols[to];
ops.push(SchemaOperation::RenameColumn {
table: after.table.clone(),
from: (*from).to_string(),
to: (*to).to_string(),
});
emit_alter_column(after, bc, ac, ops);
}
for (name, ac) in &after_cols {
let Some(bc) = before_cols.get(name) else {
continue;
};
if bc == ac {
continue;
}
emit_alter_column(after, bc, ac, ops);
}
column_rename_targets
}
fn emit_alter_column(
parent: &TableSchema,
before: &ColumnSchema,
after: &ColumnSchema,
ops: &mut Vec<SchemaOperation>,
) {
let table = parent.table.clone();
{
let mut push = |change: ColumnChange| {
ops.push(SchemaOperation::AlterColumn {
table: table.clone(),
column: after.name.clone(),
change,
});
};
let type_changed = before.sql_type != after.sql_type;
let codec_changed = before.codec != after.codec;
if type_changed && !codec_changed && before.check.is_some() {
push(ColumnChange::SetCheck {
from: before.check.clone(),
to: None,
});
}
if type_changed && !codec_changed {
push(ColumnChange::ChangeType {
from: before.sql_type.clone(),
to: after.sql_type.clone(),
using: after.type_change_using.clone(),
});
}
if codec_changed {
push(ColumnChange::CodecChange {
from_codec: before.codec.clone(),
to_codec: after.codec.clone(),
});
}
if before.nullable != after.nullable {
push(ColumnChange::SetNullable(after.nullable));
}
if before.default_sql != after.default_sql {
push(ColumnChange::SetDefault(after.default_sql.clone()));
}
if type_changed && !codec_changed && before.check.is_some() {
if let Some(check) = &after.check {
push(ColumnChange::SetCheck {
from: None,
to: Some(check.clone()),
});
}
} else {
match (&before.check, &after.check) {
(None, None) => {} (Some(b), Some(a)) if b == a => {} (Some(b), Some(a)) => {
push(ColumnChange::SetCheck {
from: Some(b.clone()),
to: None,
});
push(ColumnChange::SetCheck {
from: None,
to: Some(a.clone()),
});
}
(None, Some(a)) => {
push(ColumnChange::SetCheck {
from: None,
to: Some(a.clone()),
});
}
(Some(b), None) => {
push(ColumnChange::SetCheck {
from: Some(b.clone()),
to: None,
});
}
}
}
if before.unique != after.unique {
push(ColumnChange::SetUnique(after.unique));
}
if before.generated != after.generated {
push(ColumnChange::SetGenerated {
from: before.generated.clone(),
to: after.generated.clone(),
});
}
if before.identity != after.identity {
push(ColumnChange::SetIdentity {
from: before.identity,
to: after.identity,
});
}
if before.comment != after.comment {
push(ColumnChange::SetComment {
from: before.comment.clone(),
to: after.comment.clone(),
});
}
}
match (&before.foreign_key, &after.foreign_key) {
(None, Some(fk)) => ops.push(SchemaOperation::AddForeignKey {
table: table.clone(),
column: after.name.clone(),
fk: fk.clone(),
}),
(Some(b_fk), None) => ops.push(SchemaOperation::DropForeignKey {
table,
column: after.name.clone(),
fk: b_fk.clone(),
}),
(Some(b_fk), Some(a_fk)) if b_fk != a_fk => {
ops.push(SchemaOperation::DropForeignKey {
table: table.clone(),
column: after.name.clone(),
fk: b_fk.clone(),
});
ops.push(SchemaOperation::AddForeignKey {
table,
column: after.name.clone(),
fk: a_fk.clone(),
});
}
_ => {}
}
}
fn diff_pk_in_table(
before: &TableSchema,
after: &TableSchema,
column_renames: &BTreeMap<&str, &str>,
ops: &mut Vec<SchemaOperation>,
) {
if before.primary_key == after.primary_key {
return;
}
let normalised_before: Vec<&str> = before
.primary_key
.columns
.iter()
.map(|c| {
column_renames
.get(c.as_str())
.copied()
.unwrap_or(c.as_str())
})
.collect();
let normalised_after: Vec<&str> = after
.primary_key
.columns
.iter()
.map(|c| c.as_str())
.collect();
let columns_match = normalised_before == normalised_after;
if columns_match && is_pk_kind_flip(&before.primary_key.kind, &after.primary_key.kind) {
ops.push(SchemaOperation::PkTypeFlip {
table: after.table.clone(),
from: before.primary_key.kind.clone(),
to: after.primary_key.kind.clone(),
});
return;
}
if matches!(&before.primary_key.kind, PkKindSchema::Custom(_))
|| matches!(&after.primary_key.kind, PkKindSchema::Custom(_))
{
ops.push(SchemaOperation::Unsupported {
reason: custom_pk_unsupported_reason(
&after.table,
&before.primary_key.kind,
&after.primary_key.kind,
),
});
return;
}
ops.push(SchemaOperation::Unsupported {
reason: format!(
"table `{}`: primary key change is not auto-supported \
({:?} → {:?}). Recognised auto-flips: HeerId ↔ \
HeerIdRecencyBiased, RanjId ↔ RanjIdRecencyBiased \
with identical column lists. Hand-write this migration.",
after.table, before.primary_key, after.primary_key
),
});
}
fn custom_pk_unsupported_reason(
table: &str,
before: &PkKindSchema,
after: &PkKindSchema,
) -> String {
fn describe(kind: &PkKindSchema) -> String {
match kind {
PkKindSchema::Custom(c) => format!(
"Custom(type_name = `{}`, sql_type = `{}`, default_sql = `{}`)",
c.type_name, c.sql_type, c.default_sql,
),
other => format!("{other:?}"),
}
}
let before_desc = describe(before);
let after_desc = describe(after);
let bucket = match (before, after) {
(PkKindSchema::Custom(_), PkKindSchema::Custom(_)) => "custom-to-custom",
(PkKindSchema::Custom(_), _) => "custom-to-built-in",
(_, PkKindSchema::Custom(_)) => "built-in-to-custom",
_ => "non-custom",
};
format!(
"table `{table}`: primary key change involves a \
`djogi::primary_key!` custom newtype ({bucket}: {before_desc} → \
{after_desc}) and is not auto-supported in v0.1.0. The \
`pk_flip` family only ships migration playbooks for the four \
built-in asc↔desc pairs (HeerId ↔ HeerIdRecencyBiased, \
RanjId ↔ RanjIdRecencyBiased); transitions involving a custom \
PK newtype must be hand-written so the operator can decide on \
the value-preserving cast and the FK cascade strategy. See \
`docs/spec/migrations.md` §10.10a for the v0.1.0 support \
matrix and rationale."
)
}
fn is_pk_kind_flip(before: &PkKindSchema, after: &PkKindSchema) -> bool {
matches!(
(before, after),
(PkKindSchema::HeerId, PkKindSchema::HeerIdRecencyBiased)
| (PkKindSchema::HeerIdRecencyBiased, PkKindSchema::HeerId)
| (PkKindSchema::RanjId, PkKindSchema::RanjIdRecencyBiased)
| (PkKindSchema::RanjIdRecencyBiased, PkKindSchema::RanjId)
)
}
fn diff_app_move_in_table(
before: &TableSchema,
after: &TableSchema,
ops: &mut Vec<SchemaOperation>,
) {
if before.app == after.app {
return;
}
let from_app = before.app.clone().unwrap_or_default();
let to_app = after.app.clone().unwrap_or_default();
ops.push(SchemaOperation::MoveModelBetweenApps {
model: after.table.clone(),
from_app,
to_app,
});
}
fn diff_indexes(
before: &AppliedSchema,
after: &AppliedSchema,
table_rename_targets: &BTreeMap<&str, &str>,
ops: &mut Vec<SchemaOperation>,
) {
let before_idx: BTreeMap<&str, &IndexSchema> = before
.indexes
.iter()
.map(|i| (i.name.as_str(), i))
.collect();
let after_idx: BTreeMap<&str, &IndexSchema> =
after.indexes.iter().map(|i| (i.name.as_str(), i)).collect();
for (name, ai) in &after_idx {
let resolved_old = before_idx.get(name).copied();
match resolved_old {
Some(bi) => {
let resolved_table_match = match table_rename_targets.get(bi.table.as_str()) {
Some(new_name) => *new_name == ai.table.as_str(),
None => bi.table == ai.table,
};
let logical_eq = resolved_table_match
&& bi.target == ai.target
&& bi.kind == ai.kind
&& bi.index_type == ai.index_type
&& bi.predicate == ai.predicate
&& bi.include == ai.include
&& bi.nulls_not_distinct == ai.nulls_not_distinct
&& bi.requires_out_of_transaction == ai.requires_out_of_transaction
&& bi.extension_dependency == ai.extension_dependency;
if !logical_eq {
ops.push(SchemaOperation::DropIndex((*bi).clone()));
ops.push(SchemaOperation::AddIndex((*ai).clone()));
}
}
None => {
ops.push(SchemaOperation::AddIndex((*ai).clone()));
}
}
}
for (name, bi) in &before_idx {
if after_idx.contains_key(name) {
continue;
}
ops.push(SchemaOperation::DropIndex((*bi).clone()));
}
}
fn diff_enums(before: &AppliedSchema, after: &AppliedSchema, ops: &mut Vec<SchemaOperation>) {
for (name, ae) in &after.enums {
match before.enums.get(name) {
None => ops.push(SchemaOperation::AddEnum((*ae).clone())),
Some(be) if be == ae => {}
Some(be) => {
let before_set: BTreeSet<&str> = be.variants.iter().map(|v| v.as_str()).collect();
for (i, v) in ae.variants.iter().enumerate() {
if !before_set.contains(v.as_str()) {
let anchor = pick_enum_variant_anchor(&ae.variants, i, &before_set);
ops.push(SchemaOperation::AddEnumVariant {
enum_name: name.clone(),
variant: v.clone(),
anchor,
});
}
}
let after_set: BTreeSet<&str> = ae.variants.iter().map(|v| v.as_str()).collect();
for v in &be.variants {
if !after_set.contains(v.as_str()) {
ops.push(SchemaOperation::Unsupported {
reason: format!(
"enum `{name}`: variant `{v}` removed. Postgres has \
no `DROP VALUE`; rebuild the type via a hand-written \
migration (drop dependent columns, drop type, \
recreate type without the variant, add columns \
back)."
),
});
}
}
}
}
}
for name in before.enums.keys() {
if !after.enums.contains_key(name) {
ops.push(SchemaOperation::DropEnum(name.clone()));
}
}
}
fn pick_enum_variant_anchor(
new_variants: &[String],
pos: usize,
before_set: &BTreeSet<&str>,
) -> Option<EnumVariantAnchor> {
for v in new_variants.iter().skip(pos + 1) {
if before_set.contains(v.as_str()) {
return Some(EnumVariantAnchor {
variant: v.clone(),
kind: EnumVariantAnchorKind::Before,
});
}
}
for v in new_variants[..pos].iter().rev() {
if before_set.contains(v.as_str()) {
return Some(EnumVariantAnchor {
variant: v.clone(),
kind: EnumVariantAnchorKind::After,
});
}
}
None
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum Severity {
Additive = 0,
Reversible = 1,
Destructive = 2,
Lossy = 3,
}
fn severity_of(op: &SchemaOperation) -> Severity {
match op {
SchemaOperation::RenameTable { .. }
| SchemaOperation::RenameColumn { .. }
| SchemaOperation::RenameApp { .. }
| SchemaOperation::MoveModelBetweenApps { .. } => Severity::Reversible,
SchemaOperation::DropTable(_)
| SchemaOperation::DropColumn { .. }
| SchemaOperation::DropEnum(_)
| SchemaOperation::DropIndex(_)
| SchemaOperation::DropForeignKey { .. } => Severity::Destructive,
SchemaOperation::AlterColumn { change, .. } => {
if matches!(change, ColumnChange::SetNullable(false)) {
Severity::Lossy
} else {
Severity::Additive
}
}
SchemaOperation::AddColumn { column, .. } => {
if !column.nullable && column.default_sql.is_none() && column.generated.is_none() {
Severity::Lossy
} else {
Severity::Additive
}
}
SchemaOperation::AddTable(_)
| SchemaOperation::SetTableComment { .. }
| SchemaOperation::SetStorageParams { .. }
| SchemaOperation::SetTablespace { .. }
| SchemaOperation::AddIndex(_)
| SchemaOperation::AddEnum(_)
| SchemaOperation::AddEnumVariant { .. }
| SchemaOperation::AddForeignKey { .. }
| SchemaOperation::AddExclusionConstraint { .. }
| SchemaOperation::PkTypeFlip { .. }
| SchemaOperation::PkTypeFlipGroup(_)
| SchemaOperation::PkTypeFlipMultiGroup(_)
| SchemaOperation::Unsupported { .. } => Severity::Additive,
SchemaOperation::DropExclusionConstraint { .. } => Severity::Destructive,
}
}
pub(crate) fn classify_operations(ops: &[SchemaOperation]) -> Classification {
classify(ops)
}
fn classify(ops: &[SchemaOperation]) -> Classification {
if ops.is_empty() {
return Classification::NoOp;
}
if let Some(reason) = ops.iter().find_map(|op| match op {
SchemaOperation::Unsupported { reason } => Some(reason.clone()),
_ => None,
}) {
return Classification::Unsupported { reason };
}
let has_pk_flip = ops.iter().any(|op| {
matches!(
op,
SchemaOperation::PkTypeFlip { .. }
| SchemaOperation::PkTypeFlipGroup(_)
| SchemaOperation::PkTypeFlipMultiGroup(_)
)
});
if has_pk_flip {
let mut co_destructive = false;
let mut co_lossy = false;
for op in ops {
match severity_of(op) {
Severity::Destructive => co_destructive = true,
Severity::Lossy => co_lossy = true,
Severity::Additive | Severity::Reversible => {}
}
}
return Classification::PkTypeFlip {
co_destructive,
co_lossy,
};
}
let max = ops
.iter()
.map(severity_of)
.max()
.unwrap_or(Severity::Additive);
match max {
Severity::Additive => Classification::Additive,
Severity::Reversible => Classification::Reversible,
Severity::Destructive => Classification::Destructive,
Severity::Lossy => Classification::Lossy,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::apps::AppDescriptor;
use crate::descriptor::{
EnumDescriptor, FieldDescriptor, FieldSqlType, IndexColumnSpec, IndexKind, IndexSpec,
IndexTarget, IndexType, ModelDescriptor, PkType, field_descriptor, model_descriptor,
};
use crate::migrate::projection::project_from_iters;
use crate::migrate::schema::{IndexTypeSchema, PrimaryKeySchema, SNAPSHOT_FORMAT_VERSION};
fn synth_app(label: &'static str, database: &'static str) -> AppDescriptor {
AppDescriptor {
label,
database,
renamed_from: None,
tombstone: false,
}
}
fn synth_model(table: &'static str, type_name: &'static str) -> ModelDescriptor {
ModelDescriptor {
..model_descriptor(type_name, table, PkType::HeerIdDesc, &[])
}
}
fn empty_global() -> BucketKey {
BucketKey {
database: "main".to_string(),
app: "".to_string(),
}
}
fn project_one(m: &ModelDescriptor) -> AppliedSchema {
let mut buckets = project_from_iters(
[m],
std::iter::empty::<&EnumDescriptor>(),
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("project");
buckets.remove(&empty_global()).unwrap()
}
fn project_empty() -> AppliedSchema {
let mut buckets = project_from_iters(
std::iter::empty::<&ModelDescriptor>(),
std::iter::empty::<&EnumDescriptor>(),
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("project");
buckets.remove(&empty_global()).unwrap()
}
#[test]
fn equal_schemas_yield_noop() {
let m = synth_model("widgets", "Widget");
let s = project_one(&m);
let delta = diff_schemas(&s, &s, empty_global());
assert_eq!(delta.classification, Classification::NoOp);
assert!(delta.operations.is_empty());
}
#[test]
fn add_table_is_additive() {
let before = project_empty();
let m = synth_model("widgets", "Widget");
let after = project_one(&m);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Additive);
assert!(matches!(
delta.operations.first(),
Some(SchemaOperation::AddTable(t)) if t.table == "widgets"
));
}
#[test]
fn drop_table_is_destructive() {
let m = synth_model("widgets", "Widget");
let before = project_one(&m);
let after = project_empty();
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Destructive);
assert!(matches!(
delta.operations.first(),
Some(SchemaOperation::DropTable(t)) if t == "widgets"
));
}
#[test]
fn renamed_table_is_reversible_not_destructive() {
let old_m = synth_model("widgets", "Widget");
let new_m = ModelDescriptor {
table_name: "gadgets",
renamed_from: Some("widgets"),
..synth_model("widgets", "Widget")
};
let before = project_one(&old_m);
let after = project_one(&new_m);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Reversible);
assert_eq!(delta.operations.len(), 1);
assert!(matches!(
&delta.operations[0],
SchemaOperation::RenameTable { from, to } if from == "widgets" && to == "gadgets"
));
}
#[test]
fn unannotated_table_swap_is_destructive_drop_plus_add() {
let old_m = synth_model("widgets", "Widget");
let new_m = synth_model("gadgets", "Gadget");
let before = project_one(&old_m);
let after = project_one(&new_m);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Destructive);
let kinds: Vec<_> = delta
.operations
.iter()
.map(|op| match op {
SchemaOperation::DropTable(_) => "drop",
SchemaOperation::AddTable(_) => "add",
_ => "other",
})
.collect();
assert!(kinds.contains(&"drop"));
assert!(kinds.contains(&"add"));
}
#[test]
fn add_column_is_additive() {
const NAME: FieldDescriptor = field_descriptor("name", FieldSqlType::Text, true);
static FIELDS_WITH_NAME: &[FieldDescriptor] = &[NAME];
let bare = synth_model("widgets", "Widget");
let new_field = ModelDescriptor {
fields: FIELDS_WITH_NAME,
..synth_model("widgets", "Widget")
};
let before = project_one(&bare);
let after = project_one(&new_field);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Additive);
assert!(matches!(
delta.operations.first(),
Some(SchemaOperation::AddColumn { table, column }) if table == "widgets" && column.name == "name"
));
}
#[test]
fn pk_flip_classifies_as_pk_type_flip() {
let asc = ModelDescriptor {
pk_type: PkType::HeerId,
..synth_model("widgets", "Widget")
};
let desc = ModelDescriptor {
pk_type: PkType::HeerIdDesc,
..synth_model("widgets", "Widget")
};
let before = project_one(&asc);
let after = project_one(&desc);
let delta = diff_schemas(&before, &after, empty_global());
assert!(matches!(
delta.classification,
Classification::PkTypeFlip {
co_destructive: false,
co_lossy: false
}
));
assert!(delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::PkTypeFlip {
from: PkKindSchema::HeerId,
to: PkKindSchema::HeerIdRecencyBiased,
..
}
)));
}
#[test]
fn pk_flip_reverse_direction_also_classifies() {
let desc = ModelDescriptor {
pk_type: PkType::HeerIdDesc,
..synth_model("widgets", "Widget")
};
let asc = ModelDescriptor {
pk_type: PkType::HeerId,
..synth_model("widgets", "Widget")
};
let before = project_one(&desc);
let after = project_one(&asc);
let delta = diff_schemas(&before, &after, empty_global());
assert!(matches!(
delta.classification,
Classification::PkTypeFlip { .. }
));
}
#[test]
fn pk_unrelated_change_classifies_as_unsupported() {
let heer = ModelDescriptor {
pk_type: PkType::HeerId,
..synth_model("widgets", "Widget")
};
let serial = ModelDescriptor {
pk_type: PkType::Serial,
..synth_model("widgets", "Widget")
};
let before = project_one(&heer);
let after = project_one(&serial);
let delta = diff_schemas(&before, &after, empty_global());
assert!(matches!(
delta.classification,
Classification::Unsupported { .. }
));
}
fn synth_model_with_pk(
table: &'static str,
type_name: &'static str,
pk: PkType,
) -> ModelDescriptor {
ModelDescriptor {
pk_type: pk,
..synth_model(table, type_name)
}
}
fn unsupported_pk_reason(before_pk: PkType, after_pk: PkType) -> String {
let before = project_one(&synth_model_with_pk("users", "User", before_pk));
let after = project_one(&synth_model_with_pk("users", "User", after_pk));
let delta = diff_schemas(&before, &after, empty_global());
match delta.classification {
Classification::Unsupported { reason } => reason,
other => panic!("expected Unsupported, got {other:?}"),
}
}
const CUSTOM_USER_ID: PkType = PkType::Custom(crate::descriptor::CustomPrimaryKeyKind {
type_name: "crate::ids::UserId",
sql_type: "BIGINT",
default_sql: "user_id_next()",
});
const CUSTOM_USER_ID_V2: PkType = PkType::Custom(crate::descriptor::CustomPrimaryKeyKind {
type_name: "crate::ids::UserIdV2",
sql_type: "BIGINT",
default_sql: "user_id_next()",
});
const CUSTOM_USER_ID_UUID: PkType = PkType::Custom(crate::descriptor::CustomPrimaryKeyKind {
type_name: "crate::ids::UserId",
sql_type: "UUID",
default_sql: "gen_random_uuid()",
});
const CUSTOM_USER_ID_NEXT_V2: PkType =
PkType::Custom(crate::descriptor::CustomPrimaryKeyKind {
type_name: "crate::ids::UserId",
sql_type: "BIGINT",
default_sql: "user_id_next_v2()",
});
#[test]
fn pk_custom_same_inner_type_different_newtype_is_unsupported() {
let reason = unsupported_pk_reason(CUSTOM_USER_ID, CUSTOM_USER_ID_V2);
assert!(
reason.contains("custom-to-custom"),
"diagnostic must label the bucket; got: {reason}"
);
assert!(
reason.contains("crate::ids::UserId") && reason.contains("crate::ids::UserIdV2"),
"diagnostic must surface both type_names; got: {reason}"
);
assert!(
reason.contains("djogi::primary_key!"),
"diagnostic must mention the macro so operators know where to look; got: {reason}"
);
}
#[test]
fn pk_custom_changed_inner_sql_type_is_unsupported() {
let reason = unsupported_pk_reason(CUSTOM_USER_ID, CUSTOM_USER_ID_UUID);
assert!(
reason.contains("custom-to-custom"),
"diagnostic must label the bucket; got: {reason}"
);
assert!(
reason.contains("BIGINT") && reason.contains("UUID"),
"diagnostic must surface both inner SQL types; got: {reason}"
);
}
#[test]
fn pk_custom_changed_default_sql_is_unsupported() {
let reason = unsupported_pk_reason(CUSTOM_USER_ID, CUSTOM_USER_ID_NEXT_V2);
assert!(
reason.contains("custom-to-custom"),
"diagnostic must label the bucket; got: {reason}"
);
assert!(
reason.contains("user_id_next()") && reason.contains("user_id_next_v2()"),
"diagnostic must surface both default_sql generators so the \
operator can see what changed; got: {reason}"
);
}
#[test]
fn pk_builtin_to_custom_is_unsupported() {
let reason = unsupported_pk_reason(PkType::HeerId, CUSTOM_USER_ID);
assert!(
reason.contains("built-in-to-custom"),
"diagnostic must label the bucket; got: {reason}"
);
assert!(
reason.contains("crate::ids::UserId"),
"diagnostic must surface the custom side's type_name; got: {reason}"
);
assert!(
reason.contains("HeerId"),
"diagnostic must surface the built-in side; got: {reason}"
);
}
#[test]
fn pk_custom_to_builtin_is_unsupported() {
let reason = unsupported_pk_reason(CUSTOM_USER_ID, PkType::HeerId);
assert!(
reason.contains("custom-to-built-in"),
"diagnostic must label the bucket; got: {reason}"
);
assert!(
reason.contains("crate::ids::UserId") && reason.contains("HeerId"),
"diagnostic must surface both sides; got: {reason}"
);
}
#[test]
fn pk_custom_unchanged_is_noop() {
let m = synth_model_with_pk("users", "User", CUSTOM_USER_ID);
let s = project_one(&m);
let delta = diff_schemas(&s, &s, empty_global());
assert_eq!(delta.classification, Classification::NoOp);
assert!(delta.operations.is_empty());
}
#[test]
fn add_index_is_additive() {
let bare = synth_model("widgets", "Widget");
static IDX_SLICE: &[IndexSpec] = &[IndexSpec {
name: "widgets_name_idx",
target: IndexTarget::Columns(&[IndexColumnSpec::simple("name")]),
kind: IndexKind::NonUnique,
index_type: IndexType::BTree,
predicate: None,
include: &[],
nulls_not_distinct: false,
requires_out_of_transaction: false,
extension_dependency: None,
}];
let with_idx = ModelDescriptor {
indexes: IDX_SLICE,
..bare.clone()
};
let before = project_one(&bare);
let after = project_one(&with_idx);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Additive);
assert!(matches!(
delta.operations.first(),
Some(SchemaOperation::AddIndex(idx)) if idx.name == "widgets_name_idx"
));
}
#[test]
fn add_enum_is_additive() {
let status_field = field_descriptor("status", FieldSqlType::Custom("status"), true);
let fields: &'static [FieldDescriptor] = Box::leak(Box::new([status_field]));
let m = model_descriptor("Widget", "widgets", PkType::HeerIdDesc, fields);
let e = EnumDescriptor {
type_name: "Status",
postgres_type: "status",
variants: &["active", "deleted"],
};
let mut buckets_before = project_from_iters(
[&m],
std::iter::empty::<&EnumDescriptor>(),
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("p");
let mut buckets_after = project_from_iters(
[&m],
[&e],
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("p");
let before = buckets_before.remove(&empty_global()).unwrap();
let after = buckets_after.remove(&empty_global()).unwrap();
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Additive);
assert!(
delta
.operations
.iter()
.any(|op| matches!(op, SchemaOperation::AddEnum(_)))
);
}
#[test]
fn variant_removal_classifies_as_unsupported() {
let status_field = field_descriptor("status", FieldSqlType::Custom("status"), true);
let fields: &'static [FieldDescriptor] = Box::leak(Box::new([status_field]));
let m = model_descriptor("Widget", "widgets", PkType::HeerIdDesc, fields);
let two = EnumDescriptor {
type_name: "Status",
postgres_type: "status",
variants: &["active", "deleted"],
};
let one = EnumDescriptor {
type_name: "Status",
postgres_type: "status",
variants: &["active"],
};
let mut buckets_before = project_from_iters(
[&m],
[&two],
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("p");
let mut buckets_after = project_from_iters(
[&m],
[&one],
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("p");
let before = buckets_before.remove(&empty_global()).unwrap();
let after = buckets_after.remove(&empty_global()).unwrap();
let delta = diff_schemas(&before, &after, empty_global());
assert!(matches!(
delta.classification,
Classification::Unsupported { .. }
));
}
#[test]
fn diff_bucket_maps_handles_added_bucket() {
let m = synth_model("widgets", "Widget");
let before = BTreeMap::new();
let mut after = BTreeMap::new();
after.insert(empty_global(), project_one(&m));
let deltas = diff_bucket_maps(&before, &after).expect("differ must succeed in this test");
assert_eq!(deltas.len(), 1);
assert_eq!(deltas[0].bucket, empty_global());
assert!(matches!(deltas[0].classification, Classification::Additive));
}
#[test]
fn renamed_column_emits_rename_not_drop_add() {
const OLD_NAME: FieldDescriptor = field_descriptor("old_name", FieldSqlType::Text, true);
static OLD_SLICE: &[FieldDescriptor] = &[OLD_NAME];
const NEW_NAME: FieldDescriptor = FieldDescriptor {
renamed_from: Some("old_name"),
..field_descriptor("new_name", FieldSqlType::Text, true)
};
static NEW_SLICE: &[FieldDescriptor] = &[NEW_NAME];
let bare = ModelDescriptor {
fields: OLD_SLICE,
..synth_model("widgets", "Widget")
};
let renamed = ModelDescriptor {
fields: NEW_SLICE,
..synth_model("widgets", "Widget")
};
let before = project_one(&bare);
let after = project_one(&renamed);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Reversible);
assert!(matches!(
delta.operations.first(),
Some(SchemaOperation::RenameColumn { from, to, .. }) if from == "old_name" && to == "new_name"
));
}
#[test]
fn nullability_change_to_not_null_classifies_lossy() {
const NULLABLE: FieldDescriptor = field_descriptor("name", FieldSqlType::Text, true);
const NOT_NULL: FieldDescriptor = field_descriptor("name", FieldSqlType::Text, false);
static NULLABLE_SLICE: &[FieldDescriptor] = &[NULLABLE];
static NOT_NULL_SLICE: &[FieldDescriptor] = &[NOT_NULL];
let nullable = ModelDescriptor {
fields: NULLABLE_SLICE,
..synth_model("widgets", "Widget")
};
let not_null = ModelDescriptor {
fields: NOT_NULL_SLICE,
..synth_model("widgets", "Widget")
};
let before = project_one(&nullable);
let after = project_one(¬_null);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Lossy);
}
#[test]
fn nullability_change_to_nullable_is_additive() {
const NOT_NULL: FieldDescriptor = field_descriptor("name", FieldSqlType::Text, false);
const NULLABLE: FieldDescriptor = field_descriptor("name", FieldSqlType::Text, true);
static NOT_NULL_SLICE: &[FieldDescriptor] = &[NOT_NULL];
static NULLABLE_SLICE: &[FieldDescriptor] = &[NULLABLE];
let not_null = ModelDescriptor {
fields: NOT_NULL_SLICE,
..synth_model("widgets", "Widget")
};
let nullable = ModelDescriptor {
fields: NULLABLE_SLICE,
..synth_model("widgets", "Widget")
};
let before = project_one(¬_null);
let after = project_one(&nullable);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Additive);
}
#[test]
fn type_change_emits_alter_column_change_type() {
const TEXT: FieldDescriptor = field_descriptor("amount", FieldSqlType::Text, true);
const BIGINT: FieldDescriptor = field_descriptor("amount", FieldSqlType::BigInt, true);
static TEXT_SLICE: &[FieldDescriptor] = &[TEXT];
static BIGINT_SLICE: &[FieldDescriptor] = &[BIGINT];
let text = ModelDescriptor {
fields: TEXT_SLICE,
..synth_model("widgets", "Widget")
};
let bigint = ModelDescriptor {
fields: BIGINT_SLICE,
..synth_model("widgets", "Widget")
};
let before = project_one(&text);
let after = project_one(&bigint);
let delta = diff_schemas(&before, &after, empty_global());
assert!(delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::AlterColumn {
change: ColumnChange::ChangeType { .. },
..
}
)));
}
fn build_table_with_check(check: Option<&str>) -> crate::migrate::schema::AppliedSchema {
build_table_with_check_and_type(check, "BIGINT")
}
fn build_table_with_check_and_type(
check: Option<&str>,
sql_type: &str,
) -> crate::migrate::schema::AppliedSchema {
use crate::migrate::schema::{
AppliedSchema, ColumnSchema, PkKindSchema, PrimaryKeySchema, TableSchema,
};
use std::collections::BTreeMap;
let id_col = ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: Some("heerid_next()".to_string()),
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "id".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
};
let amount_col = ColumnSchema {
check: check.map(|s| s.to_string()),
codec: None,
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "amount".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: sql_type.to_string(),
unique: false,
type_change_using: None,
};
let mut models = BTreeMap::new();
models.insert(
"widgets".to_string(),
TableSchema {
app: None,
columns: vec![id_col, amount_col],
exclusion_constraints: vec![],
fts: None,
is_through: false,
moved_from_app: None,
partition: None,
primary_key: PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::HeerId,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: "widgets".to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
},
);
AppliedSchema {
djogi_version: "0.1.0".to_string(),
enums: BTreeMap::new(),
format_version: "1".to_string(),
generated_at: "2026-05-10T00:00:00Z".to_string(),
indexes: vec![],
models,
registered_apps: vec!["".to_string()],
}
}
fn alter_column_changes_for(
delta: &crate::migrate::diff::SchemaDelta,
column: &str,
) -> Vec<ColumnChange> {
delta
.operations
.iter()
.filter_map(|op| match op {
SchemaOperation::AlterColumn {
column: c, change, ..
} if c == column => Some(change.clone()),
_ => None,
})
.collect()
}
#[test]
fn check_unchanged_some_emits_no_set_check() {
let before = build_table_with_check(Some("\"amount\" >= 0 AND \"amount\" <= 4294967295"));
let after = build_table_with_check(Some("\"amount\" >= 0 AND \"amount\" <= 4294967295"));
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "amount");
assert!(
changes.is_empty(),
"identical CHECK on both sides must not emit any AlterColumn change: {changes:?}",
);
}
#[test]
fn check_unchanged_none_emits_no_set_check() {
let before = build_table_with_check(None);
let after = build_table_with_check(None);
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "amount");
assert!(
changes.is_empty(),
"absent CHECK on both sides must not emit any AlterColumn change: {changes:?}",
);
}
#[test]
fn check_add_emits_single_set_check_some() {
let after_expr = "\"amount\" >= 0 AND \"amount\" <= 4294967295";
let before = build_table_with_check(None);
let after = build_table_with_check(Some(after_expr));
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "amount");
assert_eq!(
changes.len(),
1,
"ADD CHECK is a single ColumnChange entry: {changes:?}"
);
assert!(
matches!(
changes.first(),
Some(ColumnChange::SetCheck { from: None, to: Some(s) }) if s == after_expr
),
"ADD CHECK emits SetCheck {{ from: None, to: Some(new) }}: {changes:?}"
);
}
#[test]
fn table_metadata_changes_emit_dedicated_operations() {
let mut before = build_table_with_check(None);
let mut after = build_table_with_check(None);
let before_table = before.models.get_mut("widgets").expect("before table");
before_table.table_comment = Some("old comment".to_string());
before_table.storage_params = Some("fillfactor=80".to_string());
before_table.tablespace = Some("slowspace".to_string());
let after_table = after.models.get_mut("widgets").expect("after table");
after_table.table_comment = Some("new comment".to_string());
after_table.storage_params = Some("fillfactor=70, autovacuum_enabled=false".to_string());
after_table.tablespace = Some("fastspace".to_string());
let delta = diff_schemas(&before, &after, empty_global());
assert!(delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::SetTableComment {
table,
from: Some(from),
to: Some(to),
} if table == "widgets" && from == "old comment" && to == "new comment"
)));
assert!(delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::SetStorageParams {
table,
from: Some(from),
to: Some(to),
} if table == "widgets"
&& from == "fillfactor=80"
&& to == "fillfactor=70, autovacuum_enabled=false"
)));
assert!(delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::SetTablespace {
table,
from: Some(from),
to: Some(to),
} if table == "widgets" && from == "slowspace" && to == "fastspace"
)));
}
#[test]
fn check_drop_emits_single_set_check_none() {
let prior_expr = "\"amount\" >= 0 AND \"amount\" <= 4294967295";
let before = build_table_with_check(Some(prior_expr));
let after = build_table_with_check(None);
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "amount");
assert_eq!(
changes.len(),
1,
"DROP CHECK is a single ColumnChange entry: {changes:?}"
);
assert!(
matches!(
changes.first(),
Some(ColumnChange::SetCheck { from: Some(s), to: None }) if s == prior_expr
),
"DROP CHECK emits SetCheck {{ from: Some(prior), to: None }}: {changes:?}"
);
}
#[test]
fn check_amend_emits_drop_then_add() {
let before_expr = "\"amount\" >= 0 AND \"amount\" <= 4294967295";
let after_expr = "\"amount\" >= 0 AND \"amount\" <= 18446744073709551615";
let before = build_table_with_check(Some(before_expr));
let after = build_table_with_check(Some(after_expr));
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "amount");
assert_eq!(
changes.len(),
2,
"AMEND CHECK emits exactly two ColumnChange entries: {changes:?}"
);
assert!(
matches!(
changes.first(),
Some(ColumnChange::SetCheck { from: Some(b), to: None }) if b == before_expr
),
"AMEND step 1: SetCheck {{ from: Some(prior), to: None }}: {changes:?}"
);
assert!(
matches!(
changes.get(1),
Some(ColumnChange::SetCheck { from: None, to: Some(a) }) if a == after_expr
),
"AMEND step 2: SetCheck {{ from: None, to: Some(new) }}: {changes:?}"
);
}
#[test]
fn check_type_change_readds_same_check_after_type_change() {
let check_expr = "\"amount\" >= 0 AND \"amount\" <= 4294967295";
let before = build_table_with_check_and_type(Some(check_expr), "INTEGER");
let after = build_table_with_check_and_type(Some(check_expr), "BIGINT");
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "amount");
assert_eq!(
changes.len(),
3,
"unchanged CHECK plus type migration must emit drop/type/readd steps: {changes:?}"
);
assert!(
matches!(
changes.first(),
Some(ColumnChange::SetCheck { from: Some(b), to: None }) if b == check_expr
),
"existing CHECK must be dropped before ALTER TYPE \
(carrying `from` for rollback): {changes:?}"
);
assert!(
matches!(changes.get(1), Some(ColumnChange::ChangeType { from, to, .. }) if from == "INTEGER" && to == "BIGINT"),
"ALTER TYPE should be between drop and re-add: {changes:?}"
);
assert!(
matches!(
changes.get(2),
Some(ColumnChange::SetCheck { from: None, to: Some(s) }) if s == check_expr
),
"same CHECK should be re-added after type conversion \
(forward-only `to`; the prior is restored by step 0's down): \
{changes:?}"
);
}
#[test]
fn check_type_change_reorders_drop_and_readd_for_changed_check() {
let before_expr = "\"amount\" >= 0 AND \"amount\" <= 4294967295";
let after_expr = "\"amount\" >= 0 AND \"amount\" <= 18446744073709551615";
let before = build_table_with_check_and_type(Some(before_expr), "INTEGER");
let after = build_table_with_check_and_type(Some(after_expr), "BIGINT");
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "amount");
assert_eq!(
changes.len(),
3,
"changed CHECK + type migration must emit drop/type/readd steps: {changes:?}"
);
assert!(
matches!(
changes.first(),
Some(ColumnChange::SetCheck { from: Some(b), to: None }) if b == before_expr
),
"existing CHECK must be dropped before ALTER TYPE \
with `from` carrying the OLD expression for rollback: {changes:?}"
);
assert!(
matches!(changes.get(1), Some(ColumnChange::ChangeType { from, to, .. }) if from == "INTEGER" && to == "BIGINT"),
"ALTER TYPE should be between drop and re-add: {changes:?}"
);
assert!(
matches!(
changes.get(2),
Some(ColumnChange::SetCheck { from: None, to: Some(s) }) if s == after_expr
),
"new CHECK should be re-added after type conversion \
(the OLD expression is restored by step 0's down side): {changes:?}"
);
}
#[test]
fn diff_detects_identity_none_to_some_by_default_as_set_identity() {
use crate::migrate::schema::{
AppliedSchema, ColumnSchema, IdentityKindSchema, PkKindSchema, PrimaryKeySchema,
TableSchema,
};
use std::collections::BTreeMap;
fn build_schema(identity: Option<IdentityKindSchema>) -> AppliedSchema {
let id_col = ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity,
index_type: None,
indexed: false,
max_length: None,
name: "id".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "INTEGER".to_string(),
unique: false,
type_change_using: None,
};
let table = TableSchema {
app: None,
columns: vec![id_col],
exclusion_constraints: Vec::new(),
fts: None,
is_through: false,
moved_from_app: None,
partition: None,
primary_key: PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::Serial,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: "countries".to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
};
let mut models = BTreeMap::new();
models.insert("countries".to_string(), table);
AppliedSchema {
djogi_version: "0.1.0".to_string(),
enums: BTreeMap::new(),
format_version: SNAPSHOT_FORMAT_VERSION.to_string(),
generated_at: "2026-05-02T00:00:00Z".to_string(),
indexes: Vec::new(),
models,
registered_apps: vec!["".to_string()],
}
}
let before = build_schema(None);
let after = build_schema(Some(IdentityKindSchema::ByDefault));
let delta = diff_schemas(&before, &after, empty_global());
let has_set_identity = delta.operations.iter().any(|op| {
matches!(
op,
SchemaOperation::AlterColumn {
change: ColumnChange::SetIdentity {
from: None,
to: Some(IdentityKindSchema::ByDefault)
},
..
}
)
});
assert!(
has_set_identity,
"expected SetIdentity {{ from: None, to: Some(ByDefault) }} in delta; got: {:?}",
delta.operations
);
}
#[test]
fn add_not_null_column_without_default_classifies_lossy() {
const NOT_NULL: FieldDescriptor = field_descriptor("required", FieldSqlType::Text, false);
static FIELDS: &[FieldDescriptor] = &[NOT_NULL];
let bare = synth_model("widgets", "Widget");
let with_required = ModelDescriptor {
fields: FIELDS,
..synth_model("widgets", "Widget")
};
let before = project_one(&bare);
let after = project_one(&with_required);
let delta = diff_schemas(&before, &after, empty_global());
assert_eq!(delta.classification, Classification::Lossy);
}
#[test]
fn pk_flip_with_concurrent_drop_surfaces_co_destructive_flag() {
let asc = ModelDescriptor {
pk_type: PkType::HeerId,
..synth_model("widgets", "Widget")
};
let asc_other = ModelDescriptor {
pk_type: PkType::HeerId,
..synth_model("decommissioned", "Decommissioned")
};
let mut before_buckets = project_from_iters(
[&asc, &asc_other],
std::iter::empty::<&EnumDescriptor>(),
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("project");
let desc = ModelDescriptor {
pk_type: PkType::HeerIdDesc,
..synth_model("widgets", "Widget")
};
let mut after_buckets = project_from_iters(
[&desc],
std::iter::empty::<&EnumDescriptor>(),
std::iter::empty::<&AppDescriptor>(),
"2026-04-25T00:00:00Z".to_string(),
)
.expect("project");
let before = before_buckets.remove(&empty_global()).unwrap();
let after = after_buckets.remove(&empty_global()).unwrap();
let delta = diff_schemas(&before, &after, empty_global());
match delta.classification {
Classification::PkTypeFlip {
co_destructive,
co_lossy,
} => {
assert!(
co_destructive,
"DropTable alongside PkTypeFlip must set co_destructive"
);
assert!(!co_lossy);
}
other => panic!("expected PkTypeFlip with co_destructive, got {other:?}"),
}
}
#[test]
fn pk_flip_with_lossy_only_op_does_not_claim_co_destructive() {
const REQUIRED: FieldDescriptor = field_descriptor("required", FieldSqlType::Text, false);
static FIELDS: &[FieldDescriptor] = &[REQUIRED];
let asc = ModelDescriptor {
pk_type: PkType::HeerId,
..synth_model("widgets", "Widget")
};
let desc_with_required = ModelDescriptor {
pk_type: PkType::HeerIdDesc,
fields: FIELDS,
..synth_model("widgets", "Widget")
};
let before = project_one(&asc);
let after = project_one(&desc_with_required);
let delta = diff_schemas(&before, &after, empty_global());
match delta.classification {
Classification::PkTypeFlip {
co_destructive,
co_lossy,
} => {
assert!(
!co_destructive,
"Lossy-only op alongside PkTypeFlip must NOT set co_destructive"
);
assert!(
co_lossy,
"AddColumn(NOT NULL, no default) alongside PkTypeFlip must set co_lossy"
);
}
other => panic!("expected PkTypeFlip with co_lossy only, got {other:?}"),
}
}
#[test]
fn drop_index_carries_full_metadata() {
static IDX_SLICE: &[IndexSpec] = &[IndexSpec {
name: "widgets_name_idx",
target: IndexTarget::Columns(&[IndexColumnSpec::simple("name")]),
kind: IndexKind::NonUnique,
index_type: IndexType::Gist,
predicate: None,
include: &[],
nulls_not_distinct: false,
requires_out_of_transaction: true,
extension_dependency: Some("postgis"),
}];
let with_idx = ModelDescriptor {
indexes: IDX_SLICE,
..synth_model("widgets", "Widget")
};
let bare = synth_model("widgets", "Widget");
let before = project_one(&with_idx);
let after = project_one(&bare);
let delta = diff_schemas(&before, &after, empty_global());
let drop = delta
.operations
.iter()
.find_map(|op| match op {
SchemaOperation::DropIndex(idx) => Some(idx),
_ => None,
})
.expect("DropIndex emitted");
assert_eq!(drop.name, "widgets_name_idx");
assert!(drop.requires_out_of_transaction);
assert_eq!(drop.extension_dependency.as_deref(), Some("postgis"));
assert_eq!(drop.index_type, IndexTypeSchema::Gist);
}
#[test]
fn cross_bucket_move_via_moved_from_app_emits_move_not_drop_add() {
let billing = synth_app("billing", "main");
let users = synth_app("users", "main");
let before_model = ModelDescriptor {
app: Some("billing"),
..synth_model("user_settings", "UserSettings")
};
let after_model = ModelDescriptor {
app: Some("users"),
moved_from_app: Some("billing"),
..synth_model("user_settings", "UserSettings")
};
let before = project_from_iters(
[&before_model],
std::iter::empty::<&EnumDescriptor>(),
[&billing, &users],
"2026-04-25T00:00:00Z".to_string(),
)
.expect("project before");
let after = project_from_iters(
[&after_model],
std::iter::empty::<&EnumDescriptor>(),
[&billing, &users],
"2026-04-25T00:00:00Z".to_string(),
)
.expect("project after");
let deltas = diff_bucket_maps(&before, &after).expect("differ must succeed in this test");
let billing_bucket = BucketKey {
database: "main".to_string(),
app: "billing".to_string(),
};
let billing_delta = deltas.iter().find(|d| d.bucket == billing_bucket);
if let Some(d) = billing_delta {
assert!(
!d.operations
.iter()
.any(|op| matches!(op, SchemaOperation::DropTable(_))),
"source bucket must not emit DropTable for moved model"
);
}
let users_bucket = BucketKey {
database: "main".to_string(),
app: "users".to_string(),
};
let users_delta = deltas.iter().find(|d| d.bucket == users_bucket).unwrap();
assert!(
users_delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::MoveModelBetweenApps { model, from_app, to_app }
if model == "user_settings" && from_app == "billing" && to_app == "users"
)),
"destination bucket must emit MoveModelBetweenApps"
);
assert!(
!users_delta
.operations
.iter()
.any(|op| matches!(op, SchemaOperation::AddTable(_))),
"destination bucket must not emit AddTable for moved model"
);
}
#[test]
fn pk_column_rename_normalises_under_kind_flip() {
let bucket = empty_global();
let mut before = AppliedSchema {
djogi_version: "0.1.0".to_string(),
enums: BTreeMap::new(),
format_version: SNAPSHOT_FORMAT_VERSION.to_string(),
generated_at: "2026-04-25T00:00:00Z".to_string(),
indexes: Vec::new(),
models: BTreeMap::new(),
registered_apps: vec!["".to_string()],
};
before.models.insert(
"widgets".to_string(),
TableSchema {
app: None,
columns: vec![ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "old_id".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
}],
exclusion_constraints: Vec::new(),
fts: None,
is_through: false,
moved_from_app: None,
partition: None,
primary_key: PrimaryKeySchema {
columns: vec!["old_id".to_string()],
kind: PkKindSchema::HeerId,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: "widgets".to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
},
);
let mut after = before.clone();
after.models.insert(
"widgets".to_string(),
TableSchema {
app: None,
columns: vec![ColumnSchema {
name: "new_id".to_string(),
renamed_from: Some("old_id".to_string()),
..before.models["widgets"].columns[0].clone()
}],
primary_key: PrimaryKeySchema {
columns: vec!["new_id".to_string()],
kind: PkKindSchema::HeerIdRecencyBiased,
},
..before.models["widgets"].clone()
},
);
let delta = diff_schemas(&before, &after, bucket);
assert!(
delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::RenameColumn { from, to, .. }
if from == "old_id" && to == "new_id"
)),
"RenameColumn must be emitted before the PK comparison"
);
assert!(
delta.operations.iter().any(|op| matches!(
op,
SchemaOperation::PkTypeFlip {
from: PkKindSchema::HeerId,
to: PkKindSchema::HeerIdRecencyBiased,
..
}
)),
"PK rename + kind flip must classify as PkTypeFlip, not Unsupported. \
Operations were: {:?}",
delta.operations
);
assert!(
!delta
.operations
.iter()
.any(|op| matches!(op, SchemaOperation::Unsupported { .. })),
"PK rename + kind flip must NOT emit Unsupported"
);
assert!(matches!(
delta.classification,
Classification::PkTypeFlip {
co_destructive: false,
co_lossy: false
}
));
}
fn schema_with_enum(name: &str, variants: &[&str]) -> AppliedSchema {
let mut enums = BTreeMap::new();
enums.insert(
name.to_string(),
EnumSchema {
name: name.to_string(),
variants: variants.iter().map(|s| s.to_string()).collect(),
},
);
AppliedSchema {
djogi_version: "0.1.0".to_string(),
enums,
format_version: SNAPSHOT_FORMAT_VERSION.to_string(),
generated_at: "2026-04-25T00:00:00Z".to_string(),
indexes: Vec::new(),
models: BTreeMap::new(),
registered_apps: vec!["".to_string()],
}
}
#[test]
fn enum_variant_inserted_at_head_anchors_before_first_existing() {
let before = schema_with_enum("status", &["b", "c"]);
let after = schema_with_enum("status", &["a", "b", "c"]);
let delta = diff_schemas(&before, &after, empty_global());
let anchor = delta
.operations
.iter()
.find_map(|op| match op {
SchemaOperation::AddEnumVariant {
variant, anchor, ..
} if variant == "a" => Some(anchor.clone()),
_ => None,
})
.expect("AddEnumVariant for `a`");
assert_eq!(
anchor,
Some(EnumVariantAnchor {
variant: "b".to_string(),
kind: EnumVariantAnchorKind::Before,
})
);
}
#[test]
fn enum_variant_inserted_in_middle_anchors_before_next_existing() {
let before = schema_with_enum("status", &["a", "c"]);
let after = schema_with_enum("status", &["a", "b", "c"]);
let delta = diff_schemas(&before, &after, empty_global());
let anchor = delta
.operations
.iter()
.find_map(|op| match op {
SchemaOperation::AddEnumVariant {
variant, anchor, ..
} if variant == "b" => Some(anchor.clone()),
_ => None,
})
.expect("AddEnumVariant for `b`");
assert_eq!(
anchor,
Some(EnumVariantAnchor {
variant: "c".to_string(),
kind: EnumVariantAnchorKind::Before,
})
);
}
#[test]
fn enum_variant_inserted_at_tail_anchors_after_last_existing() {
let before = schema_with_enum("status", &["a", "b"]);
let after = schema_with_enum("status", &["a", "b", "c"]);
let delta = diff_schemas(&before, &after, empty_global());
let anchor = delta
.operations
.iter()
.find_map(|op| match op {
SchemaOperation::AddEnumVariant {
variant, anchor, ..
} if variant == "c" => Some(anchor.clone()),
_ => None,
})
.expect("AddEnumVariant for `c`");
assert_eq!(
anchor,
Some(EnumVariantAnchor {
variant: "b".to_string(),
kind: EnumVariantAnchorKind::After,
})
);
}
#[test]
fn multiple_new_variants_each_anchor_against_an_existing_neighbour() {
let before = schema_with_enum("status", &["b"]);
let after = schema_with_enum("status", &["a", "b", "c"]);
let delta = diff_schemas(&before, &after, empty_global());
let anchor_a = delta
.operations
.iter()
.find_map(|op| match op {
SchemaOperation::AddEnumVariant {
variant, anchor, ..
} if variant == "a" => Some(anchor.clone()),
_ => None,
})
.expect("AddEnumVariant a");
let anchor_c = delta
.operations
.iter()
.find_map(|op| match op {
SchemaOperation::AddEnumVariant {
variant, anchor, ..
} if variant == "c" => Some(anchor.clone()),
_ => None,
})
.expect("AddEnumVariant c");
assert_eq!(
anchor_a,
Some(EnumVariantAnchor {
variant: "b".to_string(),
kind: EnumVariantAnchorKind::Before,
})
);
assert_eq!(
anchor_c,
Some(EnumVariantAnchor {
variant: "b".to_string(),
kind: EnumVariantAnchorKind::After,
})
);
}
fn synth_table_with_fks(
table: &str,
fks: &[(&str, &str)],
pk_kind: crate::migrate::schema::PkKindSchema,
) -> crate::migrate::schema::TableSchema {
use crate::migrate::schema::{
ColumnSchema, ForeignKeySchema, OnDeleteSchema, PrimaryKeySchema, TableSchema,
};
let mut columns = Vec::new();
columns.push(ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "id".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
});
for (col, ref_table) in fks {
columns.push(ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: Some(ForeignKeySchema {
deferrable: false,
initially_deferred: false,
on_delete: OnDeleteSchema::Restrict,
ref_column: "id".to_string(),
ref_table: ref_table.to_string(),
}),
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: col.to_string(),
nullable: false,
on_delete: Some(OnDeleteSchema::Restrict),
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
});
}
TableSchema {
app: None,
columns,
exclusion_constraints: Vec::new(),
fts: None,
is_through: false,
moved_from_app: None,
partition: None,
primary_key: PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: pk_kind,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: table.to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
}
}
fn synth_schema_with_tables(
tables: Vec<crate::migrate::schema::TableSchema>,
) -> crate::migrate::schema::AppliedSchema {
let mut models = BTreeMap::new();
for t in tables {
models.insert(t.table.clone(), t);
}
crate::migrate::schema::AppliedSchema {
djogi_version: "0.1.0".to_string(),
enums: BTreeMap::new(),
format_version: SNAPSHOT_FORMAT_VERSION.to_string(),
generated_at: "2026-04-25T00:00:00Z".to_string(),
indexes: Vec::new(),
models,
registered_apps: vec!["".to_string()],
}
}
fn synth_partitioned_cross_flipping_schema(
left_pk: crate::migrate::schema::PkKindSchema,
right_pk: crate::migrate::schema::PkKindSchema,
) -> crate::migrate::schema::AppliedSchema {
use crate::migrate::schema::{
ColumnSchema, ForeignKeySchema, OnDeleteSchema, PartitionSchema, PrimaryKeySchema,
TableSchema,
};
let left = TableSchema {
app: None,
columns: vec![
ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "id".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
},
ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "ts".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "TIMESTAMPTZ".to_string(),
unique: false,
type_change_using: None,
},
],
exclusion_constraints: Vec::new(),
fts: None,
is_through: false,
moved_from_app: None,
partition: Some(PartitionSchema::Range {
column: "ts".to_string(),
}),
primary_key: PrimaryKeySchema {
columns: vec!["ts".to_string(), "id".to_string()],
kind: left_pk,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: "left_events".to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
};
let right = synth_table_with_fks("right_tags", &[], right_pk);
let join = TableSchema {
app: None,
columns: vec![
ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: Some("heerid_next()".to_string()),
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "id".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
},
ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: Some(ForeignKeySchema {
deferrable: false,
initially_deferred: false,
on_delete: OnDeleteSchema::Restrict,
ref_column: "id".to_string(),
ref_table: "left_events".to_string(),
}),
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "left_event_id".to_string(),
nullable: false,
on_delete: Some(OnDeleteSchema::Restrict),
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
},
ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: None,
foreign_key: Some(ForeignKeySchema {
deferrable: false,
initially_deferred: false,
on_delete: OnDeleteSchema::Restrict,
ref_column: "id".to_string(),
ref_table: "right_tags".to_string(),
}),
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "right_tag_id".to_string(),
nullable: false,
on_delete: Some(OnDeleteSchema::Restrict),
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
},
],
exclusion_constraints: Vec::new(),
fts: None,
is_through: true,
moved_from_app: None,
partition: None,
primary_key: PrimaryKeySchema {
columns: vec!["left_event_id".to_string(), "right_tag_id".to_string()],
kind: crate::migrate::schema::PkKindSchema::Serial,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: "event_tags".to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
};
synth_schema_with_tables(vec![left, right, join])
}
#[test]
fn transitive_fk_closure_walks_four_level_cascade() {
use crate::migrate::schema::PkKindSchema;
let after = synth_schema_with_tables(vec![
synth_table_with_fks("a", &[], PkKindSchema::HeerId),
synth_table_with_fks("b", &[("a_id", "a")], PkKindSchema::HeerId),
synth_table_with_fks("c", &[("b_id", "b")], PkKindSchema::HeerId),
synth_table_with_fks("d", &[("c_id", "c")], PkKindSchema::HeerId),
]);
let mut ops = vec![SchemaOperation::PkTypeFlip {
table: "a".to_string(),
from: PkKindSchema::HeerId,
to: PkKindSchema::HeerIdRecencyBiased,
}];
promote_pk_flips_to_groups(&after, &mut ops).expect("closure must not error in this test");
let group = ops
.iter()
.find_map(|op| match op {
SchemaOperation::PkTypeFlipGroup(g) => Some(g),
_ => None,
})
.expect("PkTypeFlipGroup emitted");
assert_eq!(group.parent_table, "a");
assert_eq!(
group
.children
.iter()
.map(|c| c.table.as_str())
.collect::<Vec<_>>(),
vec!["b"],
"for asc↔desc only depth-1 children become shadow-column targets",
);
}
#[test]
fn transitive_fk_closure_terminates_on_cycle() {
use crate::migrate::schema::PkKindSchema;
let after = synth_schema_with_tables(vec![
synth_table_with_fks("a", &[("b_id", "b")], PkKindSchema::HeerId),
synth_table_with_fks("b", &[("a_id", "a")], PkKindSchema::HeerId),
]);
let mut ops = vec![SchemaOperation::PkTypeFlip {
table: "a".to_string(),
from: PkKindSchema::HeerId,
to: PkKindSchema::HeerIdRecencyBiased,
}];
promote_pk_flips_to_groups(&after, &mut ops).expect("closure must not error in this test");
let group = ops
.iter()
.find_map(|op| match op {
SchemaOperation::PkTypeFlipGroup(g) => Some(g),
_ => None,
})
.expect("PkTypeFlipGroup emitted");
assert_eq!(group.cycles.len(), 1, "cycle peer detected");
assert_eq!(group.cycles[0].peer_table, "b");
let cycle_children: Vec<_> = group
.children
.iter()
.filter(|c| c.cycle_flag)
.map(|c| (c.table.as_str(), c.fk_column.as_str()))
.collect();
assert_eq!(cycle_children, vec![("b", "a_id")]);
}
#[test]
fn transitive_fk_closure_handles_three_level_cascade_without_panic() {
use crate::migrate::schema::PkKindSchema;
let after = synth_schema_with_tables(vec![
synth_table_with_fks("p", &[], PkKindSchema::HeerId),
synth_table_with_fks("c1", &[("p_id", "p")], PkKindSchema::HeerId),
synth_table_with_fks("g1", &[("c_id", "c1")], PkKindSchema::HeerId),
]);
let mut ops = vec![SchemaOperation::PkTypeFlip {
table: "p".to_string(),
from: PkKindSchema::HeerId,
to: PkKindSchema::HeerIdRecencyBiased,
}];
promote_pk_flips_to_groups(&after, &mut ops).expect("closure must not error in this test");
let group = ops
.iter()
.find_map(|op| match op {
SchemaOperation::PkTypeFlipGroup(g) => Some(g),
_ => None,
})
.expect("group");
assert_eq!(group.children.len(), 1);
assert_eq!(group.children[0].table, "c1");
}
#[test]
fn diff_bucket_maps_emits_pk_flip_cascade_depth_exceeded_on_deep_graph() {
use crate::migrate::projection::BucketKey;
use crate::migrate::schema::PkKindSchema;
let build_chain = |p_pk: PkKindSchema| -> AppliedSchema {
let mut tables: Vec<crate::migrate::schema::TableSchema> = Vec::new();
tables.push(synth_table_with_fks("p", &[], p_pk));
for i in 1..=70u32 {
let prev = if i == 1 {
"p".to_string()
} else {
format!("t{}", i - 1)
};
let name = format!("t{i}");
let prev_str = prev.as_str();
let table =
synth_table_with_fks(&name, &[("ref_id", prev_str)], PkKindSchema::HeerId);
tables.push(table);
}
synth_schema_with_tables(tables)
};
let bucket = BucketKey {
database: "main".to_string(),
app: "".to_string(),
};
let before: BTreeMap<BucketKey, AppliedSchema> = {
let mut m = BTreeMap::new();
m.insert(bucket.clone(), build_chain(PkKindSchema::HeerId));
m
};
let after: BTreeMap<BucketKey, AppliedSchema> = {
let mut m = BTreeMap::new();
m.insert(
bucket.clone(),
build_chain(PkKindSchema::HeerIdRecencyBiased),
);
m
};
let err = diff_bucket_maps(&before, &after)
.expect_err("70-level chain must trigger depth contract via diff_bucket_maps");
match err {
DiffError::PkFlipCascadeDepthExceeded {
parent_table,
chain,
max_depth,
} => {
assert_eq!(parent_table, "p");
assert_eq!(max_depth, 65);
assert!(
chain.len() >= 2,
"chain must record at least the parent and one descendant: {chain:?}",
);
assert_eq!(chain[0], "p");
}
other => panic!("unexpected error: {other:?}"),
}
}
#[test]
fn diff_bucket_maps_rejects_partitioned_cross_flipping_cluster() {
use crate::migrate::projection::BucketKey;
use crate::migrate::schema::PkKindSchema;
let bucket = BucketKey {
database: "main".to_string(),
app: "".to_string(),
};
let before: BTreeMap<BucketKey, AppliedSchema> = {
let mut m = BTreeMap::new();
m.insert(
bucket.clone(),
synth_partitioned_cross_flipping_schema(PkKindSchema::HeerId, PkKindSchema::HeerId),
);
m
};
let after: BTreeMap<BucketKey, AppliedSchema> = {
let mut m = BTreeMap::new();
m.insert(
bucket.clone(),
synth_partitioned_cross_flipping_schema(
PkKindSchema::HeerIdRecencyBiased,
PkKindSchema::HeerIdRecencyBiased,
),
);
m
};
let err = diff_bucket_maps(&before, &after)
.expect_err("partitioned + cross-flipping cluster must reject");
match err {
DiffError::PartitionedMultiParentClusterUnsupported {
partitioned_parents,
cross_flipping_partners,
} => {
assert_eq!(partitioned_parents, vec!["left_events".to_string()]);
assert_eq!(
cross_flipping_partners,
vec!["left_events".to_string(), "right_tags".to_string()]
);
}
other => panic!("unexpected error: {other:?}"),
}
}
#[test]
fn pk_flip_cascade_depth_exceeded_display_renders_operator_message() {
let display = format!(
"{}",
DiffError::PkFlipCascadeDepthExceeded {
parent_table: "p".to_string(),
chain: vec!["p".to_string(), "t1".to_string()],
max_depth: 65,
}
);
assert!(display.contains("rooted at p"));
assert!(display.contains("65 levels"));
assert!(display.contains("table_chain"));
}
#[test]
fn text_to_varchar_emits_change_type() {
const TEXT: FieldDescriptor = field_descriptor("slug", FieldSqlType::Text, false);
const VARCHAR: FieldDescriptor =
field_descriptor("slug", FieldSqlType::Varchar(100), false);
static TEXT_SLICE: &[FieldDescriptor] = &[TEXT];
static VARCHAR_SLICE: &[FieldDescriptor] = &[VARCHAR];
let model_text = ModelDescriptor {
fields: TEXT_SLICE,
..synth_model("articles", "Article")
};
let model_varchar = ModelDescriptor {
fields: VARCHAR_SLICE,
..synth_model("articles", "Article")
};
let before = project_one(&model_text);
let after = project_one(&model_varchar);
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "slug");
assert!(
changes.iter().any(|c| matches!(
c,
ColumnChange::ChangeType { from, to, .. }
if from == "TEXT" && to == "VARCHAR(100)"
)),
"TEXT → VARCHAR(100) must emit ColumnChange::ChangeType; got: {changes:?}"
);
}
#[test]
fn varchar_width_change_emits_change_type() {
const V100: FieldDescriptor = field_descriptor("slug", FieldSqlType::Varchar(100), false);
const V200: FieldDescriptor = field_descriptor("slug", FieldSqlType::Varchar(200), false);
static V100_SLICE: &[FieldDescriptor] = &[V100];
static V200_SLICE: &[FieldDescriptor] = &[V200];
let before = project_one(&ModelDescriptor {
fields: V100_SLICE,
..synth_model("articles", "Article")
});
let after = project_one(&ModelDescriptor {
fields: V200_SLICE,
..synth_model("articles", "Article")
});
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "slug");
assert!(
changes.iter().any(|c| matches!(
c,
ColumnChange::ChangeType { from, to, .. }
if from == "VARCHAR(100)" && to == "VARCHAR(200)"
)),
"VARCHAR(100) → VARCHAR(200) must emit ColumnChange::ChangeType; got: {changes:?}"
);
}
fn build_codec_table(
sql_type: &str,
codec: Option<&str>,
) -> crate::migrate::schema::AppliedSchema {
use crate::migrate::schema::{
AppliedSchema, ColumnSchema, PkKindSchema, PrimaryKeySchema, TableSchema,
};
use std::collections::BTreeMap;
let id_col = ColumnSchema {
check: None,
codec: None,
comment: None,
default_sql: Some("heerid_next()".to_string()),
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "id".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: "BIGINT".to_string(),
unique: false,
type_change_using: None,
};
let token_col = ColumnSchema {
check: None,
codec: codec.map(|s| s.to_string()),
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: "token".to_string(),
nullable: false,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: sql_type.to_string(),
unique: false,
type_change_using: None,
};
let mut models = BTreeMap::new();
models.insert(
"vault".to_string(),
TableSchema {
app: None,
columns: vec![id_col, token_col],
exclusion_constraints: vec![],
fts: None,
is_through: false,
moved_from_app: None,
partition: None,
primary_key: PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::HeerId,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: "vault".to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
},
);
AppliedSchema {
djogi_version: "0.1.0".to_string(),
enums: BTreeMap::new(),
format_version: "1".to_string(),
generated_at: "2026-06-10T00:00:00Z".to_string(),
indexes: vec![],
models,
registered_apps: vec!["".to_string()],
}
}
#[test]
fn same_sql_type_codec_swap_emits_codec_change_not_nothing() {
let before = build_codec_table("BYTEA", Some("aes256_gcm_v1"));
let after = build_codec_table("BYTEA", Some("aes256_gcm_v2"));
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "token");
assert_eq!(
changes.len(),
1,
"a same-sql_type codec swap must emit exactly one op; got: {changes:?}"
);
assert!(
matches!(
&changes[0],
ColumnChange::CodecChange { from_codec: Some(f), to_codec: Some(t) }
if f == "aes256_gcm_v1" && t == "aes256_gcm_v2"
),
"expected a CodecChange(v1 -> v2); got: {:?}",
changes[0]
);
assert!(
!changes
.iter()
.any(|c| matches!(c, ColumnChange::ChangeType { .. })),
"no ChangeType op for an unchanged sql_type; got: {changes:?}"
);
}
#[test]
fn plaintext_to_codec_emits_codec_change_and_suppresses_change_type() {
let before = build_codec_table("TEXT", None);
let after = build_codec_table("BYTEA", Some("aes256_gcm_v1"));
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "token");
assert!(
changes.iter().any(|c| matches!(
c,
ColumnChange::CodecChange { from_codec: None, to_codec: Some(t) }
if t == "aes256_gcm_v1"
)),
"plaintext->codec must emit CodecChange(None -> v1); got: {changes:?}"
);
assert!(
!changes
.iter()
.any(|c| matches!(c, ColumnChange::ChangeType { .. })),
"ChangeType must be suppressed when the type change is codec-driven \
(no ::BYTEA cast); got: {changes:?}"
);
}
#[test]
fn codec_drop_emits_codec_change() {
let before = build_codec_table("BYTEA", Some("aes256_gcm_v1"));
let after = build_codec_table("TEXT", None);
let delta = diff_schemas(&before, &after, empty_global());
let changes = alter_column_changes_for(&delta, "token");
assert!(
changes.iter().any(|c| matches!(
c,
ColumnChange::CodecChange { from_codec: Some(f), to_codec: None }
if f == "aes256_gcm_v1"
)),
"codec->plaintext must emit CodecChange(v1 -> None); got: {changes:?}"
);
assert!(
!changes
.iter()
.any(|c| matches!(c, ColumnChange::ChangeType { .. })),
"ChangeType must be suppressed for a codec drop; got: {changes:?}"
);
}
#[test]
fn plaintext_to_codec_classifies_offline_only() {
use crate::live_migrate::classify::{ClassifyContext, classify_operation};
use crate::live_migrate::{LoggingProfile, TargetDatabase};
use crate::migrate::schema::OnlineSafetyClassification;
use std::collections::BTreeMap;
let before = build_codec_table("TEXT", None);
let after = build_codec_table("BYTEA", Some("aes256_gcm_v1"));
let delta = diff_schemas(&before, &after, empty_global());
let codec_op = delta
.operations
.iter()
.find(|op| {
matches!(
op,
SchemaOperation::AlterColumn {
change: ColumnChange::CodecChange { .. },
..
}
)
})
.expect("a CodecChange op must be present for plaintext->codec");
let inbound: BTreeMap<String, u32> = BTreeMap::new();
let overrides: BTreeMap<(String, String), crate::descriptor::DefaultVolatility> =
BTreeMap::new();
let ctx = ClassifyContext {
estimated_rows: Some(1_000),
validation_threshold_rows: 100_000,
multi_fk_threshold: 4,
logging_profile: LoggingProfile::Balanced,
target_database: TargetDatabase::Application,
inbound_fk_counts: &inbound,
default_volatility_overrides: &overrides,
};
assert_eq!(
classify_operation(codec_op, &ctx),
OnlineSafetyClassification::OfflineOnly,
"plaintext->codec must classify OfflineOnly"
);
}
}