linkrs_storage/edge/immutable_csr/
maintenance.rs

1use super::super::csr_shared::is_reclaimable_cold;
2use super::super::{EdgeId, Timestamp};
3use super::ImmutableCsr;
4use bitvec::vec::BitVec;
5
6impl ImmutableCsr {
7    /// Drop GC-eligible tombstones in place without unfreezing.
8    ///
9    /// Rows stay sorted because filtering preserves order; degrees, offsets
10    /// and the packed halves are rebuilt tight with no reserved gaps. Live
11    /// entries are untouched, so the live edge count does not change. Each
12    /// dropped entry is reported through `on_edge_removed` for tombstone
13    /// accounting above this layer.
14    pub fn compact_with_cutoff(
15        &mut self,
16        cutoff: Timestamp,
17        on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
18    ) -> usize {
19        if cutoff == Timestamp::MAX {
20            return 0;
21        }
22        let rows = self.degrees.len();
23        let valued = self.has_valued_entries();
24        let mut removed = 0usize;
25        let mut kept_hot = Vec::with_capacity(self.hot_entries.len());
26        let mut kept_cold = Vec::with_capacity(self.cold_entries.len());
27        let mut kept_values = Vec::with_capacity(self.values.len());
28        let mut kept_valid = BitVec::new();
29        let mut new_degrees = Vec::with_capacity(rows);
30        for vid in 0..rows {
31            let start = self.offsets[vid] as usize;
32            let degree = self.degrees[vid] as usize;
33            let end = start.saturating_add(degree).min(self.hot_entries.len());
34            let mut kept = 0usize;
35            for idx in start..end {
36                let cold = self.cold_entries[idx];
37                if is_reclaimable_cold(&cold, cutoff) {
38                    on_edge_removed(self.hot_entries[idx].edge_id, cold.delete_ts);
39                    removed += 1;
40                } else {
41                    kept_hot.push(self.hot_entries[idx]);
42                    kept_cold.push(cold);
43                    if valued {
44                        kept_values.push(self.values[idx]);
45                        kept_valid.push(self.valid.get(idx).map(|b| *b).unwrap_or(false));
46                    }
47                    kept += 1;
48                }
49            }
50            new_degrees.push(kept as u32);
51        }
52        self.hot_entries = kept_hot;
53        self.cold_entries = kept_cold;
54        self.degrees = new_degrees;
55        if valued {
56            self.values = kept_values;
57            self.valid = kept_valid;
58        }
59        self.rebuild_offsets();
60        removed
61    }
62
63    /// Reclaim a set of rows in one linear pass without unfreezing.
64    ///
65    /// Production replacement for repeated `compact_row` calls: one filter
66    /// over the packed halves plus one offset rebuild reclaims any number
67    /// of rows for a single table-proportional cost instead of one trailing
68    /// memmove per row. Rows outside `vids` are carried verbatim; listed
69    /// rows drop only cutoff-eligible tombstones and keep sorted order.
70    /// An empty or maximum-cutoff request compacts nothing. Reported
71    /// removals arrive in row order.
72    pub fn compact_rows_batched(
73        &mut self,
74        vids: &[u32],
75        cutoff: Timestamp,
76        on_edge_removed: &mut dyn FnMut(EdgeId, Timestamp),
77    ) -> usize {
78        if cutoff == Timestamp::MAX || vids.is_empty() {
79            return 0;
80        }
81        let rows = self.degrees.len();
82        let mut wanted = vec![false; rows];
83        let mut any = false;
84        for vid in vids {
85            let idx = *vid as usize;
86            if idx < rows && !wanted[idx] {
87                wanted[idx] = true;
88                any = true;
89            }
90        }
91        if !any {
92            return 0;
93        }
94        let mut removed = 0usize;
95        let valued = self.has_valued_entries();
96        let mut kept_hot = Vec::with_capacity(self.hot_entries.len());
97        let mut kept_cold = Vec::with_capacity(self.cold_entries.len());
98        let mut kept_values = Vec::with_capacity(self.values.len());
99        let mut kept_valid = BitVec::new();
100        let mut new_degrees = Vec::with_capacity(rows);
101        for (vid, want) in wanted.iter().enumerate().take(rows) {
102            let start = self.offsets[vid] as usize;
103            let degree = self.degrees[vid] as usize;
104            let end = start.saturating_add(degree).min(self.hot_entries.len());
105            let mut kept = 0usize;
106            for idx in start..end {
107                let cold = self.cold_entries[idx];
108                if *want && is_reclaimable_cold(&cold, cutoff) {
109                    on_edge_removed(self.hot_entries[idx].edge_id, cold.delete_ts);
110                    removed += 1;
111                } else {
112                    kept_hot.push(self.hot_entries[idx]);
113                    kept_cold.push(cold);
114                    if valued {
115                        kept_values.push(self.values[idx]);
116                        kept_valid.push(self.valid.get(idx).map(|b| *b).unwrap_or(false));
117                    }
118                    kept += 1;
119                }
120            }
121            new_degrees.push(kept as u32);
122        }
123        if removed == 0 {
124            return 0;
125        }
126        self.hot_entries = kept_hot;
127        self.cold_entries = kept_cold;
128        self.degrees = new_degrees;
129        if valued {
130            self.values = kept_values;
131            self.valid = kept_valid;
132        }
133        self.rebuild_offsets();
134        removed
135    }
136}
linkrs_storage/edge/immutable_csr/maintenance.rs

linkrs_storage/edge/immutable_csr/
maintenance.rs
