linkrs_storage/edge/csr_variant/
trait_impl.rs

1use linkrs_core::{StorageError, StorageResult};
2
3use super::super::{
4    EdgeId, EdgePosition, MutableCsrTrait, Nbr, Timestamp, VertexId, NO_EDGES_STORED_MSG,
5    ROW_POSITION_CROSS_VARIANT_MSG,
6};
7use super::CsrVariant;
8
9impl MutableCsrTrait for CsrVariant {
10    fn insert_edge(
11        &mut self,
12        src_vid: u32,
13        dst: VertexId,
14        edge_id: EdgeId,
15        ts: Timestamp,
16    ) -> StorageResult<()> {
17        // Capability gate first: writability decides the outcome, the match
18        // below only forwards to the concrete writable form.
19        if self.is_empty_placeholder() {
20            return Err(StorageError::invalid_operation(
21                NO_EDGES_STORED_MSG.to_string(),
22            ));
23        }
24        if !self.is_writable() {
25            return match self {
26                CsrVariant::Frozen(csr) => csr.insert_edge(src_vid, dst, edge_id, ts),
27                _ => Err(StorageError::invalid_operation(
28                    NO_EDGES_STORED_MSG.to_string(),
29                )),
30            };
31        }
32        match self {
33            CsrVariant::Multiple(csr) => csr.insert_edge(src_vid, dst, edge_id, ts),
34            CsrVariant::Single(csr) => csr.insert_edge(src_vid, dst, edge_id, ts),
35            CsrVariant::Pure(csr) => csr.insert_edge(src_vid, dst, edge_id, ts),
36            CsrVariant::Bundled(csr) => csr.insert_edge(src_vid, dst, edge_id, ts),
37            CsrVariant::Frozen(_) | CsrVariant::None { .. } => {
38                unreachable!("capability gate handles non-writable forms above")
39            }
40        }
41    }
42
43    fn delete_edge(&mut self, src_vid: u32, edge_id: EdgeId, ts: Timestamp) -> StorageResult<bool> {
44        // Same capability gate as inserts: result-returning deletes refuse
45        // read-only and placeholder forms with an error, never a silent miss.
46        // The read-only arm forwards so frozen keeps its own rejection
47        // wording; the match below only serves writable forms.
48        if self.is_empty_placeholder() {
49            return Err(StorageError::invalid_operation(
50                NO_EDGES_STORED_MSG.to_string(),
51            ));
52        }
53        if self.is_read_only() {
54            return match self {
55                CsrVariant::Frozen(csr) => csr.delete_edge(src_vid, edge_id, ts),
56                _ => Err(StorageError::invalid_operation(
57                    NO_EDGES_STORED_MSG.to_string(),
58                )),
59            };
60        }
61        match self {
62            CsrVariant::Multiple(csr) => csr.delete_edge(src_vid, edge_id, ts),
63            CsrVariant::Single(csr) => csr.delete_edge(src_vid, edge_id, ts),
64            CsrVariant::Pure(csr) => csr.delete_edge(src_vid, edge_id, ts),
65            CsrVariant::Bundled(csr) => csr.delete_edge(src_vid, edge_id, ts),
66            CsrVariant::Frozen(_) | CsrVariant::None { .. } => {
67                unreachable!("capability gate handles non-writable forms above")
68            }
69        }
70    }
71
72    fn delete_edge_by_dst(&mut self, src_vid: u32, dst: VertexId, ts: Timestamp) -> usize {
73        // Counting deletes stay silent on read-only and placeholder forms;
74        // the capability query owns that decision.
75        if self.is_read_only() || self.is_empty_placeholder() {
76            return 0;
77        }
78        dispatch!(self, delete_edge_by_dst(src_vid, dst, ts) -> 0)
79    }
80
81    fn delete_edge_by_dst_reporting(
82        &mut self,
83        src_vid: u32,
84        dst: VertexId,
85        ts: Timestamp,
86        on_deleted: &mut dyn FnMut(EdgeId),
87    ) -> usize {
88        // Same gate as the other counting deletes: read-only and placeholder
89        // forms report zero here, while result-returning deletes above refuse
90        // those forms with an error instead of a silent zero.
91        if self.is_read_only() || self.is_empty_placeholder() {
92            return 0;
93        }
94        dispatch!(self, delete_edge_by_dst_reporting(src_vid, dst, ts, on_deleted) -> 0)
95    }
96
97    fn delete_edge_by_dst_reporting_positioned(
98        &mut self,
99        src_vid: u32,
100        dst: VertexId,
101        ts: Timestamp,
102        on_deleted: &mut dyn FnMut(EdgeId, Option<EdgePosition>),
103    ) -> usize {
104        // Read-only and placeholder forms report zero directly instead of
105        // re-entering the counting path; only writable forms carry positions.
106        if self.is_read_only() || self.is_empty_placeholder() {
107            return 0;
108        }
109        match self {
110            CsrVariant::Multiple(csr) => csr.delete_edge_by_dst_reporting_positioned(
111                src_vid,
112                dst,
113                ts,
114                &mut |edge_id, position| on_deleted(edge_id, Some(position)),
115            ),
116            CsrVariant::Single(csr) => {
117                csr.delete_edge_by_dst_reporting_positioned(src_vid, dst, ts, on_deleted)
118            }
119            CsrVariant::Pure(csr) => {
120                csr.delete_edge_by_dst_reporting_positioned(src_vid, dst, ts, on_deleted)
121            }
122            CsrVariant::Bundled(csr) => {
123                csr.delete_edge_by_dst_reporting_positioned(src_vid, dst, ts, on_deleted)
124            }
125            CsrVariant::Frozen(_) | CsrVariant::None { .. } => {
126                unreachable!("capability gate handles non-writable forms above")
127            }
128        }
129    }
130
131    fn locate_edge(&self, src_vid: u32, edge_id: EdgeId) -> Option<(EdgePosition, Nbr)> {
132        match self {
133            CsrVariant::Multiple(csr) => csr.locate_edge(src_vid, edge_id),
134            CsrVariant::Single(csr) => csr.locate_edge(src_vid, edge_id),
135            CsrVariant::Pure(csr) => csr.locate_edge(src_vid, edge_id),
136            CsrVariant::Bundled(csr) => csr.locate_edge(src_vid, edge_id),
137            CsrVariant::Frozen(csr) => csr.locate_edge(src_vid, edge_id),
138            _ => None,
139        }
140    }
141
142    fn delete_edge_at_position(
143        &mut self,
144        src_vid: u32,
145        position: EdgePosition,
146        expected: EdgeId,
147        ts: Timestamp,
148    ) -> StorageResult<bool> {
149        // Positions are variant-local; forms without positional addressing
150        // refuse instead of falling back to an id scan.
151        if !self.supports_positions() {
152            return Err(StorageError::invalid_operation(
153                ROW_POSITION_CROSS_VARIANT_MSG.to_string(),
154            ));
155        }
156        match self {
157            CsrVariant::Multiple(csr) => {
158                csr.delete_edge_at_position(src_vid, position, expected, ts)
159            }
160            CsrVariant::Single(csr) => csr.delete_edge_at_position(src_vid, position, expected, ts),
161            CsrVariant::Pure(csr) => csr.delete_edge_at_position(src_vid, position, expected, ts),
162            CsrVariant::Bundled(csr) => {
163                csr.delete_edge_at_position(src_vid, position, expected, ts)
164            }
165            CsrVariant::Frozen(_) | CsrVariant::None { .. } => {
166                unreachable!("capability gate handles non-positional forms above")
167            }
168        }
169    }
170
171    fn revert_delete_at_position(
172        &mut self,
173        src_vid: u32,
174        position: EdgePosition,
175        expected: EdgeId,
176        ts: Timestamp,
177    ) -> bool {
178        if !self.supports_positions() {
179            debug_assert!(false, "{}", ROW_POSITION_CROSS_VARIANT_MSG);
180            return false;
181        }
182        match self {
183            CsrVariant::Multiple(csr) => {
184                csr.revert_delete_at_position(src_vid, position, expected, ts)
185            }
186            CsrVariant::Single(csr) => {
187                csr.revert_delete_at_position(src_vid, position, expected, ts)
188            }
189            CsrVariant::Pure(csr) => csr.revert_delete_at_position(src_vid, position, expected, ts),
190            CsrVariant::Bundled(csr) => {
191                csr.revert_delete_at_position(src_vid, position, expected, ts)
192            }
193            CsrVariant::Frozen(_) | CsrVariant::None { .. } => {
194                unreachable!("capability gate handles non-positional forms above")
195            }
196        }
197    }
198
199    fn delete_edge_by_offset(
200        &mut self,
201        src_vid: u32,
202        offset: i32,
203        ts: Timestamp,
204    ) -> StorageResult<bool> {
205        // Result-returning delete like `delete_edge`: placeholder and
206        // read-only forms are refused, with frozen keeping its own rejection
207        // wording through forwarding.
208        if self.is_empty_placeholder() {
209            return Err(StorageError::invalid_operation(
210                NO_EDGES_STORED_MSG.to_string(),
211            ));
212        }
213        if self.is_read_only() {
214            return match self {
215                CsrVariant::Frozen(csr) => csr.delete_edge_by_offset(src_vid, offset, ts),
216                _ => Err(StorageError::invalid_operation(
217                    NO_EDGES_STORED_MSG.to_string(),
218                )),
219            };
220        }
221        match self {
222            CsrVariant::Multiple(csr) => csr.delete_edge_by_offset(src_vid, offset, ts),
223            CsrVariant::Single(csr) => csr.delete_edge_by_offset(src_vid, offset, ts),
224            CsrVariant::Pure(csr) => csr.delete_edge_by_offset(src_vid, offset, ts),
225            CsrVariant::Bundled(csr) => csr.delete_edge_by_offset(src_vid, offset, ts),
226            CsrVariant::Frozen(_) | CsrVariant::None { .. } => {
227                unreachable!("capability gate handles non-writable forms above")
228            }
229        }
230    }
231
232    fn revert_delete_by_offset(&mut self, src_vid: u32, offset: i32, ts: Timestamp) -> bool {
233        dispatch!(self, revert_delete_by_offset(src_vid, offset, ts) -> false)
234    }
235
236    fn nbr_at_offset(&self, src_vid: u32, offset: i32) -> Option<Nbr> {
237        dispatch!(self, nbr_at_offset(src_vid, offset) -> None)
238    }
239
240    fn get_edge_physical(&self, src_vid: u32, dst: VertexId) -> Option<Nbr> {
241        dispatch!(self, get_edge_physical(src_vid, dst) -> None)
242    }
243
244    fn physical_edges_of(&self, src_vid: u32) -> Vec<Nbr> {
245        dispatch!(self, physical_edges_of(src_vid) -> Vec::new())
246    }
247
248    fn fill_physical_into(&self, src_vid: u32, out: &mut Vec<Nbr>) {
249        CsrVariant::fill_physical_into(self, src_vid, out)
250    }
251
252    fn has_physical_entries(&self, vid: u32) -> bool {
253        match self {
254            CsrVariant::Multiple(csr) => csr.has_physical_entries(vid),
255            CsrVariant::Single(csr) => csr.has_physical_entries(vid),
256            CsrVariant::Pure(csr) => csr.has_physical_entries(vid),
257            CsrVariant::Bundled(csr) => csr.has_physical_entries(vid),
258            CsrVariant::Frozen(csr) => csr.has_physical_entries(vid),
259            CsrVariant::None { .. } => false,
260        }
261    }
262
263    fn primary_contains(&self, src_vid: u32, edge_id: EdgeId) -> bool {
264        match self {
265            CsrVariant::Multiple(csr) => csr.primary_contains(src_vid, edge_id),
266            CsrVariant::Single(csr) => csr.primary_contains(src_vid, edge_id),
267            CsrVariant::Pure(csr) => csr.primary_contains(src_vid, edge_id),
268            CsrVariant::Bundled(csr) => csr.primary_contains(src_vid, edge_id),
269            CsrVariant::Frozen(csr) => csr.primary_contains(src_vid, edge_id),
270            CsrVariant::None { .. } => false,
271        }
272    }
273
274    fn rollback_insert(&mut self, src_vid: u32, edge_id: EdgeId) -> bool {
275        dispatch!(self, rollback_insert(src_vid, edge_id) -> false)
276    }
277
278    fn revert_delete_by_edge_id(&mut self, src_vid: u32, edge_id: EdgeId, ts: Timestamp) -> bool {
279        dispatch!(self, revert_delete_by_edge_id(src_vid, edge_id, ts) -> false)
280    }
281
282    fn get_edge(&self, src_vid: u32, dst: VertexId, ts: Timestamp) -> Option<Nbr> {
283        dispatch!(self, get_edge(src_vid, dst, ts) -> None)
284    }
285
286    fn edges_of(&self, src_vid: u32, ts: Timestamp) -> Vec<Nbr> {
287        dispatch!(self, edges_of(src_vid, ts) -> Vec::new())
288    }
289
290    fn compact_vertex_with_reporting(
291        &mut self,
292        vid: u32,
293        cutoff: Timestamp,
294        on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
295    ) -> usize {
296        // Single exemption source lives with the capability query: read-only
297        // and placeholder forms report zero here and reclaim through the
298        // batched frozen entry or a file rebuild instead.
299        if !self.supports_vertex_compact() {
300            return 0;
301        }
302        match self {
303            CsrVariant::Multiple(csr) => {
304                csr.compact_vertex_with_reporting(vid, cutoff, on_edge_removed)
305            }
306            CsrVariant::Single(csr) => {
307                csr.compact_vertex_with_reporting(vid, cutoff, on_edge_removed)
308            }
309            CsrVariant::Pure(csr) => {
310                csr.compact_vertex_with_reporting(vid, cutoff, on_edge_removed)
311            }
312            CsrVariant::Bundled(csr) => {
313                csr.compact_vertex_with_reporting(vid, cutoff, on_edge_removed)
314            }
315            CsrVariant::Frozen(_) | CsrVariant::None { .. } => 0,
316        }
317    }
318
319    fn reclaimable_count(&self, vid: u32, cutoff: Timestamp) -> usize {
320        // Per-variant probe stays unsliced: pure and bundled report hole
321        // counts, frozen reports timestamp-eligible tombstones. The capability
322        // query only gates the reporting compact entry.
323        match self {
324            CsrVariant::Multiple(csr) => csr.reclaimable_count(vid, cutoff),
325            CsrVariant::Single(csr) => csr.reclaimable_count(vid, cutoff),
326            CsrVariant::Pure(csr) => csr.reclaimable_count(vid, cutoff),
327            CsrVariant::Bundled(csr) => csr.reclaimable_count(vid, cutoff),
328            CsrVariant::Frozen(csr) => csr.reclaimable_count(vid, cutoff),
329            CsrVariant::None { .. } => 0,
330        }
331    }
332
333    fn vertex_needs_compact(&self, vid: u32, cutoff: Timestamp) -> bool {
334        self.reclaimable_count(vid, cutoff) > 0
335    }
336
337    fn vertex_census(&self, vid: u32) -> (usize, usize, usize) {
338        match self {
339            CsrVariant::Multiple(csr) => csr.vertex_census(vid),
340            CsrVariant::Single(csr) => csr.vertex_census(vid),
341            CsrVariant::Pure(csr) => csr.vertex_census(vid),
342            CsrVariant::Bundled(csr) => csr.vertex_census(vid),
343            CsrVariant::Frozen(csr) => csr.vertex_census(vid),
344            CsrVariant::None { .. } => (0, 0, 0),
345        }
346    }
347
348    fn vertex_reclaim_probe(&self, vid: u32, cutoff: Timestamp) -> (usize, usize) {
349        match self {
350            CsrVariant::Multiple(csr) => csr.vertex_reclaim_probe(vid, cutoff),
351            CsrVariant::Single(csr) => csr.vertex_reclaim_probe(vid, cutoff),
352            CsrVariant::Pure(csr) => csr.vertex_reclaim_probe(vid, cutoff),
353            CsrVariant::Bundled(csr) => csr.vertex_reclaim_probe(vid, cutoff),
354            CsrVariant::Frozen(csr) => csr.vertex_reclaim_probe(vid, cutoff),
355            CsrVariant::None { .. } => (0, 0),
356        }
357    }
358
359    fn row_gap(&self, vid: u32) -> usize {
360        match self {
361            CsrVariant::Multiple(csr) => csr.row_gap(vid),
362            CsrVariant::Pure(csr) => csr.row_gap(vid),
363            CsrVariant::Bundled(csr) => csr.row_gap(vid),
364            CsrVariant::Single(_) | CsrVariant::Frozen(_) | CsrVariant::None { .. } => 0,
365        }
366    }
367
368    fn row_density(&self, vid: u32) -> f32 {
369        match self {
370            CsrVariant::Multiple(csr) => csr.row_density(vid),
371            CsrVariant::Pure(csr) => csr.row_density(vid),
372            CsrVariant::Bundled(csr) => csr.row_density(vid),
373            CsrVariant::Single(_) | CsrVariant::Frozen(_) | CsrVariant::None { .. } => 1.0,
374        }
375    }
376
377    fn rebalance_row(&mut self, vid: u32) -> bool {
378        match self {
379            CsrVariant::Multiple(csr) => csr.rebalance_row(vid),
380            CsrVariant::Pure(csr) => csr.rebalance_row(vid),
381            CsrVariant::Bundled(csr) => csr.rebalance_row(vid),
382            CsrVariant::Single(_) | CsrVariant::Frozen(_) | CsrVariant::None { .. } => true,
383        }
384    }
385
386    fn used_memory_size(&self) -> usize {
387        match self {
388            CsrVariant::None { .. } => std::mem::size_of::<Self>(),
389            CsrVariant::Multiple(csr) => csr.used_memory_size(),
390            CsrVariant::Single(csr) => csr.used_memory_size(),
391            CsrVariant::Pure(csr) => csr.used_memory_size(),
392            CsrVariant::Bundled(csr) => csr.used_memory_size(),
393            CsrVariant::Frozen(csr) => csr.used_memory_size(),
394        }
395    }
396}
linkrs_storage/edge/csr_variant/trait_impl.rs

linkrs_storage/edge/csr_variant/
trait_impl.rs
