radixdb_executor/dispatch/
statement.rs

1//! Statement classification, fencing, routing, and statement atomicity.
2
3use std::sync::atomic::{AtomicU64, Ordering};
4
5use radixdb_core::{Error, Result};
6use radixdb_sql::ast::*;
7use radixdb_storage::traits::QueryResult;
8
9use crate::context::ExecutionContext;
10use crate::mutation::copy::CopyExecutorExt;
11use crate::mutation::ddl::DdlExecutorExt;
12use crate::mutation::dml::DmlExecutorExt;
13use crate::mutation::extension::ExtensionDdlExecutorExt;
14use crate::mutation::external_type::ExternalTypeDdlExecutorExt;
15use crate::mutation::host::MutationHost;
16use crate::mutation::operator::OperatorDdlExecutorExt;
17
18use super::transaction::TransactionControlExt;
19
20static NEXT_STATEMENT_SAVEPOINT_ID: AtomicU64 = AtomicU64::new(1);
21
22/// Internal dispatch seam between statement coordination and concrete owners.
23pub trait StatementDispatchHost: MutationHost {
24    type NavigationPlan;
25
26    fn dispatch_ddl_fence_already_held(&self) -> bool;
27
28    fn dispatch_authorize_statement(
29        &self,
30        statement: &Statement,
31        context: &ExecutionContext,
32    ) -> Result<()>;
33
34    fn dispatch_bind_navigation(
35        &self,
36        statement: &Statement,
37    ) -> Result<Option<Self::NavigationPlan>>;
38
39    fn dispatch_select(
40        &self,
41        statement: &SelectStatement,
42        navigation: Option<&Self::NavigationPlan>,
43        context: &ExecutionContext,
44    ) -> Result<Box<dyn QueryResult>>;
45    fn dispatch_call(
46        &self,
47        statement: &CallStatement,
48        context: &ExecutionContext,
49    ) -> Result<Box<dyn QueryResult>>;
50
51    fn dispatch_set(
52        &self,
53        statement: &SetStatement,
54        context: &ExecutionContext,
55    ) -> Result<Box<dyn QueryResult>>;
56    fn dispatch_show_tables(
57        &self,
58        statement: &ShowTablesStatement,
59        context: &ExecutionContext,
60    ) -> Result<Box<dyn QueryResult>>;
61    fn dispatch_show_views(
62        &self,
63        statement: &ShowViewsStatement,
64        context: &ExecutionContext,
65    ) -> Result<Box<dyn QueryResult>>;
66    fn dispatch_show_create_table(
67        &self,
68        statement: &ShowCreateTableStatement,
69        context: &ExecutionContext,
70    ) -> Result<Box<dyn QueryResult>>;
71    fn dispatch_show_create_view(
72        &self,
73        statement: &ShowCreateViewStatement,
74        context: &ExecutionContext,
75    ) -> Result<Box<dyn QueryResult>>;
76    fn dispatch_show_indexes(
77        &self,
78        statement: &ShowIndexesStatement,
79        context: &ExecutionContext,
80    ) -> Result<Box<dyn QueryResult>>;
81    fn dispatch_describe(
82        &self,
83        statement: &DescribeStatement,
84        context: &ExecutionContext,
85    ) -> Result<Box<dyn QueryResult>>;
86    fn dispatch_pragma(
87        &self,
88        statement: &PragmaStatement,
89        context: &ExecutionContext,
90    ) -> Result<Box<dyn QueryResult>>;
91    fn dispatch_expression(
92        &self,
93        statement: &ExpressionStatement,
94        context: &ExecutionContext,
95    ) -> Result<Box<dyn QueryResult>>;
96    fn dispatch_explain(
97        &self,
98        statement: &ExplainStatement,
99        context: &ExecutionContext,
100    ) -> Result<Box<dyn QueryResult>>;
101    fn dispatch_analyze(
102        &self,
103        statement: &AnalyzeStatement,
104        context: &ExecutionContext,
105    ) -> Result<Box<dyn QueryResult>>;
106    fn dispatch_vacuum(
107        &self,
108        statement: &VacuumStatement,
109        context: &ExecutionContext,
110    ) -> Result<Box<dyn QueryResult>>;
111}
112
113pub fn execute_statement<H: StatementDispatchHost + ?Sized>(
114    host: &H,
115    statement: &Statement,
116    context: &ExecutionContext,
117    navigation_checked: bool,
118    mut navigation: Option<H::NavigationPlan>,
119) -> Result<Box<dyn QueryResult>> {
120    if matches!(statement, Statement::Expression(_)) {
121        return Err(Error::parse(format!(
122            "invalid SQL: unrecognised statement: {statement}"
123        )));
124    }
125
126    let explicit_transaction = host.mutation_has_active_transaction();
127    let transaction_end = matches!(statement, Statement::Commit(_) | Statement::Rollback(_));
128    let engine_owns_fence = matches!(
129        statement,
130        Statement::Pragma(value)
131            if value.name.value.eq_ignore_ascii_case("SNAPSHOT")
132                || value.name.value.eq_ignore_ascii_case("RESTORE")
133                || value.name.value.eq_ignore_ascii_case("CHECKPOINT")
134    );
135    let transaction_owns_fence = matches!(statement, Statement::CreateTable(_))
136        || (!explicit_transaction
137            && matches!(
138                statement,
139                Statement::CreateExtension(_)
140                    | Statement::DropExtension(_)
141                    | Statement::CreateExternalType(_)
142                    | Statement::DropExternalType(_)
143                    | Statement::CreateOperator(_)
144                    | Statement::DropOperator(_)
145                    | Statement::CreateOperatorClass(_)
146                    | Statement::DropOperatorClass(_)
147                    | Statement::CreatePlannerSupport(_)
148                    | Statement::DropPlannerSupport(_)
149                    | Statement::CreateRoutine(_)
150                    | Statement::CreateTrigger(_)
151                    | Statement::CreateJob(_)
152                    | Statement::DropRoutine(_)
153                    | Statement::DropTrigger(_)
154                    | Statement::DropJob(_)
155                    | Statement::AlterJob(_)
156                    | Statement::CreateSchema(_)
157                    | Statement::CreatePrincipal(_)
158                    | Statement::CreateRole(_)
159                    | Statement::AlterSecuritySubject(_)
160                    | Statement::DropSecuritySubject(_)
161                    | Statement::Grant(_)
162                    | Statement::Revoke(_)
163                    | Statement::AlterOwner(_)
164                    | Statement::CreateIndex(_)
165                    | Statement::DropTable(_)
166                    | Statement::DropIndex(_)
167                    | Statement::AlterIndex(_)
168                    | Statement::CreateView(_)
169                    | Statement::DropView(_)
170            ))
171        || matches!(statement, Statement::AlterTable(_));
172    let coordinates_nested_statements = matches!(statement, Statement::Analyze(_));
173    // Physical DDL takes `ddl_fence` internally, so auto-commit catalog
174    // writers use an independent admission fence across the complete pin ->
175    // stage -> commit interval. Otherwise two writers can pin the same
176    // generation and the loser reports a stale catalog mutation.
177    let _catalog_write_fence = (is_catalog_mutation(statement) && !explicit_transaction)
178        .then(|| host.mutation_engine().acquire_catalog_write_fence());
179    let _catalog_fence = (!host.dispatch_ddl_fence_already_held()
180        && !transaction_end
181        && !coordinates_nested_statements
182        && !engine_owns_fence
183        && !transaction_owns_fence)
184        .then(|| {
185            let exclusive = (is_ddl(statement) && !explicit_transaction)
186                || matches!(statement, Statement::Truncate(_));
187            host.mutation_engine()
188                .acquire_ddl_statement_fence(exclusive)
189        });
190
191    host.dispatch_authorize_statement(statement, context)?;
192
193    if !navigation_checked {
194        navigation = host.dispatch_bind_navigation(statement)?;
195    }
196
197    let _statement_scope = context.enter_statement_scope();
198    let _visibility_fence = matches!(statement, Statement::Select(_))
199        .then(|| host.mutation_engine().acquire_statement_visibility_fence())
200        .flatten();
201    let context = host.mutation_active_transaction_id().map_or_else(
202        || context.clone(),
203        |id| context.with_transaction_id(id as u64),
204    );
205
206    let statement_savepoint = create_statement_savepoint(host, statement, explicit_transaction)?;
207    let result = route_statement(host, statement, &context, navigation.as_ref());
208    finalize_statement_savepoint(host, statement_savepoint, result)
209}
210
211fn is_catalog_mutation(statement: &Statement) -> bool {
212    is_ddl(statement) && !matches!(statement, Statement::Truncate(_))
213}
214
215fn is_ddl(statement: &Statement) -> bool {
216    matches!(
217        statement,
218        Statement::CreateTable(_)
219            | Statement::CreateExtension(_)
220            | Statement::DropExtension(_)
221            | Statement::CreateExternalType(_)
222            | Statement::DropExternalType(_)
223            | Statement::CreateOperator(_)
224            | Statement::DropOperator(_)
225            | Statement::CreateOperatorClass(_)
226            | Statement::DropOperatorClass(_)
227            | Statement::CreateRoutine(_)
228            | Statement::CreateTrigger(_)
229            | Statement::CreateJob(_)
230            | Statement::DropRoutine(_)
231            | Statement::DropTrigger(_)
232            | Statement::DropJob(_)
233            | Statement::AlterJob(_)
234            | Statement::CreateSchema(_)
235            | Statement::CreatePrincipal(_)
236            | Statement::CreateRole(_)
237            | Statement::AlterSecuritySubject(_)
238            | Statement::DropSecuritySubject(_)
239            | Statement::Grant(_)
240            | Statement::Revoke(_)
241            | Statement::AlterOwner(_)
242            | Statement::DropTable(_)
243            | Statement::CreateIndex(_)
244            | Statement::DropIndex(_)
245            | Statement::AlterTable(_)
246            | Statement::AlterIndex(_)
247            | Statement::CreateView(_)
248            | Statement::DropView(_)
249            | Statement::Truncate(_)
250    )
251}
252
253fn route_statement<H: StatementDispatchHost + ?Sized>(
254    host: &H,
255    statement: &Statement,
256    context: &ExecutionContext,
257    navigation: Option<&H::NavigationPlan>,
258) -> Result<Box<dyn QueryResult>> {
259    match statement {
260        Statement::CreateExtension(value) => host.execute_create_extension(value, context),
261        Statement::DropExtension(value) => host.execute_drop_extension(value, context),
262        Statement::CreateExternalType(value) => host.execute_create_external_type(value, context),
263        Statement::DropExternalType(value) => host.execute_drop_external_type(value, context),
264        Statement::CreateOperator(value) => host.execute_create_operator(value, context),
265        Statement::DropOperator(value) => host.execute_drop_operator(value, context),
266        Statement::CreateOperatorClass(value) => host.execute_create_operator_class(value, context),
267        Statement::DropOperatorClass(value) => host.execute_drop_operator_class(value, context),
268        Statement::CreatePlannerSupport(value) => {
269            host.execute_create_planner_support(value, context)
270        }
271        Statement::DropPlannerSupport(value) => host.execute_drop_planner_support(value, context),
272        Statement::CreateTable(value) => host.execute_create_table(value, context),
273        Statement::CreateRoutine(value) => host.execute_create_routine(value, context),
274        Statement::CreateTrigger(value) => host.execute_create_trigger(value, context),
275        Statement::CreateJob(value) => host.execute_create_job(value, context),
276        Statement::DropRoutine(value) => host.execute_drop_routine(value, context),
277        Statement::DropTrigger(value) => host.execute_drop_trigger(value, context),
278        Statement::DropJob(value) => host.execute_drop_job(value, context),
279        Statement::AlterJob(value) => host.execute_alter_job(value, context),
280        Statement::CreateSchema(value) => host.execute_create_schema(value, context),
281        Statement::CreatePrincipal(value) => host.execute_create_principal(value, context),
282        Statement::CreateRole(value) => host.execute_create_role(value, context),
283        Statement::AlterSecuritySubject(value) => {
284            host.execute_alter_security_subject(value, context)
285        }
286        Statement::DropSecuritySubject(value) => host.execute_drop_security_subject(value, context),
287        Statement::Grant(value) => host.execute_grant(value, context),
288        Statement::Revoke(value) => host.execute_revoke(value, context),
289        Statement::AlterOwner(value) => host.execute_alter_owner(value, context),
290        Statement::DropTable(value) => host.execute_drop_table(value, context),
291        Statement::CreateIndex(value) => host.execute_create_index(value, context),
292        Statement::DropIndex(value) => host.execute_drop_index(value, context),
293        Statement::AlterTable(value) => host.execute_alter_table(value, context),
294        Statement::AlterIndex(value) => host.execute_alter_index(value, context),
295        Statement::CreateView(value) => host.execute_create_view(value, context),
296        Statement::DropView(value) => host.execute_drop_view(value, context),
297        Statement::Insert(value) => host.execute_insert(value, context),
298        Statement::Update(value) => host.execute_update(value, context),
299        Statement::Delete(value) => host.execute_delete(value, context),
300        Statement::Truncate(value) => host.execute_truncate(value, context),
301        Statement::Select(value) => host.dispatch_select(value, navigation, context),
302        Statement::Call(value) => host.dispatch_call(value, context),
303        Statement::Begin(value) => host.execute_begin(value, context),
304        Statement::Commit(value) => host.execute_commit_stmt(value, context),
305        Statement::Rollback(value) => host.execute_rollback_stmt(value, context),
306        Statement::Savepoint(value) => host.execute_savepoint(value, context),
307        Statement::ReleaseSavepoint(value) => host.execute_release_savepoint(value, context),
308        Statement::Set(value) => host.dispatch_set(value, context),
309        Statement::ShowTables(value) => host.dispatch_show_tables(value, context),
310        Statement::ShowViews(value) => host.dispatch_show_views(value, context),
311        Statement::ShowCreateTable(value) => host.dispatch_show_create_table(value, context),
312        Statement::ShowCreateView(value) => host.dispatch_show_create_view(value, context),
313        Statement::ShowIndexes(value) => host.dispatch_show_indexes(value, context),
314        Statement::Describe(value) => host.dispatch_describe(value, context),
315        Statement::Pragma(value) => host.dispatch_pragma(value, context),
316        Statement::Expression(value) => host.dispatch_expression(value, context),
317        Statement::Explain(value) => host.dispatch_explain(value, context),
318        Statement::Analyze(value) => host.dispatch_analyze(value, context),
319        Statement::Vacuum(value) => host.dispatch_vacuum(value, context),
320        Statement::Copy(value) => host.execute_copy(value, context),
321    }
322}
323
324fn create_statement_savepoint<H: StatementDispatchHost + ?Sized>(
325    host: &H,
326    statement: &Statement,
327    explicit_transaction: bool,
328) -> Result<Option<String>> {
329    if !explicit_transaction
330        || !matches!(
331            statement,
332            Statement::Insert(_)
333                | Statement::Update(_)
334                | Statement::Delete(_)
335                | Statement::AlterTable(_)
336        )
337    {
338        return Ok(None);
339    }
340
341    let id = NEXT_STATEMENT_SAVEPOINT_ID.fetch_add(1, Ordering::Relaxed);
342    let name = format!("\0radixdb-statement-{id}");
343    let mut active = host.mutation_active_transaction().lock().unwrap();
344    let state = active.as_mut().ok_or_else(|| {
345        Error::internal("explicit transaction disappeared before statement execution")
346    })?;
347    state.create_savepoint(&name)?;
348    Ok(Some(name))
349}
350
351fn finalize_statement_savepoint<H: StatementDispatchHost + ?Sized>(
352    host: &H,
353    savepoint: Option<String>,
354    result: Result<Box<dyn QueryResult>>,
355) -> Result<Box<dyn QueryResult>> {
356    let Some(name) = savepoint else {
357        return result;
358    };
359    let mut active = host.mutation_active_transaction().lock().unwrap();
360    let state = active.as_mut().ok_or_else(|| {
361        Error::internal("explicit transaction disappeared during statement execution")
362    })?;
363
364    match result {
365        Ok(value) => match state.release_savepoint(&name) {
366            Ok(()) => Ok(value),
367            Err(release_error) => {
368                let rollback_error = state.rollback_to_savepoint(&name).err();
369                Err(Error::internal(format!(
370                    "statement completed but its atomic savepoint could not be released: {release_error}{}",
371                    rollback_error.map_or_else(String::new, |error| format!(
372                        "; rollback also failed: {error}"
373                    ))
374                )))
375            }
376        },
377        Err(statement_error) => {
378            if let Err(rollback_error) = state.rollback_to_savepoint(&name) {
379                return Err(Error::internal(format!(
380                    "statement failed: {statement_error}; atomic rollback failed: {rollback_error}"
381                )));
382            }
383            if let Err(release_error) = state.release_savepoint(&name) {
384                return Err(Error::internal(format!(
385                    "statement failed: {statement_error}; rolled back but could not release its savepoint: {release_error}"
386                )));
387            }
388            Err(statement_error)
389        }
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use std::sync::{Arc, Barrier};
396
397    use radixdb_storage::mvcc::MVCCEngine;
398    use radixdb_storage::traits::Engine;
399
400    use crate::Executor;
401
402    #[test]
403    fn concurrent_autocommit_catalog_writers_serialize_pin_to_commit() {
404        let engine = Arc::new(MVCCEngine::in_memory());
405        engine.open_engine().unwrap();
406        let barrier = Arc::new(Barrier::new(3));
407        let mut writers = Vec::new();
408
409        for writer in 0..2 {
410            let engine = Arc::clone(&engine);
411            let barrier = Arc::clone(&barrier);
412            writers.push(std::thread::spawn(move || {
413                let executor = Executor::new(engine);
414                barrier.wait();
415                for table in 0..64 {
416                    executor
417                        .execute(&format!(
418                            "CREATE TABLE concurrent_catalog_{writer}_{table} (id INTEGER PRIMARY KEY)"
419                        ))
420                        .unwrap();
421                }
422            }));
423        }
424
425        barrier.wait();
426        for writer in writers {
427            writer.join().unwrap();
428        }
429
430        for writer in 0..2 {
431            for table in 0..64 {
432                assert!(engine
433                    .table_exists(&format!("concurrent_catalog_{writer}_{table}"))
434                    .unwrap());
435            }
436        }
437    }
438}
radixdb_executor/dispatch/statement.rs

radixdb_executor/dispatch/
statement.rs
