reddb-io-engine 1.36.1

//! Single-owner SELECT over a committed native artifact. This is a local file-owner
//! API, not a server principal/tenant session or an application transaction engine.
#[cfg(feature = "full")]
use crate::catalog::codec::deserialize_collection_contracts;
use crate::catalog::enforcement::{
    INTERNAL_EXPIRES_AT_KEY, INTERNAL_TTL_MILLIS_KEY, INTERNAL_TTL_SECONDS_KEY,
};
use crate::catalog::{CollectionContract, CollectionModel};
use crate::storage::{entity::EntityData, error::StoreError};
#[cfg(feature = "full")]
use crate::storage::{manager::ManagerConfig, resident_file::EmbeddedResidentStore};
#[cfg(feature = "full")]
use reddb_file::VfsFile;
use std::collections::HashMap;
#[cfg(feature = "full")]
use std::path::PathBuf;

use super::{
    evaluator::EvalError,
    planning::build_table_plan_with_host,
    runtime::{
        projection::projected_columns,
        table_plan::{execute_table_plan_with_host, visit_table_plan_with_host},
    },
    scalar::ScalarResult,
};

pub(crate) mod admission;
pub(crate) mod host;
pub(crate) mod source;

use source::StoreView;

/// Bound materialization before cloning records, independently of SQL LIMIT.
pub const LOCAL_READ_ROWS_MAX: usize = 100_000;

/// Host-owned equality restriction, applied before the caller's SQL predicates.
/// This is a local inspection boundary, not a server authentication policy.
pub struct ReadScope {
    pub table: String,
    pub field: String,
    pub value: reddb_types::Value,
}

#[derive(Debug, PartialEq, Eq)]
pub struct StreamQueryResult {
    pub columns: Vec<String>,
    pub rows: u64,
    pub complete: bool,
}

#[derive(Debug)]
pub enum LocalReadError {
    Storage(StoreError),
    Catalog(String),
    Parse(reddb_rql::ParseError),
    Unsupported(&'static str),
    Evaluation(EvalError),
    Query(String),
    Parameters(super::parameters::UserParamError),
    ResourceLimit,
    Vault(&'static str),
}

impl LocalReadError {
    pub fn code(&self) -> &'static str {
        match self {
            Self::Vault(code) => code,
            Self::Storage(_) => "DATABASE_ERROR",
            Self::Catalog(_) => "CATALOG_ERROR",
            Self::Parse(_) => "QUERY_PARSE_ERROR",
            Self::Unsupported(_) => "QUERY_UNSUPPORTED",
            Self::Evaluation(_) | Self::Query(_) => "QUERY_EVALUATION_ERROR",
            Self::ResourceLimit => "QUERY_RESOURCE_LIMIT",
            Self::Parameters(_) => "INVALID_PARAMS",
        }
    }
}

impl std::fmt::Display for LocalReadError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Vault(code) => formatter.write_str(code),
            Self::Storage(error) => error.fmt(formatter),
            Self::Catalog(message) | Self::Query(message) => formatter.write_str(message),
            Self::Parse(error) => error.fmt(formatter),
            Self::Unsupported(message) => formatter.write_str(message),
            Self::Evaluation(error) => error.fmt(formatter),
            Self::Parameters(error) => error.fmt(formatter),
            Self::ResourceLimit => {
                formatter.write_str("local SELECT materialization exceeds 100000 rows")
            }
        }
    }
}
impl std::error::Error for LocalReadError {}
impl From<StoreError> for LocalReadError {
    fn from(error: StoreError) -> Self {
        Self::Storage(error)
    }
}
impl From<EvalError> for LocalReadError {
    fn from(error: EvalError) -> Self {
        Self::Evaluation(error)
    }
}

/// Holds the exclusively owned file and reconstructs snapshot + committed WAL once.
/// Callers retain the file lock until this value is dropped. No mutator is exposed.
#[cfg(feature = "full")]
pub struct LocalReadDatabase<F: VfsFile> {
    store: EmbeddedResidentStore<F>,
    contracts: HashMap<String, CollectionContract>,
}

