nodedb-graph 0.6.1

Shared graph engine (CSR adjacency index + traversal) for NodeDB Origin and Lite
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
// SPDX-License-Identifier: Apache-2.0

//! Surrogate-keyed, collection-scoped BFS.
//!
//! The string-keyed traversals materialize a node name for every visited node —
//! and, through `neighbors_in_collection`, one for every *edge* they walk. That
//! is the wrong currency for cross-engine work. Vector, FTS and spatial all
//! produce surrogates, so a graph result keyed on names has to be translated
//! back before it can meet them, which is exactly the per-query translation the
//! global surrogate exists to remove.
//!
//! This walk stays in the surrogate domain from end to end: it seeds from
//! surrogates, expands over the dense `u32` CSR ids, and returns a
//! [`SurrogateBitmap`] plus per-surrogate hop distances. The bitmap intersects
//! directly with any other engine's. Node names are resolved once, by the
//! caller, and only for the rows that survive fusion.

use std::collections::HashSet;

use nodedb_types::{Surrogate, SurrogateBitmap};

use crate::csr::{CsrIndex, Direction};

/// Inputs to [`CsrIndex::traverse_surrogates_in_collection`].
pub struct SurrogateBfsParams<'a> {
    /// Seed nodes as CSR-local ids. Callers holding surrogates convert with
    /// [`CsrIndex::local_id_for_surrogate`]; callers holding names use
    /// [`CsrIndex::local_id_for_node`]. Taking local ids keeps the walk
    /// independent of which currency the caller happens to have, and keeps a
    /// name-seeded walk working on an index whose surrogate bindings have not
    /// been populated.
    pub seeds: &'a [u32],
    /// Edge labels to expand over. Empty keeps every edge. Otherwise an edge
    /// whose label is any listed label passes.
    pub label_filter: &'a [&'a str],
    pub direction: Direction,
    /// Hops from the seeds. `0` returns just the addressable seeds.
    pub max_depth: usize,
    /// Cap on visited nodes. Reaching it sets [`SurrogateHops::truncated`].
    pub max_visited: usize,
    /// Only edges inserted under this collection are traversed. A collection
    /// with no edges here yields an empty result and never falls back to the
    /// merged, cross-collection view.
    pub collection: &'a str,
}

/// Outcome of a surrogate-keyed BFS.
///
/// The two result fields answer different questions and must not be conflated.
/// `distances` is the *traversal* — every node the walk reached, always
/// complete. `reached` is the *intersectable* subset: only nodes that carry a
/// surrogate can meet another engine's candidate set. Gating `distances` on the
/// surrogate binding would silently shrink a graph answer whenever bindings are
/// missing, which is a storage-state question and has nothing to do with what
/// the graph actually contains.
pub struct SurrogateHops {
    /// Reached nodes that carry a surrogate, ready to intersect with another
    /// engine's candidate set. A subset of `distances`.
    pub reached: SurrogateBitmap,
    /// `(CSR-local node id, hop distance from the nearest seed)` in discovery
    /// order — the complete reached set, so callers that rank by proximity need
    /// no second pass over the graph. Local ids resolve to names through the
    /// same index, which is why this stays complete even with no surrogates
    /// bound at all.
    pub distances: Vec<(u32, usize)>,
    /// `max_visited` cut the walk short — the result is a prefix of the
    /// reachable set, not the whole of it.
    pub truncated: bool,
    /// Reached nodes carrying no surrogate. They are traversed and reported in
    /// `distances` as normal; what they cannot do is appear in `reached`, since
    /// a node with no global identity has nothing to intersect against.
    /// Non-zero means cross-engine fusion sees less than the traversal did, and
    /// callers surface it rather than quietly narrowing the answer.
    pub unaddressable: usize,
}

impl SurrogateHops {
    fn empty() -> Self {
        Self {
            reached: SurrogateBitmap::new(),
            distances: Vec::new(),
            truncated: false,
            unaddressable: 0,
        }
    }

    /// Record a newly reached node in the traversal, and in the intersectable
    /// set when it carries a surrogate.
    fn record(&mut self, csr: &CsrIndex, local: u32, depth: usize) {
        self.distances.push((local, depth));
        let raw = csr.node_surrogate_raw(local);
        if raw == 0 {
            self.unaddressable += 1;
        } else {
            self.reached.insert(Surrogate::new(raw));
        }
    }
}

