Method exe_sql

crates/pg/src/pg_server.rs:155–197

(&self, query: String)

    T: ControlStateReadAccess + ControlStateWriteAccess + NodeDelegate + Authorization + Clone,
{
    async fn exe_sql(&self, query: String) -> PgWireResult<Vec<Response>> {
        let params = self.cached.lock().await.clone().unwrap();
        let db = SqlParams {
            name_or_identity: database::NameOrIdentity::Name(DatabaseName(params.database.clone())),
        };

        let sql = match response(
            database::sql_direct(
                self.ctx.clone(),
                db,
                SqlQueryParams { confirmed: Some(true) },
                params.caller_identity,
                params.caller_auth.clone(),
                query.to_string(),
            )
            .await,
            &params.database,
        )
        .await
        {
            Ok(sql) => sql,
            Err(PgError::Pg(PgWireError::UserError(err))) => {
                return Ok(vec![Response::Error(err)]);
            }
            Err(err) => {
                return Err(err.into());
            }
        };

        let mut result = Vec::with_capacity(sql.len());
        for sql_result in sql {
            let header = row_desc(&sql_result.schema, &Format::UnifiedText);
            if !sql_result.schema.is_empty() {
                let rows = to_rows(sql_result, header.clone())?;
                let q = QueryResponse::new(header, rows);
                result.push(Response::Query(q));
            } else {
                let tag = Tag::new(&stats(&sql_result));
                result.push(Response::Execution(tag));
            }
        }
        Ok(result)
    }
}

async fn close_client<C, E>(client: &mut C, err: E) -> PgWireResult<()>