coven_database/
migration_change.rs

1//! App-facing conversion values retain the identity and references of their source columns.
2
3use crate::migration_names::MigrationMatch;
4use crate::schema::{Schema, TableSchema};
5use crate::write_encoding::{sql_value, value};
6use crate::{DbError, RowKey};
7use coven_format::value::Value as WireValue;
8use coven_format::write::RowChange as WrittenRow;
9use coven_merge::{ColumnValue, ForeignKey, Operation, Parent};
10use rusqlite::types::Value;
11use std::collections::BTreeMap;
12
13/// One row change of a waiting write (E13).
14#[derive(Clone, Debug, PartialEq)]
15pub struct RowChange {
16    /// The table being changed.
17    pub table: String,
18    /// The row's primary key.
19    pub key: RowKey,
20    /// The row operation.
21    pub op: ChangeOp,
22    /// The columns an insert or update writes, or the cells a delete removes.
23    pub columns: Vec<ColumnChange>,
24    original: WrittenRow,
25}
26
27impl RowChange {
28    /// Rename a column, carrying its original references with it.
29    pub fn rename_column(&mut self, from: &str, to: &str) {
30        for column in &mut self.columns {
31            if column.name == from {
32                column.name = to.to_owned();
33            }
34            if let Some(source) = &mut column.source {
35                for name in source.renames.values_mut() {
36                    if name == from {
37                        *name = to.to_owned();
38                    }
39                }
40                source.renames.insert(from.to_owned(), to.to_owned());
41            }
42        }
43    }
44
45    pub(crate) fn read(row: &WrittenRow, schema: &Schema) -> Result<Self, DbError> {
46        let table = table(schema, &row.row.table)?;
47        // An insert or update shows the values it writes; a delete, the
48        // cells it removes. An update carries no value it replaced (D5.5).
49        let (op, values) = match &row.change.operation {
50            Operation::Insert(columns) => (ChangeOp::Insert, written(columns)),
51            Operation::Update(columns) => (ChangeOp::Update, written(columns)),
52            Operation::Delete => (
53                ChangeOp::Delete,
54                row.old.iter().map(|(name, value)| (name, value, None)).collect(),
55            ),
56        };
57        let columns = values
58            .into_iter()
59            .map(|(name, value, parents)| {
60                let value = sql_value(value);
61                let constraints = references(table, name, schema)?;
62                let reference = (!constraints.is_empty()).then(|| ReferenceColumn {
63                    constraints,
64                    value: value.clone(),
65                    parents: parents.cloned().unwrap_or_default(),
66                });
67                Ok(ColumnChange {
68                    name: name.clone(),
69                    value,
70                    source: Some(ColumnSource {
71                        name: name.clone(),
72                        reference,
73                        renames: BTreeMap::new(),
74                    }),
75                })
76            })
77            .collect::<Result<_, DbError>>()?;
78        Ok(Self {
79            table: row.row.table.clone(),
80            key: RowKey::decode(&row.row.key)?,
81            op,
82            columns,
83            original: row.clone(),
84        })
85    }
86
87    pub(crate) fn write(
88        &self,
89        schema: &Schema,
90        names: &MigrationMatch,
91    ) -> Result<WrittenRow, DbError> {
92        let original = &self.original;
93        let table = table(schema, &self.table)?;
94        let renames = self.sources().map(|(a, b)| (a.to_owned(), b.to_owned())).collect();
95        let mut reference_sources = std::collections::BTreeSet::new();
96        let mut old = BTreeMap::new();
97        let mut new = BTreeMap::new();
98        let mut columns_seen = std::collections::BTreeSet::new();
99        for column in &self.columns {
100            if !columns_seen.insert(&column.name) {
101                return Err(DbError::MigrationDuplicateColumn {
102                    table: self.table.clone(),
103                    column: column.name.clone(),
104                });
105            }
106            if !table.columns.iter().any(|c| c.name == column.name) {
107                return Err(DbError::MigrationColumnMissing {
108                    table: self.table.clone(),
109                    column: column.name.clone(),
110                });
111            }
112            let reference = column.source.as_ref().and_then(|source| source.reference.as_ref());
113            let constraints = references(table, &column.name, schema)?;
114            if reference.is_some_and(|r| r.value != column.value)
115                || (!constraints.is_empty() && reference.is_none())
116            {
117                return Err(reference_error(&self.table, &column.name));
118            }
119            if let Some(reference) = reference {
120                let mapped = reference
121                    .constraints
122                    .iter()
123                    .filter_map(|key| names.foreign_key(&original.row.table, key, Some(&renames)))
124                    .collect::<Vec<_>>();
125                if constraints.iter().any(|key| !mapped.contains(key))
126                    || (!constraints.is_empty()
127                        && !reference_sources
128                            .insert(&column.source.as_ref().expect("reference source").name))
129                {
130                    return Err(reference_error(&self.table, &column.name));
131                }
132            }
133            let written = value((&column.value).into())?;
134            if self.op == ChangeOp::Delete {
135                old.insert(column.name.clone(), written);
136            } else {
137                let parents = match reference {
138                    Some(reference) => rename_parents(
139                        &reference.parents,
140                        &original.row.table,
141                        &renames,
142                        names,
143                        &constraints,
144                    )?,
145                    None => BTreeMap::new(),
146                };
147                new.insert(column.name.clone(), ColumnValue { value: written, parents });
148            }
149        }
150        let key: Vec<_> =
151            self.key.values().iter().map(|v| value(v.into())).collect::<Result<_, _>>()?;
152        coven_format::key::encode_key(&key)?;
153        let primary = table
154            .indices
155            .iter()
156            .find(|index| index.primary)
157            .ok_or_else(|| DbError::MigrationPrimaryKeyMissing { table: self.table.clone() })?;
158        if key.len() != primary.columns.len() {
159            return Err(DbError::MigrationKeyArity {
160                table: self.table.clone(),
161                expected: primary.columns.len(),
162                actual: key.len(),
163            });
164        }
165        let original_op = match original.change.operation {
166            Operation::Insert(_) => ChangeOp::Insert,
167            Operation::Update(_) => ChangeOp::Update,
168            Operation::Delete => ChangeOp::Delete,
169        };
170        let key = crate::write_rows::equality_key(&key)?;
171        if self.op != original_op
172            || key != original.row.key
173            || names.tables.get(&original.row.table) != Some(&table.name)
174        {
175            return Err(coven_format::Error::Invalid {
176                field: "converted row identity or operation",
177                rule: coven_format::error::Rule::ColumnOperation,
178            }
179            .into());
180        }
181        Ok(WrittenRow {
182            row: coven_merge::RowId {
183                table: table.name.clone(),
184                key,
185                audience: original.row.audience.clone(),
186            },
187            change: coven_merge::Change {
188                generation: original.change.generation,
189                operation: match self.op {
190                    ChangeOp::Insert => Operation::Insert(new),
191                    ChangeOp::Update => Operation::Update(new),
192                    ChangeOp::Delete => Operation::Delete,
193                },
194            },
195            old,
196            sources: coven_format::write_sources::WriteSources {
197                generation: original.sources.generation.clone(),
198                setters: original
199                    .sources
200                    .setters
201                    .iter()
202                    .filter_map(|(name, source)| {
203                        let mapped = self
204                            .columns
205                            .iter()
206                            .find(|c| c.source.as_ref().is_some_and(|s| s.name == *name));
207                        match mapped {
208                            Some(column) => Some((column.name.clone(), source.clone())),
209                            None if self.op == ChangeOp::Delete => {
210                                Some((name.clone(), source.clone()))
211                            }
212                            None => None,
213                        }
214                    })
215                    .collect(),
216            },
217            // A converted copy keeps naming the original creation it copied.
218            // Operations stay in the original record, which owner and
219            // foreign-key checks read before conversion (§18.6).
220            creator: original.creator.clone(),
221            execution: Default::default(),
222        })
223    }
224
225    fn sources(&self) -> impl Iterator<Item = (&str, &str)> {
226        self.columns.iter().flat_map(|c| {
227            c.source.iter().flat_map(move |s| {
228                s.renames
229                    .iter()
230                    .map(|(from, to)| (from.as_str(), to.as_str()))
231                    .chain(std::iter::once((s.name.as_str(), c.name.as_str())))
232            })
233        })
234    }
235}
236
237/// The operation a waiting write performs on a row.
238#[derive(Clone, Copy, Debug, Eq, PartialEq)]
239pub enum ChangeOp {
240    /// Insert the row.
241    Insert,
242    /// Change columns of the row.
243    Update,
244    /// Delete the row.
245    Delete,
246}
247
248/// One column of a change, retaining its original parent generations privately.
249#[derive(Clone, Debug, PartialEq)]
250pub struct ColumnChange {
251    /// The column name.
252    pub name: String,
253    /// The value an insert or update writes, or the cell a delete removes.
254    /// `Value::Null` is SQL NULL. An update shows no value it replaced.
255    pub value: Value,
256    source: Option<ColumnSource>,
257}
258
259#[derive(Clone, Debug, PartialEq)]
260struct ColumnSource {
261    name: String,
262    reference: Option<ReferenceColumn>,
263    renames: BTreeMap<String, String>,
264}
265
266#[derive(Clone, Debug, PartialEq)]
267struct ReferenceColumn {
268    constraints: Vec<ForeignKey>,
269    value: Value,
270    parents: BTreeMap<ForeignKey, Parent>,
271}
272
273impl ColumnChange {
274    /// A column the conversion adds; it carries no references.
275    pub fn new(name: impl Into<String>, value: Value) -> Self {
276        Self { name: name.into(), value, source: None }
277    }
278}
279
280/// Each column an insert or update writes, with its value and parents.
281type Written<'a> = Vec<(&'a String, &'a WireValue, Option<&'a BTreeMap<ForeignKey, Parent>>)>;
282
283fn written(columns: &BTreeMap<String, ColumnValue<WireValue>>) -> Written<'_> {
284    columns.iter().map(|(name, column)| (name, &column.value, Some(&column.parents))).collect()
285}
286
287fn table<'a>(schema: &'a Schema, name: &str) -> Result<&'a TableSchema, DbError> {
288    schema
289        .tables
290        .get(&name.to_ascii_lowercase())
291        .ok_or_else(|| DbError::MigrationTableMissing { table: name.into() })
292}
293
294fn references(
295    source: &TableSchema,
296    name: &str,
297    schema: &Schema,
298) -> Result<Vec<ForeignKey>, DbError> {
299    source
300        .foreign_keys_of(name)
301        .map(|fk| {
302            table(schema, &fk.target)?;
303            Ok(schema.foreign_key(source, fk))
304        })
305        .collect()
306}
307
308fn rename_parents(
309    parents: &BTreeMap<ForeignKey, Parent>,
310    source_table: &str,
311    renames: &BTreeMap<String, String>,
312    names: &MigrationMatch,
313    constraints: &[ForeignKey],
314) -> Result<BTreeMap<ForeignKey, Parent>, DbError> {
315    let mut renamed = BTreeMap::new();
316    for (key, parent) in parents {
317        let Some(identity) = names.foreign_key(source_table, key, Some(renames)) else {
318            continue;
319        };
320        if !constraints.contains(&identity) {
321            continue;
322        }
323        let mut parent = parent.clone();
324        parent.row.table = identity.parent.clone();
325        match renamed.entry(identity) {
326            std::collections::btree_map::Entry::Vacant(entry) => {
327                entry.insert(parent);
328            }
329            std::collections::btree_map::Entry::Occupied(entry) => {
330                return Err(DbError::MigrationReferenceCollision {
331                    table: source_table.into(),
332                    reference: entry.key().clone(),
333                })
334            }
335        }
336    }
337    Ok(renamed)
338}
339
340fn reference_error(table: &str, column: &str) -> DbError {
341    DbError::MigrationReference { table: table.to_owned(), column: column.to_owned() }
342}
343
344#[cfg(test)]
345#[path = "migration_change_tests.rs"]
346mod tests;
coven_database/migration_change.rs

coven_database/
migration_change.rs
