sync-diff-inspector: support checkpoint

go.mod1

@@ -17,6 +17,7 @@ require (
 	github.com/pingcap/tidb v0.0.0-20190320062740-9071c7b5b9ed
 	github.com/pingcap/tipb v0.0.0-20190107072121-abbec73437b7
 	github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726
+	github.com/siddontang/go-mysql v0.0.0-20190312052122-c6ab05a85eb8
 	go.uber.org/atomic v1.3.2
 	go.uber.org/zap v1.9.1
 	golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e

go.sum1

@@ -144,10 +144,16 @@ github.com/prometheus/procfs v0.0.0-20180408092902-8b1c2da0d56d/go.mod h1:c3At6R
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/remyoudompheng/bigfft v0.0.0-20170806203942-52369c62f446/go.mod h1:uYEyJGbgTkfkS4+E/PavXkNJcbFIpEtjt2B0KDQ5+9M=
+github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/shirou/gopsutil v2.18.10+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
+github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
 github.com/shurcooL/httpfs v0.0.0-20171119174359-809beceb2371/go.mod h1:ZY1cvUeJuFPAdZ/B6v7RHavJWZn2YPVFQ1OSXhCGOkg=
 github.com/shurcooL/vfsgen v0.0.0-20181020040650-a97a25d856ca/go.mod h1:TrYk7fJVaAttu97ZZKrO9UbRa8izdowaMIZcxYMbVaw=
 github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726/go.mod h1:3yhqj7WBBfRhbBlzyOC3gUxftwsU0u8gqevxwIHQpMw=
+github.com/siddontang/go-log v0.0.0-20180807004314-8d05993dda07 h1:oI+RNwuC9jF2g2lP0u0cVEEZrc/AYBCuFdvwrLWM/6Q=
+github.com/siddontang/go-log v0.0.0-20180807004314-8d05993dda07/go.mod h1:yFdBgwXP24JziuRl2NMUahT7nGLNOKi1SIiFxMttVD4=
+github.com/siddontang/go-mysql v0.0.0-20190312052122-c6ab05a85eb8 h1:8puKTg/UOIQ+ZiowY1ywmGsI08sWqrKD7HJ/j165CUM=
+github.com/siddontang/go-mysql v0.0.0-20190312052122-c6ab05a85eb8/go.mod h1:/b8ZcWjAShCcHp2dWpjb1vTlNyiG03UeHEQr2jteOpI=
 github.com/sirupsen/logrus v0.0.0-20170323161349-3bcb09397d6d/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
 github.com/sirupsen/logrus v1.1.1/go.mod h1:zrgwTnHtNr00buQ1vSptGe8m1f/BbgsPukg8qsT7A+A=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=

pkg/dbutil/common.go

@@ -20,12 +20,19 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
+	"github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/parser/model"
+	tmysql "github.com/pingcap/parser/mysql"
+	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb-tools/pkg/utils"
+	"github.com/pingcap/tidb/ddl"
+	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/types"
+	gmysql "github.com/siddontang/go-mysql/mysql"
 	"go.uber.org/zap"
 )
 
