mysql: Handle more deprecated SQL commands (#15907)

Signed-off-by: Dirkjan Bussink <d.bussink@gmail.com>

examples/compose/fix_replication.sh

@@ -26,15 +26,15 @@ cd "$(dirname "${BASH_SOURCE[0]}")"
 
 function get_replication_status() {
     # Get replication status
-    STATUS_LINE=$(mysql -u$DB_USER -p$DB_PASS -h 127.0.0.1 -e "SHOW SLAVE STATUS\G")
+    STATUS_LINE=$(mysql -u$DB_USER -p$DB_PASS -h 127.0.0.1 -e "SHOW REPLICA STATUS\G")
     LAST_ERRNO=$(grep "Last_IO_Errno:" <<< "$STATUS_LINE" | awk '{ print $2 }')
-    SLAVE_SQL_RUNNING=$(grep "Slave_SQL_Running:" <<< "$STATUS_LINE" | awk '{ print $2 }')
-    SLAVE_IO_RUNNING=$(grep "Slave_IO_Running:" <<< "$STATUS_LINE" | awk '{ print $2 }')
-    MASTER_HOST=$(grep "Master_Host:" <<< "$STATUS_LINE" | awk '{ print $2 }')
-    MASTER_PORT=$(grep "Master_Port:" <<< "$STATUS_LINE" | awk '{ print $2 }')
+    REPLICA_SQL_RUNNING=$(grep "Replica_SQL_Running:" <<< "$STATUS_LINE" | awk '{ print $2 }')
+    REPLICA_IO_RUNNING=$(grep "Replica_IO_Running:" <<< "$STATUS_LINE" | awk '{ print $2 }')
+    SOURCE_HOST=$(grep "Source_Host:" <<< "$STATUS_LINE" | awk '{ print $2 }')
+    SOURCE_PORT=$(grep "Source_Port:" <<< "$STATUS_LINE" | awk '{ print $2 }')
 
-    echo "Slave_SQL_Running: $SLAVE_SQL_RUNNING"
-    echo "Slave_IO_Running: $SLAVE_IO_RUNNING"
+    echo "Replica_SQL_Running: $REPLICA_SQL_RUNNING"
+    echo "Replica_IO_Running: $REPLICA_IO_RUNNING"
     echo "Last_IO_Errno: $LAST_ERRNO"
 }
 
@@ -54,7 +54,7 @@ get_replication_status
 [ ${1:-''} != 'status' ] || exit 0;
 
 # Check if IO_Thread is running
-if [[ $SLAVE_IO_RUNNING = "No" && $LAST_ERRNO = 1236 ]]; then
+if [[ $REPLICA_IO_RUNNING = "No" && $LAST_ERRNO = 1236 ]]; then
 
     echo "Primary has purged bin logs that replica requires. Sync will require restore from mysqldump"
     if [[ -f $KEYSPACE.sql ]] ; then
@@ -64,7 +64,7 @@ if [[ $SLAVE_IO_RUNNING = "No" && $LAST_ERRNO = 1236 ]]; then
     else
         echo "Starting mysqldump. This may take a while.."
         # Modify flags to user's requirements
-        if mysqldump -h $MASTER_HOST -P $MASTER_PORT -u$DB_USER -p$DB_PASS --databases $KEYSPACE \
+        if mysqldump -h $SOURCE_HOST -P $SOURCE_PORT -u$DB_USER -p$DB_PASS --databases $KEYSPACE \
             --triggers --routines --events --hex-blob  --master-data=1 --quick --order-by-primary \
             --no-autocommit --skip-comments --skip-add-drop-table --skip-add-locks \
             --skip-disable-keys --single-transaction --set-gtid-purged=on --verbose > $KEYSPACE.sql; then

go/mysql/constants.go

@@ -289,6 +289,4 @@ func IsNum(typ uint8) bool {
 
 const (
 	readReplicationConnectionConfiguration = "SELECT * FROM performance_schema.replication_connection_configuration"
-	readReplicaNetTimeout                  = "select @@global.replica_net_timeout"
-	readSlaveNetTimeout                    = "select @@global.slave_net_timeout"
 )

go/mysql/endtoend/replication_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"vitess.io/vitess/go/mysql/replication"
 	"vitess.io/vitess/go/mysql/sqlerror"
 
 	"vitess.io/vitess/go/mysql"
@@ -46,15 +47,6 @@ func connectForReplication(t *testing.T, rbr bool) (*mysql.Conn, mysql.BinlogFor
 		t.Fatal(err)
 	}
 
-	// We need to know if this is MariaDB, to set the right flag.
-	if conn.IsMariaDB() {
-		// This flag is required to get GTIDs from MariaDB.
-		t.Log("MariaDB: sensing SET @mariadb_slave_capability=4")
-		if _, err := conn.ExecuteFetch("SET @mariadb_slave_capability=4", 0, false); err != nil {
-			t.Fatalf("failed to set @mariadb_slave_capability=4: %v", err)
-		}
-	}
-
 	// Switch server to RBR if needed.
 	if rbr {
 		if _, err := conn.ExecuteFetch("SET GLOBAL binlog_format='ROW'", 0, false); err != nil {
@@ -63,25 +55,21 @@ func connectForReplication(t *testing.T, rbr bool) (*mysql.Conn, mysql.BinlogFor
 	}
 
 	// First we get the current binlog position.
-	result, err := conn.ExecuteFetch("SHOW MASTER STATUS", 1, true)
-	require.NoError(t, err, "SHOW MASTER STATUS failed: %v", err)
+	status, err := conn.ShowPrimaryStatus()
+	require.NoError(t, err, "retrieving primary status failed: %v", err)
 
-	if len(result.Fields) < 2 || result.Fields[0].Name != "File" || result.Fields[1].Name != "Position" ||
-		len(result.Rows) != 1 {
-		t.Fatalf("SHOW MASTER STATUS returned unexpected result: %v", result)
-	}
-	file := result.Rows[0][0].ToString()
-	position, err := result.Rows[0][1].ToCastUint64()
-	require.NoError(t, err, "SHOW MASTER STATUS returned invalid position: %v", result.Rows[0][1])
+	filePos := status.FilePosition.GTIDSet.(replication.FilePosGTID)
+	file := filePos.File
+	position := filePos.Pos
 
 	// Tell the server that we understand the format of events
 	// that will be used if binlog_checksum is enabled on the server.
-	if _, err := conn.ExecuteFetch("SET @master_binlog_checksum=@@global.binlog_checksum", 0, false); err != nil {
-		t.Fatalf("failed to set @master_binlog_checksum=@@global.binlog_checksum: %v", err)
+	if _, err := conn.ExecuteFetch("SET @source_binlog_checksum = @@global.binlog_checksum, @master_binlog_checksum=@@global.binlog_checksum", 0, false); err != nil {
+		t.Fatalf("failed to set @source_binlog_checksum=@@global.binlog_checksum: %v", err)
 	}
 
 	// Write ComBinlogDump packet with to start streaming events from here.
-	if err := conn.WriteComBinlogDump(1, file, uint32(position), 0); err != nil {
+	if err := conn.WriteComBinlogDump(1, file, position, 0); err != nil {
 		t.Fatalf("WriteComBinlogDump failed: %v", err)
 	}
 

go/mysql/flavor.go

@@ -123,11 +123,14 @@ type flavor interface {
 	// as the new replication source (without changing any GTID position).
 	setReplicationSourceCommand(params *ConnParams, host string, port int32, heartbeatInterval float64, connectRetry int) string
 
+	// resetBinaryLogsCommand returns the command to reset the binary logs.
+	resetBinaryLogsCommand() string
+
 	// status returns the result of the appropriate status command,
 	// with parsed replication position.
 	status(c *Conn) (replication.ReplicationStatus, error)
 
-	// primaryStatus returns the result of 'SHOW MASTER STATUS',
+	// primaryStatus returns the result of 'SHOW BINARY LOG STATUS',
 	// with parsed executed position.
 	primaryStatus(c *Conn) (replication.PrimaryStatus, error)
 
@@ -185,15 +188,16 @@ func GetFlavor(serverVersion string, flavorFunc func() flavor) (f flavor, capabl
 			f = mariadbFlavor102{mariadbFlavor{serverVersion: fmt.Sprintf("%f", mariadbVersion)}}
 		}
 	case strings.HasPrefix(serverVersion, mysql8VersionPrefix):
-		recent, _ := capabilities.MySQLVersionHasCapability(serverVersion, capabilities.ReplicaTerminologyCapability)
-		if recent {
+		if latest, _ := capabilities.ServerVersionAtLeast(serverVersion, 8, 2, 0); latest {
+			f = mysqlFlavor82{mysqlFlavor{serverVersion: serverVersion}}
+		} else if recent, _ := capabilities.MySQLVersionHasCapability(serverVersion, capabilities.ReplicaTerminologyCapability); recent {
 			f = mysqlFlavor8{mysqlFlavor{serverVersion: serverVersion}}
 		} else {
-			f = mysqlFlavor8Legacy{mysqlFlavor{serverVersion: serverVersion}}
+			f = mysqlFlavor8Legacy{mysqlFlavorLegacy{mysqlFlavor{serverVersion: serverVersion}}}
 		}
 	default:
 		// If unknown, return the most basic flavor: MySQL 57.
-		f = mysqlFlavor57{mysqlFlavor{serverVersion: serverVersion}}
+		f = mysqlFlavor57{mysqlFlavorLegacy{mysqlFlavor{serverVersion: serverVersion}}}
 	}
 	return f, f.supportsCapability, canonicalVersion
 }
@@ -400,7 +404,7 @@ func (c *Conn) ShowReplicationStatus() (replication.ReplicationStatus, error) {
 	return c.flavor.status(c)
 }
 
-// ShowPrimaryStatus executes the right SHOW MASTER STATUS command,
+// ShowPrimaryStatus executes the right SHOW BINARY LOG STATUS command,
 // and returns a parsed executed Position, as well as file based Position.
 func (c *Conn) ShowPrimaryStatus() (replication.PrimaryStatus, error) {
 	return c.flavor.primaryStatus(c)

go/mysql/flavor_filepos.go

@@ -233,6 +233,11 @@ func (flv *filePosFlavor) setReplicationSourceCommand(params *ConnParams, host s
 	return "unsupported"
 }
 
+// resetBinaryLogsCommand is part of the Flavor interface.
+func (flv *filePosFlavor) resetBinaryLogsCommand() string {
+	return "unsupported"
+}
+
 // status is part of the Flavor interface.
 func (flv *filePosFlavor) status(c *Conn) (replication.ReplicationStatus, error) {
 	qr, err := c.ExecuteFetch("SHOW SLAVE STATUS", 100, true /* wantfields */)

go/mysql/flavor_mariadb.go

@@ -219,6 +219,10 @@ func (mariadbFlavor) setReplicationSourceCommand(params *ConnParams, host string
 	return "CHANGE MASTER TO\n  " + strings.Join(args, ",\n  ")
 }
 
+func (mariadbFlavor) resetBinaryLogsCommand() string {
+	return "RESET MASTER"
+}
+
 // status is part of the Flavor interface.
 func (mariadbFlavor) status(c *Conn) (replication.ReplicationStatus, error) {
 	qr, err := c.ExecuteFetch("SHOW ALL SLAVES STATUS", 100, true /* wantfields */)
@@ -288,7 +292,7 @@ func (mariadbFlavor) replicationConfiguration(c *Conn) (*replicationdata.Configu
 
 // replicationNetTimeout is part of the Flavor interface.
 func (mariadbFlavor) replicationNetTimeout(c *Conn) (int32, error) {
-	qr, err := c.ExecuteFetch(readSlaveNetTimeout, 1, false)
+	qr, err := c.ExecuteFetch("select @@global.slave_net_timeout", 1, false)
 	if err != nil {
 		return 0, err
 	}