Consolidate fetch_* methods in a trait

sqlx-core/src/query_as.rs

@@ -2,15 +2,15 @@ use std::marker::PhantomData;
 
 use either::Either;
 use futures_core::stream::BoxStream;
-use futures_util::{StreamExt, TryStreamExt};
+use futures_util::{FutureExt, StreamExt};
 
 use crate::arguments::IntoArguments;
 use crate::database::{Database, HasStatementCache};
 use crate::encode::Encode;
 use crate::error::{BoxDynError, Error};
 use crate::executor::{Execute, Executor};
 use crate::from_row::FromRow;
-use crate::query::{query, query_statement, query_statement_with, query_with_result, Query};
+use crate::query::{query, query_statement, query_statement_with, query_with_result, Fetch, Query};
 use crate::types::Type;
 
 /// A single SQL query as a prepared statement, mapping results using [`FromRow`].
@@ -77,44 +77,23 @@ where
     }
 }
 
-// FIXME: This is very close, nearly 1:1 with `Map`
-// noinspection DuplicatedCode
-impl<'q, DB, O, A> QueryAs<'q, DB, O, A>
+impl<'q, DB, O, A> Fetch<'q, DB> for QueryAs<'q, DB, O, A>
 where
     DB: Database,
     A: 'q + IntoArguments<'q, DB>,
     O: Send + Unpin + for<'r> FromRow<'r, DB::Row>,
 {
-    /// Execute the query and return the generated results as a stream.
-    pub fn fetch<'e, 'c: 'e, E>(self, executor: E) -> BoxStream<'e, Result<O, Error>>
-    where
-        'q: 'e,
-        E: 'e + Executor<'c, Database = DB>,
-        DB: 'e,
-        O: 'e,
-        A: 'e,
-    {
-        // FIXME: this should have used `executor.fetch()` but that's a breaking change
-        // because this technically allows multiple statements in one query string.
-        #[allow(deprecated)]
-        self.fetch_many(executor)
-            .try_filter_map(|step| async move { Ok(step.right()) })
-            .boxed()
-    }
+    type Output = O;
 
-    /// Execute multiple queries and return the generated results as a stream
-    /// from each query, in a stream.
-    #[deprecated = "Only the SQLite driver supports multiple statements in one prepared statement and that behavior is deprecated. Use `sqlx::raw_sql()` instead. See https://github.com/launchbadge/sqlx/issues/3108 for discussion."]
-    pub fn fetch_many<'e, 'c: 'e, E>(
+    fn fetch_many<'e, 'c: 'e, E>(
         self,
         executor: E,
-    ) -> BoxStream<'e, Result<Either<DB::QueryResult, O>, Error>>
+    ) -> BoxStream<'e, Result<Either<DB::QueryResult, Self::Output>, Error>>
     where
         'q: 'e,
         E: 'e + Executor<'c, Database = DB>,
         DB: 'e,
-        O: 'e,
-        A: 'e,
+        Self::Output: 'e,
     {
         executor
             .fetch_many(self.inner)
@@ -126,76 +105,25 @@ where
             .boxed()
     }
 
