0.8.0: MariaDB + PlanetScale, TimescaleDB, multi-result sets, streaming export, schema diff #27
@@ -912,3 +912,98 @@ pub fn pg_hypertables_query(_schema: &str) -> String {
|
||||
ORDER BY h.hypertable_name"
|
||||
.to_string()
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Schema-diff snapshot queries (0.8.0) — one set-based query per object class
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Table columns with resolved types, nullability, and defaults.
|
||||
/// Columns: table_name, column_name, data_type, is_nullable, default_expr
|
||||
pub fn pg_diff_columns_query() -> String {
|
||||
"SELECT c.relname, a.attname, \
|
||||
pg_catalog.format_type(a.atttypid, a.atttypmod), \
|
||||
NOT a.attnotnull, \
|
||||
pg_catalog.pg_get_expr(d.adbin, d.adrelid) \
|
||||
FROM pg_catalog.pg_class c \
|
||||
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
|
||||
JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid AND a.attnum > 0 AND NOT a.attisdropped \
|
||||
LEFT JOIN pg_catalog.pg_attrdef d ON d.adrelid = c.oid AND d.adnum = a.attnum \
|
||||
WHERE n.nspname = $1 AND c.relkind IN ('r', 'p') \
|
||||
ORDER BY c.relname, a.attnum"
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// Constraints (PK/UNIQUE/CHECK/FK) with exact column lists and FK metadata.
|
||||
/// Columns: table_name, constraint_name, contype, cols, ref_cols, ref_table, confdeltype, confupdtype, definition
|
||||
pub fn pg_diff_constraints_query() -> String {
|
||||
"SELECT cl.relname, con.conname, con.contype::text, \
|
||||
(SELECT string_agg(a.attname, ', ' ORDER BY k.ord) \
|
||||
FROM unnest(con.conkey) WITH ORDINALITY AS k(attnum, ord) \
|
||||
JOIN pg_attribute a ON a.attrelid = con.conrelid AND a.attnum = k.attnum), \
|
||||
(SELECT string_agg(a.attname, ', ' ORDER BY k.ord) \
|
||||
FROM unnest(con.confkey) WITH ORDINALITY AS k(attnum, ord) \
|
||||
JOIN pg_attribute a ON a.attrelid = con.confrelid AND a.attnum = k.attnum), \
|
||||
(SELECT n2.nspname || '.' || cl2.relname \
|
||||
FROM pg_class cl2 JOIN pg_namespace n2 ON n2.oid = cl2.relnamespace \
|
||||
WHERE cl2.oid = con.confrelid), \
|
||||
con.confdeltype::text, con.confupdtype::text, \
|
||||
pg_get_constraintdef(con.oid, true) \
|
||||
FROM pg_catalog.pg_constraint con \
|
||||
JOIN pg_catalog.pg_class cl ON cl.oid = con.conrelid \
|
||||
JOIN pg_catalog.pg_namespace n ON n.oid = cl.relnamespace \
|
||||
WHERE n.nspname = $1 AND con.contype IN ('p', 'u', 'c', 'f') \
|
||||
ORDER BY cl.relname, con.contype, con.conname"
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// Secondary indexes (constraint-backed indexes excluded), with column lists.
|
||||
/// Columns: table_name, index_name, is_unique, cols
|
||||
pub fn pg_diff_indexes_query() -> String {
|
||||
"SELECT t.relname, i.relname, ix.indisunique, \
|
||||
(SELECT string_agg(a.attname, ', ' ORDER BY k.ord) \
|
||||
FROM unnest(ix.indkey) WITH ORDINALITY AS k(attnum, ord) \
|
||||
JOIN pg_attribute a ON a.attrelid = ix.indrelid AND a.attnum = k.attnum \
|
||||
WHERE k.attnum > 0) \
|
||||
FROM pg_index ix \
|
||||
JOIN pg_class i ON i.oid = ix.indexrelid \
|
||||
JOIN pg_class t ON t.oid = ix.indrelid \
|
||||
JOIN pg_namespace n ON n.oid = t.relnamespace \
|
||||
WHERE n.nspname = $1 AND NOT ix.indisprimary \
|
||||
AND NOT EXISTS (SELECT 1 FROM pg_constraint con WHERE con.conindid = i.oid) \
|
||||
ORDER BY t.relname, i.relname"
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// Views + materialized views. Columns: schema, name, definition, materialized
|
||||
pub fn pg_diff_views_query() -> String {
|
||||
"SELECT v.schemaname, v.viewname, v.definition, false \
|
||||
FROM pg_views v WHERE v.schemaname = $1 \
|
||||
UNION ALL \
|
||||
SELECT v.schemaname, v.matviewname, v.definition, true \
|
||||
FROM pg_matviews v WHERE v.schemaname = $1 \
|
||||
ORDER BY 2"
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// Sequences. Columns: schema, name, start, increment, min, max, cycle
|
||||
pub fn pg_diff_sequences_query() -> String {
|
||||
"SELECT sequence_schema, sequence_name, \
|
||||
COALESCE(start_value::text, '1'), COALESCE(increment::text, '1'), \
|
||||
COALESCE(minimum_value::text, '1'), COALESCE(maximum_value::text, '9223372036854775807'), \
|
||||
cycle_option = 'YES' \
|
||||
FROM information_schema.sequences \
|
||||
WHERE sequence_schema = $1 \
|
||||
ORDER BY sequence_name"
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// Enums with ordered labels. Columns: schema, name, label (row per label)
|
||||
pub fn pg_diff_enums_query() -> String {
|
||||
"SELECT n.nspname, t.typname, e.enumlabel \
|
||||
FROM pg_type t \
|
||||
JOIN pg_namespace n ON n.oid = t.typnamespace \
|
||||
JOIN pg_enum e ON e.enumtypid = t.oid \
|
||||
WHERE t.typtype = 'e' AND n.nspname = $1 \
|
||||
ORDER BY t.typname, e.enumsortorder"
|
||||
.to_string()
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ pub mod object_crud;
|
||||
pub mod object_ddl;
|
||||
pub mod pool;
|
||||
pub mod tls;
|
||||
pub mod schema_diff;
|
||||
|
||||
#[allow(unused_imports)]
|
||||
pub use pool::{ConnectionPoolManager, DbConfig, DbHandle};
|
||||
|
||||
@@ -323,4 +323,64 @@ mod tests {
|
||||
assert_eq!(mysql_default_sort(&["name".into()]), "name");
|
||||
assert_eq!(mysql_default_sort(&[]), "");
|
||||
}
|
||||
}
|
||||
}
|
||||
// ── Schema-diff snapshot queries (0.8.0) ──────────────────────────────
|
||||
|
||||
/// Columns: table_name, column_name, column_type, is_nullable, column_default
|
||||
pub fn mysql_snapshot_columns_query(schema: &str) -> String {
|
||||
format!(
|
||||
"SELECT table_name, column_name, column_type, is_nullable, \
|
||||
COALESCE(column_default, '') \
|
||||
FROM information_schema.columns \
|
||||
WHERE table_schema = '{}' \
|
||||
ORDER BY table_name, ordinal_position",
|
||||
schema.replace('\'', "''")
|
||||
)
|
||||
}
|
||||
|
||||
/// Constraints (one row per constraint-column). Columns: table_name,
|
||||
/// constraint_name, constraint_type, column_name, referenced_table_name,
|
||||
/// delete_rule, update_rule, check_clause
|
||||
pub fn mysql_snapshot_constraints_query(schema: &str) -> String {
|
||||
format!(
|
||||
"SELECT kcu.table_name, kcu.constraint_name, tc.constraint_type, kcu.column_name, \
|
||||
COALESCE(kcu.referenced_table_name, ''), \
|
||||
COALESCE(rc.delete_rule, ''), COALESCE(rc.update_rule, ''), \
|
||||
COALESCE((SELECT cc.check_clause FROM information_schema.check_constraints cc \
|
||||
WHERE cc.constraint_schema = kcu.constraint_schema \
|
||||
AND cc.constraint_name = kcu.constraint_name), '') \
|
||||
FROM information_schema.key_column_usage kcu \
|
||||
LEFT JOIN information_schema.table_constraints tc \
|
||||
ON tc.constraint_schema = kcu.constraint_schema \
|
||||
AND tc.constraint_name = kcu.constraint_name \
|
||||
AND tc.table_name = kcu.table_name \
|
||||
LEFT JOIN information_schema.referential_constraints rc \
|
||||
ON rc.constraint_schema = kcu.constraint_schema \
|
||||
AND rc.constraint_name = kcu.constraint_name \
|
||||
WHERE kcu.constraint_schema = '{}' AND tc.constraint_type IS NOT NULL \
|
||||
ORDER BY kcu.table_name, kcu.constraint_name, kcu.ordinal_position",
|
||||
schema.replace('\'', "''")
|
||||
)
|
||||
}
|
||||
|
||||
/// Secondary indexes (one row per index-column). Columns: table_name,
|
||||
/// index_name, non_unique, column_name
|
||||
pub fn mysql_snapshot_indexes_query(schema: &str) -> String {
|
||||
format!(
|
||||
"SELECT table_name, index_name, non_unique, column_name \
|
||||
FROM information_schema.statistics \
|
||||
WHERE table_schema = '{}' AND index_name != 'PRIMARY' \
|
||||
ORDER BY table_name, index_name, seq_in_index",
|
||||
schema.replace('\'', "''")
|
||||
)
|
||||
}
|
||||
|
||||
/// Views. Columns: table_schema, table_name, view_definition
|
||||
pub fn mysql_snapshot_views_query(schema: &str) -> String {
|
||||
format!(
|
||||
"SELECT table_schema, table_name, view_definition \
|
||||
FROM information_schema.views \
|
||||
WHERE table_schema = '{}'",
|
||||
schema.replace('\'', "''")
|
||||
)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,508 @@
|
||||
//! Schema snapshot capture + the pure diff engine (engine lands in Task 14).
|
||||
//!
|
||||
//! Snapshots are read-only, single-use, and capped (5,000 objects — spec §2).
|
||||
//! Sync SQL is generated in the TARGET's dialect with identifiers quoted
|
||||
//! per engine (spec §2: never raw interpolation).
|
||||
|
||||
use crate::db::pool::DbHandle;
|
||||
use crate::models::db_viewer::*;
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
pub(crate) const SNAPSHOT_CAP: usize = 5_000;
|
||||
|
||||
/// Map a PG `conf*type` action code to its SQL name.
|
||||
pub(crate) fn pg_fk_action(code: &str) -> &'static str {
|
||||
match code {
|
||||
"r" => "RESTRICT",
|
||||
"c" => "CASCADE",
|
||||
"n" => "SET NULL",
|
||||
"d" => "SET DEFAULT",
|
||||
_ => "NO ACTION",
|
||||
}
|
||||
}
|
||||
|
||||
/// Enforce the 5,000-object cap, setting `truncated` (never silent).
|
||||
pub(crate) fn apply_snapshot_cap(snap: &mut SchemaSnapshot) {
|
||||
let mut total = snap.tables.len() + snap.views.len() + snap.sequences.len() + snap.enums.len();
|
||||
let over = total > SNAPSHOT_CAP;
|
||||
if over {
|
||||
let mut remaining = SNAPSHOT_CAP;
|
||||
fn take<T>(v: &mut Vec<T>, budget: &mut usize) {
|
||||
if v.len() > *budget {
|
||||
v.truncate(*budget);
|
||||
}
|
||||
*budget -= v.len();
|
||||
}
|
||||
take(&mut snap.tables, &mut remaining);
|
||||
take(&mut snap.views, &mut remaining);
|
||||
take(&mut snap.sequences, &mut remaining);
|
||||
take(&mut snap.enums, &mut remaining);
|
||||
total = SNAPSHOT_CAP;
|
||||
}
|
||||
snap.object_count = total;
|
||||
snap.truncated = over;
|
||||
}
|
||||
|
||||
fn csv_names(s: &str) -> Vec<String> {
|
||||
s.split(", ").map(|p| p.trim().to_string()).filter(|p| !p.is_empty()).collect()
|
||||
}
|
||||
|
||||
/// Capture a PostgreSQL schema snapshot (six set-based queries, one per class).
|
||||
pub(crate) async fn capture_pg(
|
||||
client: &tokio_postgres::Client,
|
||||
engine: &str,
|
||||
schema: &str,
|
||||
) -> Result<SchemaSnapshot, String> {
|
||||
let mut tables: BTreeMap<String, SnapshotTable> = BTreeMap::new();
|
||||
let mut views: Vec<SnapshotView> = Vec::new();
|
||||
let mut sequences: Vec<SnapshotSequence> = Vec::new();
|
||||
let mut enums: BTreeMap<String, SnapshotEnum> = BTreeMap::new();
|
||||
|
||||
// Columns
|
||||
let rows = client
|
||||
.query(&crate::db::introspection::pg_diff_columns_query(), &[&schema])
|
||||
.await
|
||||
.map_err(|e| format!("columns query failed: {e}"))?;
|
||||
for r in rows {
|
||||
let tname: String = r.get(0);
|
||||
let entry = tables.entry(tname.clone()).or_insert_with(|| SnapshotTable {
|
||||
schema: schema.to_string(),
|
||||
name: tname,
|
||||
..Default::default()
|
||||
});
|
||||
entry.columns.push(SnapshotColumn {
|
||||
name: r.get(1),
|
||||
data_type: r.get(2),
|
||||
nullable: r.get(3),
|
||||
default_value: r.get::<_, Option<String>>(4),
|
||||
});
|
||||
}
|
||||
|
||||
// Constraints (PK / UNIQUE / CHECK / FK)
|
||||
// Column order: table_name(0), constraint_name(1), contype(2), cols(3),
|
||||
// ref_cols(4), ref_table(5), confdeltype(6), confupdtype(7), definition(8)
|
||||
let rows = client
|
||||
.query(&crate::db::introspection::pg_diff_constraints_query(), &[&schema])
|
||||
.await
|
||||
.map_err(|e| format!("constraints query failed: {e}"))?;
|
||||
for r in rows {
|
||||
let tname: String = r.get(0);
|
||||
let cname: String = r.get(1);
|
||||
let contype: String = r.get(2);
|
||||
let Some(entry) = tables.get_mut(&tname) else { continue };
|
||||
match contype.as_str() {
|
||||
"p" => entry.primary_key = csv_names(&r.get::<_, Option<String>>(3).unwrap_or_default()),
|
||||
"u" => entry.uniques.push(SnapshotConstraint {
|
||||
name: cname,
|
||||
kind: "unique".into(),
|
||||
definition: r.get::<_, Option<String>>(3).unwrap_or_default(),
|
||||
}),
|
||||
"c" => entry.checks.push(SnapshotConstraint {
|
||||
name: cname,
|
||||
kind: "check".into(),
|
||||
definition: r.get::<_, Option<String>>(8).unwrap_or_default(),
|
||||
}),
|
||||
"f" => {
|
||||
let ref_qualified: Option<String> = r.get(5);
|
||||
let (ref_schema, ref_table) = ref_qualified
|
||||
.map(|q| match q.split_once('.') {
|
||||
Some((s, t)) => (s.to_string(), t.to_string()),
|
||||
None => (String::new(), q.clone()),
|
||||
})
|
||||
.unwrap_or_default();
|
||||
entry.foreign_keys.push(SnapshotFk {
|
||||
name: cname,
|
||||
columns: csv_names(&r.get::<_, Option<String>>(3).unwrap_or_default()),
|
||||
ref_schema,
|
||||
ref_table,
|
||||
ref_columns: csv_names(&r.get::<_, Option<String>>(4).unwrap_or_default()),
|
||||
on_delete: pg_fk_action(&r.get::<_, Option<String>>(6).unwrap_or_default()).to_string(),
|
||||
on_update: pg_fk_action(&r.get::<_, Option<String>>(7).unwrap_or_default()).to_string(),
|
||||
});
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
// Indexes
|
||||
let rows = client
|
||||
.query(&crate::db::introspection::pg_diff_indexes_query(), &[&schema])
|
||||
.await
|
||||
.map_err(|e| format!("indexes query failed: {e}"))?;
|
||||
for r in rows {
|
||||
let tname: String = r.get(0);
|
||||
let Some(entry) = tables.get_mut(&tname) else { continue };
|
||||
entry.indexes.push(SnapshotIndex {
|
||||
name: r.get(1),
|
||||
columns: csv_names(&r.get::<_, Option<String>>(3).unwrap_or_default()),
|
||||
unique: r.get(2),
|
||||
});
|
||||
}
|
||||
|
||||
// Views + matviews
|
||||
let rows = client
|
||||
.query(&crate::db::introspection::pg_diff_views_query(), &[&schema])
|
||||
.await
|
||||
.map_err(|e| format!("views query failed: {e}"))?;
|
||||
for r in rows {
|
||||
views.push(SnapshotView {
|
||||
schema: r.get(0),
|
||||
name: r.get(1),
|
||||
definition: r.get(2),
|
||||
materialized: r.get(3),
|
||||
});
|
||||
}
|
||||
|
||||
// Sequences
|
||||
let rows = client
|
||||
.query(&crate::db::introspection::pg_diff_sequences_query(), &[&schema])
|
||||
.await
|
||||
.map_err(|e| format!("sequences query failed: {e}"))?;
|
||||
for r in rows {
|
||||
sequences.push(SnapshotSequence {
|
||||
schema: r.get(0),
|
||||
name: r.get(1),
|
||||
start: r.get(2),
|
||||
increment: r.get(3),
|
||||
minimum: r.get(4),
|
||||
maximum: r.get(5),
|
||||
cycle: r.get(6),
|
||||
});
|
||||
}
|
||||
|
||||
// Enums (one row per label — aggregate by name)
|
||||
let rows = client
|
||||
.query(&crate::db::introspection::pg_diff_enums_query(), &[&schema])
|
||||
.await
|
||||
.map_err(|e| format!("enums query failed: {e}"))?;
|
||||
for r in rows {
|
||||
let ename: String = r.get(1);
|
||||
let entry = enums.entry(ename.clone()).or_insert_with(|| SnapshotEnum {
|
||||
schema: r.get(0),
|
||||
name: ename,
|
||||
labels: Vec::new(),
|
||||
});
|
||||
entry.labels.push(r.get(2));
|
||||
}
|
||||
|
||||
let mut snap = SchemaSnapshot {
|
||||
engine: engine.to_string(),
|
||||
schema: schema.to_string(),
|
||||
tables: tables.into_values().collect(),
|
||||
views,
|
||||
sequences,
|
||||
enums: enums.into_values().collect(),
|
||||
object_count: 0,
|
||||
truncated: false,
|
||||
};
|
||||
apply_snapshot_cap(&mut snap);
|
||||
Ok(snap)
|
||||
}
|
||||
|
||||
/// Capture a SQLite (in-file/in-memory) snapshot.
|
||||
pub(crate) fn capture_sqlite(conn: &rusqlite::Connection) -> Result<SchemaSnapshot, String> {
|
||||
crate::db::schema_diff::capture::sqlite(conn)
|
||||
}
|
||||
|
||||
/// Capture a MySQL/MariaDB snapshot.
|
||||
pub(crate) async fn capture_mysql(
|
||||
conn: &mut sqlx::mysql::MySqlConnection,
|
||||
engine: &str,
|
||||
schema: &str,
|
||||
) -> Result<SchemaSnapshot, String> {
|
||||
crate::db::schema_diff::capture::mysql(conn, engine, schema).await
|
||||
}
|
||||
|
||||
/// Dispatch a snapshot capture across engines. `engine` is the CONNECTION's
|
||||
/// db_type (mariadb vs mysql matters for the diff dialect).
|
||||
pub(crate) async fn capture_snapshot(
|
||||
handle: &DbHandle,
|
||||
engine: &str,
|
||||
schema: &str,
|
||||
) -> Result<SchemaSnapshot, String> {
|
||||
match handle {
|
||||
DbHandle::Postgresql(client, _) => capture_pg(client, engine, schema).await,
|
||||
DbHandle::MySql(pool) => {
|
||||
let mut conn = pool
|
||||
.acquire()
|
||||
.await
|
||||
.map_err(|e| format!("source/target connection failed: {e}"))?;
|
||||
capture_mysql(&mut *conn, engine, schema).await
|
||||
}
|
||||
DbHandle::Sqlite(conn) => capture_sqlite(conn),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) mod capture {
|
||||
use super::*;
|
||||
|
||||
pub(crate) fn sqlite(conn: &rusqlite::Connection) -> Result<SchemaSnapshot, String> {
|
||||
let mut snap = SchemaSnapshot {
|
||||
engine: "sqlite".into(),
|
||||
schema: "main".into(),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut stmt = conn
|
||||
.prepare("SELECT type, name, sql FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' AND type IN ('table', 'view') ORDER BY name")
|
||||
.map_err(|e| e.to_string())?;
|
||||
let objects: Vec<(String, String, Option<String>)> = stmt
|
||||
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
|
||||
.map_err(|e| e.to_string())?
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
|
||||
for (otype, name, sql) in objects {
|
||||
if otype == "view" {
|
||||
snap.views.push(SnapshotView {
|
||||
schema: "main".into(),
|
||||
name,
|
||||
definition: sql.unwrap_or_default(),
|
||||
materialized: false,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
let mut table = SnapshotTable {
|
||||
schema: "main".into(),
|
||||
name: name.clone(),
|
||||
..Default::default()
|
||||
};
|
||||
// Columns + PK + defaults
|
||||
let mut cols = conn
|
||||
.prepare(&format!("PRAGMA table_info({})", quote_double(&name)))
|
||||
.map_err(|e| e.to_string())?;
|
||||
let col_rows: Vec<(String, String, i64, Option<String>, i64)> = cols
|
||||
.query_map([], |r| Ok((r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?)))
|
||||
.map_err(|e| e.to_string())?
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
drop(cols);
|
||||
for (cname, ctype, notnull, default, pk) in col_rows {
|
||||
if pk > 0 {
|
||||
table.primary_key.push(cname.clone());
|
||||
}
|
||||
table.columns.push(SnapshotColumn {
|
||||
name: cname,
|
||||
data_type: ctype,
|
||||
nullable: notnull == 0,
|
||||
default_value: default,
|
||||
});
|
||||
}
|
||||
// Indexes (skip the PK-mandated sqlite_autoindex_*)
|
||||
let mut idx = conn
|
||||
.prepare(&format!("PRAGMA index_list({})", quote_double(&name)))
|
||||
.map_err(|e| e.to_string())?;
|
||||
let idx_rows: Vec<(String, i64)> = idx
|
||||
.query_map([], |r| Ok((r.get(1)?, r.get(2)?)))
|
||||
.map_err(|e| e.to_string())?
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
drop(idx);
|
||||
for (iname, unique) in idx_rows {
|
||||
if iname.starts_with("sqlite_autoindex") {
|
||||
continue;
|
||||
}
|
||||
let mut info = conn
|
||||
.prepare(&format!("PRAGMA index_info({})", quote_double(&iname)))
|
||||
.map_err(|e| e.to_string())?;
|
||||
let icols: Vec<String> = info
|
||||
.query_map([], |r| r.get::<_, String>(2))
|
||||
.map_err(|e| e.to_string())?
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
drop(info);
|
||||
table.indexes.push(SnapshotIndex {
|
||||
name: iname,
|
||||
columns: icols,
|
||||
unique: unique != 0,
|
||||
});
|
||||
}
|
||||
// Foreign keys
|
||||
let mut fks = conn
|
||||
.prepare(&format!("PRAGMA foreign_key_list({})", quote_double(&name)))
|
||||
.map_err(|e| e.to_string())?;
|
||||
let fk_rows: Vec<(i64, i64, String, String, String, String, String)> = fks
|
||||
.query_map([], |r| {
|
||||
Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?))
|
||||
})
|
||||
.map_err(|e| e.to_string())?
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
drop(fks);
|
||||
// (id, seq, table, from, to, on_update, on_delete) — group by id
|
||||
let mut grouped: BTreeMap<i64, SnapshotFk> = BTreeMap::new();
|
||||
for (id, _seq, reftable, from, to, on_update, on_delete) in fk_rows {
|
||||
let fk = grouped.entry(id).or_insert_with(|| SnapshotFk {
|
||||
name: format!("fk_{name}_{id}"),
|
||||
columns: Vec::new(),
|
||||
ref_schema: "main".into(),
|
||||
ref_table: reftable,
|
||||
ref_columns: Vec::new(),
|
||||
on_delete: String::new(),
|
||||
on_update: String::new(),
|
||||
});
|
||||
fk.columns.push(from);
|
||||
if !to.is_empty() {
|
||||
fk.ref_columns.push(to);
|
||||
}
|
||||
if !on_delete.is_empty() {
|
||||
fk.on_delete = on_delete.to_uppercase();
|
||||
}
|
||||
if !on_update.is_empty() {
|
||||
fk.on_update = on_update.to_uppercase();
|
||||
}
|
||||
}
|
||||
table.foreign_keys = grouped.into_values().collect();
|
||||
snap.tables.push(table);
|
||||
}
|
||||
|
||||
apply_snapshot_cap(&mut snap);
|
||||
Ok(snap)
|
||||
}
|
||||
|
||||
pub(crate) async fn mysql(
|
||||
conn: &mut sqlx::mysql::MySqlConnection,
|
||||
engine: &str,
|
||||
schema: &str,
|
||||
) -> Result<SchemaSnapshot, String> {
|
||||
use crate::db::mysql::{mysql_row_string, mysql_snapshot_columns_query, mysql_snapshot_constraints_query, mysql_snapshot_indexes_query, mysql_snapshot_views_query};
|
||||
|
||||
let mut tables: BTreeMap<String, SnapshotTable> = BTreeMap::new();
|
||||
let mut views: Vec<SnapshotView> = Vec::new();
|
||||
|
||||
let rows = sqlx::query(&mysql_snapshot_columns_query(schema))
|
||||
.fetch_all(&mut *conn)
|
||||
.await
|
||||
.map_err(|e| format!("columns query failed: {e}"))?;
|
||||
for r in rows {
|
||||
let tname = mysql_row_string(&r, 0);
|
||||
let entry = tables.entry(tname.clone()).or_insert_with(|| SnapshotTable {
|
||||
schema: schema.to_string(),
|
||||
name: tname,
|
||||
..Default::default()
|
||||
});
|
||||
entry.columns.push(SnapshotColumn {
|
||||
name: mysql_row_string(&r, 1),
|
||||
data_type: mysql_row_string(&r, 2),
|
||||
nullable: mysql_row_string(&r, 3) == "YES",
|
||||
default_value: (!mysql_row_string(&r, 4).is_empty()).then(|| mysql_row_string(&r, 4)),
|
||||
});
|
||||
}
|
||||
|
||||
let rows = sqlx::query(&mysql_snapshot_constraints_query(schema))
|
||||
.fetch_all(&mut *conn)
|
||||
.await
|
||||
.map_err(|e| format!("constraints query failed: {e}"))?;
|
||||
for r in rows {
|
||||
let tname = mysql_row_string(&r, 0);
|
||||
let cname = mysql_row_string(&r, 1);
|
||||
let ctype = mysql_row_string(&r, 2);
|
||||
let col = mysql_row_string(&r, 3);
|
||||
let Some(entry) = tables.get_mut(&tname) else { continue };
|
||||
match ctype.as_str() {
|
||||
"PRIMARY KEY" => {
|
||||
if !entry.primary_key.contains(&col) {
|
||||
entry.primary_key.push(col);
|
||||
}
|
||||
}
|
||||
"UNIQUE" => {
|
||||
if let Some(u) = entry.uniques.iter_mut().find(|u| u.name == cname) {
|
||||
u.definition = if u.definition.is_empty() { col.clone() } else { format!("{}, {col}", u.definition) };
|
||||
} else {
|
||||
entry.uniques.push(SnapshotConstraint {
|
||||
name: cname,
|
||||
kind: "unique".into(),
|
||||
definition: col,
|
||||
});
|
||||
}
|
||||
}
|
||||
"FOREIGN KEY" => {
|
||||
let ref_table = mysql_row_string(&r, 4);
|
||||
let on_delete = mysql_row_string(&r, 5).to_uppercase();
|
||||
let on_update = mysql_row_string(&r, 6).to_uppercase();
|
||||
if let Some(fk) = entry.foreign_keys.iter_mut().find(|f| f.name == cname) {
|
||||
fk.columns.push(col);
|
||||
} else {
|
||||
entry.foreign_keys.push(SnapshotFk {
|
||||
name: cname,
|
||||
columns: vec![col],
|
||||
ref_schema: String::new(),
|
||||
ref_table,
|
||||
ref_columns: Vec::new(),
|
||||
on_delete,
|
||||
on_update,
|
||||
});
|
||||
}
|
||||
}
|
||||
"CHECK" => {
|
||||
if let Some(c) = entry.checks.iter_mut().find(|c| c.name == cname) {
|
||||
c.definition = format!("{} {}", c.definition, mysql_row_string(&r, 7));
|
||||
} else {
|
||||
entry.checks.push(SnapshotConstraint {
|
||||
name: cname,
|
||||
kind: "check".into(),
|
||||
definition: mysql_row_string(&r, 7),
|
||||
});
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
let rows = sqlx::query(&mysql_snapshot_indexes_query(schema))
|
||||
.fetch_all(&mut *conn)
|
||||
.await
|
||||
.map_err(|e| format!("indexes query failed: {e}"))?;
|
||||
for r in rows {
|
||||
let tname = mysql_row_string(&r, 0);
|
||||
let iname = mysql_row_string(&r, 1);
|
||||
let unique = mysql_row_string(&r, 2) == "0";
|
||||
let col = mysql_row_string(&r, 3);
|
||||
let Some(entry) = tables.get_mut(&tname) else { continue };
|
||||
if let Some(ix) = entry.indexes.iter_mut().find(|i| i.name == iname) {
|
||||
ix.columns.push(col);
|
||||
} else {
|
||||
entry.indexes.push(SnapshotIndex {
|
||||
name: iname,
|
||||
columns: vec![col],
|
||||
unique,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
let rows = sqlx::query(&mysql_snapshot_views_query(schema))
|
||||
.fetch_all(&mut *conn)
|
||||
.await
|
||||
.map_err(|e| format!("views query failed: {e}"))?;
|
||||
for r in rows {
|
||||
views.push(SnapshotView {
|
||||
schema: mysql_row_string(&r, 0),
|
||||
name: mysql_row_string(&r, 1),
|
||||
definition: mysql_row_string(&r, 2),
|
||||
materialized: false,
|
||||
});
|
||||
}
|
||||
|
||||
let mut snap = SchemaSnapshot {
|
||||
engine: engine.to_string(),
|
||||
schema: schema.to_string(),
|
||||
tables: tables.into_values().collect(),
|
||||
views,
|
||||
sequences: Vec::new(),
|
||||
enums: Vec::new(),
|
||||
object_count: 0,
|
||||
truncated: false,
|
||||
};
|
||||
apply_snapshot_cap(&mut snap);
|
||||
Ok(snap)
|
||||
}
|
||||
}
|
||||
|
||||
/// Double-quote an identifier (PG/SQLite) with embedded-quote doubling.
|
||||
pub(crate) fn quote_double(name: &str) -> String {
|
||||
format!("\"{}\"", name.replace('"', "\"\""))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "schema_diff.test.rs"]
|
||||
mod schema_diff_tests;
|
||||
@@ -0,0 +1,159 @@
|
||||
use super::*;
|
||||
|
||||
// ── Pure helpers (PG constraint row mapping) ────────────────────────
|
||||
|
||||
#[test]
|
||||
fn pg_fk_action_maps_codes() {
|
||||
assert_eq!(pg_fk_action("a"), "NO ACTION");
|
||||
assert_eq!(pg_fk_action("r"), "RESTRICT");
|
||||
assert_eq!(pg_fk_action("c"), "CASCADE");
|
||||
assert_eq!(pg_fk_action("n"), "SET NULL");
|
||||
assert_eq!(pg_fk_action("d"), "SET DEFAULT");
|
||||
assert_eq!(pg_fk_action("x"), "NO ACTION");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn snapshot_cap_sets_truncated() {
|
||||
// 5,001 objects → truncated report, never silent (spec §2)
|
||||
let mut snap = SchemaSnapshot::default();
|
||||
snap.engine = "postgresql".into();
|
||||
snap.schema = "s".into();
|
||||
for i in 0..5_001 {
|
||||
snap.tables.push(SnapshotTable {
|
||||
schema: "s".into(),
|
||||
name: format!("t{i}"),
|
||||
columns: vec![],
|
||||
primary_key: vec![],
|
||||
uniques: vec![],
|
||||
checks: vec![],
|
||||
indexes: vec![],
|
||||
foreign_keys: vec![],
|
||||
});
|
||||
}
|
||||
apply_snapshot_cap(&mut snap);
|
||||
assert!(snap.truncated);
|
||||
assert!(snap.object_count <= 5_000);
|
||||
assert_eq!(snap.tables.len(), snap.object_count);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn query_builders_target_the_right_catalogs() {
|
||||
let cols = crate::db::introspection::pg_diff_columns_query();
|
||||
assert!(cols.contains("pg_attribute") && cols.contains("format_type"));
|
||||
let cons = crate::db::introspection::pg_diff_constraints_query();
|
||||
assert!(cons.contains("pg_constraint") && cons.contains("contype"));
|
||||
let idx = crate::db::introspection::pg_diff_indexes_query();
|
||||
assert!(idx.contains("pg_index") && idx.contains("indisunique"));
|
||||
let views = crate::db::introspection::pg_diff_views_query();
|
||||
assert!(views.contains("pg_views") && views.contains("pg_matviews"));
|
||||
let seqs = crate::db::introspection::pg_diff_sequences_query();
|
||||
assert!(seqs.contains("information_schema.sequences"));
|
||||
let enums = crate::db::introspection::pg_diff_enums_query();
|
||||
assert!(enums.contains("pg_type") && enums.contains("pg_enum"));
|
||||
}
|
||||
|
||||
// ── SQLite capture (in-memory, real, NOT ignored) ──────────────────
|
||||
|
||||
#[test]
|
||||
fn sqlite_capture_reads_pragmas() {
|
||||
let conn = rusqlite::Connection::open_in_memory().unwrap();
|
||||
conn.execute_batch(
|
||||
"CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL DEFAULT 'x', email TEXT);
|
||||
CREATE UNIQUE INDEX idx_users_email ON users(email);
|
||||
CREATE TABLE orders (id INTEGER PRIMARY KEY, user_id INTEGER REFERENCES users(id) ON DELETE CASCADE);
|
||||
CREATE VIEW v_users AS SELECT id, name FROM users;",
|
||||
)
|
||||
.unwrap();
|
||||
let snap = capture_sqlite(&conn).expect("capture");
|
||||
assert_eq!(snap.engine, "sqlite");
|
||||
assert_eq!(snap.tables.len(), 2);
|
||||
let users = snap.tables.iter().find(|t| t.name == "users").unwrap();
|
||||
assert_eq!(users.primary_key, vec!["id"]);
|
||||
assert_eq!(users.columns.len(), 3);
|
||||
assert!(users.columns.iter().any(|c| c.name == "name" && !c.nullable));
|
||||
assert!(users.columns.iter().any(|c| c.name == "email" && c.nullable));
|
||||
assert!(users.indexes.iter().any(|i| i.name == "idx_users_email" && i.unique));
|
||||
let orders = snap.tables.iter().find(|t| t.name == "orders").unwrap();
|
||||
assert_eq!(orders.foreign_keys.len(), 1);
|
||||
let fk = &orders.foreign_keys[0];
|
||||
assert!(fk.ref_table.contains("users"));
|
||||
assert_eq!(fk.on_delete, "CASCADE");
|
||||
assert_eq!(snap.views.len(), 1);
|
||||
assert!(snap.views[0].definition.contains("SELECT"));
|
||||
}
|
||||
|
||||
/// Live PG capture — mirror the env/client conventions of the live PG tests in
|
||||
/// `commands/objects.test.rs`. Create two diverged tables in a temp schema,
|
||||
/// assert capture shape (columns with defaults, PK, FK with actions, indexes,
|
||||
/// views, a sequence, an enum).
|
||||
#[tokio::test]
|
||||
#[ignore]
|
||||
async fn pg_capture_live() {
|
||||
let h = std::env::var("GRIDLINE_TEST_PG_HOST").expect("set GRIDLINE_TEST_PG_HOST");
|
||||
let p: u16 = std::env::var("GRIDLINE_TEST_PG_PORT").unwrap_or_else(|_| "5432".into()).parse().unwrap();
|
||||
let u = std::env::var("GRIDLINE_TEST_PG_USER").expect("set GRIDLINE_TEST_PG_USER");
|
||||
let d = std::env::var("GRIDLINE_TEST_PG_DB").expect("set GRIDLINE_TEST_PG_DB");
|
||||
let pw = std::env::var("GRIDLINE_TEST_PG_PASSWORD").unwrap_or_default();
|
||||
let (client, conn) = tokio_postgres::connect(
|
||||
&format!("host={h} port={p} user={u} dbname={d} password={pw}"),
|
||||
tokio_postgres::NoTls,
|
||||
)
|
||||
.await
|
||||
.expect("connect to test PG");
|
||||
let handle = tokio::spawn(async move { let _ = conn.await; });
|
||||
|
||||
let schema = "gridline_diff_test";
|
||||
client
|
||||
.batch_execute(&format!(
|
||||
"DROP SCHEMA IF EXISTS {schema} CASCADE; CREATE SCHEMA {schema};"
|
||||
))
|
||||
.await
|
||||
.expect("create temp schema");
|
||||
|
||||
// Two diverged tables: one with PK/FK/index/default, one plain.
|
||||
client
|
||||
.batch_execute(&format!(
|
||||
"CREATE TABLE {schema}.parent (id serial PRIMARY KEY, code text UNIQUE);
|
||||
CREATE TABLE {schema}.child (
|
||||
id serial PRIMARY KEY,
|
||||
parent_id integer REFERENCES {schema}.parent(id) ON DELETE CASCADE ON UPDATE RESTRICT,
|
||||
name text NOT NULL DEFAULT 'x',
|
||||
email text
|
||||
);
|
||||
CREATE INDEX idx_child_email ON {schema}.child(email);
|
||||
CREATE VIEW {schema}.v_child AS SELECT id, name FROM {schema}.child;
|
||||
CREATE SEQUENCE {schema}.seq_test START 5 INCREMENT 2;
|
||||
CREATE TYPE {schema}.mood AS ENUM ('happy', 'sad');"
|
||||
))
|
||||
.await
|
||||
.expect("create objects");
|
||||
|
||||
let snap = capture_pg(&client, "postgresql", schema).await.expect("capture");
|
||||
assert_eq!(snap.engine, "postgresql");
|
||||
assert_eq!(snap.schema, schema);
|
||||
assert_eq!(snap.tables.len(), 2);
|
||||
|
||||
let child = snap.tables.iter().find(|t| t.name == "child").expect("child table");
|
||||
assert_eq!(child.primary_key, vec!["id"]);
|
||||
assert!(child.columns.iter().any(|c| c.name == "name" && !c.nullable && c.default_value.is_some()));
|
||||
assert!(child.columns.iter().any(|c| c.name == "email" && c.nullable));
|
||||
assert!(child.indexes.iter().any(|i| i.name == "idx_child_email"));
|
||||
assert_eq!(child.foreign_keys.len(), 1);
|
||||
let fk = &child.foreign_keys[0];
|
||||
assert!(fk.ref_table.contains("parent"));
|
||||
assert_eq!(fk.on_delete, "CASCADE");
|
||||
assert_eq!(fk.on_update, "RESTRICT");
|
||||
|
||||
let parent = snap.tables.iter().find(|t| t.name == "parent").expect("parent table");
|
||||
assert!(parent.uniques.iter().any(|u| u.kind == "unique"));
|
||||
|
||||
assert!(snap.views.iter().any(|v| v.name == "v_child"));
|
||||
assert!(snap.sequences.iter().any(|s| s.name == "seq_test" && s.start == "5"));
|
||||
assert!(snap.enums.iter().any(|e| e.name == "mood" && e.labels == vec!["happy", "sad"]));
|
||||
|
||||
client
|
||||
.batch_execute(&format!("DROP SCHEMA {schema} CASCADE;"))
|
||||
.await
|
||||
.expect("drop temp schema");
|
||||
let _ = handle;
|
||||
}
|
||||
@@ -345,6 +345,125 @@ pub struct Relationship {
|
||||
pub cardinality: String,
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Schema diff (0.8.0)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotColumn {
|
||||
pub name: String,
|
||||
pub data_type: String,
|
||||
pub nullable: bool,
|
||||
pub default_value: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotIndex {
|
||||
pub name: String,
|
||||
pub columns: Vec<String>,
|
||||
pub unique: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotFk {
|
||||
pub name: String,
|
||||
pub columns: Vec<String>,
|
||||
pub ref_schema: String,
|
||||
pub ref_table: String,
|
||||
pub ref_columns: Vec<String>,
|
||||
pub on_delete: String,
|
||||
pub on_update: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotConstraint {
|
||||
pub name: String,
|
||||
/// "unique" | "check"
|
||||
pub kind: String,
|
||||
/// check: expression; unique: comma-joined columns
|
||||
pub definition: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotTable {
|
||||
pub schema: String,
|
||||
pub name: String,
|
||||
pub columns: Vec<SnapshotColumn>,
|
||||
pub primary_key: Vec<String>,
|
||||
pub uniques: Vec<SnapshotConstraint>,
|
||||
pub checks: Vec<SnapshotConstraint>,
|
||||
pub indexes: Vec<SnapshotIndex>,
|
||||
pub foreign_keys: Vec<SnapshotFk>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotView {
|
||||
pub schema: String,
|
||||
pub name: String,
|
||||
pub definition: String,
|
||||
pub materialized: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotSequence {
|
||||
pub schema: String,
|
||||
pub name: String,
|
||||
pub start: String,
|
||||
pub increment: String,
|
||||
pub minimum: String,
|
||||
pub maximum: String,
|
||||
pub cycle: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SnapshotEnum {
|
||||
pub schema: String,
|
||||
pub name: String,
|
||||
pub labels: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
|
||||
pub struct SchemaSnapshot {
|
||||
/// "postgresql" | "mysql" | "mariadb" | "sqlite" — the CONNECTION's db_type
|
||||
pub engine: String,
|
||||
pub schema: String,
|
||||
pub tables: Vec<SnapshotTable>,
|
||||
pub views: Vec<SnapshotView>,
|
||||
pub sequences: Vec<SnapshotSequence>,
|
||||
pub enums: Vec<SnapshotEnum>,
|
||||
pub object_count: usize,
|
||||
pub truncated: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||
pub struct DiffDetailLine {
|
||||
pub label: String,
|
||||
pub old: String,
|
||||
pub new: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||
pub struct DiffItem {
|
||||
/// "table" | "column" | "constraint" | "index" | "view" | "sequence" | "enum"
|
||||
pub object_type: String,
|
||||
/// For column/constraint/index items: "<table>.<name>"; else the object name
|
||||
pub name: String,
|
||||
/// "added" | "removed" | "changed"
|
||||
pub kind: String,
|
||||
pub detail: Vec<DiffDetailLine>,
|
||||
/// Target-side statements (usually exactly one per item)
|
||||
pub sync_sql: Vec<String>,
|
||||
pub destructive: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DiffReport {
|
||||
pub items: Vec<DiffItem>,
|
||||
pub source_label: String,
|
||||
pub target_label: String,
|
||||
pub truncated: bool,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
Reference in New Issue
Block a user