linkrs_storage/edge/bundled_csr/
trait_impl.rs

1use linkrs_core::types::{EdgeId, Timestamp, VertexId};
2use linkrs_core::StorageResult;
3
4use super::super::csr_trait::MutableCsrTrait;
5use super::super::{EdgePosition, Nbr};
6use super::{BundledCsr, BundledOverflowValues};
7
8impl MutableCsrTrait for BundledCsr {
9    fn insert_edge(
10        &mut self,
11        src_vid: u32,
12        dst: VertexId,
13        edge_id: EdgeId,
14        _ts: Timestamp,
15    ) -> StorageResult<()> {
16        self.insert_edge_with_value(src_vid, dst, edge_id, None)
17    }
18
19    fn delete_edge(&mut self, src_vid: u32, edge_id: EdgeId, ts: Timestamp) -> StorageResult<bool> {
20        let located = self.topology.locate_edge(src_vid, edge_id);
21        let deleted = self.topology.delete_edge(src_vid, edge_id, ts)?;
22        if deleted {
23            if let Some((position, _)) = located {
24                self.clear_value_at_position(src_vid, position);
25            }
26        }
27        Ok(deleted)
28    }
29
30    fn delete_edge_by_dst(&mut self, src_vid: u32, dst: VertexId, ts: Timestamp) -> usize {
31        let mut noop = |_: EdgeId| {};
32        self.delete_edge_by_dst_reporting(src_vid, dst, ts, &mut noop)
33    }
34
35    fn delete_edge_by_dst_reporting(
36        &mut self,
37        src_vid: u32,
38        dst: VertexId,
39        ts: Timestamp,
40        on_deleted: &mut dyn FnMut(EdgeId),
41    ) -> usize {
42        let mut positioned = |edge_id: EdgeId, position: Option<EdgePosition>| {
43            on_deleted(edge_id);
44            let _ = position;
45        };
46        self.delete_edge_by_dst_reporting_positioned(src_vid, dst, ts, &mut positioned)
47    }
48
49    fn delete_edge_by_dst_reporting_positioned(
50        &mut self,
51        src_vid: u32,
52        dst: VertexId,
53        ts: Timestamp,
54        on_deleted: &mut dyn FnMut(EdgeId, Option<EdgePosition>),
55    ) -> usize {
56        let mut stamped: Vec<(EdgeId, EdgePosition)> = Vec::new();
57        let deleted = self.topology.delete_edge_by_dst_reporting_positioned(
58            src_vid,
59            dst,
60            ts,
61            &mut |edge_id, position| {
62                if let Some(position) = position {
63                    stamped.push((edge_id, position));
64                }
65            },
66        );
67        for (edge_id, position) in &stamped {
68            self.clear_value_at_position(src_vid, *position);
69            on_deleted(*edge_id, Some(*position));
70        }
71        // Topology positions are this variant's positions, so a complete
72        // report means every stamped edge carried one.
73        debug_assert_eq!(stamped.len(), deleted);
74        deleted
75    }
76
77    fn locate_edge(&self, src_vid: u32, edge_id: EdgeId) -> Option<(EdgePosition, Nbr)> {
78        self.topology.locate_edge(src_vid, edge_id)
79    }
80
81    fn delete_edge_at_position(
82        &mut self,
83        src_vid: u32,
84        position: EdgePosition,
85        expected: EdgeId,
86        ts: Timestamp,
87    ) -> StorageResult<bool> {
88        let deleted = self
89            .topology
90            .delete_edge_at_position(src_vid, position, expected, ts)?;
91        if deleted {
92            self.clear_value_at_position(src_vid, position);
93        }
94        Ok(deleted)
95    }
96
97    fn revert_delete_at_position(
98        &mut self,
99        src_vid: u32,
100        position: EdgePosition,
101        expected: EdgeId,
102        ts: Timestamp,
103    ) -> bool {
104        if !self
105            .topology
106            .revert_delete_at_position(src_vid, position, expected, ts)
107        {
108            return false;
109        }
110        // Holes are never reused by inserts, so the stale raw word still
111        // belongs to this edge: revive its validity instead of NULLing it.
112        // Callers carrying an explicit value use
113        // `revert_delete_at_position_with_value`.
114        self.sync_primary_len();
115        self.sync_overflow_row(src_vid);
116        match position {
117            EdgePosition::Primary { slot } => {
118                if let Some(idx) = self.primary_index(src_vid, slot) {
119                    if idx < self.primary_valid.len() {
120                        self.primary_valid.set(idx, true);
121                    }
122                }
123                true
124            }
125            EdgePosition::Overflow { chunk, slot } => {
126                if let Some(chunks) = self.overflow_values.get_mut(src_vid) {
127                    if let Some(c) = chunks.get_mut(chunk as usize) {
128                        if (slot as usize) < c.len() {
129                            c.set_valid(slot as usize, true);
130                        }
131                    }
132                }
133                true
134            }
135        }
136    }
137
138    fn delete_edge_by_offset(
139        &mut self,
140        src_vid: u32,
141        offset: i32,
142        ts: Timestamp,
143    ) -> StorageResult<bool> {
144        if offset < 0 {
145            return Ok(false);
146        }
147        let position = EdgePosition::Primary {
148            slot: offset as u32,
149        };
150        let deleted = self.topology.delete_edge_by_offset(src_vid, offset, ts)?;
151        if deleted {
152            self.clear_value_at_position(src_vid, position);
153        }
154        Ok(deleted)
155    }
156
157    fn revert_delete_by_offset(&mut self, src_vid: u32, offset: i32, ts: Timestamp) -> bool {
158        // Topology slots erase the edge id on delete, so offset-only and
159        // id-only reverts cannot recover it. Use the positioned revert with
160        // the expected id and value supplied by the caller.
161        let _ = (src_vid, offset, ts);
162        false
163    }
164
165    fn nbr_at_offset(&self, src_vid: u32, offset: i32) -> Option<Nbr> {
166        self.topology.nbr_at_offset(src_vid, offset)
167    }
168
169    fn get_edge_physical(&self, src_vid: u32, dst: VertexId) -> Option<Nbr> {
170        self.topology.get_edge_physical(src_vid, dst)
171    }
172
173    fn physical_edges_of(&self, src_vid: u32) -> Vec<Nbr> {
174        self.topology.physical_edges_of(src_vid)
175    }
176
177    fn fill_physical_into(&self, src_vid: u32, out: &mut Vec<Nbr>) {
178        self.topology.fill_physical_into(src_vid, out)
179    }
180
181    fn has_physical_entries(&self, vid: u32) -> bool {
182        self.topology.has_physical_entries(vid)
183    }
184
185    fn primary_contains(&self, src_vid: u32, edge_id: EdgeId) -> bool {
186        self.topology.primary_contains(src_vid, edge_id)
187    }
188
189    fn rollback_insert(&mut self, src_vid: u32, edge_id: EdgeId) -> bool {
190        self.rollback_insert_shifted(src_vid, edge_id)
191    }
192
193    fn revert_delete_by_edge_id(&mut self, src_vid: u32, edge_id: EdgeId, ts: Timestamp) -> bool {
194        // Deleted topology slots hold the sentinel instead of the edge id,
195        // so an id-keyed scan cannot locate them. Use the positioned revert
196        // with value instead.
197        let _ = (src_vid, edge_id, ts);
198        false
199    }
200
201    fn get_edge(&self, src_vid: u32, dst: VertexId, ts: Timestamp) -> Option<Nbr> {
202        self.topology.get_edge(src_vid, dst, ts)
203    }
204
205    fn edges_of(&self, src_vid: u32, ts: Timestamp) -> Vec<Nbr> {
206        self.topology.edges_of(src_vid, ts)
207    }
208
209    fn compact_vertex_with_reporting(
210        &mut self,
211        vid: u32,
212        _cutoff: Timestamp,
213        _on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
214    ) -> usize {
215        self.compact_vertex_shifted(vid, _on_edge_removed)
216    }
217
218    fn reclaimable_count(&self, vid: u32, cutoff: Timestamp) -> usize {
219        self.topology.reclaimable_count(vid, cutoff)
220    }
221
222    fn vertex_census(&self, vid: u32) -> (usize, usize, usize) {
223        self.topology.vertex_census(vid)
224    }
225
226    fn row_gap(&self, vid: u32) -> usize {
227        self.topology.row_gap(vid)
228    }
229
230    fn row_density(&self, vid: u32) -> f32 {
231        self.topology.row_density(vid)
232    }
233
234    fn rebalance_row(&mut self, vid: u32) -> bool {
235        self.topology.rebalance_row(vid)
236    }
237
238    fn used_memory_size(&self) -> usize {
239        self.topology.used_memory_size()
240            + self.primary_values.len() * 8
241            + self.primary_valid.len().div_ceil(8)
242            + self
243                .overflow_values
244                .iter()
245                .map(|(_, chunks)| {
246                    chunks
247                        .iter()
248                        .map(BundledOverflowValues::heap_bytes)
249                        .sum::<usize>()
250                })
251                .sum::<usize>()
252            + self.overflow_values.table_bytes()
253    }
254}
linkrs_storage/edge/bundled_csr/trait_impl.rs

linkrs_storage/edge/bundled_csr/
trait_impl.rs
