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.
3use crate::catalog::enforcement::{
4    INTERNAL_EXPIRES_AT_KEY, INTERNAL_TTL_MILLIS_KEY, INTERNAL_TTL_SECONDS_KEY,
5};
6use crate::catalog::{
7    codec::deserialize_collection_contracts, CollectionContract, CollectionModel,
8};
9use crate::storage::{
10    entity::EntityData, error::StoreError, manager::ManagerConfig,
11    resident_file::EmbeddedResidentStore,
12};
13use reddb_file::VfsFile;
14use std::{collections::HashMap, path::PathBuf};
15
16use super::{
17    evaluator::EvalError,
18    planning::build_table_plan_with_host,
19    runtime::{projection::projected_columns, table_plan::execute_table_plan_with_host},
20    scalar::ScalarResult,
21};
22
23pub(crate) mod admission;
24pub(crate) mod host;
25
26/// Bound materialization before cloning records, independently of SQL LIMIT.
27pub const LOCAL_READ_ROWS_MAX: usize = 100_000;
28
29/// Host-owned equality restriction, applied before the caller's SQL predicates.
30/// This is a local inspection boundary, not a server authentication policy.
31pub struct ReadScope {
32    pub table: String,
33    pub field: String,
34    pub value: reddb_types::Value,
35}
36
37#[derive(Debug)]
38pub enum LocalReadError {
39    Storage(StoreError),
40    Catalog(String),
41    Parse(reddb_rql::ParseError),
42    Unsupported(&'static str),
43    Evaluation(EvalError),
44    Query(String),
45    Parameters(super::parameters::UserParamError),
46    ResourceLimit,
47}
48
49impl LocalReadError {
50    pub fn code(&self) -> &'static str {
51        match self {
52            Self::Storage(_) => "DATABASE_ERROR",
53            Self::Catalog(_) => "CATALOG_ERROR",
54            Self::Parse(_) => "QUERY_PARSE_ERROR",
55            Self::Unsupported(_) => "QUERY_UNSUPPORTED",
56            Self::Evaluation(_) | Self::Query(_) => "QUERY_EVALUATION_ERROR",
57            Self::ResourceLimit => "QUERY_RESOURCE_LIMIT",
58            Self::Parameters(_) => "INVALID_PARAMS",
59        }
60    }
61}
62
63impl std::fmt::Display for LocalReadError {
64    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        match self {
66            Self::Storage(error) => error.fmt(formatter),
67            Self::Catalog(message) | Self::Query(message) => formatter.write_str(message),
68            Self::Parse(error) => error.fmt(formatter),
69            Self::Unsupported(message) => formatter.write_str(message),
70            Self::Evaluation(error) => error.fmt(formatter),
71            Self::Parameters(error) => error.fmt(formatter),
72            Self::ResourceLimit => {
73                formatter.write_str("local SELECT materialization exceeds 100000 rows")
74            }
75        }
76    }
77}
78impl std::error::Error for LocalReadError {}
79impl From<StoreError> for LocalReadError {
80    fn from(error: StoreError) -> Self {
81        Self::Storage(error)
82    }
83}
84impl From<EvalError> for LocalReadError {
85    fn from(error: EvalError) -> Self {
86        Self::Evaluation(error)
87    }
88}
89
90/// Holds the exclusively owned file and reconstructs snapshot + committed WAL once.
91/// Callers retain the file lock until this value is dropped. No mutator is exposed.
92pub struct LocalReadDatabase<F: VfsFile> {
93    store: EmbeddedResidentStore<F>,
94    contracts: HashMap<String, CollectionContract>,
95}
96
97impl<F: VfsFile> LocalReadDatabase<F> {
98    pub fn open(file: F, name: impl Into<PathBuf>) -> Result<Self, LocalReadError> {
99        let store = EmbeddedResidentStore::open(file, name, ManagerConfig::default(), false, 0)?;
100        let metadata = store.committed_resident()?.aux_metadata();
101        let mut contracts = HashMap::new();
102        if !metadata.is_empty() {
103            for contract in deserialize_collection_contracts(&metadata)
104                .map_err(|error| LocalReadError::Catalog(error.to_string()))?
105            {
106                let name = contract.name.clone();
107                if contracts.insert(name.clone(), contract).is_some() {
108                    return Err(LocalReadError::Catalog(format!(
109                        "duplicate collection contract: {name}"
110                    )));
111                }
112            }
113        }
114        Ok(Self { store, contracts })
115    }
116
117    pub fn query(&self, sql: &str) -> Result<ScalarResult, LocalReadError> {
118        query_resident(self.store.committed_resident()?, &self.contracts, sql)
119    }
120}
121
122pub(crate) fn query_resident(
123    resident: &crate::storage::resident_store::ResidentStore,
124    contracts: &HashMap<String, CollectionContract>,
125    sql: &str,
126) -> Result<ScalarResult, LocalReadError> {
127    let query = admission::query(sql)?;
128    execute_resident_query(resident, contracts, query, None)
129}
130
131pub(crate) fn query_resident_bound(
132    resident: &crate::storage::resident_store::ResidentStore,
133    contracts: &HashMap<String, CollectionContract>,
134    query: reddb_rql::ast::TableQuery,
135) -> Result<ScalarResult, LocalReadError> {
136    execute_resident_query(resident, contracts, admission::table(query)?, None)
137}
138
139pub(crate) fn query_resident_scoped(
140    resident: &crate::storage::resident_store::ResidentStore,
141    contracts: &HashMap<String, CollectionContract>,
142    query: reddb_rql::ast::TableQuery,
143    scope: &ReadScope,
144) -> Result<ScalarResult, LocalReadError> {
145    execute_resident_query(resident, contracts, admission::table(query)?, Some(scope))
146}
147
148fn execute_resident_query(
149    resident: &crate::storage::resident_store::ResidentStore,
150    contracts: &HashMap<String, CollectionContract>,
151    query: reddb_rql::ast::TableQuery,
152    scope: Option<&ReadScope>,
153) -> Result<ScalarResult, LocalReadError> {
154    let contract = contracts.get(&query.table).ok_or_else(|| {
155        LocalReadError::Catalog(format!(
156            "no persisted collection contract for {}",
157            query.table
158        ))
159    })?;
160    if contract.declared_model != CollectionModel::Table
161        || contract.default_ttl_ms.is_some()
162        || contract.retention_duration_ms.is_some()
163    {
164        return Err(LocalReadError::Unsupported(
165            "local SELECT currently requires a table without TTL or retention",
166        ));
167    }
168    let columns = crate::catalog::schema::resolved_contract_columns(contract)
169        .map_err(|error| LocalReadError::Catalog(error.to_string()))?;
170    admission::expressions(&query, &columns)?;
171    if scope.is_some_and(|scope| !columns.iter().any(|column| column.name == scope.field)) {
172        return Err(LocalReadError::Catalog(
173            "inspection scope column is missing".into(),
174        ));
175    }
176    let manager = resident
177        .get_collection(&query.table)
178        .ok_or_else(|| LocalReadError::Catalog(format!("collection {} is missing", query.table)))?;
179    let schema = std::sync::Arc::new(columns.iter().map(|column| column.name.clone()).collect());
180    let mut entities = Vec::new();
181    let mut exceeded = false;
182    manager.scan_for_each(None, |entity| {
183        if let Some(scope) = scope {
184            let Some(record) = super::records::runtime_table_record_from_entity_ref_with_schema(
185                entity,
186                Some(&schema),
187            ) else {
188                return true;
189            };
190            if record.get(&scope.field) != Some(&scope.value) {
191                return true;
192            }
193        }
194        if entities.len() == LOCAL_READ_ROWS_MAX {
195            exceeded = true;
196            return false;
197        }
198        entities.push(entity.clone());
199        true
200    });
201    if exceeded {
202        return Err(LocalReadError::ResourceLimit);
203    }
204    let mut records = Vec::with_capacity(entities.len());
205    for entity in entities {
206        if !matches!(entity.data, EntityData::Row(_)) {
207            return Err(LocalReadError::Unsupported(
208                "local table contains non-row entities",
209            ));
210        }
211        if manager.get_metadata(entity.id).is_some_and(|metadata| {
212            metadata.has(INTERNAL_TTL_SECONDS_KEY)
213                || metadata.has(INTERNAL_TTL_MILLIS_KEY)
214                || metadata.has(INTERNAL_EXPIRES_AT_KEY)
215        }) {
216            return Err(LocalReadError::Unsupported(
217                "local SELECT does not yet admit row TTL metadata",
218            ));
219        }
220        if let Some(record) =
221            super::records::runtime_table_record_from_entity_ref_with_schema(&entity, Some(&schema))
222        {
223            records.push(record);
224        }
225    }
226    let projections = reddb_rql::sql_lowering::effective_table_projections(&query);
227    let host = host::ReadHost::new(&query, &columns, &projections, records);
228    let plan = build_table_plan_with_host(&host, &query);
229    let records = execute_table_plan_with_host(&host, &plan, &query)?;
230    let columns = projected_columns(&records, &projections);
231    let rows = records
232        .iter()
233        .map(|record| {
234            columns
235                .iter()
236                .map(|column| {
237                    record
238                        .get(column)
239                        .cloned()
240                        .unwrap_or(reddb_types::Value::Null)
241                })
242                .collect()
243        })
244        .collect();
245    Ok(ScalarResult { columns, rows })
246}
reddb_engine/query/local_read.rs

reddb_engine/query/
local_read.rs
