Repository: blackbeam/mysql_async (language: rust)

Issue:
Is it possible to implement a real Stream?
To make a non-panicking collect, with the actual code I have to do something like
```rust
async fn query<T: FromRow, E: From<FromRowError> + From<Error>>(conn: &mut Conn, query: &str) -> Result<Vec<T>, E> {
  conn.query_iter(query).await?
    .try_collect::<T>().await?
    .into_iter().collect::<Result<Vec<T>, E>>()
}
```
(this is only for example purposes, could contains errors)

This means allocating the entire resultset even if every single row is an error, or having to implement a wrapper to use next() as a Stream...
Wouldn't be possible, for example, to add an impl of futures_core::Stream that is the same used by tokio-stream and it seems to be a standard de-facto?
If you prefer, under a feature, so it wouldn't be default for everyone.

I can do the PR if you want


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index f5e15a86..dc7a5043 100644
--- a/README.md
+++ b/README.md
@@ -11,7 +11,7 @@ Tokio based asynchronous MySql client library for rust programming language.
 
 ### Installation
 
-Library hosted on [crates.io](https://crates.io/crates/mysql_async/).
+The library is hosted on [crates.io](https://crates.io/crates/mysql_async/).
 ```toml
 [dependencies]
 mysql_async = "<desired version>"
@@ -19,7 +19,7 @@ mysql_async = "<desired version>"
 
 ### Example
 
-See crate documentation on [docs.rs](https://docs.rs/mysql_async).
+Please see the crate docs – [docs.rs](https://docs.rs/mysql_async).
 
 ### License
 
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
index 9488fd5a..87b99a88 100644
--- a/src/conn/mod.rs
+++ b/src/conn/mod.rs
@@ -25,9 +25,10 @@ use std::{
     borrow::Cow,
     fmt,
     future::Future,
-    mem,
+    mem::{self, replace},
     pin::Pin,
     str::FromStr,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -78,6 +79,15 @@ fn disconnect(mut conn: Conn) {
     }
 }
 
+/// Pending result set.
+#[derive(Debug, Clone)]
+pub(crate) enum PendingResult {
+    /// There is a pending result set.
+    Pending(ResultSetMeta),
+    /// Result set metadata was taken but not yet consumed.
+    Taken(Arc<ResultSetMeta>),
+}
+
 /// Mysql connection
 struct ConnInner {
     stream: Option<Stream>,
@@ -90,7 +100,7 @@ struct ConnInner {
     last_ok_packet: Option<OkPacket<'static>>,
     last_err_packet: Option<mysql_common::packets::ServerError<'static>>,
     pool: Option<Pool>,
-    pending_result: Option<ResultSetMeta>,
+    pending_result: std::result::Result<Option<PendingResult>, ServerError>,
     tx_status: TxStatus,
     opts: Opts,
     last_io: Instant,
@@ -129,7 +139,7 @@ impl ConnInner {
             is_mariadb: false,
             version: (0, 0, 0),
             id: 0,
-            pending_result: None,
+            pending_result: Ok(None),
             pool: None,
             tx_status: TxStatus::None,
             last_io: Instant::now(),
@@ -269,16 +279,63 @@ impl Conn {
     /// Returns pending result metadata, if any.
     ///
     /// If `Some(_)`, then result is not yet consumed.
-    pub(crate) fn get_pending_result(&self) -> Option<&ResultSetMeta> {
-        self.inner.pending_result.as_ref()
+    pub(crate) fn use_pending_result(
+        &mut self,
+    ) -> std::result::Result<Option<&PendingResult>, ServerError> {
+        if let Err(ref e) = self.inner.pending_result {
+            let e = e.clone();
+            self.inner.pending_result = Ok(None);
+            return Err(e);
+        } else {
+            Ok(self.inner.pending_result.as_ref().unwrap().as_ref())
+        }
+    }
+
+    pub(crate) fn get_pending_result(
+        &self,
+    ) -> std::result::Result<Option<&PendingResult>, &ServerError> {
+        self.inner.pending_result.as_ref().map(|x| x.as_ref())
+    }
+
+    pub(crate) fn has_pending_result(&self) -> bool {
+        matches!(self.inner.pending_result, Err(_))
+            || matches!(self.inner.pending_result, Ok(Some(_)))
     }
 
     /// Sets the given pening result metadata for this connection. Returns the previous value.
     pub(crate) fn set_pending_result(
         &mut self,
         meta: Option<ResultSetMeta>,
-    ) -> Option<ResultSetMeta> {
-        std::mem::replace(&mut self.inner.pending_result, meta)
+    ) -> std::result::Result<Option<PendingResult>, ServerError> {
+        replace(
+            &mut self.inner.pending_result,
+            Ok(meta.map(PendingResult::Pending)),
+        )
+    }
+
+    pub(crate) fn set_pending_result_error(
+        &mut self,
+        error: ServerError,
+    ) -> std::result::Result<Option<PendingResult>, ServerError> {
+        replace(&mut self.inner.pending_result, Err(error))
+    }
+
+    /// Gives the currently pending result to a caller for consumption.
+    pub(crate) fn take_pending_result(
+        &mut self,
+    ) -> std::result::Result<Option<Arc<ResultSetMeta>>, ServerError> {
+        let mut output = None;
+
+        self.inner.pending_result = match replace(&mut self.inner.pending_result, Ok(None))? {
+            Some(PendingResult::Pending(x)) => {
+                let meta = Arc::new(x);
+                output = Some(meta.clone());
+                Ok(Some(PendingResult::Taken(meta)))
+            }
+            x => Ok(x),
+        };
+
+        Ok(output)
     }
 
     /// Returns current status flags.
@@ -597,7 +654,7 @@ impl Conn {
 
     /// Returns `true` for ProgressReport packet.
     fn handle_packet(&mut self, packet: &PooledBuf) -> Result<bool> {
-        let ok_packet = if self.get_pending_result().is_some() {
+        let ok_packet = if self.has_pending_result() {
             ParseBuf(&*packet)
                 .parse::<OkPacketDeserializer<ResultSetTerminator>>(self.capabilities())
                 .map(|x| x.into_inner())
@@ -868,23 +925,37 @@ impl Conn {
 
     /// The purpose of this function is to cleanup a pending result set
     /// for prematurely dropeed connection or query result.
+    ///
+    /// Requires that there are no other references to the pending result.
     pub(crate) async fn drop_result(&mut self) -> Result<()> {
-        match self.inner.pending_result {
-            Some(ResultSetMeta::Text(_)) => {
+        // Map everything into `PendingResult::Pending`
+        let meta = match self.set_pending_result(None)? {
+            Some(PendingResult::Pending(meta)) => Some(meta),
+            Some(PendingResult::Taken(meta)) => {
+                // This also asserts that there is only one reference left to the taken ResultSetMeta,
+                // therefore this result set must be dropped here since it won't be dropped anywhere else.
+                Some(Arc::try_unwrap(meta).expect("Conn::drop_result call on a pending result that may still be droped by someone else"))
+            }
+            None => None,
+        };
+
+        let _ = self.set_pending_result(meta);
+
+        match self.use_pending_result() {
+            Ok(Some(PendingResult::Pending(ResultSetMeta::Text(_)))) => {
                 QueryResult::<'_, '_, TextProtocol>::new(self)
                     .drop_result()
                     .await
             }
-            Some(ResultSetMeta::Binary(_)) => {
+            Ok(Some(PendingResult::Pending(ResultSetMeta::Binary(_)))) => {
                 QueryResult::<'_, '_, BinaryProtocol>::new(self)
                     .drop_result()
                     .await
             }
-            Some(ResultSetMeta::Error(_)) => match self.set_pending_result(None) {
-                Some(ResultSetMeta::Error(err)) => Err(err.into()),
-                _ => unreachable!(),
-            },
-            None => Ok(()),
+            Ok(None) => Ok((/* this case does not require an action */)),
+            Ok(Some(PendingResult::Taken(_))) | Err(_) => {
+                unreachable!("this case must be handled earlier in this function")
+            }
         }
     }
 
@@ -893,7 +964,7 @@ impl Conn {
     /// The purpose of this function, is to cleanup the connection while returning it to a [`Pool`].
     async fn cleanup_for_pool(mut self) -> Result<Self> {
         loop {
-            let result = if self.inner.pending_result.is_some() {
+            let result = if self.has_pending_result() {
                 self.drop_result().await
             } else if self.inner.tx_status != TxStatus::None {
                 self.rollback_transaction().await
@@ -953,7 +1024,7 @@ mod test {
 
     use crate::{
         from_row, params, prelude::*, test_misc::get_opts, BinlogDumpFlags, BinlogRequest, Conn,
-        Error, OptsBuilder, Pool, TxOpts, WhiteListFsLocalInfileHandler,
+        Error, OptsBuilder, Pool, WhiteListFsLocalInfileHandler,
     };
 
     async fn gen_dummy_data() -> super::Result<()> {
@@ -1317,212 +1388,6 @@ mod test {
         Ok(())
     }
 
-    #[tokio::test]
-    async fn dropped_query_result_should_emit_errors_on_cleanup() -> super::Result<()> {
-        use crate::{Error::Server, ServerError};
-        let mut conn = Conn::new(get_opts()).await?;
-        conn.query_iter("SELECT '1'; BLABLA;").await?;
-        assert!(matches!(
-            conn.query_drop("DO 42;").await.unwrap_err(),
-            Server(ServerError { code: 1064, .. })
-        ));
-        conn.disconnect().await?;
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn should_try_collect() -> super::Result<()> {
-        let mut conn = Conn::new(get_opts()).await?;
-        let mut result = conn
-            .query_iter(
-                r"SELECT 'hello', 123
-                    UNION ALL
-                    SELECT 'world', 'bar'
-                    UNION ALL
-                    SELECT 'hello', 123
-                ",
-            )
-            .await?;
-        let mut rows = result.try_collect::<(String, u8)>().await?;
-        assert!(rows.pop().unwrap().is_ok());
-        assert!(rows.pop().unwrap().is_err());
-        assert!(rows.pop().unwrap().is_ok());
-        result.drop_result().await?;
-        conn.disconnect().await?;
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn should_try_collect_and_drop() -> super::Result<()> {
-        let mut conn = Conn::new(get_opts()).await?;
-        let mut rows = conn
-            .query_iter(
-                r"SELECT 'hello', 123
-                    UNION ALL
-                    SELECT 'world', 'bar'
-                    UNION ALL
-                    SELECT 'hello', 123;
-                    SELECT 'foo', 255;
-                ",
-            )
-            .await?
-            .try_collect_and_drop::<(String, u8)>()
-            .await?;
-        assert!(rows.pop().unwrap().is_ok());
-        assert!(rows.pop().unwrap().is_err());
-        assert!(rows.pop().unwrap().is_ok());
-        conn.disconnect().await?;
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn should_handle_mutliresult_set() -> super::Result<()> {
-        let mut conn = Conn::new(get_opts()).await?;
-        let mut result = conn
-            .query_iter(
-                r"SELECT 'hello', 123
-                    UNION ALL
-                    SELECT 'world', 231;
-                    SELECT 'foo', 255;
-                ",
-            )
-            .await?;
-        let rows_1 = result.collect::<(String, u8)>().await?;
-        let rows_2 = result.collect_and_drop().await?;
-        conn.disconnect().await?;
-
-        assert_eq!((String::from("hello"), 123), rows_1[0]);
-        assert_eq!((String::from("world"), 231), rows_1[1]);
-        assert_eq!((String::from("foo"), 255), rows_2[0]);
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn should_map_resultset() -> super::Result<()> {
-        let mut conn = Conn::new(get_opts()).await?;
-        let mut result = conn
-            .query_iter(
-                r"
-                    SELECT 'hello', 123
-                    UNION ALL
-                    SELECT 'world', 231;
-                    SELECT 'foo', 255;
-                ",
-            )
-            .await?;
-
-        let rows_1 = result.map(|row| from_row::<(String, u8)>(row)).await?;
-        let rows_2 = result.map_and_drop(from_row).await?;
-        conn.disconnect().await?;
-
-        assert_eq!((String::from("hello"), 123), rows_1[0]);
-        assert_eq!((String::from("world"), 231), rows_1[1]);
-        assert_eq!((String::from("foo"), 255), rows_2[0]);
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn should_reduce_resultset() -> super::Result<()> {
-        let mut conn = Conn::new(get_opts()).await?;
-        let mut result = conn
-            .query_iter(
-                r"SELECT 5
-                    UNION ALL
-                    SELECT 6;
-                    SELECT 7;",
-            )
-            .await?;
-        let reduced = result
-            .reduce(0, |mut acc, row| {
-                acc += from_row::<i32>(row);
-                acc
-            })
-            .await?;
-        let rows_2 = result.collect_and_drop::<i32>().await?;
-        conn.disconnect().await?;
-        assert_eq!(11, reduced);
-        assert_eq!(7, rows_2[0]);
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn should_handle_multi_result_sets_where_some_results_have_no_output() -> super::Result<()>
-    {
-        const QUERY: &str = r"SELECT 1;
-            UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
-            SELECT 2;
-            SELECT 3;
-            UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
-            UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
-            SELECT 4;";
-
-        let mut c = Conn::new(get_opts()).await?;
-        c.query_drop("CREATE TEMPORARY TABLE time_zone (Time_zone_id INT)")
-            .await
-            .unwrap();
-        let mut t = c.start_transaction(TxOpts::new()).await?;
-        t.query_drop(QUERY).await?;
-        let r = t.query_iter(QUERY).await?;
-        let out = r.collect_and_drop::<u8>().await?;
-        assert_eq!(vec![1], out);
-        let r = t.query_iter(QUERY).await?;
-        r.for_each_and_drop(|x| assert_eq!(from_row::<u8>(x), 1))
-            .await?;
-        let r = t.query_iter(QUERY).await?;
-        let out = r.map_and_drop(|row| from_row::<u8>(row)).await?;
-        assert_eq!(vec![1], out);
-        let r = t.query_iter(QUERY).await?;
-        let out = r
-            .reduce_and_drop(0u8, |acc, x| acc + from_row::<u8>(x))
-            .await?;
-        assert_eq!(1, out);
-        t.query_drop(QUERY).await?;
-        t.commit().await?;
-        let result = c.exec_first("SELECT 1", ()).await?;
-        c.disconnect().await?;
-        assert_eq!(result, Some(1_u8));
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn should_iterate_over_resultset() -> super::Result<()> {
-        use std::sync::{
-            atomic::{AtomicUsize, Ordering},
-            Arc,
-        };
-
-        let acc = Arc::new(AtomicUsize::new(0));
-
-        let mut conn = Conn::new(get_opts()).await?;
-        let mut result = conn
-            .query_iter(
-                r"SELECT 2
-                    UNION ALL
-                    SELECT 3;
-                    SELECT 5;",
-            )
-            .await?;
-        result
-            .for_each({
-                let acc = acc.clone();
-                move |row| {
-                    acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
-                }
-            })
-            .await?;
-        result
-            .for_each_and_drop({
-                let acc = acc.clone();
-                move |row| {
-                    acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
-                }
-            })
-            .await?;
-        conn.disconnect().await?;
-        assert_eq!(acc.load(Ordering::SeqCst), 10);
-        Ok(())
-    }
-
     #[tokio::test]
     async fn should_prepare_statement() -> super::Result<()> {
         let mut conn = Conn::new(get_opts()).await?;
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
index 74c94e8f..34303845 100644
--- a/src/conn/pool/mod.rs
+++ b/src/conn/pool/mod.rs
@@ -184,7 +184,7 @@ impl Pool {
             && !conn.inner.disconnected
             && !conn.expired()
             && conn.inner.tx_status == TxStatus::None
-            && conn.inner.pending_result.is_none()
+            && !conn.has_pending_result()
             && !self.inner.close.load(atomic::Ordering::Acquire)
         {
             let mut exchange = self.inner.exchange.lock().unwrap();
diff --git a/src/conn/pool/recycler.rs b/src/conn/pool/recycler.rs
index b8fae75e..b60066d1 100644
--- a/src/conn/pool/recycler.rs
+++ b/src/conn/pool/recycler.rs
@@ -65,9 +65,7 @@ impl Future for Recycler {
                 if $conn.inner.stream.is_none() || $conn.inner.disconnected {
                     // drop unestablished connection
                     $self.discard.push(futures_util::future::ok(()).boxed());
-                } else if $conn.inner.tx_status != TxStatus::None
-                    || $conn.inner.pending_result.is_some()
-                {
+                } else if $conn.inner.tx_status != TxStatus::None || $conn.has_pending_result() {
                     $self.cleaning.push($conn.cleanup_for_pool().boxed());
                 } else if $conn.expired() || close {
                     $self.discard.push($conn.close_conn().boxed());
diff --git a/src/conn/routines/helpers.rs b/src/conn/routines/helpers.rs
index 16a39a01..0bf5dba9 100644
--- a/src/conn/routines/helpers.rs
+++ b/src/conn/routines/helpers.rs
@@ -10,10 +10,7 @@ use mysql_common::{
 };
 use tokio::io::AsyncReadExt;
 
-use crate::{
-    queryable::{query_result::ResultSetMeta, Protocol},
-    Conn, DriverError, Error,
-};
+use crate::{queryable::Protocol, Conn, DriverError, Error};
 
 impl Conn {
     /// Helper, that sends all `Value::Bytes` in the given list of paramenters as long data.
@@ -59,7 +56,7 @@ impl Conn {
             }
             Err(Error::Server(error)) => {
                 // error will be consumed as a part of a multi-result set
-                self.set_pending_result(Some(ResultSetMeta::Error(error)));
+                self.set_pending_result_error(error)?;
                 return Ok(());
             }
             Err(err) => {
@@ -72,7 +69,7 @@ impl Conn {
             Some(0x00) => {
                 self.set_pending_result(Some(P::result_set_meta(Arc::from(
                     Vec::new().into_boxed_slice(),
-                ))));
+                ))))?;
             }
             Some(0xFB) => self.handle_local_infile::<P>(&*packet).await?,
             _ => self.handle_result_set::<P>(&*packet).await?,
@@ -106,7 +103,7 @@ impl Conn {
         self.read_packet().await?;
         self.set_pending_result(Some(P::result_set_meta(Arc::from(
             Vec::new().into_boxed_slice(),
-        ))));
+        ))))?;
         Ok(())
     }
 
@@ -120,7 +117,7 @@ impl Conn {
         let column_count = packet.read_lenenc_int()?;
         let columns = self.read_column_defs(column_count as usize).await?;
         let meta = P::result_set_meta(Arc::from(columns.into_boxed_slice()));
-        self.set_pending_result(Some(meta));
+        self.set_pending_result(Some(meta))?;
         Ok(())
     }
 }
diff --git a/src/connection_like/mod.rs b/src/connection_like/mod.rs
index bb322187..b47adfaa 100644
--- a/src/connection_like/mod.rs
+++ b/src/connection_like/mod.rs
@@ -79,6 +79,17 @@ impl<'a, 't: 'a, T: Into<Connection<'a, 't>> + Send> ToConnection<'a, 't> for T
     }
 }
 
+impl ToConnection<'static, 'static> for Pool {
+    fn to_connection(self) -> ToConnectionResult<'static, 'static> {
+        let fut = async move {
+            let conn = self.get_conn().await?;
+            Ok(conn.into())
+        }
+        .boxed();
+        ToConnectionResult::Mediate(fut)
+    }
+}
+
 impl<'a> ToConnection<'a, 'static> for &'a Pool {
     fn to_connection(self) -> ToConnectionResult<'a, 'static> {
         let fut = async move {
diff --git a/src/lib.rs b/src/lib.rs
index bdb7b428..545f62f4 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -7,10 +7,10 @@
 // modified, or distributed except according to those terms.
 
 //! ## mysql-async
-//! Tokio based asynchronous MySql client library for rust programming language.
+//! Tokio based asynchronous MySql client library for The Rust Programming Language.
 //!
 //! ### Installation
-//! Library hosted on [crates.io](https://crates.io/crates/mysql_async/).
+//! The library is hosted on [crates.io](https://crates.io/crates/mysql_async/).
 //!
 //! ```toml
 //! [dependencies]
@@ -47,41 +47,34 @@
 //!     let pool = mysql_async::Pool::new(database_url);
 //!     let mut conn = pool.get_conn().await?;
 //!
-//!     // Create temporary table
-//!     conn.query_drop(
-//!         r"CREATE TEMPORARY TABLE payment (
-//!             customer_id int not null,
-//!             amount int not null,
-//!             account_name text
-//!         )"
-//!     ).await?;
+//!     // Create a temporary table
+//!     r"CREATE TEMPORARY TABLE payment (
+//!         customer_id int not null,
+//!         amount int not null,
+//!         account_name text
+//!     )".ignore(&mut conn).await?;
 //!
 //!     // Save payments
-//!     let params = payments.clone().into_iter().map(|payment| {
-//!         params! {
+//!     r"INSERT INTO payment (customer_id, amount, account_name)
+//!       VALUES (:customer_id, :amount, :account_name)"
+//!         .with(payments.iter().map(|payment| params! {
 //!             "customer_id" => payment.customer_id,
 //!             "amount" => payment.amount,
-//!             "account_name" => payment.account_name,
-//!         }
-//!     });
+//!             "account_name" => payment.account_name.as_ref(),
+//!         }))
+//!         .batch(&mut conn)
+//!         .await?;
 //!
-//!     conn.exec_batch(
-//!         r"INSERT INTO payment (customer_id, amount, account_name)
-//!           VALUES (:customer_id, :amount, :account_name)",
-//!         params,
-//!     ).await?;
-//!
-//!     // Load payments from database. Type inference will work here.
-//!     let loaded_payments = conn.exec_map(
-//!         "SELECT customer_id, amount, account_name FROM payment",
-//!         (),
-//!         |(customer_id, amount, account_name)| Payment { customer_id, amount, account_name },
-//!     ).await?;
+//!     // Load payments from the database. Type inference will work here.
+//!     let loaded_payments = "SELECT customer_id, amount, account_name FROM payment"
+//!         .with(())
+//!         .map(&mut conn, |(customer_id, amount, account_name)| Payment { customer_id, amount, account_name })
+//!         .await?;
 //!
 //!     // Dropped connection will go to the pool
 //!     drop(conn);
 //!
-//!     // Pool must be disconnected explicitly because
+//!     // The Pool must be disconnected explicitly because
 //!     // it's an asynchronous operation.
 //!     pool.disconnect().await?;
 //!
@@ -93,7 +86,7 @@
 //! ```
 
 #![recursion_limit = "1024"]
-#![cfg_attr(feature = "nightly", feature(test, const_fn))]
+#![cfg_attr(feature = "nightly", feature(test))]
 
 #[cfg(feature = "nightly")]
 extern crate test;
@@ -185,7 +178,7 @@ pub use mysql_common::value::convert::{from_value, from_value_opt, FromValueErro
 pub use mysql_common::value::json::{Deserialized, Serialized};
 
 #[doc(inline)]
-pub use self::queryable::query_result::QueryResult;
+pub use self::queryable::query_result::{result_set_stream::ResultSetStream, QueryResult};
 
 #[doc(inline)]
 pub use self::queryable::transaction::{Transaction, TxOpts};
@@ -236,9 +229,12 @@ pub mod prelude {
     impl<T: crate::queryable::stmt::StatementLike> StatementLike for T {}
 
     /// Everything that is a connection.
+    ///
+    /// Note that you could obtain a `'static` connection by giving away `Conn` or `Pool`.
     pub trait ToConnection<'a, 't: 'a>: crate::connection_like::ToConnection<'a, 't> {}
     // explicitly implemented because of rusdoc
     impl<'a> ToConnection<'a, 'static> for &'a crate::Pool {}
+    impl<'a> ToConnection<'static, 'static> for crate::Pool {}
     impl ToConnection<'static, 'static> for crate::Conn {}
     impl<'a> ToConnection<'a, 'static> for &'a mut crate::Conn {}
     impl<'a, 't> ToConnection<'a, 't> for &'a mut crate::Transaction<'t> {}
diff --git a/src/query.rs b/src/query.rs
index f2b97ca8..9267b547 100644
--- a/src/query.rs
+++ b/src/query.rs
@@ -12,7 +12,7 @@ use crate::{
     connection_like::ToConnectionResult,
     from_row,
     prelude::{FromRow, StatementLike, ToConnection},
-    BinaryProtocol, BoxFuture, Params, QueryResult, TextProtocol,
+    BinaryProtocol, BoxFuture, Params, QueryResult, ResultSetStream, TextProtocol,
 };
 
 /// MySql text query.
@@ -119,6 +119,32 @@ pub trait Query: Send + Sized {
         .boxed()
     }
 
+    /// Returns a stream over the first result set.
+    ///
+    /// This method corresponds to [`QueryResult::stream_and_drop`][stream_and_drop].
+    ///
+    /// [stream_and_drop]: crate::QueryResult::stream_and_drop
+    fn stream<'a, 't: 'a, T, C>(
+        self,
+        conn: C,
+    ) -> BoxFuture<'a, ResultSetStream<'a, 'a, 't, T, Self::Protocol>>
+    where
+        Self: 'a,
+        Self::Protocol: Unpin,
+        T: Unpin + FromRow + Send + 'static,
+        C: ToConnection<'a, 't> + 'a,
+    {
+        async move {
+            self.run(conn)
+                .await?
+                .stream_and_drop()
+                .await
+                .transpose()
+                .expect("At least one result set is expected")
+        }
+        .boxed()
+    }
+
     /// This method corresponds to [`Queryable::query_drop`][query_drop].
     ///
     /// [query_drop]: crate::prelude::Queryable::query_drop
diff --git a/src/queryable/mod.rs b/src/queryable/mod.rs
index c6b4cf5f..bd609bf5 100644
--- a/src/queryable/mod.rs
+++ b/src/queryable/mod.rs
@@ -29,7 +29,7 @@ use crate::{
     error::*,
     prelude::{FromRow, StatementLike},
     queryable::query_result::ResultSetMeta,
-    BoxFuture, Column, Conn, Params, Row,
+    BoxFuture, Column, Conn, Params, ResultSetStream, Row,
 };
 
 pub mod query_result;
@@ -157,167 +157,6 @@ pub trait Queryable: Send {
         P: Into<Params>;
 
     /// Performs the given query and collects the first result set.
-    fn query<'a, T, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Vec<T>>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static;
-
-    /// Performs the given query and returns the first row of the first result set.
-    fn query_first<'a, T, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Option<T>>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static;
-
-    /// Performs the given query and maps each row of the first result set.
-    fn query_map<'a, T, F, Q, U>(&'a mut self, query: Q, f: F) -> BoxFuture<'a, Vec<U>>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static,
-        F: FnMut(T) -> U + Send + 'a,
-        U: Send;
-
-    /// Performs the given query and folds the first result set to a single value.
-    fn query_fold<'a, T, F, Q, U>(&'a mut self, query: Q, init: U, f: F) -> BoxFuture<'a, U>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static,
-        F: FnMut(U, T) -> U + Send + 'a,
-        U: Send + 'a;
-
-    /// Performs the given query and drops the query result.
-    fn query_drop<'a, Q>(&'a mut self, query: Q) -> BoxFuture<'a, ()>
-    where
-        Q: AsRef<str> + Send + Sync + 'a;
-
-    /// Exectues the given statement for each item in the given params iterator.
-    ///
-    /// It'll prepare `stmt` (once), if necessary.
-    fn exec_batch<'a: 'b, 'b, S, P, I>(&'a mut self, stmt: S, params_iter: I) -> BoxFuture<'b, ()>
-    where
-        S: StatementLike + 'b,
-        I: IntoIterator<Item = P> + Send + 'b,
-        I::IntoIter: Send,
-        P: Into<Params> + Send;
-
-    /// Exectues the given statement and collects the first result set.
-    ///
-    /// It'll prepare `stmt`, if necessary.
-    fn exec<'a: 'b, 'b, T, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, Vec<T>>
-    where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static;
-
-    /// Exectues the given statement and returns the first row of the first result set.
-    ///
-    /// It'll prepare `stmt`, if necessary.
-    fn exec_first<'a: 'b, 'b, T, S, P>(
-        &'a mut self,
-        stmt: S,
-        params: P,
-    ) -> BoxFuture<'b, Option<T>>
-    where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static;
-
-    /// Exectues the given stmt and maps each row of the first result set.
-    ///
-    /// It'll prepare `stmt`, if necessary.
-    fn exec_map<'a: 'b, 'b, T, S, P, U, F>(
-        &'a mut self,
-        stmt: S,
-        params: P,
-        f: F,
-    ) -> BoxFuture<'b, Vec<U>>
-    where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static,
-        F: FnMut(T) -> U + Send + 'a,
-        U: Send + 'a;
-
-    /// Exectues the given stmt and folds the first result set to a signel value.
-    ///
-    /// It'll prepare `stmt`, if necessary.
-    fn exec_fold<'a: 'b, 'b, T, S, P, U, F>(
-        &'a mut self,
-        stmt: S,
-        params: P,
-        init: U,
-        f: F,
-    ) -> BoxFuture<'b, U>
-    where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static,
-        F: FnMut(U, T) -> U + Send + 'a,
-        U: Send + 'a;
-
-    /// Exectues the given statement and drops the result.
-    fn exec_drop<'a: 'b, 'b, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, ()>
-    where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b;
-}
-
-impl Queryable for Conn {
-    fn ping(&mut self) -> BoxFuture<'_, ()> {
-        async move {
-            self.routine(PingRoutine).await?;
-            Ok(())
-        }
-        .boxed()
-    }
-
-    fn query_iter<'a, Q>(
-        &'a mut self,
-        query: Q,
-    ) -> BoxFuture<'a, QueryResult<'a, 'static, TextProtocol>>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-    {
-        async move {
-            self.routine(QueryRoutine::new(query.as_ref().as_bytes()))
-                .await?;
-            Ok(QueryResult::new(self))
-        }
-        .boxed()
-    }
-
-    fn prep<'a, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Statement>
-    where
-        Q: AsRef<str> + Sync + Send + 'a,
-    {
-        async move { self.get_statement(query.as_ref()).await }.boxed()
-    }
-
-    fn close(&mut self, stmt: Statement) -> BoxFuture<'_, ()> {
-        async move {
-            self.stmt_cache_mut().remove(stmt.id());
-            self.close_statement(stmt.id()).await
-        }
-        .boxed()
-    }
-
-    fn exec_iter<'a: 's, 's, Q, P>(
-        &'a mut self,
-        stmt: Q,
-        params: P,
-    ) -> BoxFuture<'s, QueryResult<'a, 'static, BinaryProtocol>>
-    where
-        Q: StatementLike + 'a,
-        P: Into<Params>,
-    {
-        let params = params.into();
-        async move {
-            let statement = self.get_statement(stmt).await?;
-            self.execute_statement(&statement, params).await?;
-            Ok(QueryResult::new(self))
-        }
-        .boxed()
-    }
-
     fn query<'a, T, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Vec<T>>
     where
         Q: AsRef<str> + Send + Sync + 'a,
