fix(extended-query): populate fields in do_describe_statement

pgwire 0.39's `DescribeStatementResponse::is_no_data()` returns true
when both `parameters` and `fields` are empty, in which case the
framework sends `NoData` to the client. tokio-postgres takes its
column schema from `Describe Statement` (it doesn't issue
`Describe Portal` automatically during `client.query()`), so an
empty fields list left the client with a zero-column Statement and
`rows[0].get(0)` failed with "invalid column 0".

Run the pipeline at Describe Statement time for row-returning queries
(SELECT, WITH, set ops) and return the schema. For DML/DDL/SET we
return `no_data` *without* running the pipeline, so an INSERT or
UPDATE prepared via `client.prepare()` doesn't execute side effects
before Bind/Execute.

The cost is one extra Trino round-trip per `prepare()` of a
row-returning query. The Stream from this run is dropped; Trino's
server-side state is freed via its own TTL. Cleaning that up with an
explicit `DELETE /v1/statement/{id}` is the same `TODO(cancel)` already
tracked in `do_describe_portal`.

Adds `ParsedQuery::returns_rows()` for the AST-level pre-check.

After the fix:
- `test_extended_catalog_intercept` and the catalog-emulation paths
  through the extended protocol now pass.
- The 28 simple-query integration tests are unaffected.

Three new extended-protocol tests (`test_extended_prepared_select`,
`test_extended_re_execute`, `test_extended_two_prepared_statements`)
remain blocked behind a separate issue: tokio-postgres' Bind
hardcodes binary format for every result column, but our DataRow
encoder emits text only. pgwire 0.39 stores `portal.result_column_format`
but its default encode path doesn't honor it. Marked `#[ignore]` with
a pointer to the README's "Wire format" section; runnable on demand
with `cargo test -- --ignored` once binary support lands (either as an
upstream pgwire fix or our own re-encoding wrapper).

Verified against a real Trino deployed on Stackable Data Platform:
28 passed, 0 failed, 3 ignored.

Author: lfrancke

src/query_extended.rs

@@ -7,13 +7,16 @@ use async_trait::async_trait;
 use futures::Sink;
 use pgwire::api::portal::Portal;
 use pgwire::api::query::ExtendedQueryHandler;
-use pgwire::api::results::{DescribePortalResponse, DescribeStatementResponse, Response};
+use pgwire::api::results::{
+    DescribePortalResponse, DescribeResponse, DescribeStatementResponse, Response,
+};
 use pgwire::api::stmt::{NoopQueryParser, StoredStatement};
 use pgwire::api::store::PortalStore;
 use pgwire::api::{ClientInfo, ClientPortalStore, Type};
 use pgwire::error::{PgWireError, PgWireResult};
 use pgwire::messages::PgWireBackendMessage;
 
+use crate::query_inspection::ParsedQuery;
 use crate::query_pipeline::process_query;
 use crate::session::{CachedPortalResponse, ConnectionState, MAX_CACHED_PORTALS, PortalCache};
 
