nodedb_lite/engine/columnar/store/
mutate.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Columnar row mutations and outbound replication.
4
5use super::segments::columnar_err_to_lite;
6use super::state::ColumnarEngine;
7use crate::error::LiteError;
8use crate::storage::engine::StorageEngine;
9use nodedb_types::columnar::ColumnarProfile;
10use nodedb_types::value::Value;
11#[cfg(not(target_arch = "wasm32"))]
12use std::sync::Arc;
13
14impl<S: StorageEngine> ColumnarEngine<S> {
15    /// Insert a row into a columnar collection's memtable.
16    ///
17    /// This is a pure in-memory operation. Call [`enqueue_outbound`] from
18    /// an async context after inserting to durably enqueue the row for
19    /// replication to Origin.
20    pub fn insert(&self, collection: &str, values: &[Value]) -> Result<(), LiteError> {
21        let state_arc = self.lookup(collection)?;
22        let mut s = Self::lock_state(&state_arc)?;
23        s.mutation.insert(values).map_err(columnar_err_to_lite)?;
24        Ok(())
25    }
26
27    /// Durably enqueue a batch of inserted rows for replication to Origin.
28    ///
29    /// Must be called from an async context after one or more successful
30    /// [`insert`] calls. Timeseries-profile collections are routed to the
31    /// `timeseries_outbound` queue; all other columnar collections use the
32    /// plain `outbound` queue.
33    ///
34    /// Returns [`LiteError::Backpressure`] when the queue is at cap so the
35    /// caller can propagate back-pressure. Other enqueue errors are logged as
36    /// warnings (the local insert already succeeded) and `Ok(())` is returned.
37    #[cfg(not(target_arch = "wasm32"))]
38    pub async fn enqueue_outbound(
39        &self,
40        collection: &str,
41        rows: &[Vec<Value>],
42    ) -> Result<(), LiteError> {
43        if rows.is_empty() {
44            return Ok(());
45        }
46
47        // Read the profile and schema metadata under the lock, then drop it
48        // before any await point to satisfy the no-lock-across-await rule.
49        enum OutboundRoute {
50            Timeseries { column_names: Vec<String> },
51            Columnar { schema_bytes: Vec<u8> },
52            None,
53        }
54
55        let route: OutboundRoute = {
56            let state_arc = self.lookup(collection)?;
57            let s = Self::lock_state(&state_arc)?;
58            if matches!(s.profile, ColumnarProfile::Timeseries { .. }) {
59                if self.timeseries_outbound.is_some() {
60                    let column_names: Vec<String> = s
61                        .mutation
62                        .schema()
63                        .columns
64                        .iter()
65                        .map(|c| c.name.clone())
66                        .collect();
67                    OutboundRoute::Timeseries { column_names }
68                } else {
69                    OutboundRoute::None
70                }
71            } else if self.outbound.is_some() {
72                let schema_bytes = zerompk::to_msgpack_vec(s.mutation.schema()).unwrap_or_default();
73                OutboundRoute::Columnar { schema_bytes }
74            } else {
75                OutboundRoute::None
76            }
77            // lock `s` is dropped here at end of block
78        };
79
80        match route {
81            OutboundRoute::Timeseries { column_names } => {
82                let queue = match &self.timeseries_outbound {
83                    Some(q) => Arc::clone(q),
84                    None => return Ok(()),
85                };
86                for row in rows {
87                    crate::sync::reconcile_outbound_enqueue(
88                        queue
89                            .enqueue_row(collection, column_names.clone(), row.clone())
90                            .await,
91                        "timeseries insert",
92                        collection,
93                        "",
94                    )?;
95                }
96            }
97            OutboundRoute::Columnar { schema_bytes } => {
98                let queue = match &self.outbound {
99                    Some(q) => Arc::clone(q),
100                    None => return Ok(()),
101                };
102                for row in rows {
103                    crate::sync::reconcile_outbound_enqueue(
104                        queue
105                            .enqueue_row(collection, row.clone(), schema_bytes.clone())
106                            .await,
107                        "columnar insert",
108                        collection,
109                        "",
110                    )?;
111                }
112            }
113            OutboundRoute::None => {}
114        }
115
116        Ok(())
117    }
118
119    /// Delete a row by PK.
120    pub fn delete(&self, collection: &str, pk: &Value) -> Result<bool, LiteError> {
121        let state_arc = self.lookup(collection)?;
122        let mut s = Self::lock_state(&state_arc)?;
123
124        if matches!(s.profile, ColumnarProfile::Timeseries { .. }) {
125            return Err(LiteError::BadRequest {
126                detail: format!(
127                    "DELETE not allowed on timeseries collection '{collection}' (append-only)"
128                ),
129            });
130        }
131
132        match s.mutation.delete(pk) {
133            Ok(_) => Ok(true),
134            Err(nodedb_columnar::ColumnarError::PrimaryKeyNotFound) => Ok(false),
135            Err(e) => Err(columnar_err_to_lite(e)),
136        }
137    }
138
139    /// Update a row: DELETE old + INSERT new.
140    pub fn update(
141        &self,
142        collection: &str,
143        old_pk: &Value,
144        new_values: &[Value],
145    ) -> Result<bool, LiteError> {
146        let state_arc = self.lookup(collection)?;
147        let mut s = Self::lock_state(&state_arc)?;
148
149        if matches!(s.profile, ColumnarProfile::Timeseries { .. }) {
150            return Err(LiteError::BadRequest {
151                detail: format!(
152                    "UPDATE not allowed on timeseries collection '{collection}' (append-only)"
153                ),
154            });
155        }
156
157        // Lite keeps no per-row surrogate sidecar for flushed segments.
158        match s.mutation.update(old_pk, new_values, None) {
159            Ok(_) => Ok(true),
160            Err(nodedb_columnar::ColumnarError::PrimaryKeyNotFound) => Ok(false),
161            Err(e) => Err(columnar_err_to_lite(e)),
162        }
163    }
164}
nodedb_lite/engine/columnar/store/mutate.rs

nodedb_lite/engine/columnar/store/
mutate.rs
