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

reddb_engine/query/
local_database.rs