@@ -35,6 +42,15 @@ const (
 
 	// ImplicitColID is ID implicit column in TiDB
 	ImplicitColID = -1
+
+	// DefaultRetryTime is the default retry time to execute sql
+	DefaultRetryTime = 10
+
+	// DefaultTimeout is the default timeout for execute sql
+	DefaultTimeout time.Duration = 5 * time.Second
+
+	// SlowWarnLog defines the duration to log warn log of sql when exec time greater than
+	SlowWarnLog = 100 * time.Millisecond
 )
 
 var (
@@ -617,3 +633,122 @@ func ReplacePlaceholder(str string, args []string) string {
 	newStr := strings.Replace(str, "?", "'%s'", -1)
 	return fmt.Sprintf(newStr, utils.StringsToInterfaces(args)...)
 }
+
+// ExecSQLWithRetry executes sql with retry
+func ExecSQLWithRetry(ctx context.Context, db *sql.DB, sql string, args ...interface{}) (err error) {
+	for i := 0; i < DefaultRetryTime; i++ {
+		startTime := time.Now()
+		_, err = db.ExecContext(ctx, sql, args...)
+		takeDuration := time.Since(startTime)
+		if takeDuration > SlowWarnLog {
+			log.Warn("exec sql slow", zap.String("sql", sql), zap.Reflect("args", args), zap.Duration("take", takeDuration))
+		}
+		if err == nil {
+			return nil
+		}
+
+		if ignoreError(err) {
+			log.Debug("ignore execute sql error", zap.Error(err))
+			return nil
+		}
+
+		if !isRetryableError(err) {
+			return errors.Trace(err)
+		}
+
+		log.Warn("exe sql failed, will try again", zap.String("sql", sql), zap.Reflect("args", args), zap.Error(err))
+
+		if i == DefaultRetryTime-1 {
+			break
+		}
+
+		select {
+		case <-ctx.Done():
+			return errors.Trace(ctx.Err())
+		case <-time.After(10 * time.Millisecond):
+		}
+	}
+
+	return errors.Trace(err)
+}
+
+// ExecuteSQLs executes some sqls in one transaction
+func ExecuteSQLs(ctx context.Context, db *sql.DB, sqls []string, args [][]interface{}) error {
+	txn, err := db.Begin()
+	if err != nil {
+		log.Error("exec sqls begin", zap.Error(err))
+		return errors.Trace(err)
+	}
+
+	for i := range sqls {
+		startTime := time.Now()
+
+		_, err = txn.ExecContext(ctx, sqls[i], args[i]...)
+		if err != nil {
+			log.Error("exec sql", zap.String("sql", sqls[i]), zap.Reflect("args", args[i]), zap.Error(err))
+			rerr := txn.Rollback()
+			if rerr != nil {
+				log.Error("rollback", zap.Error(err))
+			}
+			return errors.Trace(err)
+		}
+
+		takeDuration := time.Since(startTime)
+		if takeDuration > SlowWarnLog {
+			log.Warn("exec sql slow", zap.String("sql", sqls[i]), zap.Reflect("args", args[i]), zap.Duration("take", takeDuration))
+		}
+	}
+
+	err = txn.Commit()
+	if err != nil {
+		log.Error("exec sqls commit", zap.Error(err))
+		return errors.Trace(err)
+	}
+
+	return nil
+}
+
+func isRetryableError(err error) bool {
+	err = errors.Cause(err) // check the original error
+	mysqlErr, ok := err.(*mysql.MySQLError)
+	if !ok {
+		return false
+	}
+
+	switch mysqlErr.Number {
+	// ER_LOCK_DEADLOCK can retry to commit while meet deadlock
+	case tmysql.ErrUnknown, gmysql.ER_LOCK_DEADLOCK, tmysql.ErrPDServerTimeout, tmysql.ErrTiKVServerTimeout, tmysql.ErrTiKVServerBusy, tmysql.ErrResolveLockTimeout, tmysql.ErrRegionUnavailable:
+		return true
+	default:
+		return false
+	}
+}
+
+func ignoreError(err error) bool {
+	// TODO: now only ignore some ddl error, add some dml error later
+	if ignoreDDLError(err) {
+		return true
+	}
+
+	return false
+}
+
+func ignoreDDLError(err error) bool {
+	err = errors.Cause(err)
+	mysqlErr, ok := err.(*mysql.MySQLError)
+	if !ok {
+		return false
+	}
+
+	errCode := terror.ErrCode(mysqlErr.Number)
+	switch errCode {
+	case infoschema.ErrDatabaseExists.Code(), infoschema.ErrDatabaseDropExists.Code(),
+		infoschema.ErrTableExists.Code(), infoschema.ErrTableDropExists.Code(),
+		infoschema.ErrColumnExists.Code(), infoschema.ErrIndexExists.Code():
+		return true
+	case ddl.ErrDupKeyName.Code():
+		return true
+	default:
+		return false
+	}
+}

pkg/dbutil/common_test.go

@@ -14,7 +14,14 @@
 package dbutil
 
 import (
+	"database/sql/driver"
+
+	"github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
+	"github.com/pingcap/errors"
+	tmysql "github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tidb/infoschema"
+	gmysql "github.com/siddontang/go-mysql/mysql"
 )
 
 func (*testDBSuite) TestReplacePlaceholder(c *C) {
@@ -69,3 +76,56 @@ func (*testDBSuite) TestTableName(c *C) {
 		c.Assert(tableName, Equals, testCase.expectTableName)
 	}
 }
+
+func newMysqlErr(number uint16, message string) *mysql.MySQLError {
+	return &mysql.MySQLError{
+		Number:  number,
+		Message: message,
+	}
+}
+
+func (s *testDBSuite) TestIsRetryableError(c *C) {
+	cases := []struct {
+		err         error
+		isRetryable bool
+	}{
+		{newMysqlErr(tmysql.ErrNoDB, "no db error"), false},
+		{errors.New("unknown error"), false},
+		{newMysqlErr(tmysql.ErrUnknown, "i/o timeout"), true},
+		{newMysqlErr(tmysql.ErrDBCreateExists, "db already exists"), false},
+		{driver.ErrBadConn, false},
+		{newMysqlErr(gmysql.ER_LOCK_DEADLOCK, "Deadlock found when trying to get lock; try restarting transaction"), true},
+		{newMysqlErr(tmysql.ErrPDServerTimeout, "pd server timeout"), true},
+		{newMysqlErr(tmysql.ErrTiKVServerTimeout, "tikv server timeout"), true},
+		{newMysqlErr(tmysql.ErrTiKVServerBusy, "tikv server busy"), true},
+		{newMysqlErr(tmysql.ErrResolveLockTimeout, "resolve lock timeout"), true},
+		{newMysqlErr(tmysql.ErrRegionUnavailable, "region unavailable"), true},
+	}
+
+	for _, t := range cases {
+		c.Logf("err %v, expected %v", t.err, t.isRetryable)
+		c.Assert(isRetryableError(t.err), Equals, t.isRetryable)
+	}
+}
+
+func (s *testDBSuite) TestIsIgnoreError(c *C) {
+	cases := []struct {
+		err       error
+		canIgnore bool
+	}{
+		{newMysqlErr(uint16(infoschema.ErrDatabaseExists.Code()), "Can't create database, database exists"), true},
+		{newMysqlErr(uint16(infoschema.ErrDatabaseDropExists.Code()), "Can't drop database, database doesn't exists"), true},
+		{newMysqlErr(uint16(infoschema.ErrTableExists.Code()), "Can't create table, table exists"), true},
+		{newMysqlErr(uint16(infoschema.ErrTableDropExists.Code()), "Can't drop table, table dosen't exists"), true},
+		{newMysqlErr(uint16(infoschema.ErrColumnExists.Code()), "Duplicate column name"), true},
+		{newMysqlErr(uint16(infoschema.ErrIndexExists.Code()), "Duplicate Index"), true},
+
+		{newMysqlErr(uint16(999), "fake error"), false},
+		{errors.New("unknown error"), false},
+	}
+
+	for _, t := range cases {
+		c.Logf("err %v, expected %v", t.err, t.canIgnore)
+		c.Assert(ignoreError(t.err), Equals, t.canIgnore)
+	}
+}