-    /// Execute the query and return all the resulting rows collected into a [`Vec`].
-    ///
-    /// ### Note: beware result set size.
-    /// This will attempt to collect the full result set of the query into memory.
-    ///
-    /// To avoid exhausting available memory, ensure the result set has a known upper bound,
-    /// e.g. using `LIMIT`.
-    #[inline]
-    pub async fn fetch_all<'e, 'c: 'e, E>(self, executor: E) -> Result<Vec<O>, Error>
-    where
-        'q: 'e,
-        E: 'e + Executor<'c, Database = DB>,
-        DB: 'e,
-        O: 'e,
-        A: 'e,
-    {
-        self.fetch(executor).try_collect().await
-    }
-
-    /// Execute the query, returning the first row or [`Error::RowNotFound`] otherwise.
-    ///
-    /// ### Note: for best performance, ensure the query returns at most one row.
-    /// Depending on the driver implementation, if your query can return more than one row,
-    /// it may lead to wasted CPU time and bandwidth on the database server.
-    ///
-    /// Even when the driver implementation takes this into account, ensuring the query returns at most one row
-    /// can result in a more optimal query plan.
-    ///
-    /// If your query has a `WHERE` clause filtering a unique column by a single value, you're good.
-    ///
-    /// Otherwise, you might want to add `LIMIT 1` to your query.
-    pub async fn fetch_one<'e, 'c: 'e, E>(self, executor: E) -> Result<O, Error>
-    where
-        'q: 'e,
-        E: 'e + Executor<'c, Database = DB>,
-        DB: 'e,
-        O: 'e,
-        A: 'e,
-    {
-        self.fetch_optional(executor)
-            .await
-            .and_then(|row| row.ok_or(Error::RowNotFound))
-    }
-
-    /// Execute the query, returning the first row or `None` otherwise.
-    ///
-    /// ### Note: for best performance, ensure the query returns at most one row.
-    /// Depending on the driver implementation, if your query can return more than one row,
-    /// it may lead to wasted CPU time and bandwidth on the database server.
-    ///
-    /// Even when the driver implementation takes this into account, ensuring the query returns at most one row
-    /// can result in a more optimal query plan.
-    ///
-    /// If your query has a `WHERE` clause filtering a unique column by a single value, you're good.
-    ///
-    /// Otherwise, you might want to add `LIMIT 1` to your query.
-    pub async fn fetch_optional<'e, 'c: 'e, E>(self, executor: E) -> Result<Option<O>, Error>
+    fn fetch_optional<'e, 'c: 'e, E>(
+        self,
+        executor: E,
+    ) -> futures_core::future::BoxFuture<'e, Result<Option<Self::Output>, Error>>
     where
         'q: 'e,
         E: 'e + Executor<'c, Database = DB>,
         DB: 'e,
-        O: 'e,
-        A: 'e,
+        Self::Output: 'e + Send + Unpin,
     {
-        let row = executor.fetch_optional(self.inner).await?;
-        if let Some(row) = row {
-            O::from_row(&row).map(Some)
-        } else {
-            Ok(None)
+        async {
+            let row = executor.fetch_optional(self.inner).await?;
+            if let Some(row) = row {
+                O::from_row(&row).map(Some)
+            } else {
+                Ok(None)
+            }
         }
+        .boxed()
     }
 }
 

sqlx-core/src/query_scalar.rs

@@ -1,13 +1,14 @@
 use either::Either;
 use futures_core::stream::BoxStream;
-use futures_util::{StreamExt, TryFutureExt, TryStreamExt};
+use futures_util::{FutureExt, StreamExt, TryStreamExt};
 
 use crate::arguments::IntoArguments;
 use crate::database::{Database, HasStatementCache};
 use crate::encode::Encode;
 use crate::error::{BoxDynError, Error};
 use crate::executor::{Execute, Executor};
 use crate::from_row::FromRow;
+use crate::query::Fetch;
 use crate::query_as::{
     query_as, query_as_with_result, query_statement_as, query_statement_as_with, QueryAs,
 };
@@ -74,42 +75,24 @@ where
     }
 }
 
-// FIXME: This is very close, nearly 1:1 with `Map`
-// noinspection DuplicatedCode
-impl<'q, DB, O, A> QueryScalar<'q, DB, O, A>
+impl<'q, DB, O, A> Fetch<'q, DB> for QueryScalar<'q, DB, O, A>
 where
     DB: Database,
     O: Send + Unpin,
     A: 'q + IntoArguments<'q, DB>,
     (O,): Send + Unpin + for<'r> FromRow<'r, DB::Row>,
 {
-    /// Execute the query and return the generated results as a stream.
-    #[inline]
-    pub fn fetch<'e, 'c: 'e, E>(self, executor: E) -> BoxStream<'e, Result<O, Error>>
-    where
-        'q: 'e,
-        E: 'e + Executor<'c, Database = DB>,
-        DB: 'e,
-        A: 'e,
-        O: 'e,
-    {
-        self.inner.fetch(executor).map_ok(|it| it.0).boxed()
-    }
+    type Output = O;
 
-    /// Execute multiple queries and return the generated results as a stream
-    /// from each query, in a stream.
-    #[inline]
-    #[deprecated = "Only the SQLite driver supports multiple statements in one prepared statement and that behavior is deprecated. Use `sqlx::raw_sql()` instead. See https://github.com/launchbadge/sqlx/issues/3108 for discussion."]
-    pub fn fetch_many<'e, 'c: 'e, E>(
+    fn fetch_many<'e, 'c: 'e, E>(
         self,
         executor: E,
-    ) -> BoxStream<'e, Result<Either<DB::QueryResult, O>, Error>>
+    ) -> BoxStream<'e, Result<Either<DB::QueryResult, Self::Output>, Error>>
     where
         'q: 'e,
         E: 'e + Executor<'c, Database = DB>,
         DB: 'e,
