Add query_raw_txt client method

It takes all the extended protocol params as text and passes them to
postgres to sort out types. With that we can avoid situations when
postgres derived different type compared to what was passed in arguments.
There is also propare_typed method, but since we receive data in text
format anyway it makes more sense to avoid dealing with types in params.

This way we also can save on roundtrip and send Parse+Bind+Describe+Execute
right away without waiting for params description before Bind.

Also use text protocol for responses -- that allows to grab
postgres-provided serializations for types.

Author: kelvich

postgres-types/src/lib.rs

@@ -395,6 +395,22 @@ impl WrongType {
     }
 }
 
+/// An error indicating that a as_text conversion was attempted on a binary
+/// result.
+#[derive(Debug)]
+pub struct WrongFormat {}
+
+impl Error for WrongFormat {}
+
+impl fmt::Display for WrongFormat {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            fmt,
+            "cannot read column as text while it is in binary format"
+        )
+    }
+}
+
 /// A trait for types that can be created from a Postgres value.
 ///
 /// # Types
@@ -846,7 +862,7 @@ pub trait ToSql: fmt::Debug {
 /// Supported Postgres message format types
 ///
 /// Using Text format in a message assumes a Postgres `SERVER_ENCODING` of `UTF8`
-#[derive(Clone, Copy, Debug)]
+#[derive(Clone, Copy, Debug, PartialEq)]
 pub enum Format {
     /// Text format (UTF-8)
     Text,

tokio-postgres/src/client.rs

@@ -7,8 +7,10 @@ use crate::copy_both::CopyBothDuplex;
 use crate::copy_out::CopyOutStream;
 #[cfg(feature = "runtime")]
 use crate::keepalive::KeepaliveConfig;
+use crate::prepare::get_type;
 use crate::query::RowStream;
 use crate::simple_query::SimpleQueryStream;
+use crate::statement::Column;
 #[cfg(feature = "runtime")]
 use crate::tls::MakeTlsConnect;
 use crate::tls::TlsConnect;
@@ -20,7 +22,7 @@ use crate::{
     CopyInSink, Error, Row, SimpleQueryMessage, Statement, ToStatement, Transaction,
     TransactionBuilder,
 };
-use bytes::{Buf, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use fallible_iterator::FallibleIterator;
 use futures_channel::mpsc;
 use futures_util::{future, pin_mut, ready, StreamExt, TryStreamExt};
@@ -374,6 +376,87 @@ impl Client {
         query::query(&self.inner, statement, params).await
     }
 
+    /// Pass text directly to the Postgres backend to allow it to sort out typing itself and
+    /// to save a roundtrip
+    pub async fn query_raw_txt<'a, S, I>(&self, query: S, params: I) -> Result<RowStream, Error>
+    where
+        S: AsRef<str>,
+        I: IntoIterator<Item = S>,
+        I::IntoIter: ExactSizeIterator,
+    {
+        let params = params.into_iter();
+        let params_len = params.len();
+
+        let buf = self.inner.with_buf(|buf| {
+            // Parse, anonymous portal
+            frontend::parse("", query.as_ref(), std::iter::empty(), buf).map_err(Error::encode)?;
+            // Bind, pass params as text, retrieve as binary
+            match frontend::bind(
+                "",                 // empty string selects the unnamed portal
+                "",                 // empty string selects the unnamed prepared statement
+                std::iter::empty(), // all parameters use the default format (text)
+                params,
+                |param, buf| {
+                    buf.put_slice(param.as_ref().as_bytes());
+                    Ok(postgres_protocol::IsNull::No)
+                },
+                Some(0), // all text
+                buf,
+            ) {
+                Ok(()) => Ok(()),
+                Err(frontend::BindError::Conversion(e)) => Err(Error::to_sql(e, 0)),
+                Err(frontend::BindError::Serialization(e)) => Err(Error::encode(e)),
+            }?;
+
+            // Describe portal to typecast results
+            frontend::describe(b'P', "", buf).map_err(Error::encode)?;
+            // Execute
+            frontend::execute("", 0, buf).map_err(Error::encode)?;
+            // Sync
+            frontend::sync(buf);
+
+            Ok(buf.split().freeze())
+        })?;
+
+        let mut responses = self
+            .inner
+            .send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+
+        // now read the responses
+
+        match responses.next().await? {
+            Message::ParseComplete => {}
+            _ => return Err(Error::unexpected_message()),
+        }
+        match responses.next().await? {
+            Message::BindComplete => {}
+            _ => return Err(Error::unexpected_message()),
+        }
+        let row_description = match responses.next().await? {
+            Message::RowDescription(body) => Some(body),
+            Message::NoData => None,
+            _ => return Err(Error::unexpected_message()),
+        };
+
+        // construct statement object
+
+        let parameters = vec![Type::UNKNOWN; params_len];
+
+        let mut columns = vec![];
+        if let Some(row_description) = row_description {
+            let mut it = row_description.fields();
+            while let Some(field) = it.next().map_err(Error::parse)? {
+                let type_ = get_type(&self.inner, field.type_oid()).await?;
+                let column = Column::new(field.name().to_string(), type_);
+                columns.push(column);
+            }
+        }
+
+        let statement = Statement::new_text(&self.inner, "".to_owned(), parameters, columns);
+
+        Ok(RowStream::new(statement, responses))
+    }
+
     /// Executes a statement, returning the number of rows modified.
     ///
     /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list

tokio-postgres/src/prepare.rs

@@ -126,7 +126,7 @@ fn encode(client: &InnerClient, name: &str, query: &str, types: &[Type]) -> Resu
     })
 }
 