@@ -326,6 +165,7 @@ impl Queryable for Conn {
         async move { self.query_iter(query).await?.collect_and_drop::<T>().await }.boxed()
     }
 
+    /// Performs the given query and returns the first row of the first result set.
     fn query_first<'a, T, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Option<T>>
     where
         Q: AsRef<str> + Send + Sync + 'a,
@@ -344,6 +184,7 @@ impl Queryable for Conn {
         .boxed()
     }
 
+    /// Performs the given query and maps each row of the first result set.
     fn query_map<'a, T, F, Q, U>(&'a mut self, query: Q, mut f: F) -> BoxFuture<'a, Vec<U>>
     where
         Q: AsRef<str> + Send + Sync + 'a,
@@ -361,6 +202,7 @@ impl Queryable for Conn {
         .boxed()
     }
 
+    /// Performs the given query and folds the first result set to a single value.
     fn query_fold<'a, T, F, Q, U>(&'a mut self, query: Q, init: U, mut f: F) -> BoxFuture<'a, U>
     where
         Q: AsRef<str> + Send + Sync + 'a,
@@ -377,6 +219,7 @@ impl Queryable for Conn {
         .boxed()
     }
 
+    /// Performs the given query and drops the query result.
     fn query_drop<'a, Q>(&'a mut self, query: Q) -> BoxFuture<'a, ()>
     where
         Q: AsRef<str> + Send + Sync + 'a,