-        A: 'e,
-        O: 'e,
+        Self::Output: 'e,
     {
         #[allow(deprecated)]
         self.inner
@@ -118,75 +101,17 @@ where
             .boxed()
     }
 
-    /// Execute the query and return all the resulting rows collected into a [`Vec`].
-    ///
-    /// ### Note: beware result set size.
-    /// This will attempt to collect the full result set of the query into memory.
-    ///
-    /// To avoid exhausting available memory, ensure the result set has a known upper bound,
-    /// e.g. using `LIMIT`.
-    #[inline]
-    pub async fn fetch_all<'e, 'c: 'e, E>(self, executor: E) -> Result<Vec<O>, Error>
-    where
-        'q: 'e,
-        E: 'e + Executor<'c, Database = DB>,
-        DB: 'e,
-        (O,): 'e,
-        A: 'e,
-    {
-        self.inner
-            .fetch(executor)
-            .map_ok(|it| it.0)
-            .try_collect()
-            .await
-    }
-
-    /// Execute the query, returning the first row or [`Error::RowNotFound`] otherwise.
-    ///
-    /// ### Note: for best performance, ensure the query returns at most one row.
-    /// Depending on the driver implementation, if your query can return more than one row,
-    /// it may lead to wasted CPU time and bandwidth on the database server.
-    ///
-    /// Even when the driver implementation takes this into account, ensuring the query returns at most one row
-    /// can result in a more optimal query plan.
-    ///
-    /// If your query has a `WHERE` clause filtering a unique column by a single value, you're good.
-    ///
-    /// Otherwise, you might want to add `LIMIT 1` to your query.
-    #[inline]
-    pub async fn fetch_one<'e, 'c: 'e, E>(self, executor: E) -> Result<O, Error>
-    where
-        'q: 'e,
-        E: 'e + Executor<'c, Database = DB>,
-        DB: 'e,
-        O: 'e,
-        A: 'e,
-    {
-        self.inner.fetch_one(executor).map_ok(|it| it.0).await
-    }
-
-    /// Execute the query, returning the first row or `None` otherwise.
-    ///
-    /// ### Note: for best performance, ensure the query returns at most one row.
-    /// Depending on the driver implementation, if your query can return more than one row,
-    /// it may lead to wasted CPU time and bandwidth on the database server.
-    ///
-    /// Even when the driver implementation takes this into account, ensuring the query returns at most one row
-    /// can result in a more optimal query plan.
-    ///
-    /// If your query has a `WHERE` clause filtering a unique column by a single value, you're good.
-    ///
-    /// Otherwise, you might want to add `LIMIT 1` to your query.
-    #[inline]
-    pub async fn fetch_optional<'e, 'c: 'e, E>(self, executor: E) -> Result<Option<O>, Error>
+    fn fetch_optional<'e, 'c: 'e, E>(
+        self,
+        executor: E,
+    ) -> futures_core::future::BoxFuture<'e, Result<Option<Self::Output>, Error>>
     where
         'q: 'e,
         E: 'e + Executor<'c, Database = DB>,
         DB: 'e,
-        O: 'e,
-        A: 'e,
+        Self::Output: 'e + Send + Unpin,
     {
-        Ok(self.inner.fetch_optional(executor).await?.map(|it| it.0))
+        async { Ok(self.inner.fetch_optional(executor).await?.map(|it| it.0)) }.boxed()
     }
 }
 

sqlx-mysql/src/migrate.rs

@@ -8,7 +8,7 @@ pub(crate) use sqlx_core::migrate::*;
 use crate::connection::{ConnectOptions, Connection};
 use crate::error::Error;
 use crate::executor::Executor;
-use crate::query::query;
+use crate::query::{query, Fetch};
 use crate::query_as::query_as;
 use crate::query_scalar::query_scalar;
 use crate::{MySql, MySqlConnectOptions, MySqlConnection};

sqlx-mysql/src/testing/mod.rs

