linkrs_storage/edge/single_mutable_csr/
read.rs

1//! Read path for `SingleMutableCsr`: physical reads, threshold scans and
2//! per-vertex accessors.
3
4use super::super::csr_shared::decode_endpoint_pair;
5use super::super::{EdgeId, EdgePosition, Nbr, Timestamp, VertexId, INVALID_EDGE_ID};
6use super::SingleMutableCsr;
7
8impl SingleMutableCsr {
9    pub fn get_edge(&self, src: u32, dst: VertexId, ts: Timestamp) -> Option<Nbr> {
10        let src_idx = src as usize;
11
12        if src_idx >= self.vertex_capacity() {
13            return None;
14        }
15
16        let (dst_ep, dst_rank) = decode_endpoint_pair(dst)?;
17        let probe = self.slot_at(src_idx)?;
18
19        if !probe.is_alive_at(ts) {
20            return None;
21        }
22
23        if probe.endpoint == dst_ep && probe.rank == dst_rank {
24            Some(probe)
25        } else {
26            None
27        }
28    }
29
30    /// Locate the single slot holding `edge_id`, if any.
31    ///
32    /// Returns the row-local primary position alongside a copy, including
33    /// tombstoned slots so delete and revert callers can act on them.
34    pub fn locate_edge(&self, src: u32, edge_id: EdgeId) -> Option<(EdgePosition, Nbr)> {
35        let probe = self.slot_at(src as usize)?;
36        if probe.edge_id == INVALID_EDGE_ID || probe.edge_id != edge_id {
37            return None;
38        }
39        Some((EdgePosition::Primary { slot: 0 }, probe))
40    }
41
42    pub fn nbr_at_offset(&self, src: u32, offset: i32) -> Option<Nbr> {
43        if offset != 0 {
44            return None;
45        }
46        self.slot_at(src as usize)
47    }
48
49    pub fn get_edge_physical(&self, src: u32, dst: VertexId) -> Option<Nbr> {
50        let probe = self.slot_at(src as usize)?;
51        if probe.edge_id == INVALID_EDGE_ID || probe.delete_ts != Timestamp::MAX {
52            return None;
53        }
54        let (dst_ep, dst_rank) = decode_endpoint_pair(dst)?;
55        if probe.endpoint == dst_ep && probe.rank == dst_rank {
56            Some(probe)
57        } else {
58            None
59        }
60    }
61
62    /// Test and offline use; production scans use `fill_physical_into` or
63    /// the visitor paths instead of this allocating accessor.
64    pub fn physical_edges_of(&self, src: u32) -> Vec<Nbr> {
65        let mut out = Vec::new();
66        self.fill_physical_into(src, &mut out);
67        out
68    }
69
70    /// Fill a caller buffer with the physically stored entry of one slot.
71    ///
72    /// Same content as the allocating accessor above, without the per-vertex
73    /// allocation. Lets batch scans share one buffer across vertices.
74    pub fn fill_physical_into(&self, src: u32, out: &mut Vec<Nbr>) {
75        out.clear();
76        if let Some(probe) = self.slot_at(src as usize) {
77            if probe.edge_id != INVALID_EDGE_ID {
78                out.push(probe);
79            }
80        }
81    }
82
83    pub fn visit_physical<F>(&self, src: u32, mut f: F)
84    where
85        F: FnMut(Nbr) -> bool,
86    {
87        if let Some(probe) = self.slot_at(src as usize) {
88            if probe.edge_id != INVALID_EDGE_ID {
89                let _ = f(probe);
90            }
91        }
92    }
93
94    /// Visit the single hot half of one vertex without allocating.
95    pub fn visit_hot<F>(&self, src: u32, mut f: F)
96    where
97        F: FnMut(super::super::HotNbr) -> bool,
98    {
99        let idx = src as usize;
100        if idx >= self.vertex_capacity || !self.has_present(idx) {
101            return;
102        }
103        let (seg, off) = Self::locate(idx);
104        if let Some(segment) = self.segments.get(seg).and_then(|s| s.as_ref()) {
105            let hot = segment.hot[off];
106            if hot.edge_id != INVALID_EDGE_ID {
107                let _ = f(hot);
108            }
109        }
110    }
111
112    /// Single-slot rows hold at most one entry and are trivially ordered.
113    pub fn is_row_sorted(&self, _src: u32) -> bool {
114        true
115    }
116
117    /// Threshold visit over at most one live entry.
118    pub fn visit_threshold<F>(
119        &self,
120        src: u32,
121        lower: Option<(u32, i64)>,
122        upper: Option<(u32, i64)>,
123        mut f: F,
124    ) where
125        F: FnMut(Nbr) -> bool,
126    {
127        let Some(probe) = self.slot_at(src as usize) else {
128            return;
129        };
130        if probe.edge_id == INVALID_EDGE_ID || probe.delete_ts != Timestamp::MAX {
131            return;
132        }
133        if let Some((lo_ep, lo_rank)) = lower {
134            if (probe.endpoint, probe.rank) < (lo_ep, lo_rank) {
135                return;
136            }
137        }
138        if let Some((hi_ep, hi_rank)) = upper {
139            if (probe.endpoint, probe.rank) > (hi_ep, hi_rank) {
140                return;
141            }
142        }
143        let _ = f(probe);
144    }
145
146    /// Fill a caller buffer with the same range content as `visit_threshold`.
147    pub fn fill_threshold_into(
148        &self,
149        src: u32,
150        lower: Option<(u32, i64)>,
151        upper: Option<(u32, i64)>,
152        out: &mut Vec<Nbr>,
153    ) {
154        out.clear();
155        self.visit_threshold(src, lower, upper, |nbr| {
156            out.push(nbr);
157            true
158        });
159    }
160
161    pub fn has_physical_entries(&self, vid: u32) -> bool {
162        self.hot_at(vid as usize)
163            .is_some_and(|hot| hot.edge_id != INVALID_EDGE_ID)
164    }
165
166    pub fn primary_contains(&self, src: u32, edge_id: EdgeId) -> bool {
167        self.hot_at(src as usize)
168            .is_some_and(|hot| hot.edge_id == edge_id)
169    }
170
171    /// UNFILTERED single-predicate read for tests and offline tools only.
172    /// Production reads must use `iter_edges_of` with the authority gate
173    /// instead of collecting through this allocating accessor.
174    pub fn edges_of(&self, src: u32, ts: Timestamp) -> Vec<Nbr> {
175        self.iter_edges_of(src, ts).into_iter().collect()
176    }
177
178    /// Read the single slot of one vertex when it is alive at `ts`.
179    ///
180    /// Allocation-free counterpart of `edges_of`: no vector is built for the
181    /// one-entry row. Returns `None` for out-of-range rows and for slots
182    /// that are empty or not alive at `ts`.
183    pub fn iter_edges_of(&self, src: u32, ts: Timestamp) -> Option<Nbr> {
184        let probe = self.slot_at(src as usize)?;
185        probe.is_alive_at(ts).then_some(probe)
186    }
187}
linkrs_storage/edge/single_mutable_csr/read.rs

linkrs_storage/edge/single_mutable_csr/
read.rs
