Method do_query

nodedb/src/control/server/pgwire/handler/core.rs:305–416

(&self, client: &mut C, query: &str)

#[async_trait]
impl SimpleQueryHandler for NodeDbPgHandler {
    async fn do_query<C>(&self, client: &mut C, query: &str) -> PgWireResult<Vec<Response>>
    where
        C: ClientInfo + ClientPortalStore + Sink<PgWireBackendMessage> + Unpin + Send + Sync,
        C::Error: Debug,
        PgWireError: From<<C as Sink<PgWireBackendMessage>>::Error>,
    {
        let addr = client.socket_addr();
        self.sessions.ensure_session(addr);

        let identity = self.resolve_identity(client, &addr)?;
        self.enforce_database_access(&identity, &addr)?;

        // Emit db.id / db.name trace fields at session bind so that any
        // downstream spans inherit the database context.
        let current_db = self
            .sessions
            .get_current_database(&addr)
            .unwrap_or(crate::types::DatabaseId::DEFAULT);
        let db_name: String = self
            .state
            .credentials
            .catalog()
            .as_ref()
            .and_then(|cat| cat.get_database(current_db).ok().flatten())
            .map(|d| d.name.clone())
            .unwrap_or_else(|| "default".to_string());
        tracing::debug!(
            db.id = current_db.as_u64(),
            db.name = %db_name,
            user = %identity.username,
            "session query dispatch",
        );

        // Send notice if BEGIN is called (advisory transactions).
        let upper = query.trim().to_uppercase();
        if (upper == "BEGIN" || upper == "BEGIN TRANSACTION" || upper == "START TRANSACTION")
            && self.sessions.transaction_state(&addr) == TransactionState::InBlock
        {
            let notice = notice_warning("there is already a transaction in progress");
            let _ = client
                .send(PgWireBackendMessage::NoticeResponse(notice))
                .await;
        }

        if (upper == "COMMIT" || upper == "END")
            && self.sessions.transaction_state(&addr) == TransactionState::Idle
        {
            let notice = notice_warning("there is no transaction in progress");
            let _ = client
                .send(PgWireBackendMessage::NoticeResponse(notice))
                .await;
        }

        // J.4: install the DDL audit context for this statement. Any
        // `propose_catalog_entry` call reached from `execute_sql`
        // picks up the identity + raw SQL so the applier can emit a
        // full audit record on every replica. The guard auto-clears
        // on scope exit.