reddb_engine/query/
local_read.rs

1//! Single-owner SELECT over a committed native artifact. This is a local file-owner
2//! API, not a server principal/tenant session or an application transaction engine.
3#[cfg(feature = "full")]
4use crate::catalog::codec::deserialize_collection_contracts;
5use crate::catalog::enforcement::{
6    INTERNAL_EXPIRES_AT_KEY, INTERNAL_TTL_MILLIS_KEY, INTERNAL_TTL_SECONDS_KEY,
7};
8use crate::catalog::{CollectionContract, CollectionModel};
9use crate::storage::{entity::EntityData, error::StoreError};
10#[cfg(feature = "full")]
11use crate::storage::{manager::ManagerConfig, resident_file::EmbeddedResidentStore};
12#[cfg(feature = "full")]
13use reddb_file::VfsFile;
14use std::collections::HashMap;
15#[cfg(feature = "full")]
16use std::path::PathBuf;
17
18use super::{
19    evaluator::EvalError,
20    planning::build_table_plan_with_host,
21    runtime::{
22        projection::projected_columns,
23        table_plan::{execute_table_plan_with_host, visit_table_plan_with_host},
24    },
25    scalar::ScalarResult,
26};
27
28pub(crate) mod admission;
29pub(crate) mod host;
30pub(crate) mod source;
31
32use source::StoreView;
33
34/// Bound materialization before cloning records, independently of SQL LIMIT.
35pub const LOCAL_READ_ROWS_MAX: usize = 100_000;
36
37/// Host-owned equality restriction, applied before the caller's SQL predicates.
38/// This is a local inspection boundary, not a server authentication policy.
39pub struct ReadScope {
40    pub table: String,
41    pub field: String,
42    pub value: reddb_types::Value,
43}
44
45#[derive(Debug, PartialEq, Eq)]
46pub struct StreamQueryResult {
47    pub columns: Vec<String>,
48    pub rows: u64,
49    pub complete: bool,
50}
51
52#[derive(Debug)]
53pub enum LocalReadError {
54    Storage(StoreError),
55    Catalog(String),
56    Parse(reddb_rql::ParseError),
57    Unsupported(&'static str),
58    Evaluation(EvalError),
59    Query(String),
60    Parameters(super::parameters::UserParamError),
61    ResourceLimit,
62    Vault(&'static str),
63}
64
65impl LocalReadError {
66    pub fn code(&self) -> &'static str {
67        match self {
68            Self::Vault(code) => code,
69            Self::Storage(_) => "DATABASE_ERROR",
70            Self::Catalog(_) => "CATALOG_ERROR",
71            Self::Parse(_) => "QUERY_PARSE_ERROR",
72            Self::Unsupported(_) => "QUERY_UNSUPPORTED",
73            Self::Evaluation(_) | Self::Query(_) => "QUERY_EVALUATION_ERROR",
74            Self::ResourceLimit => "QUERY_RESOURCE_LIMIT",
75            Self::Parameters(_) => "INVALID_PARAMS",
76        }
77    }
78}
79
80impl std::fmt::Display for LocalReadError {
81    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
82        match self {
83            Self::Vault(code) => formatter.write_str(code),
84            Self::Storage(error) => error.fmt(formatter),
85            Self::Catalog(message) | Self::Query(message) => formatter.write_str(message),
86            Self::Parse(error) => error.fmt(formatter),
87            Self::Unsupported(message) => formatter.write_str(message),
88            Self::Evaluation(error) => error.fmt(formatter),
89            Self::Parameters(error) => error.fmt(formatter),
90            Self::ResourceLimit => {
91                formatter.write_str("local SELECT materialization exceeds 100000 rows")
92            }
93        }
94    }
95}
96impl std::error::Error for LocalReadError {}
97impl From<StoreError> for LocalReadError {
98    fn from(error: StoreError) -> Self {
99        Self::Storage(error)
100    }
101}
102impl From<EvalError> for LocalReadError {
103    fn from(error: EvalError) -> Self {
104        Self::Evaluation(error)
105    }
106}
107
108/// Holds the exclusively owned file and reconstructs snapshot + committed WAL once.
109/// Callers retain the file lock until this value is dropped. No mutator is exposed.
110#[cfg(feature = "full")]
111pub struct LocalReadDatabase<F: VfsFile> {
112    store: EmbeddedResidentStore<F>,
113    contracts: HashMap<String, CollectionContract>,
114}
115
116#[cfg(feature = "full")]
117impl<F: VfsFile> LocalReadDatabase<F> {
118    pub fn open(file: F, name: impl Into<PathBuf>) -> Result<Self, LocalReadError> {
119        let store = EmbeddedResidentStore::open(file, name, ManagerConfig::default(), false, 0)?;
120        let metadata = store.committed_resident()?.aux_metadata();
121        let mut contracts = HashMap::new();
122        if !metadata.is_empty() {
123            for contract in deserialize_collection_contracts(&metadata)
124                .map_err(|error| LocalReadError::Catalog(error.to_string()))?
125            {
126                let name = contract.name.clone();
127                if contracts.insert(name.clone(), contract).is_some() {
128                    return Err(LocalReadError::Catalog(format!(
129                        "duplicate collection contract: {name}"
130                    )));
131                }
132            }
133        }
134        Ok(Self { store, contracts })
135    }
136
137    pub fn query(&self, sql: &str) -> Result<ScalarResult, LocalReadError> {
138        query_resident(self.store.committed_resident()?, &self.contracts, sql)
139    }
140}
141
142#[cfg(feature = "full")]
143pub(crate) fn query_resident(
144    resident: &dyn StoreView,
145    contracts: &HashMap<String, CollectionContract>,
146    sql: &str,
147) -> Result<ScalarResult, LocalReadError> {
148    let query = admission::query(sql)?;
149    execute_resident_query(resident, contracts, query, None)
150}
151
152pub(crate) fn query_resident_bound(
153    resident: &dyn StoreView,
154    contracts: &HashMap<String, CollectionContract>,
155    query: reddb_rql::ast::TableQuery,
156) -> Result<ScalarResult, LocalReadError> {
157    execute_resident_query(resident, contracts, admission::table(query)?, None)
158}
159
160#[cfg(feature = "full")]
161pub(crate) fn query_resident_scoped(
162    resident: &dyn StoreView,
163    contracts: &HashMap<String, CollectionContract>,
164    query: reddb_rql::ast::TableQuery,
165    scope: &ReadScope,
166) -> Result<ScalarResult, LocalReadError> {
167    execute_resident_query(resident, contracts, admission::table(query)?, Some(scope))
168}
169
170pub(crate) fn query_store_stream(
171    store: &dyn StoreView,
172    contracts: &HashMap<String, CollectionContract>,
173    query: reddb_rql::ast::TableQuery,
174    visit: &mut dyn FnMut(&[reddb_types::Value]) -> Result<bool, LocalReadError>,
175) -> Result<StreamQueryResult, LocalReadError> {
176    let query = admission::table(query)?;
177    let contract = contracts.get(&query.table).ok_or_else(|| {
178        LocalReadError::Catalog(format!(
179            "no persisted collection contract for {}",
180            query.table
181        ))
182    })?;
183    if contract.declared_model != CollectionModel::Table
184        || contract.default_ttl_ms.is_some()
185        || contract.retention_duration_ms.is_some()
186    {
187        return Err(LocalReadError::Unsupported(
188            "local SELECT currently requires a table without TTL or retention",
189        ));
190    }
191    let columns = crate::catalog::schema::resolved_contract_columns(contract)
192        .map_err(|error| LocalReadError::Catalog(error.to_string()))?;
193    admission::expressions(&query, &columns)?;
194    let schema = std::sync::Arc::new(columns.iter().map(|column| column.name.clone()).collect());
195    let source = source::StoreRows {
196        store,
197        table: &query.table,
198        schema,
199        allow_materialization: false,
200    };
201    let projections = reddb_rql::sql_lowering::effective_table_projections(&query);
202    if projections
203        .iter()
204        .any(|projection| matches!(projection, reddb_rql::ast::Projection::All))
205    {
206        return Err(LocalReadError::Unsupported(
207            "streaming SELECT requires explicit output columns",
208        ));
209    }
210    let names = projected_columns(&[], &projections);
211    let host = host::ReadHost::with_source(&query, &columns, &projections, &source);
212    let plan = build_table_plan_with_host(&host, &query);
213    let mut rows = 0u64;
214    let mut complete = true;
215    visit_table_plan_with_host(&host, &plan, &query, &mut |record| {
216        let row = names
217            .iter()
218            .map(|name| {
219                record
220                    .get(name)
221                    .cloned()
222                    .unwrap_or(reddb_types::Value::Null)
223            })
224            .collect::<Vec<_>>();
225        rows = rows.checked_add(1).ok_or(LocalReadError::ResourceLimit)?;
226        if !visit(&row)? {
227            complete = false;
228            return Ok(false);
229        }
230        Ok(true)
231    })?;
232    Ok(StreamQueryResult {
233        columns: names,
234        rows,
235        complete,
236    })
237}
238
239fn execute_resident_query(
240    resident: &dyn StoreView,
241    contracts: &HashMap<String, CollectionContract>,
242    query: reddb_rql::ast::TableQuery,
243    scope: Option<&ReadScope>,
244) -> Result<ScalarResult, LocalReadError> {
245    let contract = contracts.get(&query.table).ok_or_else(|| {
246        LocalReadError::Catalog(format!(
247            "no persisted collection contract for {}",
248            query.table
249        ))
250    })?;
251    if contract.declared_model != CollectionModel::Table
252        || contract.default_ttl_ms.is_some()
253        || contract.retention_duration_ms.is_some()
254    {
255        return Err(LocalReadError::Unsupported(
256            "local SELECT currently requires a table without TTL or retention",
257        ));
258    }
259    let columns = crate::catalog::schema::resolved_contract_columns(contract)
260        .map_err(|error| LocalReadError::Catalog(error.to_string()))?;
261    admission::expressions(&query, &columns)?;
262    if scope.is_some_and(|scope| !columns.iter().any(|column| column.name == scope.field)) {
263        return Err(LocalReadError::Catalog(
264            "inspection scope column is missing".into(),
265        ));
266    }
267    let schema = std::sync::Arc::new(columns.iter().map(|column| column.name.clone()).collect());
268    let mut entities = Vec::new();
269    let mut exceeded = false;
270    resident.scan_collection(&query.table, &mut |entity, metadata| {
271        if let Some(scope) = scope {
272            let Some(record) = super::records::runtime_table_record_from_entity_ref_with_schema(
273                entity,
274                Some(&schema),
275            ) else {
276                return true;
277            };
278            if record.get(&scope.field) != Some(&scope.value) {
279                return true;
280            }
281        }
282        if entities.len() == LOCAL_READ_ROWS_MAX {
283            exceeded = true;
284            return false;
285        }
286        let has_ttl = metadata.is_some_and(|metadata| {
287            metadata.has(INTERNAL_TTL_SECONDS_KEY)
288                || metadata.has(INTERNAL_TTL_MILLIS_KEY)
289                || metadata.has(INTERNAL_EXPIRES_AT_KEY)
290        });
291        entities.push((entity.clone(), has_ttl));
292        true
293    })?;
294    if exceeded {
295        return Err(LocalReadError::ResourceLimit);
296    }
297    let mut records = Vec::with_capacity(entities.len());
298    for (entity, has_ttl) in entities {
299        if !matches!(entity.data, EntityData::Row(_)) {
300            return Err(LocalReadError::Unsupported(
301                "local table contains non-row entities",
302            ));
303        }
304        if has_ttl {
305            return Err(LocalReadError::Unsupported(
306                "local SELECT does not yet admit row TTL metadata",
307            ));
308        }
309        if let Some(record) =
310            super::records::runtime_table_record_from_entity_ref_with_schema(&entity, Some(&schema))
311        {
312            records.push(record);
313        }
314    }
315    let projections = reddb_rql::sql_lowering::effective_table_projections(&query);
316    let host = host::ReadHost::new(&query, &columns, &projections, records);
317    let plan = build_table_plan_with_host(&host, &query);
318    let records = execute_table_plan_with_host(&host, &plan, &query)?;
319    let columns = projected_columns(&records, &projections);
320    let rows = records
321        .iter()
322        .map(|record| {
323            columns
324                .iter()
325                .map(|column| {
326                    record
327                        .get(column)
328                        .cloned()
329                        .unwrap_or(reddb_types::Value::Null)
330                })
331                .collect()
332        })
333        .collect();
334    Ok(ScalarResult { columns, rows })
335}
reddb_engine/query/local_read.rs

reddb_engine/query/
local_read.rs
