rindle/
changes.rs

1//! Owned, fully-materialized change events off the dataflow pipeline.
2//!
3//! These types ([`CaughtChange`] / [`CaughtNode`]) and the [`expand_change`] /
4//! [`expand_node`] materializers were originally part of the spec-`11` test oracle
5//! (`crate::testkit`). They are **graduated into the production crate** here because
6//! they are the canonical *owned* form of a downstream [`Change`](crate::change::Change):
7//! the production change-stream sink ([`Graph::add_change_sink`](crate::graph::Graph))
8//! and out-of-crate consumers (the `rindle-replica` live-query wrapper) need them outside
9//! `cfg(test)`/`testkit` builds. `testkit` re-exports them, so existing oracle code and
10//! `rindle::testkit::*` paths are unchanged.
11//!
12//! Why owned: a dataflow [`Change<'g>`](crate::change::Change) borrows the graph (its
13//! relationship thunks are `+ 'g`), so it cannot survive past the push that produced it.
14//! [`expand_change`] drains every lazy relationship thunk eagerly into an owned tree,
15//! which *can* cross a callback / thread / process boundary.
16
17use std::cmp::Ordering;
18use std::collections::BTreeMap;
19
20use crate::change::{Change, Node};
21use crate::value::{compare_values, OwnedRow, RelId};
22
23// ---------------------------------------------------------------------------
24// The comparison tree: CaughtNode / CaughtChange (catch.ts `CaughtNode` /
25// `CaughtChange`, with the `'yield'` variant gone — Primitive #6)
26// ---------------------------------------------------------------------------
27
28/// A fully-materialized node: the row cells plus **eagerly drained**
29/// relationships. The comparison unit for a fetch/push assertion, and the owned
30/// node a [`Graph::add_change_sink`](crate::graph::Graph) consumer receives.
31///
32/// Relationships are keyed by their resolved [`RelId`] slot in a `BTreeMap`, so
33/// the diff is slot-stable regardless of insertion order; the child `Vec` within
34/// a relationship preserves the **operator's sort order** (the order *is* part of
35/// the contract — never re-sort it, `11` §3.8).
36#[derive(Clone, Debug)]
37pub struct CaughtNode {
38    pub row: OwnedRow,
39    pub relationships: BTreeMap<RelId, Vec<CaughtNode>>,
40}
41
42/// A caught downstream change. Mirrors `catch.ts` `expandChange` output: an
43/// `Edit` carries only the two rows (no node, `catch.ts:104-109`); a `Child`
44/// carries the parent row, the relationship slot, and the nested change
45/// (`catch.ts:110-118`).
46#[derive(Clone, Debug)]
47pub enum CaughtChange {
48    Add(CaughtNode),
49    Remove(CaughtNode),
50    Edit {
51        old: OwnedRow,
52        row: OwnedRow,
53    },
54    Child {
55        row: OwnedRow,
56        rel: RelId,
57        change: Box<CaughtChange>,
58    },
59}
60
61impl CaughtChange {
62    /// The **root** row this change is routed by: the node's row for `Add`/`Remove`, the
63    /// new row for `Edit` (a partition-key change never reaches a consumer as an `Edit` —
64    /// the source splits it — so `old` agrees on every routing column), and the
65    /// **parent** row for `Child` (the outermost `Child` of a nested change carries the
66    /// top-level row). This is what a parameterized query family's consumer demuxes
67    /// partitions by (design 310 §4.2 / impl plan D8): every change is routable by a
68    /// column read of this row.
69    pub fn root_row(&self) -> &OwnedRow {
70        match self {
71            CaughtChange::Add(n) | CaughtChange::Remove(n) => &n.row,
72            CaughtChange::Edit { row, .. } => row,
73            CaughtChange::Child { row, .. } => row,
74        }
75    }
76}
77
78/// `expandNode` (`catch.ts:124-136`): clone the row and **drain every
79/// relationship `NodeStream` thunk eagerly**, recursing. This is the only place a
80/// consumer fully forces the lazy thunks — exactly what reveals overlay/position
81/// bugs. No graph handle is needed: each thunk already owns its `&'g Graph`
82/// (Primitive #5, captured by value at `Node` construction).
83pub fn expand_node(node: &Node<'_>) -> CaughtNode {
84    let mut relationships: BTreeMap<RelId, Vec<CaughtNode>> = BTreeMap::new();
85    for r in &node.rels {
86        let children: Vec<CaughtNode> = (r.thunk)().map(|c| expand_node(&c)).collect();
87        // A node never carries two relationships for the same slot; if it somehow
88        // did, the later one wins (matching JS object-key overwrite).
89        relationships.insert(r.slot, children);
90    }
91    CaughtNode {
92        row: node.row.clone(),
93        relationships,
94    }
95}
96
97/// `expandChange` (`catch.ts:92-122`).
98pub fn expand_change(change: &Change<'_>) -> CaughtChange {
99    match change {
100        Change::Add(n) => CaughtChange::Add(expand_node(n)),
101        Change::Remove(n) => CaughtChange::Remove(expand_node(n)),
102        Change::Edit { node, old } => CaughtChange::Edit {
103            old: old.row.clone(),
104            row: node.row.clone(),
105        },
106        Change::Child { node, rel, child } => CaughtChange::Child {
107            row: node.row.clone(),
108            rel: *rel,
109            change: Box::new(expand_change(child)),
110        },
111    }
112}
113
114// ---------------------------------------------------------------------------
115// The diff predicate (§3.8): structural equality with full-cell content
116// comparison. `OwnedValue` has no derived `PartialEq` (it would invite the wrong
117// comparator), so we spell row equality out via `compare_values` (null == null,
118// floats via total_cmp — `11` §3.8).
119// ---------------------------------------------------------------------------
120
121#[inline]
122fn row_eq(a: &OwnedRow, b: &OwnedRow) -> bool {
123    a.len() == b.len()
124        && a.cells()
125            .zip(b.cells())
126            .all(|(x, y)| compare_values(x, y) == Ordering::Equal)
127}
128
129impl PartialEq for CaughtNode {
130    fn eq(&self, o: &Self) -> bool {
131        // `BTreeMap`/`Vec<CaughtNode>` equality recurses into this impl; the only
132        // hand-written part is the row (no `OwnedValue: Eq`).
133        row_eq(&self.row, &o.row) && self.relationships == o.relationships
134    }
135}
136impl Eq for CaughtNode {}
137
138impl PartialEq for CaughtChange {
139    fn eq(&self, o: &Self) -> bool {
140        use CaughtChange::*;
141        match (self, o) {
142            (Add(a), Add(b)) | (Remove(a), Remove(b)) => a == b,
143            (Edit { old: o1, row: r1 }, Edit { old: o2, row: r2 }) => {
144                row_eq(o1, o2) && row_eq(r1, r2)
145            }
146            (
147                Child {
148                    row: r1,
149                    rel: s1,
150                    change: c1,
151                },
152                Child {
153                    row: r2,
154                    rel: s2,
155                    change: c2,
156                },
157            ) => row_eq(r1, r2) && s1 == s2 && c1 == c2,
158            _ => false,
159        }
160    }
161}
162impl Eq for CaughtChange {}
rindle/changes.rs

rindle/
changes.rs
