Bump mysql_common

Cargo.toml

@@ -23,7 +23,7 @@ keyed_priority_queue = "0.4"
 lazy_static = "1"
 lru = "0.11.0"
 mio = { version = "0.8.0", features = ["os-poll", "net"] }
-mysql_common = { version = "0.30", default-features = false }
+mysql_common = { version = "0.31", default-features = false }
 once_cell = "1.7.2"
 pem = "3.0"
 percent-encoding = "2.1.0"
@@ -109,7 +109,7 @@ rustls-tls = [
 tracing = ["dep:tracing"]
 derive = ["mysql_common/derive"]
 nightly = []
-binlog = []
+binlog = ["mysql_common/binlog"]
 
 [lib]
 name = "mysql_async"

src/conn/binlog_stream/mod.rs

@@ -9,8 +9,8 @@
 use futures_core::ready;
 use mysql_common::{
     binlog::{
-        consts::BinlogVersion::Version4,
-        events::{Event, TableMapEvent},
+        consts::{BinlogVersion::Version4, EventType},
+        events::{Event, TableMapEvent, TransactionPayloadEvent},
         EventStreamReader,
     },
     io::ParseBuf,
@@ -19,7 +19,7 @@ use mysql_common::{
 
 use std::{
     future::Future,
-    io::ErrorKind,
+    io::{Cursor, ErrorKind},
     pin::Pin,
     task::{Context, Poll},
 };
@@ -71,6 +71,9 @@ impl super::Conn {
 pub struct BinlogStream {
     read_packet: ReadPacket<'static, 'static>,
     esr: EventStreamReader,
+    // TODO: Use 'static reader here (requires impl on the mysql_common side).
+    /// Uncompressed Transaction_payload_event we are iterating over (if any).
+    tpe: Option<Cursor<Vec<u8>>>,
 }
 
 impl BinlogStream {
@@ -79,6 +82,7 @@ impl BinlogStream {
         BinlogStream {
             read_packet: ReadPacket::new(conn),
             esr: EventStreamReader::new(Version4),
+            tpe: None,
         }
     }
 
@@ -114,6 +118,22 @@ impl futures_core::stream::Stream for BinlogStream {
     type Item = Result<Event>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        {
+            let Self {
+                ref mut tpe,
+                ref mut esr,
+                ..
+            } = *self;
+
+            if let Some(tpe) = tpe.as_mut() {
+                match esr.read_decompressed(tpe) {
+                    Ok(Some(event)) => return Poll::Ready(Some(Ok(event))),
+                    Ok(None) => self.tpe = None,
+                    Err(err) => return Poll::Ready(Some(Err(err.into()))),
+                }
+            }
+        }
+
         let packet = match ready!(Pin::new(&mut self.read_packet).poll(cx)) {
             Ok(packet) => packet,
             Err(err) => return Poll::Ready(Some(Err(err.into()))),
@@ -143,9 +163,17 @@ impl futures_core::stream::Stream for BinlogStream {
         if first_byte == Some(0) {
             let event_data = &packet[1..];
             match self.esr.read(event_data) {
-                Ok(event) => {
+                Ok(Some(event)) => {
+                    if event.header().event_type_raw() == EventType::TRANSACTION_PAYLOAD_EVENT as u8
+                    {
+                        match event.read_event::<TransactionPayloadEvent<'_>>() {
+                            Ok(e) => self.tpe = Some(Cursor::new(e.danger_decompress())),
+                            Err(_) => (/* TODO: Log the error */),
+                        }
+                    }
                     return Poll::Ready(Some(Ok(event)));
                 }
+                Ok(None) => return Poll::Ready(None),
                 Err(err) => return Poll::Ready(Some(Err(err.into()))),
             }
         } else {
@@ -168,21 +196,21 @@ mod tests {
     use crate::prelude::*;
     use crate::{test_misc::get_opts, *};
 
-    async fn gen_dummy_data() -> super::Result<()> {
-        let mut conn = Conn::new(get_opts()).await?;
-
+    async fn gen_dummy_data(conn: &mut Conn) -> super::Result<()> {
         "CREATE TABLE IF NOT EXISTS customers (customer_id int not null)"
-            .ignore(&mut conn)
+            .ignore(&mut *conn)
             .await?;
 
+        let mut tx = conn.start_transaction(Default::default()).await?;
         for i in 0_u8..100 {
             "INSERT INTO customers(customer_id) VALUES (?)"
                 .with((i,))
-                .ignore(&mut conn)
+                .ignore(&mut tx)
                 .await?;
         }
+        tx.commit().await?;
 
-        "DROP TABLE customers".ignore(&mut conn).await?;
+        "DROP TABLE customers".ignore(conn).await?;
 
         Ok(())
     }
@@ -193,6 +221,12 @@ mod tests {
             Some(pool) => pool.get_conn().await.unwrap(),
         };
 
+        if conn.server_version() >= (8, 0, 31) && conn.server_version() < (9, 0, 0) {
+            let _ = "SET binlog_transaction_compression=ON"
+                .ignore(&mut conn)
+                .await;
+        }
+
         if let Ok(Some(gtid_mode)) = "SELECT @@GLOBAL.GTID_MODE"
             .first::<String, _>(&mut conn)
             .await
@@ -209,7 +243,7 @@ mod tests {
         let filename = row.get(0).unwrap();
         let position = row.get(1).unwrap();
 
-        gen_dummy_data().await.unwrap();
+        gen_dummy_data(&mut conn).await.unwrap();
         Ok((conn, filename, position))
     }
 

src/conn/routines/exec.rs

@@ -71,14 +71,13 @@ impl Routine<()> for ExecRoutine<'_> {
                         break;
                     }
                     Params::Named(_) => {
-                        if self.stmt.named_params.is_none() {
+                        if self.stmt.named_params.is_empty() {
                             let error = DriverError::NamedParamsForPositionalQuery.into();
                             return Err(error);
                         }
 
                         let named = mem::replace(&mut self.params, Params::Empty);
-                        self.params =
-                            named.into_positional(self.stmt.named_params.as_ref().unwrap())?;
+                        self.params = named.into_positional(&self.stmt.named_params)?;
 
                         continue;
                     }

src/io/read_packet.rs

@@ -15,7 +15,7 @@ use std::{
     task::{Context, Poll},
 };
 
-use crate::{buffer_pool::PooledBuf, connection_like::Connection, error::IoError, Conn};
+use crate::{buffer_pool::PooledBuf, connection_like::Connection, error::IoError};
 
 /// Reads a packet.
 #[derive(Debug)]
@@ -27,7 +27,8 @@ impl<'a, 't> ReadPacket<'a, 't> {
         Self(conn.into())
     }
 
-    pub(crate) fn conn_ref(&self) -> &Conn {
+    #[cfg(feature = "binlog")]
+    pub(crate) fn conn_ref(&self) -> &crate::Conn {
         &*self.0
     }
 }

src/queryable/stmt.rs

@@ -9,7 +9,7 @@
 use futures_util::FutureExt;
 use mysql_common::{
     io::ParseBuf,
-    named_params::parse_named_params,
+    named_params::ParsedNamedParams,
     packets::{ComStmtClose, StmtPacket},
 };
 
@@ -45,12 +45,22 @@ fn to_statement_move<'a, T: AsQuery + 'a>(
 ) -> ToStatementResult<'a> {
     let fut = async move {
         let query = stmt.as_query();
-        let (named_params, raw_query) = parse_named_params(query.as_ref())?;
-        let inner_stmt = match conn.get_cached_stmt(&*raw_query) {
+        let parsed = ParsedNamedParams::parse(query.as_ref())?;
+        let inner_stmt = match conn.get_cached_stmt(parsed.query()) {
             Some(inner_stmt) => inner_stmt,
-            None => conn.prepare_statement(raw_query).await?,
+            None => {
+                conn.prepare_statement(Cow::Borrowed(parsed.query()))
+                    .await?
+            }
         };
-        Ok(Statement::new(inner_stmt, named_params))
+        Ok(Statement::new(
+            inner_stmt,
+            parsed
+                .params()
+                .iter()
+                .map(|x| x.as_ref().to_vec())
+                .collect::<Vec<_>>(),
+        ))
     }
     .boxed();
     ToStatementResult::Mediate(fut)
@@ -240,11 +250,12 @@ impl StmtInner {
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct Statement {
     pub(crate) inner: Arc<StmtInner>,
-    pub(crate) named_params: Option<Vec<Vec<u8>>>,
+    /// An empty vector in case of no named params.
+    pub(crate) named_params: Vec<Vec<u8>>,
 }
 
 impl Statement {
-    pub(crate) fn new(inner: Arc<StmtInner>, named_params: Option<Vec<Vec<u8>>>) -> Self {
+    pub(crate) fn new(inner: Arc<StmtInner>, named_params: Vec<Vec<u8>>) -> Self {
         Self {
             inner,
             named_params,