@@ -384,26 +227,19 @@ impl Queryable for Conn {
         async move { self.query_iter(query).await?.drop_result().await }.boxed()
     }
 
+    /// Exectues the given statement for each item in the given params iterator.
+    ///
+    /// It'll prepare `stmt` (once), if necessary.
     fn exec_batch<'a: 'b, 'b, S, P, I>(&'a mut self, stmt: S, params_iter: I) -> BoxFuture<'b, ()>
     where
         S: StatementLike + 'b,
         I: IntoIterator<Item = P> + Send + 'b,
         I::IntoIter: Send,
-        P: Into<Params> + Send,
-    {
-        async move {
-            let statement = self.get_statement(stmt).await?;
-            for params in params_iter {
-                self.execute_statement(&statement, params).await?;
-                QueryResult::<BinaryProtocol>::new(&mut *self)
-                    .drop_result()
-                    .await?;
-            }
-            Ok(())
-        }
-        .boxed()
-    }
+        P: Into<Params> + Send;
 
+    /// Exectues the given statement and collects the first result set.
+    ///
+    /// It'll prepare `stmt`, if necessary.
     fn exec<'a: 'b, 'b, T, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, Vec<T>>
     where
         S: StatementLike + 'b,
@@ -419,6 +255,9 @@ impl Queryable for Conn {
         .boxed()
     }
 
