coven_database/store/store_session/merge_materialization_transaction/
changeset_application.rs

1//! Apply a changeset to the connection, resolving conflicts as each row lands.
2//!
3//! Two stages. First a column-level three-way premerge
4//! (`premerge_losing_update_columns`): when an incoming UPDATE loses row
5//! arbitration, the columns it moved away from a base the local row still holds
6//! are folded into the local row, so concurrent edits to *different* columns of
7//! one row both survive. Then the changeset is applied with
8//! `arbitrate_row_conflict` as the conflict handler, which picks the winning row
9//! by `_updated_at` (remove-wins for deletes, a future-skew bound on the
10//! comparison) for every collision the premerge did not already fold in.
11//!
12//! Within a single changeset, SQLite defers FK checks — parent and child rows in
13//! the same changeset are applied in recording order. Cross-changeset FK
14//! dependencies are handled by applying changesets in seq order (parents are
15//! always in earlier changesets than children).
16//!
17//! If a FK violation remains after applying a changeset, the conflict handler
18//! reports it via `FOREIGN_KEY`. The production materializer validates the
19//! deferred foreign keys after its whole atomic replay step; the test wrapper
20//! also returns a flag so isolated changeset tests can roll back. A non-FK
21//! constraint conflict marks the whole changeset rejected; the caller rolls its
22//! transaction back instead of committing the rows that happened not to conflict.
23
24use std::collections::HashSet;
25#[cfg(any(test, feature = "test-utils"))]
26use std::sync::atomic::{AtomicBool, Ordering};
27use std::sync::{Arc, Mutex};
28
29use fallible_streaming_iterator::FallibleStreamingIterator;
30use rusqlite::hooks::Action;
31use rusqlite::session::{ChangesetItem, ChangesetIter, ConflictAction, ConflictType};
32use rusqlite::types::{Value, ValueRef};
33use rusqlite::{params_from_iter, Connection, OptionalExtension, ToSql};
34use tracing::warn;
35
36use super::conflict::{
37    arbitrate_row_conflict, compare_lww_stamps, IncomingTimestampPolicy, LwwComparison, TableSchema,
38};
39use crate::changeset::{value_ref_to_string, UpdateValue};
40use crate::changeset_identity::validate_changeset_row_identities;
41use crate::{quote_ident, ChangesetIdentityError, DbError};
42use coven_protocol::hlc::Timestamp;
43#[cfg(any(test, feature = "test-utils"))]
44use coven_protocol::synced_schema::SyncedTable;
45
46use super::MergeMaterializationTransaction;
47
48/// Result of applying a changeset.
49#[cfg_attr(
50    not(any(test, feature = "test-utils")),
51    allow(unreachable_pub),
52    doc = "Public when the `test-utils` feature exposes changeset application."
53)]
54pub struct ApplyResult {
55    /// True if any FK violations were reported. The caller may retry this
56    /// changeset after applying other changesets that contain the missing parent
57    /// rows.
58    #[cfg(any(test, feature = "test-utils"))]
59    pub had_fk_violations: bool,
60    /// Tables that hit non-retryable SQLite constraint conflicts. The caller must
61    /// roll back the transaction when this is non-empty.
62    pub constraint_conflict_tables: Vec<String>,
63    /// Incoming rows whose exact row value won arbitration. A missing stamp
64    /// identifies a winning deletion.
65    #[cfg(any(test, feature = "test-utils"))]
66    pub winning_rows: Vec<WinningRow>,
67}
68
69#[derive(Clone, Debug, PartialEq, Eq)]
70pub struct WinningRow {
71    pub table: String,
72    pub row_id: String,
73    pub row_stamp: Option<String>,
74}
75
76#[derive(Clone, Debug)]
77struct IncomingRow {
78    table: String,
79    row_id: String,
80    row_stamp: Option<String>,
81}
82
83/// Changeset bytes paired with the exact synced schema that validated their row
84/// identities. Construction is the one identity-validation boundary; apply only
85/// accepts this type, so callers cannot parse once for classification and then
86/// parse the same bytes again during mutation.
87pub struct ValidatedChangeset<B> {
88    bytes: B,
89    schema: Arc<TableSchema>,
90}
91
92impl<B: AsRef<[u8]>> ValidatedChangeset<B> {
93    pub fn new(bytes: B, schema: Arc<TableSchema>) -> Result<Self, ChangesetIdentityError> {
94        validate_changeset_row_identities(bytes.as_ref(), schema.synced_tables())?;
95        Ok(Self { bytes, schema })
96    }
97
98    pub fn bytes(&self) -> &[u8] {
99        self.bytes.as_ref()
100    }
101
102    pub fn schema(&self) -> &TableSchema {
103        &self.schema
104    }
105
106    pub fn validate_subset<C: AsRef<[u8]>>(
107        &self,
108        bytes: C,
109    ) -> Result<ValidatedChangeset<C>, ChangesetIdentityError> {
110        ValidatedChangeset::new(bytes, self.schema.clone())
111    }
112}
113
114/// Apply `bytes` to `conn`, resolving conflicts (premerge + row arbitration),
115/// building the [`TableSchema`] from `tables` once. A convenience wrapper over
116/// `resolve_and_apply_changeset_with_schema` for callers that apply a single
117/// changeset and don't already hold a schema (tests, snapshot round-trips).
118///
119/// `receiver_wall_ms` is the receiver's current wall-clock millis, against which
120/// a grossly-future incoming `_updated_at` is refused (see `arbitrate_row_conflict`).
121#[cfg(any(test, feature = "test-utils"))]
122pub fn resolve_and_apply_changeset(
123    conn: &Connection,
124    store_dir: &coven_foundation::store_dir::StoreDir,
125    bytes: &[u8],
126    tables: &[SyncedTable],
127    receiver_wall_ms: u64,
128) -> Result<ApplyResult, DbError> {
129    let schema = Arc::new(TableSchema::from_db(conn, tables)?);
130    resolve_and_apply_changeset_with_schema(conn, store_dir, bytes, schema, receiver_wall_ms)
131}
132
133/// Apply `bytes` to `conn`, resolving conflicts against a pre-built
134/// [`TableSchema`]: a column-level premerge of losing UPDATEs
135/// (`premerge_losing_update_columns`) followed by an apply whose conflict
136/// closure arbitrates every remaining row collision.
137///
138/// The schema's per-table `_updated_at` column index map is derived once (from
139/// the live schema, so future migrations that add columns are safe) and reused
140/// across every changeset in a pull, rather than re-querying `PRAGMA table_info`
141/// per changeset. The conflict closure resolves each conflicting row's table from
142/// its operation and decides REPLACE/OMIT by comparing `_updated_at`;
143/// FK violations flip a shared flag for the caller to retry; non-FK constraint
144/// conflicts are collected so the caller can surface the rejected changeset.
145///
146/// `schema` is an `Arc` so the same map moves into the `'static` conflict closure
147/// without re-deriving it per call. `receiver_wall_ms` is the receiver's current
148/// wall-clock millis, read once by the caller and moved into the closure to bound
149/// a grossly-future incoming `_updated_at` (see `arbitrate_row_conflict`).
150#[cfg(any(test, feature = "test-utils"))]
151pub(crate) fn resolve_and_apply_changeset_with_schema(
152    conn: &Connection,
153    store_dir: &coven_foundation::store_dir::StoreDir,
154    bytes: &[u8],
155    schema: Arc<TableSchema>,
156    receiver_wall_ms: u64,
157) -> Result<ApplyResult, DbError> {
158    let changeset = ValidatedChangeset::new(bytes, schema).map_err(DbError::from)?;
159    let tx = conn.unchecked_transaction().map_err(DbError::from)?;
160    let result = MergeMaterializationTransaction::from_store(
161        crate::store::store_session::StoreTransaction::new(&tx, store_dir),
162    )
163    .apply_changeset(
164        changeset,
165        IncomingTimestampPolicy::Received { receiver_wall_ms },
166    )?;
167    if result.had_fk_violations || !result.constraint_conflict_tables.is_empty() {
168        tx.rollback().map_err(DbError::from)?;
169    } else {
170        tx.commit().map_err(DbError::from)?;
171    }
172    Ok(result)
173}
174
175impl MergeMaterializationTransaction<'_, '_> {
176    pub(crate) fn apply_changeset<B: AsRef<[u8]>>(
177        &self,
178        changeset: ValidatedChangeset<B>,
179        timestamp_policy: IncomingTimestampPolicy,
180    ) -> Result<ApplyResult, DbError> {
181        let conn = self.store.transaction;
182        let ValidatedChangeset { bytes, schema } = changeset;
183        let bytes = bytes.as_ref();
184        #[cfg(any(test, feature = "test-utils"))]
185        let incoming_rows = incoming_rows(bytes, &schema)?;
186
187        #[cfg(any(test, feature = "test-utils"))]
188        let fk_flag = Arc::new(AtomicBool::new(false));
189        let constraint_conflict_tables = Arc::new(Mutex::new(Vec::new()));
190        let premerged_updates =
191            premerge_losing_update_columns(conn, bytes, &schema, timestamp_policy)?;
192
193        #[cfg(any(test, feature = "test-utils"))]
194        let closure_flag = fk_flag.clone();
195        let closure_constraint_conflict_tables = constraint_conflict_tables.clone();
196        let closure_schema = schema.clone();
197        conn.apply_strm(
198            &mut &bytes[..],
199            None::<fn(&str) -> bool>,
200            move |conflict_type, item| {
201                // A FOREIGN_KEY conflict's iterator supports ONLY `fk_conflicts()`;
202                // calling `op()`/`new_value()`/`conflict()` on it is undefined (it
203                // crashes the process). Resolve it first, without touching the row.
204                if conflict_type == ConflictType::SQLITE_CHANGESET_FOREIGN_KEY {
205                    #[cfg(any(test, feature = "test-utils"))]
206                    closure_flag.store(true, Ordering::Relaxed);
207                    return ConflictAction::SQLITE_CHANGESET_OMIT;
208                }
209                // Every other conflict type exposes the operation, so the table name
210                // (needed to find the `_updated_at` column) is readable.
211                let (table, op_code) = match item.op() {
212                    Ok(op) => (op.table_name().to_string(), op.code()),
213                    Err(error) => {
214                        warn!(error = %error, "failed to read changeset conflict operation; aborting apply");
215                        return ConflictAction::SQLITE_CHANGESET_ABORT;
216                    }
217                };
218                if conflict_type == ConflictType::SQLITE_CHANGESET_CONSTRAINT {
219                    warn!(
220                        table = %table,
221                        "changeset hit a non-retryable SQLite constraint conflict; rejecting changeset"
222                    );
223                    match closure_constraint_conflict_tables.lock() {
224                        Ok(mut tables) => tables.push(table),
225                        Err(error) => {
226                            warn!(error = %error, "failed to record changeset constraint conflict; aborting apply");
227                            return ConflictAction::SQLITE_CHANGESET_ABORT;
228                        }
229                    }
230                    return ConflictAction::SQLITE_CHANGESET_OMIT;
231                }
232                if conflict_type == ConflictType::SQLITE_CHANGESET_DATA
233                    && op_code == Action::SQLITE_UPDATE
234                {
235                    match update_pk_key(&item, &table).map(|pk| {
236                        premerged_updates.contains(&RowKey {
237                            table: table.clone(),
238                            pk,
239                        })
240                    }) {
241                        Ok(true) => return ConflictAction::SQLITE_CHANGESET_OMIT,
242                        Ok(false) => {}
243                        Err(error) => {
244                            warn!(table, error = %error, "failed to read premerged UPDATE primary key; aborting apply");
245                            return ConflictAction::SQLITE_CHANGESET_ABORT;
246                        }
247                    }
248                }
249                arbitrate_row_conflict(
250                    conflict_type,
251                    item,
252                    &table,
253                    &closure_schema,
254                    timestamp_policy,
255                )
256            },
257        )
258        .map_err(DbError::from)?;
259        #[cfg(any(test, feature = "test-utils"))]
260        let had_fk_violations = fk_flag.load(Ordering::Relaxed);
261        let constraint_conflict_tables = constraint_conflict_tables
262            .lock()
263            .map_err(|_| {
264                DbError::Message("constraint conflict table collection is poisoned".to_string())
265            })?
266            .clone();
267        #[cfg(any(test, feature = "test-utils"))]
268        let winning_rows = resolve_winning_rows(conn, &schema, incoming_rows)?;
269
270        Ok(ApplyResult {
271            #[cfg(any(test, feature = "test-utils"))]
272            had_fk_violations,
273            constraint_conflict_tables,
274            #[cfg(any(test, feature = "test-utils"))]
275            winning_rows,
276        })
277    }
278
279    pub(crate) fn current_winning_rows<B: AsRef<[u8]>>(
280        &self,
281        schema: &TableSchema,
282        changeset: B,
283    ) -> Result<Vec<WinningRow>, DbError> {
284        resolve_winning_rows(
285            self.store.transaction,
286            schema,
287            incoming_rows(changeset.as_ref(), schema)?,
288        )
289    }
290
291    pub(crate) fn apply_changeset_strict<B: AsRef<[u8]>>(
292        &self,
293        changeset: ValidatedChangeset<B>,
294        blob_decls: &crate::BlobDecls,
295    ) -> Result<(), DbError> {
296        let bytes = changeset.bytes();
297        let old_changes = crate::walk_old_changeset(bytes).map_err(DbError::Changeset)?;
298        let new_changes = crate::walk_changeset(bytes).map_err(DbError::Changeset)?;
299        let old_exact_bindings = super::exact_blob_bindings_on(self.store.transaction)?;
300        let obsolete = crate::local_blob_cleanup_intents::intents_from_changes(
301            blob_decls,
302            &old_changes,
303            &new_changes,
304        )?;
305        self.store
306            .transaction
307            .apply_strm(
308                &mut &bytes[..],
309                None::<fn(&str) -> bool>,
310                |_conflict_type, _item| ConflictAction::SQLITE_CHANGESET_ABORT,
311            )
312            .map_err(DbError::from)?;
313        for intent in obsolete {
314            super::record_obsolete_copy_intents_from_bindings_on(
315                self.store.transaction,
316                blob_decls,
317                &intent,
318                &old_exact_bindings,
319            )?;
320        }
321        Ok(())
322    }
323}
324
325fn incoming_rows(bytes: &[u8], schema: &TableSchema) -> Result<Vec<IncomingRow>, DbError> {
326    if bytes.is_empty() {
327        return Ok(Vec::new());
328    }
329    let input: &mut dyn std::io::Read = &mut &bytes[..];
330    let mut iter = ChangesetIter::start_strm(&input).map_err(DbError::from)?;
331    let mut rows = Vec::new();
332    while let Some(item) = iter.next().map_err(DbError::from)? {
333        let op = item.op().map_err(DbError::from)?;
334        let table = op.table_name();
335        let updated_at = schema.updated_at(table).ok_or_else(|| {
336            DbError::Message(format!("changeset contains undeclared table {table:?}"))
337        })?;
338        let (id_side, stamp_side) = match op.code() {
339            Action::SQLITE_INSERT => (UpdateValue::New, Some(UpdateValue::New)),
340            Action::SQLITE_UPDATE => (UpdateValue::Old, Some(UpdateValue::New)),
341            Action::SQLITE_DELETE => (UpdateValue::Old, None),
342            code => {
343                return Err(DbError::Message(format!(
344                    "changeset for {table:?} contains unsupported operation {code:?}"
345                )))
346            }
347        };
348        let row_id = required_text_changeset_value(item, table, 0, id_side, "row id")?;
349        let row_stamp = stamp_side
350            .map(|side| required_text_changeset_value(item, table, updated_at, side, "row stamp"))
351            .transpose()?;
352        rows.push(IncomingRow {
353            table: table.to_string(),
354            row_id,
355            row_stamp,
356        });
357    }
358    Ok(rows)
359}
360
361fn required_text_changeset_value(
362    item: &ChangesetItem,
363    table: &str,
364    column: usize,
365    side: UpdateValue,
366    field: &str,
367) -> Result<String, DbError> {
368    let value = changeset_value(item, column, side)?.ok_or_else(|| {
369        DbError::Message(format!("changeset for {table:?} has no {side:?} {field}"))
370    })?;
371    let Value::Text(value) = value else {
372        return Err(DbError::Message(format!(
373            "changeset for {table:?} has non-TEXT {side:?} {field}"
374        )));
375    };
376    Ok(value)
377}
378
379fn resolve_winning_rows(
380    conn: &Connection,
381    schema: &TableSchema,
382    incoming: Vec<IncomingRow>,
383) -> Result<Vec<WinningRow>, DbError> {
384    let mut winners = Vec::new();
385    for row in incoming {
386        let columns = schema.columns(&row.table).ok_or_else(|| {
387            DbError::Message(format!("synced table {:?} has no column map", row.table))
388        })?;
389        let updated_at = schema.updated_at(&row.table).ok_or_else(|| {
390            DbError::Message(format!(
391                "synced table {:?} has no _updated_at column index",
392                row.table
393            ))
394        })?;
395        let sql = format!(
396            "SELECT {} FROM {} WHERE {} = ?1",
397            quote_ident(&columns[updated_at]),
398            quote_ident(&row.table),
399            quote_ident(&columns[0])
400        );
401        let live_stamp = conn
402            .query_row(&sql, [&row.row_id], |result| result.get::<_, String>(0))
403            .optional()
404            .map_err(DbError::from)?;
405        let incoming_won = match (&row.row_stamp, &live_stamp) {
406            (None, None) => true,
407            (Some(expected), Some(actual)) => expected == actual,
408            _ => false,
409        };
410        if incoming_won {
411            winners.push(WinningRow {
412                table: row.table,
413                row_id: row.row_id,
414                row_stamp: row.row_stamp,
415            });
416        }
417    }
418    Ok(winners)
419}
420
421#[derive(Clone, Debug, Eq, PartialEq, Hash)]
422struct RowKey {
423    table: String,
424    pk: String,
425}
426
427struct ChangedColumn {
428    index: usize,
429    base: Value,
430    incoming: Value,
431}
432
433struct IncomingUpdate {
434    table: String,
435    pk: String,
436    changed_columns: Vec<ChangedColumn>,
437    incoming_updated_at: Timestamp,
438}
439
440fn premerge_losing_update_columns(
441    conn: &Connection,
442    bytes: &[u8],
443    schema: &TableSchema,
444    timestamp_policy: IncomingTimestampPolicy,
445) -> Result<HashSet<RowKey>, DbError> {
446    if bytes.is_empty() {
447        return Ok(HashSet::new());
448    }
449
450    let input: &mut dyn std::io::Read = &mut &bytes[..];
451    let mut iter = ChangesetIter::start_strm(&input).map_err(DbError::from)?;
452    let mut handled = HashSet::new();
453
454    while let Some(item) = iter.next().map_err(DbError::from)? {
455        let Some(update) = incoming_update(item, schema)? else {
456            continue;
457        };
458        if merge_losing_update(conn, schema, &update, timestamp_policy)? {
459            handled.insert(RowKey {
460                table: update.table,
461                pk: update.pk,
462            });
463        }
464    }
465
466    Ok(handled)
467}
468
469fn incoming_update(
470    item: &ChangesetItem,
471    schema: &TableSchema,
472) -> Result<Option<IncomingUpdate>, DbError> {
473    let op = item.op().map_err(DbError::from)?;
474    if op.code() != Action::SQLITE_UPDATE {
475        return Ok(None);
476    }
477
478    let table = op.table_name();
479    let Some(updated_at) = schema.updated_at(table) else {
480        warn!(
481            table,
482            "UPDATE changeset table is not in the local synced schema"
483        );
484        return Ok(None);
485    };
486
487    let Some(incoming_updated_at_value) = changeset_value(item, updated_at, UpdateValue::New)?
488    else {
489        warn!(table, "UPDATE changeset has no incoming _updated_at value");
490        return Ok(None);
491    };
492    let Some(incoming_updated_at) = timestamp_from_value(&incoming_updated_at_value) else {
493        warn!(
494            table,
495            "UPDATE changeset has an incoming _updated_at value that does not parse"
496        );
497        return Ok(None);
498    };
499
500    let pk = update_pk_key(item, table)?;
501
502    let mut changed_columns = Vec::new();
503    for index in 0..op.number_of_columns() as usize {
504        if index == 0 || index == updated_at {
505            continue;
506        }
507        let base = changeset_value(item, index, UpdateValue::Old)?;
508        let incoming = changeset_value(item, index, UpdateValue::New)?;
509        match (base, incoming) {
510            (Some(base), Some(incoming)) => changed_columns.push(ChangedColumn {
511                index,
512                base,
513                incoming,
514            }),
515            (None, None) => {}
516            _ => {
517                return Err(DbError::Message(format!(
518                    "UPDATE changeset for {table} has only one side for column {index}"
519                )));
520            }
521        }
522    }
523
524    Ok(Some(IncomingUpdate {
525        table: table.to_string(),
526        pk,
527        changed_columns,
528        incoming_updated_at,
529    }))
530}
531
532fn merge_losing_update(
533    conn: &Connection,
534    schema: &TableSchema,
535    update: &IncomingUpdate,
536    timestamp_policy: IncomingTimestampPolicy,
537) -> Result<bool, DbError> {
538    let columns = schema.columns(&update.table).ok_or_else(|| {
539        DbError::Message(format!("synced table {} has no column map", update.table))
540    })?;
541    let updated_at_index = schema.updated_at(&update.table).ok_or_else(|| {
542        DbError::Message(format!(
543            "synced table {} has no _updated_at column index",
544            update.table
545        ))
546    })?;
547    if update
548        .changed_columns
549        .iter()
550        .any(|c| c.index >= columns.len())
551        || updated_at_index >= columns.len()
552    {
553        return Err(DbError::Message(format!(
554            "UPDATE changeset for {} names a column outside the local schema",
555            update.table
556        )));
557    }
558
559    let mut selected_indices = update
560        .changed_columns
561        .iter()
562        .map(|column| column.index)
563        .collect::<Vec<_>>();
564    selected_indices.push(updated_at_index);
565    let select_columns = selected_indices
566        .iter()
567        .map(|index| quote_ident(&columns[*index]))
568        .collect::<Vec<_>>()
569        .join(", ");
570    let sql = format!(
571        "SELECT {select_columns} FROM {} WHERE {} = ?1",
572        quote_ident(&update.table),
573        quote_ident(&columns[0])
574    );
575    let local_values = conn
576        .query_row(&sql, rusqlite::params![&update.pk], |row| {
577            (0..selected_indices.len())
578                .map(|index| row.get::<_, Value>(index))
579                .collect::<rusqlite::Result<Vec<_>>>()
580        })
581        .optional()
582        .map_err(DbError::from)?;
583    let Some(local_values) = local_values else {
584        return Ok(false);
585    };
586
587    let local_updated_at = local_values
588        .last()
589        .and_then(timestamp_from_value)
590        .ok_or_else(|| {
591            DbError::Message(format!(
592                "local row in {} has no parseable _updated_at",
593                update.table
594            ))
595        })?;
596    match compare_lww_stamps(
597        &update.table,
598        update.incoming_updated_at.clone(),
599        local_updated_at,
600        timestamp_policy,
601    ) {
602        LwwComparison::IncomingWins | LwwComparison::IncomingGrossFuture => return Ok(false),
603        LwwComparison::LocalWins => {}
604    }
605
606    let mut applied = Vec::new();
607    for (column, local_value) in update.changed_columns.iter().zip(local_values.iter()) {
608        if *local_value == column.base {
609            applied.push(column);
610        }
611    }
612    if !applied.is_empty() {
613        // Fold the losing writer's columns into the local row WITHOUT bumping its
614        // `_updated_at`. The local row won row arbitration, so its stamp already
615        // dominates the incoming one; re-stamping the merged row could only lower
616        // it, and the row winner's stamp is what future arbitration must compare
617        // against. The merge changes a losing column's value, not the row's clock.
618        let assignments = applied
619            .iter()
620            .enumerate()
621            .map(|(offset, column)| {
622                format!("{} = ?{}", quote_ident(&columns[column.index]), offset + 1)
623            })
624            .collect::<Vec<_>>()
625            .join(", ");
626        let sql = format!(
627            "UPDATE {} SET {assignments} WHERE {} = ?{}",
628            quote_ident(&update.table),
629            quote_ident(&columns[0]),
630            applied.len() + 1
631        );
632        let mut params: Vec<&dyn ToSql> = applied
633            .iter()
634            .map(|column| &column.incoming as &dyn ToSql)
635            .collect();
636        params.push(&update.pk);
637        conn.execute(&sql, params_from_iter(params))
638            .map_err(DbError::from)?;
639    }
640
641    Ok(true)
642}
643
644fn changeset_value(
645    item: &ChangesetItem,
646    column: usize,
647    side: UpdateValue,
648) -> Result<Option<Value>, DbError> {
649    let value = match side {
650        UpdateValue::Old => item.old_value(column),
651        UpdateValue::New => item.new_value(column),
652    };
653    match value {
654        Ok(value) => Value::try_from(value).map(Some).map_err(|error| {
655            DbError::context(
656                format!("changeset {side:?} value conversion failed for column {column}"),
657                error,
658            )
659        }),
660        Err(rusqlite::Error::InvalidColumnIndex(_)) => Ok(None),
661        Err(error) => Err(DbError::context(
662            format!("changeset {side:?} value read failed for column {column}"),
663            error,
664        )),
665    }
666}
667
668fn update_pk_key(item: &ChangesetItem, table: &str) -> Result<String, DbError> {
669    match item.old_value(0) {
670        Ok(value) => text_id_from_value_ref(table, value),
671        Err(rusqlite::Error::InvalidColumnIndex(_)) => Err(DbError::Message(format!(
672            "UPDATE changeset for {table} has no old-side primary key"
673        ))),
674        Err(error) => Err(DbError::context(
675            format!("UPDATE changeset for {table} primary key read failed"),
676            error,
677        )),
678    }
679}
680
681fn text_id_from_value_ref(table: &str, value: ValueRef<'_>) -> Result<String, DbError> {
682    let ValueRef::Text(bytes) = value else {
683        return Err(DbError::Message(format!(
684            "UPDATE changeset for {table} primary key is not TEXT"
685        )));
686    };
687    std::str::from_utf8(bytes)
688        .map(str::to_owned)
689        .map_err(|error| {
690            DbError::context(
691                format!("UPDATE changeset for {table} primary key is not UTF-8"),
692                error,
693            )
694        })
695}
696
697fn timestamp_from_value(value: &Value) -> Option<Timestamp> {
698    value_ref_to_string(ValueRef::from(value)).and_then(|s| Timestamp::parse(&s))
699}
coven_database/store/store_session/merge_materialization_transaction/changeset_application.rs

coven_database/store/store_session/merge_materialization_transaction/
changeset_application.rs
