Try to handle a SQL statement as a Control Plane DDL command. These execute directly on the Control Plane without going through DataFusion or the Data Plane. Returns `None` if not recognized. Async because DSL commands (SEARCH, CRDT) dispatch to the Data Plane and must await the response without blocking the Tokio runtime.
(
state: &SharedState,
identity: &AuthenticatedIdentity,
sql: &str,
database_id: DatabaseId,
)
| 26 | /// Async because DSL commands (SEARCH, CRDT) dispatch to the Data Plane |
| 27 | /// and must await the response without blocking the Tokio runtime. |
| 28 | pub async fn dispatch( |
| 29 | state: &SharedState, |
| 30 | identity: &AuthenticatedIdentity, |
| 31 | sql: &str, |
| 32 | database_id: DatabaseId, |
| 33 | ) -> Option<PgWireResult<Vec<Response>>> { |
| 34 | // AST-typed fast path: parse once, handle IF [NOT] EXISTS at the |
| 35 | // dispatch level, then fall through to legacy handlers for the |
| 36 | // actual execution. This is the incremental migration path — |
| 37 | // once every legacy handler has been ported to accept a typed |
| 38 | // NodedbStatement, the string-prefix routers below can be |
| 39 | // removed entirely. |
| 40 | match nodedb_sql::ddl_ast::parse(sql) { |
| 41 | Some(Err(e)) => { |
| 42 | // UnsupportedConstraint → 0A000 (feature_not_supported). |
| 43 | // All other parse errors → 42601 (syntax error). |
| 44 | let sqlstate = match &e { |
| 45 | nodedb_sql::SqlError::UnsupportedConstraint { .. } => "0A000", |
| 46 | _ => "42601", |
| 47 | }; |
| 48 | return Some(Err(super::super::types::sqlstate_error( |
| 49 | sqlstate, |
| 50 | &e.to_string(), |
| 51 | ))); |
| 52 | } |
| 53 | Some(Ok(stmt)) => { |
| 54 | if let Some(r) = ast::try_dispatch(state, identity, &stmt, database_id).await { |
| 55 | return Some(r); |
| 56 | } |
| 57 | } |
| 58 | None => {} |
| 59 | } |
| 60 | |
| 61 | let upper = sql.to_uppercase(); |
| 62 | let parts: Vec<&str> = sql.split_whitespace().collect(); |
| 63 | |
| 64 | if let Some(r) = auth::dispatch(state, identity, sql, &upper, &parts).await { |
| 65 | return Some(r); |
| 66 | } |
| 67 | |
| 68 | if let Some(r) = function::dispatch(state, identity, sql, &upper, &parts).await { |
| 69 | return Some(r); |
| 70 | } |
| 71 | |
| 72 | if let Some(r) = streaming::dispatch(state, identity, sql, &upper, &parts).await { |
| 73 | return Some(r); |
| 74 | } |
| 75 | |
| 76 | if let Some(r) = engine_ops::dispatch(state, identity, sql, &upper, &parts, database_id).await { |
| 77 | return Some(r); |
| 78 | } |
| 79 | |
| 80 | if let Some(r) = schema::dispatch(state, identity, sql, &upper, &parts).await { |
| 81 | return Some(r); |
| 82 | } |
| 83 | |
| 84 | if let Some(r) = collaborative::dispatch(state, identity, sql, &upper, &parts).await { |
| 85 | return Some(r); |