+    /// Exectues the given statement and returns the first row of the first result set.
+    ///
+    /// It'll prepare `stmt`, if necessary.
     fn exec_first<'a: 'b, 'b, T, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, Option<T>>
     where
         S: StatementLike + 'b,
@@ -438,6 +277,9 @@ impl Queryable for Conn {
         .boxed()
     }
 
+    /// Exectues the given stmt and maps each row of the first result set.
+    ///
+    /// It'll prepare `stmt`, if necessary.
     fn exec_map<'a: 'b, 'b, T, S, P, U, F>(
         &'a mut self,
         stmt: S,
@@ -461,6 +303,9 @@ impl Queryable for Conn {
         .boxed()
     }
 
+    /// Exectues the given stmt and folds the first result set to a signel value.
+    ///
+    /// It'll prepare `stmt`, if necessary.
     fn exec_fold<'a: 'b, 'b, T, S, P, U, F>(
         &'a mut self,
         stmt: S,
@@ -484,6 +329,7 @@ impl Queryable for Conn {
         .boxed()
     }
 
+    /// Exectues the given statement and drops the result.
     fn exec_drop<'a: 'b, 'b, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, ()>
     where
         S: StatementLike + 'b,
@@ -491,11 +337,65 @@ impl Queryable for Conn {
     {
         async move { self.exec_iter(stmt, params).await?.drop_result().await }.boxed()
     }
+
+    /// Returns a stream over the first result set.
+    ///
+    /// This method corresponds to [`QueryResult::stream_and_drop`][stream_and_drop].
+    ///
+    /// [stream_and_drop]: crate::QueryResult::stream_and_drop
+    fn query_stream<'a, T, Q>(
+        &'a mut self,
+        query: Q,
+    ) -> BoxFuture<'a, ResultSetStream<'a, 'a, 'static, T, TextProtocol>>
+    where
+        T: Unpin + FromRow + Send + 'static,
+        Q: AsRef<str> + Send + Sync + 'a,
+    {
+        async move {
+            self.query_iter(query)
+                .await?
+                .stream_and_drop()
+                .await
+                .transpose()
+                .expect("At least one result set is expected")
+        }
+        .boxed()
+    }
+
+    /// Returns a stream over the first result set.
+    ///
+    /// This method corresponds to [`QueryResult::stream_and_drop`][stream_and_drop].
+    ///
+    /// [stream_and_drop]: crate::QueryResult::stream_and_drop
+    fn exec_stream<'a: 's, 's, T, Q, P>(
+        &'a mut self,
+        stmt: Q,
+        params: P,
+    ) -> BoxFuture<'s, ResultSetStream<'a, 'a, 'static, T, BinaryProtocol>>
+    where
+        T: Unpin + FromRow + Send + 'static,
+        Q: StatementLike + 'a,
+        P: Into<Params> + Send + 's,
+    {
+        async move {
+            self.exec_iter(stmt, params)
+                .await?
+                .stream_and_drop()
+                .await
+                .transpose()
+                .expect("At least one result set is expected")
+        }
+        .boxed()
+    }
 }
 
-impl Queryable for Transaction<'_> {
+impl Queryable for Conn {
     fn ping(&mut self) -> BoxFuture<'_, ()> {
-        self.0.ping()
+        async move {
+            self.routine(PingRoutine).await?;
+            Ok(())
+        }
+        .boxed()
     }
 
     fn query_iter<'a, Q>(
