Function execute_planned

nodedb/src/control/server/native/dispatch/sql.rs:142–249

Plan SQL via DataFusion and dispatch tasks to the Data Plane.

(
    ctx: &DispatchCtx<'_>,
    seq: u64,
    sql: &str,
    database_id: crate::types::DatabaseId,
)

/// Plan SQL via DataFusion and dispatch tasks to the Data Plane.
async fn execute_planned(
    ctx: &DispatchCtx<'_>,
    seq: u64,
    sql: &str,
    database_id: crate::types::DatabaseId,
) -> NativeResponse {
    // Extract per-query ON DENY override (e.g., SELECT ... ON DENY ERROR 'CODE' MESSAGE '...').
    let mut auth_ctx = ctx.auth_context.clone();
    let clean_sql =
        crate::control::server::session_auth::extract_and_apply_on_deny(sql, &mut auth_ctx);

    let perm_cache = ctx.state.permission_cache.read().await;
    let sec = crate::control::planner::context::PlanSecurityContext {
        identity: ctx.identity,
        auth: &auth_ctx,
        rls_store: &ctx.state.rls,
        permissions: &ctx.state.permissions,
        roles: &ctx.state.roles,
        permission_cache: Some(&*perm_cache),
    };
    let tasks = match ctx
        .query_ctx
        .plan_sql_with_rls(&clean_sql, ctx.tenant_id(), database_id, &sec)
        .await
    {
        Ok(t) => t,
        Err(e) => return error_to_native(seq, &e),
    };

    if tasks.is_empty() {
        return NativeResponse::status_row(seq, "OK");
    }

    let mut all_columns: Option<Vec<String>> = None;
    let mut all_rows: Vec<Vec<Value>> = Vec::new();
    let mut last_lsn = 0u64;
    let mut total_affected = 0u64;

    for task in tasks {
        if task.tenant_id != ctx.tenant_id() {
            return NativeResponse::error(seq, "42501", "tenant isolation violation");
        }

        // In transaction: buffer writes.
        if ctx.sessions.transaction_state(ctx.peer_addr) == TransactionState::InBlock {
            let is_write = crate::control::wal_replication::to_replicated_entry(
                task.tenant_id,
                task.vshard_id,
                &task.plan,
            )
            .is_some();
            if is_write {
                ctx.sessions.buffer_write(ctx.peer_addr, task);
                total_affected += 1;
                continue;
            }
        }