Update mysql_common to 0.21.0

Author: blackbeam

Cargo.toml

@@ -19,7 +19,7 @@ futures-util = "0.3"
 futures-sink = "0.3"
 lazy_static = "1"
 mio-named-pipes = "0.1.6"
-mysql_common = "0.20.0"
+mysql_common = "0.21.0"
 native-tls = "0.2"
 percent-encoding = "2.1.0"
 pin-project = "0.4.6"

src/conn/mod.rs

@@ -68,6 +68,13 @@ fn disconnect(mut conn: Conn) {
     }
 }
 
+#[derive(Debug)]
+pub enum PendingResult {
+    Text(Arc<Vec<Column>>),
+    Binary(Arc<Vec<Column>>, StmtCacheResult),
+    Empty,
+}
+
 /// Mysql connection
 struct ConnInner {
     stream: Option<Stream>,
@@ -79,7 +86,7 @@ struct ConnInner {
     status: consts::StatusFlags,
     last_ok_packet: Option<OkPacket<'static>>,
     pool: Option<Pool>,
-    has_result: Option<(Arc<Vec<Column>>, Option<StmtCacheResult>)>,
+    has_result: Option<PendingResult>,
     in_transaction: bool,
     opts: Opts,
     last_io: Instant,
@@ -521,13 +528,18 @@ impl Conn {
 
     async fn drop_result(mut self) -> Result<Conn> {
         match self.inner.has_result.take() {
-            Some((columns, None)) => {
+            Some(PendingResult::Text(columns)) => {
                 query_result::assemble::<_, TextProtocol>(self, Some(columns), None)
                     .drop_result()
                     .await
             }
-            Some((columns, cached)) => {
-                query_result::assemble::<_, BinaryProtocol>(self, Some(columns), cached)
+            Some(PendingResult::Binary(columns, cached)) => {
+                query_result::assemble::<_, BinaryProtocol>(self, Some(columns), Some(cached))
+                    .drop_result()
+                    .await
+            }
+            Some(PendingResult::Empty) => {
+                query_result::assemble::<_, TextProtocol>(self, None, None)
                     .drop_result()
                     .await
             }
@@ -619,7 +631,7 @@ impl ConnectionLike for Conn {
         &self.inner.opts
     }
 
-    fn get_pending_result(&self) -> Option<&(Arc<Vec<Column>>, Option<StmtCacheResult>)> {
+    fn get_pending_result(&self) -> Option<&PendingResult> {
         self.inner.has_result.as_ref()
     }
 
@@ -639,7 +651,7 @@ impl ConnectionLike for Conn {
         self.inner.in_transaction = in_transaction;
     }
 
-    fn set_pending_result(&mut self, meta: Option<(Arc<Vec<Column>>, Option<StmtCacheResult>)>) {
+    fn set_pending_result(&mut self, meta: Option<PendingResult>) {
         self.inner.has_result = meta;
     }
 
@@ -1194,6 +1206,52 @@ mod test {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn issue_107() -> super::Result<()> {
+        let conn = Conn::new(get_opts()).await?;
+        let conn = conn
+            .drop_query(
+                r"CREATE TEMPORARY TABLE mysql.issue (
+                    a BIGINT(20) UNSIGNED,
+                    b VARBINARY(16),
+                    c BINARY(32),
+                    d BIGINT(20) UNSIGNED,
+                    e BINARY(32)
+                )",
+            )
+            .await?;
+        let conn = conn
+            .drop_query(
+                r"INSERT INTO mysql.issue VALUES (
+                    0,
+                    0xC066F966B0860000,
+                    0x7939DA98E524C5F969FC2DE8D905FD9501EBC6F20001B0A9C941E0BE6D50CF44,
+                    0,
+                    ''
+                ), (
+                    1,
+                    '',
+                    0x076311DF4D407B0854371BA13A5F3FB1A4555AC22B361375FD47B263F31822F2,
+                    0,
+                    ''
+                )",
+            )
+            .await?;
+
+        let q = "SELECT b, c, d, e FROM mysql.issue";
+        let result = conn.query(q).await?;
+
+        let (conn, loaded_structs) = result
+            .map_and_drop(|row| crate::from_row::<(Vec<u8>, Vec<u8>, u64, Vec<u8>)>(dbg!(row)))
+            .await?;
+
+        conn.disconnect().await?;
+
+        assert_eq!(loaded_structs.len(), 2);
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn should_run_transactions() -> super::Result<()> {
         let conn = Conn::new(get_opts()).await?;

src/connection_like/mod.rs

@@ -6,6 +6,7 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
+use crate::conn::PendingResult;
 use futures_util::future::ok;
 use mysql_common::{
     io::ReadMysqlExt,
@@ -51,7 +52,7 @@ pub mod streamless {
 }
 pub mod write_packet;
 
-#[derive(Debug)]
+#[derive(Debug, Clone, Copy)]
 pub enum StmtCacheResult {
     Cached,
     NotCached(u32),
@@ -130,7 +131,7 @@ where
         self.conn_like_ref().get_opts()
     }
 
-    fn get_pending_result(&self) -> Option<&(Arc<Vec<Column>>, Option<StmtCacheResult>)> {
+    fn get_pending_result(&self) -> Option<&PendingResult> {
         self.conn_like_ref().get_pending_result()
     }
 
@@ -150,7 +151,7 @@ where
         self.conn_like_mut().set_in_transaction(in_transaction);
     }
 
-    fn set_pending_result(&mut self, meta: Option<(Arc<Vec<Column>>, Option<StmtCacheResult>)>) {
+    fn set_pending_result(&mut self, meta: Option<PendingResult>) {
         self.conn_like_mut().set_pending_result(meta);
     }
 
@@ -191,12 +192,12 @@ pub trait ConnectionLike: Send {
     fn get_local_infile_handler(&self) -> Option<Arc<dyn LocalInfileHandler>>;
     fn get_max_allowed_packet(&self) -> usize;
     fn get_opts(&self) -> &Opts;
-    fn get_pending_result(&self) -> Option<&(Arc<Vec<Column>>, Option<StmtCacheResult>)>;
+    fn get_pending_result(&self) -> Option<&PendingResult>;
     fn get_server_version(&self) -> (u16, u16, u16);
     fn get_status(&self) -> StatusFlags;
     fn set_last_ok_packet(&mut self, ok_packet: Option<OkPacket<'static>>);
     fn set_in_transaction(&mut self, in_transaction: bool);
-    fn set_pending_result(&mut self, meta: Option<(Arc<Vec<Column>>, Option<StmtCacheResult>)>);
+    fn set_pending_result(&mut self, meta: Option<PendingResult>);
     fn set_status(&mut self, status: StatusFlags);
     fn reset_seq_id(&mut self);
     fn sync_seq_id(&mut self);
@@ -435,7 +436,17 @@ where
         this = this.read_packet().await?.0;
     }
 
-    let columns = Arc::new(columns);
-    this.set_pending_result(Some((Clone::clone(&columns), None)));
-    Ok(query_result::new(this, Some(columns), cached))
+    if column_count > 0 {
+        let columns = Arc::new(columns);
+        match cached {
+            Some(cached) => {
+                this.set_pending_result(Some(PendingResult::Binary(columns.clone(), cached)))
+            }
+            None => this.set_pending_result(Some(PendingResult::Text(columns.clone()))),
+        }
+        Ok(query_result::new(this, Some(columns), cached))
+    } else {
+        this.set_pending_result(Some(PendingResult::Empty));
+        Ok(query_result::new(this, None, cached))
+    }
 }

src/connection_like/read_packet.rs

@@ -8,7 +8,7 @@
 
 use futures_core::ready;
 use futures_util::stream::{StreamExt, StreamFuture};
-use mysql_common::packets::{parse_err_packet, parse_ok_packet};
+use mysql_common::packets::{parse_err_packet, parse_ok_packet, OkPacketKind};
 use pin_project::pin_project;
 use std::{
     future::Future,
@@ -50,7 +50,13 @@ impl<T: ConnectionLike> Future for ReadPacket<T> {
         let mut conn_like = this.conn_like.take().unwrap().return_stream(stream);
         match packet_opt {
             Some(packet) => {
-                if let Ok(ok_packet) = parse_ok_packet(&*packet, conn_like.get_capabilities()) {
+                let kind = if conn_like.get_pending_result().is_some() {
+                    OkPacketKind::ResultSetTerminator
+                } else {
+                    OkPacketKind::Other
+                };
+                if let Ok(ok_packet) = parse_ok_packet(&*packet, conn_like.get_capabilities(), kind)
+                {
                     conn_like.set_status(ok_packet.status_flags());
                     conn_like.set_last_ok_packet(Some(ok_packet.into_owned()));
                 } else if let Ok(err_packet) =

src/queryable/mod.rs

@@ -7,7 +7,7 @@
 // modified, or distributed except according to those terms.
 
 use mysql_common::{
-    packets::parse_ok_packet,
+    packets::{parse_ok_packet, OkPacketKind},
     row::new_row,
     value::{read_bin_values, read_text_values},
 };
@@ -35,7 +35,12 @@ pub trait Protocol: Send + 'static {
     where
         T: ConnectionLike,
     {
-        parse_ok_packet(packet, conn_like.get_capabilities()).is_ok()
+        parse_ok_packet(
+            packet,
+            conn_like.get_capabilities(),
+            OkPacketKind::ResultSetTerminator,
+        )
+        .is_ok()
     }
 }
 
@@ -52,20 +57,13 @@ impl Protocol for TextProtocol {
             .map_err(Into::into)
     }
 }
+
 impl Protocol for BinaryProtocol {
     fn read_result_set_row(packet: &[u8], columns: Arc<Vec<Column>>) -> Result<Row> {
         read_bin_values::<ServerSide>(packet, &*columns)
             .map(|values| new_row(values, columns))
             .map_err(Into::into)
     }
-
-    fn is_last_result_set_packet<T>(conn_like: &T, packet: &[u8]) -> bool
-    where
-        T: ConnectionLike,
-    {
-        parse_ok_packet(packet, conn_like.get_capabilities()).is_ok()
-            && packet.get(0).cloned() == Some(0xFE)
-    }
 }
 
 /// Represents something queryable like connection or transaction.