linkrs_storage/edge/node_group/
iter.rs

1//! Cross-group physical scan iterator and its constructors.
2//!
3//! Chains every existing group in group order, translating local rows back
4//! to global vertex ids. Missing groups are absent and never visited. The
5//! scan is physical: it yields every stored entry including tombstones, and
6//! visibility is decided above this layer by the version authority, never
7//! from row stamps here.
8
9use linkrs_core::types::{Timestamp, VertexId};
10use std::collections::BTreeMap;
11
12use super::super::csr_variant::{CsrIterator, CsrRowIter};
13use super::super::Nbr;
14use super::{group_base, CsrShardSet, Shard};
15
16/// Iterator chaining every existing group in group order, translating local
17/// rows to global vertex ids. Missing groups are absent and never visited.
18///
19/// Physical scan: yields every stored entry including tombstones. Callers
20/// apply the version authority to decide visibility.
21pub struct ShardCsrIterator<'a> {
22    shards: &'a BTreeMap<usize, Shard>,
23    order: Vec<usize>,
24    group_bits: u32,
25    group_pos: usize,
26    base: u32,
27    inner: CsrIterator<'a>,
28}
29
30impl<'a> ShardCsrIterator<'a> {
31    fn new(shards: &'a BTreeMap<usize, Shard>, group_bits: u32) -> Self {
32        Self {
33            shards,
34            order: shards.keys().copied().collect(),
35            group_bits,
36            group_pos: 0,
37            base: 0,
38            inner: CsrIterator::None,
39        }
40    }
41
42    fn advance_group(&mut self) -> bool {
43        if self.group_pos >= self.order.len() {
44            return false;
45        }
46        let gid = self.order[self.group_pos];
47        self.group_pos += 1;
48        self.base = group_base(gid, self.group_bits);
49        let Some(shard) = self.shards.get(&gid) else {
50            return self.advance_group();
51        };
52        self.inner = shard.variant.iter_all();
53        true
54    }
55}
56
57impl<'a> Iterator for ShardCsrIterator<'a> {
58    type Item = (VertexId, Nbr);
59
60    fn next(&mut self) -> Option<Self::Item> {
61        loop {
62            if let Some((local, nbr)) = self.inner.next() {
63                let Some(local) = local.as_internal_u32() else {
64                    continue;
65                };
66                let global = local.saturating_add(self.base);
67                return Some((VertexId::from_u32(global), nbr));
68            }
69            if !self.advance_group() {
70                return None;
71            }
72        }
73    }
74}
75
76impl CsrShardSet {
77    /// Iterate edges of a vertex without allocating, for every strategy.
78    /// Test-only row-stamp filtered iterator; production scans go through the version authority.
79    pub fn iter_edges_of(&self, src_vid: u32, ts: Timestamp) -> Option<CsrRowIter<'_>> {
80        let (gid, local) = self.route(src_vid)?;
81        self.shards.get(&gid)?.variant.iter_edges_of(local, ts)
82    }
83
84    /// Iterate every physically present entry across groups, including
85    /// tombstoned ones, in group order. Visibility is decided above this
86    /// layer by the version authority.
87    pub fn iter_all(&self) -> ShardCsrIterator<'_> {
88        ShardCsrIterator::new(&self.shards, self.group_bits)
89    }
90}
linkrs_storage/edge/node_group/iter.rs

linkrs_storage/edge/node_group/
iter.rs
