feat: model pooler mode in ConnectionParams for PgBouncer + custom types

src/client.rs

@@ -71,6 +71,44 @@ pub(crate) fn format_pg_error(e: &tokio_postgres::Error) -> String {
     }
 }
 
+/// Format a `tokio_postgres::Error` from a query path, rewriting
+/// prepared-statement errors (SQLSTATE 26000 / 42P05) with actionable
+/// guidance when the connection is through a transaction-mode pooler.
+///
+/// Under PgBouncer `pool_mode = transaction`, tokio-postgres's internal
+/// `typeinfo_statement` (fired during `Row` deserialization for non-builtin
+/// column types — custom enums, composites, extension types) splits across
+/// two Syncs and can be orphaned by pooler reassignment, producing
+/// `ERROR: prepared statement "typeinfo_statement" does not exist`. The fix
+/// is pooler-side: `max_prepared_statements > 0` in pgbouncer.ini makes
+/// PgBouncer track prepared statements across reassignments. This rewrites
+/// the cryptic error into that guidance so the user knows what to configure
+/// (#157, tabularis#1097).
+///
+/// For non-pooler connections or non-prepared-statement errors, falls
+/// through to `format_pg_error` unchanged.
+pub(crate) fn format_query_error(e: &tokio_postgres::Error, is_transaction_pooler: bool) -> String {
+    if is_transaction_pooler {
+        if let Some(code) = e.code() {
+            if *code == tokio_postgres::error::SqlState::INVALID_SQL_STATEMENT_NAME
+                || *code == tokio_postgres::error::SqlState::DUPLICATE_PSTATEMENT
+            {
+                return format!(
+                    "{}\n\n\
+                     This error occurs under transaction-mode poolers (e.g. PgBouncer) \
+                     when a named prepared statement is lost to connection reassignment. \
+                     For queries returning custom types (enums, composites, extension types), \
+                     set `max_prepared_statements > 0` in pgbouncer.ini — this enables \
+                     server-side prepared statement caching so named statements survive \
+                     reassignment. See: https://www.pgbouncer.org/config.html#max_prepared_statements",
+                    format_pg_error(e)
+                );
+            }
+        }
+    }
+    format_pg_error(e)
+}
+
 /// Format a `pool.get()` failure. The pool's own connection-establishment
 /// handshake (bad database name, bad password, ...) surfaces as a
 /// `tokio_postgres::Error` wrapped in `PoolError::Backend` — the exact same

src/client_live_tests.rs

@@ -48,6 +48,7 @@ fn live_params(password: &str) -> Option<ConnectionParams> {
         ssl_key: None,
         connection_string: None,
         startup_script: None,
+        pooler_mode: None,
     })
 }
 

src/client_tests.rs

@@ -36,6 +36,7 @@ fn params(host: &str, port: u16, db: &str, user: &str) -> ConnectionParams {
         ssl_key: None,
         connection_string: None,
         startup_script: None,
+        pooler_mode: None,
     }
 }
 
@@ -804,6 +805,7 @@ fn params_with_ssl(ssl_mode: &str) -> ConnectionParams {
         ssl_key: None,
         connection_string: None,
         startup_script: None,
+        pooler_mode: None,
     }
 }
 

src/handlers/query.rs

