Function handle_sql

nodedb/src/control/server/native/dispatch/sql.rs:29–139

Handle a SQL statement: transaction control, SET/SHOW, DDL, or DataFusion. `sql_params`, when present, carries the caller's bound values for `$1`, `$2`, … placeholders in `sql`. The handler renders each value as a SQL literal via `value_to_sql_literal` and substitutes the placeholders before any other dispatch — DDL routing, planner, transaction buffer — so every downstream sees one canonical SQL

(
    ctx: &DispatchCtx<'_>,
    seq: u64,
    sql: &str,
    sql_params: Option<&[Value]>,
)

/// string with literal values in place of placeholders. `None` (the
/// common case) routes the SQL through unmodified.
pub(crate) async fn handle_sql(
    ctx: &DispatchCtx<'_>,
    seq: u64,
    sql: &str,
    sql_params: Option<&[Value]>,
) -> NativeResponse {
    // Inline bound parameters before any dispatch — keeps the
    // substitution invariant in one place so the DDL router, planner,
    // and transaction buffer all see the same SQL shape regardless of
    // whether the caller sent params or inlined values directly.
    let substituted: Option<String> = match sql_params {
        Some(params) if !params.is_empty() => match inline_params(sql, params) {
            Ok(s) => Some(s),
            Err(msg) => return NativeResponse::error(seq, "42P02", msg),
        },
        _ => None,
    };
    let sql = substituted.as_deref().unwrap_or(sql);
    let sql_trimmed = sql.trim();
    let upper = sql_trimmed.to_uppercase();

    ctx.sessions.ensure_session(*ctx.peer_addr);

    if sql_trimmed.is_empty() || sql_trimmed == ";" {
        return NativeResponse::ok(seq);
    }

    // Transaction control.
    if upper == "BEGIN" || upper == "BEGIN TRANSACTION" || upper == "START TRANSACTION" {
        return handle_begin(ctx, seq);
    }
    if upper == "COMMIT" || upper == "END" || upper == "END TRANSACTION" {
        return handle_commit(ctx, seq).await;
    }
    if upper == "ROLLBACK" || upper == "ABORT" {
        return handle_rollback(ctx, seq);
    }
    if upper.starts_with("SAVEPOINT ") {
        return NativeResponse::status_row(seq, "SAVEPOINT");
    }
    if upper.starts_with("RELEASE SAVEPOINT ") || upper.starts_with("RELEASE ") {
        return NativeResponse::status_row(seq, "RELEASE");
    }
    if upper.starts_with("ROLLBACK TO ") {
        return NativeResponse::status_row(seq, "ROLLBACK");
    }

    if ctx.sessions.transaction_state(ctx.peer_addr) == TransactionState::Failed {
        return NativeResponse::error(
            seq,
            "25P02",
            "current transaction is aborted, commands ignored until end of transaction block",
        );
    }

    // SET / SHOW / RESET.
    if upper.starts_with("SET ") {
        return handle_set_sql(ctx, seq, sql_trimmed);