radixdb_executor/procedural/
host.rs

1use std::collections::BTreeMap;
2use std::sync::atomic::{AtomicU64, Ordering};
3use std::sync::Arc;
4
5use radixdb_catalog::{ObjectId, ObjectKind, SecurityMode, Volatility};
6use radixdb_core::{Error, Row, Value};
7use radixdb_procedural::{
8    admit_embedded_sql, AuditEvent, AuditHost, BudgetOwner, CancellationProbe, CursorHost,
9    CursorToken, Diagnostic, DiagnosticKind, Interpreter, OutboxHost, OutboxMessage,
10    PrincipalContext, PrincipalHost, ProceduralResult, RoutineCallHost, RuntimeValue,
11    SavepointToken, SqlHost, SqlOutcome, SqlRowSink, TransactionHost,
12};
13use radixdb_sql::{
14    parse_sql, walk_statement_physical_table_sources, walk_statement_tree, Expression,
15    InfixExpression, InfixOperator, Position, Statement, Token, TokenType,
16};
17use radixdb_storage::traits::QueryResult;
18
19use crate::context::{CancellationHandle, ExecutionContext, TimeoutGuard};
20use crate::expression::ExpressionEval;
21use crate::Executor;
22
23use super::error::map_executor_error;
24use super::function::ExecutorStoredFunctionInvoker;
25use super::load_published_routine;
26use super::value::{runtime_row, scalar_parameters, scalar_value};
27
28static NEXT_HOST_RESOURCE_ID: AtomicU64 = AtomicU64::new(1);
29
30struct CursorEntry {
31    result: Box<dyn QueryResult>,
32    _timeout: Option<TimeoutGuard>,
33}
34
35pub(super) struct ExecutorProceduralHost<'a> {
36    executor: &'a Executor,
37    context: ExecutionContext,
38    principals: PrincipalContext,
39    definers: Vec<ObjectId>,
40    cursors: BTreeMap<u64, CursorEntry>,
41    savepoints: BTreeMap<u64, String>,
42    function_volatility: Option<Volatility>,
43}
44
45impl<'a> ExecutorProceduralHost<'a> {
46    pub(super) fn new(
47        executor: &'a Executor,
48        context: &ExecutionContext,
49        principals: PrincipalContext,
50        function_volatility: Option<Volatility>,
51    ) -> Self {
52        Self {
53            executor,
54            context: context.clone(),
55            principals,
56            definers: Vec::new(),
57            cursors: BTreeMap::new(),
58            savepoints: BTreeMap::new(),
59            function_volatility,
60        }
61    }
62
63    fn sql_context(
64        &self,
65        parameters: &[RuntimeValue],
66        budget: &BudgetOwner,
67    ) -> ProceduralResult<ExecutionContext> {
68        budget.check_boundary()?;
69        self.context.check_cancelled().map_err(map_executor_error)?;
70        let mut context = self.context.clone();
71        context.set_params(scalar_parameters(parameters)?);
72        let remaining = budget.remaining_deadline()?;
73        let remaining_ms = u64::try_from(remaining.as_millis())
74            .unwrap_or(u64::MAX)
75            .max(1);
76        context.set_timeout_ms(remaining_ms);
77        context = context.with_procedural_budget(budget.clone());
78        context = context.with_effective_principal_id(self.principal_context().effective_principal);
79        let invoker = Arc::new(ExecutorStoredFunctionInvoker::new_with_principals(
80            self.executor,
81            &context,
82            self.principal_context(),
83            self.function_volatility,
84        ));
85        context = context.with_stored_function_invoker(invoker);
86        Ok(context)
87    }
88
89    fn execute_result(
90        &self,
91        statement: &Statement,
92        parameters: &[RuntimeValue],
93        budget: &BudgetOwner,
94    ) -> ProceduralResult<(Box<dyn QueryResult>, Option<TimeoutGuard>)> {
95        admit_embedded_sql(statement)?;
96        let context = self.sql_context(parameters, budget)?;
97        let timeout = TimeoutGuard::new(&context);
98        let result = self
99            .executor
100            .execute_statement(statement, &context)
101            .map_err(|error| map_controlled_error(error, budget))?;
102        Ok((result, timeout))
103    }
104
105    fn check_statement_capability(&self, statement: &Statement) -> ProceduralResult<()> {
106        let Some(caller) = self.function_volatility else {
107            return Ok(());
108        };
109        if caller != Volatility::Volatile
110            && matches!(
111                statement,
112                Statement::Insert(_) | Statement::Update(_) | Statement::Delete(_)
113            )
114        {
115            return Err(Diagnostic::new(
116                DiagnosticKind::VerifyCapabilityDenied,
117                "IMMUTABLE and STABLE functions cannot execute DML",
118            ));
119        }
120        if caller != Volatility::Volatile && matches!(statement, Statement::Call(_)) {
121            return Err(Diagnostic::new(
122                DiagnosticKind::VerifyCapabilityDenied,
123                "IMMUTABLE and STABLE functions cannot call procedures",
124            ));
125        }
126        if caller == Volatility::Immutable {
127            let mut reads_relation = false;
128            walk_statement_physical_table_sources(statement, &mut |_| reads_relation = true);
129            if reads_relation {
130                return Err(Diagnostic::new(
131                    DiagnosticKind::VerifyCapabilityDenied,
132                    "IMMUTABLE functions cannot read tables or views",
133                ));
134            }
135        }
136        let mut denied = None;
137        walk_statement_tree(statement, &mut |expression| {
138            let Expression::FunctionCall(function) = expression else {
139                return;
140            };
141            let Some(info) = self.executor.function_registry.get_info(&function.function) else {
142                return;
143            };
144            let target = match info.volatility {
145                radixdb_functions::FunctionVolatility::Immutable => Volatility::Immutable,
146                radixdb_functions::FunctionVolatility::Stable => Volatility::Stable,
147                radixdb_functions::FunctionVolatility::Volatile => Volatility::Volatile,
148            };
149            if !volatility_allows(caller, target) {
150                denied = Some(Diagnostic::new(
151                    DiagnosticKind::VerifyCapabilityDenied,
152                    format!(
153                        "{caller:?} function cannot invoke {target:?} built-in {}",
154                        function.function
155                    ),
156                ));
157            }
158        });
159        if let Some(error) = denied {
160            return Err(error);
161        }
162        Ok(())
163    }
164
165    fn evaluate(
166        &self,
167        expression: &Expression,
168        context: &ExecutionContext,
169    ) -> ProceduralResult<Value> {
170        ExpressionEval::compile(expression, &[])
171            .map_err(map_executor_error)?
172            .with_context(context)
173            .eval_slice(&Row::new())
174            .map_err(map_executor_error)
175    }
176}
177
178fn volatility_allows(caller: Volatility, target: Volatility) -> bool {
179    match caller {
180        Volatility::Volatile => true,
181        Volatility::Stable => target != Volatility::Volatile,
182        Volatility::Immutable => target == Volatility::Immutable,
183    }
184}
185
186impl CancellationProbe for CancellationHandle {
187    fn is_cancelled(&self) -> bool {
188        CancellationHandle::is_cancelled(self)
189    }
190}
191
192impl SqlHost for ExecutorProceduralHost<'_> {
193    fn evaluate_expression(
194        &mut self,
195        expression: &Expression,
196        parameters: &[RuntimeValue],
197        budget: &BudgetOwner,
198    ) -> ProceduralResult<RuntimeValue> {
199        let context = self.sql_context(parameters, budget)?;
200        let value = self.evaluate(expression, &context)?;
201        budget.check_boundary()?;
202        Ok(RuntimeValue::scalar(value))
203    }
204
205    fn evaluate_binary(
206        &mut self,
207        operator: InfixOperator,
208        left: &RuntimeValue,
209        right: &RuntimeValue,
210        budget: &BudgetOwner,
211    ) -> ProceduralResult<RuntimeValue> {
212        let spelling = operator_spelling(operator).ok_or_else(|| {
213            Diagnostic::new(
214                DiagnosticKind::VerifyCapabilityDenied,
215                "unbound SQL binary operator reached the executor bridge",
216            )
217        })?;
218        let token = Token::new(TokenType::Operator, spelling, Position::default());
219        let expression = Expression::Infix(InfixExpression::new(
220            token,
221            Box::new(Expression::BoundValue(Box::new(scalar_value(left)?))),
222            spelling,
223            Box::new(Expression::BoundValue(Box::new(scalar_value(right)?))),
224        ));
225        let context = self.sql_context(&[], budget)?;
226        let value = self.evaluate(&expression, &context)?;
227        budget.check_boundary()?;
228        Ok(RuntimeValue::scalar(value))
229    }
230
231    fn execute_sql(
232        &mut self,
233        statement: &Statement,
234        parameters: &[RuntimeValue],
235        rows: &mut dyn SqlRowSink,
236        budget: &BudgetOwner,
237    ) -> ProceduralResult<SqlOutcome> {
238        self.check_statement_capability(statement)?;
239        let (mut result, _timeout) = self.execute_result(statement, parameters, budget)?;
240        let affected_rows = u64::try_from(result.rows_affected()).unwrap_or(0);
241        while result.next() {
242            if let Err(error) = budget.check_boundary() {
243                let _ = result.close();
244                return Err(error);
245            }
246            if let Err(error) = rows.push_row(runtime_row(result.take_row())) {
247                let _ = result.close();
248                return Err(error);
249            }
250        }
251        if let Some(error) = result.last_error() {
252            let _ = result.close();
253            return Err(map_controlled_error(error, budget));
254        }
255        result.close().map_err(map_executor_error)?;
256        budget.check_boundary()?;
257        Ok(SqlOutcome { affected_rows })
258    }
259
260    fn execute_dynamic_sql(
261        &mut self,
262        source: &str,
263        parameters: &[RuntimeValue],
264        rows: &mut dyn SqlRowSink,
265        budget: &BudgetOwner,
266    ) -> ProceduralResult<SqlOutcome> {
267        budget.check_boundary()?;
268        let mut statements = parse_sql(source).map_err(|error| {
269            Diagnostic::new(DiagnosticKind::ParseExpectedToken, error.to_string())
270        })?;
271        if statements.len() != 1 {
272            return Err(Diagnostic::new(
273                DiagnosticKind::ParseUnsupportedSyntax,
274                "dynamic SQL must contain exactly one statement",
275            ));
276        }
277        let statement = statements.remove(0);
278        admit_embedded_sql(&statement)?;
279        self.check_statement_capability(&statement)?;
280        self.execute_sql(&statement, parameters, rows, budget)
281    }
282}
283
284impl CursorHost for ExecutorProceduralHost<'_> {
285    fn open_cursor(
286        &mut self,
287        statement: &Statement,
288        parameters: &[RuntimeValue],
289        budget: &BudgetOwner,
290    ) -> ProceduralResult<CursorToken> {
291        let (result, timeout) = self.execute_result(statement, parameters, budget)?;
292        let token = NEXT_HOST_RESOURCE_ID.fetch_add(1, Ordering::Relaxed);
293        self.cursors.insert(
294            token,
295            CursorEntry {
296                result,
297                _timeout: timeout,
298            },
299        );
300        Ok(CursorToken(token))
301    }
302
303    fn fetch_cursor(
304        &mut self,
305        cursor: CursorToken,
306        budget: &BudgetOwner,
307    ) -> ProceduralResult<Option<Vec<RuntimeValue>>> {
308        budget.check_boundary()?;
309        let entry = self.cursors.get_mut(&cursor.0).ok_or_else(|| {
310            Diagnostic::new(DiagnosticKind::RuntimeInvalidState, "cursor is not open")
311        })?;
312        if entry.result.next() {
313            budget.charge_rows(1)?;
314            return Ok(Some(runtime_row(entry.result.take_row())));
315        }
316        if let Some(error) = entry.result.last_error() {
317            return Err(map_controlled_error(error, budget));
318        }
319        Ok(None)
320    }
321
322    fn close_cursor(&mut self, cursor: CursorToken) -> ProceduralResult<()> {
323        let mut entry = self.cursors.remove(&cursor.0).ok_or_else(|| {
324            Diagnostic::new(DiagnosticKind::RuntimeInvalidState, "cursor is not open")
325        })?;
326        entry.result.close().map_err(map_executor_error)
327    }
328}
329
330impl TransactionHost for ExecutorProceduralHost<'_> {
331    fn create_savepoint(&mut self) -> ProceduralResult<SavepointToken> {
332        let token = NEXT_HOST_RESOURCE_ID.fetch_add(1, Ordering::Relaxed);
333        let name = format!("\0radixdb-procedural-{token}");
334        self.executor
335            .create_active_savepoint(&name)
336            .map_err(map_executor_error)?;
337        self.savepoints.insert(token, name);
338        Ok(SavepointToken(token))
339    }
340
341    fn rollback_savepoint(&mut self, savepoint: SavepointToken) -> ProceduralResult<()> {
342        let name = self.savepoints.get(&savepoint.0).ok_or_else(|| {
343            Diagnostic::new(
344                DiagnosticKind::RuntimeInvalidState,
345                "procedural savepoint is not active",
346            )
347        })?;
348        self.executor
349            .rollback_active_to_savepoint(name)
350            .map_err(map_executor_error)
351    }
352
353    fn release_savepoint(&mut self, savepoint: SavepointToken) -> ProceduralResult<()> {
354        let name = self.savepoints.remove(&savepoint.0).ok_or_else(|| {
355            Diagnostic::new(
356                DiagnosticKind::RuntimeInvalidState,
357                "procedural savepoint is not active",
358            )
359        })?;
360        self.executor
361            .release_active_savepoint(&name)
362            .map_err(map_executor_error)
363    }
364}
365
366impl PrincipalHost for ExecutorProceduralHost<'_> {
367    fn principal_context(&self) -> PrincipalContext {
368        let mut context = self.principals;
369        context.effective_principal = self
370            .definers
371            .last()
372            .copied()
373            .unwrap_or(context.effective_principal);
374        context
375    }
376
377    fn push_definer(&mut self, owner: ObjectId) -> ProceduralResult<()> {
378        self.definers.push(owner);
379        Ok(())
380    }
381
382    fn pop_definer(&mut self) -> ProceduralResult<()> {
383        self.definers.pop().map(|_| ()).ok_or_else(|| {
384            Diagnostic::new(
385                DiagnosticKind::RuntimeInvalidState,
386                "definer stack is empty",
387            )
388        })
389    }
390}
391
392impl RoutineCallHost for ExecutorProceduralHost<'_> {
393    fn call_routine(
394        &mut self,
395        routine: ObjectId,
396        arguments: &[RuntimeValue],
397        budget: &BudgetOwner,
398    ) -> ProceduralResult<Vec<RuntimeValue>> {
399        budget.check_boundary()?;
400        let principals = self.principal_context();
401        crate::authorization::authorize_routine_invocation(
402            self.executor,
403            principals.session_principal,
404            principals.effective_principal,
405            routine,
406        )
407        .map_err(map_executor_error)?;
408        let published = load_published_routine(self.executor, routine, ObjectKind::Procedure)
409            .map_err(map_executor_error)?;
410        let use_definer = published.security == SecurityMode::Definer;
411        if use_definer {
412            self.push_definer(published.owner)?;
413        }
414        let execution = Interpreter.execute(&published.program, arguments.to_vec(), self, budget);
415        let cleanup = if use_definer {
416            self.pop_definer()
417        } else {
418            Ok(())
419        };
420        match (execution, cleanup) {
421            (Ok(outcome), Ok(())) => Ok(outcome.output_values),
422            (Err(primary), Ok(())) => Err(primary),
423            (Ok(_), Err(cleanup)) => Err(cleanup),
424            (Err(primary), Err(cleanup)) => {
425                Err(primary.with_detail("definer_cleanup_error", cleanup.to_string()))
426            }
427        }
428    }
429}
430
431impl AuditHost for ExecutorProceduralHost<'_> {
432    fn append_audit(&mut self, event: AuditEvent) -> ProceduralResult<()> {
433        if self
434            .function_volatility
435            .is_some_and(|volatility| volatility != Volatility::Volatile)
436        {
437            return Err(Diagnostic::new(
438                DiagnosticKind::VerifyCapabilityDenied,
439                "IMMUTABLE and STABLE functions cannot append audit records",
440            ));
441        }
442        let context = self
443            .context
444            .clone()
445            .with_effective_principal_id(self.principal_context().effective_principal);
446        self.executor
447            .append_audit_event(&context, event)
448            .map(|_| ())
449            .map_err(map_executor_error)
450    }
451}
452
453impl OutboxHost for ExecutorProceduralHost<'_> {
454    fn append_outbox(&mut self, message: OutboxMessage) -> ProceduralResult<()> {
455        if self
456            .function_volatility
457            .is_some_and(|volatility| volatility != Volatility::Volatile)
458        {
459            return Err(Diagnostic::new(
460                DiagnosticKind::VerifyCapabilityDenied,
461                "IMMUTABLE and STABLE functions cannot append outbox records",
462            ));
463        }
464        self.executor
465            .append_outbox_message(message)
466            .map(|_| ())
467            .map_err(map_executor_error)
468    }
469}
470
471impl Drop for ExecutorProceduralHost<'_> {
472    fn drop(&mut self) {
473        for (_, mut entry) in std::mem::take(&mut self.cursors) {
474            let _ = entry.result.close();
475        }
476    }
477}
478
479pub(super) fn operator_spelling(operator: InfixOperator) -> Option<&'static str> {
480    Some(match operator {
481        InfixOperator::Equal => "=",
482        InfixOperator::NotEqual => "<>",
483        InfixOperator::LessThan => "<",
484        InfixOperator::LessEqual => "<=",
485        InfixOperator::GreaterThan => ">",
486        InfixOperator::GreaterEqual => ">=",
487        InfixOperator::And => "AND",
488        InfixOperator::Or => "OR",
489        InfixOperator::Xor => "XOR",
490        InfixOperator::Add => "+",
491        InfixOperator::Subtract => "-",
492        InfixOperator::Multiply => "*",
493        InfixOperator::Divide => "/",
494        InfixOperator::Modulo => "%",
495        InfixOperator::Concat => "||",
496        InfixOperator::Like => "LIKE",
497        InfixOperator::ILike => "ILIKE",
498        InfixOperator::NotLike => "NOT LIKE",
499        InfixOperator::NotILike => "NOT ILIKE",
500        InfixOperator::Glob => "GLOB",
501        InfixOperator::NotGlob => "NOT GLOB",
502        InfixOperator::Regexp => "REGEXP",
503        InfixOperator::NotRegexp => "NOT REGEXP",
504        InfixOperator::Is => "IS",
505        InfixOperator::IsNot => "IS NOT",
506        InfixOperator::IsDistinctFrom => "IS DISTINCT FROM",
507        InfixOperator::IsNotDistinctFrom => "IS NOT DISTINCT FROM",
508        InfixOperator::Index => "[]",
509        InfixOperator::JsonAccess => "->",
510        InfixOperator::JsonAccessText => "->>",
511        InfixOperator::VectorDistance => "<=>",
512        InfixOperator::BitwiseAnd => "&",
513        InfixOperator::BitwiseOr => "|",
514        InfixOperator::BitwiseXor => "^",
515        InfixOperator::LeftShift => "<<",
516        InfixOperator::RightShift => ">>",
517        InfixOperator::Other => return None,
518    })
519}
520
521fn map_controlled_error(error: Error, budget: &BudgetOwner) -> Diagnostic {
522    if matches!(error, Error::QueryCancelled) {
523        if let Err(control) = budget.check_boundary() {
524            return control;
525        }
526    }
527    map_executor_error(error)
528}
radixdb_executor/procedural/host.rs

radixdb_executor/procedural/
host.rs
