Fixes and improvements for replication (go-mysql-org#342)

* Properly parsing empty binary values from mysqldump output

Author: destel

canal/canal.go

@@ -200,6 +200,8 @@ func (c *Canal) run() error {
 c.cancel()
 }()
 
+c.master.UpdateTimestamp(uint32(time.Now().Unix()))
+
 if !c.dumped {
 c.dumped = true
 
@@ -396,21 +398,9 @@ func (c *Canal) checkBinlogRowFormat() error {
 }
 
 func (c *Canal) prepareSyncer() error {
-seps := strings.Split(c.cfg.Addr, ":")
-if len(seps) != 2 {
-return errors.Errorf("invalid mysql addr format %s, must host:port", c.cfg.Addr)
-}
-
-port, err := strconv.ParseUint(seps[1], 10, 16)
-if err != nil {
-return errors.Trace(err)
-}
-
 cfg := replication.BinlogSyncerConfig{
 ServerID: c.cfg.ServerID,
 Flavor: c.cfg.Flavor,
-Host: seps[0],
-Port: uint16(port),
 User: c.cfg.User,
 Password: c.cfg.Password,
 Charset: c.cfg.Charset,
@@ -421,6 +411,23 @@ func (c *Canal) prepareSyncer() error {
 SemiSyncEnabled: c.cfg.SemiSyncEnabled,
 }
 
+if strings.Contains(c.cfg.Addr, "/") {
+cfg.Host = c.cfg.Addr
+} else {
+seps := strings.Split(c.cfg.Addr, ":")
+if len(seps) != 2 {
+return errors.Errorf("invalid mysql addr format %s, must host:port", c.cfg.Addr)
+}
+
+port, err := strconv.ParseUint(seps[1], 10, 16)
+if err != nil {
+return errors.Trace(err)
+}
+
+cfg.Host = seps[0]
+cfg.Port = uint16(port)
+}
+
 c.syncer = replication.NewBinlogSyncer(cfg)
 
 return nil
@@ -458,6 +465,10 @@ func (c *Canal) SyncedPosition() mysql.Position {
 return c.master.Position()
 }
 
+func (c *Canal) SyncedTimestamp() uint32 {
+return c.master.timestamp
+}
+
 func (c *Canal) SyncedGTIDSet() mysql.GTIDSet {
 return c.master.GTIDSet()
 }

canal/dump.go

@@ -10,7 +10,6 @@ import (
 "github.com/juju/errors"
 "github.com/shopspring/decimal"
 "github.com/siddontang/go-log/log"
-"github.com/siddontang/go-mysql/dump"
 "github.com/siddontang/go-mysql/mysql"
 "github.com/siddontang/go-mysql/schema"
 )
@@ -50,47 +49,43 @@ func (h *dumpParseHandler) Data(db string, table string, values []string) error
 for i, v := range values {
 if v == "NULL" {
 vs[i] = nil
+} else if v == "_binary ''" {
+vs[i] = []byte{}
 } else if v[0] != '\'' {
 if tableInfo.Columns[i].Type == schema.TYPE_NUMBER {
 n, err := strconv.ParseInt(v, 10, 64)
 if err != nil {
-log.Errorf("parse row %v at %d error %v, skip", values, i, err)
-return dump.ErrSkip
+return fmt.Errorf("parse row %v at %d error %v, int expected", values, i, err)
 }
 vs[i] = n
 } else if tableInfo.Columns[i].Type == schema.TYPE_FLOAT {
 f, err := strconv.ParseFloat(v, 64)
 if err != nil {
-log.Errorf("parse row %v at %d error %v, skip", values, i, err)
-return dump.ErrSkip
+return fmt.Errorf("parse row %v at %d error %v, float expected", values, i, err)
 }
 vs[i] = f
 } else if tableInfo.Columns[i].Type == schema.TYPE_DECIMAL {
 if h.c.cfg.UseDecimal {
 d, err := decimal.NewFromString(v)
 if err != nil {
-log.Errorf("parse row %v at %d error %v, skip", values, i, err)
-return dump.ErrSkip
+return fmt.Errorf("parse row %v at %d error %v, decimal expected", values, i, err)
 }
 vs[i] = d
 } else {
 f, err := strconv.ParseFloat(v, 64)
 if err != nil {
-log.Errorf("parse row %v at %d error %v, skip", values, i, err)
-return dump.ErrSkip
+return fmt.Errorf("parse row %v at %d error %v, float expected", values, i, err)
 }
 vs[i] = f
 }
 } else if strings.HasPrefix(v, "0x") {
 buf, err := hex.DecodeString(v[2:])
 if err != nil {
-log.Errorf("parse row %v at %d error %v, skip", values, i, err)
-return dump.ErrSkip
+return fmt.Errorf("parse row %v at %d error %v, hex literal expected", values, i, err)
 }
 vs[i] = string(buf)
 } else {
-log.Errorf("parse row %v error, invalid type at %d, skip", values, i)
-return dump.ErrSkip
+return fmt.Errorf("parse row %v error, invalid type at %d", values, i)
 }
 } else {
 vs[i] = v[1 : len(v)-1]
@@ -130,6 +125,8 @@ func (c *Canal) dump() error {
 return errors.New("mysqldump does not exist")
 }
 
+c.master.UpdateTimestamp(uint32(time.Now().Unix()))
+
 h := &dumpParseHandler{c: c}
 // If users call StartFromGTID with empty position to start dumping with gtid,
 // we record the current gtid position before dump starts.
@@ -161,7 +158,9 @@ func (c *Canal) dump() error {
 
 pos := mysql.Position{Name: h.name, Pos: uint32(h.pos)}
 c.master.Update(pos)
-c.eventHandler.OnPosSynced(pos, true)
+if err := c.eventHandler.OnPosSynced(pos, true); err != nil {
+return errors.Trace(err)
+}
 var startPos fmt.Stringer = pos
 if h.gset != nil {
 c.master.UpdateGTIDSet(h.gset)

canal/master.go

@@ -13,6 +13,8 @@ type masterInfo struct {
 pos mysql.Position
 
 gset mysql.GTIDSet
+
+timestamp uint32
 }
 
 func (m *masterInfo) Update(pos mysql.Position) {
@@ -23,6 +25,14 @@ func (m *masterInfo) Update(pos mysql.Position) {
 m.Unlock()
 }
 
+func (m *masterInfo) UpdateTimestamp(ts uint32) {
+log.Debugf("update master timestamp %s", ts)
+
+m.Lock()
+m.timestamp = ts
+m.Unlock()
+}
+
 func (m *masterInfo) UpdateGTIDSet(gset mysql.GTIDSet) {
 log.Debugf("update master gtid set %s", gset)
 
@@ -38,6 +48,13 @@ func (m *masterInfo) Position() mysql.Position {
 return m.pos
 }
 
+func (m *masterInfo) Timestamp() uint32 {
+m.RLock()
+defer m.RUnlock()
+
+return m.timestamp
+}
+
 func (m *masterInfo) GTIDSet() mysql.GTIDSet {
 m.RLock()
 defer m.RUnlock()

canal/sync.go

@@ -14,10 +14,11 @@ import (
 )
 
 var (
-expCreateTable = regexp.MustCompile("(?i)^CREATE\\sTABLE(\\sIF\\sNOT\\sEXISTS)?\\s`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s.*")
-expAlterTable = regexp.MustCompile("(?i)^ALTER\\sTABLE\\s.*?`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s.*")
-expRenameTable = regexp.MustCompile("(?i)^RENAME\\sTABLE\\s.*?`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s{1,}TO\\s.*?")
-expDropTable = regexp.MustCompile("(?i)^DROP\\sTABLE(\\sIF\\sEXISTS){0,1}\\s`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}(?:$|\\s)")
+expCreateTable = regexp.MustCompile("(?i)^CREATE\\sTABLE(\\sIF\\sNOT\\sEXISTS)?\\s`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s.*")
+expAlterTable = regexp.MustCompile("(?i)^ALTER\\sTABLE\\s.*?`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s.*")
+expRenameTable = regexp.MustCompile("(?i)^RENAME\\sTABLE\\s.*?`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s{1,}TO\\s.*?")
+expDropTable = regexp.MustCompile("(?i)^DROP\\sTABLE(\\sIF\\sEXISTS){0,1}\\s`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}(?:$|\\s)")
+expTruncateTable = regexp.MustCompile("(?i)^TRUNCATE\\s+(?:TABLE\\s+)?(?:`?([^`\\s]+)`?\\.`?)?([^`\\s]+)`?")
 )
 
 func (c *Canal) startSyncer() (*replication.BinlogStreamer, error) {
@@ -126,7 +127,7 @@ func (c *Canal) runSyncBinlog() error {
 db []byte
 table []byte
 )
-regexps := []regexp.Regexp{*expCreateTable, *expAlterTable, *expRenameTable, *expDropTable}
+regexps := []regexp.Regexp{*expCreateTable, *expAlterTable, *expRenameTable, *expDropTable, *expTruncateTable}
 for _, reg := range regexps {
 mb = reg.FindSubmatch(e.Query)
 if len(mb) != 0 {
@@ -164,7 +165,10 @@ func (c *Canal) runSyncBinlog() error {
 
 if savePos {
 c.master.Update(pos)
-c.eventHandler.OnPosSynced(pos, force)
+c.master.UpdateTimestamp(ev.Header.Timestamp)
+if err := c.eventHandler.OnPosSynced(pos, force); err != nil {
+return errors.Trace(err)
+}
 }
 }
 

dump/dump.go

@@ -121,10 +121,14 @@ func (d *Dumper) Dump(w io.Writer) error {
 args := make([]string, 0, 16)
 
 // Common args
-seps := strings.Split(d.Addr, ":")
-args = append(args, fmt.Sprintf("--host=%s", seps[0]))
-if len(seps) > 1 {
-args = append(args, fmt.Sprintf("--port=%s", seps[1]))
+if strings.Contains(d.Addr, "/") {
+args = append(args, fmt.Sprintf("--socket=%s", d.Addr))
+} else {
+seps := strings.SplitN(d.Addr, ":", 2)
+args = append(args, fmt.Sprintf("--host=%s", seps[0]))
+if len(seps) > 1 {
+args = append(args, fmt.Sprintf("--port=%s", seps[1]))
+}
 }
 
 args = append(args, fmt.Sprintf("--user=%s", d.User))

replication/binlogsyncer.go

@@ -7,6 +7,7 @@ import (
 "fmt"
 "net"
 "os"
+"strings"
 "sync"
 "time"
 
@@ -192,9 +193,16 @@ func (b *BinlogSyncer) registerSlave() error {
 b.c.Close()
 }
 
-log.Infof("register slave for master server %s:%d", b.cfg.Host, b.cfg.Port)
+addr := ""
+if strings.Contains(b.cfg.Host, "/") {
+addr = b.cfg.Host
+} else {
+addr = fmt.Sprintf("%s:%d", b.cfg.Host, b.cfg.Port)
+}
+
+log.Infof("register slave for master server %s", addr)
 var err error
-b.c, err = client.Connect(fmt.Sprintf("%s:%d", b.cfg.Host, b.cfg.Port), b.cfg.User, b.cfg.Password, "", func(c *client.Conn) {
+b.c, err = client.Connect(addr, b.cfg.User, b.cfg.Password, "", func(c *client.Conn) {
 c.TLSConfig = b.cfg.TLSConfig
 })
 if err != nil {