impl CsrIndex {
    /// BFS from surrogate seeds over one collection's edges, returning
    /// surrogates and hop distances.
    ///
    /// The frontier is the dense `u32` CSR id, exactly as in the durable
    /// string-keyed path; the difference is that nothing is ever converted to a
    /// name. Nodes without a surrogate stay traversable — they just cannot be
    /// reported (see [`SurrogateHops::unaddressable`]).
    ///
    /// The walk admits nodes in `(depth, node name)` order: each level's new
    /// nodes in name order. When `max_visited` cuts the walk, the admitted set
    /// is the same however the edges are stored, which lets a cluster
    /// coordinator walking the same edges across partitions admit the same
    /// nodes.
    pub fn traverse_surrogates_in_collection(
        &self,
        params: SurrogateBfsParams<'_>,
    ) -> SurrogateHops {
        let SurrogateBfsParams {
            seeds,
            label_filter,
            direction,
            max_depth,
            max_visited,
            collection,
        } = params;

        let mut hops = SurrogateHops::empty();
        let Some(collection_id) = self.collection_id(collection) else {
            return hops;
        };
        // A set of unknown labels matches no edge. Falling through to
        // unfiltered expansion would return other labels' neighbourhoods under
        // the caller's labels.
        let labels = self.label_filter(label_filter);
        if labels.keeps_none() {
            self.seed_only(seeds, &mut hops);
            return hops;
        }

        let mut visited: HashSet<u32> = HashSet::with_capacity(max_visited.min(1024));
        let mut frontier: Vec<u32> = Vec::new();
        for &local in seeds {
            if self.is_local_node(local) && visited.insert(local) {
                frontier.push(local);
            }
        }
        self.sort_by_name(&mut frontier);
        for &local in &frontier {
            hops.record(self, local, 0);
        }

        let want_out = matches!(direction, Direction::Out | Direction::Both);
        let want_in = matches!(direction, Direction::In | Direction::Both);

        // Level by level. Each level's new nodes are admitted in node-name
        // order, so under the visit cap the admitted set is the same whatever
        // order the edges are stored in, here or split across partitions.
        for depth in 0..max_depth {
            if frontier.is_empty() {
                break;
            }
            let mut candidates: Vec<u32> = Vec::new();
            let mut offered: HashSet<u32> = HashSet::new();
            for &node in &frontier {
                self.record_access(node);
                let mut neighbors: Vec<(u32, u32)> = Vec::new();
                if want_out {
                    neighbors.extend(self.iter_out_edges_raw_in(node, collection_id));
                }
                if want_in {
                    neighbors.extend(self.iter_in_edges_raw_in(node, collection_id));
                }
                for (lid, other) in neighbors {
                    if !labels.keeps(lid) || visited.contains(&other) || !offered.insert(other) {
                        continue;
                    }
                    candidates.push(other);
                }
            }
            self.sort_by_name(&mut candidates);
            let mut next: Vec<u32> = Vec::with_capacity(candidates.len());
            for other in candidates {
                if visited.len() >= max_visited {
                    hops.truncated = true;
                    return hops;
                }
                visited.insert(other);
                hops.record(self, other, depth + 1);
                self.prefetch_node(other);
                next.push(other);
            }
            frontier = next;
        }

        hops
    }

    /// Order CSR-local ids by node name.
    pub(crate) fn sort_by_name(&self, nodes: &mut [u32]) {
        nodes.sort_by(|a, b| self.node_name_checked(*a).cmp(&self.node_name_checked(*b)));
    }

