*: support multiple sources replication for mariadb (go-mysql-org#272)

Author: IANTHEREAL

canal/sync.go

@@ -98,7 +98,11 @@ func (c *Canal) runSyncBinlog() error {
 }
 case *replication.MariadbGTIDEvent:
 // try to save the GTID later
-gtid := &e.GTID
+gtid, err := mysql.ParseMariadbGTIDSet(e.GTID.String())
+if err != nil {
+return errors.Trace(err)
+}
+
 c.master.UpdateGTID(gtid)
 if err := c.eventHandler.OnGTID(gtid); err != nil {
 return errors.Trace(err)

failover/mariadb_gtid_handler.go

@@ -46,12 +46,12 @@ func (h *MariadbGTIDHandler) FindBestSlaves(slaves []*Server) ([]*Server, error)
 if len(str) == 0 {
 seq = 0
 } else {
-g, err := ParseMariadbGTIDSet(str)
+g, err := ParseMariadbGTID(str)
 if err != nil {
 return nil, errors.Trace(err)
 }
 
-seq = g.(*MariadbGTID).SequenceNumber
+seq = g.SequenceNumber
 }
 
 ps[i] = seq

mysql/mariadb_gtid.go

@@ -1,21 +1,24 @@
 package mysql
 
 import (
+"bytes"
 "fmt"
 "strconv"
 "strings"
 
 "github.com/juju/errors"
+"github.com/siddontang/go/hack"
 )
 
+// MariadbGTID represent mariadb gtid, [domain ID]-[server-id]-[sequence]
 type MariadbGTID struct {
 DomainID uint32
 ServerID uint32
 SequenceNumber uint64
 }
 
-// We don't support multi source replication, so the mariadb gtid set may have only domain-server-sequence
-func ParseMariadbGTIDSet(str string) (GTIDSet, error) {
+// ParseMariadbGTID parses mariadb gtid, [domain ID]-[server-id]-[sequence]
+func ParseMariadbGTID(str string) (*MariadbGTID, error) {
 if len(str) == 0 {
 return &MariadbGTID{0, 0, 0}, nil
 }
@@ -57,41 +60,158 @@ func (gtid *MariadbGTID) String() string {
 return fmt.Sprintf("%d-%d-%d", gtid.DomainID, gtid.ServerID, gtid.SequenceNumber)
 }
 
-func (gtid *MariadbGTID) Encode() []byte {
-return []byte(gtid.String())
+// Contain return whether one mariadb gtid covers another mariadb gtid
+func (gtid *MariadbGTID) Contain(other *MariadbGTID) bool {
+return gtid.DomainID == other.DomainID && gtid.SequenceNumber >= other.SequenceNumber
 }
 
-func (gtid *MariadbGTID) Equal(o GTIDSet) bool {
-other, ok := o.(*MariadbGTID)
-if !ok {
-return false
+// Clone clones a mariadb gtid
+func (gtid *MariadbGTID) Clone() *MariadbGTID {
+o := new(MariadbGTID)
+*o = *gtid
+return o
+}
+
+func (gtid *MariadbGTID) forward(newer *MariadbGTID) error {
+if newer.DomainID != gtid.DomainID {
+return errors.Errorf("%s is not same with doamin of %s", newer, gtid)
+}
+
+if newer.SequenceNumber <= gtid.SequenceNumber {
+return errors.Errorf("out of order binlog appears with gtid %s vs current position gtid %s", newer, gtid)
 }
 
-return *gtid == *other
+gtid.ServerID = newer.ServerID
+gtid.SequenceNumber = newer.SequenceNumber
+return nil
 }
 
-func (gtid *MariadbGTID) Contain(o GTIDSet) bool {
-other, ok := o.(*MariadbGTID)
-if !ok {
-return false
+// MariadbGTIDSet is a set of mariadb gtid
+type MariadbGTIDSet struct {
+Sets map[uint32]*MariadbGTID
+}
+
+// ParseMariadbGTIDSet parses str into mariadb gtid sets
+func ParseMariadbGTIDSet(str string) (GTIDSet, error) {
+s := new(MariadbGTIDSet)
+s.Sets = make(map[uint32]*MariadbGTID)
+if str == "" {
+return s, nil
 }
 
-return gtid.DomainID == other.DomainID && gtid.SequenceNumber >= other.SequenceNumber
+sp := strings.Split(str, ",")
+
+//todo, handle redundant same uuid
+for i := 0; i < len(sp); i++ {
+err := s.Update(sp[i])
+if err != nil {
+return nil, errors.Trace(err)
+}
+}
+return s, nil
 }
+// AddSet adds mariadb gtid into mariadb gtid set
+func (s *MariadbGTIDSet) AddSet(gtid *MariadbGTID) error {
+if gtid == nil {
+return nil
+}
 
-func (gtid *MariadbGTID) Update(GTIDStr string) error {
-newGTID, err := ParseMariadbGTIDSet(GTIDStr)
+o, ok := s.Sets[gtid.DomainID]
+if ok {
+err := o.forward(gtid)
+if err != nil {
+return errors.Trace(err)
+}
+} else {
+s.Sets[gtid.DomainID] = gtid
+}
+
+return nil
+}
+
+// Update updates mariadb gtid set
+func (s *MariadbGTIDSet) Update(GTIDStr string) error {
+gtid, err := ParseMariadbGTID(GTIDStr)
 if err != nil {
 return err
 }
 
-*gtid = *(newGTID.(*MariadbGTID))
+err = s.AddSet(gtid)
+return errors.Trace(err)
+}
+
 
-return nil
+func (s *MariadbGTIDSet) String() string {
+return hack.String(s.Encode())
 }
 
-func (gtid *MariadbGTID) Clone() GTIDSet {
-clone := new(MariadbGTID)
-*clone = *gtid
+// Encode encodes mariadb gtid set
+func (s *MariadbGTIDSet) Encode() []byte {
+var buf bytes.Buffer
+sep := ""
+for _, gtid := range s.Sets {
+buf.WriteString(sep)
+buf.WriteString(gtid.String())
+sep = ","
+}
+
+return buf.Bytes()
+}
+
+// Clone clones a mariadb gtid set
+func (s *MariadbGTIDSet) Clone() GTIDSet {
+clone := &MariadbGTIDSet{
+Sets: make(map[uint32]*MariadbGTID),
+}
+for domainID, gtid := range s.Sets {
+clone.Sets[domainID] = gtid.Clone()
+}
+
 return clone
 }
+
+// Equal returns true if two mariadb gtid set is same, otherwise return false
+func (s *MariadbGTIDSet) Equal(o GTIDSet) bool {
+other, ok := o.(*MariadbGTIDSet)
+if !ok {
+return false
+}
+
+if len(other.Sets) != len(s.Sets) {
+return false
+}
+
+for domainID, gtid := range other.Sets {
+o, ok := s.Sets[domainID]
+if !ok {
+return false
+}
+
+if *gtid != *o {
+return false
+}
+}
+
+return true
+}
+
+// Contain return whether one mariadb gtid set covers another mariadb gtid set
+func (s *MariadbGTIDSet) Contain(o GTIDSet) bool {
+other, ok := o.(*MariadbGTIDSet)
+if !ok {
+return false
+}
+
+for doaminID, gtid := range other.Sets {
+o, ok := s.Sets[doaminID]
+if !ok {
+return false
+}
+
+if !o.Contain(gtid) {
+return false
+}
+}
+
+return true
+}