-async fn get_type(client: &Arc<InnerClient>, oid: Oid) -> Result<Type, Error> {
+pub async fn get_type(client: &Arc<InnerClient>, oid: Oid) -> Result<Type, Error> {
     if let Some(type_) = Type::from_oid(oid) {
         return Ok(type_);
     }

tokio-postgres/src/query.rs

@@ -207,6 +207,17 @@ pin_project! {
     }
 }
 
+impl RowStream {
+    /// Creates a new `RowStream`.
+    pub fn new(statement: Statement, responses: Responses) -> Self {
+        RowStream {
+            statement,
+            responses,
+            _p: PhantomPinned,
+        }
+    }
+}
+
 impl Stream for RowStream {
     type Item = Result<Row, Error>;
 

tokio-postgres/src/row.rs

@@ -7,6 +7,7 @@ use crate::types::{FromSql, Type, WrongType};
 use crate::{Error, Statement};
 use fallible_iterator::FallibleIterator;
 use postgres_protocol::message::backend::DataRowBody;
+use postgres_types::{Format, WrongFormat};
 use std::fmt;
 use std::ops::Range;
 use std::str;
@@ -187,6 +188,22 @@ impl Row {
         let range = self.ranges[idx].to_owned()?;
         Some(&self.body.buffer()[range])
     }
+
+    /// Interpret the column at the given index as text
+    ///
+    /// Useful when using query_raw_txt() which sets text transfer mode
+    pub fn as_text(&self, idx: usize) -> Result<Option<&str>, Error> {
+        if self.statement.output_format() == Format::Text {
+            match self.col_buffer(idx) {
+                Some(raw) => {
+                    FromSql::from_sql(&Type::TEXT, raw).map_err(|e| Error::from_sql(e, idx))
+                }
+                None => Ok(None),
+            }
+        } else {
+            Err(Error::from_sql(Box::new(WrongFormat {}), idx))
+        }
+    }
 }
 
 impl AsName for SimpleColumn {

tokio-postgres/src/statement.rs

@@ -3,6 +3,7 @@ use crate::codec::FrontendMessage;
 use crate::connection::RequestMessages;
 use crate::types::Type;
 use postgres_protocol::message::frontend;
+use postgres_types::Format;
 use std::{
     fmt,
     sync::{Arc, Weak},
@@ -13,6 +14,7 @@ struct StatementInner {
     name: String,
     params: Vec<Type>,
     columns: Vec<Column>,
+    output_format: Format,
 }
 
 impl Drop for StatementInner {
@@ -46,6 +48,22 @@ impl Statement {
             name,
             params,
             columns,
+            output_format: Format::Binary,
+        }))
+    }
+
+    pub(crate) fn new_text(
+        inner: &Arc<InnerClient>,
+        name: String,
+        params: Vec<Type>,
+        columns: Vec<Column>,
+    ) -> Statement {
+        Statement(Arc::new(StatementInner {
+            client: Arc::downgrade(inner),
+            name,
+            params,
+            columns,
+            output_format: Format::Text,
         }))
     }
 
@@ -62,6 +80,11 @@ impl Statement {
     pub fn columns(&self) -> &[Column] {
         &self.0.columns
     }
+
+    /// Returns output format for the statement.
+    pub fn output_format(&self) -> Format {
+        self.0.output_format
+    }
 }
 
 /// Information about a column of a query.

tokio-postgres/tests/test/main.rs

@@ -251,6 +251,34 @@ async fn custom_array() {
     }
 }
 
+#[tokio::test]
+async fn query_raw_txt() {
+    let client = connect("user=postgres").await;
+
+    let rows: Vec<tokio_postgres::Row> = client
+        .query_raw_txt("SELECT 55 * $1", ["42"])
+        .await
+        .unwrap()
+        .try_collect()
+        .await
+        .unwrap();
+
+    assert_eq!(rows.len(), 1);
+    let res: i32 = rows[0].as_text(0).unwrap().parse::<i32>().unwrap();
+    assert_eq!(res, 55 * 42);
+
+    let rows: Vec<tokio_postgres::Row> = client
+        .query_raw_txt("SELECT $1", ["42"])
+        .await
+        .unwrap()
+        .try_collect()
+        .await
+        .unwrap();
+
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].get::<_, &str>(0), "42");
+}
+
 #[tokio::test]
 async fn custom_composite() {
     let client = connect("user=postgres").await;