    /// Record the addressable seeds and nothing else. Used when no requested
    /// edge label exists in this partition, so no expansion is possible
    /// but the seeds themselves are still legitimately reachable at depth 0.
    fn seed_only(&self, seeds: &[u32], hops: &mut SurrogateHops) {
        let mut seen: HashSet<u32> = HashSet::new();
        for &local in seeds {
            if !self.is_local_node(local) {
                continue;
            }
            if seen.insert(local) {
                hops.record(self, local, 0);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_support::test_memory;

    /// `a -knows-> b -knows-> c` in `people`, plus `a -knows-> x` in `other`.
    /// Every node gets a surrogate except `x`, which deliberately has none.
    fn seeded_csr() -> CsrIndex {
        let mut csr = CsrIndex::new(test_memory());
        for (src, dst, coll) in [
            ("a", "b", "people"),
            ("b", "c", "people"),
            ("a", "x", "other"),
        ] {
            csr.add_edge_in_collection(src, "knows", dst, coll)
                .unwrap_or_else(|e| panic!("seed edge {src}->{dst} in {coll}: {e}"));
        }
        for (node, sur) in [("a", 10u32), ("b", 20), ("c", 30)] {
            csr.set_node_surrogate(node, Surrogate::new(sur));
        }
        csr
    }

    /// Seed by name, the way a name-seeded read does.
    fn local(csr: &CsrIndex, node: &str) -> u32 {
        csr.local_id_for_node(node)
            .unwrap_or_else(|| panic!("node {node} not interned"))
    }

    fn params<'a>(seeds: &'a [u32], collection: &'a str) -> SurrogateBfsParams<'a> {
        SurrogateBfsParams {
            seeds,
            label_filter: &[],
            direction: Direction::Out,
            max_depth: 5,
            max_visited: 1000,
            collection,
        }
    }

    #[test]
    fn walk_returns_surrogates_and_hop_distances() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let hops = csr.traverse_surrogates_in_collection(params(&seeds, "people"));

        assert!(hops.reached.contains(Surrogate::new(20)));
        assert!(hops.reached.contains(Surrogate::new(30)));
        let depth_of = |node: &str| {
            let id = local(&csr, node);
            hops.distances
                .iter()
                .find(|(l, _)| *l == id)
                .map(|(_, d)| *d)
        };
        assert_eq!(depth_of("a"), Some(0), "seed sits at depth 0");
        assert_eq!(depth_of("b"), Some(1));
        assert_eq!(depth_of("c"), Some(2));
    }

    /// A surrogate-holding caller enters through the same door, without ever
    /// materializing a name.
    #[test]
    fn a_surrogate_seed_resolves_to_the_same_walk() {
        let csr = seeded_csr();
        let seeds = [csr
            .local_id_for_surrogate(Surrogate::new(10))
            .expect("surrogate 10 is bound to `a`")];
        let hops = csr.traverse_surrogates_in_collection(params(&seeds, "people"));

        assert!(hops.reached.contains(Surrogate::new(20)));
        assert!(hops.reached.contains(Surrogate::new(30)));
    }

    /// The whole point of the surrogate domain: the result meets another
    /// engine's candidate set with an intersection and no name lookup.
    #[test]
    fn reached_set_intersects_another_engines_bitmap() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let hops = csr.traverse_surrogates_in_collection(params(&seeds, "people"));

        let mut vector_hits = SurrogateBitmap::new();
        vector_hits.insert(Surrogate::new(30));
        vector_hits.insert(Surrogate::new(99));

        let both = hops.reached.intersect(&vector_hits);
        assert_eq!(both.len(), 1);
        assert!(both.contains(Surrogate::new(30)));
    }

    /// Edges of another collection are not walked, even though both live in
    /// one partition under a shared node space.
    #[test]
    fn walk_does_not_cross_a_collection_boundary() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let hops = csr.traverse_surrogates_in_collection(params(&seeds, "people"));
        // `x` is only reachable through the `other` collection, and carries no
        // surrogate — so a leak would show up as an unaddressable node.
        assert_eq!(hops.unaddressable, 0, "crossed into another collection");
        assert!(
            !hops.distances.iter().any(|&(l, _)| l == local(&csr, "x")),
            "`x` belongs to the `other` collection"
        );
    }

