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.

Author: kelvich

tokio-postgres/src/client.rs

@@ -7,6 +7,7 @@ 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;
 #[cfg(feature = "runtime")]
@@ -20,11 +21,12 @@ use crate::{
     CopyInSink, Error, Row, SimpleQueryMessage, Statement, ToStatement, Transaction,
     TransactionBuilder,
 };
-use bytes::{Buf, BytesMut};
+use bytes::{Buf, BytesMut, BufMut};
 use fallible_iterator::FallibleIterator;
 use futures_channel::mpsc;
 use futures_util::{future, pin_mut, ready, StreamExt, TryStreamExt};
 use parking_lot::Mutex;
+use crate::statement::Column;
 use postgres_protocol::message::{backend::Message, frontend};
 use postgres_types::BorrowToSql;
 use std::collections::HashMap;
@@ -374,6 +376,90 @@ 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 parems_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(1), // all binary
+                buf,
+            ) {
+                Ok(()) => Ok(()),
+                Err(frontend::BindError::Conversion(e)) => Err(Error::to_sql(e, 0)), // TODO
+                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::TEXT; parems_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(
+            &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/tests/test/main.rs

@@ -251,6 +251,29 @@ 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);
+    assert_eq!(rows[0].get::<_, i32>(0), 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;