1//! The plan that brings a sidecar's foreign tables in line with the
2//! target's catalog, one change at a time. Unchanged tables are never
3//! touched, and nothing is dropped and imported again, so the sidecar's
4//! views, grants and functions over a foreign table survive a sync.
5//!
6//! A column is matched by its remote name (postgres_fdw's `column_name`
7//! option, else the local name). A rename is recognised when exactly one
8//! column disappeared and one appeared at the same position with the
9//! same type and nullability; anything else is a drop and an add, which
10//! the dependency check refuses when a view uses the dropped column.
11
12use std::fmt;
13
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct Column {
16    /// The name on the target (the `column_name` option locally).
17    pub name: String,
18    /// `format_type` of the column, as the target reports it.
19    pub type_sql: String,
20    pub not_null: bool,
21}
22
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct Table {
25    pub schema: String,
26    pub name: String,
27    /// In attnum order, dropped columns left out.
28    pub columns: Vec<Column>,
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub enum Change {
33    CreateTable { table: Table },
34    DropTable { schema: String, name: String },
35    AddColumn { schema: String, table: String, column: Column },
36    DropColumn { schema: String, table: String, column: String },
37    RenameColumn { schema: String, table: String, from: String, to: String },
38    AlterType { schema: String, table: String, column: String, type_sql: String },
39    SetNotNull { schema: String, table: String, column: String, not_null: bool },
40}
41
42/// What a change can break: the local object whose dependents
43/// (views, grants on columns, functions) must be checked in pg_depend
44/// before it is applied. `None` for changes nothing can depend on.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct Touches<'a> {
47    pub schema: &'a str,
48    pub table: &'a str,
49    /// `None` for the whole table.
50    pub column: Option<&'a str>,
51}
52
53impl Change {
54    pub fn touches(&self) -> Option<Touches<'_>> {
55        match self {
56            Change::DropTable { schema, name } => Some(Touches { schema, table: name, column: None }),
57            Change::DropColumn { schema, table, column }
58            | Change::AlterType { schema, table, column, .. } => {
59                Some(Touches { schema, table, column: Some(column) })
60            }
61            _ => None,
62        }
63    }
64
65    /// The statement that applies the change on the sidecar.
66    pub fn sql(&self, server: &str) -> String {
67        match self {
68            Change::CreateTable { table } => {
69                let cols: Vec<String> = table
70                    .columns
71                    .iter()
72                    .map(|c| {
73                        format!(
74                            "{} {} OPTIONS (column_name {}){}",
75                            ident(&c.name),
76                            c.type_sql,
77                            lit(&c.name),
78                            if c.not_null { " NOT NULL" } else { "" }
79                        )
80                    })
81                    .collect();
82                format!(
83                    "CREATE FOREIGN TABLE {}.{} ({}) SERVER {} OPTIONS (schema_name {}, table_name {})",
84                    ident(&table.schema),
85                    ident(&table.name),
86                    cols.join(", "),
87                    ident(server),
88                    lit(&table.schema),
89                    lit(&table.name)
90                )
91            }
92            Change::DropTable { schema, name } => {
93                format!("DROP FOREIGN TABLE {}.{}", ident(schema), ident(name))
94            }
95            Change::AddColumn { schema, table, column } => format!(
96                "ALTER FOREIGN TABLE {}.{} ADD COLUMN {} {} OPTIONS (column_name {}){}",
97                ident(schema),
98                ident(table),
99                ident(&column.name),
100                column.type_sql,
101                lit(&column.name),
102                if column.not_null { " NOT NULL" } else { "" }
103            ),
104            Change::DropColumn { schema, table, column } => format!(
105                "ALTER FOREIGN TABLE {}.{} DROP COLUMN {}",
106                ident(schema),
107                ident(table),
108                ident(column)
109            ),
110            Change::RenameColumn { schema, table, from, to } => format!(
111                "ALTER FOREIGN TABLE {s}.{t} RENAME COLUMN {f} TO {n}; \
112                 ALTER FOREIGN TABLE {s}.{t} ALTER COLUMN {n} OPTIONS (SET column_name {l})",
113                s = ident(schema),
114                t = ident(table),
115                f = ident(from),
116                n = ident(to),
117                l = lit(to)
118            ),
119            Change::AlterType { schema, table, column, type_sql } => format!(
120                "ALTER FOREIGN TABLE {}.{} ALTER COLUMN {} TYPE {}",
121                ident(schema),
122                ident(table),
123                ident(column),
124                type_sql
125            ),
126            Change::SetNotNull { schema, table, column, not_null } => format!(
127                "ALTER FOREIGN TABLE {}.{} ALTER COLUMN {} {} NOT NULL",
128                ident(schema),
129                ident(table),
130                ident(column),
131                if *not_null { "SET" } else { "DROP" }
132            ),
133        }
134    }
135}
136
137impl fmt::Display for Change {
138    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
139        f.write_str(&self.sql("<server>"))
140    }
141}
142
143/// The changes that make `local` (the sidecar's foreign tables in the
144/// imported schemas) match `target` (the target's tables in the same
145/// schemas). Tables are compared by schema and name; order follows
146/// `target`, then drops in `local`'s order.
147pub fn plan(target: &[Table], local: &[Table]) -> Vec<Change> {
148    let find = |set: &'_ [Table], t: &Table| -> Option<usize> {
149        set.iter().position(|o| o.schema == t.schema && o.name == t.name)
150    };
151    let mut changes = Vec::new();
152    for t in target {
153        match find(local, t) {
154            None => changes.push(Change::CreateTable { table: t.clone() }),
155            Some(i) => columns(t, &local[i], &mut changes),
156        }
157    }
158    for l in local {
159        if find(target, l).is_none() {
160            changes.push(Change::DropTable { schema: l.schema.clone(), name: l.name.clone() });
161        }
162    }
163    changes
164}
165
166fn columns(target: &Table, local: &Table, out: &mut Vec<Change>) {
167    let (schema, table) = (target.schema.clone(), target.name.clone());
168    let has = |cols: &[Column], name: &str| cols.iter().any(|c| c.name == name);
169    let gone: Vec<(usize, &Column)> =
170        local.columns.iter().enumerate().filter(|(_, c)| !has(&target.columns, &c.name)).collect();
171    let new: Vec<(usize, &Column)> =
172        target.columns.iter().enumerate().filter(|(_, c)| !has(&local.columns, &c.name)).collect();
173
174    let renamed = match (gone.as_slice(), new.as_slice()) {
175        ([(i, g)], [(j, n)]) if i == j && g.type_sql == n.type_sql && g.not_null == n.not_null => {
176            out.push(Change::RenameColumn {
177                schema: schema.clone(),
178                table: table.clone(),
179                from: g.name.clone(),
180                to: n.name.clone(),
181            });
182            true
183        }
184        _ => false,
185    };
186    if !renamed {
187        for (_, g) in &gone {
188            out.push(Change::DropColumn {
189                schema: schema.clone(),
190                table: table.clone(),
191                column: g.name.clone(),
192            });
193        }
194    }
195    for t in &target.columns {
196        let Some(l) = local.columns.iter().find(|l| l.name == t.name) else {
197            if !renamed {
198                out.push(Change::AddColumn {
199                    schema: schema.clone(),
200                    table: table.clone(),
201                    column: t.clone(),
202                });
203            }
204            continue;
205        };
206        if l.type_sql != t.type_sql {
207            out.push(Change::AlterType {
208                schema: schema.clone(),
209                table: table.clone(),
210                column: t.name.clone(),
211                type_sql: t.type_sql.clone(),
212            });
213        }
214        if l.not_null != t.not_null {
215            out.push(Change::SetNotNull {
216                schema: schema.clone(),
217                table: table.clone(),
218                column: t.name.clone(),
219                not_null: t.not_null,
220            });
221        }
222    }
223}
224
225pub fn ident(s: &str) -> String {
226    format!("\"{}\"", s.replace('"', "\"\""))
227}
228
229pub fn lit(s: &str) -> String {
230    format!("'{}'", s.replace('\'', "''"))
231}
232
233#[cfg(test)]
234mod tests {
235    use super::*;
236
237    fn col(name: &str, ty: &str) -> Column {
238        Column { name: name.into(), type_sql: ty.into(), not_null: false }
239    }
240    fn table(name: &str, cols: &[Column]) -> Table {
241        Table { schema: "public".into(), name: name.into(), columns: cols.to_vec() }
242    }
243
244    #[test]
245    fn unchanged_tables_plan_nothing() {
246        let t = [table("a", &[col("id", "integer")])];
247        assert!(plan(&t, &t).is_empty());
248    }
249
250    #[test]
251    fn new_table_new_column_dropped_column_and_dropped_table() {
252        let local = [
253            table("people", &[col("id", "integer"), col("old", "text"), col("name", "text")]),
254            table("gone", &[col("id", "integer")]),
255            table("same", &[col("id", "integer")]),
256        ];
257        let target = [
258            table("people", &[col("id", "integer"), col("name", "text"), col("email", "text")]),
259            table("same", &[col("id", "integer")]),
260            table("fresh", &[col("id", "bigint")]),
261        ];
262        let p = plan(&target, &local);
263        assert_eq!(
264            p,
265            vec![
266                Change::DropColumn { schema: "public".into(), table: "people".into(), column: "old".into() },
267                Change::AddColumn { schema: "public".into(), table: "people".into(), column: col("email", "text") },
268                Change::CreateTable { table: target[2].clone() },
269                Change::DropTable { schema: "public".into(), name: "gone".into() },
270            ]
271        );
272        assert!(p.iter().all(|c| !c.sql("s").contains("\"same\"")));
273    }
274
275    #[test]
276    fn rename_in_place_and_type_change() {
277        let local = [table("t", &[col("id", "integer"), col("nm", "text")])];
278        let target = [table("t", &[col("id", "bigint"), col("name", "text")])];
279        let p = plan(&target, &local);
280        assert!(matches!(&p[0], Change::RenameColumn { from, to, .. } if from == "nm" && to == "name"));
281        assert!(matches!(&p[1], Change::AlterType { column, .. } if column == "id"));
282        assert_eq!(p[0].touches(), None);
283        assert!(p[1].touches().is_some());
284    }
285
286    #[test]
287    fn quoting() {
288        assert_eq!(ident("a\"b"), "\"a\"\"b\"");
289        assert_eq!(lit("o'k"), "'o''k'");
290    }
291}