nodedb_lite/engine/columnar/store/
lifecycle.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Collection restoration and engine lock access.
4
5use super::decode::rebuild_pk_from_column;
6use super::segments::load_segment_bytes;
7use super::state::{CollectionMap, CollectionState, ColumnarEngine, SegmentMeta};
8use crate::error::LiteError;
9use crate::storage::engine::StorageEngine;
10#[cfg(not(target_arch = "wasm32"))]
11use crate::sync::outbound::columnar::ColumnarOutbound;
12#[cfg(not(target_arch = "wasm32"))]
13use crate::sync::outbound::timeseries::TimeseriesOutbound;
14use nodedb_columnar::delete_bitmap::DeleteBitmap;
15use nodedb_columnar::mutation::MutationEngine;
16use nodedb_columnar::reader::SegmentReader;
17use nodedb_mem::ScopedMemory;
18use nodedb_types::Namespace;
19use nodedb_types::columnar::{ColumnarProfile, ColumnarSchema};
20use std::collections::HashMap;
21use std::sync::{Arc, Mutex, RwLock};
22
23pub(super) const META_COLUMNAR_SCHEMA_PREFIX: &str = "columnar_schema:";
24pub(super) const META_COLUMNAR_COLLECTIONS: &[u8] = b"meta:columnar_collections";
25
26impl<S: StorageEngine> ColumnarEngine<S> {
27    /// Create a new empty columnar engine.
28    pub fn new(storage: Arc<S>, memory: ScopedMemory) -> Self {
29        Self {
30            storage,
31            collections: RwLock::new(HashMap::new()),
32            #[cfg(not(target_arch = "wasm32"))]
33            outbound: None,
34            #[cfg(not(target_arch = "wasm32"))]
35            timeseries_outbound: None,
36            memory,
37        }
38    }
39
40    /// Attach a sync outbound queue for plain columnar collections.
41    ///
42    /// Must be called before any inserts if columnar sync is desired.
43    #[cfg(not(target_arch = "wasm32"))]
44    pub fn set_outbound(&mut self, outbound: Arc<ColumnarOutbound<S>>) {
45        self.outbound = Some(outbound);
46    }
47
48    /// Attach a sync outbound queue for timeseries-profile collections.
49    ///
50    /// When set, inserts into collections with `ColumnarProfile::Timeseries`
51    /// are routed here instead of `outbound`, so the transport can send them
52    /// as `TimeseriesPush` frames to Origin's timeseries engine.
53    #[cfg(not(target_arch = "wasm32"))]
54    pub fn set_timeseries_outbound(&mut self, outbound: Arc<TimeseriesOutbound<S>>) {
55        self.timeseries_outbound = Some(outbound);
56    }
57
58    /// Restore columnar collections from storage on startup.
59    pub async fn restore(storage: Arc<S>, memory: ScopedMemory) -> Result<Self, LiteError> {
60        let engine = Self::new(Arc::clone(&storage), memory);
61
62        let list_bytes = storage
63            .get(Namespace::Meta, META_COLUMNAR_COLLECTIONS)
64            .await?;
65        let names: Vec<String> = match list_bytes {
66            Some(bytes) => zerompk::from_msgpack(&bytes).map_err(|e| LiteError::Storage {
67                detail: format!("columnar collection list deserialization: {e}"),
68            })?,
69            None => Vec::new(),
70        };
71
72        let mut loaded: CollectionMap = HashMap::new();
73        for name in names {
74            let meta_key = format!("{META_COLUMNAR_SCHEMA_PREFIX}{name}");
75            #[derive(serde::Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack)]
76            struct StoredSchema {
77                schema: ColumnarSchema,
78                profile: ColumnarProfile,
79                #[serde(default)]
80                bitemporal: bool,
81            }
82            if let Some(schema_bytes) = storage.get(Namespace::Meta, meta_key.as_bytes()).await?
83                && let Ok(stored) = zerompk::from_msgpack::<StoredSchema>(&schema_bytes)
84            {
85                let seg_meta_key = format!("{name}:meta");
86                let segments: Vec<SegmentMeta> = storage
87                    .get(Namespace::Columnar, seg_meta_key.as_bytes())
88                    .await?
89                    .and_then(|b| zerompk::from_msgpack(&b).ok())
90                    .unwrap_or_default();
91
92                let next_id = segments.iter().map(|s| s.segment_id + 1).max().unwrap_or(1);
93
94                let mut mutation = MutationEngine::new(name.clone(), stored.schema.clone());
95
96                for seg_meta in &segments {
97                    // Skip fully-deleted segments — they have no physical segment file.
98                    if seg_meta.fully_deleted_at_ms.is_some() {
99                        continue;
100                    }
101
102                    if let Some(seg_bytes) =
103                        load_segment_bytes(&*storage, &name, seg_meta.segment_id).await?
104                        && let Ok(reader) = SegmentReader::open(&seg_bytes)
105                        && let Ok(pk_col) = reader.read_column(0)
106                    {
107                        rebuild_pk_from_column(&mut mutation, &pk_col, seg_meta.segment_id);
108                    }
109
110                    let del_key = format!("{name}:del:{}", seg_meta.segment_id);
111                    if let Some(del_bytes) =
112                        storage.get(Namespace::Columnar, del_key.as_bytes()).await?
113                        && let Ok(bitmap) = DeleteBitmap::from_bytes(&del_bytes)
114                    {
115                        for row_idx in bitmap.iter() {
116                            let _ = row_idx;
117                        }
118                    }
119                }
120
121                loaded.insert(
122                    name,
123                    Arc::new(Mutex::new(CollectionState {
124                        mutation,
125                        profile: stored.profile,
126                        bitemporal: stored.bitemporal,
127                        segments,
128                        next_segment_id: next_id,
129                    })),
130                );
131            }
132        }
133
134        *engine
135            .collections
136            .write()
137            .map_err(|_| LiteError::LockPoisoned)? = loaded;
138
139        // outbound is wired after restore by the caller (NodeDbLite::open_inner).
140        Ok(engine)
141    }
142
143    // -- Internal helpers --
144
145    pub(in crate::engine::columnar) fn lookup(
146        &self,
147        name: &str,
148    ) -> Result<Arc<Mutex<CollectionState>>, LiteError> {
149        let guard = self
150            .collections
151            .read()
152            .map_err(|_| LiteError::LockPoisoned)?;
153        guard
154            .get(name)
155            .cloned()
156            .ok_or_else(|| LiteError::collection_not_found("columnar", name))
157    }
158
159    pub(in crate::engine::columnar) fn lock_state<'a>(
160        state: &'a Arc<Mutex<CollectionState>>,
161    ) -> Result<std::sync::MutexGuard<'a, CollectionState>, LiteError> {
162        state.lock().map_err(|_| LiteError::LockPoisoned)
163    }
164}
nodedb_lite/engine/columnar/store/lifecycle.rs

nodedb_lite/engine/columnar/store/
lifecycle.rs
