nodedb_columnar/mutation/
flush.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Post-write coordination: memtable flush.
4
5use crate::error::ColumnarError;
6use crate::pk_index::RowLocation;
7use crate::wal_record::ColumnarWalRecord;
8
9use super::engine::{MutationEngine, MutationResult};
10
11impl MutationEngine {
12    /// Notify the engine that the memtable was flushed to a new segment.
13    ///
14    /// Remaps the PK index entries of the memtable's virtual segment to
15    /// `new_segment_id` and moves the memtable's delete bitmap with them, so
16    /// a row tombstoned while in the memtable stays tombstoned in the
17    /// segment. The virtual segment id itself never changes: it names the
18    /// memtable, not a flushed segment, so no flushed segment can ever share
19    /// it and a later remap can never sweep flushed rows along.
20    ///
21    /// Returns the WAL record for the flush event, or `SegmentIdExhausted`
22    /// if the u64 segment ID counter has wrapped past its maximum.
23    pub fn on_memtable_flushed(
24        &mut self,
25        new_segment_id: u64,
26    ) -> Result<MutationResult, ColumnarError> {
27        let row_count = self.memtable_row_counter;
28
29        // Remap PK index entries from virtual memtable segment to real segment.
30        self.pk_index
31            .remap_segment(self.memtable_segment_id, |old_row| {
32                Some(RowLocation {
33                    segment_id: new_segment_id,
34                    row_index: old_row,
35                })
36            });
37        if let Some(bitmap) = self.delete_bitmaps.remove(&self.memtable_segment_id) {
38            self.delete_bitmaps.insert(new_segment_id, bitmap);
39        }
40
41        // Advance the segment ID counter with overflow protection.
42        let next = self
43            .next_segment_id
44            .checked_add(1)
45            .ok_or(ColumnarError::SegmentIdExhausted)?;
46
47        // Reset memtable tracking.
48        self.next_segment_id = next;
49        self.memtable_row_counter = 0;
50        self.memtable_surrogates.clear();
51
52        let wal = ColumnarWalRecord::MemtableFlushed {
53            collection: self.collection.clone(),
54            segment_id: new_segment_id,
55            row_count: row_count as u64,
56        };
57
58        Ok(MutationResult {
59            wal_records: vec![wal],
60        })
61    }
62}
63
64#[cfg(test)]
65mod tests {
66    use nodedb_types::columnar::{ColumnDef, ColumnType, ColumnarSchema};
67    use nodedb_types::value::Value;
68
69    use super::*;
70    use crate::pk_index::encode_pk;
71
72    fn engine() -> MutationEngine {
73        let schema = ColumnarSchema::new(vec![
74            ColumnDef::required("id", ColumnType::Int64).with_primary_key(),
75        ])
76        .expect("valid");
77        MutationEngine::new("t".into(), schema)
78    }
79
80    /// Two flushes must land in two distinct segments: the memtable keeps its
81    /// virtual id, so the second remap touches only the rows that were still
82    /// in the memtable.
83    #[test]
84    fn second_flush_leaves_first_segment_rows_in_place() {
85        let mut e = engine();
86        e.insert(&[Value::Integer(1)]).expect("insert");
87        let first = e.next_segment_id();
88        e.on_memtable_flushed(first).expect("flush 1");
89        e.insert(&[Value::Integer(2)]).expect("insert");
90        let second = e.next_segment_id();
91        e.on_memtable_flushed(second).expect("flush 2");
92
93        assert_ne!(first, second);
94        assert_eq!(e.memtable_segment_id(), 0);
95        assert_eq!(
96            e.pk_index()
97                .get(&encode_pk(&Value::Integer(1)))
98                .map(|l| l.segment_id),
99            Some(first)
100        );
101        assert_eq!(
102            e.pk_index()
103                .get(&encode_pk(&Value::Integer(2)))
104                .map(|l| l.segment_id),
105            Some(second)
106        );
107    }
108
109    /// A row tombstoned in the memtable stays tombstoned once flushed.
110    #[test]
111    fn memtable_tombstones_follow_the_flushed_segment() {
112        let mut e = engine();
113        e.insert(&[Value::Integer(1)]).expect("insert");
114        e.insert(&[Value::Integer(2)]).expect("insert");
115        e.delete(&Value::Integer(1)).expect("delete");
116        let seg = e.next_segment_id();
117        e.on_memtable_flushed(seg).expect("flush");
118
119        assert!(e.delete_bitmap(0).is_none(), "memtable bitmap moved out");
120        assert!(
121            e.delete_bitmap(seg).is_some_and(|bm| bm.is_deleted(0)),
122            "row 0 of the flushed segment is the tombstoned row"
123        );
124        assert!(!e.delete_bitmap(seg).is_some_and(|bm| bm.is_deleted(1)));
125    }
126}
nodedb_columnar/mutation/flush.rs

nodedb_columnar/mutation/
flush.rs