@@ -158,7 +158,14 @@ async fn run_batch_in_session(
     let mut results: Vec<Value> = Vec::new();
     for query in queries {
         let start = Instant::now();
-        let outcome = exec_query_on_client(&pg_client, query, limit, page).await;
+        let outcome = exec_query_on_client(
+            &pg_client,
+            query,
+            limit,
+            page,
+            conn_params.is_transaction_pooler(),
+        )
+        .await;
         let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
 
         in_transaction =
@@ -296,7 +303,14 @@ async fn exec_query(
         apply_search_path(&pg_client, s).await?;
     }
 
-    exec_query_on_client(&pg_client, query, limit, page).await
+    exec_query_on_client(
+        &pg_client,
+        query,
+        limit,
+        page,
+        conn_params.is_transaction_pooler(),
+    )
+    .await
 }
 
 /// Build a `SET search_path TO` statement that prepends `schema` onto the
@@ -386,6 +400,7 @@ async fn exec_query_on_client(
     query: &str,
     limit: Option<u32>,
     page: u32,
+    is_transaction_pooler: bool,
 ) -> Result<Value, String> {
     // Check if the statement returns a result set
     if !returns_result_set(query) {
@@ -427,10 +442,15 @@ async fn exec_query_on_client(
     // Unnamed-statement protocol (`query_typed`) — pooler-safe under
     // transaction-mode poolers (PgBouncer); see `client`'s module note
     // (tabularis#1097). No params for user-issued queries.
+    //
+    // On a transaction-mode pooler, a prepared-statement error (26000/42P05)
+    // from tokio-postgres's internal `typeinfo_statement` (fired by
+    // non-builtin column types) is rewritten with actionable guidance via
+    // `format_query_error` (#157).
     let rows = pg_client
         .query_typed(&final_query, &[])
         .await
-        .map_err(|e| client::format_pg_error(&e))?;
+        .map_err(|e| client::format_query_error(&e, is_transaction_pooler))?;
 
     if rows.is_empty() {
         // Get columns from the statement if possible. `prepare`'s own
@@ -445,14 +465,27 @@ async fn exec_query_on_client(
         // statement prepared and executed across separate Syncs, the same
         // pooler-unsafe pattern this PR eliminates elsewhere. That path is
         // a tokio-postgres 0.7 limitation (see the module note in
-        // `client.rs`); this `prepare` is the only remaining named-
-        // statement call, used only here for column discovery on empty
-        // result sets.
-        let columns: Vec<String> = if let Ok(stmt) = pg_client.prepare(&final_query).await {
-            stmt.columns()
-                .iter()
-                .map(|c| c.name().to_string())
-                .collect()
+        // `client.rs`).
+        //
+        // On a transaction-mode pooler connection, skip `prepare()`
+        // entirely and return `columns: []` — the host handles empty
+        // columns gracefully, and avoiding `prepare()` eliminates the
+        // plugin's own named-statement trigger (#157, tabularis#1097).
+        // The remaining trigger (tokio-postgres's internal
+        // `typeinfo_statement` during `Row` deserialization on the
+        // non-empty-rows path) is handled by the error-message rewrite
+        // above + `max_prepared_statements` on the pooler side. On direct
+        // connections, `prepare()` is still used (it's safe there and
+        // gives better column discovery for empty result sets).
+        let columns: Vec<String> = if !is_transaction_pooler {
+            if let Ok(stmt) = pg_client.prepare(&final_query).await {
+                stmt.columns()
+                    .iter()
+                    .map(|c| c.name().to_string())
+                    .collect()
+            } else {
+                vec![]
+            }
         } else {
             vec![]
         };

src/models.rs

@@ -20,6 +20,16 @@ pub struct ConnectionParams {
     pub ssl_key: Option<String>,
     pub connection_string: Option<String>,
     pub startup_script: Option<String>,
+    /// Connection pooler mode, set by the host when the user declares the
+    /// connection goes through a transaction-mode pooler (e.g. PgBouncer
+    /// `pool_mode = transaction`). When `"transaction"`, the plugin avoids
+    /// its own named-statement `prepare()` calls (which split across two
+    /// Syncs and can be orphaned by pooler reassignment) and surfaces
+    /// actionable error messages for tokio-postgres's internal
+    /// `typeinfo_statement` (which requires `max_prepared_statements > 0`
+    /// on the pooler side for non-builtin column types — see #157,
+    /// tabularis#1097). `None` for direct connections — no behavior change.
+    pub pooler_mode: Option<String>,
 }
 
 impl ConnectionParams {
@@ -48,8 +58,20 @@ impl ConnectionParams {
             ssl_key: get_str("ssl_key"),
             connection_string: get_str("connection_string"),
             startup_script: get_str("startup_script"),
+            pooler_mode: get_str("pooler_mode"),
         }
     }
+
+    /// True when the connection goes through a transaction-mode pooler
+    /// (e.g. PgBouncer `pool_mode = transaction`). Drives pooler-safe
+    /// behavior: the plugin avoids its own named-statement `prepare()` calls
+    /// and surfaces actionable error messages for tokio-postgres's internal
+    /// `typeinfo_statement` (#157, tabularis#1097).
+    pub fn is_transaction_pooler(&self) -> bool {
+        self.pooler_mode
+            .as_deref()
+            .is_some_and(|m| m.eq_ignore_ascii_case("transaction"))
+    }
 }
 
 /// Extract the nested `params` object every RPC method receives.
@@ -82,3 +104,7 @@ pub struct RoutineCallArg {
     #[serde(default)]
     pub is_raw: bool,
 }
+
+#[cfg(test)]
+#[path = "models_tests.rs"]
+mod tests;

src/models_tests.rs

@@ -0,0 +1,74 @@
+//! Unit tests for `ConnectionParams::is_transaction_pooler` and the
+//! `pooler_mode` field parsing (#157).
+
+use crate::models::ConnectionParams;
+use serde_json::json;
+
+fn params_with_pooler_mode(mode: Option<&str>) -> ConnectionParams {
+    ConnectionParams::from_value(&json!({
+        "driver": "postgres",
+        "host": "localhost",
+        "port": 5432,
+        "database": "db",
+        "username": "user",
+        "pooler_mode": mode,
+    }))
+}
+
+#[test]
+fn is_transaction_pooler_true_for_transaction() {
+    assert!(params_with_pooler_mode(Some("transaction")).is_transaction_pooler());
+}
+
+#[test]
+fn is_transaction_pooler_false_for_none() {
+    let params = ConnectionParams::from_value(&json!({
+        "driver": "postgres",
+        "host": "localhost",
+        "port": 5432,
+        "database": "db",
+        "username": "user",
+    }));
+    assert!(!params.is_transaction_pooler());
+}
+
+#[test]
+fn is_transaction_pooler_false_for_session() {
+    assert!(!params_with_pooler_mode(Some("session")).is_transaction_pooler());
+}
+
+#[test]
+fn is_transaction_pooler_false_for_empty_string() {
+    assert!(!params_with_pooler_mode(Some("")).is_transaction_pooler());
+}
+
+#[test]
+fn is_transaction_pooler_false_for_typo() {
+    assert!(!params_with_pooler_mode(Some("transation")).is_transaction_pooler());
+}
+
+/// Case-insensitive: the host might send "Transaction" or "TRANSACTION"
+/// from a differently-cased dropdown label or config file. The comparison
+/// must not silently disable pooler-safe behavior on a valid-but-cased
+/// value (boundary-and-edge-cases finding from the #158 ensemble review).
+#[test]
+fn is_transaction_pooler_case_insensitive() {
+    assert!(params_with_pooler_mode(Some("Transaction")).is_transaction_pooler());
+    assert!(params_with_pooler_mode(Some("TRANSACTION")).is_transaction_pooler());
+    assert!(params_with_pooler_mode(Some("Transaction")).is_transaction_pooler());
+}
+
+/// Non-string JSON for `pooler_mode` (number, bool, null) is silently
+/// treated as None — not a transaction pooler.
+#[test]
+fn is_transaction_pooler_false_for_non_string_json() {
+    let params = ConnectionParams::from_value(&json!({
+        "driver": "postgres",
+        "host": "localhost",
+        "port": 5432,
+        "database": "db",
+        "username": "user",
+        "pooler_mode": 42,
+    }));
+    assert!(!params.is_transaction_pooler());
+}