#[cfg(feature = "full")]
impl<F: VfsFile> LocalReadDatabase<F> {
    pub fn open(file: F, name: impl Into<PathBuf>) -> Result<Self, LocalReadError> {
        let store = EmbeddedResidentStore::open(file, name, ManagerConfig::default(), false, 0)?;
        let metadata = store.committed_resident()?.aux_metadata();
        let mut contracts = HashMap::new();
        if !metadata.is_empty() {
            for contract in deserialize_collection_contracts(&metadata)
                .map_err(|error| LocalReadError::Catalog(error.to_string()))?
            {
                let name = contract.name.clone();
                if contracts.insert(name.clone(), contract).is_some() {
                    return Err(LocalReadError::Catalog(format!(
                        "duplicate collection contract: {name}"
                    )));
                }
            }
        }
        Ok(Self { store, contracts })
    }

    pub fn query(&self, sql: &str) -> Result<ScalarResult, LocalReadError> {
        query_resident(self.store.committed_resident()?, &self.contracts, sql)
    }
}

#[cfg(feature = "full")]
pub(crate) fn query_resident(
    resident: &dyn StoreView,
    contracts: &HashMap<String, CollectionContract>,
    sql: &str,
) -> Result<ScalarResult, LocalReadError> {
    let query = admission::query(sql)?;
    execute_resident_query(resident, contracts, query, None)
}

pub(crate) fn query_resident_bound(
    resident: &dyn StoreView,
    contracts: &HashMap<String, CollectionContract>,
    query: reddb_rql::ast::TableQuery,
) -> Result<ScalarResult, LocalReadError> {
    execute_resident_query(resident, contracts, admission::table(query)?, None)
}

#[cfg(feature = "full")]
pub(crate) fn query_resident_scoped(
    resident: &dyn StoreView,
    contracts: &HashMap<String, CollectionContract>,
    query: reddb_rql::ast::TableQuery,
    scope: &ReadScope,
) -> Result<ScalarResult, LocalReadError> {
    execute_resident_query(resident, contracts, admission::table(query)?, Some(scope))
}

pub(crate) fn query_store_stream(
    store: &dyn StoreView,
    contracts: &HashMap<String, CollectionContract>,
    query: reddb_rql::ast::TableQuery,
    visit: &mut dyn FnMut(&[reddb_types::Value]) -> Result<bool, LocalReadError>,
) -> Result<StreamQueryResult, LocalReadError> {
    let query = admission::table(query)?;
    let contract = contracts.get(&query.table).ok_or_else(|| {
        LocalReadError::Catalog(format!(
            "no persisted collection contract for {}",
            query.table
        ))
    })?;
    if contract.declared_model != CollectionModel::Table
        || contract.default_ttl_ms.is_some()
        || contract.retention_duration_ms.is_some()
    {
        return Err(LocalReadError::Unsupported(
            "local SELECT currently requires a table without TTL or retention",
        ));
    }
    let columns = crate::catalog::schema::resolved_contract_columns(contract)
        .map_err(|error| LocalReadError::Catalog(error.to_string()))?;
    admission::expressions(&query, &columns)?;
    let schema = std::sync::Arc::new(columns.iter().map(|column| column.name.clone()).collect());
    let source = source::StoreRows {
        store,
        table: &query.table,
        schema,
        allow_materialization: false,
    };
    let projections = reddb_rql::sql_lowering::effective_table_projections(&query);
    if projections
        .iter()
        .any(|projection| matches!(projection, reddb_rql::ast::Projection::All))
    {
        return Err(LocalReadError::Unsupported(
            "streaming SELECT requires explicit output columns",
        ));
    }
    let names = projected_columns(&[], &projections);
    let host = host::ReadHost::with_source(&query, &columns, &projections, &source);
    let plan = build_table_plan_with_host(&host, &query);
    let mut rows = 0u64;
    let mut complete = true;
    visit_table_plan_with_host(&host, &plan, &query, &mut |record| {
        let row = names
            .iter()
            .map(|name| {
                record
                    .get(name)
                    .cloned()
                    .unwrap_or(reddb_types::Value::Null)
            })
            .collect::<Vec<_>>();
        rows = rows.checked_add(1).ok_or(LocalReadError::ResourceLimit)?;
        if !visit(&row)? {
            complete = false;
            return Ok(false);
        }
        Ok(true)
    })?;
    Ok(StreamQueryResult {
        columns: names,
        rows,
        complete,
    })
}

