GTID 5/5: GTID-aware replica catchup

config.go

@@ -689,11 +689,18 @@ type Config struct {
 	//
 	// Valid values:
 	//   - "" or "file_position": the classic (file, position) coordinate. This
-	//     is the default and current behavior.
-	//   - "gtid": GTID-set based coordinates (experimental, MySQL only). At this
-	//     stage GTID mode only affects coordinate representation, validation, and
-	//     the ability to read the current GTID set; binlog streaming still uses
-	//     file/position until a later stage wires GTID streaming.
+	//     is the default.
+	//   - "gtid": GTID-set based coordinates (experimental, MySQL only). GTID
+	//     mode drives binlog streaming start/stop, state persistence and resume,
+	//     progress reporting, and replica catchup via executed-set containment.
+	//     It requires @@GLOBAL.gtid_mode=ON on the source (and target when
+	//     target verification is enabled, and the replication master when
+	//     running from a replica).
+	//
+	// Note: in GTID mode the legacy file/position progress fields
+	// (LastSuccessfulBinlogPos, FinalBinlogPos) are not populated; use the
+	// coordinate fields (LastSuccessfulBinlogCoordinate, FinalBinlogCoordinate)
+	// instead.
 	//
 	// Prefer BinlogCoordinateMode over any future boolean flag: it keeps the
 	// file/position and GTID paths cleanly separated and leaves room for

copydb/config.go

@@ -70,6 +70,14 @@ type Config struct {
 	// SELECT file, position FROM meta.ptheartbeat WHERE server_id = master_server_id
 	ReplicatedMasterPositionQuery string
 
+	// This is the SQL query used to read the master's executed GTID set that
+	// has been replicated to the Source. It is only used when
+	// BinlogCoordinateMode is "gtid". The query must return a single column
+	// containing the GTID set string, e.g.
+	//
+	// SELECT gtid_executed FROM meta.ptheartbeat WHERE server_id = master_server_id
+	ReplicatedMasterGTIDQuery string
+
 	// The duration to wait for the replication to catchup before aborting. Only use if RunFerryFromReplica is true.
 	WaitForReplicationTimeout string
 

copydb/copydb.go

@@ -114,8 +114,6 @@ func (this *CopydbFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection(
 		return err
 	}
 
-	positionFetcher := ghostferry.ReplicatedMasterPositionViaCustomQuery{Query: this.config.ReplicatedMasterPositionQuery}
-
 	var timeout time.Duration
 	if this.config.WaitForReplicationTimeout == "" {
 		timeout = time.Duration(0)
@@ -126,11 +124,29 @@ func (this *CopydbFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection(
 		}
 	}
 
-	this.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{
-		MasterDB:                        masterDB,
-		Timeout:                         timeout,
-		ReplicatedMasterPositionFetcher: positionFetcher,
+	wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{
+		MasterDB:             masterDB,
+		Timeout:              timeout,
+		BinlogCoordinateMode: this.config.BinlogCoordinateMode,
 	}
+
+	if this.config.BinlogCoordinateMode == ghostferry.BinlogCoordinateGTID {
+		if this.config.ReplicatedMasterGTIDQuery == "" {
+			return fmt.Errorf("ReplicatedMasterGTIDQuery must be set when running from a replica in GTID mode")
+		}
+		wait.ReplicatedMasterCoordinateFetcher = ghostferry.ReplicatedMasterGTIDViaCustomQuery{
+			Query: this.config.ReplicatedMasterGTIDQuery,
+		}
+	} else {
+		if this.config.ReplicatedMasterPositionQuery == "" {
+			return fmt.Errorf("ReplicatedMasterPositionQuery must be set when running from a replica")
+		}
+		wait.ReplicatedMasterPositionFetcher = ghostferry.ReplicatedMasterPositionViaCustomQuery{
+			Query: this.config.ReplicatedMasterPositionQuery,
+		}
+	}
+
+	this.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait
 	return nil
 }
 

ferry.go

@@ -423,6 +423,7 @@ func (f *Ferry) Initialize() (err error) {
 	// of the MySQL databases.
 	if f.WaitUntilReplicaIsCaughtUpToMaster != nil {
 		f.WaitUntilReplicaIsCaughtUpToMaster.ReplicaDB = f.SourceDB
+		f.WaitUntilReplicaIsCaughtUpToMaster.BinlogCoordinateMode = f.Config.BinlogCoordinateMode
 
 		if f.WaitUntilReplicaIsCaughtUpToMaster.MasterDB == nil {
 			err = errors.New("must specify a MasterDB")
@@ -436,8 +437,25 @@ func (f *Ferry) Initialize() (err error) {
 			return err
 		}
 
-		// Ensures the query to check for position is executable.
-		_, err = f.WaitUntilReplicaIsCaughtUpToMaster.IsCaughtUp(zeroPosition, 1)
+		// In GTID mode the master's executed GTID set is the cutover target, so
+		// the master must have GTID mode enabled.
+		if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID {
+			if err = CheckServerGTIDModeEnabled(f.WaitUntilReplicaIsCaughtUpToMaster.MasterDB); err != nil {
+				f.logger.WithError(err).Error("source master is not configured for GTID mode")
+				return err
+			}
+		}
+
+		// Ensures the query to check for the replicated coordinate is
+		// executable. A zero target coordinate of the active mode is used only
+		// as a probe; the boolean result is ignored.
+		var probeTarget BinlogCoordinate
+		if f.Config.BinlogCoordinateMode == BinlogCoordinateGTID {
+			probeTarget = NewGTIDCoordinate("")
+		} else {
+			probeTarget = NewFilePositionCoordinate(zeroPosition)
+		}
+		_, err = f.WaitUntilReplicaIsCaughtUpToMaster.IsCaughtUpToCoordinate(probeTarget, 1)
 		if err != nil {
 			f.logger.WithError(err).Error("cannot check replicated master position on the source database")
 			return err

sharding/config.go

@@ -17,7 +17,11 @@ type Config struct {
 
 	SourceReplicationMaster       *ghostferry.DatabaseConfig
 	ReplicatedMasterPositionQuery string
-	RunFerryFromReplica           bool
+	// ReplicatedMasterGTIDQuery is used instead of ReplicatedMasterPositionQuery
+	// when BinlogCoordinateMode is "gtid". It must return a single column
+	// containing the master's executed GTID set string.
+	ReplicatedMasterGTIDQuery string
+	RunFerryFromReplica       bool
 
 	StatsDAddress string
 

sharding/sharding.go

@@ -264,12 +264,25 @@ func (r *ShardingFerry) initializeWaitUntilReplicaIsCaughtUpToMasterConnection()
 		return err
 	}
 
-	positionFetcher := ghostferry.ReplicatedMasterPositionViaCustomQuery{Query: r.config.ReplicatedMasterPositionQuery}
+	wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{
+		MasterDB:             masterDB,
+		BinlogCoordinateMode: r.config.BinlogCoordinateMode,
+	}
 
-	r.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{
-		MasterDB:                        masterDB,
-		ReplicatedMasterPositionFetcher: positionFetcher,
+	if r.config.BinlogCoordinateMode == ghostferry.BinlogCoordinateGTID {
+		if r.config.ReplicatedMasterGTIDQuery == "" {
+			return fmt.Errorf("ReplicatedMasterGTIDQuery must be set when running from a replica in GTID mode")
+		}
+		wait.ReplicatedMasterCoordinateFetcher = ghostferry.ReplicatedMasterGTIDViaCustomQuery{
+			Query: r.config.ReplicatedMasterGTIDQuery,
+		}
+	} else {
+		wait.ReplicatedMasterPositionFetcher = ghostferry.ReplicatedMasterPositionViaCustomQuery{
+			Query: r.config.ReplicatedMasterPositionQuery,
+		}
 	}
+
+	r.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait
 	return nil
 }
 

test/go/race_conditions_integration_test.go

@@ -2,6 +2,7 @@ package test
 
 import (
 	"fmt"
+	"os"
 	"testing"
 	"time"
 
@@ -84,6 +85,14 @@ func TestUpdateBinlogSelectCopy(t *testing.T) {
 }
 
 func TestMasterChangingBeforeStoppingBinlogStreaming(t *testing.T) {
+	// This test fakes an unreachable replicated-master coordinate to exercise
+	// master-demotion detection, which is independent of the binlog coordinate
+	// mode. The file/position run provides full coverage; the fabricated GTID
+	// timing differs enough that we skip the redundant GTID variant.
+	if os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") == string(ghostferry.BinlogCoordinateGTID) {
+		t.Skip("master-demotion detection is mode-independent; covered by the file/position run")
+	}
+
 	ferry := testhelpers.NewTestFerry()
 
 	sourceDB, err := ferry.Source.SqlDB(nil)

test/go/replication_config_test.go

@@ -2,10 +2,12 @@ package test
 
 import (
 	"fmt"
-	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math"
+	"os"
 	"testing"
 
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
+
 	"github.com/Shopify/ghostferry"
 	"github.com/Shopify/ghostferry/testhelpers"
 	"github.com/stretchr/testify/suite"
@@ -16,6 +18,8 @@ type ReplicationConfigTestSuite struct {
 
 	SourceDB                        *sql.DB
 	ReplicatedMasterPositionFetcher *ghostferry.ReplicatedMasterPositionViaCustomQuery
+	ReplicatedMasterGTIDFetcher     *ghostferry.ReplicatedMasterGTIDViaCustomQuery
+	gtidMode                        bool
 }
 
 func (t *ReplicationConfigTestSuite) SetupTest() {
@@ -25,14 +29,27 @@ func (t *ReplicationConfigTestSuite) SetupTest() {
 	t.SourceDB, err = t.Ferry.Source.SqlDB(nil)
 	t.Require().Nil(err)
 
-	t.ReplicatedMasterPositionFetcher = &ghostferry.ReplicatedMasterPositionViaCustomQuery{
-		Query: "SELECT 'mysql-bin.000003', 483685",
+	t.gtidMode = os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") == string(ghostferry.BinlogCoordinateGTID)
+
+	wait := &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{
+		MasterDB:             t.SourceDB,
+		BinlogCoordinateMode: t.Ferry.Config.BinlogCoordinateMode,
 	}
 
-	t.Ferry.WaitUntilReplicaIsCaughtUpToMaster = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{
-		MasterDB:                        t.SourceDB,
-		ReplicatedMasterPositionFetcher: t.ReplicatedMasterPositionFetcher,
+	if t.gtidMode {
+		// A high, unreachable GTID set so the replica is never "caught up".
+		t.ReplicatedMasterGTIDFetcher = &ghostferry.ReplicatedMasterGTIDViaCustomQuery{
+			Query: "SELECT '3e11fa47-71ca-11e1-9e33-c80aa9429562:1-1'",
+		}
+		wait.ReplicatedMasterCoordinateFetcher = t.ReplicatedMasterGTIDFetcher
+	} else {
+		t.ReplicatedMasterPositionFetcher = &ghostferry.ReplicatedMasterPositionViaCustomQuery{
+			Query: "SELECT 'mysql-bin.000003', 483685",
+		}
+		wait.ReplicatedMasterPositionFetcher = t.ReplicatedMasterPositionFetcher
 	}
+
+	t.Ferry.WaitUntilReplicaIsCaughtUpToMaster = wait
 }
 
 func (t *ReplicationConfigTestSuite) TearDownTest() {
@@ -57,6 +74,17 @@ func (t *ReplicationConfigTestSuite) TestErrorsIfItsRunFromAReplicaWithoutSettin
 }
 
 func (t *ReplicationConfigTestSuite) TestErrorsIfPositionFetcherQueryIsNotProvided() {
+	if t.gtidMode {
+		// The GTID fetcher scans a single column, so a two-column query is the
+		// error condition here.
+		t.ReplicatedMasterGTIDFetcher.Query = "SELECT 1, 2"
+
+		err := t.Ferry.Initialize()
+		t.Require().NotNil(err)
+		t.Require().Equal("sql: expected 2 destination arguments in Scan, not 1", err.Error())
+		return
+	}
+
 	t.ReplicatedMasterPositionFetcher.Query = "SELECT 1"
 
 	err := t.Ferry.Initialize()
@@ -73,6 +101,16 @@ func (t *ReplicationConfigTestSuite) TestErrorsIfProvidedMasterIsReadOnly() {
 }
 
 func (t *ReplicationConfigTestSuite) TestCanInitializeFerryWithValidConfig() {
+	if t.gtidMode {
+		// A high, valid GTID set that parses cleanly; Initialize only probes
+		// that the fetcher query is executable, so it must succeed.
+		t.ReplicatedMasterGTIDFetcher.Query = "SELECT '3e11fa47-71ca-11e1-9e33-c80aa9429562:1-999999999'"
+
+		err := t.Ferry.Initialize()
+		t.Require().Nil(err)
+		return
+	}
+
 	t.ReplicatedMasterPositionFetcher.Query = fmt.Sprintf("SELECT 'mysql-bin.999999',%d", math.MaxUint32)
 
 	err := t.Ferry.Initialize()

test/go/wait_until_replica_is_caught_up_to_master_test.go

@@ -79,3 +79,68 @@ func (s *WaitUntilReplicaIsCaughtUpToMasterSuite) TestIsCaughtUpIsCorrect() {
 func TestWaitUntilReplicaIsCaughtUpToMaster(t *testing.T) {
 	suite.Run(t, new(WaitUntilReplicaIsCaughtUpToMasterSuite))
 }
+
+type WaitUntilReplicaIsCaughtUpToMasterGTIDSuite struct {
+	suite.Suite
+
+	w *ghostferry.WaitUntilReplicaIsCaughtUpToMaster
+}
+
+func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) SetupTest() {
+	config := testhelpers.NewTestConfig()
+
+	masterDB, err := config.Source.SqlDB(nil)
+	s.Require().Nil(err)
+	replicaDB, err := config.Target.SqlDB(nil)
+	s.Require().Nil(err)
+
+	s.w = &ghostferry.WaitUntilReplicaIsCaughtUpToMaster{
+		MasterDB:             masterDB,
+		ReplicaDB:            replicaDB,
+		BinlogCoordinateMode: ghostferry.BinlogCoordinateGTID,
+		ReplicatedMasterCoordinateFetcher: ghostferry.ReplicatedMasterGTIDViaCustomQuery{
+			Query: "SELECT gtid_executed FROM meta.heartbeat_gtid WHERE server_id = 1",
+		},
+	}
+
+	s.recreateTable(s.w.MasterDB)
+	s.recreateTable(s.w.ReplicaDB)
+}
+
+func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) recreateTable(db *sql.DB) {
+	_, err := db.Exec("DROP DATABASE IF EXISTS meta")
+	s.Require().Nil(err)
+
+	_, err = db.Exec("CREATE DATABASE meta")
+	s.Require().Nil(err)
+
+	_, err = db.Exec("CREATE TABLE meta.heartbeat_gtid (server_id int unsigned NOT NULL, gtid_executed longtext, PRIMARY KEY (server_id))")
+	s.Require().Nil(err)
+}
+
+func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) updateHeartbeatGTID(db *sql.DB, gtidSet string) {
+	_, err := db.Exec("REPLACE INTO meta.heartbeat_gtid (server_id, gtid_executed) VALUES (1, ?)", gtidSet)
+	s.Require().Nil(err)
+}
+
+func (s *WaitUntilReplicaIsCaughtUpToMasterGTIDSuite) TestIsCaughtUpToCoordinateGTID() {
+	uuid := "3e11fa47-71ca-11e1-9e33-c80aa9429562"
+	target := ghostferry.NewGTIDCoordinate(uuid + ":1-100")
+
+	// Replica has replicated only up to :1-50, which does not contain the
+	// target :1-100.
+	s.updateHeartbeatGTID(s.w.ReplicaDB, uuid+":1-50")
+	isCaughtUp, err := s.w.IsCaughtUpToCoordinate(target, 1)
+	s.Require().Nil(err)
+	s.Require().False(isCaughtUp)
+
+	// Replica now contains the full target set.
+	s.updateHeartbeatGTID(s.w.ReplicaDB, uuid+":1-150")
+	isCaughtUp, err = s.w.IsCaughtUpToCoordinate(target, 1)
+	s.Require().Nil(err)
+	s.Require().True(isCaughtUp)
+}
+
+func TestWaitUntilReplicaIsCaughtUpToMasterGTID(t *testing.T) {
+	suite.Run(t, new(WaitUntilReplicaIsCaughtUpToMasterGTIDSuite))
+}