rindle/family.rs
1//! Parameterized query families — the engine half of design 310
2//! (`designs/310-PARAMETERIZED-QUERY-FAMILIES-DESIGN.md` §4).
3//!
4//! V subscriptions that are the same query but for a root-level equality literal
5//! (`albums.where(artistId = ?a).related(tracks)` for V values of `?a`) compile to
6//! **one** pipeline whose parameter columns are a *partition dimension* instead of a
7//! predicate. The pipeline maintains results for every currently-bound parameter tuple
8//! at once; the root connection's holed conjuncts are replaced by one membership test
9//! — *row's partition-key tuple ∈ [`BindingSet`](crate::family::BindingSet)* — and the root `limit` becomes a
10//! per-binding top-N (a `Take` partitioned by the parameter columns). Everything below
11//! the root is naturally partition-safe: root partitions are disjoint (a root row has
12//! exactly one partition-key tuple), so no operator other than the root limiter learns
13//! about partitions.
14//!
15//! Entry points: [`build_family_pipeline`](crate::build_family_pipeline) lowers a
16//! template (the stripped AST + parameter names — `rindle-wire`'s `FamilyTemplate`,
17//! split into its two fields because `rindle-wire` depends on this crate and not the
18//! reverse) into a [`FamilyPipeline`](crate::family::FamilyPipeline); `Graph::bind_family_partition` /
19//! `Graph::unbind_family_partition` / `Graph::hydrate_family` drive it.
20//!
21//! **Binding mutations happen only between pushes** (impl plan D3). The
22//! [`BindingSet`](crate::family::BindingSet)
23//! is interior-mutable so the root connection's predicate can read it through a shared
24//! `Rc`, but nothing mutates it while a push or a fetch is in flight: bind/unbind are
25//! host commands, never operator side effects. That is what makes the reentrant
26//! fetch-during-push compose with the membership predicate exactly as it does with any
27//! other predicate — the `RefCell` borrow inside [`BindingSet::contains_row`](crate::family::BindingSet::contains_row) is a
28//! momentary shared borrow that no writer ever contends with.
29
30use std::cell::RefCell;
31use std::collections::HashSet;
32use std::rc::Rc;
33
34use crate::canon::{canonical_key, CanonKey};
35use crate::change::Constraint;
36use crate::graph::NodeId;
37use crate::value::{ColId, OwnedRow};
38
39/// The currently-bound partition-key tuples of one family pipeline (design §4.1's
40/// "binding-set handle"). Shared (`Rc`) between the root connection's predicate and
41/// the graph's bind/unbind path; interior-mutable, but mutated ONLY between pushes
42/// (impl plan D3) — the predicate takes a momentary shared borrow per row.
43#[derive(Default)]
44pub struct BindingSet {
45 inner: RefCell<HashSet<CanonKey>>,
46}
47
48impl BindingSet {
49 pub fn new() -> Rc<BindingSet> {
50 Rc::new(BindingSet::default())
51 }
52
53 /// The membership test of design §4.1: is `row`'s partition-key tuple (its `cols`
54 /// cells, canonicalized) currently bound? Never holds the borrow across a vend.
55 #[inline]
56 pub fn contains_row(&self, row: &OwnedRow, cols: &[ColId]) -> bool {
57 let key = canonical_key(row, cols);
58 self.inner.borrow().contains(&key)
59 }
60
61 pub fn contains(&self, b: &CanonKey) -> bool {
62 self.inner.borrow().contains(b)
63 }
64
65 /// Bind `b`. `false` iff it was already bound.
66 pub fn insert(&self, b: CanonKey) -> bool {
67 self.inner.borrow_mut().insert(b)
68 }
69
70 /// Unbind `b`. `false` iff it was not bound.
71 pub fn remove(&self, b: &CanonKey) -> bool {
72 self.inner.borrow_mut().remove(b)
73 }
74
75 pub fn len(&self) -> usize {
76 self.inner.borrow().len()
77 }
78
79 pub fn is_empty(&self) -> bool {
80 self.inner.borrow().is_empty()
81 }
82
83 /// Every bound tuple, in a deterministic (sorted) order.
84 pub fn snapshot(&self) -> Vec<CanonKey> {
85 let mut out: Vec<CanonKey> = self.inner.borrow().iter().cloned().collect();
86 out.sort();
87 out
88 }
89}
90
91impl std::fmt::Debug for BindingSet {
92 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
93 f.debug_set().entries(self.inner.borrow().iter()).finish()
94 }
95}
96
97/// A built family pipeline: what `build_family_pipeline` returns and what the graph's
98/// bind/unbind/hydrate entry points take. The node ids are the pipeline's own (the
99/// caller records them in a `PipelineManifest` exactly as for a singleton).
100#[derive(Debug)]
101pub struct FamilyPipeline {
102 /// What the sink is wired to (the pipeline top), as `build_pipeline` returns it.
103 pub top: NodeId,
104 /// The root `SourceConn` — the connection carrying the membership predicate (and,
105 /// with dynamic guards, the growing push-index guard).
106 pub root_conn: NodeId,
107 /// The partitioned root limiter, when the template carries a `limit`.
108 pub root_take: Option<NodeId>,
109 /// The partition key: the parameter columns resolved on the root schema, in
110 /// parameter order. Every root row carries its own partition value, which is what
111 /// makes output routing a column read (design §4.2).
112 pub param_cols: Vec<ColId>,
113 /// The binding-set handle the root connection's predicate reads.
114 pub bindings: Rc<BindingSet>,
115 /// The last spine node below the `related` join chain (the root `Take` when there
116 /// is a `limit`, else the last of connection / `Skip` / EXISTS gates). Its output
117 /// edge feeds the first `related` join, or the sink. Unbind injects a partition's
118 /// synthetic removes here (impl plan D5), *above* the root limiter and the gates
119 /// the rows already passed, so the relationship joins run their ordinary
120 /// parent-left cleanup.
121 pub(crate) spine_tail: NodeId,
122 /// The non-flipped EXISTS relationship joins built on the spine (below the root
123 /// limiter). Their child limiters are partitioned per parent; unbind evicts those
124 /// partitions for the partition's rows directly, since the synthetic removes enter
125 /// above them.
126 pub(crate) spine_joins: Vec<NodeId>,
127}
128
129impl FamilyPipeline {
130 /// The fetch constraint that names binding `b`'s partition: the parameter columns
131 /// paired with the binding's canonical representatives (`CanonVal::to_owned_value`).
132 pub fn constraint_for(&self, b: &CanonKey) -> Constraint {
133 assert_eq!(
134 b.len(),
135 self.param_cols.len(),
136 "binding arity must match the family's parameter count"
137 );
138 self.param_cols
139 .iter()
140 .zip(b)
141 .map(|(&c, v)| (c, v.to_owned_value()))
142 .collect()
143 }
144}
145
146#[cfg(test)]
147mod tests {
148 use super::*;
149 use crate::canon::CanonVal;
150 use crate::value::{owned_row, OwnedValue};
151
152 #[test]
153 fn binding_set_membership_is_canonical() {
154 let set = BindingSet::new();
155 assert!(set.is_empty());
156 assert!(set.insert(vec![CanonVal::Int(7)]));
157 assert!(!set.insert(vec![CanonVal::Int(7)]), "already bound");
158 assert_eq!(set.len(), 1);
159 // A `Float(7.0)` cell is in the `Int(7)` class.
160 let row = owned_row(vec![OwnedValue::Int(1), OwnedValue::Float(7.0)]);
161 assert!(set.contains_row(&row, &[1]));
162 let other = owned_row(vec![OwnedValue::Int(1), OwnedValue::Int(8)]);
163 assert!(!set.contains_row(&other, &[1]));
164 assert!(set.remove(&vec![CanonVal::Int(7)]));
165 assert!(!set.remove(&vec![CanonVal::Int(7)]), "not bound");
166 assert!(!set.contains_row(&row, &[1]));
167 }
168
169 #[test]
170 fn snapshot_is_sorted_and_complete() {
171 let set = BindingSet::new();
172 set.insert(vec![CanonVal::Int(9)]);
173 set.insert(vec![CanonVal::Int(2)]);
174 set.insert(vec![CanonVal::Str("a".into())]);
175 let snap = set.snapshot();
176 assert_eq!(snap.len(), 3);
177 assert_eq!(snap, {
178 let mut s = snap.clone();
179 s.sort();
180 s
181 });
182 }
183}