rindle/op/
cap.rs

1//! `Cap` (`ivm/cap.ts`, spec `07` §3.8) — the **unordered** count-based limiter for
2//! non-flipped `EXISTS` child pipelines. Cap is the sibling of [`Take`](crate::op::Take):
3//! both bound a partition to `limit` rows and refill on remove, but Cap tracks
4//! membership by a **primary-key set** (`CapState{size, pks}`) instead of a sorted
5//! bound. Because only the *count* matters to `Exists` (`> 0` vs `== 0` up to the
6//! limit), Cap needs no comparator — which lets the SQLite leaf skip `ORDER BY`.
7//!
8//! Cap asserts **no `start`, no `reverse`** (its only consumer is a join fetching
9//! with a constraint that matches the partition key), and does **PK point-lookups**
10//! on fetch rather than an ordered scan. Stateful: it reads/writes a [`StorageId`]
11//! slot (the shipped [`StorageValue::Cap`] shape — `pks` are serialized strings, so
12//! the point-lookup round-trips each through [`serialize_pk`]/[`deserialize_pk`]).
13
14use std::cell::Cell;
15use std::sync::Arc;
16
17use crate::change::{Change, Constraint, FetchRequest, Node, NodeStream, OutEdge};
18use crate::graph::{Graph, NodeId, StorageId};
19use crate::op::partition::{
20    constraint_matches_partition_key, encode_state_key, partition_key_unchanged,
21};
22use crate::storage::StorageValue;
23use crate::value::{ColId, OwnedRow as Row, OwnedValue, Value};
24
25/// `Cap`: unordered top-N by PK-set membership, backed by a [`StorageId`] slot.
26pub struct Cap {
27    /// Upstream input (the EXISTS-child connection, unordered in production).
28    pub input: NodeId,
29    /// Handle into the [`Graph`](crate::graph::Graph) storage arena holding this op's per-partition
30    /// `CapState{size, pks}` under `encode_state_key`("cap", …).
31    pub storage: StorageId,
32    /// The `EXISTS_LIMIT` count. `0` ⇒ yields nothing and stores no state.
33    pub limit: u32,
34    /// The partition columns (the correlation **child** field), or `None` for a
35    /// single global partition.
36    pub partition_key: Option<Vec<ColId>>,
37    /// The input's primary-key columns — the identity Cap tracks by
38    /// (`serializePK`, `cap.ts:315`).
39    pub primary_key: Vec<ColId>,
40    /// The single downstream edge as a **port-carrying** [`OutEdge`] (Cap feeds a
41    /// relationship join's child port). Wired via
42    /// [`Graph::set_output`](crate::graph::Graph::set_output) /
43    /// [`Graph::set_out_edge`](crate::graph::Graph::set_out_edge).
44    pub output: Cell<Option<OutEdge>>,
45}
46
47impl Cap {
48    pub fn new(
49        input: NodeId,
50        storage: StorageId,
51        limit: u32,
52        partition_key: Option<Vec<ColId>>,
53        primary_key: Vec<ColId>,
54    ) -> Cap {
55        Cap {
56            input,
57            storage,
58            limit,
59            partition_key,
60            primary_key,
61            output: Cell::new(None),
62        }
63    }
64
65    // --- storage helpers ----------------------------------------------------
66
67    fn get_state(&self, g: &Graph, key: &str) -> Option<(u32, Vec<Box<str>>)> {
68        match g.storage(self.storage).get(key) {
69            None => None,
70            Some(StorageValue::Cap { size, pks }) => Some((size, pks)),
71            Some(other) => unreachable!("cap state slot held {other:?}"),
72        }
73    }
74
75    fn set_state(&self, g: &Graph, key: &str, size: u32, pks: Vec<Box<str>>) {
76        g.storage(self.storage)
77            .set(key, StorageValue::Cap { size, pks });
78    }
79
80    /// The cap-state key for a partition identified by a **row** (`getCapStateKey`
81    /// over a `Row`, `cap.ts:300`). Unpartitioned ⇒ the single global key.
82    fn key_from_row(&self, row: &Row) -> String {
83        match &self.partition_key {
84            None => encode_state_key("cap", &[]),
85            Some(pk) => {
86                let vals: Vec<OwnedValue> = pk.iter().map(|&c| row.col(c).to_owned()).collect();
87                encode_state_key("cap", &vals)
88            }
89        }
90    }
91
92    /// The cap-state key for a partition identified by a fetch **constraint**.
93    fn key_from_constraint(&self, c: Option<&Constraint>) -> String {
94        match (&self.partition_key, c) {
95            (Some(pk), Some(c)) => {
96                let vals: Vec<OwnedValue> = pk
97                    .iter()
98                    .map(|&col| {
99                        c.iter()
100                            .find(|(cc, _)| *cc == col)
101                            .map(|(_, v)| v.clone())
102                            .unwrap_or(OwnedValue::Null)
103                    })
104                    .collect();
105                encode_state_key("cap", &vals)
106            }
107            _ => encode_state_key("cap", &[]),
108        }
109    }
110
111    /// The partition constraint for a push on `row` (the reentrant refill fetch),
112    /// or `None` when unpartitioned (a bare scan).
113    fn constraint_for(&self, row: &Row) -> Option<Constraint> {
114        self.partition_key
115            .as_ref()
116            .map(|pk| pk.iter().map(|&c| (c, row.col(c).to_owned())).collect())
117    }
118
119    /// Delete the per-partition slot identified by `constraint`, if this Cap is
120    /// partitioned and the constraint covers its partition key — the sibling of
121    /// [`Take::evict_partition`](crate::op::Take::evict_partition), called by the
122    /// relationship join when a parent leaves a bounded parent view. Safe for the same
123    /// reason: an un-hydrated Cap partition is re-hydrated by the join's constrained
124    /// `fetch` on parent re-entry, and a push to a missing partition is dropped.
125    ///
126    /// This addresses the parent-left-the-*view* source of a zombie partition, which is
127    /// the only one that reaches a `related` child limiter. A **spine EXISTS** `Cap` is
128    /// upstream of the root `Take` instead, so no parent `Remove` ever reaches its join;
129    /// that side is reclaimed by the mirror walk
130    /// ([`Graph::evict_upstream_child_partitions`](crate::graph::Graph)), driven from the
131    /// `Take`'s drop/displace points. See `follow-ups/04-cap-partition-leak.md`.
132    pub fn evict_partition(&self, g: &Graph, constraint: &Constraint) {
133        let Some(pk) = &self.partition_key else {
134            return;
135        };
136        if !constraint_matches_partition_key(constraint, pk) {
137            return;
138        }
139        let key = self.key_from_constraint(Some(constraint));
140        g.storage(self.storage).del(&key);
141    }
142
143    // --- fetch --------------------------------------------------------------
144
145    /// Lazy pull (`cap.ts:87`). Asserts no `start`/`reverse`. With state, fetches
146    /// each tracked PK by a point-lookup; without state, hydrates.
147    pub fn fetch<'g>(&'g self, g: &'g Graph, req: &FetchRequest) -> NodeStream<'g> {
148        assert!(req.start.is_none(), "Cap does not support start");
149        assert!(!req.reverse, "Cap does not support reverse");
150        debug_assert!(
151            self.partition_key.as_ref().is_none_or(|pk| req
152                .constraint
153                .as_ref()
154                .is_some_and(|c| constraint_matches_partition_key(c, pk))),
155            "Cap fetch: constraint must match partition key when partitioned"
156        );
157
158        let key = self.key_from_constraint(req.constraint.as_ref());
159        match self.get_state(g, &key) {
160            None => self.initial_fetch(g, req),
161            Some((0, _)) => Box::new(std::iter::empty()),
162            Some((_, pks)) => {
163                // PK point-lookups: fetch each tracked row by its PK directly
164                // (`cap.ts:113`). Eager — `limit` is tiny (EXISTS_LIMIT = 3).
165                let mut out: Vec<Node<'g>> = Vec::new();
166                for pk in &pks {
167                    let c = deserialize_pk(pk, &self.primary_key);
168                    out.extend(g.fetch(self.input, &FetchRequest::with_constraint(c)));
169                }
170                Box::new(out.into_iter())
171            }
172        }
173    }
174
175    /// `#initialFetch` (`cap.ts:125`): hydrate by recording the first `limit` rows'
176    /// PKs. Eager (like [`Take::initial_fetch`](crate::op::Take)) — every real
177    /// consumer fully drains a hydrate.
178    fn initial_fetch<'g>(&'g self, g: &'g Graph, req: &FetchRequest) -> NodeStream<'g> {
179        if self.limit == 0 {
180            return Box::new(std::iter::empty());
181        }
182        let key = self.key_from_constraint(req.constraint.as_ref());
183        debug_assert!(
184            self.get_state(g, &key).is_none(),
185            "Cap initial fetch: state should be undefined"
186        );
187        let mut size = 0u32;
188        let mut pks: Vec<Box<str>> = Vec::new();
189        let mut out: Vec<Node<'g>> = Vec::new();
190        for node in g.fetch(self.input, req) {
191            pks.push(serialize_pk(&node.row, &self.primary_key));
192            out.push(node);
193            size += 1;
194            if size == self.limit {
195                break;
196            }
197        }
198        self.set_state(g, &key, size, pks);
199        Box::new(out.into_iter())
200    }
201
202    // --- push ---------------------------------------------------------------
203
204    /// Eager push (`cap.ts:177`). Edit routes to `Cap::push_edit`; otherwise look
205    /// up the partition's state (drop if never hydrated) and dispatch Add / Remove /
206    /// Child by PK-set membership.
207    pub fn push<'g>(&'g self, g: &'g Graph, change: Change<'g>) {
208        if matches!(change, Change::Edit { .. }) {
209            return self.push_edit(g, change);
210        }
211
212        let row = match &change {
213            Change::Add(n) | Change::Remove(n) => &n.row,
214            Change::Child { node, .. } => &node.row,
215            Change::Edit { .. } => unreachable!(),
216        };
217        let key = self.key_from_row(row);
218        let Some((size, pks)) = self.get_state(g, &key) else {
219            return;
220        };
221        let pk = serialize_pk(row, &self.primary_key);
222        let out = self.output.get().expect("Cap output not wired");
223
224        match change {
225            Change::Add(node) => {
226                if size < self.limit {
227                    let mut pks = pks;
228                    pks.push(pk);
229                    self.set_state(g, &key, size + 1, pks);
230                    g.push(out.node, Change::Add(node), out.port);
231                }
232                // else full ⇒ drop
233            }
234            Change::Remove(node) => self.push_remove(g, node, out, &key, size, pks, &pk),
235            Change::Child { node, rel, child } => {
236                if pks.iter().any(|p| **p == *pk) {
237                    g.push(out.node, Change::Child { node, rel, child }, out.port);
238                }
239            }
240            Change::Edit { .. } => unreachable!(),
241        }
242    }
243
244    /// Remove (`cap.ts:203`). Untracked PK ⇒ drop. Else remove the PK and try to
245    /// refill from the first partition row whose PK is not in the (post-removal)
246    /// set: store WITHOUT the replacement, forward the Remove, then add the
247    /// replacement's PK and forward its Add (the state-write ordering that hides the
248    /// in-flight change from a reentrant refetch). No refill ⇒ just the Remove.
249    #[allow(clippy::too_many_arguments)]
250    fn push_remove<'g>(
251        &'g self,
252        g: &'g Graph,
253        node: Node<'g>,
254        out: OutEdge,
255        key: &str,
256        size: u32,
257        pks: Vec<Box<str>>,
258        pk: &str,
259    ) {
260        let Some(idx) = pks.iter().position(|p| **p == *pk) else {
261            return; // not in our set ⇒ drop
262        };
263        let mut pks = pks;
264        pks.remove(idx);
265        let new_size = size - 1;
266
267        // Refill: scan the partition for the first row whose PK is not tracked.
268        let req = FetchRequest {
269            constraint: self.constraint_for(&node.row),
270            ..Default::default()
271        };
272        let mut replacement: Option<Node<'g>> = None;
273        for n in g.fetch(self.input, &req) {
274            let npk = serialize_pk(&n.row, &self.primary_key);
275            if !pks.iter().any(|p| **p == *npk) {
276                replacement = Some(n);
277                break;
278            }
279        }
280
281        match replacement {
282            Some(repl) => {
283                // Store WITHOUT the replacement, forward the Remove, then add it.
284                self.set_state(g, key, new_size, pks.clone());
285                g.push(out.node, Change::Remove(node), out.port);
286                let repl_pk = serialize_pk(&repl.row, &self.primary_key);
287                let mut pks = pks;
288                pks.push(repl_pk);
289                self.set_state(g, key, new_size + 1, pks);
290                g.push(out.node, Change::Add(repl), out.port);
291            }
292            None => {
293                self.set_state(g, key, new_size, pks);
294                g.push(out.node, Change::Remove(node), out.port);
295            }
296        }
297    }
298
299    /// `#pushEditChange` (`cap.ts:260`). Asserts the partition key is unchanged. If
300    /// the old PK is tracked, swap it for the new PK (when changed) and forward the
301    /// Edit; else drop.
302    fn push_edit<'g>(&'g self, g: &'g Graph, change: Change<'g>) {
303        let Change::Edit { node, old } = change else {
304            unreachable!()
305        };
306        if let Some(pk) = &self.partition_key {
307            debug_assert!(
308                partition_key_unchanged(&old.row, &node.row, pk),
309                "Cap: unexpected change of partition key"
310            );
311        }
312        let key = self.key_from_row(&old.row);
313        let Some((size, pks)) = self.get_state(g, &key) else {
314            return;
315        };
316        let old_pk = serialize_pk(&old.row, &self.primary_key);
317        let out = self.output.get().expect("Cap output not wired");
318
319        if pks.iter().any(|p| **p == *old_pk) {
320            let new_pk = serialize_pk(&node.row, &self.primary_key);
321            if new_pk != old_pk {
322                let pks: Vec<Box<str>> = pks
323                    .into_iter()
324                    .map(|p| if *p == *old_pk { new_pk.clone() } else { p })
325                    .collect();
326                self.set_state(g, &key, size, pks);
327            }
328            g.push(out.node, Change::Edit { node, old }, out.port);
329        }
330        // else not tracked ⇒ drop
331    }
332}
333
334/// `serializePK` (`cap.ts:315`): a reversible, self-describing encoding of the
335/// row's primary-key values — the JS `JSON.stringify(pk.map(k => row[k]))`. Each
336/// value is `<tag><byte-len>:<content>` (tag selects the type; numeric/bool/null
337/// content is ASCII, str/json content is the raw UTF-8), so [`deserialize_pk`]
338/// round-trips it exactly without needing the schema's column types.
339pub(crate) fn serialize_pk(row: &Row, pk: &[ColId]) -> Box<str> {
340    let mut s = String::new();
341    for &c in pk {
342        let (tag, content): (char, std::borrow::Cow<'_, str>) = match row.col(c) {
343            // A PK cell is never `Absent` in practice (presence-required); a distinct
344            // round-trippable tag keeps the codec total.
345            Value::Absent => ('a', "".into()),
346            Value::Null => ('n', "".into()),
347            Value::Bool(b) => ('b', if b { "1".into() } else { "0".into() }),
348            Value::Int(i) => ('i', i.to_string().into()),
349            Value::Float(f) => ('f', f.to_bits().to_string().into()),
350            // Row text is UTF-8-validated at construction.
351            Value::Str(x) => ('s', String::from_utf8_lossy(x)),
352            Value::Json(x) => ('j', String::from_utf8_lossy(x)),
353        };
354        s.push(tag);
355        s.push_str(&content.len().to_string());
356        s.push(':');
357        s.push_str(&content);
358    }
359    s.into_boxed_str()
360}
361
362/// `deserializePKToConstraint` (`cap.ts:319`): parse a [`serialize_pk`] string back
363/// into a [`Constraint`] mapping each primary-key column to its value, for the
364/// point-lookup fetch.
365pub(crate) fn deserialize_pk(pk: &str, primary_key: &[ColId]) -> Constraint {
366    let bytes = pk.as_bytes();
367    let mut i = 0;
368    let mut out = Vec::with_capacity(primary_key.len());
369    for &col in primary_key {
370        let tag = bytes[i];
371        i += 1;
372        let mut len = 0usize;
373        while bytes[i] != b':' {
374            len = len * 10 + usize::from(bytes[i] - b'0');
375            i += 1;
376        }
377        i += 1; // skip ':'
378        let content = &pk[i..i + len];
379        i += len;
380        let v = match tag {
381            b'a' => OwnedValue::Absent,
382            b'n' => OwnedValue::Null,
383            b'b' => OwnedValue::Bool(content == "1"),
384            b'i' => OwnedValue::Int(content.parse().expect("pk int")),
385            b'f' => OwnedValue::Float(f64::from_bits(content.parse().expect("pk float bits"))),
386            b's' => OwnedValue::str(content),
387            b'j' => OwnedValue::Json(Arc::from(content)),
388            other => unreachable!("bad pk tag {other}"),
389        };
390        out.push((col, v));
391    }
392    out
393}
394
395#[cfg(test)]
396mod tests {
397    use super::{deserialize_pk, serialize_pk};
398    use crate::change::constraint_matches;
399    use crate::value::{owned_row, ColId, OwnedValue};
400
401    #[test]
402    fn pk_codec_round_trips() {
403        // Each PK shape must deserialize to a constraint that matches its source row
404        // (the embedded ':' and multibyte string prove the byte-length prefix holds).
405        let cases: Vec<(Vec<OwnedValue>, Vec<ColId>)> = vec![
406            (vec![OwnedValue::Int(42)], vec![0]),
407            (vec![OwnedValue::Int(-7)], vec![0]),
408            (vec![OwnedValue::str("héllo:1")], vec![0]),
409            (vec![OwnedValue::str("a"), OwnedValue::Int(2)], vec![0, 1]),
410            (vec![OwnedValue::Int(1), OwnedValue::str("x:y")], vec![0, 1]),
411        ];
412        for (vals, pk) in cases {
413            let r = owned_row(vals.clone());
414            let s = serialize_pk(&r, &pk);
415            let c = deserialize_pk(&s, &pk);
416            assert!(
417                constraint_matches(&r, &c),
418                "round-trip failed for {vals:?} -> {s:?}"
419            );
420        }
421    }
422
423    #[test]
424    fn pk_codec_distinguishes_tuples_and_types() {
425        // Length-prefixing keeps distinct tuples distinct, and type tags keep
426        // `1` (int) apart from `"1"` (str).
427        let r2 = owned_row(vec![OwnedValue::str("a"), OwnedValue::str("b")]);
428        let r1 = owned_row(vec![OwnedValue::str("ab")]);
429        assert_ne!(serialize_pk(&r2, &[0, 1]), serialize_pk(&r1, &[0]));
430
431        let int1 = owned_row(vec![OwnedValue::Int(1)]);
432        let str1 = owned_row(vec![OwnedValue::str("1")]);
433        assert_ne!(serialize_pk(&int1, &[0]), serialize_pk(&str1, &[0]));
434    }
435}
rindle/op/cap.rs

rindle/op/
cap.rs
