reddb_engine/query/
local_database.rs

1//! Single-owner local SQL. Native formats and publication barriers remain canonical.
2#[cfg(feature = "full")]
3use super::local_read::{query_resident_scoped, ReadScope};
4use super::{
5    local_read::{
6        query_resident_bound, query_store_stream, source::StoreView, LocalReadError,
7        StreamQueryResult,
8    },
9    scalar::ScalarResult,
10};
11use crate::{
12    catalog::{
13        codec::{deserialize_collection_contracts, serialize_collection_contracts},
14        CollectionContract,
15    },
16    storage::store_wal::StoreWalAction,
17};
18use reddb_file::VfsFile;
19use reddb_rql::ast::{QueryExpr, TxnControl};
20use reddb_types::Value;
21use std::collections::HashMap;
22#[cfg(feature = "full")]
23use std::{collections::HashSet, path::PathBuf};
24
25mod mutation;
26#[cfg(all(feature = "full", not(target_arch = "wasm32")))]
27pub mod service;
28mod strategy;
29#[cfg(feature = "full")]
30pub mod vault;
31
32#[cfg(feature = "full")]
33use strategy::ResidentStrategy;
34use strategy::{PagedStrategy, StorageStrategy};
35
36type Result<T> = std::result::Result<T, LocalReadError>;
37
38/// Same positional-parameter count bound as the native wire protocol.
39pub const MAX_QUERY_PARAMETERS: usize = 65_536;
40
41struct Transaction<T> {
42    store: T,
43    contracts: HashMap<String, CollectionContract>,
44    actions: Vec<StoreWalAction>,
45    catalog_changed: bool,
46    aborted: bool,
47}
48
49/// One exclusively locked database handle is the transaction/session boundary.
50/// Closing drops uncommitted state. Failed I/O fences the controller until reopen.
51#[cfg(feature = "full")]
52pub struct LocalDatabase<F: VfsFile> {
53    database: Database<ResidentStrategy<F>>,
54}
55
56/// Bounded-page local SQL over the canonical `.rdb` pager and B-tree.
57/// Transactions and non-append mutations remain explicitly unsupported.
58#[doc(hidden)]
59pub struct PagedLocalDatabase<F: VfsFile + 'static> {
60    database: Database<PagedStrategy<F>>,
61}
62
63struct Database<S: StorageStrategy> {
64    storage: S,
65    contracts: HashMap<String, CollectionContract>,
66    transaction: Option<Transaction<S::Transaction>>,
67}
68
69#[cfg(feature = "full")]
70impl<F: VfsFile> LocalDatabase<F> {
71    pub fn create(file: F, name: impl Into<PathBuf>, now_ms: u64) -> Result<Self> {
72        Ok(Self {
73            database: Database::create(ResidentStrategy::create(file, name, now_ms)?),
74        })
75    }
76
77    pub fn open(file: F, name: impl Into<PathBuf>) -> Result<Self> {
78        Ok(Self {
79            database: Database::open(ResidentStrategy::open(file, name)?)?,
80        })
81    }
82
83    pub fn query(&mut self, sql: &str, now_ms: u64) -> Result<ScalarResult> {
84        self.database.query(sql, now_ms)
85    }
86
87    /// Host-side read operations honor the interrupted-publication fence.
88    pub fn check_healthy(&self) -> Result<()> {
89        self.database.check_healthy()
90    }
91
92    /// Read persisted contracts without inferring schemas from row payloads.
93    pub fn catalog(&self) -> Result<ScalarResult> {
94        self.database.catalog()
95    }
96
97    pub fn query_stream(
98        &self,
99        sql: &str,
100        parameters: &[Value],
101        visit: &mut dyn FnMut(&[Value]) -> Result<bool>,
102    ) -> Result<StreamQueryResult> {
103        self.database.query_stream(sql, parameters, visit)
104    }
105
106    /// Only SELECT is admitted; callers cannot widen the host's row scope.
107    pub fn read_query(
108        &self,
109        sql: &str,
110        parameters: &[Value],
111        scopes: &[ReadScope],
112    ) -> Result<ScalarResult> {
113        self.database.read_query(sql, parameters, scopes)
114    }
115
116    pub fn query_with_params(
117        &mut self,
118        sql: &str,
119        parameters: &[Value],
120        now_ms: u64,
121    ) -> Result<ScalarResult> {
122        self.database.query_with_params(sql, parameters, now_ms)
123    }
124
125    pub fn into_inner(self) -> F {
126        self.database.storage.into_inner()
127    }
128}
129
130impl<F: VfsFile + 'static> PagedLocalDatabase<F> {
131    pub fn create(file: F) -> Result<Self> {
132        Ok(Self {
133            database: Database::create(PagedStrategy::create(file)?),
134        })
135    }
136
137    pub fn open(file: F) -> Result<Self> {
138        Ok(Self {
139            database: Database::open(PagedStrategy::open(file)?)?,
140        })
141    }
142
143    pub fn query(&mut self, sql: &str, now_ms: u64) -> Result<ScalarResult> {
144        self.database.query(sql, now_ms)
145    }
146
147    pub fn query_with_params(
148        &mut self,
149        sql: &str,
150        parameters: &[Value],
151        now_ms: u64,
152    ) -> Result<ScalarResult> {
153        self.database.query_with_params(sql, parameters, now_ms)
154    }
155
156    pub fn catalog(&self) -> Result<ScalarResult> {
157        self.database.catalog()
158    }
159
160    pub fn query_stream(
161        &self,
162        sql: &str,
163        parameters: &[Value],
164        visit: &mut dyn FnMut(&[Value]) -> Result<bool>,
165    ) -> Result<StreamQueryResult> {
166        self.database.query_stream(sql, parameters, visit)
167    }
168
169    pub fn cache_pages(&self) -> usize {
170        self.database.storage.cache_pages()
171    }
172
173    pub fn into_inner(self) -> F {
174        self.database.storage.into_inner()
175    }
176}
177
178impl<S: StorageStrategy> Database<S> {
179    fn create(storage: S) -> Self {
180        Self {
181            storage,
182            contracts: HashMap::new(),
183            transaction: None,
184        }
185    }
186
187    fn open(storage: S) -> Result<Self> {
188        let bytes = storage.committed_view()?.aux_metadata()?;
189        let mut contracts = HashMap::new();
190        if !bytes.is_empty() {
191            for contract in deserialize_collection_contracts(&bytes)
192                .map_err(|error| LocalReadError::Catalog(error.to_string()))?
193            {
194                let name = contract.name.clone();
195                if contracts.insert(name.clone(), contract).is_some() {
196                    return Err(LocalReadError::Catalog(format!(
197                        "duplicate collection contract: {name}"
198                    )));
199                }
200            }
201        }
202        Ok(Self {
203            storage,
204            contracts,
205            transaction: None,
206        })
207    }
208
209    fn query(&mut self, sql: &str, now_ms: u64) -> Result<ScalarResult> {
210        self.query_with_params(sql, &[], now_ms)
211    }
212
213    /// Host-side read operations must honor the same interrupted-publication fence.
214    fn check_healthy(&self) -> Result<()> {
215        self.storage.committed_view()?;
216        Ok(())
217    }
218
219    /// Read actual persisted contracts without inferring schemas from row payloads.
220    fn catalog(&self) -> Result<ScalarResult> {
221        self.check_healthy()?;
222        if self
223            .transaction
224            .as_ref()
225            .is_some_and(|transaction| transaction.aborted)
226        {
227            return Err(query_error("transaction is aborted; ROLLBACK is required"));
228        }
229        let (_, contracts) = self.visible()?;
230        let mut tables: Vec<_> = contracts.values().collect();
231        tables.sort_by(|left, right| left.name.cmp(&right.name));
232        let mut rows = Vec::new();
233        for table in tables {
234            if table.declared_model != crate::catalog::CollectionModel::Table {
235                continue;
236            }
237            for column in crate::catalog::schema::resolved_contract_columns(table)
238                .map_err(|error| LocalReadError::Catalog(error.to_string()))?
239            {
240                rows.push(vec![
241                    Value::text(table.name.as_str()),
242                    Value::text(column.name),
243                    Value::text(column.data_type_name),
244                    Value::Boolean(!column.not_null),
245                ]);
246            }
247        }
248        Ok(ScalarResult {
249            columns: ["table_name", "column_name", "data_type", "nullable"]
250                .map(String::from)
251                .to_vec(),
252            rows,
253        })
254    }
255
256    /// Only SELECT is admitted; a caller cannot widen the host's source-row scope.
257    #[cfg(feature = "full")]
258    fn read_query(
259        &self,
260        sql: &str,
261        parameters: &[Value],
262        scopes: &[ReadScope],
263    ) -> Result<ScalarResult> {
264        self.check_healthy()?;
265        if parameters.len() > MAX_QUERY_PARAMETERS || scopes.len() > 64 {
266            return Err(query_error("inspection arguments exceed their limit"));
267        }
268        if self
269            .transaction
270            .as_ref()
271            .is_some_and(|transaction| transaction.aborted)
272        {
273            return Err(query_error("transaction is aborted; ROLLBACK is required"));
274        }
275        let statement = reddb_rql::parse_frontend(sql)
276            .map_err(LocalReadError::Parse)?
277            .into_query_expr();
278        let statement =
279            super::parameters::bind(&statement, parameters).map_err(LocalReadError::Parameters)?;
280        let QueryExpr::Table(query) = statement else {
281            return Err(LocalReadError::Unsupported(
282                "inspection accepts table SELECT only",
283            ));
284        };
285        let mut tables = HashSet::new();
286        if scopes.iter().any(|scope| !tables.insert(&scope.table)) {
287            return Err(query_error("duplicate inspection table scope"));
288        }
289        let scope = scopes
290            .iter()
291            .find(|scope| scope.table == query.table)
292            .ok_or(LocalReadError::Unsupported(
293                "table is outside the inspection scope",
294            ))?;
295        let (resident, contracts) = self.visible()?;
296        query_resident_scoped(resident, contracts, query, scope)
297    }
298
299    fn query_stream(
300        &self,
301        sql: &str,
302        parameters: &[Value],
303        visit: &mut dyn FnMut(&[Value]) -> Result<bool>,
304    ) -> Result<StreamQueryResult> {
305        self.check_healthy()?;
306        if parameters.len() > MAX_QUERY_PARAMETERS {
307            return Err(query_error("local SQL accepts at most 65536 parameters"));
308        }
309        if self
310            .transaction
311            .as_ref()
312            .is_some_and(|transaction| transaction.aborted)
313        {
314            return Err(query_error("transaction is aborted; ROLLBACK is required"));
315        }
316        let statement = reddb_rql::parse_frontend(sql)
317            .map_err(LocalReadError::Parse)?
318            .into_query_expr();
319        let statement =
320            super::parameters::bind(&statement, parameters).map_err(LocalReadError::Parameters)?;
321        let QueryExpr::Table(query) = statement else {
322            return Err(LocalReadError::Unsupported(
323                "streaming execution requires SELECT",
324            ));
325        };
326        let (store, contracts) = self.visible()?;
327        query_store_stream(store, contracts, query, visit)
328    }
329
330    fn query_with_params(
331        &mut self,
332        sql: &str,
333        parameters: &[Value],
334        now_ms: u64,
335    ) -> Result<ScalarResult> {
336        let result = self.query_inner(sql, parameters, now_ms);
337        if result.is_err() {
338            if let Some(transaction) = &mut self.transaction {
339                transaction.aborted = true;
340            }
341        }
342        result
343    }
344
345    fn query_inner(
346        &mut self,
347        sql: &str,
348        parameters: &[Value],
349        now_ms: u64,
350    ) -> Result<ScalarResult> {
351        // Also fence staged reads after an ambiguous durable commit.
352        self.check_healthy()?;
353        if parameters.len() > MAX_QUERY_PARAMETERS {
354            return Err(query_error("local SQL accepts at most 65536 parameters"));
355        }
356        let statement = reddb_rql::parse_frontend(sql)
357            .map_err(LocalReadError::Parse)?
358            .into_query_expr();
359        let statement =
360            super::parameters::bind(&statement, parameters).map_err(LocalReadError::Parameters)?;
361        self.storage.admit(&statement)?;
362        if matches!(
363            statement,
364            QueryExpr::TransactionControl(TxnControl::Rollback)
365        ) {
366            if self.transaction.take().is_none() {
367                return Err(query_error("no active transaction"));
368            }
369            return Ok(affected(0));
370        }
371        if self
372            .transaction
373            .as_ref()
374            .is_some_and(|transaction| transaction.aborted)
375        {
376            return Err(query_error("transaction is aborted; ROLLBACK is required"));
377        }
378        match statement {
379            QueryExpr::TransactionControl(control) => self.control(control),
380            QueryExpr::Table(query) => {
381                let (resident, contracts) = self.visible()?;
382                query_resident_bound(resident, contracts, query)
383            }
384            QueryExpr::CreateTable(query) => self.create_table(&query, now_ms),
385            QueryExpr::Insert(_) | QueryExpr::Update(_) | QueryExpr::Delete(_) => {
386                let (resident, contracts) = self.visible()?;
387                let actions = mutation::plan(&statement, resident, contracts, now_ms)?;
388                let count = actions.len();
389                if let Some(transaction) = &mut self.transaction {
390                    if transaction.actions.len().checked_add(count).is_none_or(|count| count > super::local_read::LOCAL_READ_ROWS_MAX) {
391                        return Err(LocalReadError::ResourceLimit);
392                    }
393                    // An apply error aborts the transaction; provisional state is
394                    // inaccessible until rollback, including through COMMIT.
395                    self.storage.apply_transaction(&transaction.store, &actions)?;
396                    transaction.actions.extend(actions);
397                } else {
398                    self.storage.publish_actions(&actions)?;
399                }
400                Ok(affected(count))
401            }
402            _ => Err(LocalReadError::Unsupported("local SQL supports table SELECT, CREATE TABLE, INSERT, UPDATE, DELETE and BEGIN/COMMIT/ROLLBACK")),
403        }
404    }
405
406    fn visible(&self) -> Result<(&dyn StoreView, &HashMap<String, CollectionContract>)> {
407        match &self.transaction {
408            Some(transaction) => Ok((&transaction.store, &transaction.contracts)),
409            None => Ok((self.storage.committed_view()?, &self.contracts)),
410        }
411    }
412
413    fn control(&mut self, control: TxnControl) -> Result<ScalarResult> {
414        match control {
415            TxnControl::Begin(None) => {
416                if self.transaction.is_some() {
417                    return Err(query_error("a transaction is already active"));
418                }
419                self.transaction = Some(Transaction {
420                    store: self.storage.begin_transaction()?,
421                    contracts: self.contracts.clone(),
422                    actions: Vec::new(),
423                    catalog_changed: false,
424                    aborted: false,
425                });
426            }
427            TxnControl::Commit => {
428                let transaction = self.transaction.as_ref().ok_or_else(|| query_error("no active transaction"))?;
429                let catalog = transaction
430                    .catalog_changed
431                    .then(|| catalog_bytes(&transaction.contracts));
432                self.storage.commit_transaction(
433                    &transaction.store,
434                    &transaction.actions,
435                    catalog,
436                )?;
437                self.contracts = self.transaction.take().expect("active transaction committed").contracts;
438            }
439            _ => return Err(LocalReadError::Unsupported("local transactions support BEGIN, COMMIT and ROLLBACK without isolation options or savepoints")),
440        }
441        Ok(affected(0))
442    }
443
444    fn create_table(
445        &mut self,
446        query: &reddb_rql::ast::CreateTableQuery,
447        now_ms: u64,
448    ) -> Result<ScalarResult> {
449        mutation::admit_create(query)?;
450        let (exists, catalog_exists) = {
451            let (resident, contracts) = self.visible()?;
452            (
453                resident.contains_collection(&query.name)?,
454                contracts.contains_key(&query.name),
455            )
456        };
457        if exists || catalog_exists {
458            if query.if_not_exists && catalog_exists {
459                return Ok(affected(0));
460            }
461            return Err(query_error(format!(
462                "table '{}' already exists",
463                query.name
464            )));
465        }
466        let contract =
467            crate::catalog::ddl::collection_contract_from_create_table(query, u128::from(now_ms))
468                .map_err(query_error)?;
469        crate::catalog::schema::validate_contract_expressions(&contract)
470            .map_err(|error| query_error(error.message))?;
471        let action = StoreWalAction::CreateCollection {
472            name: query.name.clone(),
473        };
474        if let Some(transaction) = &mut self.transaction {
475            self.storage
476                .apply_transaction(&transaction.store, &[action])?;
477            transaction.contracts.insert(query.name.clone(), contract);
478            transaction.catalog_changed = true;
479        } else {
480            let mut contracts = self.contracts.clone();
481            contracts.insert(query.name.clone(), contract);
482            self.storage
483                .publish_catalog_actions(&[action], catalog_bytes(&contracts))?;
484            self.contracts = contracts;
485        }
486        Ok(affected(0))
487    }
488}
489
490fn affected(count: usize) -> ScalarResult {
491    ScalarResult {
492        columns: vec!["affectedRows".into()],
493        rows: vec![vec![Value::UnsignedInteger(
494            u64::try_from(count).expect("row count fits u64"),
495        )]],
496    }
497}
498fn query_error(message: impl Into<String>) -> LocalReadError {
499    LocalReadError::Query(message.into())
500}
501fn catalog_bytes(contracts: &HashMap<String, CollectionContract>) -> Vec<u8> {
502    let mut contracts: Vec<_> = contracts.values().cloned().collect();
503    contracts.sort_by(|left, right| left.name.cmp(&right.name));
504    serialize_collection_contracts(&contracts)
505}
reddb_engine/query/local_database.rs

reddb_engine/query/
local_database.rs