    /// A node with no surrogate is still traversed and still reported by name —
    /// only its cross-engine addressability is missing, and that is counted.
    #[test]
    fn surrogateless_node_is_traversed_and_counted_not_dropped() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let hops = csr.traverse_surrogates_in_collection(params(&seeds, "other"));
        assert_eq!(hops.unaddressable, 1, "`x` has no surrogate to intersect");
        assert!(
            hops.distances.iter().any(|&(l, _)| l == local(&csr, "x")),
            "the traversal itself must stay complete"
        );
        assert!(!hops.reached.contains(Surrogate::new(0)));
    }

    #[test]
    fn seed_that_is_no_node_of_this_partition_is_skipped_not_fatal() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a"), 4242];
        let hops = csr.traverse_surrogates_in_collection(params(&seeds, "people"));
        assert!(hops.reached.contains(Surrogate::new(10)));
        assert!(
            !hops.distances.iter().any(|&(l, _)| l == 4242),
            "an out-of-range seed must not enter the result"
        );
        assert_eq!(hops.unaddressable, 0, "it is absent, not unaddressable");
    }

    #[test]
    fn unknown_collection_yields_empty_not_the_merged_view() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let hops = csr.traverse_surrogates_in_collection(params(&seeds, "absent"));
        assert!(hops.reached.is_empty());
        assert!(hops.distances.is_empty());
    }

    /// An unknown label must not fall through to unfiltered expansion — that
    /// would answer with another label's neighbourhood.
    #[test]
    fn unknown_label_expands_nothing_but_keeps_the_seed() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let mut p = params(&seeds, "people");
        p.label_filter = &["never_inserted"];
        let hops = csr.traverse_surrogates_in_collection(p);
        assert!(hops.reached.contains(Surrogate::new(10)));
        assert!(!hops.reached.contains(Surrogate::new(20)));
    }

    /// `a -knows-> b -likes-> c -hates-> d` in `people`.
    fn three_label_csr() -> CsrIndex {
        let mut csr = CsrIndex::new(test_memory());
        for (src, label, dst) in [
            ("a", "knows", "b"),
            ("b", "likes", "c"),
            ("c", "hates", "d"),
        ] {
            csr.add_edge_in_collection(src, label, dst, "people")
                .unwrap_or_else(|e| panic!("seed edge {src}->{dst}: {e}"));
        }
        csr
    }

    fn reached_names<'a>(csr: &'a CsrIndex, hops: &SurrogateHops) -> Vec<&'a str> {
        let mut names: Vec<&str> = hops
            .distances
            .iter()
            .filter_map(|&(l, _)| csr.node_name_checked(l))
            .collect();
        names.sort_unstable();
        names
    }

    #[test]
    fn a_label_set_expands_over_every_listed_label() {
        let csr = three_label_csr();
        let seeds = [local(&csr, "a")];
        let mut p = params(&seeds, "people");
        p.label_filter = &["knows", "likes"];
        let hops = csr.traverse_surrogates_in_collection(p);
        assert_eq!(reached_names(&csr, &hops), vec!["a", "b", "c"]);
    }

    #[test]
    fn an_unknown_label_in_a_set_adds_nothing() {
        let csr = three_label_csr();
        let seeds = [local(&csr, "a")];
        let mut p = params(&seeds, "people");
        p.label_filter = &["knows", "never_inserted"];
        let hops = csr.traverse_surrogates_in_collection(p);
        assert_eq!(reached_names(&csr, &hops), vec!["a", "b"]);
    }

    #[test]
    fn max_depth_zero_returns_only_the_seeds() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let mut p = params(&seeds, "people");
        p.max_depth = 0;
        let hops = csr.traverse_surrogates_in_collection(p);
        assert_eq!(hops.reached.len(), 1);
        assert!(hops.reached.contains(Surrogate::new(10)));
    }

    /// Under the cap, a level admits its nodes in name order, whatever order
    /// their edges were inserted in.
    #[test]
    fn a_capped_level_admits_nodes_in_name_order() {
        let mut csr = CsrIndex::new(test_memory());
        for dst in ["z", "m", "b"] {
            csr.add_edge_in_collection("a", "knows", dst, "people")
                .unwrap_or_else(|e| panic!("seed edge a->{dst}: {e}"));
        }
        let seeds = [local(&csr, "a")];
        let mut p = params(&seeds, "people");
        p.max_visited = 3;
        let hops = csr.traverse_surrogates_in_collection(p);
        assert!(hops.truncated);
        let admitted: Vec<&str> = hops
            .distances
            .iter()
            .filter_map(|&(l, _)| csr.node_name_checked(l))
            .collect();
        assert_eq!(admitted, vec!["a", "b", "m"]);
    }

    #[test]
    fn hitting_max_visited_reports_truncation() {
        let csr = seeded_csr();
        let seeds = [local(&csr, "a")];
        let mut p = params(&seeds, "people");
        p.max_visited = 2;
        let hops = csr.traverse_surrogates_in_collection(p);
        assert!(hops.truncated, "a cut-short walk must say so");
    }
}