citadel_sql/executor/
mod.rs

1//! SQL executor: DDL and DML operations.
2
3mod aggregate;
4pub(crate) mod compile;
5mod correlated;
6mod cte;
7mod ddl;
8mod dml;
9mod explain;
10pub(crate) mod helpers;
11mod join;
12mod scan;
13mod select;
14mod view;
15mod window;
16pub(crate) mod write;
17pub use compile::{compile, CompiledPlan};
18use cte::*;
19use ddl::*;
20pub use dml::exec_insert_in_txn;
21use dml::*;
22use explain::*;
23use join::*;
24use scan::*;
25use select::*;
26use view::*;
27use window::*;
28use write::*;
29
30use citadel::Database;
31use rustc_hash::FxHashMap;
32
33use crate::error::{Result, SqlError};
34use crate::parser::*;
35use crate::schema::SchemaManager;
36use crate::types::*;
37
38type CteContext = FxHashMap<String, QueryResult>;
39type ScanTableFn<'a> = &'a mut dyn FnMut(&str) -> Result<(TableSchema, Vec<Vec<Value>>)>;
40
41pub fn execute(
42    db: &Database,
43    schema: &mut SchemaManager,
44    stmt: &Statement,
45    params: &[Value],
46) -> Result<ExecutionResult> {
47    match stmt {
48        Statement::CreateTable(ct) => exec_create_table(db, schema, ct),
49        Statement::DropTable(dt) => exec_drop_table(db, schema, dt),
50        Statement::CreateIndex(ci) => exec_create_index(db, schema, ci),
51        Statement::DropIndex(di) => exec_drop_index(db, schema, di),
52        Statement::CreateView(cv) => exec_create_view(db, schema, cv),
53        Statement::DropView(dv) => exec_drop_view(db, schema, dv),
54        Statement::AlterTable(at) => exec_alter_table(db, schema, at),
55        Statement::Insert(ins) => exec_insert(db, schema, ins, params),
56        Statement::Select(sq) => exec_select_query(db, schema, sq),
57        Statement::Update(upd) => exec_update(db, schema, upd),
58        Statement::Delete(del) => exec_delete(db, schema, del),
59        Statement::Explain(inner) => explain(schema, inner),
60        Statement::Begin
61        | Statement::Commit
62        | Statement::Rollback
63        | Statement::Savepoint(_)
64        | Statement::ReleaseSavepoint(_)
65        | Statement::RollbackTo(_)
66        | Statement::SetTimezone(_) => Err(SqlError::Unsupported(
67            "transaction / session control handled by Connection".into(),
68        )),
69    }
70}
71
72/// Execute a parsed SQL statement within an existing write transaction.
73pub fn execute_in_txn(
74    wtx: &mut citadel_txn::write_txn::WriteTxn<'_>,
75    schema: &mut SchemaManager,
76    stmt: &Statement,
77    params: &[Value],
78) -> Result<ExecutionResult> {
79    match stmt {
80        Statement::CreateTable(ct) => exec_create_table_in_txn(wtx, schema, ct),
81        Statement::DropTable(dt) => exec_drop_table_in_txn(wtx, schema, dt),
82        Statement::CreateIndex(ci) => exec_create_index_in_txn(wtx, schema, ci),
83        Statement::DropIndex(di) => exec_drop_index_in_txn(wtx, schema, di),
84        Statement::CreateView(cv) => exec_create_view_in_txn(wtx, schema, cv),
85        Statement::DropView(dv) => exec_drop_view_in_txn(wtx, schema, dv),
86        Statement::AlterTable(at) => exec_alter_table_in_txn(wtx, schema, at),
87        Statement::Insert(ins) => exec_insert_in_txn(wtx, schema, ins, params),
88        Statement::Select(sq) => exec_select_query_in_txn(wtx, schema, sq),
89        Statement::Update(upd) => exec_update_in_txn(wtx, schema, upd),
90        Statement::Delete(del) => exec_delete_in_txn(wtx, schema, del),
91        Statement::Explain(inner) => explain(schema, inner),
92        Statement::Begin
93        | Statement::Commit
94        | Statement::Rollback
95        | Statement::Savepoint(_)
96        | Statement::ReleaseSavepoint(_)
97        | Statement::RollbackTo(_)
98        | Statement::SetTimezone(_) => {
99            Err(SqlError::Unsupported("nested transaction control".into()))
100        }
101    }
102}
103
104pub(super) fn scan_table_read(
105    db: &Database,
106    schema: &SchemaManager,
107    name: &str,
108) -> Result<(TableSchema, Vec<Vec<Value>>)> {
109    let table_schema = schema
110        .get(name)
111        .ok_or_else(|| SqlError::TableNotFound(name.to_string()))?;
112    let (rows, _) = collect_rows_read(db, table_schema, &None, None)?;
113    Ok((table_schema.clone(), rows))
114}
115
116pub(super) fn scan_table_read_or_view(
117    db: &Database,
118    schema: &SchemaManager,
119    name: &str,
120) -> Result<(TableSchema, Vec<Vec<Value>>)> {
121    if let Some(ts) = schema.get(name) {
122        let (rows, _) = collect_rows_read(db, ts, &None, None)?;
123        return Ok((ts.clone(), rows));
124    }
125    if let Some(vd) = schema.get_view(name) {
126        let qr = exec_view_read(db, schema, vd)?;
127        let vs = build_view_schema(name, &qr);
128        return Ok((vs, qr.rows));
129    }
130    Err(SqlError::TableNotFound(name.to_string()))
131}
132
133pub(super) fn scan_table_write(
134    wtx: &mut citadel_txn::write_txn::WriteTxn<'_>,
135    schema: &SchemaManager,
136    name: &str,
137) -> Result<(TableSchema, Vec<Vec<Value>>)> {
138    let table_schema = schema
139        .get(name)
140        .ok_or_else(|| SqlError::TableNotFound(name.to_string()))?;
141    let (rows, _) = collect_rows_write(wtx, table_schema, &None, None)?;
142    Ok((table_schema.clone(), rows))
143}
144
145pub(super) fn scan_table_write_or_view(
146    wtx: &mut citadel_txn::write_txn::WriteTxn<'_>,
147    schema: &SchemaManager,
148    name: &str,
149) -> Result<(TableSchema, Vec<Vec<Value>>)> {
150    if let Some(ts) = schema.get(name) {
151        let (rows, _) = collect_rows_write(wtx, ts, &None, None)?;
152        return Ok((ts.clone(), rows));
153    }
154    if let Some(vd) = schema.get_view(name) {
155        let qr = exec_view_write(wtx, schema, vd)?;
156        let vs = build_view_schema(name, &qr);
157        return Ok((vs, qr.rows));
158    }
159    Err(SqlError::TableNotFound(name.to_string()))
160}
161
162pub(super) fn resolve_table_or_cte(
163    name: &str,
164    ctes: &CteContext,
165    scan_table: ScanTableFn<'_>,
166) -> Result<(TableSchema, Vec<Vec<Value>>)> {
167    let lower = name.to_ascii_lowercase();
168    if let Some(cte_result) = ctes.get(&lower) {
169        let schema = build_cte_schema(&lower, cte_result);
170        Ok((schema, cte_result.rows.clone()))
171    } else {
172        scan_table(&lower)
173    }
174}
175
176pub(super) fn exec_select_join_with_ctes(
177    stmt: &SelectStmt,
178    ctes: &CteContext,
179    scan_table: ScanTableFn<'_>,
180) -> Result<ExecutionResult> {
181    let (from_schema, from_rows) = resolve_table_or_cte(&stmt.from, ctes, scan_table)?;
182    let from_alias = table_alias_or_name(&stmt.from, &stmt.from_alias);
183
184    let mut tables: Vec<(String, TableSchema)> = vec![(from_alias.clone(), from_schema)];
185    let mut join_rows: Vec<Vec<Vec<Value>>> = Vec::new();
186
187    for join in &stmt.joins {
188        let jname = &join.table.name;
189        let (js, jrows) = resolve_table_or_cte(jname, ctes, scan_table)?;
190        let jalias = table_alias_or_name(jname, &join.table.alias);
191        tables.push((jalias, js));
192        join_rows.push(jrows);
193    }
194
195    let mut outer_rows = from_rows;
196    let mut cur_tables: Vec<(String, &TableSchema)> = vec![(from_alias.clone(), &tables[0].1)];
197
198    for (ji, join) in stmt.joins.iter().enumerate() {
199        let inner_schema = &tables[ji + 1].1;
200        let inner_alias = &tables[ji + 1].0;
201        let inner_rows = &join_rows[ji];
202
203        let mut preview_tables = cur_tables.clone();
204        preview_tables.push((inner_alias.clone(), inner_schema));
205        let combined_cols = build_joined_columns(&preview_tables);
206
207        let outer_col_count = if outer_rows.is_empty() {
208            cur_tables.iter().map(|(_, s)| s.columns.len()).sum()
209        } else {
210            outer_rows[0].len()
211        };
212        let inner_col_count = inner_schema.columns.len();
213
214        outer_rows = exec_join_step(
215            outer_rows,
216            inner_rows,
217            join,
218            &combined_cols,
219            outer_col_count,
220            inner_col_count,
221            None,
222            None,
223        );
224        cur_tables.push((inner_alias.clone(), inner_schema));
225    }
226
227    let joined_cols = build_joined_columns(&cur_tables);
228    process_select(&joined_cols, outer_rows, stmt, false)
229}
citadel_sql/executor/mod.rs

citadel_sql/executor/
mod.rs