@@ -13,7 +13,7 @@ use crate::connection::Connection;
 use crate::error::Error;
 use crate::executor::Executor;
 use crate::pool::{Pool, PoolOptions};
-use crate::query::query;
+use crate::query::{query, Fetch};
 use crate::query_builder::QueryBuilder;
 use crate::query_scalar::query_scalar;
 use crate::{MySql, MySqlConnectOptions, MySqlConnection};

sqlx-postgres/src/advisory_lock.rs

@@ -4,6 +4,7 @@ use crate::PgConnection;
 use hkdf::Hkdf;
 use once_cell::sync::OnceCell;
 use sha2::Sha256;
+use sqlx_core::query::Fetch;
 use std::ops::{Deref, DerefMut};
 
 /// A mutex-like type utilizing [Postgres advisory locks].

sqlx-postgres/src/connection/describe.rs

@@ -2,6 +2,7 @@ use crate::error::Error;
 use crate::ext::ustr::UStr;
 use crate::io::StatementId;
 use crate::message::{ParameterDescription, RowDescription};
+use crate::query::Fetch;
 use crate::query_as::query_as;
 use crate::query_scalar::query_scalar;
 use crate::statement::PgStatementMetadata;

sqlx-postgres/src/migrate.rs

@@ -11,7 +11,7 @@ pub(crate) use sqlx_core::migrate::{Migrate, MigrateDatabase};
 use crate::connection::{ConnectOptions, Connection};
 use crate::error::Error;
 use crate::executor::Executor;
-use crate::query::query;
+use crate::query::{query, Fetch};
 use crate::query_as::query_as;
 use crate::query_scalar::query_scalar;
 use crate::{PgConnectOptions, PgConnection, Postgres};

sqlx-postgres/src/testing/mod.rs

@@ -13,7 +13,7 @@ use crate::connection::Connection;
 use crate::error::Error;
 use crate::executor::Executor;
 use crate::pool::{Pool, PoolOptions};
-use crate::query::query;
+use crate::query::{query, Fetch};
 use crate::query_scalar::query_scalar;
 use crate::{PgConnectOptions, PgConnection, Postgres};
 

sqlx-sqlite/src/migrate.rs

@@ -5,7 +5,7 @@ use crate::fs;
 use crate::migrate::MigrateError;
 use crate::migrate::{AppliedMigration, Migration};
 use crate::migrate::{Migrate, MigrateDatabase};
-use crate::query::query;
+use crate::query::{query, Fetch};
 use crate::query_as::query_as;
 use crate::{Sqlite, SqliteConnectOptions, SqliteConnection, SqliteJournalMode};
 use futures_core::future::BoxFuture;

src/lib.rs

@@ -13,7 +13,7 @@ pub use sqlx_core::from_row::FromRow;
 pub use sqlx_core::pool::{self, Pool};
 #[doc(hidden)]
 pub use sqlx_core::query::query_with_result as __query_with_result;
-pub use sqlx_core::query::{query, query_with};
+pub use sqlx_core::query::{query, query_with, Fetch};
 pub use sqlx_core::query_as::{query_as, query_as_with};
 pub use sqlx_core::query_builder::{self, QueryBuilder};
 #[doc(hidden)]

tests/any/any.rs

@@ -1,5 +1,5 @@
 use sqlx::any::AnyRow;
-use sqlx::{Any, Connection, Executor, Row};
+use sqlx::{Any, Connection, Executor, Fetch, Row};
 use sqlx_test::new;
 
 #[sqlx_macros::test]

tests/any/pool.rs

@@ -1,5 +1,5 @@
 use sqlx::any::{AnyConnectOptions, AnyPoolOptions};
-use sqlx::Executor;
+use sqlx::{Executor, Fetch};
 use std::sync::{
     atomic::{AtomicI32, AtomicUsize, Ordering},
     Arc, Mutex,

tests/postgres/derives.rs

@@ -1,6 +1,6 @@
 use futures::TryStreamExt;
 use sqlx::postgres::types::PgRange;
-use sqlx::{Connection, Executor, FromRow, Postgres};
+use sqlx::{Connection, Executor, Fetch, FromRow, Postgres};
 use sqlx_postgres::PgHasArrayType;
 use sqlx_test::{new, test_type};
 use std::fmt::Debug;