@@ -505,18 +405,29 @@ impl Queryable for Transaction<'_> {
     where
         Q: AsRef<str> + Send + Sync + 'a,
     {
-        self.0.query_iter(query)
+        async move {
+            self.routine(QueryRoutine::new(query.as_ref().as_bytes()))
+                .await?;
+            Ok(QueryResult::new(self))
+        }
+        .boxed()
     }
 
     fn prep<'a, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Statement>
     where
         Q: AsRef<str> + Sync + Send + 'a,
     {
-        self.0.prep(query)
+        async move { self.get_statement(query.as_ref()).await }.boxed()
     }
+
     fn close(&mut self, stmt: Statement) -> BoxFuture<'_, ()> {
-        self.0.close(stmt)
+        async move {
+            self.stmt_cache_mut().remove(stmt.id());
+            self.close_statement(stmt.id()).await
+        }
+        .boxed()
     }
+
     fn exec_iter<'a: 's, 's, Q, P>(
         &'a mut self,
         stmt: Q,
@@ -526,46 +437,15 @@ impl Queryable for Transaction<'_> {
         Q: StatementLike + 'a,
         P: Into<Params>,
     {
-        self.0.exec_iter(stmt, params)
-    }
-    fn query<'a, T, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Vec<T>>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static,
-    {
-        self.0.query(query)
-    }
-    fn query_first<'a, T, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Option<T>>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static,
-    {
-        self.0.query_first(query)
-    }
-    fn query_map<'a, T, F, Q, U>(&'a mut self, query: Q, f: F) -> BoxFuture<'a, Vec<U>>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static,
-        F: FnMut(T) -> U + Send + 'a,
-        U: Send,
-    {
-        self.0.query_map(query, f)
-    }
-    fn query_fold<'a, T, F, Q, U>(&'a mut self, query: Q, init: U, f: F) -> BoxFuture<'a, U>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-        T: FromRow + Send + 'static,
-        F: FnMut(U, T) -> U + Send + 'a,
-        U: Send + 'a,
-    {
-        self.0.query_fold(query, init, f)
-    }
-    fn query_drop<'a, Q>(&'a mut self, query: Q) -> BoxFuture<'a, ()>
-    where
-        Q: AsRef<str> + Send + Sync + 'a,
-    {
-        self.0.query_drop(query)
+        let params = params.into();
+        async move {
+            let statement = self.get_statement(stmt).await?;
+            self.execute_statement(&statement, params).await?;
+            Ok(QueryResult::new(self))
+        }
+        .boxed()
     }
+
     fn exec_batch<'a: 'b, 'b, S, P, I>(&'a mut self, stmt: S, params_iter: I) -> BoxFuture<'b, ()>
     where
         S: StatementLike + 'b,
@@ -573,61 +453,66 @@ impl Queryable for Transaction<'_> {
         I::IntoIter: Send,
         P: Into<Params> + Send,
     {
-        self.0.exec_batch(stmt, params_iter)
+        async move {
+            let statement = self.get_statement(stmt).await?;
+            for params in params_iter {
+                self.execute_statement(&statement, params).await?;
+                QueryResult::<BinaryProtocol>::new(&mut *self)
+                    .drop_result()
+                    .await?;
+            }
+            Ok(())
+        }
+        .boxed()
     }
-    fn exec<'a: 'b, 'b, T, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, Vec<T>>
-    where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static,
-    {
-        self.0.exec(stmt, params)
+}
+
+impl Queryable for Transaction<'_> {
+    fn ping(&mut self) -> BoxFuture<'_, ()> {
+        self.0.ping()
     }
-    fn exec_first<'a: 'b, 'b, T, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, Option<T>>
+
+    fn query_iter<'a, Q>(
+        &'a mut self,
+        query: Q,
+    ) -> BoxFuture<'a, QueryResult<'a, 'static, TextProtocol>>
     where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static,
+        Q: AsRef<str> + Send + Sync + 'a,
     {
-        self.0.exec_first(stmt, params)
+        self.0.query_iter(query)
     }
-    fn exec_map<'a: 'b, 'b, T, S, P, U, F>(
-        &'a mut self,
-        stmt: S,
-        params: P,
-        f: F,
-    ) -> BoxFuture<'b, Vec<U>>
+
+    fn prep<'a, Q>(&'a mut self, query: Q) -> BoxFuture<'a, Statement>
     where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static,
-        F: FnMut(T) -> U + Send + 'a,
-        U: Send + 'a,
+        Q: AsRef<str> + Sync + Send + 'a,
     {
-        self.0.exec_map(stmt, params, f)
+        self.0.prep(query)
     }
-    fn exec_fold<'a: 'b, 'b, T, S, P, U, F>(
+
+    fn close(&mut self, stmt: Statement) -> BoxFuture<'_, ()> {
+        self.0.close(stmt)
+    }
+
+    fn exec_iter<'a: 's, 's, Q, P>(
         &'a mut self,
-        stmt: S,
+        stmt: Q,
         params: P,
-        init: U,
-        f: F,
-    ) -> BoxFuture<'b, U>
+    ) -> BoxFuture<'s, QueryResult<'a, 'static, BinaryProtocol>>
     where
-        S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
-        T: FromRow + Send + 'static,
-        F: FnMut(U, T) -> U + Send + 'a,
-        U: Send + 'a,
+        Q: StatementLike + 'a,
+        P: Into<Params>,
     {
-        self.0.exec_fold(stmt, params, init, f)
+        self.0.exec_iter(stmt, params)
     }
-    fn exec_drop<'a: 'b, 'b, S, P>(&'a mut self, stmt: S, params: P) -> BoxFuture<'b, ()>
+
+    fn exec_batch<'a: 'b, 'b, S, P, I>(&'a mut self, stmt: S, params_iter: I) -> BoxFuture<'b, ()>
     where
         S: StatementLike + 'b,
-        P: Into<Params> + Send + 'b,
+        I: IntoIterator<Item = P> + Send + 'b,
+        I::IntoIter: Send,
+        P: Into<Params> + Send,
     {
-        self.0.exec_drop(stmt, params)
+        self.0.exec_batch(stmt, params_iter)
     }
 }
 
diff --git a/src/queryable/query_result.rs b/src/queryable/query_result/mod.rs
similarity index 68%
rename from src/queryable/query_result.rs
rename to src/queryable/query_result/mod.rs
index 806e5715..21f45f96 100644
--- a/src/queryable/query_result.rs
+++ b/src/queryable/query_result/mod.rs
@@ -8,16 +8,19 @@
 
 use mysql_common::row::convert::FromRowError;
 
-use std::{borrow::Cow, marker::PhantomData, result::Result as StdResult, sync::Arc};
+use std::{borrow::Cow, fmt, marker::PhantomData, result::Result as StdResult, sync::Arc};
 
 use crate::{
-    conn::routines::NextSetRoutine,
+    conn::{routines::NextSetRoutine, PendingResult},
     connection_like::Connection,
     error::*,
     prelude::{FromRow, Protocol},
     Column, Row,
 };
 
+pub mod result_set_stream;
+mod tests;
+
 /// Result set metadata.
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum ResultSetMeta {
@@ -25,15 +28,12 @@ pub enum ResultSetMeta {
     Text(Arc<[Column]>),
     /// Binary result set, that may contain rows.
     Binary(Arc<[Column]>),
-    /// Error result set.
-    Error(ServerError),
 }
 
 impl ResultSetMeta {
-    fn columns(&self) -> StdResult<&Arc<[Column]>, &ServerError> {
+    fn columns(&self) -> &Arc<[Column]> {
         match self {
-            ResultSetMeta::Text(cols) | ResultSetMeta::Binary(cols) => Ok(cols),
-            ResultSetMeta::Error(err) => Err(err),
+            ResultSetMeta::Text(cols) | ResultSetMeta::Binary(cols) => cols,
         }
     }
 }
@@ -68,12 +68,20 @@ impl ResultSetMeta {
 ///
 /// # conn.disconnect().await }
 /// ```
-#[derive(Debug)]
 pub struct QueryResult<'a, 't: 'a, P> {
     conn: Connection<'a, 't>,
     __phantom: PhantomData<P>,
 }
 
+impl<'a, 't: 'a, P> fmt::Debug for QueryResult<'a, 't, P> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("QueryResult")
+            .field("conn", &self.conn)
+            .field("__phantom", &"PhantomData<P>")
+            .finish()
+    }
+}
+
 impl<'a, 't: 'a, P> QueryResult<'a, 't, P>
 where
     P: Protocol,
@@ -91,66 +99,101 @@ where
     fn has_rows(&self) -> bool {
         self.conn
             .get_pending_result()
-            .and_then(|meta| meta.columns().map(|columns| columns.len() > 0).ok())
+            .map(|pending_result| match pending_result {
+                Some(PendingResult::Pending(meta)) => meta.columns().len() > 0,
+                Some(PendingResult::Taken(meta)) => meta.columns().len() > 0,
+                None => false,
+            })
             .unwrap_or(false)
     }
 
     /// `true` if there are no more rows nor result sets in this query.
+    ///
+    /// This function will return `false` if the last result set was taken
+    /// by the [`QueryResult::stream`] that was dropped before being fully consumed
+    /// (i.e. caller will get `false` even if QueryResult data is reachable only for library internals).
     pub fn is_empty(&self) -> bool {
         !self.has_rows() && !self.conn.more_results_exists()
     }
 
