Merge pull request #398 from Shopify/tyeh/make-mysql8.4-compatible

Update for MySQl 8.4 compatibility

Author: chenhuiyeh

.github/workflows/start-mysql.sh

@@ -3,6 +3,8 @@ set -xe
 
 if [ "$MYSQL_VERSION" == "8.0" ]; then
   docker compose -f docker-compose_8.0.yml up -d mysql-1 mysql-2
+elif [ "$MYSQL_VERSION" == "8.4" ]; then
+  docker compose -f docker-compose_8.4.yml up -d mysql-1 mysql-2
 else
   docker compose up -d mysql-1 mysql-2
 fi

.github/workflows/tests.yml

@@ -10,7 +10,7 @@ jobs:
   gh-285:
     strategy:
       matrix:
-        mysql: ["5.7", "8.0"]
+        mysql: ["5.7", "8.0", "8.4"]
 
     runs-on: ubuntu-latest
     continue-on-error: true
@@ -35,7 +35,7 @@ jobs:
   go-test:
     strategy:
       matrix:
-        mysql: ["5.7", "8.0"]
+        mysql: ["5.7", "8.0", "8.4"]
 
     runs-on: ubuntu-latest
     timeout-minutes: 15
@@ -60,7 +60,7 @@ jobs:
   ruby-test:
     strategy:
       matrix:
-        mysql: ["5.7", "8.0"]
+        mysql: ["5.7", "8.0", "8.4"]
 
     runs-on: ubuntu-latest
     timeout-minutes: 15

binlog_streamer.go

