radixdb_executor/dispatch/
program.rs

1//! SQL parsing, parsed-plan admission, and multi-statement sequencing.
2
3use radixdb_core::{Error, Result, Value};
4use radixdb_sql::ast::{Program, Statement};
5use radixdb_sql::Parser;
6use radixdb_storage::traits::QueryResult;
7
8use crate::context::ExecutionContext;
9
10use super::cache::{CachedPlanRef, ParameterContract, QueryCache};
11
12/// Internal callbacks joining cached-program admission to the executor owner.
13pub trait CachedExecutionHost<B> {
14    fn dispatch_query_cache(&self) -> &QueryCache<B>;
15
16    fn dispatch_execute_bound_plan(
17        &self,
18        plan: &CachedPlanRef<B>,
19        context: &ExecutionContext,
20    ) -> Result<Box<dyn QueryResult>>;
21
22    fn dispatch_execute_statement(
23        &self,
24        statement: &Statement,
25        context: &ExecutionContext,
26    ) -> Result<Box<dyn QueryResult>>;
27}
28
29/// Cache admission callbacks for the borrowed-parameter fast path.
30pub trait CachedFastPathHost<B>: CachedExecutionHost<B> {
31    fn dispatch_fast_path_blocked(&self) -> bool;
32
33    fn dispatch_fast_path_binding_is_nonempty(&self, plan: &CachedPlanRef<B>) -> Result<bool>;
34
35    fn dispatch_try_borrowed_param_fast_path(
36        &self,
37        plan: &CachedPlanRef<B>,
38        params: &[Value],
39    ) -> Option<Result<Box<dyn QueryResult>>>;
40}
41
42pub fn try_fast_path_with_params<H, B>(
43    host: &H,
44    sql: &str,
45    params: &[Value],
46) -> Option<Result<Box<dyn QueryResult>>>
47where
48    H: CachedFastPathHost<B> + ?Sized,
49    B: Default,
50{
51    if host.dispatch_fast_path_blocked() {
52        return None;
53    }
54    let plan = host.dispatch_query_cache().get(sql)?;
55    if plan.parameter_contract().positional_count() != params.len()
56        || !plan.parameter_contract().named_params().is_empty()
57    {
58        return None;
59    }
60    match host.dispatch_fast_path_binding_is_nonempty(&plan) {
61        Ok(true) => return None,
62        Ok(false) => {}
63        Err(error) => return Some(Err(error)),
64    }
65    host.dispatch_try_borrowed_param_fast_path(&plan, params)
66}
67
68pub fn execute_sql<H, B>(
69    host: &H,
70    sql: &str,
71    context: &ExecutionContext,
72) -> Result<Box<dyn QueryResult>>
73where
74    H: CachedExecutionHost<B> + ?Sized,
75    B: Default,
76{
77    if let Some(plan) = host.dispatch_query_cache().get(sql) {
78        plan.parameter_contract().validate(context)?;
79        return host.dispatch_execute_bound_plan(&plan, context);
80    }
81
82    let mut program = parse_program(sql)?;
83    if program.statements.len() != 1 {
84        return execute_program(host, &program, context);
85    }
86
87    let statement = program
88        .statements
89        .pop()
90        .expect("single-statement length was checked");
91    if matches!(statement, Statement::Expression(_)) {
92        return host.dispatch_execute_statement(&statement, context);
93    }
94    let plan = host
95        .dispatch_query_cache()
96        .put(sql, std::sync::Arc::new(statement), false, 0);
97    plan.parameter_contract().validate(context)?;
98    host.dispatch_execute_bound_plan(&plan, context)
99}
100
101pub fn execute_program<H, B>(
102    host: &H,
103    program: &Program,
104    context: &ExecutionContext,
105) -> Result<Box<dyn QueryResult>>
106where
107    H: CachedExecutionHost<B> + ?Sized,
108{
109    if program.statements.is_empty() {
110        return Err(Error::NoStatementsToExecute);
111    }
112
113    let mut last_result: Option<Box<dyn QueryResult>> = None;
114    for statement in &program.statements {
115        if let Some(mut previous) = last_result.take() {
116            while previous.next() {}
117            if let Some(error) = previous.last_error() {
118                return Err(error);
119            }
120            previous.close()?;
121        }
122        last_result = Some(host.dispatch_execute_statement(statement, context)?);
123    }
124    Ok(last_result.expect("non-empty program produces a result"))
125}
126
127pub fn get_or_create_plan<H, B>(host: &H, sql: &str) -> Result<CachedPlanRef<B>>
128where
129    H: CachedExecutionHost<B> + ?Sized,
130    B: Default,
131{
132    if let Some(plan) = host.dispatch_query_cache().get(sql) {
133        return Ok(plan);
134    }
135    let mut program = parse_program(sql)?;
136    if program.statements.len() != 1 {
137        return Err(Error::parse(
138            "Prepared statements must contain exactly one statement",
139        ));
140    }
141    let statement = program
142        .statements
143        .pop()
144        .expect("single-statement length was checked");
145    Ok(host
146        .dispatch_query_cache()
147        .put(sql, std::sync::Arc::new(statement), false, 0))
148}
149
150pub fn execute_prepared_plan<H, B>(
151    host: &H,
152    plan: &CachedPlanRef<B>,
153    context: &ExecutionContext,
154) -> Result<Box<dyn QueryResult>>
155where
156    H: CachedExecutionHost<B> + ?Sized,
157    B: Default,
158{
159    if !host.dispatch_query_cache().owns(plan) {
160        return Err(Error::invalid_argument(
161            "cached plan belongs to a different Database owner",
162        ));
163    }
164    plan.parameter_contract().validate(context)?;
165    host.dispatch_execute_bound_plan(plan, context)
166}
167
168pub fn count_parameters(statement: &Statement) -> (bool, usize) {
169    let contract = ParameterContract::from_statement(statement);
170    (contract.has_params(), contract.positional_count())
171}
172
173pub(crate) fn parse_program(sql: &str) -> Result<Program> {
174    Parser::new(sql)
175        .parse_program()
176        .map_err(|error| Error::parse(error.to_string()))
177}
radixdb_executor/dispatch/program.rs

radixdb_executor/dispatch/
program.rs