fn execute_resident_query(
    resident: &dyn StoreView,
    contracts: &HashMap<String, CollectionContract>,
    query: reddb_rql::ast::TableQuery,
    scope: Option<&ReadScope>,
) -> Result<ScalarResult, LocalReadError> {
    let contract = contracts.get(&query.table).ok_or_else(|| {
        LocalReadError::Catalog(format!(
            "no persisted collection contract for {}",
            query.table
        ))
    })?;
    if contract.declared_model != CollectionModel::Table
        || contract.default_ttl_ms.is_some()
        || contract.retention_duration_ms.is_some()
    {
        return Err(LocalReadError::Unsupported(
            "local SELECT currently requires a table without TTL or retention",
        ));
    }
    let columns = crate::catalog::schema::resolved_contract_columns(contract)
        .map_err(|error| LocalReadError::Catalog(error.to_string()))?;
    admission::expressions(&query, &columns)?;
    if scope.is_some_and(|scope| !columns.iter().any(|column| column.name == scope.field)) {
        return Err(LocalReadError::Catalog(
            "inspection scope column is missing".into(),
        ));
    }
    let schema = std::sync::Arc::new(columns.iter().map(|column| column.name.clone()).collect());
    let mut entities = Vec::new();
    let mut exceeded = false;
    resident.scan_collection(&query.table, &mut |entity, metadata| {
        if let Some(scope) = scope {
            let Some(record) = super::records::runtime_table_record_from_entity_ref_with_schema(
                entity,
                Some(&schema),
            ) else {
                return true;
            };
            if record.get(&scope.field) != Some(&scope.value) {
                return true;
            }
        }
        if entities.len() == LOCAL_READ_ROWS_MAX {
            exceeded = true;
            return false;
        }
        let has_ttl = metadata.is_some_and(|metadata| {
            metadata.has(INTERNAL_TTL_SECONDS_KEY)
                || metadata.has(INTERNAL_TTL_MILLIS_KEY)
                || metadata.has(INTERNAL_EXPIRES_AT_KEY)
        });
        entities.push((entity.clone(), has_ttl));
        true
    })?;
    if exceeded {
        return Err(LocalReadError::ResourceLimit);
    }
    let mut records = Vec::with_capacity(entities.len());
    for (entity, has_ttl) in entities {
        if !matches!(entity.data, EntityData::Row(_)) {
            return Err(LocalReadError::Unsupported(
                "local table contains non-row entities",
            ));
        }
        if has_ttl {
            return Err(LocalReadError::Unsupported(
                "local SELECT does not yet admit row TTL metadata",
            ));
        }
        if let Some(record) =
            super::records::runtime_table_record_from_entity_ref_with_schema(&entity, Some(&schema))
        {
            records.push(record);
        }
    }
    let projections = reddb_rql::sql_lowering::effective_table_projections(&query);
    let host = host::ReadHost::new(&query, &columns, &projections, records);
    let plan = build_table_plan_with_host(&host, &query);
    let records = execute_table_plan_with_host(&host, &plan, &query)?;
    let columns = projected_columns(&records, &projections);
    let rows = records
        .iter()
        .map(|record| {
            columns
                .iter()
                .map(|column| {
                    record
                        .get(column)
                        .cloned()
                        .unwrap_or(reddb_types::Value::Null)
                })
                .collect()
        })
        .collect();
    Ok(ScalarResult { columns, rows })
}