nodedb_lite/engine/columnar/
truncate.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Whole-collection `TRUNCATE` on the Lite columnar engine.
4//!
5//! The collection stays registered with its schema, profile, and bitemporal
6//! flag. Segment ids keep counting from where they were, so a stale reader
7//! can never name a new segment by an old id.
8
9use nodedb_types::Namespace;
10
11use crate::error::LiteError;
12use crate::storage::engine::{StorageEngine, WriteOp};
13
14use super::store::{ColumnarEngine, SegmentMeta, remove_segment_bytes};
15
16impl<S: StorageEngine> ColumnarEngine<S> {
17    /// Remove every row of `collection`: the memtable, the primary-key
18    /// index, every delete bitmap, every bitemporal version, and every
19    /// flushed segment with its metadata. Returns the number of rows that
20    /// existed across the memtable and live segments.
21    ///
22    /// Errors when `collection` is not a columnar collection.
23    pub async fn truncate(&self, collection: &str) -> Result<usize, LiteError> {
24        let state_arc = self.lookup(collection)?;
25
26        // Drain in-memory state under the lock, then do storage I/O with the
27        // lock dropped.
28        let (segments, truncated): (Vec<SegmentMeta>, usize) = {
29            let mut s = Self::lock_state(&state_arc)?;
30            let segment_rows: u64 = s
31                .segments
32                .iter()
33                .filter(|m| m.fully_deleted_at_ms.is_none())
34                .map(|m| m.row_count)
35                .sum();
36            let truncated = segment_rows as usize + s.mutation.memtable().row_count();
37            // Lite runs no transaction undo log, so the pre-image is dropped.
38            drop(s.mutation.truncate());
39            (std::mem::take(&mut s.segments), truncated)
40        };
41
42        for seg in &segments {
43            if seg.fully_deleted_at_ms.is_none() {
44                remove_segment_bytes(&*self.storage, collection, seg.segment_id).await?;
45            }
46        }
47
48        let mut ops: Vec<WriteOp> = Vec::with_capacity(segments.len() + 1);
49        for seg in &segments {
50            ops.push(WriteOp::Delete {
51                ns: Namespace::Columnar,
52                key: format!("{collection}:del:{}", seg.segment_id).into_bytes(),
53            });
54        }
55        ops.push(WriteOp::Delete {
56            ns: Namespace::Columnar,
57            key: format!("{collection}:meta").into_bytes(),
58        });
59        self.storage.batch_write(&ops).await?;
60
61        Ok(truncated)
62    }
63}
64
65#[cfg(test)]
66mod tests {
67    use std::sync::Arc;
68
69    use nodedb_types::columnar::{ColumnDef, ColumnType, ColumnarProfile, ColumnarSchema};
70    use nodedb_types::value::Value;
71
72    use super::*;
73    use crate::PagedbStorageMem;
74    use crate::query::engine::{test_governor, test_scoped_memory};
75
76    async fn engine() -> ColumnarEngine<PagedbStorageMem> {
77        let storage = Arc::new(
78            PagedbStorageMem::open_in_memory()
79                .await
80                .expect("in-memory pagedb"),
81        );
82        let governor = test_governor();
83        let engine = ColumnarEngine::new(
84            storage,
85            test_scoped_memory(&governor, nodedb_mem::EngineId::Columnar),
86        );
87        let schema = ColumnarSchema::new(vec![
88            ColumnDef::required("id", ColumnType::Int64).with_primary_key(),
89            ColumnDef::nullable("v", ColumnType::Int64),
90        ])
91        .expect("schema");
92        engine
93            .create_collection("t", schema, ColumnarProfile::Plain, false)
94            .await
95            .expect("create");
96        engine
97    }
98
99    fn row(i: i64) -> Vec<Value> {
100        vec![Value::Integer(i), Value::Integer(i * 10)]
101    }
102
103    #[tokio::test]
104    async fn truncate_empties_memtable_and_keeps_schema() {
105        let engine = engine().await;
106        for i in 1..=3 {
107            engine.insert("t", &row(i)).expect("insert");
108        }
109        assert_eq!(engine.truncate("t").await.expect("truncate"), 3);
110        assert_eq!(engine.row_count("t"), 0);
111        assert!(engine.list_rows("t").await.expect("rows").is_empty());
112        assert!(engine.schema("t").is_some(), "schema survives");
113        engine.insert("t", &row(1)).expect("insert after truncate");
114        assert_eq!(engine.row_count("t"), 1);
115    }
116
117    #[tokio::test]
118    async fn truncate_drops_flushed_segments() {
119        let engine = engine().await;
120        for i in 1..=3 {
121            engine.insert("t", &row(i)).expect("insert");
122        }
123        engine.flush_collection("t").await.expect("flush");
124        engine.insert("t", &row(4)).expect("insert");
125        assert_eq!(engine.truncate("t").await.expect("truncate"), 4);
126        assert!(
127            engine
128                .read_segments("t")
129                .await
130                .expect("segments")
131                .is_empty()
132        );
133        assert!(engine.list_rows("t").await.expect("rows").is_empty());
134        let meta = engine
135            .storage
136            .get(Namespace::Columnar, b"t:meta")
137            .await
138            .expect("get");
139        assert!(meta.is_none(), "segment metadata is removed");
140    }
141
142    #[tokio::test]
143    async fn truncate_unknown_collection_is_an_error() {
144        let engine = engine().await;
145        assert!(engine.truncate("nope").await.is_err());
146    }
147}
nodedb_lite/engine/columnar/truncate.rs

nodedb_lite/engine/columnar/
truncate.rs
