binlog replication: support GTID

_utils/terror_gen/errors_release.txt

@@ -150,7 +150,7 @@ ErrBinlogExtractPosition,[code=22001:class=binlog-op:scope=internal:level=high],
 ErrBinlogInvalidFilename,[code=22002:class=binlog-op:scope=internal:level=high],"invalid binlog filename"
 ErrBinlogParsePosFromStr,[code=22003:class=binlog-op:scope=internal:level=high],""
 ErrCheckpointInvalidTaskMode,[code=24001:class=checkpoint:scope=internal:level=medium],"invalid task mode: %s"
-ErrCheckpointSaveInvalidPos,[code=24002:class=checkpoint:scope=internal:level=high],"save point %v is older than current pos %v"
+ErrCheckpointSaveInvalidPos,[code=24002:class=checkpoint:scope=internal:level=high],"save point %s is older than current location %s"
 ErrCheckpointInvalidTableFile,[code=24003:class=checkpoint:scope=internal:level=medium],"invalid db table sql file - %s"
 ErrCheckpointDBNotExistInFile,[code=24004:class=checkpoint:scope=internal:level=medium],"db (%s) not exist in data files, but in checkpoint"
 ErrCheckpointTableNotExistInFile,[code=24005:class=checkpoint:scope=internal:level=medium],"table (%s) not exist in db (%s) data files, but in checkpoint"

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
+// TODO: get min gtidSet
 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,15 @@
 package binlog
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
 	gmysql "github.com/siddontang/go-mysql/mysql"
+	"go.uber.org/zap"
 
+	"github.com/pingcap/dm/pkg/gtid"
+	"github.com/pingcap/dm/pkg/log"
 	"github.com/pingcap/dm/pkg/terror"
 	"github.com/pingcap/dm/pkg/utils"
 )
@@ -33,6 +37,11 @@ const (
 	posUUIDSuffixSeparator = "|"
 )
 
+var (
+	// MinPosition is the min binlog position
+	MinPosition = gmysql.Position{Pos: 4}
+)
+
 // PositionFromStr constructs a mysql.Position from a string representation like `mysql-bin.000001:2345`
 func PositionFromStr(s string) (gmysql.Position, error) {
 	parsed := strings.Split(s, ":")
@@ -160,3 +169,84 @@ 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
+}
+
+// NewLocation returns a new Location
+func NewLocation(flavor string) Location {
+	return Location{
+		Position: MinPosition,
+		GTIDSet:  gtid.MinGTIDSet(flavor),
+	}
+}
+
+func (l Location) String() string {
+	return fmt.Sprintf("position: %v, gtid-set: %s", l.Position, l.GTIDSetStr())
+}
+
+// GTIDSetStr returns gtid set's string
+func (l Location) GTIDSetStr() string {
+	gsetStr := ""
+	if l.GTIDSet != nil {
+		gsetStr = l.GTIDSet.String()
+	}
+
+	return gsetStr
+}
+
+// Clone clones a same Location
+func (l Location) Clone() Location {
+	return l.CloneWithFlavor("")
+}
+
+// CloneWithFlavor clones the location, and if the GTIDSet is nil, will create a GTIDSet with specified flavor.
+func (l Location) CloneWithFlavor(flavor string) Location {
+	var newGTIDSet gtid.Set
+	if len(flavor) != 0 {
+		newGTIDSet = gtid.MinGTIDSet(flavor)
+	}
+
+	if l.GTIDSet != nil {
+		newGTIDSet = l.GTIDSet.Clone()
+	}
+
+	return Location{
+		Position: gmysql.Position{
+			Name: l.Position.Name,
+			Pos:  l.Position.Pos,
+		},
+		GTIDSet: newGTIDSet,
+	}
+}
+
+// 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
+			}
+
+			// can't compare location by gtid, and will compare by position
+			log.L().Warn("gtidSet can't be compared", zap.Stringer("location1", location1), zap.Stringer("location2", location2))
+		}
+	}
+
+	return ComparePosition(location1.Position, location2.Position)
+}

pkg/binlog/position_test.go

