rindle/op/exists.rs
1//! `Exists` (`ivm/exists.ts`, spec `07` §3.5) — keep a parent row iff its child
2//! relationship is non-empty (`EXISTS`) or empty (`NOT EXISTS`). A **`FilterChain`
3//! link**: it has no `fetch`; it `filter(node) -> bool` gates the upstream scan and
4//! `push`es incremental changes. The child relationship is already on the `Node`
5//! (attached by an upstream `Join` whose child pipeline is bounded to `EXISTS_LIMIT`
6//! by [`Cap`](crate::op::Cap)/[`Take`](crate::op::Take)), so Exists just **counts**
7//! it.
8//!
9//! Dispatched through the graph's filter-chain seam: [`Graph`](crate::graph::Graph)'s
10//! `begin_filter`/`chain_filter`/`end_filter`/`chain_push` each carry a one-line
11//! `Operator::Exists` arm that delegates here, and this file drives the rest of the
12//! chain back through the same `pub(crate)` graph methods (`g.chain_filter(out, …)`
13//! / `g.chain_push(out, …)`). All Exists logic lives here (the fan-out seam); only
14//! the dispatch arms live in `graph.rs`.
15//!
16//! Three subtleties ported faithfully:
17//! - a **per-fetch-loop size cache** (`begin`/`end` bracket it; reset on `end`),
18//! suppressed when `no_size_reuse` (the parent join key is the PK ⇒ each parent is
19//! unique) or `in_push` (relationships are transiently inconsistent mid-push);
20//! - the **`in_push` reentrancy guard** (a push within a push is a bug);
21//! - the **boundary-flip forced-relationship rewrite**: when a child Add/Remove
22//! flips membership, Exists converts it into a parent Add/Remove and rewrites the
23//! parent's relationship so the downstream view stays consistent (an EXISTS-remove
24//! carries just the removed child; a NOT-EXISTS-add carries an empty relationship).
25
26use std::cell::{Cell, RefCell};
27use std::collections::HashMap;
28
29use crate::change::{Change, ChangeType, Node, NodeStream, Relationship};
30use crate::graph::{Graph, NodeId};
31use crate::op::partition::encode_state_key;
32use crate::value::{ColId, OwnedValue, RelId};
33
34/// `Exists`: a relationship-size gate in the `where` filter sub-graph.
35pub struct Exists {
36 /// Upstream `FilterInput` (the `FilterStart` / prior link) — for `input_schema`.
37 pub input: NodeId,
38 /// The relationship slot to count (resolved from the subquery alias against the
39 /// parent schema at build time).
40 pub rel: RelId,
41 /// `NOT EXISTS`.
42 pub not: bool,
43 /// The parent-join-key columns — the cache key (`condition.parentField`).
44 pub parent_join_key: Vec<ColId>,
45 /// `parentJoinKey == primaryKey` ⇒ each parent is unique ⇒ the cache can never
46 /// hit, so skip it entirely (`exists.ts:60`).
47 no_size_reuse: bool,
48 /// Per-fetch-loop size cache keyed by the parent-join-key tuple (reset on
49 /// `end`). Ephemeral to one fetch loop — never persisted to `Storage`.
50 cache: RefCell<HashMap<String, bool>>,
51 /// `true` while processing a push: doubles as the reentrancy guard and the
52 /// cache-suppression flag (`exists.ts:39`).
53 in_push: Cell<bool>,
54 /// The downstream `FilterOutput` chain link (wired late via
55 /// [`Graph::set_output`](crate::graph::Graph::set_output)).
56 pub output: Cell<Option<NodeId>>,
57}
58
59/// Clears [`Exists::in_push`] on drop — the RAII rendering of the JS `finally`
60/// (`exists.ts:205`).
61struct InPushGuard<'a>(&'a Cell<bool>);
62impl Drop for InPushGuard<'_> {
63 fn drop(&mut self) {
64 self.0.set(false);
65 }
66}
67
68impl Exists {
69 pub fn new(
70 input: NodeId,
71 rel: RelId,
72 parent_join_key: Vec<ColId>,
73 not: bool,
74 primary_key: &[ColId],
75 ) -> Exists {
76 let no_size_reuse = parent_join_key.as_slice() == primary_key;
77 Exists {
78 input,
79 rel,
80 not,
81 parent_join_key,
82 no_size_reuse,
83 cache: RefCell::new(HashMap::new()),
84 in_push: Cell::new(false),
85 output: Cell::new(None),
86 }
87 }
88
89 fn out(&self) -> NodeId {
90 self.output.get().expect("Exists output not wired")
91 }
92
93 // --- filter-chain lifecycle (driven by graph.rs dispatch arms) ----------
94
95 /// `beginFilter` (`exists.ts:71`): just delegate downstream (the cache is reset
96 /// on `end`, not begin).
97 pub fn begin(&self, g: &Graph) {
98 g.begin_filter(self.out());
99 }
100
101 /// `endFilter` (`exists.ts:75`): reset the per-loop cache, then delegate.
102 pub fn end(&self, g: &Graph) {
103 self.cache.borrow_mut().clear();
104 g.end_filter(self.out());
105 }
106
107 /// `filter(node)` (`exists.ts:80`): the epoch-gated cache lookup, then
108 /// `#filter(node) && output.filter(node)`.
109 pub fn filter(&self, g: &Graph, node: &Node) -> bool {
110 let exists = if !self.no_size_reuse && !self.in_push.get() {
111 let key = self.cache_key(node);
112 // Drop the immutable borrow (via `.copied()`) before any `borrow_mut`.
113 let cached = self.cache.borrow().get(&key).copied();
114 Some(cached.unwrap_or_else(|| {
115 let e = self.fetch_exists(node);
116 self.cache.borrow_mut().insert(key, e);
117 e
118 }))
119 } else {
120 None
121 };
122 self.gate(node, exists) && g.chain_filter(self.out(), node)
123 }
124
125 // --- push ---------------------------------------------------------------
126
127 /// `push` (`exists.ts:109`). Add/Edit/Remove of the *parent* cannot change the
128 /// child relationship's size → `#pushWithFilter`. A Child **Add/Remove to this
129 /// relationship** can flip membership → refetch the size and either pass it
130 /// through filtered or convert it into a parent Add/Remove. Returns the change(s)
131 /// to forward (the chassis bubbles them up; a terminal `FilterEnd` consumes them).
132 pub fn push_chain<'g>(&'g self, g: &'g Graph, change: Change<'g>) -> Vec<Change<'g>> {
133 assert!(!self.in_push.get(), "Exists: unexpected re-entrancy");
134 self.in_push.set(true);
135 let _guard = InPushGuard(&self.in_push);
136
137 match change {
138 Change::Add(_) | Change::Edit { .. } | Change::Remove(_) => {
139 self.push_with_filter(g, change, None)
140 }
141 Change::Child { node, rel, child } => {
142 let child_type = child.change_type();
143 // A change to a *different* relationship, or a nested Edit/Child to
144 // this one, cannot change this relationship's size.
145 if rel != self.rel || matches!(child_type, ChangeType::Edit | ChangeType::Child) {
146 return self.push_with_filter(g, Change::Child { node, rel, child }, None);
147 }
148 match child_type {
149 ChangeType::Add => {
150 let size = self.fetch_size(&node);
151 if size == 1 {
152 // The relationship just became non-empty.
153 if self.not {
154 // NOT EXISTS: parent newly fails. The just-added child
155 // was never in the output, so force the relationship
156 // empty on the Remove.
157 g.chain_push(
158 self.out(),
159 Change::Remove(force_rel_empty(node, self.rel)),
160 )
161 } else {
162 // EXISTS: parent newly satisfies → Add (its relationship
163 // thunk already includes the added child via the overlay).
164 g.chain_push(self.out(), Change::Add(node))
165 }
166 } else {
167 self.push_with_filter(
168 g,
169 Change::Child { node, rel, child },
170 Some(size > 0),
171 )
172 }
173 }
174 ChangeType::Remove => {
175 let size = self.fetch_size(&node);
176 if size == 0 {
177 // The relationship just became empty.
178 if self.not {
179 // NOT EXISTS: parent newly satisfies → Add.
180 g.chain_push(self.out(), Change::Add(node))
181 } else {
182 // EXISTS: parent newly fails → Remove, carrying just the
183 // removed child (still present in the view's copy).
184 let Change::Remove(removed) = *child else {
185 unreachable!()
186 };
187 let forced = force_rel_single(node, self.rel, removed);
188 g.chain_push(self.out(), Change::Remove(forced))
189 }
190 } else {
191 self.push_with_filter(
192 g,
193 Change::Child { node, rel, child },
194 Some(size > 0),
195 )
196 }
197 }
198 ChangeType::Edit | ChangeType::Child => unreachable!("handled above"),
199 }
200 }
201 }
202 }
203
204 /// `#pushWithFilter` (`exists.ts:235`): forward iff `#filter` passes for the
205 /// change's node (using `exists` if supplied, else computing the size).
206 fn push_with_filter<'g>(
207 &'g self,
208 g: &'g Graph,
209 change: Change<'g>,
210 exists: Option<bool>,
211 ) -> Vec<Change<'g>> {
212 let pass = {
213 let node = match &change {
214 Change::Add(n) | Change::Remove(n) => n,
215 Change::Edit { node, .. } | Change::Child { node, .. } => node,
216 };
217 self.gate(node, exists)
218 };
219 if pass {
220 g.chain_push(self.out(), change)
221 } else {
222 Vec::new()
223 }
224 }
225
226 // --- size / gate / cache ------------------------------------------------
227
228 /// `#filter` (`exists.ts:219`): `not ? !exists : exists`, computing the size if
229 /// `exists` was not supplied.
230 fn gate(&self, node: &Node, exists: Option<bool>) -> bool {
231 let e = exists.unwrap_or_else(|| self.fetch_exists(node));
232 // `not ? !exists : exists` == `not XOR exists`.
233 self.not ^ e
234 }
235
236 fn fetch_exists(&self, node: &Node) -> bool {
237 // Can't early-return at the first child: Take/Cap don't support early return
238 // during initial fetch, so count up to the limit (`exists.ts:241`).
239 self.fetch_size(node) > 0
240 }
241
242 fn fetch_size(&self, node: &Node) -> usize {
243 let rel = node
244 .rels
245 .iter()
246 .find(|r| r.slot == self.rel)
247 .expect("Exists: relationship not found on node");
248 (rel.thunk)().count()
249 }
250
251 fn cache_key(&self, node: &Node) -> String {
252 let vals: Vec<OwnedValue> = self
253 .parent_join_key
254 .iter()
255 .map(|&c| node.row.col(c).to_owned())
256 .collect();
257 // The prefix is irrelevant (each Exists has its own cache); reuse the shared
258 // value-tuple encoder for the JSON-of-normalized-values key.
259 encode_state_key("", &vals)
260 }
261}
262
263/// Replace `node`'s `slot` relationship with one that yields a single `child` (the
264/// EXISTS-remove forced relationship, `exists.ts:181`). Single-use: the held node is
265/// not `Clone` (its own thunks aren't), so the stream yields once then is empty —
266/// sufficient for the materialize-once consumers (the view / `Catch`).
267fn force_rel_single<'g>(mut node: Node<'g>, slot: RelId, child: Node<'g>) -> Node<'g> {
268 node.rels.retain(|r| r.slot != slot);
269 let held = RefCell::new(Some(vec![child]));
270 node.rels.push(Relationship {
271 slot,
272 thunk: Box::new(move || {
273 let v: Vec<Node<'g>> = held.borrow_mut().take().unwrap_or_default();
274 Box::new(v.into_iter()) as NodeStream<'g>
275 }),
276 });
277 node
278}
279
280/// Replace `node`'s `slot` relationship with an empty one (the NOT-EXISTS-add forced
281/// relationship, `exists.ts:152`).
282fn force_rel_empty(mut node: Node<'_>, slot: RelId) -> Node<'_> {
283 node.rels.retain(|r| r.slot != slot);
284 node.rels.push(Relationship {
285 slot,
286 thunk: Box::new(|| Box::new(std::iter::empty())),
287 });
288 node
289}