Add support for MySQL 5.7

This commits fixes some issues that prevent us from connecting and
replicating from a 5.7 instance. The main issue with 5.7 is the lack
of support for the `LOCK INSTANCE FOR BACKUP` which allows tables to be
created during the backup.
During snapshot we do the following:
- get the list of tables
- for each table run a select 1 from table_name to add a metadata lock
  and prevent the table from being altered or dropped.
- get the current binlog position

If a new table happens to be created between the time we get the list
and before we get the current binlog position, we will not snapshot
the table. This is a rare case and the worst that can happen is the
table be missing from readyset in the first snapshot. If the user
restart readyset it will snapshot the table.

Fixes: REA-4910
Fixes: #1393

Release-Note-Core: Fixes issues that prevent us from connecting and
  replicating from a MySQL 5.7 instance.
Change-Id: I89b74edb9dd9a4ccfe00d55d0e32cd2efc491c54
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/8131
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <jason.b@readyset.io>

Author: altmannmarcelo

readyset-errors/src/lib.rs

@@ -678,8 +678,13 @@ pub enum ReadySetError {
     MirNodeMustHaveDfNodeAssigned { mir_node_index: usize },
 
     /// Error that the upstream database is using an unsupported server version.
-    #[error("Upstream server version {major}-{minor} is too low. Minimum supported major version is {min}")]
-    UnsupportedServerVersion { major: u16, minor: String, min: u16 },
+    #[error("Upstream server version {major}-{minor} is too low. Minimum supported version is {min_major}.{min_minor}")]
+    UnsupportedServerVersion {
+        major: u16,
+        minor: String,
+        min_major: u16,
+        min_minor: u16,
+    },
 
     /// Error that the upstream database reports a server version the ReadySet could not parse.
     #[error("Upstream server version could not be parsed")]

readyset-mysql/src/upstream.rs

@@ -26,7 +26,8 @@ type StatementID = u32;
 
 /// Indicates the minimum upstream server version that we currently support. Used to error out
 /// during connection phase if the version for the upstream server is too low.
-const MIN_UPSTREAM_VERSION: u16 = 8;
+const MIN_UPSTREAM_MAJOR_VERSION: u16 = 5;
+const MIN_UPSTREAM_MINOR_VERSION: u16 = 7;
 
 fn dt_to_value_params(dt: &[DfValue]) -> ReadySetResult<Vec<mysql_async::Value>> {
     dt.iter().map(|v| v.try_into()).collect()
@@ -202,11 +203,14 @@ impl MySqlUpstream {
 
         // Check that the server version is supported.
         let (major, minor, _) = conn.server_version();
-        if major < MIN_UPSTREAM_VERSION {
+        if major < MIN_UPSTREAM_MAJOR_VERSION
+            || (major == MIN_UPSTREAM_MAJOR_VERSION && minor < MIN_UPSTREAM_MINOR_VERSION)
+        {
             return Err(Error::ReadySet(ReadySetError::UnsupportedServerVersion {
                 major,
                 minor: minor.to_string(),
-                min: MIN_UPSTREAM_VERSION,
+                min_major: MIN_UPSTREAM_MAJOR_VERSION,
+                min_minor: MIN_UPSTREAM_MINOR_VERSION,
             }));
         }
 

readyset-psql/src/upstream.rs

@@ -25,7 +25,8 @@ use crate::Error;
 
 /// Indicates the minimum upstream server version that we currently support. Used to error out
 /// during connection phase if the version for the upstream server is too low.
-const MIN_UPSTREAM_VERSION: u16 = 13;
+const MIN_UPSTREAM_MAJOR_VERSION: u16 = 13;
+const MIN_UPSTREAM_MINOR_VERSION: u16 = 0;
 
 /// A connector to an underlying PostgreSQL database
 pub struct PostgreSqlUpstream {
@@ -208,11 +209,22 @@ impl UpstreamDatabase for PostgreSqlUpstream {
         let major = major
             .parse()
             .map_err(|_| Error::ReadySet(ReadySetError::UnparseableServerVersion))?;
-        if major < MIN_UPSTREAM_VERSION {
+        let minor: u16 = minor
+            .chars()
+            .take_while(|c| c.is_ascii_digit())
+            .collect::<String>()
+            .parse()
+            .map_err(|_| Error::ReadySet(ReadySetError::UnparseableServerVersion))?;
+
+        #[allow(clippy::absurd_extreme_comparisons)]
+        if major < MIN_UPSTREAM_MAJOR_VERSION
+            || (major == MIN_UPSTREAM_MAJOR_VERSION && minor < MIN_UPSTREAM_MINOR_VERSION)
+        {
             return Err(Error::ReadySet(ReadySetError::UnsupportedServerVersion {
                 major,
-                minor: minor.to_owned(),
-                min: MIN_UPSTREAM_VERSION,
+                minor: minor.to_string(),
+                min_major: MIN_UPSTREAM_MAJOR_VERSION,
+                min_minor: MIN_UPSTREAM_MINOR_VERSION,
             }));
         }
         let version = format!("{version} ReadySet");

replicators/src/mysql_connector/connector.rs

@@ -35,7 +35,8 @@ use replication_offset::ReplicationOffset;
 use crate::mysql_connector::utils::mysql_pad_collation_column;
 use crate::noria_adapter::{Connector, ReplicationAction};
 
-const CHECKSUM_QUERY: &str = "SET @source_binlog_checksum='CRC32'";
+use super::utils::get_mysql_version;
+
 const DEFAULT_SERVER_ID: u32 = u32::MAX - 55;
 const MAX_POSITION_TIME: u64 = 10;
 
@@ -84,7 +85,21 @@ impl MySqlBinlogConnector {
     /// know what type of checksum we support (NONE and CRC32 are the options), NONE seems to work
     /// but others use CRC32 🤷‍♂️
     async fn register_as_replica(&mut self) -> mysql::Result<()> {
-        self.connection.query_drop(CHECKSUM_QUERY).await?;
+        let query = match get_mysql_version(&mut self.connection).await {
+            Ok(version) => {
+                if version >= 80400 {
+                    // MySQL 8.4.0 and above
+                    "SET @source_binlog_checksum='CRC32'"
+                } else {
+                    // MySQL 8.3.0 and below
+                    "SET @master_binlog_checksum='CRC32'"
+                }
+            }
+            Err(err) => {
+                return Err(err);
+            }
+        };
+        self.connection.query_drop(query).await?;
 
         let cmd = mysql_common::packets::ComRegisterSlave::new(self.server_id());
         self.connection.write_command(&cmd).await?;

replicators/src/mysql_connector/snapshot.rs

@@ -27,7 +27,7 @@ use tokio::task::JoinHandle;
 use tracing::{debug, error, info, info_span, warn};
 use tracing_futures::Instrument;
 
-use super::utils::{mysql_pad_collation_column, parse_mysql_version};
+use super::utils::{get_mysql_version, mysql_pad_collation_column};
 use crate::db_util::DatabaseSchemas;
 use crate::mysql_connector::snapshot_type::SnapshotType;
 use crate::mysql_connector::utils::MYSQL_BATCH_SIZE;
@@ -350,19 +350,11 @@ impl MySqlReplicator {
         Ok(tx)
     }
 
-    /// Get MySQL Server Version
-    async fn get_mysql_version(&self) -> mysql::Result<u32> {
-        let mut conn = self.pool.get_conn().await?;
-        let version: mysql::Row = conn.query_first("SELECT VERSION()").await?.unwrap();
-        let version: String = version.get(0).expect("MySQL version");
-        parse_mysql_version(&version)
-    }
-
     /// Use the SHOW MASTER STATUS or SHOW BINARY LOG STATUS statement to determine
     /// the current binary log file name and position.
     async fn get_binlog_position(&self) -> mysql::Result<MySqlPosition> {
         let mut conn = self.pool.get_conn().await?;
-        let query = match self.get_mysql_version().await {
+        let query = match get_mysql_version(&mut conn).await {
             Ok(version) => {
                 if version >= 80400 {
                     // MySQL 8.4.0 and above
@@ -581,7 +573,7 @@ impl MySqlReplicator {
             match conn.query_drop("LOCK INSTANCE FOR BACKUP").await {
                 Ok(_) => Some(conn),
                 Err(err) => {
-                    warn!(%err, "Failed to acquire instance lock, DDL changes may cause inconsistency");
+                    warn!(%err, "Failed to acquire instance lock, if new tables are created, we might not detect them.");
                     None
                 }
             }

replicators/src/mysql_connector/utils.rs

@@ -1,9 +1,9 @@
-use std::string::FromUtf8Error;
-use std::sync::Arc;
-
+use mysql_async::{self as mysql, prelude::Queryable};
 use mysql_common::collations::{self, Collation, CollationId};
 use mysql_srv::ColumnType;
 use readyset_data::DfValue;
+use std::string::FromUtf8Error;
+use std::sync::Arc;
 
 //TODO(marce): Make this a configuration parameter or dynamically adjust based on the table size
 pub const MYSQL_BATCH_SIZE: usize = 100_000; // How many rows to fetch at a time from MySQL
@@ -58,6 +58,13 @@ pub fn parse_mysql_version(version: &str) -> mysql_async::Result<u32> {
     Ok(major * 10000 + minor * 100 + patch)
 }
 
+/// Get MySQL Server Version
+pub async fn get_mysql_version(conn: &mut mysql_async::Conn) -> mysql::Result<u32> {
+    let version: mysql::Row = conn.query_first("SELECT VERSION()").await?.unwrap();
+    let version: String = version.get(0).expect("MySQL version");
+    parse_mysql_version(&version)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;