zeppelin_embed/meta/
alive.rs

1//! Per-segment alive and tombstone sets with exact counters.
2
3use super::bitmap::DocBitmap;
4
5/// A tombstone operation referred outside the segment document-id space.
6#[derive(Clone, Copy, Debug, Eq, PartialEq)]
7pub struct AliveError {
8    document: u32,
9    row_count: u32,
10}
11
12impl AliveError {
13    /// Returns the invalid document identifier.
14    #[must_use]
15    pub const fn document(self) -> u32 {
16        self.document
17    }
18
19    /// Returns the segment row count that rejected it.
20    #[must_use]
21    pub const fn row_count(self) -> u32 {
22        self.row_count
23    }
24}
25
26impl std::fmt::Display for AliveError {
27    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
28        write!(
29            formatter,
30            "document {} is outside segment row count {}",
31            self.document, self.row_count
32        )
33    }
34}
35
36impl std::error::Error for AliveError {}
37
38/// Disjoint per-segment alive and tombstone bitmaps.
39#[derive(Clone, Debug, Eq, PartialEq)]
40pub struct AliveSet {
41    row_count: u32,
42    alive: DocBitmap,
43    tombstones: DocBitmap,
44}
45
46impl AliveSet {
47    pub(crate) fn compact_for_cache(&mut self) {
48        self.alive.compact_for_cache();
49        self.tombstones.compact_for_cache();
50    }
51
52    pub(crate) fn resident_bytes(&self) -> Option<usize> {
53        self.alive
54            .resident_bytes()?
55            .checked_add(self.tombstones.resident_bytes()?)
56    }
57
58    /// Creates a segment state in which every document is alive.
59    #[must_use]
60    pub fn new(row_count: u32) -> Self {
61        let state = Self {
62            row_count,
63            alive: DocBitmap::full(row_count),
64            tombstones: DocBitmap::new(),
65        };
66        state.debug_assert_consistent();
67        state
68    }
69
70    /// Returns the immutable segment row count.
71    #[must_use]
72    pub const fn row_count(&self) -> u32 {
73        self.row_count
74    }
75
76    /// Marks one live document as tombstoned.
77    ///
78    /// Returns `true` only for the first tombstone of that identifier.
79    pub fn tombstone(&mut self, document: u32) -> Result<bool, AliveError> {
80        if document >= self.row_count {
81            return Err(AliveError {
82                document,
83                row_count: self.row_count,
84            });
85        }
86        let changed = self.alive.remove(document);
87        if changed {
88            self.tombstones.insert(document);
89        }
90        self.debug_assert_consistent();
91        Ok(changed)
92    }
93
94    /// Returns whether a document is currently alive.
95    #[must_use]
96    pub fn is_alive(&self, document: u32) -> bool {
97        self.alive.contains(document)
98    }
99
100    /// Returns the live-document bitmap.
101    #[must_use]
102    pub fn alive_bitmap(&self) -> &DocBitmap {
103        &self.alive
104    }
105
106    /// Returns the tombstone bitmap.
107    #[must_use]
108    pub fn tombstone_bitmap(&self) -> &DocBitmap {
109        &self.tombstones
110    }
111
112    /// Returns the exact live-document count.
113    #[must_use]
114    pub fn live_count(&self) -> u64 {
115        self.alive.cardinality()
116    }
117
118    /// Returns the exact tombstone count.
119    #[must_use]
120    pub fn tombstone_count(&self) -> u64 {
121        self.tombstones.cardinality()
122    }
123
124    /// Iterates live document identifiers in ascending order.
125    pub fn iter_alive(&self) -> impl Iterator<Item = u32> + '_ {
126        self.alive.iter()
127    }
128
129    pub(crate) fn retain(&mut self, rows: &DocBitmap) {
130        let mut removed = self.alive.clone();
131        removed.subtract(rows);
132        self.tombstones.union_with(&removed);
133        self.alive.intersect_with(rows);
134    }
135
136    pub(crate) const fn scan_mask(&self) -> &roaring::RoaringBitmap {
137        self.alive.as_roaring()
138    }
139
140    /// Audits the disjoint-union counter invariant in debug builds.
141    pub fn debug_assert_consistent(&self) {
142        debug_assert_eq!(
143            self.alive.cardinality() + self.tombstones.cardinality(),
144            u64::from(self.row_count)
145        );
146        debug_assert!(self.alive.iter().all(|id| !self.tombstones.contains(id)));
147        debug_assert!(self.tombstones.iter().all(|id| id < self.row_count));
148    }
149}
150
151#[allow(clippy::expect_used)]
152#[cfg(test)]
153mod tests {
154    use super::*;
155
156    #[test]
157    fn tombstone_then_alive_iteration_skips() {
158        let mut alive = AliveSet::new(4);
159        alive.tombstone(1).expect("row exists");
160        assert_eq!(alive.iter_alive().collect::<Vec<_>>(), vec![0, 2, 3]);
161    }
162
163    #[test]
164    fn alive_counters_remain_consistent() {
165        let mut alive = AliveSet::new(4);
166        alive.tombstone(1).expect("row exists");
167        alive.tombstone(1).expect("idempotent tombstone");
168        alive.debug_assert_consistent();
169        assert_eq!(alive.tombstone_count(), 1);
170        assert_eq!(alive.live_count(), 3);
171        assert_eq!(
172            alive.tombstone_count(),
173            alive.tombstone_bitmap().cardinality()
174        );
175        assert_eq!(alive.live_count(), alive.alive_bitmap().cardinality());
176    }
177}
zeppelin_embed/meta/alive.rs

zeppelin_embed/meta/
alive.rs
