binlog replication: support GTID

dm/config/task.go

@@ -65,6 +65,7 @@ var (
 type Meta struct {
 	BinLogName string `toml:"binlog-name" yaml:"binlog-name"`
 	BinLogPos  uint32 `toml:"binlog-pos" yaml:"binlog-pos"`
+	BinLogGTID string `toml:"binlog-gtid" yaml:"binlog-gtid"`
 }
 
 // Verify does verification on configs

dm/proto/dmworker.proto

@@ -181,7 +181,7 @@ message LoadStatus {
 message ShardingGroup {
     string target = 1;
     repeated string DDLs = 2;
-    string firstPos = 3;
+    string firstLocation = 3;
     repeated string synced = 4;
     repeated string unsynced = 5;
 }

dm/worker/server.go

@@ -808,6 +808,7 @@ func makeCommonWorkerResponse(reqErr error) *pb.CommonWorkerResponse {
 
 // all subTask in subTaskCfgs should have same source
 // this function return the min position in all subtasks, used for relay's position
+// Notes: used for relay, so don't need to use GTID
 func getMinPosInAllSubTasks(ctx context.Context, subTaskCfgs []*config.SubTaskConfig) (minPos *mysql.Position, err error) {
 	for _, subTaskCfg := range subTaskCfgs {
 		pos, err := getMinPosForSubTaskFunc(ctx, subTaskCfg)
@@ -853,6 +854,6 @@ func getMinPosForSubTask(ctx context.Context, subTaskCfg *config.SubTaskConfig)
 		return nil, errors.Annotate(err, "get min position from checkpoint")
 	}
 
-	pos := checkpoint.GlobalPoint()
-	return &pos, nil
+	location := checkpoint.GlobalPoint()
+	return &location.Position, nil
 }

dm/worker/worker.go

@@ -763,6 +763,7 @@ func (w *Worker) copyConfigFromWorker(cfg *config.SubTaskConfig) {
 	cfg.ServerID = w.cfg.ServerID
 	cfg.RelayDir = w.cfg.RelayDir
 	cfg.EnableGTID = w.cfg.EnableGTID
+	cfg.SyncerConfig.EnableGTID = w.cfg.EnableGTID
 	cfg.UseRelay = w.cfg.EnableRelay
 
 	// we can remove this from SubTaskConfig later, because syncer will always read from relay

loader/loader.go

@@ -1194,7 +1194,7 @@ func (l *Loader) checkpointID() string {
 
 func (l *Loader) getMydumpMetadata() error {
 	metafile := filepath.Join(l.cfg.LoaderConfig.Dir, "metadata")
-	pos, err := utils.ParseMetaData(metafile)
+	pos, _, err := utils.ParseMetaData(metafile)
 	if err != nil {
 		l.logCtx.L().Error("fail to parse dump metadata", log.ShortError(err))
 		return err

pkg/binlog/position.go

@@ -14,11 +14,13 @@
 package binlog
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	gmysql "github.com/siddontang/go-mysql/mysql"
 
+	"github.com/pingcap/dm/pkg/gtid"
 	"github.com/pingcap/dm/pkg/terror"
 	"github.com/pingcap/dm/pkg/utils"
 )
@@ -160,3 +162,38 @@ func ComparePosition(pos1, pos2 gmysql.Position) int {
 
 	return adjustedPos1.Compare(adjustedPos2)
 }
+
+// Location is used for save binlog's position and gtid
+type Location struct {
+	Position gmysql.Position
+
+	GTIDSet gtid.Set
+}
+
+func (p Location) String() string {
+	return fmt.Sprintf("Position: %v, GTIDSet: %s", p.Position, p.GTIDSet)
+}
+
+// CompareLocation returns:
+//   1 if point1 is bigger than point2
+//   0 if point1 is equal to point2
+//   -1 if point1 is less than point2
+func CompareLocation(location1, location2 Location) int {
+	if location1.GTIDSet != nil && len(location1.GTIDSet.String()) != 0 &&
+		location2.GTIDSet != nil && len(location2.GTIDSet.String()) != 0 {
+		contain1 := location1.GTIDSet.Contain(location2.GTIDSet)
+		contain2 := location2.GTIDSet.Contain(location1.GTIDSet)
+		if contain1 && contain2 {
+			// gtidSet1 contains gtidSet2 and gtidSet2 contains gtidSet1 means gtidSet1 equals to gtidSet2,
+			// then need to compare by position.
+		} else {
+			if contain1 {
+				return 1
+			} else if contain2 {
+				return -1
+			}
+		}
+	}
+
+	return ComparePosition(location1.Position, location2.Position)
+}

pkg/gtid/gtid.go

@@ -58,9 +58,9 @@ func ParserGTID(flavor, gtidStr string) (Set, error) {
 
 	switch flavor {
 	case mysql.MariaDBFlavor:
-		m = &mariadbGTIDSet{}
+		m = &MariadbGTIDSet{}
 	case mysql.MySQLFlavor:
-		m = &mySQLGTIDSet{}
+		m = &MySQLGTIDSet{}
 	default:
 		return nil, terror.ErrNotSupportedFlavor.Generate(flavor)
 	}
@@ -72,12 +72,12 @@ func ParserGTID(flavor, gtidStr string) (Set, error) {
 
 // MySQLGTIDSet wraps mysql.MysqlGTIDSet to implement gtidSet interface
 // extend some functions to retrieve and compute an intersection with other MySQL GTID Set
-type mySQLGTIDSet struct {
+type MySQLGTIDSet struct {
 	set *mysql.MysqlGTIDSet
 }
 
-// replace g by other
-func (g *mySQLGTIDSet) Set(other mysql.GTIDSet) error {
+// Set implements Set.Set, replace g by other
+func (g *MySQLGTIDSet) Set(other mysql.GTIDSet) error {
 	if other == nil {
 		return nil
 	}
@@ -91,12 +91,13 @@ func (g *mySQLGTIDSet) Set(other mysql.GTIDSet) error {
 	return nil
 }
 
-func (g *mySQLGTIDSet) Replace(other Set, masters []interface{}) error {
+// Replace implements Set.Replace
+func (g *MySQLGTIDSet) Replace(other Set, masters []interface{}) error {
 	if other == nil {
 		return nil
 	}
 
-	otherGS, ok := other.(*mySQLGTIDSet)
+	otherGS, ok := other.(*MySQLGTIDSet)
 	if !ok {
 		return terror.ErrNotMySQLGTID.Generate(other)
 	}
@@ -124,29 +125,32 @@ func (g *mySQLGTIDSet) Replace(other Set, masters []interface{}) error {
 	return nil
 }
 
-func (g *mySQLGTIDSet) delete(uuid string) {
+func (g *MySQLGTIDSet) delete(uuid string) {
 	delete(g.set.Sets, uuid)
 }
 
-func (g *mySQLGTIDSet) get(uuid string) (*mysql.UUIDSet, bool) {
+func (g *MySQLGTIDSet) get(uuid string) (*mysql.UUIDSet, bool) {
 	uuidSet, ok := g.set.Sets[uuid]
 	return uuidSet, ok
 }
 
-func (g *mySQLGTIDSet) Clone() Set {
-	return &mySQLGTIDSet{
+// Clone implements Set.Clone
+func (g *MySQLGTIDSet) Clone() Set {
+	return &MySQLGTIDSet{
 		set: g.set.Clone().(*mysql.MysqlGTIDSet),
 	}
 }
 
-func (g *mySQLGTIDSet) Origin() mysql.GTIDSet {
+// Origin implements Set.Origin
+func (g *MySQLGTIDSet) Origin() mysql.GTIDSet {
 	return g.set.Clone().(*mysql.MysqlGTIDSet)
 }
 
-func (g *mySQLGTIDSet) Equal(other Set) bool {
+// Equal implements Set.Equal
+func (g *MySQLGTIDSet) Equal(other Set) bool {
 	otherIsNil := other == nil
 	if !otherIsNil {
-		otherGS, ok := other.(*mySQLGTIDSet)
+		otherGS, ok := other.(*MySQLGTIDSet)
 		if !ok {
 			return false
 		}
@@ -162,10 +166,11 @@ func (g *mySQLGTIDSet) Equal(other Set) bool {
 	return g.set.Equal(other.Origin())
 }
 
-func (g *mySQLGTIDSet) Contain(other Set) bool {
+// Contain implements Set.Contain
+func (g *MySQLGTIDSet) Contain(other Set) bool {
 	otherIsNil := other == nil
 	if !otherIsNil {
-		otherGs, ok := other.(*mySQLGTIDSet)
+		otherGs, ok := other.(*MySQLGTIDSet)
 		if !ok {
 			return false
 		}
@@ -179,14 +184,15 @@ func (g *mySQLGTIDSet) Contain(other Set) bool {
 	return g.set.Contain(other.Origin())
 }
 
-func (g *mySQLGTIDSet) Truncate(end Set) error {
+// Truncate implements Set.Truncate
+func (g *MySQLGTIDSet) Truncate(end Set) error {
 	if end == nil {
 		return nil // do nothing
 	}
 	if !g.Contain(end) {
 		return terror.ErrGTIDTruncateInvalid.Generate(g, end)
 	}
-	endGs := end.(*mySQLGTIDSet) // already verify the type is `*mySQLGTIDSet` in `Contain`.
+	endGs := end.(*MySQLGTIDSet) // already verify the type is `*MySQLGTIDSet` in `Contain`.
 	if endGs == nil {
 		return nil // do nothing
 	}
@@ -209,20 +215,23 @@ func (g *mySQLGTIDSet) Truncate(end Set) error {
 	return nil
 }
 
-func (g *mySQLGTIDSet) String() string {
+func (g *MySQLGTIDSet) String() string {
 	if g.set == nil {
 		return ""
 	}
 	return g.set.String()
 }
 
 /************************ mariadb gtid set ***************************/
-type mariadbGTIDSet struct {
+
+// MariadbGTIDSet wraps mysql.MariadbGTIDSet to implement gtidSet interface
+// extend some functions to retrieve and compute an intersection with other Mariadb GTID Set
+type MariadbGTIDSet struct {
 	set *mysql.MariadbGTIDSet
 }
 
-// replace g by other
-func (m *mariadbGTIDSet) Set(other mysql.GTIDSet) error {
+// Set implements Set.Set, replace g by other
+func (m *MariadbGTIDSet) Set(other mysql.GTIDSet) error {
 	if other == nil {
 		return nil
 	}
@@ -236,12 +245,13 @@ func (m *mariadbGTIDSet) Set(other mysql.GTIDSet) error {
 	return nil
 }
 
-func (m *mariadbGTIDSet) Replace(other Set, masters []interface{}) error {
+// Replace implements Set.Replace
+func (m *MariadbGTIDSet) Replace(other Set, masters []interface{}) error {
 	if other == nil {
 		return nil
 	}
 
-	otherGS, ok := other.(*mariadbGTIDSet)
+	otherGS, ok := other.(*MariadbGTIDSet)
 	if !ok {
 		return terror.ErrNotMariaDBGTID.Generate(other)
 	}
@@ -269,29 +279,32 @@ func (m *mariadbGTIDSet) Replace(other Set, masters []interface{}) error {
 	return nil
 }
 
-func (m *mariadbGTIDSet) delete(domainID uint32) {
+func (m *MariadbGTIDSet) delete(domainID uint32) {
 	delete(m.set.Sets, domainID)
 }
 
-func (m *mariadbGTIDSet) get(domainID uint32) (*mysql.MariadbGTID, bool) {
+func (m *MariadbGTIDSet) get(domainID uint32) (*mysql.MariadbGTID, bool) {
 	gtid, ok := m.set.Sets[domainID]
 	return gtid, ok
 }
 
-func (m *mariadbGTIDSet) Clone() Set {
-	return &mariadbGTIDSet{
+// Clone implements Set.Clone
+func (m *MariadbGTIDSet) Clone() Set {
+	return &MariadbGTIDSet{
 		set: m.set.Clone().(*mysql.MariadbGTIDSet),
 	}
 }
 
-func (m *mariadbGTIDSet) Origin() mysql.GTIDSet {
+// Origin implements Set.Origin
+func (m *MariadbGTIDSet) Origin() mysql.GTIDSet {
 	return m.set.Clone().(*mysql.MariadbGTIDSet)
 }
 
-func (m *mariadbGTIDSet) Equal(other Set) bool {
+// Equal implements Set.Equal
+func (m *MariadbGTIDSet) Equal(other Set) bool {
 	otherIsNil := other == nil
 	if !otherIsNil {
-		otherGS, ok := other.(*mariadbGTIDSet)
+		otherGS, ok := other.(*MariadbGTIDSet)
 		if !ok {
 			return false
 		}
@@ -307,10 +320,11 @@ func (m *mariadbGTIDSet) Equal(other Set) bool {
 	return m.set.Equal(other.Origin())
 }
 
-func (m *mariadbGTIDSet) Contain(other Set) bool {
+// Contain implements Set.Contain
+func (m *MariadbGTIDSet) Contain(other Set) bool {
 	otherIsNil := other == nil
 	if !otherIsNil {
-		otherGS, ok := other.(*mariadbGTIDSet)
+		otherGS, ok := other.(*MariadbGTIDSet)
 		if !ok {
 			return false
 		}
@@ -324,14 +338,15 @@ func (m *mariadbGTIDSet) Contain(other Set) bool {
 	return m.set.Contain(other.Origin())
 }
 
-func (m *mariadbGTIDSet) Truncate(end Set) error {
+// Truncate implements Set.Truncate
+func (m *MariadbGTIDSet) Truncate(end Set) error {
 	if end == nil {
 		return nil // do nothing
 	}
 	if !m.Contain(end) {
 		return terror.ErrGTIDTruncateInvalid.Generate(m, end)
 	}
-	endGs := end.(*mariadbGTIDSet) // already verify the type is `*mariadbGTIDSet` in `Contain`.
+	endGs := end.(*MariadbGTIDSet) // already verify the type is `*MariadbGTIDSet` in `Contain`.
 	if endGs == nil {
 		return nil // do nothing
 	}
@@ -350,7 +365,7 @@ func (m *mariadbGTIDSet) Truncate(end Set) error {
 	return nil
 }
 
-func (m *mariadbGTIDSet) String() string {
+func (m *MariadbGTIDSet) String() string {
 	if m.set == nil {
 		return ""
 	}