linkrs_storage/edge/
single_mutable_csr.rs

1//! Single-edge mutable CSR (the `Single` variant).
2//!
3//! Each vertex holds at most one live edge in a direct slot array, giving
4//! O(1) addressing. Serves one-to-one relationships (spouse, current
5//! employer); a second live insert into an occupied slot is rejected with a
6//! conflict error, never silently overwritten.
7//!
8//! Sparse cost: logical slots stay dense for O(1) addressing, but physical
9//! segments allocate lazily behind the `present` bitmap. Untouched sparse
10//! slots read as absent without allocating, written slots allocate their
11//! 1024-row segment on demand, and emptied segments release back to `None`.
12//! Sparse wide spans therefore pay only for materialized segments plus the
13//! bitmap, never for the full logical span. The fixed-slot addressing is
14//! kept and documented rather than replaced.
15//!
16//! Contract, unified with the table layer:
17//! - Each vertex holds at most one live edge. A second live insert into an
18//!   occupied slot is rejected with a conflict error, never silently
19//!   overwritten. Callers must delete before rebuilding.
20//! - Deletion requires the exact edge id; endpoint plus rank addressing goes
21//!   through `delete_edge_by_dst`. No wildcard edge id is supported.
22//! - `delete_edge_by_dst` deletes the single matching live entry and reports
23//!   the deleted count (0 or 1) so callers can reconcile.
24//! - Rebuilding a tombstoned slot needs no timestamp ordering at this layer;
25//!   snapshot visibility is decided by the version authority above.
26//!
27//! If concurrent writes are needed, use MutableCsr (accepts multiple edges).
28//!
29//! Layout by responsibility (`single_mutable_csr/` subdirectory):
30//! - `write` implements insert, delete, revert and rollback.
31//! - `read` implements physical reads, threshold scans and row iterators.
32//! - `maintenance` handles reclaim probes and compaction.
33//! - `persistence` handles dump and load.
34//! - `trait_impl` adapts the type to CSR traits.
35//! - `iter` holds the full-table iterator.
36
37use super::csr_shared::{grown_vertex_capacity, DEFAULT_VERTEX_CAPACITY};
38use super::{ColdStamps, HotNbr, Nbr};
39
40pub(crate) mod iter;
41pub(crate) mod maintenance;
42pub(crate) mod persistence;
43pub(crate) mod read;
44pub(crate) mod trait_impl;
45pub(crate) mod write;
46
47pub use iter::SingleMutableCsrIterator;
48
49/// Unassigned single slot: no edge id, never alive at any timestamp.
50pub(super) fn empty_slot() -> Nbr {
51    Nbr::dead_gap()
52}
53
54#[derive(Debug, Clone)]
55pub(super) struct SingleSegment {
56    pub(super) hot: [HotNbr; 1024],
57    pub(super) cold: [ColdStamps; 1024],
58}
59
60impl SingleSegment {
61    pub(super) fn fresh() -> Self {
62        Self {
63            hot: [HotNbr::dead_gap(); 1024],
64            cold: [ColdStamps::dead_gap(); 1024],
65        }
66    }
67}
68
69pub struct SingleMutableCsr {
70    pub(super) segments: Vec<Option<Box<SingleSegment>>>,
71    pub(super) present: Vec<u64>,
72    pub(super) vertex_capacity: usize,
73    pub(super) edge_count: u64,
74}
75
76impl Clone for SingleMutableCsr {
77    fn clone(&self) -> Self {
78        Self {
79            segments: self.segments.clone(),
80            present: self.present.clone(),
81            vertex_capacity: self.vertex_capacity,
82            edge_count: self.edge_count,
83        }
84    }
85}
86
87impl std::fmt::Debug for SingleMutableCsr {
88    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
89        f.debug_struct("SingleMutableCsr")
90            .field("vertex_capacity", &self.vertex_capacity())
91            .field("edge_count", &self.edge_count)
92            .field("allocated_segments", &self.allocated_segments())
93            .finish_non_exhaustive()
94    }
95}
96
97impl SingleMutableCsr {
98    pub(super) const SEGMENT_ROWS: usize = 1024;
99    pub(super) const SEGMENT_SHIFT: u32 = 10;
100    pub(super) const SEGMENT_MASK: usize = 1024 - 1;
101
102    #[inline]
103    pub(super) fn locate(vid: usize) -> (usize, usize) {
104        (vid >> Self::SEGMENT_SHIFT, vid & Self::SEGMENT_MASK)
105    }
106
107    #[inline]
108    pub(super) fn has_present(&self, idx: usize) -> bool {
109        let word = idx / 64;
110        let bit = idx % 64;
111        self.present
112            .get(word)
113            .is_some_and(|w| w & (1u64 << bit) != 0)
114    }
115
116    #[inline]
117    pub(super) fn set_present(&mut self, idx: usize, value: bool) {
118        let word = idx / 64;
119        let bit = idx % 64;
120        if self.present.len() <= word {
121            self.present.resize(word + 1, 0);
122        }
123        if value {
124            self.present[word] |= 1u64 << bit;
125        } else {
126            self.present[word] &= !(1u64 << bit);
127        }
128    }
129
130    pub(super) fn ensure_table_len(&mut self) {
131        let need_segments = self.vertex_capacity.div_ceil(Self::SEGMENT_ROWS);
132        if self.segments.len() < need_segments {
133            self.segments.resize_with(need_segments, || None);
134        }
135        // The present bitmap stays lazy: missing words read as absent through
136        // `has_present`, and `set_present` grows on demand. Pre-sizing it
137        // here would charge every sparse wide span for its full logical width.
138    }
139
140    pub(crate) fn allocated_segments(&self) -> usize {
141        self.segments.iter().filter(|seg| seg.is_some()).count()
142    }
143
144    pub(crate) fn sparse_memory_bytes(&self) -> usize {
145        self.segments.capacity() * std::mem::size_of::<Option<Box<SingleSegment>>>()
146            + self.allocated_segments()
147                * Self::SEGMENT_ROWS
148                * (std::mem::size_of::<HotNbr>() + std::mem::size_of::<ColdStamps>())
149            + self.present.capacity() * std::mem::size_of::<u64>()
150    }
151
152    pub fn new() -> Self {
153        Self::with_capacity(DEFAULT_VERTEX_CAPACITY)
154    }
155
156    pub fn with_capacity(vertex_capacity: usize) -> Self {
157        let vertex_cap = vertex_capacity.max(1);
158        let mut csr = Self {
159            segments: Vec::new(),
160            present: Vec::new(),
161            vertex_capacity: vertex_cap,
162            edge_count: 0,
163        };
164        csr.ensure_table_len();
165        csr
166    }
167
168    pub fn vertex_capacity(&self) -> usize {
169        self.vertex_capacity
170    }
171
172    pub fn edge_count(&self) -> u64 {
173        self.edge_count
174    }
175
176    pub fn resize(&mut self, new_vertex_capacity: usize) {
177        if new_vertex_capacity <= self.vertex_capacity {
178            return;
179        }
180        self.vertex_capacity = new_vertex_capacity;
181        self.ensure_table_len();
182    }
183
184    pub fn ensure_vertex_capacity(&mut self, min_capacity: usize) {
185        if min_capacity > self.vertex_capacity {
186            self.resize(grown_vertex_capacity(min_capacity));
187        }
188    }
189
190    /// Assembled slot copy at one index. Untouched sparse slots read as
191    /// absent, matching the empty-slot miss path without allocating.
192    pub(super) fn slot_at(&self, idx: usize) -> Option<Nbr> {
193        if idx >= self.vertex_capacity {
194            return None;
195        }
196        if !self.has_present(idx) {
197            return None;
198        }
199        let (seg, off) = Self::locate(idx);
200        let segment = self.segments.get(seg)?.as_ref()?;
201        let probe = Nbr::from_parts(segment.hot[off], segment.cold[off]);
202        (probe.edge_id != INVALID_EDGE_ID).then_some(probe)
203    }
204
205    pub(super) fn hot_at(&self, idx: usize) -> Option<HotNbr> {
206        if idx >= self.vertex_capacity || !self.has_present(idx) {
207            return None;
208        }
209        let (seg, off) = Self::locate(idx);
210        self.segments.get(seg)?.as_ref().map(|s| s.hot[off])
211    }
212
213    pub(super) fn cold_at(&self, idx: usize) -> Option<ColdStamps> {
214        if idx >= self.vertex_capacity || !self.has_present(idx) {
215            return None;
216        }
217        let (seg, off) = Self::locate(idx);
218        self.segments.get(seg)?.as_ref().map(|s| s.cold[off])
219    }
220
221    /// Paired writer for one slot: hot and cold stay in lockstep.
222    pub(super) fn set_slot(&mut self, idx: usize, nbr: Nbr) {
223        let (seg, off) = Self::locate(idx);
224        if seg >= self.segments.len() {
225            self.segments.resize_with(seg + 1, || None);
226        }
227        let is_absent = nbr.edge_id == INVALID_EDGE_ID;
228        {
229            let segment =
230                self.segments[seg].get_or_insert_with(|| Box::new(SingleSegment::fresh()));
231            segment.hot[off] = nbr.hot();
232            segment.cold[off] = nbr.cold();
233        }
234        self.set_present(idx, !is_absent);
235        if is_absent && self.segment_is_empty(seg) {
236            self.segments[seg] = None;
237        }
238    }
239
240    pub(super) fn segment_is_empty(&self, seg: usize) -> bool {
241        self.segments
242            .get(seg)
243            .and_then(|s| s.as_ref())
244            .is_some_and(|s| s.hot.iter().all(|h| h.edge_id == INVALID_EDGE_ID))
245    }
246
247    pub(super) fn slot_mut_or_alloc(&mut self, idx: usize) -> &mut SingleSegment {
248        let (seg, _) = Self::locate(idx);
249        if seg >= self.segments.len() {
250            self.segments.resize_with(seg + 1, || None);
251        }
252        self.segments[seg].get_or_insert_with(|| Box::new(SingleSegment::fresh()))
253    }
254
255    pub(super) fn stamp_delete(&mut self, idx: usize, ts: Timestamp) {
256        let (seg, off) = Self::locate(idx);
257        if let Some(segment) = self.segments.get_mut(seg).and_then(|s| s.as_mut()) {
258            segment.cold[off].delete_ts = ts;
259        }
260    }
261
262    pub(super) fn clear_delete(&mut self, idx: usize) {
263        let (seg, off) = Self::locate(idx);
264        if let Some(segment) = self.segments.get_mut(seg).and_then(|s| s.as_mut()) {
265            segment.cold[off].delete_ts = Timestamp::MAX;
266        }
267    }
268
269    pub(super) fn cold_is_live_at(&self, idx: usize) -> bool {
270        self.cold_at(idx).is_some_and(|cold| cold.is_live())
271    }
272
273    pub(super) fn get_edge_any_dst(&self, src: u32, ts: Timestamp) -> Option<Nbr> {
274        let probe = self.slot_at(src as usize)?;
275
276        if probe.is_alive_at(ts) {
277            Some(probe)
278        } else {
279            None
280        }
281    }
282}
283
284impl Default for SingleMutableCsr {
285    fn default() -> Self {
286        Self::new()
287    }
288}
289
290use super::{Timestamp, INVALID_EDGE_ID};
291
292#[cfg(test)]
293mod tests;
linkrs_storage/edge/single_mutable_csr.rs

linkrs_storage/edge/
single_mutable_csr.rs
