fix(pgwire): decode binary-format extended-query parameters

Previously the bind layer decoded every portal parameter as UTF-8 text
regardless of wire format, so clients that declare parameter types via
prepare_typed (which causes tokio-postgres to send binary-encoded
values) had their BOOL/INT2/INT4/INT8/FLOAT4/FLOAT8 parameters rejected
with a bogus "invalid UTF-8" error. Binary scalars are now decoded via
their postgres_types::FromSql encodings; NUMERIC/TIMESTAMP/TIMESTAMPTZ
and other unmodeled binary types are explicitly rejected rather than
silently misread as text.

Extracts the parameter conversion logic into its own param_bind module
and adds regression coverage for binary-format decoding across the
scalar types.

Author: farhan-syah

Cargo.lock

@@ -242,6 +242,7 @@ pco = "1"
 pgwire = { version = "0.38", default-features = false, features = [
   "server-api-aws-lc-rs",
 ] }
+postgres-types = "0.2"
 tokio-postgres = "0.7"
 async-trait = "0.1"
 futures = "0.3"

Cargo.toml

@@ -125,6 +125,7 @@ reqwest = { workspace = true }
 
 # PostgreSQL wire protocol
 pgwire = { workspace = true }
+postgres-types = { workspace = true }
 futures = { workspace = true }
 async-stream = { workspace = true }
 

nodedb/Cargo.toml

@@ -8,18 +8,18 @@
 
 use std::fmt::Debug;
 
-use bytes::Bytes;
 use futures::sink::Sink;
 use pgwire::api::portal::Portal;
 use pgwire::api::results::Response;
-use pgwire::api::{ClientInfo, ClientPortalStore, Type};
+use pgwire::api::{ClientInfo, ClientPortalStore};
 use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
 use pgwire::messages::PgWireBackendMessage;
 
 use crate::control::server::response_shape::schema::{OutputColumn, OutputSchema};
 
 use super::super::core::NodeDbPgHandler;
 use super::super::routing::result_shaping::ResultShaping;
+use super::param_bind::convert_portal_params;
 use super::result_format::{pg_type_to_ddl_col_type, resolve_result_formats};
 use super::statement::ParsedStatement;
 
@@ -149,326 +149,3 @@ impl NodeDbPgHandler {
         Ok(results.pop().unwrap_or(Response::EmptyQuery))
     }
 }