@@ -499,9 +499,19 @@ func idExistsOnServer(id uint32, db *sql.DB) (bool, error) {
 }
 
 func idsOnServer(db *sql.DB) ([]uint32, error) {
-	rows, err := db.Query("SHOW SLAVE HOSTS")
+	var query string
+	var errorMsg string
+	version, _ := db.QueryMySQLVersion()
+	if isVersionAtLeast(version, "8.4.0") {
+		query = "SHOW REPLICAS"
+		errorMsg = "replicas"
+	} else {
+		query = "SHOW SLAVE HOSTS"
+		errorMsg = "slave hosts"
+	}
+	rows, err := db.Query(query)
 	if err != nil {
-		return nil, fmt.Errorf("could not get slave hosts: %s", err)
+		return nil, fmt.Errorf("could not get %s: %s", errorMsg, err)
 	}
 	defer rows.Close()
 
@@ -517,16 +527,16 @@ func idsOnServer(db *sql.DB) ([]uint32, error) {
 		// i.e MySQL/Percona have 5 columns as it includes slave_uuid for MariaDB slave_uuid is omitted
 		// since all other values are not used check for the amount of columns and gather only what is possible
 		if err != nil {
-			return nil, fmt.Errorf("could not get columns from slave hosts: %v", err)
+			return nil, fmt.Errorf("could not get columns from %s: %v", errorMsg, err)
 		} else if len(columns) == 5 {
 			err = rows.Scan(&server_id, &host, &port, &master_id, &slave_uuid)
 		} else if len(columns) == 4 {
 			err = rows.Scan(&server_id, &host, &port, &master_id)
 		} else {
-			return nil, fmt.Errorf("could not scan SHOW SLAVE HOSTS row, err: unknown result set with %d columns: %v", len(columns), columns)
+			return nil, fmt.Errorf("could not scan %s row, err: unknown result set with %d columns: %v", query, len(columns), columns)
 		}
 		if err != nil {
-			return nil, fmt.Errorf("could not scan SHOW SLAVE HOSTS row, err: %s", err.Error())
+			return nil, fmt.Errorf("could not scan %s row, err: %s", query, err.Error())
 		}
 
 		server_ids = append(server_ids, server_id)

copydb/test/copydb_test.go

@@ -140,7 +140,7 @@ func (t *CopydbTestSuite) TestCreateDatabaseCopiesTheRightCollation() {
 	err = row.Scan(&characterSet, &collation)
 	t.Require().Nil(err)
 
-	if os.Getenv("MYSQL_VERSION") == "8.0" {
+	if os.Getenv("MYSQL_VERSION") == "8.0" || os.Getenv("MYSQL_VERSION") == "8.4" {
 		t.Require().Equal(characterSet, "utf8mb3")
 		t.Require().Equal(collation, "utf8mb3_general_ci")
 	} else {

docker-compose_8.4.yml

@@ -0,0 +1,68 @@
+version: "3"
+services:
+  mysql-1:
+    image: percona/percona-server:8.4.4-4.1
+    command: --server-id=1
+      --log-bin=mysql-bin
+      --max-binlog-size=4096
+      --binlog-format=ROW
+      --sync-binlog=1
+      --log-replica-updates=ON
+      --gtid-mode=ON
+      --enforce-gtid-consistency=ON
+      --character-set-server=utf8mb4
+      --collation-server=utf8mb4_unicode_ci
+      --max-connections=1000
+      --read-only=OFF
+      --binlog-rows-query-log-events=ON
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: "yes"
+      MYSQL_ROOT_HOST: "%"
+    volumes:
+      - /var/lib/mysql
+    ports:
+      - "29291:3306"
+
+  mysql-2:
+    image: percona/percona-server:8.4.4-4.1
+    command: --server-id=2
+      --log-bin=mysql-bin
+      --binlog-format=ROW
+      --max-binlog-size=4096
+      --sync-binlog=1
+      --log-replica-updates=ON
+      --gtid-mode=ON
+      --enforce-gtid-consistency=ON
+      --character-set-server=utf8mb4
+      --collation-server=utf8mb4_unicode_ci
+      --max-connections=1000
+      --binlog-rows-query-log-events=ON
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: "yes"
+      MYSQL_ROOT_HOST: "%"
+    volumes:
+      - /var/lib/mysql
+    ports:
+      - "29292:3306"
+
+  mysql-3:
+    image: percona/percona-server:8.4.4-4.1
+    command: --server-id=3
+      --log-bin=mysql-bin
+      --binlog-format=ROW
+      --max-binlog-size=4096
+      --sync-binlog=1
+      --log-replica-updates=ON
+      --gtid-mode=ON
+      --enforce-gtid-consistency=ON
+      --character-set-server=utf8mb4
+      --collation-server=utf8mb4_unicode_ci
+      --max-connections=1000
+      --binlog-rows-query-log-events=ON
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: "yes"
+      MYSQL_ROOT_HOST: "%"
+    volumes:
+      - /var/lib/mysql
+    ports:
+      - "29293:3306"

test/go/lag_throttler_test.go

@@ -145,7 +145,7 @@ func TestThrottlerRunErrors(t *testing.T) {
 	assert.NotNil(t, err)
 	expectedError := "Error 1146: Table 'meta.lag_table' doesn't exist"
 
-	if os.Getenv("MYSQL_VERSION") == "8.0" {
+	if os.Getenv("MYSQL_VERSION") == "8.0" || os.Getenv("MYSQL_VERSION") == "8.4" {
 		expectedError = "Error 1049: Unknown database 'meta'"
 	}
 

test/integration/inline_verifier_test.rb

@@ -590,7 +590,7 @@ def test_utfmb3_data_from_utfmb4_to_utfmb3
   # https://github.com/Shopify/ghostferry/pull/328#discussion_r791197939
   def test_utfmb4_data_from_utfmb4_to_utfmb3
     run_collation_test(UTF8MB4DATA, "utf8mb4", "utf8mb3", identical: false)
-  end unless ENV['MYSQL_VERSION'] == '8.0'
+  end unless ENV['MYSQL_VERSION'] == '8.0' || ENV['MYSQL_VERSION'] == '8.4'
 
   private
 

test/integration/interrupt_resume_test.rb

@@ -311,8 +311,9 @@ def test_interrupt_resume_inline_verifier_will_verify_additional_rows_changed_on
   # originally taken from @kolbitsch-lastline in https://github.com/Shopify/ghostferry/pull/160
   def test_interrupt_resume_between_consecutive_rows_events
     ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY, config: { verifier_type: "Inline" })
-
-    start_binlog_status = source_db.query('SHOW MASTER STATUS').first
+    version, _ = source_db.query("SELECT @@version").first["@@version"]
+    query =  version < "8.4.0" ? "SHOW MASTER STATUS" : "SHOW BINARY LOG STATUS"
+    start_binlog_status = source_db.query(query).first
 
     # create a series of rows-events that do not have interleaved table-map
     # events. This is the case when multiple rows are affected in a single

test/integration/types_test.rb

@@ -412,7 +412,7 @@ def test_decimal
 
   def format_float_based_on_mysql_version(value)
     # mysql 5.7 removes the trailing zeros when `cast...as json` is used
-    ENV["MYSQL_VERSION"] == "8.0" ? value.to_s : value.to_i.to_s
+    ENV["MYSQL_VERSION"] == "8.0" || ENV["MYSQL_VERSION"] == "8.4" ? value.to_s : value.to_i.to_s
   end
 
   def insert_json_on_source

utils.go

@@ -6,6 +6,7 @@ import (
 	sqlorig "database/sql"
 	"encoding/binary"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -188,10 +189,28 @@ func (c *StmtCache) getStmt(query string) (*sqlorig.Stmt, bool) {
 	return stmt, exists
 }
 
+func isVersionAtLeast(version string, targetVersion string) bool {
+	result, err := mysql.CompareServerVersions(version, targetVersion)
+	if err != nil {
+		return false
+	}
+	return result >= 0
+}
+
+func getBinlogStatusCommand(db *sql.DB) string {
+	version, _ := db.QueryMySQLVersion()
+	if isVersionAtLeast(version, "8.4.0") {
+		return "SHOW BINARY LOG STATUS"
+	} else {
+		return "SHOW MASTER STATUS"
+	}
+}
+
 func ShowMasterStatusBinlogPosition(db *sql.DB) (mysql.Position, error) {
-	rows, err := db.Query("SHOW MASTER STATUS")
+	query := getBinlogStatusCommand(db)
+	rows, err := db.Query(query)
 	if err != nil {
-		return NewMysqlPosition("", 0, err)
+		return NewMysqlPosition("", 0, err, db)
 	}
 	defer rows.Close()
 	var file string
@@ -201,7 +220,7 @@ func ShowMasterStatusBinlogPosition(db *sql.DB) (mysql.Position, error) {
 	if rows.Next() {
 		cols, err = rows.Columns()
 		if err != nil {
-			return NewMysqlPosition(file, position, err)
+			return NewMysqlPosition(file, position, err, db)
 		}
 		switch len(cols) {
 		case 4:
@@ -210,18 +229,20 @@ func ShowMasterStatusBinlogPosition(db *sql.DB) (mysql.Position, error) {
 			err = rows.Scan(&file, &position, &binlog_do_db, &binlog_ignore_db, &executed_gtid_set)
 		}
 	}
-	return NewMysqlPosition(file, position, err)
+	return NewMysqlPosition(file, position, err, db)
 }
 
-func NewMysqlPosition(file string, position uint32, err error) (mysql.Position, error) {
+func NewMysqlPosition(file string, position uint32, err error, db *sql.DB) (mysql.Position, error) {
+	var binlogStatusCmd string
+	binlogStatusCmd = getBinlogStatusCommand(db)
 	switch {
 	case err == sqlorig.ErrNoRows:
-		return mysql.Position{}, fmt.Errorf("no results from show master status")
+		return mysql.Position{}, fmt.Errorf("no results from %s", strings.ToLower(binlogStatusCmd))
 	case err != nil:
 		return mysql.Position{}, err
 	default:
 		if file == "" {
-			return mysql.Position{}, fmt.Errorf("show master status does not show a file")
+			return mysql.Position{}, fmt.Errorf("%s does not show a file", strings.ToLower(binlogStatusCmd))
 		}
 
 		return mysql.Position{