0.8.0: MariaDB + PlanetScale, TimescaleDB, multi-result sets, streaming export, schema diff #27
@@ -503,6 +503,490 @@ pub(crate) fn quote_double(name: &str) -> String {
|
|||||||
format!("\"{}\"", name.replace('"', "\"\""))
|
format!("\"{}\"", name.replace('"', "\"\""))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Pure diff engine (spec §4/§5A)
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Destructive classification (binding): DROP TABLE/VIEW/INDEX/SEQUENCE/TYPE/
|
||||||
|
// CONSTRAINT, DROP COLUMN, and ALTER COLUMN TYPE are destructive (copy-only);
|
||||||
|
// everything else (CREATE, ADD COLUMN, ADD CONSTRAINT, RENAME, SET/DROP
|
||||||
|
// DEFAULT, SET/DROP NOT NULL) is stageable.
|
||||||
|
|
||||||
|
pub(crate) fn engine_family(engine: &str) -> &'static str {
|
||||||
|
match engine {
|
||||||
|
"mysql" | "mariadb" => "mysql",
|
||||||
|
"sqlite" => "sqlite",
|
||||||
|
_ => "postgresql",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn quote(family: &str, name: &str) -> String {
|
||||||
|
match family {
|
||||||
|
"mysql" => format!("`{}`", name.replace('`', "``")),
|
||||||
|
_ => format!("\"{}\"", name.replace('"', "\"\"")),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn qualified(family: &str, schema: &str, name: &str) -> String {
|
||||||
|
match family {
|
||||||
|
"mysql" => format!("`{}`.`{}`", schema.replace('`', "``"), name.replace('`', "``")),
|
||||||
|
"sqlite" => quote(family, name),
|
||||||
|
_ => format!("\"{}\".\"{}\"", schema.replace('"', "\"\""), name.replace('"', "\"\"")),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn detail_line(label: &str, old: &str, new: &str) -> DiffDetailLine {
|
||||||
|
DiffDetailLine { label: label.into(), old: old.into(), new: new.into() }
|
||||||
|
}
|
||||||
|
|
||||||
|
fn item(ty: &str, name: &str, kind: &str, detail: Vec<DiffDetailLine>, sql: Vec<String>, destructive: bool) -> DiffItem {
|
||||||
|
DiffItem {
|
||||||
|
object_type: ty.into(),
|
||||||
|
name: name.into(),
|
||||||
|
kind: kind.into(),
|
||||||
|
detail,
|
||||||
|
sync_sql: sql,
|
||||||
|
destructive,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Diff two snapshots into a report. Sync SQL is in the TARGET's dialect.
|
||||||
|
pub fn diff_snapshots(source: &SchemaSnapshot, target: &SchemaSnapshot) -> DiffReport {
|
||||||
|
let family = engine_family(&target.engine);
|
||||||
|
let schema = &target.schema;
|
||||||
|
let mut items: Vec<DiffItem> = Vec::new();
|
||||||
|
|
||||||
|
// ── Enums (PG) — must come first (tables may depend on them) ──────────
|
||||||
|
for (s_enum, t_enum) in enum_pairs(&source.enums, &target.enums) {
|
||||||
|
match (s_enum, t_enum) {
|
||||||
|
(Some(s), None) => items.push(item(
|
||||||
|
"enum", &s.name, "added", vec![],
|
||||||
|
vec![format!(
|
||||||
|
"CREATE TYPE {}.{} AS ENUM ({});",
|
||||||
|
quote(family, &s.schema),
|
||||||
|
quote(family, &s.name),
|
||||||
|
s.labels.iter().map(|l| format!("'{}'", l.replace('\'', "''"))).collect::<Vec<_>>().join(", ")
|
||||||
|
)],
|
||||||
|
false,
|
||||||
|
)),
|
||||||
|
(None, Some(t)) => items.push(item(
|
||||||
|
"enum", &t.name, "removed", vec![],
|
||||||
|
vec![format!("DROP TYPE {}.{};", quote(family, &t.schema), quote(family, &t.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(s), Some(t)) => {
|
||||||
|
let mut added = Vec::new();
|
||||||
|
let mut removed_or_reordered = false;
|
||||||
|
for l in &s.labels {
|
||||||
|
if !t.labels.contains(l) { added.push(l.clone()); }
|
||||||
|
}
|
||||||
|
if t.labels.iter().any(|l| !s.labels.contains(l)) || !common_prefix(&s.labels, &t.labels) {
|
||||||
|
removed_or_reordered = true;
|
||||||
|
}
|
||||||
|
if !added.is_empty() {
|
||||||
|
if removed_or_reordered {
|
||||||
|
items.push(item("enum", &s.name, "changed", vec![detail_line("labels", &t.labels.join(", "), &s.labels.join(", "))], vec![], true));
|
||||||
|
} else {
|
||||||
|
items.push(item(
|
||||||
|
"enum", &s.name, "changed",
|
||||||
|
vec![detail_line("added labels", "", &added.join(", "))],
|
||||||
|
added.iter().map(|l| format!("ALTER TYPE {}.{} ADD VALUE '{}';", quote(family, &s.schema), quote(family, &s.name), l.replace('\'', "''"))).collect(),
|
||||||
|
false,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
(None, None) => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── Sequences (PG) ─────────────────────────────────────────────────────
|
||||||
|
for (s_seq, t_seq) in pair_by(&source.sequences, &target.sequences, |s| s.name.clone()) {
|
||||||
|
match (s_seq, t_seq) {
|
||||||
|
(Some(s), None) => items.push(item(
|
||||||
|
"sequence", &s.name, "added", vec![],
|
||||||
|
vec![format!(
|
||||||
|
"CREATE SEQUENCE {q} START WITH {start} INCREMENT BY {inc} MINVALUE {min} MAXVALUE {max} {cycle};",
|
||||||
|
q = qualified(family, schema, &s.name),
|
||||||
|
start = s.start, inc = s.increment, min = s.minimum, max = s.maximum,
|
||||||
|
cycle = if s.cycle { "CYCLE" } else { "NO CYCLE" }
|
||||||
|
)],
|
||||||
|
false,
|
||||||
|
)),
|
||||||
|
(None, Some(t)) => items.push(item(
|
||||||
|
"sequence", &t.name, "removed", vec![],
|
||||||
|
vec![format!("DROP SEQUENCE {};", qualified(family, schema, &t.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(s), Some(t)) => {
|
||||||
|
let mut sql = String::new();
|
||||||
|
let mut detail = Vec::new();
|
||||||
|
for (label, sv, tv) in [
|
||||||
|
("start", &s.start, &t.start),
|
||||||
|
("increment", &s.increment, &t.increment),
|
||||||
|
("minimum", &s.minimum, &t.minimum),
|
||||||
|
("maximum", &s.maximum, &t.maximum),
|
||||||
|
] {
|
||||||
|
if sv != tv {
|
||||||
|
detail.push(detail_line(label, tv, sv));
|
||||||
|
if !sql.is_empty() { sql.push(' '); }
|
||||||
|
match label {
|
||||||
|
"start" => sql.push_str(&format!("START WITH {sv}")),
|
||||||
|
"increment" => sql.push_str(&format!("INCREMENT BY {sv}")),
|
||||||
|
"minimum" => sql.push_str(&format!("MINVALUE {sv}")),
|
||||||
|
"maximum" => sql.push_str(&format!("MAXVALUE {sv}")),
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if s.cycle != t.cycle {
|
||||||
|
detail.push(detail_line("cycle", &t.cycle.to_string(), &s.cycle.to_string()));
|
||||||
|
if !sql.is_empty() { sql.push(' '); }
|
||||||
|
sql.push_str(if s.cycle { "CYCLE" } else { "NO CYCLE" });
|
||||||
|
}
|
||||||
|
if !sql.is_empty() {
|
||||||
|
items.push(item("sequence", &s.name, "changed", detail, vec![format!("ALTER SEQUENCE {} {};", qualified(family, schema, &s.name), sql)], false));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
(None, None) => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── Tables ─────────────────────────────────────────────────────────────
|
||||||
|
for (s_t, t_t) in pair_by(&source.tables, &target.tables, |t| t.name.clone()) {
|
||||||
|
match (s_t, t_t) {
|
||||||
|
(Some(s), None) => {
|
||||||
|
let mut defs: Vec<String> = s.columns.iter().map(|c| {
|
||||||
|
let mut d = format!("{} {}", quote(family, &c.name), c.data_type);
|
||||||
|
if !c.nullable { d.push_str(" NOT NULL"); }
|
||||||
|
if let Some(def) = &c.default_value { d.push_str(&format!(" DEFAULT {def}")); }
|
||||||
|
d
|
||||||
|
}).collect();
|
||||||
|
if !s.primary_key.is_empty() {
|
||||||
|
defs.push(format!("PRIMARY KEY ({})", s.primary_key.iter().map(|p| quote(family, p)).collect::<Vec<_>>().join(", ")));
|
||||||
|
}
|
||||||
|
items.push(item(
|
||||||
|
"table", &s.name, "added", vec![],
|
||||||
|
vec![format!("CREATE TABLE {} ({});", qualified(family, schema, &s.name), defs.join(", "))],
|
||||||
|
false,
|
||||||
|
));
|
||||||
|
push_constraint_items(&mut items, family, schema, s, "added");
|
||||||
|
}
|
||||||
|
(None, Some(t)) => items.push(item(
|
||||||
|
"table", &t.name, "removed", vec![],
|
||||||
|
vec![format!("DROP TABLE {};", qualified(family, schema, &t.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(s), Some(t)) => diff_table(&mut items, family, schema, s, t),
|
||||||
|
(None, None) => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── Views ──────────────────────────────────────────────────────────────
|
||||||
|
for (s_v, t_v) in pair_by(&source.views, &target.views, |v| v.name.clone()) {
|
||||||
|
match (s_v, t_v) {
|
||||||
|
(Some(s), None) => items.push(item(
|
||||||
|
"view", &s.name, "added", vec![],
|
||||||
|
vec![format!("CREATE {}VIEW {} AS {};", if s.materialized { "MATERIALIZED " } else { "" }, qualified(family, schema, &s.name), s.definition.trim_end_matches(';'))],
|
||||||
|
false,
|
||||||
|
)),
|
||||||
|
(None, Some(t)) => items.push(item(
|
||||||
|
"view", &t.name, "removed", vec![],
|
||||||
|
vec![format!("DROP {}VIEW {};", if t.materialized { "MATERIALIZED " } else { "" }, qualified(family, schema, &t.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(s), Some(t)) if s.definition.trim() != t.definition.trim() => {
|
||||||
|
if family == "postgresql" || family == "mysql" {
|
||||||
|
if s.materialized || t.materialized {
|
||||||
|
items.push(item("view", &s.name, "changed", vec![detail_line("definition", &t.definition, &s.definition)],
|
||||||
|
vec![format!("DROP MATERIALIZED VIEW {q}; CREATE MATERIALIZED VIEW {q} AS {d};", q = qualified(family, schema, &s.name), d = s.definition.trim_end_matches(';'))], true));
|
||||||
|
} else {
|
||||||
|
items.push(item("view", &s.name, "changed", vec![detail_line("definition", &t.definition, &s.definition)],
|
||||||
|
vec![format!("CREATE OR REPLACE VIEW {} AS {};", qualified(family, schema, &s.name), s.definition.trim_end_matches(';'))], false));
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
items.push(item("view", &s.name, "changed", vec![detail_line("definition", &t.definition, &s.definition)], vec![], true));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
DiffReport {
|
||||||
|
items,
|
||||||
|
source_label: format!("{} ({})", source.schema, source.engine),
|
||||||
|
target_label: format!("{} ({})", target.schema, target.engine),
|
||||||
|
truncated: source.truncated || target.truncated,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Items safe to stage: non-destructive AND carrying SQL.
|
||||||
|
pub fn stageable_items(report: &DiffReport) -> Vec<&DiffItem> {
|
||||||
|
report.items.iter().filter(|i| !i.destructive && !i.sync_sql.is_empty()).collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn common_prefix(source: &[String], target: &[String]) -> bool {
|
||||||
|
let mut ti = 0usize;
|
||||||
|
for l in target {
|
||||||
|
if source.contains(l) {
|
||||||
|
if ti >= source.len() || &source[ti] != l {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
ti += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
true
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Pair two keyed lists into (source, target) options by key.
|
||||||
|
fn pair_by<'a, T, K: Ord + Clone + Eq>(src: &'a [T], tgt: &'a [T], key: impl Fn(&T) -> K) -> Vec<(Option<&'a T>, Option<&'a T>)> {
|
||||||
|
use std::collections::BTreeMap;
|
||||||
|
let mut map: BTreeMap<K, (Option<&T>, Option<&T>)> = BTreeMap::new();
|
||||||
|
for s in src { map.entry(key(s)).or_insert((None, None)).0 = Some(s); }
|
||||||
|
for t in tgt { map.entry(key(t)).or_insert((None, None)).1 = Some(t); }
|
||||||
|
map.into_values().collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn enum_pairs<'a>(src: &'a [SnapshotEnum], tgt: &'a [SnapshotEnum]) -> Vec<(Option<&'a SnapshotEnum>, Option<&'a SnapshotEnum>)> {
|
||||||
|
pair_by(src, tgt, |e| e.name.clone())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn push_constraint_items(items: &mut Vec<DiffItem>, family: &str, schema: &str, table: &SnapshotTable, kind: &str) {
|
||||||
|
let qtn = qualified(family, schema, &table.name);
|
||||||
|
for fk in &table.foreign_keys {
|
||||||
|
items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", table.name, fk.name), kind, vec![],
|
||||||
|
vec![constraint_fk_sql(family, schema, &qtn, fk)],
|
||||||
|
false,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
for u in &table.uniques {
|
||||||
|
let cols = u.definition.split(", ").map(|c| quote(family, c.trim())).collect::<Vec<_>>().join(", ");
|
||||||
|
items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", table.name, u.name), kind, vec![],
|
||||||
|
vec![format!("ALTER TABLE {qtn} ADD CONSTRAINT {n} UNIQUE ({cols});", n = quote(family, &u.name), cols = cols)],
|
||||||
|
false,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
for c in &table.checks {
|
||||||
|
items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", table.name, c.name), kind, vec![],
|
||||||
|
vec![format!("ALTER TABLE {qtn} ADD CONSTRAINT {n} CHECK ({d});", n = quote(family, &c.name), d = c.definition.trim())],
|
||||||
|
false,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
for ix in &table.indexes {
|
||||||
|
let cols = ix.columns.iter().map(|c| quote(family, c)).collect::<Vec<_>>().join(", ");
|
||||||
|
items.push(item(
|
||||||
|
"index", &format!("{}.{}", table.name, ix.name), kind, vec![],
|
||||||
|
vec![format!("CREATE {u}INDEX {n} ON {qtn} ({cols});", u = if ix.unique { "UNIQUE " } else { "" }, n = quote(family, &ix.name), cols = cols)],
|
||||||
|
false,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn constraint_fk_sql(family: &str, _schema: &str, qtn: &str, fk: &SnapshotFk) -> String {
|
||||||
|
let cols = fk.columns.iter().map(|c| quote(family, c)).collect::<Vec<_>>().join(", ");
|
||||||
|
let ref_cols = fk.ref_columns.iter().map(|c| quote(family, c)).collect::<Vec<_>>().join(", ");
|
||||||
|
let ref_qualified = match family {
|
||||||
|
"mysql" => format!("`{}`", fk.ref_table.replace('`', "``")),
|
||||||
|
"sqlite" => quote(family, &fk.ref_table),
|
||||||
|
_ => qualified(family, &fk.ref_schema, &fk.ref_table),
|
||||||
|
};
|
||||||
|
format!(
|
||||||
|
"ALTER TABLE {qtn} ADD CONSTRAINT {n} FOREIGN KEY ({cols}) REFERENCES {r} ({rc}) ON DELETE {od} ON UPDATE {ou};",
|
||||||
|
n = quote(family, &fk.name), cols = cols, r = ref_qualified, rc = ref_cols,
|
||||||
|
od = fk.on_delete, ou = fk.on_update
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn diff_table(items: &mut Vec<DiffItem>, family: &str, schema: &str, s: &SnapshotTable, t: &SnapshotTable) {
|
||||||
|
let qtn = qualified(family, schema, &s.name);
|
||||||
|
|
||||||
|
for (s_c, t_c) in pair_by(&s.columns, &t.columns, |c| c.name.clone()) {
|
||||||
|
match (s_c, t_c) {
|
||||||
|
(Some(c), None) => {
|
||||||
|
let mut sql = format!("ALTER TABLE {qtn} ADD COLUMN {n} {ty}", n = quote(family, &c.name), ty = c.data_type);
|
||||||
|
if !c.nullable {
|
||||||
|
if family == "sqlite" && c.default_value.is_none() {
|
||||||
|
items.push(item("column", &format!("{}.{}", s.name, c.name), "added",
|
||||||
|
vec![detail_line("nullable", "false", "true (SQLite cannot ADD COLUMN NOT NULL without DEFAULT)")],
|
||||||
|
vec![format!("{sql};")], false));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
sql.push_str(" NOT NULL");
|
||||||
|
}
|
||||||
|
if let Some(d) = &c.default_value { sql.push_str(&format!(" DEFAULT {d}")); }
|
||||||
|
items.push(item("column", &format!("{}.{}", s.name, c.name), "added", vec![], vec![format!("{sql};")], false));
|
||||||
|
}
|
||||||
|
(None, Some(c)) => items.push(item(
|
||||||
|
"column", &format!("{}.{}", s.name, c.name), "removed", vec![],
|
||||||
|
vec![format!("ALTER TABLE {qtn} DROP COLUMN {n};", n = quote(family, &c.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(sc), Some(tc)) => {
|
||||||
|
if sc.data_type != tc.data_type {
|
||||||
|
let sql = match family {
|
||||||
|
"postgresql" => format!("ALTER TABLE {qtn} ALTER COLUMN {n} TYPE {ty} USING {n}::{ty};", n = quote(family, &sc.name), ty = sc.data_type),
|
||||||
|
"mysql" => format!("ALTER TABLE {qtn} MODIFY COLUMN {n} {ty}{nn};", n = quote(family, &sc.name), ty = sc.data_type, nn = if sc.nullable { String::new() } else { " NOT NULL".into() }),
|
||||||
|
_ => String::new(),
|
||||||
|
};
|
||||||
|
items.push(item("column", &format!("{}.{}", s.name, sc.name), "changed",
|
||||||
|
vec![detail_line("type", &tc.data_type, &sc.data_type)],
|
||||||
|
if sql.is_empty() { vec![] } else { vec![sql] }, true));
|
||||||
|
}
|
||||||
|
if sc.nullable != tc.nullable {
|
||||||
|
let sql = match family {
|
||||||
|
"postgresql" => format!("ALTER TABLE {qtn} ALTER COLUMN {n} {op};", n = quote(family, &sc.name), op = if sc.nullable { "DROP NOT NULL" } else { "SET NOT NULL" }),
|
||||||
|
"mysql" => format!("ALTER TABLE {qtn} MODIFY COLUMN {n} {ty}{nn};", n = quote(family, &sc.name), ty = sc.data_type, nn = if sc.nullable { String::new() } else { " NOT NULL".into() }),
|
||||||
|
_ => String::new(),
|
||||||
|
};
|
||||||
|
if !sql.is_empty() {
|
||||||
|
items.push(item("column", &format!("{}.{}", s.name, sc.name), "changed",
|
||||||
|
vec![detail_line("nullable", &tc.nullable.to_string(), &sc.nullable.to_string())],
|
||||||
|
vec![sql], false));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if sc.default_value != tc.default_value {
|
||||||
|
let sql = match family {
|
||||||
|
"postgresql" => match &sc.default_value {
|
||||||
|
Some(d) => format!("ALTER TABLE {qtn} ALTER COLUMN {n} SET DEFAULT {d};", n = quote(family, &sc.name)),
|
||||||
|
None => format!("ALTER TABLE {qtn} ALTER COLUMN {n} DROP DEFAULT;", n = quote(family, &sc.name)),
|
||||||
|
},
|
||||||
|
"mysql" => match &sc.default_value {
|
||||||
|
Some(d) => format!("ALTER TABLE {qtn} ALTER COLUMN {n} SET DEFAULT {d};", n = quote(family, &sc.name)),
|
||||||
|
None => format!("ALTER TABLE {qtn} ALTER COLUMN {n} DROP DEFAULT;", n = quote(family, &sc.name)),
|
||||||
|
},
|
||||||
|
_ => String::new(),
|
||||||
|
};
|
||||||
|
if !sql.is_empty() {
|
||||||
|
items.push(item("column", &format!("{}.{}", s.name, sc.name), "changed",
|
||||||
|
vec![detail_line("default", &tc.default_value.as_deref().unwrap_or_default(), &sc.default_value.clone().unwrap_or_default())],
|
||||||
|
vec![sql], false));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
(None, None) => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if s.primary_key != t.primary_key {
|
||||||
|
items.push(item("table", &s.name, "changed",
|
||||||
|
vec![detail_line("primary key", &t.primary_key.join(", "), &s.primary_key.join(", "))],
|
||||||
|
vec![], true));
|
||||||
|
}
|
||||||
|
|
||||||
|
for (s_c, t_c) in pair_by(&s.uniques, &t.uniques, |c| c.name.clone()) {
|
||||||
|
push_constraint_delta(items, family, s, "UNIQUE", s_c, t_c);
|
||||||
|
}
|
||||||
|
for (s_c, t_c) in pair_by(&s.checks, &t.checks, |c| c.name.clone()) {
|
||||||
|
push_constraint_delta(items, family, s, "CHECK", s_c, t_c);
|
||||||
|
}
|
||||||
|
|
||||||
|
for (s_f, t_f) in pair_by(&s.foreign_keys, &t.foreign_keys, |f| f.name.clone()) {
|
||||||
|
let qtn2 = qualified(family, schema, &s.name);
|
||||||
|
match (s_f, t_f) {
|
||||||
|
(Some(f), None) => {
|
||||||
|
if family == "sqlite" {
|
||||||
|
items.push(item("constraint", &format!("{}.{}", s.name, f.name), "added",
|
||||||
|
vec![detail_line("note", "", "SQLite cannot add foreign keys to existing tables — table rebuild required")],
|
||||||
|
vec![], false));
|
||||||
|
} else {
|
||||||
|
items.push(item("constraint", &format!("{}.{}", s.name, f.name), "added", vec![],
|
||||||
|
vec![constraint_fk_sql(family, schema, &qtn2, f)], false));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
(None, Some(f)) => items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", s.name, f.name), "removed", vec![],
|
||||||
|
vec![format!("ALTER TABLE {qtn2} DROP CONSTRAINT {n};", n = quote(family, &f.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(sf), Some(tf)) if sf.definition_differs(tf) => {
|
||||||
|
items.push(item("constraint", &format!("{}.{}", s.name, sf.name), "changed",
|
||||||
|
vec![detail_line("definition", &tf.on_delete, &sf.on_delete)],
|
||||||
|
vec![format!("ALTER TABLE {qtn2} DROP CONSTRAINT {n}; ALTER TABLE {qtn2} ADD CONSTRAINT {n2} FOREIGN KEY ({cols}) REFERENCES {r} ({rc}) ON DELETE {od} ON UPDATE {ou};",
|
||||||
|
n = quote(family, &sf.name), n2 = quote(family, &sf.name),
|
||||||
|
cols = sf.columns.iter().map(|c| quote(family, c)).collect::<Vec<_>>().join(", "),
|
||||||
|
r = qualified(family, &sf.ref_schema, &sf.ref_table),
|
||||||
|
rc = sf.ref_columns.iter().map(|c| quote(family, c)).collect::<Vec<_>>().join(", "),
|
||||||
|
od = sf.on_delete, ou = sf.on_update)],
|
||||||
|
true));
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for (s_i, t_i) in pair_by(&s.indexes, &t.indexes, |i| i.name.clone()) {
|
||||||
|
match (s_i, t_i) {
|
||||||
|
(Some(ix), None) => items.push(item(
|
||||||
|
"index", &format!("{}.{}", s.name, ix.name), "added", vec![],
|
||||||
|
vec![format!("CREATE {u}INDEX {n} ON {qtn} ({cols});", u = if ix.unique { "UNIQUE " } else { "" }, n = qualified(family, schema, &ix.name), cols = ix.columns.iter().map(|c| quote(family, c)).collect::<Vec<_>>().join(", "))],
|
||||||
|
false,
|
||||||
|
)),
|
||||||
|
(None, Some(ix)) => items.push(item(
|
||||||
|
"index", &format!("{}.{}", s.name, ix.name), "removed", vec![],
|
||||||
|
vec![format!("DROP INDEX {n};", n = quote(family, &ix.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(si), Some(ti)) if si.columns != ti.columns || si.unique != ti.unique => {
|
||||||
|
items.push(item("index", &format!("{}.{}", s.name, si.name), "changed",
|
||||||
|
vec![detail_line("columns", &ti.columns.join(", "), &si.columns.join(", "))],
|
||||||
|
vec![format!("DROP INDEX {n}; CREATE {u}INDEX {n} ON {qtn} ({cols});", n = qualified(family, schema, &si.name), u = if si.unique { "UNIQUE " } else { "" }, cols = si.columns.iter().map(|c| quote(family, c)).collect::<Vec<_>>().join(", "))],
|
||||||
|
true));
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn push_constraint_delta(items: &mut Vec<DiffItem>, family: &str, table: &SnapshotTable, kind: &str,
|
||||||
|
s_c: Option<&SnapshotConstraint>, t_c: Option<&SnapshotConstraint>) {
|
||||||
|
let qtn = qualified(family, &table.schema, &table.name);
|
||||||
|
match (s_c, t_c) {
|
||||||
|
(Some(c), None) if kind == "UNIQUE" => items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", table.name, c.name), "added", vec![],
|
||||||
|
vec![format!("ALTER TABLE {qtn} ADD CONSTRAINT {n} UNIQUE ({cols});", n = quote(family, &c.name), cols = c.definition.split(", ").map(|x| quote(family, x.trim())).collect::<Vec<_>>().join(", "))],
|
||||||
|
false,
|
||||||
|
)),
|
||||||
|
(Some(c), None) if kind == "CHECK" => items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", table.name, c.name), "added", vec![],
|
||||||
|
vec![format!("ALTER TABLE {qtn} ADD CONSTRAINT {n} CHECK ({d});", n = quote(family, &c.name), d = c.definition.trim())],
|
||||||
|
false,
|
||||||
|
)),
|
||||||
|
(Some(c), None) if family == "sqlite" => items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", table.name, c.name), "added",
|
||||||
|
vec![detail_line("note", "", "SQLite cannot add constraints to existing tables — table rebuild required")],
|
||||||
|
vec![], false)),
|
||||||
|
(None, Some(c)) => items.push(item(
|
||||||
|
"constraint", &format!("{}.{}", table.name, c.name), "removed", vec![],
|
||||||
|
vec![format!("ALTER TABLE {qtn} DROP CONSTRAINT {n};", n = quote(family, &c.name))],
|
||||||
|
true,
|
||||||
|
)),
|
||||||
|
(Some(sc), Some(tc)) if sc.definition.trim() != tc.definition.trim() => {
|
||||||
|
let sql = if kind == "UNIQUE" {
|
||||||
|
format!("ALTER TABLE {qtn} DROP CONSTRAINT {n}; ALTER TABLE {qtn} ADD CONSTRAINT {n} UNIQUE ({cols});",
|
||||||
|
n = quote(family, &sc.name), cols = sc.definition.split(", ").map(|x| quote(family, x.trim())).collect::<Vec<_>>().join(", "))
|
||||||
|
} else {
|
||||||
|
format!("ALTER TABLE {qtn} DROP CONSTRAINT {n}; ALTER TABLE {qtn} ADD CONSTRAINT {n} CHECK ({d});",
|
||||||
|
n = quote(family, &sc.name), d = sc.definition.trim())
|
||||||
|
};
|
||||||
|
items.push(item("constraint", &format!("{}.{}", table.name, sc.name), "changed",
|
||||||
|
vec![detail_line("definition", &tc.definition, &sc.definition)],
|
||||||
|
vec![sql], true));
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SnapshotFk {
|
||||||
|
fn definition_differs(&self, other: &SnapshotFk) -> bool {
|
||||||
|
self.columns != other.columns
|
||||||
|
|| self.ref_table != other.ref_table
|
||||||
|
|| self.ref_columns != other.ref_columns
|
||||||
|
|| self.on_delete != other.on_delete
|
||||||
|
|| self.on_update != other.on_update
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
#[path = "schema_diff.test.rs"]
|
#[path = "schema_diff.test.rs"]
|
||||||
mod schema_diff_tests;
|
mod schema_diff_tests;
|
||||||
|
|||||||
@@ -157,3 +157,206 @@ async fn pg_capture_live() {
|
|||||||
.expect("drop temp schema");
|
.expect("drop temp schema");
|
||||||
let _ = handle;
|
let _ = handle;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── Task 14: pure diff engine + sync-SQL generation ─────────────────
|
||||||
|
|
||||||
|
fn col(name: &str, ty: &str, nullable: bool, default: Option<&str>) -> SnapshotColumn {
|
||||||
|
SnapshotColumn {
|
||||||
|
name: name.into(),
|
||||||
|
data_type: ty.into(),
|
||||||
|
nullable,
|
||||||
|
default_value: default.map(|d| d.into()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn table(name: &str, columns: Vec<SnapshotColumn>) -> SnapshotTable {
|
||||||
|
SnapshotTable {
|
||||||
|
schema: "s".into(),
|
||||||
|
name: name.into(),
|
||||||
|
columns,
|
||||||
|
primary_key: vec![],
|
||||||
|
uniques: vec![],
|
||||||
|
checks: vec![],
|
||||||
|
indexes: vec![],
|
||||||
|
foreign_keys: vec![],
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn pg_snap(tables: Vec<SnapshotTable>) -> SchemaSnapshot {
|
||||||
|
let object_count = tables.len();
|
||||||
|
SchemaSnapshot {
|
||||||
|
engine: "postgresql".into(),
|
||||||
|
schema: "s".into(),
|
||||||
|
tables,
|
||||||
|
views: vec![],
|
||||||
|
sequences: vec![],
|
||||||
|
enums: vec![],
|
||||||
|
object_count,
|
||||||
|
truncated: false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn added_table_generates_create_with_pk_and_is_stageable() {
|
||||||
|
let mut src = pg_snap(vec![]);
|
||||||
|
let mut t = table("users", vec![col("id", "integer", false, None), col("email", "text", true, Some("''"))]);
|
||||||
|
t.primary_key = vec!["id".into()];
|
||||||
|
src.tables.push(t);
|
||||||
|
let tgt = pg_snap(vec![]);
|
||||||
|
let r = diff_snapshots(&src, &tgt);
|
||||||
|
let item = r.items.iter().find(|i| i.object_type == "table" && i.name == "users").unwrap();
|
||||||
|
assert_eq!(item.kind, "added");
|
||||||
|
assert!(!item.destructive);
|
||||||
|
assert_eq!(item.sync_sql.len(), 1);
|
||||||
|
let sql = &item.sync_sql[0];
|
||||||
|
assert!(sql.starts_with("CREATE TABLE \"s\".\"users\""), "got: {sql}");
|
||||||
|
assert!(sql.contains("\"id\" integer NOT NULL"));
|
||||||
|
assert!(sql.contains("\"email\" text DEFAULT ''"));
|
||||||
|
assert!(sql.contains("PRIMARY KEY (\"id\")"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn removed_table_is_destructive_copy_only() {
|
||||||
|
let tgt = pg_snap(vec![table("old", vec![col("id", "integer", false, None)])]);
|
||||||
|
let src = pg_snap(vec![]);
|
||||||
|
let r = diff_snapshots(&src, &tgt);
|
||||||
|
let item = r.items.iter().find(|i| i.name == "old").unwrap();
|
||||||
|
assert_eq!(item.kind, "removed");
|
||||||
|
assert!(item.destructive);
|
||||||
|
assert_eq!(item.sync_sql, vec!["DROP TABLE \"s\".\"old\";"]);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn column_diffs_classify_exactly_per_spec() {
|
||||||
|
let mut src = pg_snap(vec![table(
|
||||||
|
"t",
|
||||||
|
vec![
|
||||||
|
col("kept", "int", true, None),
|
||||||
|
col("added", "int", true, None),
|
||||||
|
col("retyped", "bigint", true, None),
|
||||||
|
col("tightened", "int", false, None),
|
||||||
|
col("defchanged", "int", true, Some("2")),
|
||||||
|
],
|
||||||
|
)]);
|
||||||
|
src.tables[0].name = "t".into();
|
||||||
|
let tgt = pg_snap(vec![table(
|
||||||
|
"t",
|
||||||
|
vec![
|
||||||
|
col("kept", "int", true, None),
|
||||||
|
col("gone", "int", true, None),
|
||||||
|
col("retyped", "int", true, None),
|
||||||
|
col("tightened", "int", true, None),
|
||||||
|
col("defchanged", "int", true, Some("1")),
|
||||||
|
],
|
||||||
|
)]);
|
||||||
|
|
||||||
|
let r = diff_snapshots(&src, &tgt);
|
||||||
|
|
||||||
|
let add = r.items.iter().find(|i| i.name == "t.added").unwrap();
|
||||||
|
assert_eq!((add.kind.as_str(), add.destructive), ("added", false));
|
||||||
|
assert_eq!(add.sync_sql[0], "ALTER TABLE \"s\".\"t\" ADD COLUMN \"added\" int;");
|
||||||
|
|
||||||
|
let gone = r.items.iter().find(|i| i.name == "t.gone").unwrap();
|
||||||
|
assert_eq!((gone.kind.as_str(), gone.destructive), ("removed", true));
|
||||||
|
assert_eq!(gone.sync_sql[0], "ALTER TABLE \"s\".\"t\" DROP COLUMN \"gone\";");
|
||||||
|
|
||||||
|
let retyped = r.items.iter().find(|i| i.name == "t.retyped").unwrap();
|
||||||
|
assert!(retyped.destructive, "ALTER COLUMN TYPE is destructive per spec");
|
||||||
|
assert!(retyped.sync_sql[0].contains("ALTER COLUMN \"retyped\" TYPE bigint"));
|
||||||
|
|
||||||
|
let tight = r.items.iter().find(|i| i.name == "t.tightened").unwrap();
|
||||||
|
assert!(!tight.destructive);
|
||||||
|
assert!(tight.sync_sql[0].contains("SET NOT NULL"));
|
||||||
|
|
||||||
|
let def = r.items.iter().find(|i| i.name == "t.defchanged").unwrap();
|
||||||
|
assert!(!def.destructive);
|
||||||
|
assert!(def.sync_sql[0].contains("SET DEFAULT 2"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn constraint_index_fk_view_sequence_enum_diffs() {
|
||||||
|
let mut src = pg_snap(vec![table("t", vec![col("id", "int", false, None)])]);
|
||||||
|
let mut tgt = pg_snap(vec![table("t", vec![col("id", "int", false, None)])]);
|
||||||
|
|
||||||
|
src.tables[0].foreign_keys.push(SnapshotFk {
|
||||||
|
name: "fk_t_ref".into(),
|
||||||
|
columns: vec!["id".into()],
|
||||||
|
ref_schema: "s".into(),
|
||||||
|
ref_table: "other".into(),
|
||||||
|
ref_columns: vec!["id".into()],
|
||||||
|
on_delete: "CASCADE".into(),
|
||||||
|
on_update: "NO ACTION".into(),
|
||||||
|
});
|
||||||
|
src.tables[0].indexes.push(SnapshotIndex { name: "ix_t".into(), columns: vec!["id".into()], unique: false });
|
||||||
|
src.views.push(SnapshotView { schema: "s".into(), name: "v".into(), definition: "SELECT 1".into(), materialized: false });
|
||||||
|
src.sequences.push(SnapshotSequence { schema: "s".into(), name: "seq".into(), start: "1".into(), increment: "1".into(), minimum: "1".into(), maximum: "100".into(), cycle: false });
|
||||||
|
src.enums.push(SnapshotEnum { schema: "s".into(), name: "color".into(), labels: vec!["red".into(), "green".into()] });
|
||||||
|
tgt.enums.push(SnapshotEnum { schema: "s".into(), name: "color".into(), labels: vec!["red".into()] });
|
||||||
|
|
||||||
|
let r = diff_snapshots(&src, &tgt);
|
||||||
|
|
||||||
|
let fk = r.items.iter().find(|i| i.object_type == "constraint" && i.name == "t.fk_t_ref").unwrap();
|
||||||
|
assert!(!fk.destructive);
|
||||||
|
assert!(fk.sync_sql[0].contains("ADD CONSTRAINT \"fk_t_ref\" FOREIGN KEY (\"id\")"));
|
||||||
|
assert!(fk.sync_sql[0].contains("REFERENCES \"s\".\"other\" (\"id\")"));
|
||||||
|
assert!(fk.sync_sql[0].contains("ON DELETE CASCADE"));
|
||||||
|
|
||||||
|
let ix = r.items.iter().find(|i| i.object_type == "index" && i.name == "t.ix_t").unwrap();
|
||||||
|
assert_eq!(ix.sync_sql[0], "CREATE INDEX \"s\".\"ix_t\" ON \"s\".\"t\" (\"id\");");
|
||||||
|
|
||||||
|
let view = r.items.iter().find(|i| i.object_type == "view" && i.name == "v").unwrap();
|
||||||
|
assert!(view.sync_sql[0].starts_with("CREATE VIEW \"s\".\"v\" AS"));
|
||||||
|
|
||||||
|
let seq = r.items.iter().find(|i| i.object_type == "sequence" && i.name == "seq").unwrap();
|
||||||
|
assert!(seq.sync_sql[0].starts_with("CREATE SEQUENCE \"s\".\"seq\""));
|
||||||
|
|
||||||
|
let en = r.items.iter().find(|i| i.object_type == "enum" && i.name == "color").unwrap();
|
||||||
|
assert!(!en.destructive);
|
||||||
|
assert!(en.sync_sql[0].contains("ALTER TYPE \"s\".\"color\" ADD VALUE 'green'"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn mysql_family_uses_backticks() {
|
||||||
|
let mut src = SchemaSnapshot { engine: "mariadb".into(), schema: "app".into(), ..Default::default() };
|
||||||
|
src.tables.push(SnapshotTable {
|
||||||
|
schema: "app".into(),
|
||||||
|
name: "users".into(),
|
||||||
|
columns: vec![SnapshotColumn { name: "id".into(), data_type: "int".into(), nullable: false, default_value: None }],
|
||||||
|
primary_key: vec!["id".into()],
|
||||||
|
uniques: vec![],
|
||||||
|
checks: vec![],
|
||||||
|
indexes: vec![],
|
||||||
|
foreign_keys: vec![],
|
||||||
|
});
|
||||||
|
let tgt = SchemaSnapshot { engine: "mysql".into(), schema: "app".into(), ..Default::default() };
|
||||||
|
let r = diff_snapshots(&src, &tgt);
|
||||||
|
let item = r.items.iter().find(|i| i.name == "users").unwrap();
|
||||||
|
assert!(item.sync_sql[0].starts_with("CREATE TABLE `app`.`users`"), "got: {}", item.sync_sql[0]);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn sqlite_changed_view_is_destructive_note_only() {
|
||||||
|
let mut src = SchemaSnapshot { engine: "sqlite".into(), schema: "main".into(), ..Default::default() };
|
||||||
|
src.views.push(SnapshotView { schema: "main".into(), name: "v".into(), definition: "SELECT 2".into(), materialized: false });
|
||||||
|
let mut tgt = SchemaSnapshot { engine: "sqlite".into(), schema: "main".into(), ..Default::default() };
|
||||||
|
tgt.views.push(SnapshotView { schema: "main".into(), name: "v".into(), definition: "SELECT 1".into(), materialized: false });
|
||||||
|
let r = diff_snapshots(&src, &tgt);
|
||||||
|
let item = r.items.iter().find(|i| i.object_type == "view").unwrap();
|
||||||
|
assert_eq!(item.kind, "changed");
|
||||||
|
assert!(item.destructive);
|
||||||
|
assert!(item.sync_sql.is_empty(), "SQLite cannot CREATE OR REPLACE a view");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn identical_snapshots_diff_empty_and_stageable_filter() {
|
||||||
|
let snap = pg_snap(vec![table("t", vec![col("id", "int", false, None)])]);
|
||||||
|
let r = diff_snapshots(&snap, &snap);
|
||||||
|
assert!(r.items.is_empty());
|
||||||
|
assert!(stageable_items(&r).is_empty());
|
||||||
|
|
||||||
|
let mut src2 = pg_snap(vec![]);
|
||||||
|
src2.tables.push(table("new", vec![col("id", "int", false, None)]));
|
||||||
|
src2.tables.push(table("dropme", vec![col("id", "int", false, None)]));
|
||||||
|
let r2 = diff_snapshots(&src2, &snap);
|
||||||
|
assert!(stageable_items(&r2).iter().all(|i| !i.destructive && !i.sync_sql.is_empty()));
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user