@@ -18,6 +18,8 @@ import (
 
 	. "github.com/pingcap/check"
 	gmysql "github.com/siddontang/go-mysql/mysql"
+
+	"github.com/pingcap/dm/pkg/gtid"
 )
 
 var _ = Suite(&testPositionSuite{})
@@ -347,3 +349,196 @@ func (t *testPositionSuite) TestComparePosition(c *C) {
 		c.Assert(cmp, Equals, cs.cmp)
 	}
 }
+
+func (t *testPositionSuite) TestCompareCompareLocation(c *C) {
+	testCases := []struct {
+		flavor string
+		pos1   gmysql.Position
+		gset1  string
+		pos2   gmysql.Position
+		gset2  string
+		cmp    int
+	}{
+		{
+			// pos1 = pos2
+			gmysql.MySQLFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"",
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"",
+			0,
+		}, {
+			// pos1 = pos2
+			gmysql.MariaDBFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"",
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"",
+			0,
+		}, {
+			// pos1 < pos2
+			gmysql.MariaDBFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  122,
+			},
+			"",
+			-1,
+		}, {
+			// pos1 > pos2
+			gmysql.MySQLFlavor,
+			gmysql.Position{
+				Name: "binlog.00003",
+				Pos:  123,
+			},
+			"",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  122,
+			},
+			"",
+			1,
+		}, {
+			// gset1 = gset2, pos1 < pos2
+			gmysql.MySQLFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-4",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  122,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-4",
+			-1,
+		}, {
+			// compare by gtid set, gset2 contains gset1
+			gmysql.MySQLFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-2,53ea0ed1-9bf8-11e6-8bea-64006a897c74:1-2",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  124,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-4,53ea0ed1-9bf8-11e6-8bea-64006a897c74:1-3",
+			-1,
+		}, {
+			// compare by gtid set, gset1 contains gset2
+			gmysql.MySQLFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-2,53ea0ed1-9bf8-11e6-8bea-64006a897c74:1-3",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  124,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-2",
+			1,
+		}, {
+			// can't compare by gtid set, will compare by position, pos1 < pos2
+			gmysql.MySQLFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-2,53ea0ed1-9bf8-11e6-8bea-64006a897c74:2-4",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  124,
+			},
+			"53ea0ed1-9bf8-11e6-8bea-64006a897c73:1-2,53ea0ed1-9bf8-11e6-8bea-64006a897c74:1-3",
+			-1,
+		}, {
+			// gset1 = gset2, pos1 < pos2
+			gmysql.MariaDBFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"1-1-1,2-2-2",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  122,
+			},
+			"1-1-1,2-2-2",
+			-1,
+		}, {
+			// compare by gtid set, gset2 contains gset1
+			gmysql.MariaDBFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"1-1-1,2-2-2",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  124,
+			},
+			"1-1-1,2-2-2,3-3-3",
+			-1,
+		}, {
+			// compare by gtid set, gset1 contains gset2
+			gmysql.MariaDBFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"1-1-1,2-2-3",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  124,
+			},
+			"1-1-1,2-2-2",
+			1,
+		}, {
+			// can't compare by gtid set, will compare by position, pos1 < pos2
+			gmysql.MariaDBFlavor,
+			gmysql.Position{
+				Name: "binlog.00001",
+				Pos:  123,
+			},
+			"1-1-1,2-2-2",
+			gmysql.Position{
+				Name: "binlog.00002",
+				Pos:  124,
+			},
+			"2-2-2,3-3-3",
+			-1,
+		},
+	}
+
+	for _, cs := range testCases {
+		c.Log(cs)
+		gset1, err := gtid.ParserGTID(cs.flavor, cs.gset1)
+		c.Assert(err, IsNil)
+		gset2, err := gtid.ParserGTID(cs.flavor, cs.gset2)
+		c.Assert(err, IsNil)
+
+		cmp := CompareLocation(Location{cs.pos1, gset1}, Location{cs.pos2, gset2})
+		c.Assert(cmp, Equals, cs.cmp)
+	}
+
+}