rindle/
flat_receiver.rs

1//! A **reference receiver**: a host-language-agnostic port of the View's
2//! `apply_change` (`src/view.rs`) fed by the flat change stream ([`FlatChange`]),
3//! reconstructing the exact `ArrayView` tree. It is the executable form of the
4//! receiver contract in `FLAT-CHANGES-DESIGN.md` §6, and doubles as the conformance
5//! oracle (`tests/flat_oracle.rs` differentially checks it against the real View).
6//!
7//! Per `FLAT-CHANGES-DESIGN.md` §3 the receiver deliberately **drops** all the
8//! sender-only machinery — `Arc` reference stability, copy-on-write, `TxnGen`,
9//! `EntryId`, singular/`Format` unwrapping. It keeps only the *logical* operations:
10//! `binary_search`-locate over the level's `Sort`, `rc` multi-path counting, and the
11//! edit-as-move ghost/adjusted-position/merge protocol. A receiver rebuilding its own
12//! copy mutates freely; if it ever wants reference stability for its own UI it can
13//! re-introduce COW exactly as the sender does.
14//!
15//! The comparator is the engine's own [`compare_rows`]/[`compare_values`] — this is a
16//! Rust reference impl, so it holds the comparator fixed and tests the
17//! *flatten + apply* algorithm. A cross-language receiver instead reimplements
18//! `compare_values` byte-for-byte (`FLAT-CHANGES-DESIGN.md` §4).
19
20use std::cmp::Ordering;
21
22use crate::flat::{FlatChange, FlatOp, PathSeg, WireNode, WireRow};
23use crate::value::{compare_rows, owned_row, OwnedRow, Schema, Sort};
24
25/// A reconstructed node: row + multi-path refcount + per-slot child lists.
26///
27/// `rels` is indexed by [`RelId`](crate::value::RelId) slot, in the same order as the
28/// level's `Schema::relationships` (one list per declared slot; out-of-view / gating
29/// slots stay empty — mirroring the sender's [`Entry`](crate::view::Entry)).
30#[derive(Clone, Debug)]
31pub struct RecvNode {
32    pub row: OwnedRow,
33    pub rc: u32,
34    pub rels: Vec<Vec<RecvNode>>,
35}
36
37/// The reference receiver: holds the hierarchical view [`Schema`] (the shipped query
38/// schema) and the reconstructed top-level list (`root[""]`).
39#[derive(Debug)]
40pub struct Receiver {
41    schema: Schema,
42    top: Vec<RecvNode>,
43}
44
45impl Receiver {
46    /// A fresh receiver over the given hierarchical view schema (the same `Schema` the
47    /// sender's View was built with — its per-level `sort` is the comparator input and
48    /// `rel_child(slot)` is the in-view gate).
49    pub fn new(schema: Schema) -> Receiver {
50        Receiver {
51            schema,
52            top: Vec::new(),
53        }
54    }
55
56    /// Apply one [`FlatChange`]: descend its path, then fold the op at the reached
57    /// level. Panics on an inconsistent stream (mirrors the differ's "node does not
58    /// exist" — `view.rs:450`).
59    pub fn apply(&mut self, change: &FlatChange) {
60        apply_at(&mut self.top, &self.schema, &change.path, &change.op);
61    }
62
63    /// Apply a batch in order (the per-transaction delta, or the hydrate snapshot).
64    /// Order is significant — see `FLAT-CHANGES-DESIGN.md` §5.4.
65    pub fn apply_all(&mut self, changes: &[FlatChange]) {
66        for c in changes {
67            self.apply(c);
68        }
69    }
70
71    /// The reconstructed top-level result (`root[""]`).
72    pub fn top(&self) -> &[RecvNode] {
73        &self.top
74    }
75
76    /// The hierarchical view schema this receiver reconstructs against.
77    pub fn schema(&self) -> &Schema {
78        &self.schema
79    }
80}
81
82#[inline]
83fn binary_search(list: &[RecvNode], row: &OwnedRow, sort: &Sort) -> Result<usize, usize> {
84    list.binary_search_by(|n| compare_rows(sort, &n.row, row))
85}
86
87/// Descend `path` from `list` (whose entries have schema `schema`), then apply `op` at
88/// the reached level. Mirrors the `apply` recursion in `view.rs`, but operates on a
89/// plain `&mut Vec<RecvNode>` (no COW/`Arc`).
90fn apply_at(list: &mut Vec<RecvNode>, schema: &Schema, path: &[PathSeg], op: &FlatOp) {
91    let Some((seg, rest)) = path.split_first() else {
92        return apply_op(list, schema, op);
93    };
94    // In-view gate (`view.rs:486`): a slot with no child schema is join-only / gating
95    // (a non-flipped EXISTS forwards a Child on its own out-of-view slot) — drop it.
96    let Some(child_schema) = schema.rel_child(seg.rel) else {
97        return;
98    };
99    // Locate the parent by its FULL row (the sort-key locator) — never by PK.
100    let pr = owned_row(seg.parent_row.clone());
101    let pos = match binary_search(list, &pr, &schema.sort) {
102        Ok(p) => p,
103        Err(_) => panic!("flat receiver: parent not found at path hop (inconsistent stream)"),
104    };
105    apply_at(&mut list[pos].rels[seg.rel.ix()], child_schema, rest, op);
106}
107
108fn apply_op(list: &mut Vec<RecvNode>, schema: &Schema, op: &FlatOp) {
109    match op {
110        FlatOp::Add(node) => apply_add(list, schema, node),
111        FlatOp::Remove { row } => apply_remove(list, schema, row),
112        FlatOp::Edit { old, new } => apply_edit(list, schema, old, new),
113    }
114}
115
116/// `apply_add` (`view.rs:412`): duplicate sort-key ⇒ `rc += 1`; else build + insert.
117fn apply_add(list: &mut Vec<RecvNode>, schema: &Schema, node: &WireNode) {
118    let row = owned_row(node.row.clone());
119    match binary_search(list, &row, &schema.sort) {
120        Ok(p) => list[p].rc += 1,
121        Err(ins) => list.insert(ins, build_node(node, schema)),
122    }
123}
124
125/// Build a fresh node (`rc = 1`) with its inline subtree, mirroring
126/// `init_rels_for_new_entry`: one empty list per declared slot; fill in-view slots
127/// from the payload, folding same-sort-key children to `rc += 1`; children arrive in
128/// the operator's sort order and are NOT re-sorted.
129fn build_node(node: &WireNode, schema: &Schema) -> RecvNode {
130    let mut rels: Vec<Vec<RecvNode>> = (0..schema.relationships.len())
131        .map(|_| Vec::new())
132        .collect();
133    for (slot, children) in &node.rels {
134        let Some(child_schema) = schema.rel_child(*slot) else {
135            continue; // join-only / out of view (init_rels `continue`s)
136        };
137        let mut built: Vec<RecvNode> = Vec::with_capacity(children.len());
138        for child in children {
139            let crow = owned_row(child.row.clone());
140            match binary_search(&built, &crow, &child_schema.sort) {
141                Ok(p) => built[p].rc += 1,
142                Err(ins) => built.insert(ins, build_node(child, child_schema)),
143            }
144        }
145        rels[slot.ix()] = built;
146    }
147    RecvNode {
148        row: owned_row(node.row.clone()),
149        rc: 1,
150        rels,
151    }
152}
153
154/// `apply_remove` (`view.rs:440`): physical removal at `rc == 1`, else decrement.
155fn apply_remove(list: &mut Vec<RecvNode>, schema: &Schema, row: &WireRow) {
156    let r = owned_row(row.clone());
157    let pos = match binary_search(list, &r, &schema.sort) {
158        Ok(p) => p,
159        Err(_) => panic!("flat receiver: remove of non-existent node"),
160    };
161    if list[pos].rc == 1 {
162        list.remove(pos);
163    } else {
164        list[pos].rc -= 1;
165    }
166}
167
168/// `apply_edit_change` (`view.rs:540`): `old` locates, `new` places. Branches on
169/// whether the sort key changed (in-place vs the move/ghost/merge protocol).
170fn apply_edit(list: &mut Vec<RecvNode>, schema: &Schema, old: &WireRow, new: &WireRow) {
171    let sort = &schema.sort;
172    let old_r = owned_row(old.clone());
173    let new_r = owned_row(new.clone());
174
175    if compare_rows(sort, &old_r, &new_r) == Ordering::Equal {
176        // Sort key unchanged → edit in place (keeps position + children + rc).
177        let pos = match binary_search(list, &old_r, sort) {
178            Ok(p) => p,
179            Err(_) => panic!("flat receiver: edit of non-existent node"),
180        };
181        list[pos].row = new_r;
182        return;
183    }
184
185    // Sort key changed → the row may move; rc may be > 1.
186    let old_pos = match binary_search(list, &old_r, sort) {
187        Ok(p) => p,
188        Err(_) => panic!("flat receiver: edit old node does not exist"),
189    };
190    let raw = binary_search(list, &new_r, sort);
191    let old_rc = list[old_pos].rc;
192    let found = raw.is_ok();
193    let pos = raw.unwrap_or_else(|e| e);
194
195    // Fast path (`view.rs:566`): rc==1 and the row lands in the same slot after removal.
196    if old_rc == 1 && (pos == old_pos || pos.checked_sub(1) == Some(old_pos)) {
197        list[old_pos].row = new_r;
198        return;
199    }
200
201    // General move (`view.rs:581`).
202    let old_entry = list[old_pos].clone(); // capture before mutating
203    let new_rc = old_rc - 1;
204    let adjusted_pos;
205    if new_rc == 0 {
206        list.remove(old_pos);
207        adjusted_pos = if old_pos < pos { pos - 1 } else { pos };
208    } else {
209        list[old_pos].rc = new_rc; // ghost survives for the other path(s)
210        adjusted_pos = pos;
211    }
212    if found {
213        // Merge into the existing destination entry (keep its children), bump its rc.
214        let existing_rc = list[adjusted_pos].rc;
215        list[adjusted_pos].row = new_r;
216        list[adjusted_pos].rc = existing_rc + 1;
217    } else {
218        // Move the (edited) old entry — keeping its children — to the new position.
219        let mut moved = old_entry;
220        moved.row = new_r;
221        moved.rc = 1;
222        list.insert(adjusted_pos, moved);
223    }
224}
225
226#[cfg(test)]
227mod tests {
228    use super::*;
229    use crate::flat::{FlatChange, FlatOp, PathSeg, WireNode};
230    use crate::value::{OwnedValue as V, RelDef, RelId, Schema};
231
232    // issue(id, val) ordered by (val asc, id asc) — a NON-PK leading sort, so edits to
233    // `val` move rows (exercising the move/ghost/merge protocol). One in-view rel
234    // "kids" → kid(id, parent) ordered by id, plus one GATING slot "gate" (no child
235    // schema) to exercise the in-view gate.
236    fn kid_schema() -> Schema {
237        Schema::new(vec!["id", "parent"], vec![0], vec![(0, true)])
238    }
239    fn root_schema() -> Schema {
240        Schema::new(vec!["id", "val"], vec![0], vec![(1, true), (0, true)]).with_relationships(
241            vec![RelDef::related("kids", kid_schema()), RelDef::new("gate")],
242        )
243    }
244
245    fn wnode(cells: Vec<V>, rels: Vec<(RelId, Vec<WireNode>)>) -> WireNode {
246        WireNode { row: cells, rels }
247    }
248    fn add(cells: Vec<V>) -> FlatChange {
249        FlatChange {
250            path: vec![],
251            op: FlatOp::Add(wnode(cells, vec![])),
252        }
253    }
254    fn add_node(node: WireNode) -> FlatChange {
255        FlatChange {
256            path: vec![],
257            op: FlatOp::Add(node),
258        }
259    }
260    fn remove(cells: Vec<V>) -> FlatChange {
261        FlatChange {
262            path: vec![],
263            op: FlatOp::Remove { row: cells },
264        }
265    }
266    fn edit(old: Vec<V>, new: Vec<V>) -> FlatChange {
267        FlatChange {
268            path: vec![],
269            op: FlatOp::Edit { old, new },
270        }
271    }
272
273    fn ids(list: &[RecvNode]) -> Vec<i64> {
274        list.iter()
275            .map(|n| match n.row.col(0) {
276                crate::value::Value::Int(i) => i,
277                o => panic!("{o:?}"),
278            })
279            .collect()
280    }
281    fn rcs(list: &[RecvNode]) -> Vec<u32> {
282        list.iter().map(|n| n.rc).collect()
283    }
284
285    #[test]
286    fn add_remove_keep_sort_order_and_rc() {
287        let mut r = Receiver::new(root_schema());
288        // Insert out of order; sorted by val (col 1).
289        r.apply(&add(vec![V::Int(1), V::Int(30)]));
290        r.apply(&add(vec![V::Int(2), V::Int(10)]));
291        r.apply(&add(vec![V::Int(3), V::Int(20)]));
292        assert_eq!(ids(r.top()), vec![2, 3, 1]); // by val 10,20,30
293        assert_eq!(rcs(r.top()), vec![1, 1, 1]);
294
295        // Duplicate sort-key add (same val+id) → rc++, no new node.
296        r.apply(&add(vec![V::Int(3), V::Int(20)]));
297        assert_eq!(ids(r.top()), vec![2, 3, 1]);
298        assert_eq!(rcs(r.top()), vec![1, 2, 1]);
299
300        // Remove the rc==2 row once → survives at rc 1.
301        r.apply(&remove(vec![V::Int(3), V::Int(20)]));
302        assert_eq!(rcs(r.top()), vec![1, 1, 1]);
303        // Remove again → physically gone.
304        r.apply(&remove(vec![V::Int(3), V::Int(20)]));
305        assert_eq!(ids(r.top()), vec![2, 1]);
306    }
307
308    #[test]
309    fn edit_in_place_when_sort_key_unchanged() {
310        let mut r = Receiver::new(root_schema());
311        r.apply(&add(vec![V::Int(1), V::Int(10)]));
312        r.apply(&add(vec![V::Int(2), V::Int(20)]));
313        // Edit id=1's PK-irrelevant... here val is the sort key; change a NON-sort col?
314        // Both cols are in the sort (val, id), so to keep the sort key we must keep both.
315        // Instead edit with identical sort key (no-op move) — exercises the Equal branch.
316        r.apply(&edit(
317            vec![V::Int(1), V::Int(10)],
318            vec![V::Int(1), V::Int(10)],
319        ));
320        assert_eq!(ids(r.top()), vec![1, 2]);
321        assert_eq!(rcs(r.top()), vec![1, 1]);
322    }
323
324    #[test]
325    fn edit_move_rc1_relocates() {
326        let mut r = Receiver::new(root_schema());
327        r.apply(&add(vec![V::Int(1), V::Int(10)]));
328        r.apply(&add(vec![V::Int(2), V::Int(20)]));
329        r.apply(&add(vec![V::Int(3), V::Int(30)]));
330        assert_eq!(ids(r.top()), vec![1, 2, 3]);
331        // Move id=1 from val 10 → 25 (between 20 and 30).
332        r.apply(&edit(
333            vec![V::Int(1), V::Int(10)],
334            vec![V::Int(1), V::Int(25)],
335        ));
336        assert_eq!(ids(r.top()), vec![2, 1, 3]); // 20, 25, 30
337        assert_eq!(rcs(r.top()), vec![1, 1, 1]);
338    }
339
340    #[test]
341    fn edit_move_with_ghost_when_rc_gt_1() {
342        let mut r = Receiver::new(root_schema());
343        r.apply(&add(vec![V::Int(1), V::Int(10)]));
344        r.apply(&add(vec![V::Int(1), V::Int(10)])); // rc=2 at (val10,id1)
345        r.apply(&add(vec![V::Int(2), V::Int(30)]));
346        assert_eq!(ids(r.top()), vec![1, 2]);
347        assert_eq!(rcs(r.top()), vec![2, 1]);
348        // Edit ONE ref of id=1 to val 40 (past id=2). rc>1 → ghost stays at val10 (rc1),
349        // a fresh entry inserted at the new position (rc1).
350        r.apply(&edit(
351            vec![V::Int(1), V::Int(10)],
352            vec![V::Int(1), V::Int(40)],
353        ));
354        assert_eq!(ids(r.top()), vec![1, 2, 1]); // val 10 (ghost), 30, 40
355        assert_eq!(rcs(r.top()), vec![1, 1, 1]);
356    }
357
358    #[test]
359    fn edit_move_merges_into_existing_destination() {
360        let mut r = Receiver::new(root_schema());
361        r.apply(&add(vec![V::Int(1), V::Int(10)])); // rc1
362        r.apply(&add(vec![V::Int(1), V::Int(10)])); // rc2 at (10,1)
363        r.apply(&add(vec![V::Int(1), V::Int(20)])); // a DIFFERENT sort key, same id col0
364        assert_eq!(rcs(r.top()), vec![2, 1]); // (10,1)=2, (20,1)=1
365                                              // Move one ref of (10,1) to sort key (20,1) — destination already exists → merge.
366        r.apply(&edit(
367            vec![V::Int(1), V::Int(10)],
368            vec![V::Int(1), V::Int(20)],
369        ));
370        // ghost (10,1) drops to rc1; destination (20,1) bumps to rc2.
371        assert_eq!(ids(r.top()), vec![1, 1]);
372        assert_eq!(rcs(r.top()), vec![1, 2]);
373    }
374
375    #[test]
376    fn child_path_descends_and_in_view_gate_drops_gating_slot() {
377        let mut r = Receiver::new(root_schema());
378        // A parent with one kid inline.
379        let parent = wnode(
380            vec![V::Int(1), V::Int(10)],
381            vec![(RelId(0), vec![wnode(vec![V::Int(100), V::Int(1)], vec![])])],
382        );
383        r.apply(&add_node(parent));
384        assert_eq!(ids(r.top()), vec![1]);
385        assert_eq!(ids(&r.top()[0].rels[0]), vec![100]); // kid present under slot 0
386
387        // Child Add of a second kid, addressed via the "kids" slot (RelId 0).
388        r.apply(&FlatChange {
389            path: vec![PathSeg {
390                rel: RelId(0),
391                parent_row: vec![V::Int(1), V::Int(10)],
392            }],
393            op: FlatOp::Add(wnode(vec![V::Int(50), V::Int(1)], vec![])),
394        });
395        assert_eq!(ids(&r.top()[0].rels[0]), vec![50, 100]); // sorted by kid id
396
397        // Child Add addressed via the GATING slot (RelId 1, child schema None) → no-op.
398        r.apply(&FlatChange {
399            path: vec![PathSeg {
400                rel: RelId(1),
401                parent_row: vec![V::Int(1), V::Int(10)],
402            }],
403            op: FlatOp::Add(wnode(vec![V::Int(999), V::Int(1)], vec![])),
404        });
405        // The gating slot stays empty; kids unchanged.
406        assert!(r.top()[0].rels[1].is_empty());
407        assert_eq!(ids(&r.top()[0].rels[0]), vec![50, 100]);
408    }
409}
rindle/flat_receiver.rs

rindle/
flat_receiver.rs
