rindle/
flat.rs

1//! Flat change events for cross-process / cross-language view reconstruction.
2//!
3//! See `FLAT-CHANGES-DESIGN.md` for the full wire format **and** the receiver
4//! `apply_change` contract. This module is the **sender-side** transform: it
5//! linearizes the nested owned [`CaughtChange`] tree (the form emitted by
6//! [`Graph::add_change_sink`](crate::graph::Graph)) into a flat list of
7//! [`FlatChange`]s, each carrying the root→leaf path to its mutation point plus the
8//! op (Add / Remove / Edit).
9//!
10//! The nesting in [`CaughtChange::Child`] **is** the path; [`flatten`] peels it into
11//! [`PathSeg`]s. Each segment carries the parent's **full row** — the sort-key
12//! locator *and* PK identity — NOT just the PK, because the View locates every parent
13//! by `binary_search` over the level's `Sort`, whose leading columns are the
14//! `orderBy`, not the PK (`FLAT-CHANGES-DESIGN.md` §5.3). That full row is already in
15//! `CaughtChange::Child.row`, so the linearization is lossless and free.
16//!
17//! Serde derives are gated behind the `serde` / `testkit` feature (the same
18//! `cfg_attr` pattern as [`crate::ast::Ast`]); the cell encoding is the externally
19//! tagged, faithful [`OwnedValue`](crate::value::OwnedValue) form (§5.1).
20
21use crate::changes::{CaughtChange, CaughtNode};
22use crate::value::{OwnedValue, RelId};
23
24/// A row on the wire: positional cells, aligned to the level's `Schema.columns`.
25pub type WireRow = Vec<OwnedValue>;
26
27/// A materialized node on the wire: its row plus slot-keyed, pre-sorted child
28/// subtrees — the wire-shaped twin of [`CaughtNode`] (`Vec` rows; relationships as a
29/// sorted `Vec<(slot, children)>` rather than a `BTreeMap`).
30///
31/// Children within a slot are in the **operator's sort order** and MUST NOT be
32/// re-sorted by the receiver (`FLAT-CHANGES-DESIGN.md` §7.5).
33#[cfg_attr(
34    any(feature = "testkit", feature = "serde"),
35    derive(serde::Serialize, serde::Deserialize)
36)]
37#[derive(Clone, Debug)]
38pub struct WireNode {
39    pub row: WireRow,
40    pub rels: Vec<(RelId, Vec<WireNode>)>,
41}
42
43/// One hop of a [`FlatChange`] path: descend relationship `rel`, locating the parent
44/// at the current level by `parent_row` (the full sort-key locator).
45#[cfg_attr(
46    any(feature = "testkit", feature = "serde"),
47    derive(serde::Serialize, serde::Deserialize)
48)]
49#[derive(Clone, Debug)]
50pub struct PathSeg {
51    pub rel: RelId,
52    pub parent_row: WireRow,
53}
54
55/// The op at the end of a flat change's path (the reached level).
56#[cfg_attr(
57    any(feature = "testkit", feature = "serde"),
58    derive(serde::Serialize, serde::Deserialize)
59)]
60#[derive(Clone, Debug)]
61pub enum FlatOp {
62    /// Insert `node` (with its inline, slot-keyed, pre-sorted subtree).
63    Add(WireNode),
64    /// Remove the row at this level. **Row-only** — the View's remove path consumes
65    /// only `node.row` (the drained subtree the nested [`CaughtChange::Remove`] carries
66    /// is intentionally dropped).
67    Remove { row: WireRow },
68    /// Edit: `old` **locates**, `new` **places** (both required, at every depth — the
69    /// receiver needs `old` to find the entry even when the sort key moved).
70    Edit { old: WireRow, new: WireRow },
71}
72
73/// One flat change: the root→leaf parent path, then the op applied at the reached
74/// level. An empty `path` means the op applies at the top level.
75#[cfg_attr(
76    any(feature = "testkit", feature = "serde"),
77    derive(serde::Serialize, serde::Deserialize)
78)]
79#[derive(Clone, Debug)]
80pub struct FlatChange {
81    pub path: Vec<PathSeg>,
82    pub op: FlatOp,
83}
84
85/// Linearize one [`CaughtChange`] into a [`FlatChange`]: peel each
86/// [`CaughtChange::Child`] into a [`PathSeg`] until a non-`Child` op is reached.
87pub fn flatten(change: &CaughtChange) -> FlatChange {
88    let mut path = Vec::new();
89    let mut cur = change;
90    loop {
91        match cur {
92            CaughtChange::Child { row, rel, change } => {
93                path.push(PathSeg {
94                    rel: *rel,
95                    parent_row: row.to_value_vec(),
96                });
97                cur = &**change;
98            }
99            CaughtChange::Add(node) => {
100                return FlatChange {
101                    path,
102                    op: FlatOp::Add(wire_node(node)),
103                };
104            }
105            CaughtChange::Remove(node) => {
106                return FlatChange {
107                    path,
108                    op: FlatOp::Remove {
109                        row: node.row.to_value_vec(),
110                    },
111                };
112            }
113            CaughtChange::Edit { old, row } => {
114                return FlatChange {
115                    path,
116                    op: FlatOp::Edit {
117                        old: old.to_value_vec(),
118                        new: row.to_value_vec(),
119                    },
120                };
121            }
122        }
123    }
124}
125
126/// Linearize a batch of [`CaughtChange`]s (e.g. one transaction's drained sink),
127/// preserving order. **The order is significant** — the receiver must apply the
128/// resulting flat changes in exactly this order (`FLAT-CHANGES-DESIGN.md` §5.4).
129pub fn flatten_all(changes: &[CaughtChange]) -> Vec<FlatChange> {
130    changes.iter().map(flatten).collect()
131}
132
133/// Convert a [`CaughtNode`] into its wire twin. The `BTreeMap<RelId, _>` iterates in
134/// ascending slot order (deterministic); the child `Vec` preserves the operator's
135/// sort order (never re-sorted).
136fn wire_node(node: &CaughtNode) -> WireNode {
137    WireNode {
138        row: node.row.to_value_vec(),
139        rels: node
140            .relationships
141            .iter()
142            .map(|(slot, children)| (*slot, children.iter().map(wire_node).collect()))
143            .collect(),
144    }
145}
146
147#[cfg(test)]
148mod tests {
149    use super::*;
150    use crate::changes::{CaughtChange, CaughtNode};
151    use crate::value::{owned_row, OwnedValue as V};
152
153    fn cnode(row: Vec<V>, rels: Vec<(RelId, Vec<CaughtNode>)>) -> CaughtNode {
154        CaughtNode {
155            row: owned_row(row),
156            relationships: rels.into_iter().collect(),
157        }
158    }
159
160    fn as_int(v: &V) -> i64 {
161        match v {
162            V::Int(i) => *i,
163            other => panic!("expected Int, got {other:?}"),
164        }
165    }
166
167    #[test]
168    fn flatten_peels_child_chain_into_path() {
169        // Child(parentA) -> Child(parentB) -> Edit(leaf): the nesting linearizes into
170        // a 2-hop path with the leaf Edit as the terminal op.
171        let change = CaughtChange::Child {
172            row: owned_row(vec![V::Int(1)]),
173            rel: RelId(0),
174            change: Box::new(CaughtChange::Child {
175                row: owned_row(vec![V::Int(2)]),
176                rel: RelId(1),
177                change: Box::new(CaughtChange::Edit {
178                    old: owned_row(vec![V::Int(3), V::Int(10)]),
179                    row: owned_row(vec![V::Int(3), V::Int(20)]),
180                }),
181            }),
182        };
183        let flat = flatten(&change);
184        assert_eq!(flat.path.len(), 2);
185        assert_eq!(flat.path[0].rel, RelId(0));
186        assert_eq!(as_int(&flat.path[0].parent_row[0]), 1);
187        assert_eq!(flat.path[1].rel, RelId(1));
188        assert_eq!(as_int(&flat.path[1].parent_row[0]), 2);
189        match &flat.op {
190            FlatOp::Edit { old, new } => {
191                assert_eq!(as_int(&old[1]), 10);
192                assert_eq!(as_int(&new[1]), 20);
193            }
194            other => panic!("expected Edit, got {other:?}"),
195        }
196    }
197
198    #[test]
199    fn flatten_add_carries_slot_keyed_subtree() {
200        // Add( parent { rel0: [childX] } ): the inline subtree survives, slot-keyed.
201        let node = cnode(
202            vec![V::Int(1)],
203            vec![(RelId(0), vec![cnode(vec![V::Int(100)], vec![])])],
204        );
205        let flat = flatten(&CaughtChange::Add(node));
206        assert!(flat.path.is_empty());
207        match &flat.op {
208            FlatOp::Add(wn) => {
209                assert_eq!(as_int(&wn.row[0]), 1);
210                assert_eq!(wn.rels.len(), 1);
211                assert_eq!(wn.rels[0].0, RelId(0));
212                assert_eq!(wn.rels[0].1.len(), 1);
213                assert_eq!(as_int(&wn.rels[0].1[0].row[0]), 100);
214            }
215            other => panic!("expected Add, got {other:?}"),
216        }
217    }
218
219    #[test]
220    fn flatten_remove_is_row_only() {
221        // The drained subtree on a nested Remove is dropped; only the row crosses.
222        let node = cnode(
223            vec![V::Int(7)],
224            vec![(RelId(0), vec![cnode(vec![V::Int(1)], vec![])])],
225        );
226        let flat = flatten(&CaughtChange::Remove(node));
227        match &flat.op {
228            FlatOp::Remove { row } => assert_eq!(as_int(&row[0]), 7),
229            other => panic!("expected Remove, got {other:?}"),
230        }
231    }
232
233    #[cfg(any(feature = "testkit", feature = "serde"))]
234    #[test]
235    fn flat_change_serde_round_trips() {
236        // A Child→Add carrying every cell variant must survive serialize→deserialize
237        // byte-identically (the tagged encoding preserves Int/Float/Str/Json/Null/Bool).
238        let change = CaughtChange::Child {
239            row: owned_row(vec![V::Int(1), V::str("a")]),
240            rel: RelId(0),
241            change: Box::new(CaughtChange::Add(cnode(
242                vec![
243                    V::Int(2),
244                    V::Float(1.5),
245                    V::Null,
246                    V::Bool(true),
247                    V::str("hi"),
248                    V::Json(std::sync::Arc::from("{\"k\":1}")),
249                ],
250                vec![(RelId(0), vec![cnode(vec![V::Int(3)], vec![])])],
251            ))),
252        };
253        let flat = flatten(&change);
254        let json = serde_json::to_value(&flat).expect("serialize");
255        let back: FlatChange = serde_json::from_value(json.clone()).expect("deserialize");
256        let json2 = serde_json::to_value(&back).expect("re-serialize");
257        assert_eq!(json, json2, "flat change must round-trip through serde");
258    }
259}
rindle/flat.rs

rindle/
flat.rs