-    pub async fn next(&mut self) -> Result<Option<Row>> {
-        loop {
-            let columns = match self.conn.get_pending_result() {
-                Some(ResultSetMeta::Text(cols)) | Some(ResultSetMeta::Binary(cols)) => {
-                    Ok(Some(cols.clone()))
-                }
-                Some(ResultSetMeta::Error(err)) => Err(Error::from(err.clone())),
-                None => Ok(None),
-            };
+    /// Low-level function that reads a result set row.
+    ///
+    /// Returns `None` if there are no more rows in the current set.
+    async fn next_row(&mut self, columns: Arc<[Column]>) -> crate::Result<Option<Row>> {
+        let mut row = None;
 
-            match columns {
-                Ok(Some(columns)) => {
-                    if columns.is_empty() {
-                        // Empty, but not yet consumed result set.
-                        self.conn.set_pending_result(None);
-                    } else {
-                        // Not yet consumed non-empty result set.
-                        let packet = match self.conn.read_packet().await {
-                            Ok(packet) => packet,
-                            Err(err) => {
-                                // Next row contained an error. No more data will follow.
-                                self.conn.set_pending_result(None);
-                                return Err(err);
-                            }
-                        };
-
-                        if P::is_last_result_set_packet(self.conn.capabilities(), &packet) {
-                            // `packet` is a result set terminator.
-                            self.conn.set_pending_result(None);
-                        } else {
-                            // `packet` is a result set row.
-                            return Ok(Some(P::read_result_set_row(&packet, columns)?));
-                        }
-                    }
-                }
-                Ok(None) => {
-                    // Consumed result set.
-                    if self.conn.more_results_exists() {
-                        // More data will follow.
-                        self.conn.routine(NextSetRoutine::<P>::new()).await?;
-                        return Ok(None);
-                    } else {
-                        // The end of a query result.
-                        return Ok(None);
-                    }
-                }
+        if columns.is_empty() {
+            // Empty, but not yet consumed result set.
+            self.conn.set_pending_result(None)?;
+        } else {
+            // Not yet consumed non-empty result set.
+            let packet = match self.conn.read_packet().await {
+                Ok(packet) => packet,
                 Err(err) => {
-                    // Error result set. No more data will follow.
-                    self.conn.set_pending_result(None);
+                    // Next row contained an error. No more data will follow.
+                    self.conn.set_pending_result(None)?;
                     return Err(err);
                 }
+            };
+
+            if P::is_last_result_set_packet(self.conn.capabilities(), &packet) {
+                // `packet` is a result set terminator.
+                self.conn.set_pending_result(None)?;
+            } else {
+                // `packet` is a result set row.
+                row = Some(P::read_result_set_row(&packet, columns)?);
+            }
+        }
+
+        Ok(row)
+    }
+
+    /// Low-level function that jumps to the next result set.
+    ///
+    /// Returns `false` if there are no more result sets.
+    async fn next_set(&mut self) -> crate::Result<bool> {
+        if self.conn.more_results_exists() {
+            // More data will follow.
+            self.conn.routine(NextSetRoutine::<P>::new()).await?;
+        }
+        Ok(self.conn.has_pending_result())
+    }
+
+    /// Low-level function that reads a next row and tries to jump
+    /// to the next result set if the current one is exhausted.
+    async fn next_row_or_next_set(&mut self, meta: ResultSetMeta) -> crate::Result<Option<Row>> {
+        let columns = meta.columns().clone();
+
+        self.next_row_or_next_set2(columns).await
+    }
+
+    /// Low-level function that reads a next row and tries to jump
+    /// to the next result set if the current one is exhausted.
+    async fn next_row_or_next_set2(
+        &mut self,
+        columns: Arc<[Column]>,
+    ) -> crate::Result<Option<Row>> {
+        if let Some(row) = self.next_row(columns).await? {
+            return Ok(Some(row));
+        } else {
+            self.next_set().await?;
+            return Ok(None);
+        }
+    }
+
+    /// Skips the taken result set.
+    async fn skip_taken(&mut self, meta: Arc<ResultSetMeta>) -> crate::Result<()> {
+        while let Some(_) = self.next_row_or_next_set((*meta).clone()).await? {}
+        Ok(())
+    }
+
+    #[doc(hidden)]
+    pub async fn next(&mut self) -> Result<Option<Row>> {
+        loop {
+            match self.conn.use_pending_result()?.cloned() {
+                Some(PendingResult::Pending(meta)) => return self.next_row_or_next_set(meta).await,
+                Some(PendingResult::Taken(meta)) => self.skip_taken(meta).await?,
+                None => return Ok(None),
             }
         }
     }
@@ -324,7 +367,7 @@ where
     pub async fn drop_result(mut self) -> Result<()> {
         loop {
             while self.next().await?.is_some() {}
-            if self.conn.get_pending_result().is_none() {
+            if !self.conn.has_pending_result() {
                 break Ok(());
             }
         }
@@ -336,7 +379,12 @@ where
     pub fn columns_ref(&self) -> &[Column] {
         self.conn
             .get_pending_result()
-            .and_then(|meta| meta.columns().map(|cols| &cols[..]).ok())
+            .ok()
+            .flatten()
+            .map(|meta| match meta {
+                PendingResult::Pending(meta) => &meta.columns()[..],
+                PendingResult::Taken(meta) => &meta.columns()[..],
+            })
             .unwrap_or_default()
     }
 
@@ -344,6 +392,12 @@ where
     pub fn columns(&self) -> Option<Arc<[Column]>> {
         self.conn
             .get_pending_result()
-            .and_then(|meta| meta.columns().map(|columns| columns.clone()).ok())
+            .ok()
+            .flatten()
+            .map(|meta| match meta {
+                PendingResult::Pending(meta) => meta.columns(),
+                PendingResult::Taken(meta) => meta.columns(),
+            })
+            .cloned()
     }
 }
diff --git a/src/queryable/query_result/result_set_stream.rs b/src/queryable/query_result/result_set_stream.rs
new file mode 100644
index 00000000..dbcb953a
--- /dev/null
+++ b/src/queryable/query_result/result_set_stream.rs
@@ -0,0 +1,348 @@
+// Copyright (c) 2021 mysql_async developers.
+//
+// Licensed under the Apache License, Version 2.0
+// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
+// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. All files in the project carrying such notice may not be copied,
+// modified, or distributed except according to those terms.
+
+use std::any::type_name;
+use std::borrow::Cow;
+use std::sync::Arc;
+use std::task::Poll;
+use std::{fmt, marker::PhantomData};
+
+use futures_core::FusedStream;
+use futures_core::{future::BoxFuture, Stream};
+use futures_util::FutureExt;
+use mysql_common::packets::{Column, OkPacket};
+
+use crate::{
+    conn::PendingResult,
+    prelude::{FromRow, Protocol},
+    QueryResult, Row,
+};
+
+enum CowMut<'r, 'a: 'r, 't: 'a, P> {
+    Borrowed(&'r mut QueryResult<'a, 't, P>),
+    Owned(QueryResult<'a, 't, P>),
+}
+
+impl<'r, 'a: 'r, 't: 'a, P> fmt::Debug for CowMut<'r, 'a, 't, P> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Borrowed(arg0) => f.debug_tuple("Borrowed").field(arg0).finish(),
+            Self::Owned(arg0) => f.debug_tuple("Owned").field(arg0).finish(),
+        }
+    }
+}
+
+impl<'r, 'a: 'r, 't: 'a, P> AsMut<QueryResult<'a, 't, P>> for CowMut<'r, 'a, 't, P> {
+    fn as_mut(&mut self) -> &mut QueryResult<'a, 't, P> {
+        match self {
+            CowMut::Borrowed(q) => q,
+            CowMut::Owned(q) => q,
+        }
+    }
+}
+
+enum ResultSetStreamState<'r, 'a: 'r, 't: 'a, P> {
+    Idle(CowMut<'r, 'a, 't, P>),
+    NextFut(BoxFuture<'r, (crate::Result<Option<Row>>, CowMut<'r, 'a, 't, P>)>),
+}
+
+impl<'r, 'a: 'r, 't: 'a, P> fmt::Debug for ResultSetStreamState<'r, 'a, 't, P> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Idle(arg0) => f.debug_tuple("Idle").field(arg0).finish(),
+            Self::NextFut(_arg0) => f
+                .debug_tuple("NextFut")
+                .field(&type_name::<
+                    BoxFuture<'r, (crate::Result<Option<Row>>, CowMut<'r, 'a, 't, P>)>,
+                >())
+                .finish(),
+        }
+    }
+}
+
+#[derive(Debug)]
+/// Rows stream for a single result set.
+pub struct ResultSetStream<'r, 'a: 'r, 't: 'a, T, P> {
+    query_result: Option<ResultSetStreamState<'r, 'a, 't, P>>,
+    ok_packet: Option<OkPacket<'static>>,
+    columns: Arc<[Column]>,
+    __from_row_type: PhantomData<T>,
+}
+
+impl<'r, 'a: 'r, 't: 'a, T, P> FusedStream for ResultSetStream<'r, 'a, 't, T, P>
+where
+    P: Protocol + Unpin,
+    T: FromRow + Unpin + Send + 'static,
+{
+    fn is_terminated(&self) -> bool {
+        self.query_result.is_none()
+    }
+}
+
+impl<'r, 'a: 'r, 't: 'a, T, P> ResultSetStream<'r, 'a, 't, T, P> {
+    /// See [`Conn::last_insert_id`][1].
+    ///
+    /// [1]: crate::Conn::last_insert_id
+    pub fn last_insert_id(&self) -> Option<u64> {
+        self.ok_packet.as_ref().and_then(|ok| ok.last_insert_id())
+    }
+
+    /// See [`Conn::affected_rows`][1].
+    ///
+    /// [1]: crate::Conn::affected_rows
+    pub fn affected_rows(&self) -> u64 {
+        self.ok_packet
+            .as_ref()
+            .map(|ok| ok.affected_rows())
+            .unwrap_or_default()
+    }
+
+    /// See [`Conn::info`][1].
+    ///
+    /// [1]: crate::Conn::info
+    pub fn info(&self) -> Cow<'_, str> {
+        self.ok_packet
+            .as_ref()
+            .and_then(|ok| ok.info_str())
+            .unwrap_or_default()
+    }
+
+    /// See [`Conn::get_warnings`][1].
+    ///
+    /// [1]: crate::Conn::get_warnings
+    pub fn get_warnings(&self) -> u16 {
+        self.ok_packet
+            .as_ref()
+            .map(|ok| ok.warnings())
+            .unwrap_or_default()
+    }
+
+    /// Returns an `OkPacket` corresponding to this result set.
+    ///
+    /// Will be `None` if there is no OK packet (result set contains an error).
+    pub fn ok_packet(&self) -> Option<&OkPacket<'static>> {
+        self.ok_packet.as_ref()
+    }
+}
+
+impl<'r, 'a: 'r, 't: 'a, T, P> Stream for ResultSetStream<'r, 'a, 't, T, P>
+where
+    P: Protocol + Unpin,
+    T: FromRow + Unpin + Send + 'static,
+{
+    type Item = crate::Result<T>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        let this = self.get_mut();
+        loop {
+            let columns = this.columns.clone();
+            match this.query_result.take() {
+                Some(ResultSetStreamState::Idle(mut query_result)) => {
+                    let fut = Box::pin(async move {
+                        let row = query_result.as_mut().next_row_or_next_set2(columns).await;
+                        (row, query_result)
+                    });
+                    this.query_result = Some(ResultSetStreamState::NextFut(fut));
+                }
+                Some(ResultSetStreamState::NextFut(mut fut)) => match fut.poll_unpin(cx) {
+                    Poll::Ready((row, query_result)) => match row {
+                        Ok(Some(row)) => {
+                            this.query_result = Some(ResultSetStreamState::Idle(query_result));
+                            return Poll::Ready(Some(Ok(crate::from_row(row))));
+                        }
+                        Ok(None) => return Poll::Ready(None),
+                        Err(err) => return Poll::Ready(Some(Err(err))),
+                    },
+                    Poll::Pending => {
+                        this.query_result = Some(ResultSetStreamState::NextFut(fut));
+                        return Poll::Pending;
+                    }
+                },
+                None => return Poll::Ready(None),
+            }
+        }
+    }
+}
+
+impl<'a, 't: 'a, P> QueryResult<'a, 't, P>
+where
+    P: Protocol + Unpin,
+{
+    async fn setup_stream(
+        &mut self,
+    ) -> crate::Result<Option<(Option<OkPacket<'static>>, Arc<[Column]>)>> {
+        match self.conn.use_pending_result()? {
+            Some(PendingResult::Taken(meta)) => {
+                let meta = (*meta).clone();
+                self.skip_taken(meta).await?;
+            }
+            Some(_) => (),
+            None => return Ok(None),
+        }
+
+        let ok_packet = self.conn.last_ok_packet().cloned();
+        let columns = match self.conn.take_pending_result()? {
+            Some(meta) => meta.columns().clone(),
+            None => return Ok(None),
+        };
+
+        Ok(Some((ok_packet, columns)))
+    }
+
+    /// Returns a [`Stream`] for the current result set.
+    ///
+    /// The returned stream satisfies [`futures_util::TryStream`],
+    /// so you can use [`futures_util::TryStreamExt`] functions on it.
+    ///
+    /// # Behavior
+    ///
+    /// ## Conversion
+    ///
+    /// This stream will convert each row into `T` using [`FromRow`] implementation.
+    /// Please use the [`Row`] type for `T` to make this conversion infallible.
+    ///
+    /// ## Consumption
+    ///
+    /// The call to [`QueryResult::stream`] entails the consumption of the current result set,
+    /// practicly this means that the second call to [`QueryResult::stream`] will return
+    /// the next result set stream even if the stream returned from the first call wasn't
+    /// explicitly consumed:
+    ///
+    /// ```rust
+    /// # use mysql_async::test_misc::get_opts;
+    /// # #[tokio::main]
+    /// # async fn main() -> mysql_async::Result<()> {
+    /// # use mysql_async::*;
+    /// # use mysql_async::prelude::*;
+    /// # use futures_util::StreamExt;
+    /// let mut conn = Conn::new(get_opts()).await?;
+    ///
+    /// // This query result will contain two result sets.
+    /// let mut result = conn.query_iter("SELECT 1; SELECT 2;").await?;
+    ///
+    /// // The first result set stream is dropped here without being consumed,
+    /// let _ = result.stream::<u8>().await?;
+    /// // so it will be implicitly consumed here.
+    /// let mut stream = result.stream::<u8>().await?.expect("the second result set must be here");
+    /// assert_eq!(2_u8, stream.next().await.unwrap()?);
+    ///
+    /// # drop(stream); drop(result); conn.disconnect().await }
+    /// ```
+    ///
+    /// ## Errors
+    ///
+    /// Note, that [`QueryResult::stream`] may error if:
+    ///
+    /// - current result set contains an error,
+    /// - previously unconsumed result set stream contained an error.
+    ///
+    /// ```rust
+    /// # use mysql_async::test_misc::get_opts;
+    /// # #[tokio::main]
+    /// # async fn main() -> mysql_async::Result<()> {
+    /// # use mysql_async::*;
+    /// # use mysql_async::prelude::*;
+    /// # use futures_util::StreamExt;
+    /// let mut conn = Conn::new(get_opts()).await?;
+    ///
+    /// // The second result set of this query will contain an error.
+    /// let mut result = conn.query_iter("SELECT 1; SELECT FOO(); SELECT 2;").await?;
+    ///
+    /// // First result set stream is dropped here without being consumed,
+    /// let _ = result.stream::<Row>().await?;
+    /// // so it will be implicitly consumed on the second call to `QueryResult::stream`
+    /// // that will error complaining about unknown FOO
+    /// assert!(result.stream::<Row>().await.unwrap_err().to_string().contains("FOO"));
+    ///
+    /// # drop(result); conn.disconnect().await }
+    /// ```
+    pub fn stream<'r, T: Unpin + FromRow + Send + 'static>(
+        &'r mut self,
+    ) -> BoxFuture<'r, crate::Result<Option<ResultSetStream<'r, 'a, 't, T, P>>>> {
+        async move {
+            Ok(self
+                .setup_stream()
+                .await?
+                .map(
+                    move |(ok_packet, columns)| ResultSetStream::<'r, 'a, 't, T, P> {
+                        ok_packet,
+                        columns,
+                        query_result: Some(ResultSetStreamState::Idle(CowMut::Borrowed(self))),
+                        __from_row_type: PhantomData,
+                    },
+                ))
+        }
+        .boxed()
+    }
+
+    /// Owned version of the [`QueryResult::stream`].
+    ///
+    /// Returned stream will stop iteration on the first result set boundary.
+    ///
+    /// See also [`Query::stream`][query_stream], [`Queryable::query_stream`][queryable_query_stream],
+    /// [`Queryable::exec_stream`][queryable_exec_stream].
+    ///
+    /// The following example uses the [`Query::stream`][query_stream] function
+    /// that is based on the [`QueryResult::stream_and_drop`]:
+    ///
+    /// ```rust
+    /// # use mysql_async::test_misc::get_opts;
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    /// # use mysql_async::*;
+    /// # use mysql_async::prelude::*;
+    /// # use futures_util::TryStreamExt;
+    /// let pool = Pool::new(get_opts());
+    /// let mut conn = pool.get_conn().await?;
+    ///
+    /// // This example uses the `Query::stream` function that is based on `QueryResult::stream_and_drop`:
+    /// let mut stream = "SELECT 1 UNION ALL SELECT 2".stream::<u8, _>(&mut conn).await?;
+    /// let rows = stream.try_collect::<Vec<_>>().await.unwrap();
+    /// assert_eq!(vec![1, 2], rows);
+    ///
+    /// // Only the first result set will go into the stream:
+    /// let mut stream = r"
+    ///     SELECT 'foo' UNION ALL SELECT 'bar';
+    ///     SELECT 'baz' UNION ALL SELECT 'quux';".stream::<String, _>(&mut conn).await?;
+    /// let rows = stream.try_collect::<Vec<_>>().await.unwrap();
+    /// assert_eq!(vec!["foo".to_owned(), "bar".to_owned()], rows);
+    ///
+    /// // We can also build a `'static` stream by giving away the connection:
+    /// let stream = "SELECT 2 UNION ALL SELECT 3".stream::<u8, _>(conn).await?;
+    /// // `tokio::spawn` requires `'static`
+    /// let handle = tokio::spawn(async move {
+    ///     stream.try_collect::<Vec<_>>().await.unwrap()
+    /// });
+    /// assert_eq!(vec![2, 3], handle.await?);
+    ///
+    /// # Ok(()) }
+    /// ```
+    ///
+    /// [queryable_query_stream]: crate::prelude::Queryable::query_stream
+    /// [queryable_exec_stream]: crate::prelude::Queryable::exec_stream
+    /// [query_stream]: crate::prelude::Query::stream
+    pub fn stream_and_drop<T: Unpin + FromRow + Send + 'static>(
+        mut self,
+    ) -> BoxFuture<'a, crate::Result<Option<ResultSetStream<'a, 'a, 't, T, P>>>> {
+        async move {
+            Ok(self
+                .setup_stream()
+                .await?
+                .map(|(ok_packet, columns)| ResultSetStream::<'a, 'a, 't, T, P> {
+                    ok_packet,
+                    columns,
+                    query_result: Some(ResultSetStreamState::Idle(CowMut::Owned(self))),
+                    __from_row_type: PhantomData,
+                }))
+        }
+        .boxed()
+    }
+}


