radixdb_executor/procedural/
mod.rs

1//! Atomic bridge from verified procedural IR into the canonical SQL executor.
2//!
3//! The bridge owns no SQL semantics and no storage handle. It adapts values,
4//! streams SQL rows, and coordinates one existing executor transaction.
5
6mod binding;
7mod binding_types;
8mod cache;
9mod call;
10mod error;
11mod external;
12pub(crate) mod function;
13mod function_binding;
14mod host;
15mod job;
16mod native_function;
17mod trigger;
18mod value;
19
20#[cfg(test)]
21mod call_binding_tests;
22
23use std::collections::BTreeSet;
24use std::sync::Arc;
25
26use radixdb_catalog::{
27    CatalogGeneration, CatalogPayload, ObjectId, ObjectKind, ResourcePolicy, SecurityMode,
28    Volatility,
29};
30use radixdb_procedural::{CompileIdentity, VerifiedProgram};
31use radixdb_sql::{parse_sql, CreateRoutineStatement, Statement};
32
33pub(crate) use cache::ProceduralProgramCache;
34pub(crate) use call::CallBoundary;
35pub use call::{ProceduralCallOutcome, ProceduralResultStage};
36pub(crate) use job::bind_job_definition;
37pub use job::{
38    JobAttemptMetadata, JobAttemptOutcome, ScheduledJobDefinition, ScheduledJobSchedule,
39};
40pub(crate) use native_function::catalog_type_name;
41pub use radixdb_procedural::{Diagnostic, DiagnosticKind};
42pub(crate) use trigger::{
43    fire_after_row_triggers, fire_before_row_triggers, fire_statement_triggers,
44    prepare_dml_triggers, validate_trigger_attachment, DmlTriggerEvent, DmlTriggerPlan,
45};
46
47use crate::catalog::{
48    validate_routine_source_contract, PROCEDURAL_COMPILER_ABI, PROCEDURAL_RUNTIME_ABI,
49};
50
51#[derive(Debug, Clone)]
52pub(super) struct PublishedRoutine {
53    pub program: VerifiedProgram,
54    pub owner: ObjectId,
55    pub security: SecurityMode,
56    pub volatility: Volatility,
57    pub resource_policy: ResourcePolicy,
58}
59
60pub(crate) fn compile_catalog_routine(
61    executor: &crate::Executor,
62    statement: &radixdb_sql::CreateRoutineStatement,
63    catalog: &radixdb_catalog::CatalogGeneration,
64    identity: radixdb_procedural::CompileIdentity,
65    search_path: Vec<radixdb_catalog::ObjectId>,
66) -> radixdb_core::Result<Vec<radixdb_catalog::ObjectId>> {
67    if matches!(
68        statement.returns,
69        Some(radixdb_sql::RoutineReturnSyntax::Trigger)
70    ) {
71        if statement.kind != radixdb_sql::RoutineKindSyntax::Function
72            || !statement.arguments.is_empty()
73            || !matches!(
74                statement.volatility,
75                None | Some(radixdb_sql::RoutineVolatilitySyntax::Volatile)
76            )
77        {
78            return Err(radixdb_core::Error::InvalidArgument(
79                "RETURNS TRIGGER requires a zero-argument VOLATILE Function".to_string(),
80            ));
81        }
82        // OLD/NEW are typed by CREATE TRIGGER ... ON <table>. The complete
83        // verified program is therefore admitted at that attachment boundary.
84        return Ok(Vec::new());
85    }
86    let (program, dependencies) =
87        compile_verified_catalog_routine(executor, statement, catalog, identity, search_path)?;
88    drop(program);
89    Ok(dependencies)
90}
91
92fn compile_verified_catalog_routine(
93    executor: &crate::Executor,
94    statement: &CreateRoutineStatement,
95    catalog: &CatalogGeneration,
96    identity: CompileIdentity,
97    search_path: Vec<ObjectId>,
98) -> radixdb_core::Result<(VerifiedProgram, Vec<ObjectId>)> {
99    let volatility = match statement.kind {
100        radixdb_sql::RoutineKindSyntax::Procedure => None,
101        radixdb_sql::RoutineKindSyntax::Function => Some(match statement.volatility {
102            Some(radixdb_sql::RoutineVolatilitySyntax::Immutable) => Volatility::Immutable,
103            Some(radixdb_sql::RoutineVolatilitySyntax::Stable) => Volatility::Stable,
104            None | Some(radixdb_sql::RoutineVolatilitySyntax::Volatile) => Volatility::Volatile,
105        }),
106    };
107    let mut resolver = binding::ExecutorSemanticResolver::with_search_path(
108        executor,
109        catalog,
110        search_path,
111        volatility,
112    );
113    let compiled = radixdb_procedural::compile_routine(statement, identity, &mut resolver)
114        .map_err(procedural_compile_error)?;
115    let program = radixdb_procedural::verify(compiled.program).map_err(procedural_compile_error)?;
116    Ok((program, compiled.dependencies))
117}
118
119pub(super) fn load_published_routine(
120    executor: &crate::Executor,
121    routine_id: ObjectId,
122    expected_kind: ObjectKind,
123) -> radixdb_core::Result<Arc<PublishedRoutine>> {
124    let (catalog, cacheable) = transaction_visible_catalog(executor)?;
125    let object = catalog.object(routine_id).cloned().ok_or_else(|| {
126        radixdb_core::Error::InvalidArgument(format!(
127            "routine object {routine_id} does not exist in the transaction-visible catalog"
128        ))
129    })?;
130    if object.kind() != expected_kind {
131        return Err(radixdb_core::Error::InvalidArgument(format!(
132            "catalog object {routine_id} is {:?}, expected {expected_kind:?}",
133            object.kind()
134        )));
135    }
136    let definition = match object.payload() {
137        CatalogPayload::Function(payload) => payload.procedural_definition().ok_or_else(|| {
138            radixdb_core::Error::InvalidArgument(format!(
139                "native function {routine_id} cannot be loaded as a procedural routine"
140            ))
141        })?,
142        CatalogPayload::Procedure(payload) => payload.definition(),
143        _ => {
144            return Err(radixdb_core::Error::InvalidArgument(format!(
145                "catalog object {routine_id} has no executable routine definition"
146            )))
147        }
148    };
149    if definition.compiler_abi() != PROCEDURAL_COMPILER_ABI
150        || definition.runtime_abi() != PROCEDURAL_RUNTIME_ABI
151    {
152        return Err(radixdb_core::Error::InvalidArgument(format!(
153            "routine {routine_id} requires unsupported compiler/runtime ABI {}/{}",
154            definition.compiler_abi(),
155            definition.runtime_abi()
156        )));
157    }
158
159    let cache_key = routine_cache_key(catalog.as_ref(), &object)?;
160    if cacheable {
161        if let Some(cached) = executor.procedural_cache.get(&cache_key) {
162            // Cached IR is private and immutable, but verification remains the
163            // fail-closed admission boundary if that invariant ever changes.
164            radixdb_procedural::verify(cached.program.program().clone())
165                .map_err(procedural_compile_error)?;
166            return Ok(cached);
167        }
168    }
169
170    let mut statements = parse_sql(definition.source().as_str())
171        .map_err(|error| radixdb_core::Error::Parse(error.to_string()))?;
172    if statements.len() != 1 {
173        return Err(radixdb_core::Error::InvalidArgument(format!(
174            "routine {routine_id} durable source must contain exactly one CREATE statement"
175        )));
176    }
177    let Statement::CreateRoutine(statement) = statements.remove(0) else {
178        return Err(radixdb_core::Error::InvalidArgument(format!(
179            "routine {routine_id} durable source is not CREATE FUNCTION/PROCEDURE"
180        )));
181    };
182    validate_routine_source_contract(&statement, &object, catalog.as_ref())?;
183
184    let identity = CompileIdentity {
185        object_id: object.id(),
186        definition_revision: object.definition_revision(),
187        display_name: statement.name.to_string(),
188    };
189    let (program, dependencies) = compile_verified_catalog_routine(
190        executor,
191        &statement,
192        catalog.as_ref(),
193        identity,
194        definition.search_path().to_vec(),
195    )?;
196    let search_path = definition
197        .search_path()
198        .iter()
199        .copied()
200        .collect::<BTreeSet<_>>();
201    let compiled_dependencies = dependencies
202        .into_iter()
203        .filter(|id| !search_path.contains(id))
204        .collect::<BTreeSet<_>>()
205        .into_iter()
206        .collect::<Vec<_>>();
207    if compiled_dependencies != definition.dependency_ids() {
208        return Err(radixdb_core::Error::InvalidArgument(format!(
209            "routine {routine_id} durable dependency contract is stale or corrupt"
210        )));
211    }
212
213    let routine = Arc::new(PublishedRoutine {
214        program,
215        owner: object.owner_principal_id(),
216        security: definition.security(),
217        volatility: definition.volatility(),
218        resource_policy: definition.resource_policy(),
219    });
220    if cacheable {
221        Ok(executor.procedural_cache.insert(cache_key, routine))
222    } else {
223        Ok(routine)
224    }
225}
226
227pub(crate) fn transaction_visible_catalog(
228    executor: &crate::Executor,
229) -> radixdb_core::Result<(Arc<CatalogGeneration>, bool)> {
230    let active = executor.active_transaction.lock().unwrap();
231    if let Some(state) = active.as_ref() {
232        return Ok((
233            state.catalog.working_generation_shared(),
234            !state.catalog.has_pending_catalog_changes(),
235        ));
236    }
237    drop(active);
238    executor.engine.pin_catalog().map(|catalog| (catalog, true))
239}
240
241fn routine_cache_key(
242    catalog: &CatalogGeneration,
243    object: &radixdb_catalog::CatalogObject,
244) -> radixdb_core::Result<cache::RoutineCacheKey> {
245    let definition = match object.payload() {
246        CatalogPayload::Function(payload) => payload.procedural_definition().ok_or_else(|| {
247            radixdb_core::Error::InvalidArgument(format!(
248                "native function {} has no procedural cache key",
249                object.id()
250            ))
251        })?,
252        CatalogPayload::Procedure(payload) => payload.definition(),
253        _ => {
254            return Err(radixdb_core::Error::InvalidArgument(format!(
255                "catalog object {} has no executable routine definition",
256                object.id()
257            )))
258        }
259    };
260    let mut dependency_ids = definition
261        .search_path()
262        .iter()
263        .chain(definition.dependency_ids())
264        .copied()
265        .collect::<BTreeSet<_>>();
266    dependency_ids.insert(object.id());
267    let dependency_versions = dependency_ids
268        .into_iter()
269        .map(|id| {
270            catalog
271                .object(id)
272                .map(|dependency| (id, dependency.definition_revision()))
273                .ok_or_else(|| {
274                    radixdb_core::Error::InvalidArgument(format!(
275                        "routine {} dependency {id} is missing",
276                        object.id()
277                    ))
278                })
279        })
280        .collect::<radixdb_core::Result<Vec<_>>>()?;
281    let meta = catalog.meta();
282    Ok(cache::RoutineCacheKey {
283        database_id: meta.database_id(),
284        catalog_id: meta.catalog_id(),
285        catalog_generation: meta.catalog_generation(),
286        object_id: object.id(),
287        definition_revision: object.definition_revision(),
288        source_digest: *definition.source().digest(),
289        dependency_versions,
290        compiler_abi: definition.compiler_abi(),
291        runtime_abi: definition.runtime_abi(),
292    })
293}
294
295fn procedural_compile_error(diagnostic: radixdb_procedural::Diagnostic) -> radixdb_core::Error {
296    radixdb_core::Error::InvalidArgument(format!("procedural compilation failed: {diagnostic}"))
297}
298
299#[cfg(test)]
300mod tests;
radixdb_executor/procedural/mod.rs

radixdb_executor/procedural/
mod.rs