-
-/// Convert pgwire portal parameters to typed `ParamValue` for AST-level binding.
-///
-/// Uses per-parameter format codes from the pgwire 0.38 `Format` API to determine
-/// whether each parameter was sent in text or binary format.
-///
-/// Binary-format NUMERIC, TIMESTAMP, and TIMESTAMPTZ parameters are explicitly
-/// rejected with SQLSTATE 0A000 — their binary encodings are client-library-specific
-/// structs that would produce corrupt values if decoded naively. Clients must use
-/// text format for these types.
-fn convert_portal_params(
-    params: &[Option<Bytes>],
-    param_types: &[Option<Type>],
-    param_format: &pgwire::api::portal::Format,
-) -> PgWireResult<Vec<nodedb_sql::ParamValue>> {
-    let mut result = Vec::with_capacity(params.len());
-    for (i, param) in params.iter().enumerate() {
-        let pg_type = param_types
-            .get(i)
-            .and_then(|t| t.as_ref())
-            .unwrap_or(&Type::UNKNOWN);
-
-        let pv = match param {
-            None => nodedb_sql::ParamValue::Null,
-            Some(bytes) => {
-                // Reject binary format for types whose binary encoding is
-                // client-library-specific and cannot be decoded portably.
-                if param_format.is_binary(i) {
-                    let type_name = if *pg_type == Type::NUMERIC {
-                        Some("NUMERIC")
-                    } else if *pg_type == Type::TIMESTAMP {
-                        Some("TIMESTAMP")
-                    } else if *pg_type == Type::TIMESTAMPTZ {
-                        Some("TIMESTAMPTZ")
-                    } else {
-                        None
-                    };
-                    if let Some(name) = type_name {
-                        return Err(PgWireError::UserError(Box::new(ErrorInfo::new(
-                            "ERROR".to_owned(),
-                            "0A000".to_owned(),
-                            format!(
-                                "binary {name} parameter format is not supported for \
-                                 parameter ${n}; use text format",
-                                n = i + 1
-                            ),
-                        ))));
-                    }
-                }
-
-                let text = std::str::from_utf8(bytes).map_err(|_| {
-                    PgWireError::UserError(Box::new(ErrorInfo::new(
-                        "ERROR".to_owned(),
-                        "22021".to_owned(),
-                        format!("invalid UTF-8 in parameter ${}", i + 1),
-                    )))
-                })?;
-
-                pgwire_text_to_param(text, pg_type)
-            }
-        };
-        result.push(pv);
-    }
-    Ok(result)
-}
-
-/// Convert a pgwire text parameter + declared type to a typed
-/// `ParamValue` for AST/DSL binding.
-///
-/// # Type coverage
-///
-/// Natively decoded: `BOOL`, `INT2`/`INT4`/`INT8`, `FLOAT4`/`FLOAT8`/
-/// `NUMERIC`, `TIMESTAMP`, `TIMESTAMPTZ`, `TEXT`/`VARCHAR` (implicit via
-/// fall-through), and `UNKNOWN` (the untyped-driver path).
-///
-/// # TIMESTAMP / TIMESTAMPTZ
-///
-/// Text-format TIMESTAMP and TIMESTAMPTZ parameters are parsed directly to
-/// `ParamValue::Timestamp` / `ParamValue::Timestamptz`. This produces the
-/// correct typed `SqlValue` variant (Timestamp vs Timestamptz) through the
-/// resolver, ensuring the planner and engine see the right column type rather
-/// than a generic string that must be coerced.
-///
-/// If parsing fails the text is passed through as `ParamValue::Text` so the
-/// engine's string-coercion path can attempt a best-effort conversion — the
-/// same as all other text-passthrough types.
-///
-/// # Fallback policy (catch-all arm)
-///
-/// Types the bind layer does not decode natively — `DATE`, `TIME`, `BYTEA`,
-/// `UUID`, `JSON`, `JSONB`, `INTERVAL`, array types, and user-defined types —
-/// fall through to `ParamValue::Text(text)`. The pgwire text representation of
-/// these types is well-defined and the AST bind emits it as a
-/// `SingleQuotedString`. Downstream, the planner/engine type-coerces the text
-/// via the same path used for literal strings in simple-query SQL.
-///
-/// Binary-format parameters are handled at a layer above this function
-/// (see `convert_portal_params`); they never reach this function.
-///
-/// # Why not error on unknown types
-///
-/// Postgres itself accepts text representations of every built-in type through
-/// the extended-query protocol; refusing here would break drivers that
-/// legitimately send dates/UUIDs/etc. as text.
-fn pgwire_text_to_param(text: &str, pg_type: &Type) -> nodedb_sql::ParamValue {
-    match *pg_type {
-        Type::BOOL => {
-            let lower = text.to_lowercase();
-            if lower == "t" || lower == "true" || lower == "1" {
-                return nodedb_sql::ParamValue::Bool(true);
-            }
-            if lower == "f" || lower == "false" || lower == "0" {
-                return nodedb_sql::ParamValue::Bool(false);
-            }
-            nodedb_sql::ParamValue::Text(text.to_string())
-        }
-        Type::INT2 | Type::INT4 | Type::INT8 => {
-            if let Ok(n) = text.parse::<i64>() {
-                return nodedb_sql::ParamValue::Int64(n);
-            }
-            nodedb_sql::ParamValue::Text(text.to_string())
-        }
-        Type::FLOAT4 | Type::FLOAT8 => {
-            if let Ok(f) = text.parse::<f64>() {
-                return nodedb_sql::ParamValue::Float64(f);
-            }
-            nodedb_sql::ParamValue::Text(text.to_string())
-        }
-        Type::NUMERIC => {
-            // Parse NUMERIC as exact Decimal, not lossy f64.
-            if let Ok(d) = rust_decimal::Decimal::from_str_exact(text) {
-                return nodedb_sql::ParamValue::Decimal(d);
-            }
-            // If parsing fails, return typed error — do not fall back to Float
-            // since that would silently lose precision.
-            nodedb_sql::ParamValue::Text(text.to_string())
-        }
-        Type::TIMESTAMP => {
-            // Parse ISO 8601 / PostgreSQL timestamp text to a typed NaiveDateTime.
-            if let Some(dt) = nodedb_types::datetime::NdbDateTime::parse(text) {
-                return nodedb_sql::ParamValue::Timestamp(dt);
-            }
-            nodedb_sql::ParamValue::Text(text.to_string())
-        }
-        Type::TIMESTAMPTZ => {
-            // Parse ISO 8601 / PostgreSQL timestamptz text to a typed DateTime (UTC).
-            if let Some(dt) = nodedb_types::datetime::NdbDateTime::parse(text) {
-                return nodedb_sql::ParamValue::Timestamptz(dt);
-            }
-            nodedb_sql::ParamValue::Text(text.to_string())
-        }
-        // Text-passthrough types: wire-format text is already the
-        // canonical representation. Engine performs type coercion.
-        _ => nodedb_sql::ParamValue::Text(text.to_string()),
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use pgwire::api::portal::Format;
-
-    use super::*;
-
-    fn text_format() -> Format {
-        Format::UnifiedText
-    }
-
-    fn binary_format() -> Format {
-        Format::UnifiedBinary
-    }
-
-    #[test]
-    fn convert_null_param() {
-        let params = vec![None];
-        let types = vec![Some(Type::INT8)];
-        let result = convert_portal_params(&params, &types, &text_format()).unwrap();
-        assert_eq!(result.len(), 1);
-        assert!(matches!(result[0], nodedb_sql::ParamValue::Null));
-    }
-
-    #[test]
-    fn convert_typed_params() {
-        let params = vec![
-            Some(Bytes::from_static(b"42")),
-            Some(Bytes::from_static(b"hello")),
-            Some(Bytes::from_static(b"true")),
-        ];
-        let types = vec![Some(Type::INT8), Some(Type::TEXT), Some(Type::BOOL)];
-        let result = convert_portal_params(&params, &types, &text_format()).unwrap();
-        assert!(matches!(result[0], nodedb_sql::ParamValue::Int64(42)));
-        assert!(matches!(&result[1], nodedb_sql::ParamValue::Text(s) if s == "hello"));
-        assert!(matches!(result[2], nodedb_sql::ParamValue::Bool(true)));
-    }
-
-    #[test]
-    fn convert_float_param() {
-        let params = vec![Some(Bytes::from_static(b"2.78"))];
-        let types = vec![Some(Type::FLOAT8)];
-        let result = convert_portal_params(&params, &types, &text_format()).unwrap();
-        assert!(
-            matches!(result[0], nodedb_sql::ParamValue::Float64(f) if (f - 2.78).abs() < f64::EPSILON)
-        );
-    }
-
-    #[test]
-    fn convert_numeric_text_to_decimal() {
-        let params = vec![Some(Bytes::from_static(b"123.45"))];
-        let types = vec![Some(Type::NUMERIC)];
-        let result = convert_portal_params(&params, &types, &text_format()).unwrap();
-        match &result[0] {
-            nodedb_sql::ParamValue::Decimal(decimal) => assert_eq!(decimal.to_string(), "123.45"),
-            other => panic!("expected Decimal, got {other:?}"),
-        }
-    }
-
-    fn assert_binary_type_rejected(ty: Type, bytes: &'static [u8], name: &str) {
-        let params = vec![Some(Bytes::from_static(bytes))];
-        let types = vec![Some(ty)];
-        let error = convert_portal_params(&params, &types, &binary_format()).unwrap_err();
-        let message = error.to_string();
-        assert!(message.contains(name) || message.contains("0A000"));
-    }
-
-    #[test]
-    fn convert_numeric_binary_returns_error() {
-        assert_binary_type_rejected(Type::NUMERIC, &[0x00, 0x03, 0x00, 0x02], "NUMERIC");
-    }
-
-    #[test]
-    fn convert_timestamp_binary_returns_error() {
-        assert_binary_type_rejected(Type::TIMESTAMP, &[0; 8], "TIMESTAMP");
-    }
-
-    #[test]
-    fn convert_timestamptz_binary_returns_error() {
-        assert_binary_type_rejected(Type::TIMESTAMPTZ, &[0; 8], "TIMESTAMPTZ");
-    }
-
-    fn assert_text_param(
-        input: &'static [u8],
-        ty: Type,
-        expected: fn(&nodedb_sql::ParamValue) -> bool,
-    ) {
-        let params = vec![Some(Bytes::from_static(input))];
-        let types = vec![Some(ty)];
-        let result = convert_portal_params(&params, &types, &text_format()).unwrap();
-        assert!(expected(&result[0]));
-    }
-
-    #[test]
-    fn convert_timestamp_text_to_typed() {
-        assert_text_param(b"2024-01-01 00:00:00", Type::TIMESTAMP, |value| {
-            matches!(value, nodedb_sql::ParamValue::Timestamp(_))
-        });
-    }
-
-    #[test]
-    fn convert_timestamptz_text_to_typed() {
-        assert_text_param(b"2024-01-01 00:00:00+00", Type::TIMESTAMPTZ, |value| {
-            matches!(value, nodedb_sql::ParamValue::Timestamptz(_))
-        });
-    }
-
-    #[test]
-    fn convert_bool_variants() {
-        for (input, expected) in [("t", true), ("f", false), ("1", true), ("0", false)] {
-            let params = vec![Some(Bytes::from(input))];
-            let types = vec![Some(Type::BOOL)];
-            let result = convert_portal_params(&params, &types, &text_format()).unwrap();
-            assert!(matches!(result[0], nodedb_sql::ParamValue::Bool(value) if value == expected));
-        }
-    }
-
-    #[test]
-    fn passthrough_date_text() {
-        let value = pgwire_text_to_param("2026-04-19", &Type::DATE);
-        assert!(matches!(&value, nodedb_sql::ParamValue::Text(text) if text == "2026-04-19"));
-    }
-
-    #[test]
-    fn timestamp_text_parses_to_typed() {
-        let value = pgwire_text_to_param("2026-04-19 12:00:00", &Type::TIMESTAMP);
-        assert!(matches!(value, nodedb_sql::ParamValue::Timestamp(_)));
-    }
-
-    #[test]
-    fn timestamptz_text_parses_to_typed() {
-        let value = pgwire_text_to_param("2026-04-19 12:00:00+00", &Type::TIMESTAMPTZ);
-        assert!(matches!(value, nodedb_sql::ParamValue::Timestamptz(_)));
-    }
-
-    #[test]
-    fn passthrough_uuid_text() {
-        let uuid = "550e8400-e29b-41d4-a716-446655440000";
-        let value = pgwire_text_to_param(uuid, &Type::UUID);
-        assert!(matches!(&value, nodedb_sql::ParamValue::Text(text) if text == uuid));
-    }
-
-    #[test]
-    fn passthrough_jsonb_text() {
-        let json = r#"{"a":1}"#;
-        let value = pgwire_text_to_param(json, &Type::JSONB);
-        assert!(matches!(&value, nodedb_sql::ParamValue::Text(text) if text == json));
-    }
-
-    #[test]
-    fn passthrough_bytea_hex_text() {
-        let value = pgwire_text_to_param("\\xDEADBEEF", &Type::BYTEA);
-        assert!(matches!(&value, nodedb_sql::ParamValue::Text(text) if text == "\\xDEADBEEF"));
-    }
-
-    #[test]
-    fn int_parse_failure_falls_back_to_text() {
-        let value = pgwire_text_to_param("abc", &Type::INT8);
-        assert!(matches!(&value, nodedb_sql::ParamValue::Text(text) if text == "abc"));
-    }
-
-    #[test]
-    fn unknown_type_routes_to_text() {
-        let value = pgwire_text_to_param("42", &Type::UNKNOWN);
-        assert!(matches!(&value, nodedb_sql::ParamValue::Text(text) if text == "42"));
-    }
-}

nodedb/src/control/server/pgwire/handler/prepared/execute.rs

@@ -2,6 +2,7 @@
 
 pub mod describe;
 pub mod execute;
+pub mod param_bind;
 pub mod parser;
 pub mod result_format;
 pub mod statement;