From f3976680c6d7004b04b3ba39f90f2956ce6d7010 Mon Sep 17 00:00:00 2001 From: Ramnivas Laddad Date: Sun, 26 May 2024 11:05:00 -0700 Subject: [PATCH 1/5] Work with pools that don't support prepared statements Introduce a new `query_with_param_types` method that allows to specify Postgres type parameters. This obviated the need to use prepared statementsjust to obtain parameter types for a query. It then combines parse, bind, and execute in a single packet. Related: #1017, #1067 --- tokio-postgres/src/client.rs | 82 +++++++++++++++ tokio-postgres/src/generic_client.rs | 46 +++++++++ tokio-postgres/src/prepare.rs | 2 +- tokio-postgres/src/query.rs | 146 ++++++++++++++++++++++++++- tokio-postgres/src/statement.rs | 13 +++ tokio-postgres/src/transaction.rs | 27 +++++ tokio-postgres/tests/test/main.rs | 106 +++++++++++++++++++ 7 files changed, 416 insertions(+), 6 deletions(-) diff --git a/tokio-postgres/src/client.rs b/tokio-postgres/src/client.rs index d48a23a60..431bfa792 100644 --- a/tokio-postgres/src/client.rs +++ b/tokio-postgres/src/client.rs @@ -364,6 +364,88 @@ impl Client { query::query(&self.inner, statement, params).await } + /// Like `query`, but requires the types of query parameters to be explicitly specified. + /// + /// Compared to `query`, this method allows performing queries without three round trips (for prepare, execute, and close). Thus, + /// this is suitable in environments where prepared statements aren't supported (such as Cloudflare Workers with Hyperdrive). + /// + /// # Examples + /// + /// ```no_run + /// # async fn async_main(client: &tokio_postgres::Client) -> Result<(), tokio_postgres::Error> { + /// use tokio_postgres::types::ToSql; + /// use tokio_postgres::types::Type; + /// use futures_util::{pin_mut, TryStreamExt}; + /// + /// let rows = client.query_with_param_types( + /// "SELECT foo FROM bar WHERE biz = $1 AND baz = $2", + /// &[(&"first param", Type::TEXT), (&2i32, Type::INT4)], + /// ).await?; + /// + /// for row in rows { + /// let foo: i32 = row.get("foo"); + /// println!("foo: {}", foo); + /// } + /// # Ok(()) + /// # } + /// ``` + pub async fn query_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result, Error> { + self.query_raw_with_param_types(statement, params) + .await? + .try_collect() + .await + } + + /// The maximally flexible version of [`query_with_param_types`]. + /// + /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list + /// provided, 1-indexed. + /// + /// The parameters must specify value along with their Postgres type. This allows performing + /// queries without three round trips (for prepare, execute, and close). + /// + /// [`query_with_param_types`]: #method.query_with_param_types + /// + /// # Examples + /// + /// ```no_run + /// # async fn async_main(client: &tokio_postgres::Client) -> Result<(), tokio_postgres::Error> { + /// use tokio_postgres::types::ToSql; + /// use tokio_postgres::types::Type; + /// use futures_util::{pin_mut, TryStreamExt}; + /// + /// let mut it = client.query_raw_with_param_types( + /// "SELECT foo FROM bar WHERE biz = $1 AND baz = $2", + /// &[(&"first param", Type::TEXT), (&2i32, Type::INT4)], + /// ).await?; + /// + /// pin_mut!(it); + /// while let Some(row) = it.try_next().await? { + /// let foo: i32 = row.get("foo"); + /// println!("foo: {}", foo); + /// } + /// # Ok(()) + /// # } + /// ``` + pub async fn query_raw_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result { + fn slice_iter<'a>( + s: &'a [(&'a (dyn ToSql + Sync), Type)], + ) -> impl ExactSizeIterator + 'a { + s.iter() + .map(|(param, param_type)| (*param as _, param_type.clone())) + } + + query::query_with_param_types(&self.inner, statement, slice_iter(params)).await + } + /// 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 diff --git a/tokio-postgres/src/generic_client.rs b/tokio-postgres/src/generic_client.rs index 50cff9712..3a0b09233 100644 --- a/tokio-postgres/src/generic_client.rs +++ b/tokio-postgres/src/generic_client.rs @@ -56,6 +56,20 @@ pub trait GenericClient: private::Sealed { I: IntoIterator + Sync + Send, I::IntoIter: ExactSizeIterator; + /// Like `Client::query_with_param_types` + async fn query_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result, Error>; + + /// Like `Client::query_raw_with_param_types`. + async fn query_raw_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result; + /// Like `Client::prepare`. async fn prepare(&self, query: &str) -> Result; @@ -136,6 +150,22 @@ impl GenericClient for Client { self.query_raw(statement, params).await } + async fn query_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result, Error> { + self.query_with_param_types(statement, params).await + } + + async fn query_raw_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result { + self.query_raw_with_param_types(statement, params).await + } + async fn prepare(&self, query: &str) -> Result { self.prepare(query).await } @@ -222,6 +252,22 @@ impl GenericClient for Transaction<'_> { self.query_raw(statement, params).await } + async fn query_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result, Error> { + self.query_with_param_types(statement, params).await + } + + async fn query_raw_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result { + self.query_raw_with_param_types(statement, params).await + } + async fn prepare(&self, query: &str) -> Result { self.prepare(query).await } diff --git a/tokio-postgres/src/prepare.rs b/tokio-postgres/src/prepare.rs index 07fb45694..1d9bacb16 100644 --- a/tokio-postgres/src/prepare.rs +++ b/tokio-postgres/src/prepare.rs @@ -131,7 +131,7 @@ fn encode(client: &InnerClient, name: &str, query: &str, types: &[Type]) -> Resu }) } -async fn get_type(client: &Arc, oid: Oid) -> Result { +pub(crate) async fn get_type(client: &Arc, oid: Oid) -> Result { if let Some(type_) = Type::from_oid(oid) { return Ok(type_); } diff --git a/tokio-postgres/src/query.rs b/tokio-postgres/src/query.rs index e6e1d00a8..b9cc66405 100644 --- a/tokio-postgres/src/query.rs +++ b/tokio-postgres/src/query.rs @@ -1,17 +1,21 @@ use crate::client::{InnerClient, Responses}; use crate::codec::FrontendMessage; use crate::connection::RequestMessages; +use crate::prepare::get_type; use crate::types::{BorrowToSql, IsNull}; -use crate::{Error, Portal, Row, Statement}; +use crate::{Column, Error, Portal, Row, Statement}; use bytes::{Bytes, BytesMut}; +use fallible_iterator::FallibleIterator; use futures_util::{ready, Stream}; use log::{debug, log_enabled, Level}; use pin_project_lite::pin_project; -use postgres_protocol::message::backend::{CommandCompleteBody, Message}; +use postgres_protocol::message::backend::{CommandCompleteBody, Message, RowDescriptionBody}; use postgres_protocol::message::frontend; +use postgres_types::Type; use std::fmt; use std::marker::PhantomPinned; use std::pin::Pin; +use std::sync::Arc; use std::task::{Context, Poll}; struct BorrowToSqlParamsDebug<'a, T>(&'a [T]); @@ -50,13 +54,125 @@ where }; let responses = start(client, buf).await?; Ok(RowStream { - statement, + statement: statement, responses, rows_affected: None, _p: PhantomPinned, }) } +enum QueryProcessingState { + Empty, + ParseCompleted, + BindCompleted, + ParameterDescribed, + Final(Vec), +} + +/// State machine for processing messages for `query_with_param_types`. +impl QueryProcessingState { + pub async fn process_message( + self, + client: &Arc, + message: Message, + ) -> Result { + match (self, message) { + (QueryProcessingState::Empty, Message::ParseComplete) => { + Ok(QueryProcessingState::ParseCompleted) + } + (QueryProcessingState::ParseCompleted, Message::BindComplete) => { + Ok(QueryProcessingState::BindCompleted) + } + (QueryProcessingState::BindCompleted, Message::ParameterDescription(_)) => { + Ok(QueryProcessingState::ParameterDescribed) + } + ( + QueryProcessingState::ParameterDescribed, + Message::RowDescription(row_description), + ) => Self::form_final(client, Some(row_description)).await, + (QueryProcessingState::ParameterDescribed, Message::NoData) => { + Self::form_final(client, None).await + } + (_, Message::ErrorResponse(body)) => Err(Error::db(body)), + _ => Err(Error::unexpected_message()), + } + } + + async fn form_final( + client: &Arc, + row_description: Option, + ) -> Result { + 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(client, field.type_oid()).await?; + let column = Column { + name: field.name().to_string(), + table_oid: Some(field.table_oid()).filter(|n| *n != 0), + column_id: Some(field.column_id()).filter(|n| *n != 0), + r#type: type_, + }; + columns.push(column); + } + } + + Ok(Self::Final(columns)) + } +} + +pub async fn query_with_param_types<'a, P, I>( + client: &Arc, + query: &str, + params: I, +) -> Result +where + P: BorrowToSql, + I: IntoIterator, + I::IntoIter: ExactSizeIterator, +{ + let (params, param_types): (Vec<_>, Vec<_>) = params.into_iter().unzip(); + + let params = params.into_iter(); + + let param_oids = param_types.iter().map(|t| t.oid()).collect::>(); + + let params = params.into_iter(); + + let buf = client.with_buf(|buf| { + frontend::parse("", query, param_oids.into_iter(), buf).map_err(Error::parse)?; + + encode_bind_with_statement_name_and_param_types("", ¶m_types, params, "", buf)?; + + frontend::describe(b'S', "", buf).map_err(Error::encode)?; + + frontend::execute("", 0, buf).map_err(Error::encode)?; + + frontend::sync(buf); + + Ok(buf.split().freeze()) + })?; + + let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?; + + let mut state = QueryProcessingState::Empty; + + loop { + let message = responses.next().await?; + + state = state.process_message(client, message).await?; + + if let QueryProcessingState::Final(columns) = state { + return Ok(RowStream { + statement: Statement::unnamed(vec![], columns), + responses, + rows_affected: None, + _p: PhantomPinned, + }); + } + } +} + pub async fn query_portal( client: &InnerClient, portal: &Portal, @@ -164,7 +280,27 @@ where I: IntoIterator, I::IntoIter: ExactSizeIterator, { - let param_types = statement.params(); + encode_bind_with_statement_name_and_param_types( + statement.name(), + statement.params(), + params, + portal, + buf, + ) +} + +fn encode_bind_with_statement_name_and_param_types( + statement_name: &str, + param_types: &[Type], + params: I, + portal: &str, + buf: &mut BytesMut, +) -> Result<(), Error> +where + P: BorrowToSql, + I: IntoIterator, + I::IntoIter: ExactSizeIterator, +{ let params = params.into_iter(); if param_types.len() != params.len() { @@ -181,7 +317,7 @@ where let mut error_idx = 0; let r = frontend::bind( portal, - statement.name(), + statement_name, param_formats, params.zip(param_types).enumerate(), |(idx, (param, ty)), buf| match param.borrow_to_sql().to_sql_checked(ty, buf) { diff --git a/tokio-postgres/src/statement.rs b/tokio-postgres/src/statement.rs index c5d657738..2b88ecd3b 100644 --- a/tokio-postgres/src/statement.rs +++ b/tokio-postgres/src/statement.rs @@ -14,6 +14,10 @@ struct StatementInner { impl Drop for StatementInner { fn drop(&mut self) { + if self.name.is_empty() { + // Unnamed statements don't need to be closed + return; + } if let Some(client) = self.client.upgrade() { let buf = client.with_buf(|buf| { frontend::close(b'S', &self.name, buf).unwrap(); @@ -46,6 +50,15 @@ impl Statement { })) } + pub(crate) fn unnamed(params: Vec, columns: Vec) -> Statement { + Statement(Arc::new(StatementInner { + client: Weak::new(), + name: String::new(), + params, + columns, + })) + } + pub(crate) fn name(&self) -> &str { &self.0.name } diff --git a/tokio-postgres/src/transaction.rs b/tokio-postgres/src/transaction.rs index 96a324652..5a6094b56 100644 --- a/tokio-postgres/src/transaction.rs +++ b/tokio-postgres/src/transaction.rs @@ -227,6 +227,33 @@ impl<'a> Transaction<'a> { query::query_portal(self.client.inner(), portal, max_rows).await } + /// Like `Client::query_with_param_types`. + pub async fn query_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result, Error> { + self.query_raw_with_param_types(statement, params) + .await? + .try_collect() + .await + } + + /// Like `Client::query_raw_with_param_types`. + pub async fn query_raw_with_param_types( + &self, + statement: &str, + params: &[(&(dyn ToSql + Sync), Type)], + ) -> Result { + fn slice_iter<'a>( + s: &'a [(&'a (dyn ToSql + Sync), Type)], + ) -> impl ExactSizeIterator + 'a { + s.iter() + .map(|(param, param_type)| (*param as _, param_type.clone())) + } + query::query_with_param_types(self.client.inner(), statement, slice_iter(params)).await + } + /// Like `Client::copy_in`. pub async fn copy_in(&self, statement: &T) -> Result, Error> where diff --git a/tokio-postgres/tests/test/main.rs b/tokio-postgres/tests/test/main.rs index 737f46631..925c99206 100644 --- a/tokio-postgres/tests/test/main.rs +++ b/tokio-postgres/tests/test/main.rs @@ -952,3 +952,109 @@ async fn deferred_constraint() { .await .unwrap_err(); } + +#[tokio::test] +async fn query_with_param_types_no_transaction() { + let client = connect("user=postgres").await; + + client + .batch_execute( + " + CREATE TEMPORARY TABLE foo ( + name TEXT, + age INT + ); + INSERT INTO foo (name, age) VALUES ('alice', 20), ('bob', 30), ('carol', 40); + ", + ) + .await + .unwrap(); + + let rows: Vec = client + .query_with_param_types( + "SELECT name, age, 'literal', 5 FROM foo WHERE name <> $1 AND age < $2 ORDER BY age", + &[(&"alice", Type::TEXT), (&50i32, Type::INT4)], + ) + .await + .unwrap(); + + assert_eq!(rows.len(), 2); + let first_row = &rows[0]; + assert_eq!(first_row.get::<_, &str>(0), "bob"); + assert_eq!(first_row.get::<_, i32>(1), 30); + assert_eq!(first_row.get::<_, &str>(2), "literal"); + assert_eq!(first_row.get::<_, i32>(3), 5); + + let second_row = &rows[1]; + assert_eq!(second_row.get::<_, &str>(0), "carol"); + assert_eq!(second_row.get::<_, i32>(1), 40); + assert_eq!(second_row.get::<_, &str>(2), "literal"); + assert_eq!(second_row.get::<_, i32>(3), 5); +} + +#[tokio::test] +async fn query_with_param_types_with_transaction() { + let mut client = connect("user=postgres").await; + + client + .batch_execute( + " + CREATE TEMPORARY TABLE foo ( + name TEXT, + age INT + ); + ", + ) + .await + .unwrap(); + + let transaction = client.transaction().await.unwrap(); + + let rows: Vec = transaction + .query_with_param_types( + "INSERT INTO foo (name, age) VALUES ($1, $2), ($3, $4), ($5, $6) returning name, age", + &[ + (&"alice", Type::TEXT), + (&20i32, Type::INT4), + (&"bob", Type::TEXT), + (&30i32, Type::INT4), + (&"carol", Type::TEXT), + (&40i32, Type::INT4), + ], + ) + .await + .unwrap(); + let inserted_values: Vec<(String, i32)> = rows + .iter() + .map(|row| (row.get::<_, String>(0), row.get::<_, i32>(1))) + .collect(); + assert_eq!( + inserted_values, + [ + ("alice".to_string(), 20), + ("bob".to_string(), 30), + ("carol".to_string(), 40) + ] + ); + + let rows: Vec = transaction + .query_with_param_types( + "SELECT name, age, 'literal', 5 FROM foo WHERE name <> $1 AND age < $2 ORDER BY age", + &[(&"alice", Type::TEXT), (&50i32, Type::INT4)], + ) + .await + .unwrap(); + + assert_eq!(rows.len(), 2); + let first_row = &rows[0]; + assert_eq!(first_row.get::<_, &str>(0), "bob"); + assert_eq!(first_row.get::<_, i32>(1), 30); + assert_eq!(first_row.get::<_, &str>(2), "literal"); + assert_eq!(first_row.get::<_, i32>(3), 5); + + let second_row = &rows[1]; + assert_eq!(second_row.get::<_, &str>(0), "carol"); + assert_eq!(second_row.get::<_, i32>(1), 40); + assert_eq!(second_row.get::<_, &str>(2), "literal"); + assert_eq!(second_row.get::<_, i32>(3), 5); +} From 3f8f5ded337a0122959f6e4a3dc9343bf6c6ee70 Mon Sep 17 00:00:00 2001 From: Ramnivas Laddad Date: Sun, 7 Jul 2024 16:21:40 -0700 Subject: [PATCH 2/5] Replace the state machine to process messages with a direct match statements --- tokio-postgres/src/query.rs | 101 ++++++++++-------------------------- 1 file changed, 27 insertions(+), 74 deletions(-) diff --git a/tokio-postgres/src/query.rs b/tokio-postgres/src/query.rs index b9cc66405..2bdfa14cc 100644 --- a/tokio-postgres/src/query.rs +++ b/tokio-postgres/src/query.rs @@ -9,7 +9,7 @@ use fallible_iterator::FallibleIterator; use futures_util::{ready, Stream}; use log::{debug, log_enabled, Level}; use pin_project_lite::pin_project; -use postgres_protocol::message::backend::{CommandCompleteBody, Message, RowDescriptionBody}; +use postgres_protocol::message::backend::{CommandCompleteBody, Message}; use postgres_protocol::message::frontend; use postgres_types::Type; use std::fmt; @@ -61,66 +61,6 @@ where }) } -enum QueryProcessingState { - Empty, - ParseCompleted, - BindCompleted, - ParameterDescribed, - Final(Vec), -} - -/// State machine for processing messages for `query_with_param_types`. -impl QueryProcessingState { - pub async fn process_message( - self, - client: &Arc, - message: Message, - ) -> Result { - match (self, message) { - (QueryProcessingState::Empty, Message::ParseComplete) => { - Ok(QueryProcessingState::ParseCompleted) - } - (QueryProcessingState::ParseCompleted, Message::BindComplete) => { - Ok(QueryProcessingState::BindCompleted) - } - (QueryProcessingState::BindCompleted, Message::ParameterDescription(_)) => { - Ok(QueryProcessingState::ParameterDescribed) - } - ( - QueryProcessingState::ParameterDescribed, - Message::RowDescription(row_description), - ) => Self::form_final(client, Some(row_description)).await, - (QueryProcessingState::ParameterDescribed, Message::NoData) => { - Self::form_final(client, None).await - } - (_, Message::ErrorResponse(body)) => Err(Error::db(body)), - _ => Err(Error::unexpected_message()), - } - } - - async fn form_final( - client: &Arc, - row_description: Option, - ) -> Result { - 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(client, field.type_oid()).await?; - let column = Column { - name: field.name().to_string(), - table_oid: Some(field.table_oid()).filter(|n| *n != 0), - column_id: Some(field.column_id()).filter(|n| *n != 0), - r#type: type_, - }; - columns.push(column); - } - } - - Ok(Self::Final(columns)) - } -} - pub async fn query_with_param_types<'a, P, I>( client: &Arc, query: &str, @@ -155,20 +95,33 @@ where let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?; - let mut state = QueryProcessingState::Empty; - loop { - let message = responses.next().await?; - - state = state.process_message(client, message).await?; - - if let QueryProcessingState::Final(columns) = state { - return Ok(RowStream { - statement: Statement::unnamed(vec![], columns), - responses, - rows_affected: None, - _p: PhantomPinned, - }); + match responses.next().await? { + Message::ParseComplete + | Message::BindComplete + | Message::ParameterDescription(_) + | Message::NoData => {} + Message::RowDescription(row_description) => { + let mut columns: Vec = vec![]; + let mut it = row_description.fields(); + while let Some(field) = it.next().map_err(Error::parse)? { + let type_ = get_type(client, field.type_oid()).await?; + let column = Column { + name: field.name().to_string(), + table_oid: Some(field.table_oid()).filter(|n| *n != 0), + column_id: Some(field.column_id()).filter(|n| *n != 0), + r#type: type_, + }; + columns.push(column); + } + return Ok(RowStream { + statement: Statement::unnamed(vec![], columns), + responses, + rows_affected: None, + _p: PhantomPinned, + }); + } + _ => return Err(Error::unexpected_message()), } } } From 74eb4dbf7399cb96500f2b60a2b838805471a26a Mon Sep 17 00:00:00 2001 From: Ramnivas Laddad Date: Sun, 7 Jul 2024 16:43:41 -0700 Subject: [PATCH 3/5] Remove query_raw_with_param_types as per PR feedback --- tokio-postgres/src/client.rs | 56 ++++++---------------------- tokio-postgres/src/generic_client.rs | 23 ------------ tokio-postgres/src/transaction.rs | 20 +--------- 3 files changed, 12 insertions(+), 87 deletions(-) diff --git a/tokio-postgres/src/client.rs b/tokio-postgres/src/client.rs index 431bfa792..e420bcf2f 100644 --- a/tokio-postgres/src/client.rs +++ b/tokio-postgres/src/client.rs @@ -366,8 +366,13 @@ impl Client { /// Like `query`, but requires the types of query parameters to be explicitly specified. /// - /// Compared to `query`, this method allows performing queries without three round trips (for prepare, execute, and close). Thus, - /// this is suitable in environments where prepared statements aren't supported (such as Cloudflare Workers with Hyperdrive). + /// Compared to `query`, this method allows performing queries without three round trips (for + /// prepare, execute, and close) by requiring the caller to specify parameter values along with + /// their Postgres type. Thus, this is suitable in environments where prepared statements aren't + /// supported (such as Cloudflare Workers with Hyperdrive). + /// + /// A statement may contain parameters, specified by `$n`, where `n` is the index of the + /// parameter of the list provided, 1-indexed. /// /// # Examples /// @@ -394,48 +399,6 @@ impl Client { statement: &str, params: &[(&(dyn ToSql + Sync), Type)], ) -> Result, Error> { - self.query_raw_with_param_types(statement, params) - .await? - .try_collect() - .await - } - - /// The maximally flexible version of [`query_with_param_types`]. - /// - /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list - /// provided, 1-indexed. - /// - /// The parameters must specify value along with their Postgres type. This allows performing - /// queries without three round trips (for prepare, execute, and close). - /// - /// [`query_with_param_types`]: #method.query_with_param_types - /// - /// # Examples - /// - /// ```no_run - /// # async fn async_main(client: &tokio_postgres::Client) -> Result<(), tokio_postgres::Error> { - /// use tokio_postgres::types::ToSql; - /// use tokio_postgres::types::Type; - /// use futures_util::{pin_mut, TryStreamExt}; - /// - /// let mut it = client.query_raw_with_param_types( - /// "SELECT foo FROM bar WHERE biz = $1 AND baz = $2", - /// &[(&"first param", Type::TEXT), (&2i32, Type::INT4)], - /// ).await?; - /// - /// pin_mut!(it); - /// while let Some(row) = it.try_next().await? { - /// let foo: i32 = row.get("foo"); - /// println!("foo: {}", foo); - /// } - /// # Ok(()) - /// # } - /// ``` - pub async fn query_raw_with_param_types( - &self, - statement: &str, - params: &[(&(dyn ToSql + Sync), Type)], - ) -> Result { fn slice_iter<'a>( s: &'a [(&'a (dyn ToSql + Sync), Type)], ) -> impl ExactSizeIterator + 'a { @@ -443,7 +406,10 @@ impl Client { .map(|(param, param_type)| (*param as _, param_type.clone())) } - query::query_with_param_types(&self.inner, statement, slice_iter(params)).await + query::query_with_param_types(&self.inner, statement, slice_iter(params)) + .await? + .try_collect() + .await } /// Executes a statement, returning the number of rows modified. diff --git a/tokio-postgres/src/generic_client.rs b/tokio-postgres/src/generic_client.rs index 3a0b09233..b892015dc 100644 --- a/tokio-postgres/src/generic_client.rs +++ b/tokio-postgres/src/generic_client.rs @@ -63,13 +63,6 @@ pub trait GenericClient: private::Sealed { params: &[(&(dyn ToSql + Sync), Type)], ) -> Result, Error>; - /// Like `Client::query_raw_with_param_types`. - async fn query_raw_with_param_types( - &self, - statement: &str, - params: &[(&(dyn ToSql + Sync), Type)], - ) -> Result; - /// Like `Client::prepare`. async fn prepare(&self, query: &str) -> Result; @@ -158,14 +151,6 @@ impl GenericClient for Client { self.query_with_param_types(statement, params).await } - async fn query_raw_with_param_types( - &self, - statement: &str, - params: &[(&(dyn ToSql + Sync), Type)], - ) -> Result { - self.query_raw_with_param_types(statement, params).await - } - async fn prepare(&self, query: &str) -> Result { self.prepare(query).await } @@ -260,14 +245,6 @@ impl GenericClient for Transaction<'_> { self.query_with_param_types(statement, params).await } - async fn query_raw_with_param_types( - &self, - statement: &str, - params: &[(&(dyn ToSql + Sync), Type)], - ) -> Result { - self.query_raw_with_param_types(statement, params).await - } - async fn prepare(&self, query: &str) -> Result { self.prepare(query).await } diff --git a/tokio-postgres/src/transaction.rs b/tokio-postgres/src/transaction.rs index 5a6094b56..8a0ad2224 100644 --- a/tokio-postgres/src/transaction.rs +++ b/tokio-postgres/src/transaction.rs @@ -233,25 +233,7 @@ impl<'a> Transaction<'a> { statement: &str, params: &[(&(dyn ToSql + Sync), Type)], ) -> Result, Error> { - self.query_raw_with_param_types(statement, params) - .await? - .try_collect() - .await - } - - /// Like `Client::query_raw_with_param_types`. - pub async fn query_raw_with_param_types( - &self, - statement: &str, - params: &[(&(dyn ToSql + Sync), Type)], - ) -> Result { - fn slice_iter<'a>( - s: &'a [(&'a (dyn ToSql + Sync), Type)], - ) -> impl ExactSizeIterator + 'a { - s.iter() - .map(|(param, param_type)| (*param as _, param_type.clone())) - } - query::query_with_param_types(self.client.inner(), statement, slice_iter(params)).await + self.client.query_with_param_types(statement, params).await } /// Like `Client::copy_in`. From dbd4d02e2f3a367b949e356e9dda40c08272d954 Mon Sep 17 00:00:00 2001 From: Ramnivas Laddad Date: Mon, 8 Jul 2024 17:21:32 -0700 Subject: [PATCH 4/5] Address review comment to rename query_with_param_types to query_typed --- tokio-postgres/src/client.rs | 6 +++--- tokio-postgres/src/generic_client.rs | 12 ++++++------ tokio-postgres/src/query.rs | 2 +- tokio-postgres/src/transaction.rs | 6 +++--- tokio-postgres/tests/test/main.rs | 10 +++++----- 5 files changed, 18 insertions(+), 18 deletions(-) diff --git a/tokio-postgres/src/client.rs b/tokio-postgres/src/client.rs index e420bcf2f..2b29351a5 100644 --- a/tokio-postgres/src/client.rs +++ b/tokio-postgres/src/client.rs @@ -382,7 +382,7 @@ impl Client { /// use tokio_postgres::types::Type; /// use futures_util::{pin_mut, TryStreamExt}; /// - /// let rows = client.query_with_param_types( + /// let rows = client.query_typed( /// "SELECT foo FROM bar WHERE biz = $1 AND baz = $2", /// &[(&"first param", Type::TEXT), (&2i32, Type::INT4)], /// ).await?; @@ -394,7 +394,7 @@ impl Client { /// # Ok(()) /// # } /// ``` - pub async fn query_with_param_types( + pub async fn query_typed( &self, statement: &str, params: &[(&(dyn ToSql + Sync), Type)], @@ -406,7 +406,7 @@ impl Client { .map(|(param, param_type)| (*param as _, param_type.clone())) } - query::query_with_param_types(&self.inner, statement, slice_iter(params)) + query::query_typed(&self.inner, statement, slice_iter(params)) .await? .try_collect() .await diff --git a/tokio-postgres/src/generic_client.rs b/tokio-postgres/src/generic_client.rs index e43bddfea..b91d78064 100644 --- a/tokio-postgres/src/generic_client.rs +++ b/tokio-postgres/src/generic_client.rs @@ -56,8 +56,8 @@ pub trait GenericClient: private::Sealed { I: IntoIterator + Sync + Send, I::IntoIter: ExactSizeIterator; - /// Like [`Client::query_with_param_types`] - async fn query_with_param_types( + /// Like [`Client::query_typed`] + async fn query_typed( &self, statement: &str, params: &[(&(dyn ToSql + Sync), Type)], @@ -146,12 +146,12 @@ impl GenericClient for Client { self.query_raw(statement, params).await } - async fn query_with_param_types( + async fn query_typed( &self, statement: &str, params: &[(&(dyn ToSql + Sync), Type)], ) -> Result, Error> { - self.query_with_param_types(statement, params).await + self.query_typed(statement, params).await } async fn prepare(&self, query: &str) -> Result { @@ -244,12 +244,12 @@ impl GenericClient for Transaction<'_> { self.query_raw(statement, params).await } - async fn query_with_param_types( + async fn query_typed( &self, statement: &str, params: &[(&(dyn ToSql + Sync), Type)], ) -> Result, Error> { - self.query_with_param_types(statement, params).await + self.query_typed(statement, params).await } async fn prepare(&self, query: &str) -> Result { diff --git a/tokio-postgres/src/query.rs b/tokio-postgres/src/query.rs index 2bdfa14cc..b54e095df 100644 --- a/tokio-postgres/src/query.rs +++ b/tokio-postgres/src/query.rs @@ -61,7 +61,7 @@ where }) } -pub async fn query_with_param_types<'a, P, I>( +pub async fn query_typed<'a, P, I>( client: &Arc, query: &str, params: I, diff --git a/tokio-postgres/src/transaction.rs b/tokio-postgres/src/transaction.rs index 8a0ad2224..3e62b2ac7 100644 --- a/tokio-postgres/src/transaction.rs +++ b/tokio-postgres/src/transaction.rs @@ -227,13 +227,13 @@ impl<'a> Transaction<'a> { query::query_portal(self.client.inner(), portal, max_rows).await } - /// Like `Client::query_with_param_types`. - pub async fn query_with_param_types( + /// Like `Client::query_typed`. + pub async fn query_typed( &self, statement: &str, params: &[(&(dyn ToSql + Sync), Type)], ) -> Result, Error> { - self.client.query_with_param_types(statement, params).await + self.client.query_typed(statement, params).await } /// Like `Client::copy_in`. diff --git a/tokio-postgres/tests/test/main.rs b/tokio-postgres/tests/test/main.rs index 925c99206..7ddb7a36a 100644 --- a/tokio-postgres/tests/test/main.rs +++ b/tokio-postgres/tests/test/main.rs @@ -954,7 +954,7 @@ async fn deferred_constraint() { } #[tokio::test] -async fn query_with_param_types_no_transaction() { +async fn query_typed_no_transaction() { let client = connect("user=postgres").await; client @@ -971,7 +971,7 @@ async fn query_with_param_types_no_transaction() { .unwrap(); let rows: Vec = client - .query_with_param_types( + .query_typed( "SELECT name, age, 'literal', 5 FROM foo WHERE name <> $1 AND age < $2 ORDER BY age", &[(&"alice", Type::TEXT), (&50i32, Type::INT4)], ) @@ -993,7 +993,7 @@ async fn query_with_param_types_no_transaction() { } #[tokio::test] -async fn query_with_param_types_with_transaction() { +async fn query_typed_with_transaction() { let mut client = connect("user=postgres").await; client @@ -1011,7 +1011,7 @@ async fn query_with_param_types_with_transaction() { let transaction = client.transaction().await.unwrap(); let rows: Vec = transaction - .query_with_param_types( + .query_typed( "INSERT INTO foo (name, age) VALUES ($1, $2), ($3, $4), ($5, $6) returning name, age", &[ (&"alice", Type::TEXT), @@ -1038,7 +1038,7 @@ async fn query_with_param_types_with_transaction() { ); let rows: Vec = transaction - .query_with_param_types( + .query_typed( "SELECT name, age, 'literal', 5 FROM foo WHERE name <> $1 AND age < $2 ORDER BY age", &[(&"alice", Type::TEXT), (&50i32, Type::INT4)], ) From 0fa32471ef2e20b7f2e554d6d97cde3a67f1d494 Mon Sep 17 00:00:00 2001 From: Ramnivas Laddad Date: Tue, 9 Jul 2024 17:59:39 -0700 Subject: [PATCH 5/5] Fix a clippy warning --- tokio-postgres/src/query.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tokio-postgres/src/query.rs b/tokio-postgres/src/query.rs index b54e095df..e304bbaea 100644 --- a/tokio-postgres/src/query.rs +++ b/tokio-postgres/src/query.rs @@ -54,7 +54,7 @@ where }; let responses = start(client, buf).await?; Ok(RowStream { - statement: statement, + statement, responses, rows_affected: None, _p: PhantomPinned,