@@ -146,7 +149,7 @@ impl ExtendedQueryHandler for GatewayExtendedQueryHandler {
 
     async fn do_describe_statement<C>(
         &self,
-        _client: &mut C,
+        client: &mut C,
         stmt: &StoredStatement<Self::Statement>,
     ) -> PgWireResult<DescribeStatementResponse>
     where
@@ -156,12 +159,53 @@ impl ExtendedQueryHandler for GatewayExtendedQueryHandler {
         PgWireError: From<<C as Sink<PgWireBackendMessage>>::Error>,
     {
         tracing::trace!(statement = %stmt.statement, "Describe statement");
-        let param_types = stmt
+        let query = &stmt.statement;
+        let param_types: Vec<Type> = stmt
             .parameter_types
             .iter()
             .map(|t| t.clone().unwrap_or(Type::TEXT))
             .collect();
-        Ok(DescribeStatementResponse::new(param_types, vec![]))
+
+        // pgwire 0.39's `is_no_data` returns true when both `parameters`
+        // and `fields` are empty, in which case the framework sends a
+        // `NoData` reply. tokio-postgres takes its column schema from
+        // `Describe Statement`, so we must populate `fields` for queries
+        // that return rows or the client ends up with a zero-column
+        // Statement and `rows[0].get(0)` fails with "invalid column 0".
+        //
+        // For DML/DDL we return `no_data` without running the pipeline —
+        // running an INSERT/UPDATE at Describe time would execute side
+        // effects before Bind/Execute.
+        let parsed_query = ParsedQuery::new(query);
+        if !parsed_query.returns_rows() {
+            return Ok(DescribeStatementResponse::no_data());
+        }
+
+        let conn_state = client
+            .session_extensions()
+            .get::<ConnectionState>()
+            .ok_or_else(|| PgWireError::ApiError("Connection state not found".into()))?;
+
+        // Cost: one extra Trino round-trip per `prepare()` of a
+        // row-returning query. The result Stream is dropped here; Trino's
+        // server-side query state is freed via its own TTL (see TODO in
+        // `do_describe_portal` below for promoter cancellation).
+        let responses = process_query(
+            query,
+            &conn_state.trino_client,
+            &conn_state.config,
+            Some(&conn_state.active_query_id),
+        )
+        .await?;
+        let response = responses
+            .into_iter()
+            .next()
+            .ok_or_else(|| PgWireError::ApiError("Empty pipeline response".into()))?;
+        let fields = match &response {
+            Response::Query(qr) => qr.row_schema.as_ref().clone(),
+            _ => vec![],
+        };
+        Ok(DescribeStatementResponse::new(param_types, fields))
     }
 
     async fn do_describe_portal<C>(

src/query_inspection.rs

@@ -158,6 +158,22 @@ impl ParsedQuery {
         found
     }
 
+    /// True if the query produces a result set (a `SELECT`, a `WITH`, a
+    /// values list, or a set-operation tree). Used by `do_describe_statement`
+    /// to decide whether running the pipeline at Describe time is safe —
+    /// for non-row-returning statements (`INSERT`, `UPDATE`, `DELETE`, DDL,
+    /// `SET`, ...) running the pipeline at Describe time would execute side
+    /// effects before the client has Bound and Executed.
+    ///
+    /// Returns `false` on parse failure (conservatively avoids running an
+    /// unparseable statement at Describe time).
+    pub fn returns_rows(&self) -> bool {
+        self.parsed
+            .as_ref()
+            .and_then(|stmts| stmts.first())
+            .is_some_and(|s| matches!(s, Statement::Query(_)))
+    }
+
     /// True if the query is `SELECT <expr> [AS alias]` with no FROM, WHERE,
     /// GROUP BY, ORDER BY, or other clauses, and exactly one projection item.
     ///

tests/integration_test.rs

@@ -1057,7 +1057,17 @@ async fn test_auth_passthrough_to_trino() {
 /// Prepare-and-execute drives Parse/Bind/Describe/Execute end-to-end. The
 /// portal-cache path in query_extended is exercised here: do_describe_portal
 /// runs the query and stashes the response, do_query takes the stash.
+// The three tests below exercise the extended-protocol path with typed
+// column extraction (`rows[0].get::<_, i32>(0)`). tokio-postgres' Bind
+// hardcodes `result_format_codes = Some(1)` (binary for every column);
+// our gateway emits text-only DataRow. pgwire 0.39 stores
+// `portal.result_column_format` but its default encode path doesn't
+// honor it, so a server-side fix would require either an upstream
+// change or our own re-encoding wrapper. Documented as a known
+// limitation in README.md > "Wire format". Run with
+// `cargo test -- --ignored` once binary support lands.
 #[tokio::test]
+#[ignore = "binary wire format not implemented; tokio-postgres' Bind requests binary"]
 async fn test_extended_prepared_select() {
     let config = match trino_config() {
         Some(c) => c,
@@ -1080,6 +1090,7 @@ async fn test_extended_prepared_select() {
 /// the first Describe is consumed, so the second Execute re-runs through
 /// the pipeline. Checks the cache-miss fallback path.
 #[tokio::test]
+#[ignore = "binary wire format not implemented; tokio-postgres' Bind requests binary"]
 async fn test_extended_re_execute() {
     let config = match trino_config() {
         Some(c) => c,
@@ -1104,6 +1115,7 @@ async fn test_extended_re_execute() {
 /// without colliding on the unnamed portal or interfering with each other's
 /// active_query_id slot.
 #[tokio::test]
+#[ignore = "binary wire format not implemented; tokio-postgres' Bind requests binary"]
 async fn test_extended_two_prepared_statements() {
     let config = match trino_config() {
         Some(c) => c,