Test patch:
diff --git a/src/queryable/query_result/tests.rs b/src/queryable/query_result/tests.rs
new file mode 100644
index 00000000..24c870cf
--- /dev/null
+++ b/src/queryable/query_result/tests.rs
@@ -0,0 +1,274 @@
+#![cfg(test)]
+
+use futures_util::TryStreamExt;
+
+use crate::{from_row, prelude::*, test_misc::get_opts, Conn, Row, TxOpts};
+
+#[tokio::test]
+async fn should_stream_text_result_sets() -> crate::Result<()> {
+    const QUERY: &str = r"
+        SELECT 1;
+        SELECT 'foo' UNION ALL SELECT 'bar';
+        SELECT 3, 4 UNION ALL SELECT 5, 6;
+        SELECT FOO();
+        SELECT 3.14;
+    ";
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = QUERY.run(&mut conn).await?;
+
+    let stream = result.stream::<u8>().await?.unwrap();
+    assert_eq!(vec![1], stream.try_collect::<Vec<_>>().await?);
+
+    let _ = result.stream::<Row>().await?.unwrap();
+
+    let stream = result.stream::<(u8, u8)>().await?.unwrap();
+    assert_eq!(vec![(3, 4), (5, 6)], stream.try_collect::<Vec<_>>().await?);
+
+    let _ = result.stream::<Row>().await.unwrap_err();
+    assert!(result.stream::<Row>().await?.is_none());
+
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = QUERY.run(&mut conn).await?;
+
+    assert_eq!(vec![1], result.collect::<u8>().await?);
+
+    assert_eq!("foo", from_row::<String>(result.next().await?.unwrap()));
+    let _ = result.stream::<Row>().await?.unwrap();
+
+    assert_eq!((3_u8, 4_u8), from_row(result.next().await?.unwrap()));
+    assert_eq!(vec![(5, 6)], result.collect::<(u8, u8)>().await?);
+
+    result.collect::<Row>().await.unwrap_err();
+    assert!(result.stream::<Row>().await?.is_none());
+
+    conn.disconnect().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_stream_binary_result_sets() -> crate::Result<()> {
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = "SELECT ?".with((1_u8,)).run(&mut conn).await?;
+    assert_eq!(vec![1], result.collect::<u8>().await?);
+    assert!(result.stream::<Row>().await?.is_none());
+
+    let mut result = "SELECT ?".with((1_u8,)).run(&mut conn).await?;
+    assert_eq!(
+        vec![1],
+        result
+            .stream::<u8>()
+            .await?
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await?,
+    );
+    assert_eq!(result.collect::<u8>().await?, Vec::<u8>::new());
+
+    conn.disconnect().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn dropped_query_result_should_emit_errors_on_cleanup() -> super::Result<()> {
+    use crate::{Error::Server, ServerError};
+    let mut conn = Conn::new(get_opts()).await?;
+    conn.query_iter("SELECT '1'; BLABLA;").await?;
+    assert!(matches!(
+        conn.query_drop("DO 42;").await.unwrap_err(),
+        Server(ServerError { code: 1064, .. })
+    ));
+    conn.disconnect().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_try_collect() -> super::Result<()> {
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = conn
+        .query_iter(
+            r"SELECT 'hello', 123
+                UNION ALL
+                SELECT 'world', 'bar'
+                UNION ALL
+                SELECT 'hello', 123
+            ",
+        )
+        .await?;
+    let mut rows = result.try_collect::<(String, u8)>().await?;
+    assert!(rows.pop().unwrap().is_ok());
+    assert!(rows.pop().unwrap().is_err());
+    assert!(rows.pop().unwrap().is_ok());
+    result.drop_result().await?;
+    conn.disconnect().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_try_collect_and_drop() -> super::Result<()> {
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut rows = conn
+        .query_iter(
+            r"SELECT 'hello', 123
+                UNION ALL
+                SELECT 'world', 'bar'
+                UNION ALL
+                SELECT 'hello', 123;
+                SELECT 'foo', 255;
+            ",
+        )
+        .await?
+        .try_collect_and_drop::<(String, u8)>()
+        .await?;
+    assert!(rows.pop().unwrap().is_ok());
+    assert!(rows.pop().unwrap().is_err());
+    assert!(rows.pop().unwrap().is_ok());
+    conn.disconnect().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_handle_mutliresult_set() -> super::Result<()> {
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = conn
+        .query_iter(
+            r"SELECT 'hello', 123
+                UNION ALL
+                SELECT 'world', 231;
+                SELECT 'foo', 255;
+            ",
+        )
+        .await?;
+    let rows_1 = result.collect::<(String, u8)>().await?;
+    let rows_2 = result.collect_and_drop().await?;
+    conn.disconnect().await?;
+
+    assert_eq!((String::from("hello"), 123), rows_1[0]);
+    assert_eq!((String::from("world"), 231), rows_1[1]);
+    assert_eq!((String::from("foo"), 255), rows_2[0]);
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_map_resultset() -> super::Result<()> {
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = conn
+        .query_iter(
+            r"
+                SELECT 'hello', 123
+                UNION ALL
+                SELECT 'world', 231;
+                SELECT 'foo', 255;
+            ",
+        )
+        .await?;
+
+    let rows_1 = result.map(|row| from_row::<(String, u8)>(row)).await?;
+    let rows_2 = result.map_and_drop(from_row).await?;
+    conn.disconnect().await?;
+
+    assert_eq!((String::from("hello"), 123), rows_1[0]);
+    assert_eq!((String::from("world"), 231), rows_1[1]);
+    assert_eq!((String::from("foo"), 255), rows_2[0]);
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_reduce_resultset() -> super::Result<()> {
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = conn
+        .query_iter(
+            r"SELECT 5
+                UNION ALL
+                SELECT 6;
+                SELECT 7;",
+        )
+        .await?;
+    let reduced = result
+        .reduce(0, |mut acc, row| {
+            acc += from_row::<i32>(row);
+            acc
+        })
+        .await?;
+    let rows_2 = result.collect_and_drop::<i32>().await?;
+    conn.disconnect().await?;
+    assert_eq!(11, reduced);
+    assert_eq!(7, rows_2[0]);
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_handle_multi_result_sets_where_some_results_have_no_output() -> super::Result<()> {
+    const QUERY: &str = r"SELECT 1;
+        UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
+        SELECT 2;
+        SELECT 3;
+        UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
+        UPDATE time_zone SET Time_zone_id = 1 WHERE Time_zone_id = 1;
+        SELECT 4;";
+
+    let mut c = Conn::new(get_opts()).await?;
+    c.query_drop("CREATE TEMPORARY TABLE time_zone (Time_zone_id INT)")
+        .await
+        .unwrap();
+    let mut t = c.start_transaction(TxOpts::new()).await?;
+    t.query_drop(QUERY).await?;
+    let r = t.query_iter(QUERY).await?;
+    let out = r.collect_and_drop::<u8>().await?;
+    assert_eq!(vec![1], out);
+    let r = t.query_iter(QUERY).await?;
+    r.for_each_and_drop(|x| assert_eq!(from_row::<u8>(x), 1))
+        .await?;
+    let r = t.query_iter(QUERY).await?;
+    let out = r.map_and_drop(|row| from_row::<u8>(row)).await?;
+    assert_eq!(vec![1], out);
+    let r = t.query_iter(QUERY).await?;
+    let out = r
+        .reduce_and_drop(0u8, |acc, x| acc + from_row::<u8>(x))
+        .await?;
+    assert_eq!(1, out);
+    t.query_drop(QUERY).await?;
+    t.commit().await?;
+    let result = c.exec_first("SELECT 1", ()).await?;
+    c.disconnect().await?;
+    assert_eq!(result, Some(1_u8));
+    Ok(())
+}
+
+#[tokio::test]
+async fn should_iterate_over_resultset() -> super::Result<()> {
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    };
+
+    let acc = Arc::new(AtomicUsize::new(0));
+
+    let mut conn = Conn::new(get_opts()).await?;
+    let mut result = conn
+        .query_iter(
+            r"SELECT 2
+                UNION ALL
+                SELECT 3;
+                SELECT 5;",
+        )
+        .await?;
+    result
+        .for_each({
+            let acc = acc.clone();
+            move |row| {
+                acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
+            }
+        })
+        .await?;
+    result
+        .for_each_and_drop({
+            let acc = acc.clone();
+            move |row| {
+                acc.fetch_add(from_row::<usize>(row), Ordering::SeqCst);
+            }
+        })
+        .await?;
+    conn.disconnect().await?;
+    assert_eq!(acc.load(Ordering::SeqCst), 10);
+    Ok(())
+}