reddb_engine/query/
local_database.rs

1//! Single-owner local SQL. Native formats and publication barriers remain canonical.
2use super::{
3    local_read::{query_resident_bound, query_resident_scoped, LocalReadError, ReadScope},
4    scalar::ScalarResult,
5};
6use crate::{
7    catalog::{
8        codec::{deserialize_collection_contracts, serialize_collection_contracts},
9        CollectionContract,
10    },
11    storage::{
12        manager::ManagerConfig, resident_file::EmbeddedResidentStore,
13        resident_store::ResidentStore, store_wal::StoreWalAction,
14    },
15};
16use reddb_file::VfsFile;
17use reddb_rql::ast::{QueryExpr, TxnControl};
18use reddb_types::Value;
19use std::{
20    collections::{HashMap, HashSet},
21    path::PathBuf,
22};
23
24mod mutation;
25#[cfg(not(target_arch = "wasm32"))]
26pub mod service;
27pub mod vault;
28
29type Result<T> = std::result::Result<T, LocalReadError>;
30
31/// Same positional-parameter count bound as the native wire protocol.
32pub const MAX_QUERY_PARAMETERS: usize = 65_536;
33
34struct Transaction {
35    resident: ResidentStore,
36    contracts: HashMap<String, CollectionContract>,
37    actions: Vec<StoreWalAction>,
38    catalog_changed: bool,
39    aborted: bool,
40}
41
42/// One exclusively locked database handle is the transaction/session boundary.
43/// Closing drops uncommitted state. Failed I/O fences the controller until reopen.
44pub struct LocalDatabase<F: VfsFile> {
45    store: EmbeddedResidentStore<F>,
46    contracts: HashMap<String, CollectionContract>,
47    transaction: Option<Transaction>,
48}
49
50impl<F: VfsFile> LocalDatabase<F> {
51    pub fn create(file: F, name: impl Into<PathBuf>, now_ms: u64) -> Result<Self> {
52        Ok(Self {
53            store: EmbeddedResidentStore::create(
54                file,
55                name,
56                u128::from(now_ms),
57                ManagerConfig::default(),
58                false,
59                0,
60            )?,
61            contracts: HashMap::new(),
62            transaction: None,
63        })
64    }
65
66    pub fn open(file: F, name: impl Into<PathBuf>) -> Result<Self> {
67        let store = EmbeddedResidentStore::open(file, name, ManagerConfig::default(), false, 0)?;
68        let bytes = store.committed_resident()?.aux_metadata();
69        let mut contracts = HashMap::new();
70        if !bytes.is_empty() {
71            for contract in deserialize_collection_contracts(&bytes)
72                .map_err(|error| LocalReadError::Catalog(error.to_string()))?
73            {
74                let name = contract.name.clone();
75                if contracts.insert(name.clone(), contract).is_some() {
76                    return Err(LocalReadError::Catalog(format!(
77                        "duplicate collection contract: {name}"
78                    )));
79                }
80            }
81        }
82        Ok(Self {
83            store,
84            contracts,
85            transaction: None,
86        })
87    }
88
89    pub fn query(&mut self, sql: &str, now_ms: u64) -> Result<ScalarResult> {
90        self.query_with_params(sql, &[], now_ms)
91    }
92
93    /// Host-side read operations must honor the same interrupted-publication fence.
94    pub fn check_healthy(&self) -> Result<()> {
95        self.store.committed_resident()?;
96        Ok(())
97    }
98
99    /// Read actual persisted contracts without inferring schemas from row payloads.
100    pub fn catalog(&self) -> Result<ScalarResult> {
101        self.check_healthy()?;
102        if self
103            .transaction
104            .as_ref()
105            .is_some_and(|transaction| transaction.aborted)
106        {
107            return Err(query_error("transaction is aborted; ROLLBACK is required"));
108        }
109        let (_, contracts) = self.visible()?;
110        let mut tables: Vec<_> = contracts.values().collect();
111        tables.sort_by(|left, right| left.name.cmp(&right.name));
112        let mut rows = Vec::new();
113        for table in tables {
114            if table.declared_model != crate::catalog::CollectionModel::Table {
115                continue;
116            }
117            for column in crate::catalog::schema::resolved_contract_columns(table)
118                .map_err(|error| LocalReadError::Catalog(error.to_string()))?
119            {
120                rows.push(vec![
121                    Value::text(table.name.as_str()),
122                    Value::text(column.name),
123                    Value::text(column.data_type_name),
124                    Value::Boolean(!column.not_null),
125                ]);
126            }
127        }
128        Ok(ScalarResult {
129            columns: ["table_name", "column_name", "data_type", "nullable"]
130                .map(String::from)
131                .to_vec(),
132            rows,
133        })
134    }
135
136    /// Only SELECT is admitted; a caller cannot widen the host's source-row scope.
137    pub fn read_query(
138        &self,
139        sql: &str,
140        parameters: &[Value],
141        scopes: &[ReadScope],
142    ) -> Result<ScalarResult> {
143        self.check_healthy()?;
144        if parameters.len() > MAX_QUERY_PARAMETERS || scopes.len() > 64 {
145            return Err(query_error("inspection arguments exceed their limit"));
146        }
147        if self
148            .transaction
149            .as_ref()
150            .is_some_and(|transaction| transaction.aborted)
151        {
152            return Err(query_error("transaction is aborted; ROLLBACK is required"));
153        }
154        let statement = reddb_rql::parse_frontend(sql)
155            .map_err(LocalReadError::Parse)?
156            .into_query_expr();
157        let statement =
158            super::parameters::bind(&statement, parameters).map_err(LocalReadError::Parameters)?;
159        let QueryExpr::Table(query) = statement else {
160            return Err(LocalReadError::Unsupported(
161                "inspection accepts table SELECT only",
162            ));
163        };
164        let mut tables = HashSet::new();
165        if scopes.iter().any(|scope| !tables.insert(&scope.table)) {
166            return Err(query_error("duplicate inspection table scope"));
167        }
168        let scope = scopes
169            .iter()
170            .find(|scope| scope.table == query.table)
171            .ok_or(LocalReadError::Unsupported(
172                "table is outside the inspection scope",
173            ))?;
174        let (resident, contracts) = self.visible()?;
175        query_resident_scoped(resident, contracts, query, scope)
176    }
177
178    pub fn query_with_params(
179        &mut self,
180        sql: &str,
181        parameters: &[Value],
182        now_ms: u64,
183    ) -> Result<ScalarResult> {
184        let result = self.query_inner(sql, parameters, now_ms);
185        if result.is_err() {
186            if let Some(transaction) = &mut self.transaction {
187                transaction.aborted = true;
188            }
189        }
190        result
191    }
192
193    fn query_inner(
194        &mut self,
195        sql: &str,
196        parameters: &[Value],
197        now_ms: u64,
198    ) -> Result<ScalarResult> {
199        // Also fence staged reads after an ambiguous durable commit.
200        self.check_healthy()?;
201        if parameters.len() > MAX_QUERY_PARAMETERS {
202            return Err(query_error("local SQL accepts at most 65536 parameters"));
203        }
204        let statement = reddb_rql::parse_frontend(sql)
205            .map_err(LocalReadError::Parse)?
206            .into_query_expr();
207        let statement =
208            super::parameters::bind(&statement, parameters).map_err(LocalReadError::Parameters)?;
209        if matches!(
210            statement,
211            QueryExpr::TransactionControl(TxnControl::Rollback)
212        ) {
213            if self.transaction.take().is_none() {
214                return Err(query_error("no active transaction"));
215            }
216            return Ok(affected(0));
217        }
218        if self
219            .transaction
220            .as_ref()
221            .is_some_and(|transaction| transaction.aborted)
222        {
223            return Err(query_error("transaction is aborted; ROLLBACK is required"));
224        }
225        match statement {
226            QueryExpr::TransactionControl(control) => self.control(control),
227            QueryExpr::Table(query) => {
228                let (resident, contracts) = self.visible()?;
229                query_resident_bound(resident, contracts, query)
230            }
231            QueryExpr::CreateTable(query) => self.create_table(&query, now_ms),
232            QueryExpr::Insert(_) | QueryExpr::Update(_) | QueryExpr::Delete(_) => {
233                let (resident, contracts) = self.visible()?;
234                let actions = mutation::plan(&statement, resident, contracts, now_ms)?;
235                let count = actions.len();
236                if let Some(transaction) = &mut self.transaction {
237                    if transaction.actions.len().checked_add(count).is_none_or(|count| count > super::local_read::LOCAL_READ_ROWS_MAX) {
238                        return Err(LocalReadError::ResourceLimit);
239                    }
240                    // An apply error aborts the transaction; provisional state is
241                    // inaccessible until rollback, including through COMMIT.
242                    apply(&transaction.resident, &actions)?;
243                    transaction.actions.extend(actions);
244                } else {
245                    self.store.publish_actions(&actions)?;
246                }
247                Ok(affected(count))
248            }
249            _ => Err(LocalReadError::Unsupported("local SQL supports table SELECT, CREATE TABLE, INSERT, UPDATE, DELETE and BEGIN/COMMIT/ROLLBACK")),
250        }
251    }
252
253    fn visible(&self) -> Result<(&ResidentStore, &HashMap<String, CollectionContract>)> {
254        match &self.transaction {
255            Some(transaction) => Ok((&transaction.resident, &transaction.contracts)),
256            None => Ok((self.store.committed_resident()?, &self.contracts)),
257        }
258    }
259
260    fn control(&mut self, control: TxnControl) -> Result<ScalarResult> {
261        match control {
262            TxnControl::Begin(None) => {
263                if self.transaction.is_some() {
264                    return Err(query_error("a transaction is already active"));
265                }
266                self.transaction = Some(Transaction {
267                    resident: copy_resident(self.store.committed_resident()?)?,
268                    contracts: self.contracts.clone(),
269                    actions: Vec::new(),
270                    catalog_changed: false,
271                    aborted: false,
272                });
273            }
274            TxnControl::Commit => {
275                let transaction = self.transaction.as_ref().ok_or_else(|| query_error("no active transaction"))?;
276                if transaction.catalog_changed {
277                    self.store.publish_snapshot(&snapshot(&transaction.resident))?;
278                } else {
279                    self.store.publish_actions(&transaction.actions)?;
280                }
281                self.contracts = self.transaction.take().expect("active transaction committed").contracts;
282            }
283            _ => return Err(LocalReadError::Unsupported("local transactions support BEGIN, COMMIT and ROLLBACK without isolation options or savepoints")),
284        }
285        Ok(affected(0))
286    }
287
288    fn create_table(
289        &mut self,
290        query: &reddb_rql::ast::CreateTableQuery,
291        now_ms: u64,
292    ) -> Result<ScalarResult> {
293        mutation::admit_create(query)?;
294        let (exists, catalog_exists) = {
295            let (resident, contracts) = self.visible()?;
296            (
297                resident.get_collection(&query.name).is_some(),
298                contracts.contains_key(&query.name),
299            )
300        };
301        if exists || catalog_exists {
302            if query.if_not_exists && catalog_exists {
303                return Ok(affected(0));
304            }
305            return Err(query_error(format!(
306                "table '{}' already exists",
307                query.name
308            )));
309        }
310        let contract =
311            crate::catalog::ddl::collection_contract_from_create_table(query, u128::from(now_ms))
312                .map_err(query_error)?;
313        crate::catalog::schema::validate_contract_expressions(&contract)
314            .map_err(|error| query_error(error.message))?;
315        let action = StoreWalAction::CreateCollection {
316            name: query.name.clone(),
317        };
318        if let Some(transaction) = &mut self.transaction {
319            apply(&transaction.resident, &[action])?;
320            transaction.contracts.insert(query.name.clone(), contract);
321            set_contracts(&transaction.resident, &transaction.contracts);
322            transaction.catalog_changed = true;
323        } else {
324            let (resident, contracts) = self.visible()?;
325            let staged = copy_resident(resident)?;
326            let mut contracts = contracts.clone();
327            apply(&staged, &[action])?;
328            contracts.insert(query.name.clone(), contract);
329            set_contracts(&staged, &contracts);
330            self.store.publish_snapshot(&snapshot(&staged))?;
331            self.contracts = contracts;
332        }
333        Ok(affected(0))
334    }
335
336    pub fn into_inner(self) -> F {
337        self.store.into_inner()
338    }
339}
340
341fn affected(count: usize) -> ScalarResult {
342    ScalarResult {
343        columns: vec!["affectedRows".into()],
344        rows: vec![vec![Value::UnsignedInteger(
345            u64::try_from(count).expect("row count fits u64"),
346        )]],
347    }
348}
349fn query_error(message: impl Into<String>) -> LocalReadError {
350    LocalReadError::Query(message.into())
351}
352fn snapshot(resident: &ResidentStore) -> Vec<u8> {
353    resident.snapshot_bytes(&HashSet::new(), &HashSet::new(), || {})
354}
355fn copy_resident(resident: &ResidentStore) -> Result<ResidentStore> {
356    ResidentStore::from_snapshot(&snapshot(resident), &ManagerConfig::default(), false, 0)
357        .map(|(resident, _)| resident)
358        .map_err(|error| query_error(error.to_string()))
359}
360fn apply(resident: &ResidentStore, actions: &[StoreWalAction]) -> Result<()> {
361    for action in actions {
362        resident.apply_replayed_action(
363            action,
364            reddb_file::STORE_VERSION_CURRENT,
365            &ManagerConfig::default(),
366        )?;
367    }
368    Ok(())
369}
370fn set_contracts(resident: &ResidentStore, contracts: &HashMap<String, CollectionContract>) {
371    let mut contracts: Vec<_> = contracts.values().cloned().collect();
372    contracts.sort_by(|left, right| left.name.cmp(&right.name));
373    resident.set_aux_metadata(serialize_collection_contracts(&contracts));
374}
reddb_engine/query/local_database.rs

reddb_engine/query/
local_database.rs
