nodedb_lite/engine/columnar/store/
flush.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Memtable segment writes and collection flushing.
4
5use super::segments::{columnar_err_to_lite, store_segment_bytes};
6use super::state::{ColumnarEngine, SegmentMeta};
7use crate::error::LiteError;
8use crate::runtime::now_millis_i64;
9use crate::storage::engine::StorageEngine;
10use nodedb_columnar::writer::SegmentWriter;
11use nodedb_types::Namespace;
12use nodedb_types::columnar::ColumnarProfile;
13
14impl<S: StorageEngine> ColumnarEngine<S> {
15    /// Flush the memtable for a collection to a new segment.
16    pub async fn flush_collection(&self, collection: &str) -> Result<(), LiteError> {
17        let state_arc = self.lookup(collection)?;
18
19        // Drain memtable + collect everything we need under the inner lock.
20        struct FlushPayload {
21            segment_id: u32,
22            segment_bytes: Vec<u8>,
23            meta_key: String,
24            meta_bytes: Vec<u8>,
25            del_ops: Vec<(String, Vec<u8>)>,
26        }
27
28        let payload = {
29            let mut s = Self::lock_state(&state_arc)?;
30            if s.mutation.memtable().is_empty() {
31                return Ok(());
32            }
33
34            let segment_id = s.next_segment_id;
35            s.next_segment_id += 1;
36
37            let (schema, columns, row_count) = s.mutation.memtable_mut().drain_optimized();
38
39            let profile_tag = match &s.profile {
40                ColumnarProfile::Plain => 0,
41                ColumnarProfile::Timeseries { .. } => 1,
42                ColumnarProfile::Spatial { .. } => 2,
43            };
44
45            let writer = SegmentWriter::new(profile_tag, self.memory.clone());
46            let segment_bytes = writer
47                .write_segment(&schema, &columns, row_count, None)
48                .map_err(columnar_err_to_lite)?;
49
50            let system_time_from_ms = if s.bitemporal { now_millis_i64() } else { 0 };
51            s.segments.push(SegmentMeta {
52                segment_id,
53                row_count: row_count as u64,
54                system_time_from_ms,
55                fully_deleted_at_ms: None,
56            });
57            let meta_key = format!("{collection}:meta");
58            let meta_bytes =
59                zerompk::to_msgpack_vec(&s.segments).map_err(|e| LiteError::Serialization {
60                    detail: e.to_string(),
61                })?;
62
63            s.mutation
64                .on_memtable_flushed(segment_id as u64)
65                .map_err(|e| LiteError::Storage {
66                    detail: format!("on_memtable_flushed: {e}"),
67                })?;
68
69            let mut del_ops: Vec<(String, Vec<u8>)> = Vec::new();
70            for (&seg_id, bitmap) in s.mutation.delete_bitmaps() {
71                if !bitmap.is_empty() {
72                    let del_key = format!("{collection}:del:{seg_id}");
73                    let del_bytes = bitmap.to_bytes().map_err(columnar_err_to_lite)?;
74                    del_ops.push((del_key, del_bytes));
75                }
76            }
77
78            FlushPayload {
79                segment_id,
80                segment_bytes,
81                meta_key,
82                meta_bytes,
83                del_ops,
84            }
85        };
86
87        // Storage I/O with lock dropped.
88        // Large segment bytes go through the segment ext (or KV fallback).
89        store_segment_bytes(
90            &*self.storage,
91            collection,
92            payload.segment_id,
93            &payload.segment_bytes,
94        )
95        .await?;
96
97        // Small B+ tree entries: segment metadata list and delete bitmaps.
98        self.storage
99            .put(
100                Namespace::Columnar,
101                payload.meta_key.as_bytes(),
102                &payload.meta_bytes,
103            )
104            .await?;
105        for (del_key, del_bytes) in &payload.del_ops {
106            self.storage
107                .put(Namespace::Columnar, del_key.as_bytes(), del_bytes)
108                .await?;
109        }
110
111        Ok(())
112    }
113
114    /// Flush all collections' memtables.
115    pub async fn flush_all(&self) -> Result<(), LiteError> {
116        let names: Vec<String> = self
117            .collections
118            .read()
119            .map_err(|_| LiteError::LockPoisoned)?
120            .keys()
121            .cloned()
122            .collect();
123        for name in names {
124            self.flush_collection(&name).await?;
125        }
126        Ok(())
127    }
128}
nodedb_lite/engine/columnar/store/flush.rs

nodedb_lite/engine/columnar/store/
flush.rs
