Add Row Description Fields to SimpleQuery response

Adds the field information from the Row Description message to
SimpleQueryStream and propagates this information to SimpleQueryRow.
Previously, we took the column name from the fields and stored this as a
SimpleColumn.

This change was designed to limit the impact to the current SimpleQuery
interface. The field information is stored in a new OwnedField struct,
which is identical to the Field struct except that the name is stored as
a String. This allows the SimpleQueryMessage to outlive the
RowDescriptionBody.

We could instead store the RowDescriptionBody and attempt to parse it
when SimpleQueryRow::columns() is called. This is analogous to how the
row body is handled by SimpleQueryRow::get()/try_get(). However, this
approach was not taken to limit the changes to the current interface.

SimpleQueryRow::columns() now requires an allocation to return each
column as a SimpleColumn. This can be avoided by using the new
SimpleQueryRow::fields() method.

Author: Dan Wilbanks

postgres-protocol/src/message/backend.rs

@@ -557,6 +557,7 @@ impl CopyBothResponseBody {
     }
 }
 
+#[derive(PartialEq, Eq)]
 pub struct DataRowBody {
     storage: Bytes,
     len: u16,
@@ -901,6 +902,70 @@ impl<'a> Field<'a> {
     }
 }
 
+/// A struct representing the fields of a RowDescription message. Clones the fields to avoid keeping
+/// the `RowDescriptionMessage`.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct OwnedField {
+    name: String,
+    table_oid: Oid,
+    column_id: i16,
+    type_oid: Oid,
+    type_size: i16,
+    type_modifier: i32,
+    format: i16,
+}
+
+impl OwnedField {
+    #[inline]
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    #[inline]
+    pub fn table_oid(&self) -> Oid {
+        self.table_oid
+    }
+
+    #[inline]
+    pub fn column_id(&self) -> i16 {
+        self.column_id
+    }
+
+    #[inline]
+    pub fn type_oid(&self) -> Oid {
+        self.type_oid
+    }
+
+    #[inline]
+    pub fn type_size(&self) -> i16 {
+        self.type_size
+    }
+
+    #[inline]
+    pub fn type_modifier(&self) -> i32 {
+        self.type_modifier
+    }
+
+    #[inline]
+    pub fn format(&self) -> i16 {
+        self.format
+    }
+}
+
+impl From<Field<'_>> for OwnedField {
+    fn from(f: Field<'_>) -> Self {
+        OwnedField {
+            name: f.name().to_string(),
+            table_oid: f.table_oid(),
+            column_id: f.column_id(),
+            type_oid: f.type_oid(),
+            type_size: f.type_size(),
+            type_modifier: f.type_modifier(),
+            format: f.format(),
+        }
+    }
+}
+
 #[inline]
 fn find_null(buf: &[u8], start: usize) -> io::Result<usize> {
     match memchr(0, &buf[start..]) {

tokio-postgres/src/lib.rs

@@ -141,6 +141,7 @@ pub use crate::to_statement::ToStatement;
 pub use crate::transaction::Transaction;
 pub use crate::transaction_builder::{IsolationLevel, TransactionBuilder};
 use crate::types::ToSql;
+pub use postgres_protocol::message::backend::OwnedField;
 pub use postgres_protocol::message::backend::Message;
 
 pub mod binary_copy;
@@ -240,6 +241,7 @@ pub enum AsyncMessage {
 
 /// Message returned by the `SimpleQuery` stream.
 #[non_exhaustive]
+#[derive(Debug)]
 pub enum SimpleQueryMessage {
     /// A row of data.
     Row(SimpleQueryRow),

tokio-postgres/src/row.rs

@@ -6,7 +6,7 @@ use crate::statement::Column;
 use crate::types::{FromSql, Type, WrongType};
 use crate::{Error, Statement};
 use fallible_iterator::FallibleIterator;
-use postgres_protocol::message::backend::DataRowBody;
+use postgres_protocol::message::backend::{OwnedField, DataRowBody};
 use std::fmt;
 use std::ops::Range;
 use std::str;
@@ -32,6 +32,12 @@ impl AsName for String {
     }
 }
 
+impl AsName for OwnedField {
+    fn as_name(&self) -> &str {
+        self.name()
+    }
+}
+
 /// A trait implemented by types that can index into columns of a row.
 ///
 /// This cannot be implemented outside of this crate.
@@ -196,29 +202,39 @@ impl AsName for SimpleColumn {
 }
 
 /// A row of data returned from the database by a simple query.
+#[derive(Eq, PartialEq)]
 pub struct SimpleQueryRow {
-    columns: Arc<[SimpleColumn]>,
+    fields: Arc<[OwnedField]>,
     body: DataRowBody,
     ranges: Vec<Option<Range<usize>>>,
 }
 
 impl SimpleQueryRow {
     #[allow(clippy::new_ret_no_self)]
     pub(crate) fn new(
-        columns: Arc<[SimpleColumn]>,
+        fields: Arc<[OwnedField]>,
         body: DataRowBody,
     ) -> Result<SimpleQueryRow, Error> {
         let ranges = body.ranges().collect().map_err(Error::parse)?;
         Ok(SimpleQueryRow {
-            columns,
+            fields,
             body,
             ranges,
         })
     }
 
-    /// Returns information about the columns of data in the row.
-    pub fn columns(&self) -> &[SimpleColumn] {
-        &self.columns
+    /// Returns information about the columns of data in the row. Performs an allocation to return
+    /// each name as a `SimpleColumn`. To avoid the allocation, use `SimpleQueryRow::fields()`.
+    pub fn columns(&self) -> Vec<SimpleColumn> {
+        self.fields
+            .iter()
+            .map(|f| SimpleColumn::new(f.name().to_string()))
+            .collect()
+    }
+
+    /// Returns the fields found in the RowDescription message with information on each column.
+    pub fn fields(&self) -> &[OwnedField] {
+        &self.fields
     }
 
     /// Determines if the row contains no values.
@@ -228,7 +244,7 @@ impl SimpleQueryRow {
 
     /// Returns the number of values in the row.
     pub fn len(&self) -> usize {
-        self.columns.len()
+        self.fields.len()
     }
 
     /// Returns a value from the row.
@@ -260,7 +276,7 @@ impl SimpleQueryRow {
     where
         I: RowIndex + fmt::Display,
     {
-        let idx = match idx.__idx(&self.columns) {
+        let idx = match idx.__idx(&self.fields) {
             Some(idx) => idx,
             None => return Err(Error::column(idx.to_string())),
         };
@@ -269,3 +285,13 @@ impl SimpleQueryRow {
         FromSql::from_sql_nullable(&Type::TEXT, buf).map_err(|e| Error::from_sql(e, idx))
     }
 }
+
+impl std::fmt::Debug for SimpleQueryRow {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> std::fmt::Result {
+        let mut values = Vec::with_capacity(self.len());
+        for i in 0..self.len() {
+            values.push(self.get(i).unwrap_or(""))
+        }
+        write!(f, "{:?}", values)
+    }
+}

tokio-postgres/src/simple_query.rs

@@ -7,7 +7,7 @@ use fallible_iterator::FallibleIterator;
 use futures::{ready, Stream};
 use log::debug;
 use pin_project_lite::pin_project;
-use postgres_protocol::message::backend::Message;
+use postgres_protocol::message::backend::{OwnedField, Message};
 use postgres_protocol::message::frontend;
 use std::marker::PhantomPinned;
 use std::pin::Pin;
@@ -38,7 +38,7 @@ pub async fn simple_query(client: &InnerClient, query: &str) -> Result<SimpleQue
 
     Ok(SimpleQueryStream {
         responses,
-        columns: None,
+        fields: None,
         _p: PhantomPinned,
     })
 }
@@ -72,7 +72,7 @@ pin_project! {
     /// A stream of simple query results.
     pub struct SimpleQueryStream {
         responses: Responses,
-        columns: Option<Arc<[SimpleColumn]>>,
+        fields: Option<Arc<[OwnedField]>>,
         #[pin]
         _p: PhantomPinned,
     }
@@ -100,18 +100,18 @@ impl Stream for SimpleQueryStream {
                     return Poll::Ready(Some(Ok(SimpleQueryMessage::CommandComplete(0))));
                 }
                 Message::RowDescription(body) => {
-                    let columns = body
+                    let fields = body
                         .fields()
-                        .map(|f| Ok(SimpleColumn::new(f.name().to_string())))
+                        .map(|f| Ok(f.into()))
                         .collect::<Vec<_>>()
                         .map_err(Error::parse)?
                         .into();
 
-                    *this.columns = Some(columns);
+                    *this.fields = Some(fields);
                 }
                 Message::DataRow(body) => {
-                    let row = match &this.columns {
-                        Some(columns) => SimpleQueryRow::new(columns.clone(), body)?,
+                    let row = match &this.fields {
+                        Some(fields) => SimpleQueryRow::new(fields.clone(), body)?,
                         None => return Poll::Ready(Some(Err(Error::unexpected_message()))),
                     };
                     return Poll::Ready(Some(Ok(SimpleQueryMessage::Row(row))));