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

reddb_engine/query/
local_read.rs
