use std::sync::Arc;
use radixdb_core::{Error, Result};
use radixdb_sql::ast::{Program, Statement};
use crate::context::{ExecutionContext, TimeoutGuard};
use crate::query_cache::{CachedPlanRef, ParameterContract};
use crate::result::{ExecutionResult, TimedQueryResult};
use crate::Executor;
#[doc(hidden)]
pub fn program_contains_transaction_control(sql: &str) -> bool {
crate::dispatch::program::parse_program(sql).is_ok_and(|program| {
program.statements.iter().any(|statement| {
matches!(
statement,
Statement::Begin(_)
| Statement::Commit(_)
| Statement::Rollback(_)
| Statement::Savepoint(_)
| Statement::ReleaseSavepoint(_)
)
})
})
}
#[derive(Clone)]
pub struct PreparedProgram {
inner: PreparedProgramInner,
}
#[derive(Clone)]
enum PreparedProgramInner {
Single(CachedPlanRef),
Multi {
program: Arc<Program>,
parameter_contract: ParameterContract,
},
}
impl Executor {
#[doc(hidden)]
pub fn prepare_program(&self, sql: &str) -> Result<PreparedProgram> {
if let Some(cached) = self.query_cache().get(sql) {
reject_expression_statement(cached.statement(), sql)?;
return Ok(PreparedProgram {
inner: PreparedProgramInner::Single(cached),
});
}
let mut program = crate::dispatch::program::parse_program(sql)?;
if program.statements.is_empty() {
return Err(Error::NoStatementsToExecute);
}
if program.statements.len() == 1 {
let statement = program
.statements
.pop()
.expect("single-statement length was checked");
reject_expression_statement(&statement, sql)?;
let plan = self.query_cache().put(sql, Arc::new(statement), false, 0);
return Ok(PreparedProgram {
inner: PreparedProgramInner::Single(plan),
});
}
for statement in &program.statements {
reject_expression_statement(statement, sql)?;
}
let parameter_contract = ParameterContract::from_statements(&program.statements);
Ok(PreparedProgram {
inner: PreparedProgramInner::Multi {
program: Arc::new(program),
parameter_contract,
},
})
}
#[doc(hidden)]
pub fn execute_prepared_program(
&self,
program: &PreparedProgram,
context: &ExecutionContext,
) -> Result<ExecutionResult> {
match &program.inner {
PreparedProgramInner::Single(plan) => self.execute_with_cached_plan(plan, context),
PreparedProgramInner::Multi {
program,
parameter_contract,
} => {
parameter_contract.validate(context)?;
self.execute_program_with_context(program, context)
}
}
}
#[doc(hidden)]
pub fn execute_installed_transaction_sql(
&self,
sql: &str,
context: &ExecutionContext,
) -> Result<ExecutionResult> {
let program = crate::dispatch::program::parse_program(sql)?;
let parameter_contract = ParameterContract::from_statements(&program.statements);
self.execute_installed_statements(&program.statements, ¶meter_contract, context, sql)
}
#[doc(hidden)]
pub fn execute_installed_transaction_prepared(
&self,
program: &PreparedProgram,
context: &ExecutionContext,
sql: &str,
) -> Result<ExecutionResult> {
match &program.inner {
PreparedProgramInner::Single(plan) => self.execute_installed_statements(
std::slice::from_ref(plan.statement()),
plan.parameter_contract(),
context,
sql,
),
PreparedProgramInner::Multi {
program,
parameter_contract,
} => self.execute_installed_statements(
&program.statements,
parameter_contract,
context,
sql,
),
}
}
fn execute_installed_statements(
&self,
statements: &[Statement],
parameter_contract: &ParameterContract,
context: &ExecutionContext,
sql: &str,
) -> Result<ExecutionResult> {
if statements.is_empty() {
return Err(Error::NoStatementsToExecute);
}
parameter_contract.validate(context)?;
let timeout_guard = TimeoutGuard::new(context);
let mut last_result: Option<ExecutionResult> = None;
for statement in statements {
reject_embedded_transaction_control(statement)?;
if let Some(mut previous) = last_result.take() {
while previous.next() {}
if let Some(error) = previous.last_error() {
return Err(error);
}
previous.close()?;
}
last_result = Some(self.execute_statement(statement, context)?);
}
let result = last_result.ok_or(Error::NoStatementsToExecute)?;
Ok(TimedQueryResult::wrap_with_workload(
result,
timeout_guard,
context.cancellation_handle(),
sql,
))
}
}
fn reject_expression_statement(statement: &Statement, sql: &str) -> Result<()> {
if matches!(statement, Statement::Expression(_)) {
Err(Error::parse(format!(
"invalid SQL: unrecognised statement: {sql}"
)))
} else {
Ok(())
}
}
fn reject_embedded_transaction_control(statement: &Statement) -> Result<()> {
if matches!(
statement,
Statement::Begin(_)
| Statement::Commit(_)
| Statement::Rollback(_)
| Statement::Savepoint(_)
| Statement::ReleaseSavepoint(_)
) {
Err(Error::NotSupported(
"use the Transaction commit, rollback and savepoint methods for transaction control"
.to_string(),
))
} else {
Ok(())
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use super::*;
use radixdb_storage::mvcc::engine::MVCCEngine;
fn executor() -> Executor {
let engine = MVCCEngine::in_memory();
engine.open_engine().expect("open in-memory engine");
Executor::new(Arc::new(engine))
}
#[test]
fn prepared_program_rejects_empty_and_expression_input() {
let executor = executor();
assert_eq!(
executor.prepare_program("").err(),
Some(Error::NoStatementsToExecute)
);
assert!(matches!(
executor.prepare_program("SELECTX invalid").err(),
Some(Error::Parse(_))
));
}
#[test]
fn prepared_program_preserves_single_and_multi_parameter_contracts() {
let executor = executor();
executor
.execute("CREATE TABLE prepared_owner (id INTEGER PRIMARY KEY, value TEXT)")
.expect("create table");
let single = executor
.prepare_program("INSERT INTO prepared_owner VALUES ($1, $2)")
.expect("prepare single");
executor
.execute_prepared_program(
&single,
&ExecutionContext::with_params(vec![1.into(), "one".into()].into()),
)
.expect("execute single");
let multi = executor
.prepare_program(
"INSERT INTO prepared_owner VALUES ($1, 'two'); SELECT value FROM prepared_owner WHERE id = $1",
)
.expect("prepare multi");
let mut result = executor
.execute_prepared_program(
&multi,
&ExecutionContext::with_params(vec![2.into()].into()),
)
.expect("execute multi");
assert!(result.next());
assert_eq!(result.row().get(0), Some(&"two".into()));
}
#[test]
fn installed_transaction_rejects_sql_lifecycle_control() {
let executor = executor();
let transaction = executor.begin_transaction().expect("begin transaction");
executor.install_transaction(transaction);
let error =
match executor.execute_installed_transaction_sql("COMMIT", &ExecutionContext::new()) {
Err(error) => error,
Ok(_) => panic!("facade must remain transaction lifecycle owner"),
};
assert!(matches!(error, Error::NotSupported(_)));
executor
.rollback_installed_transaction()
.expect("rollback installed transaction");
}
#[test]
fn transaction_control_classification_stays_inside_executor() {
assert!(program_contains_transaction_control("SELECT 1; COMMIT"));
assert!(program_contains_transaction_control(
"SAVEPOINT before_update"
));
assert!(!program_contains_transaction_control("SELECT 'COMMIT'"));
assert!(!program_contains_transaction_control(
"malformed select from"
));
}
}