func ExecSQLWithRetry(ctx context.Context, db DBExecutor, 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 > SlowLogThreshold {
			log.Debug("exec sql slow", zap.String("sql", sql), zap.Reflect("args", args), zap.Duration("take", takeDuration))
		}
		if err == nil {
			return nil
		}
		if ignoreError(err) {
			log.Warn("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)
}
ExecSQLWithRetry executes sql with retry
package dbutil
import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"
	"github.com/go-sql-driver/mysql"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb-tools/pkg/utils"
	"github.com/pingcap/tidb/ddl"
	"github.com/pingcap/tidb/infoschema"
	"github.com/pingcap/tidb/parser"
	"github.com/pingcap/tidb/parser/model"
	tmysql "github.com/pingcap/tidb/parser/mysql"
	"github.com/pingcap/tidb/sessionctx/stmtctx"
	"github.com/pingcap/tidb/types"
	"go.uber.org/zap"
)
const (
	DefaultRetryTime = 10
	DefaultTimeout time.Duration = 10 * time.Second
	SlowLogThreshold = 200 * time.Millisecond
	DefaultDeleteRowsNum int64 = 100000
)
var (
	ErrVersionNotFound = errors.New("can't get the database's version")
	ErrNoData = errors.New("no data found in table")
)
type DBConfig struct {
	Host string `toml:"host" json:"host"`
	Port int `toml:"port" json:"port"`
	User string `toml:"user" json:"user"`
	Password string `toml:"password" json:"-"`  
	Schema string `toml:"schema" json:"schema"`
	Snapshot string `toml:"snapshot" json:"snapshot"`
}
func (c *DBConfig) String() string {
	cfg, err := json.Marshal(c)
	if err != nil {
		return "<nil>"
	}
	return string(cfg)
}
func GetDBConfigFromEnv(schema string) DBConfig {
	host := os.Getenv("MYSQL_HOST")
	if host == "" {
		host = "127.0.0.1"
	}
	port, _ := strconv.Atoi(os.Getenv("MYSQL_PORT"))
	if port == 0 {
		port = 3306
	}
	user := os.Getenv("MYSQL_USER")
	if user == "" {
		user = "root"
	}
	pswd := os.Getenv("MYSQL_PSWD")
	return DBConfig{
		Host:     host,
		Port:     port,
		User:     user,
		Password: pswd,
		Schema:   schema,
	}
}
func OpenDB(cfg DBConfig, vars map[string]string) (*sql.DB, error) {
	var dbDSN string
	if len(cfg.Snapshot) != 0 {
		log.Info("create connection with snapshot", zap.String("snapshot", cfg.Snapshot))
		dbDSN = fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4&tidb_snapshot=%s", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Snapshot)
	} else {
		dbDSN = fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8mb4", cfg.User, cfg.Password, cfg.Host, cfg.Port)
	}
	for key, val := range vars {
		dbDSN += fmt.Sprintf("&%s=%%27%s%%27", key, url.QueryEscape(val))
	}
	dbConn, err := sql.Open("mysql", dbDSN)
	if err != nil {
		return nil, errors.Trace(err)
	}
	err = dbConn.Ping()
	return dbConn, errors.Trace(err)
}
func CloseDB(db *sql.DB) error {
	if db == nil {
		return nil
	}
	return errors.Trace(db.Close())
}
func GetCreateTableSQL(ctx context.Context, db QueryExecutor, schemaName string, tableName string) (string, error) {
	query := fmt.Sprintf("SHOW CREATE TABLE %s", TableName(schemaName, tableName))
	var tbl, createTable sql.NullString
	err := db.QueryRowContext(ctx, query).Scan(&tbl, &createTable)
	if err != nil {
		return "", errors.Trace(err)
	}
	if !tbl.Valid || !createTable.Valid {
		return "", errors.NotFoundf("table %s", tableName)
	}
	return createTable.String, nil
}
func GetRowCount(ctx context.Context, db QueryExecutor, schemaName string, tableName string, where string, args []interface{}) (int64, error) {
	query := fmt.Sprintf("SELECT COUNT(1) cnt FROM %s", TableName(schemaName, tableName))
	if len(where) > 0 {
		query += fmt.Sprintf(" WHERE %s", where)
	}
	log.Debug("get row count", zap.String("sql", query), zap.Reflect("args", args))
	var cnt sql.NullInt64
	err := db.QueryRowContext(ctx, query, args...).Scan(&cnt)
	if err != nil {
		return 0, errors.Trace(err)
	}
	if !cnt.Valid {
		return 0, errors.NotFoundf("table `%s`.`%s`", schemaName, tableName)
	}
	return cnt.Int64, nil
}
func GetRandomValues(ctx context.Context, db QueryExecutor, schemaName, table, column string, num int, limitRange string, limitArgs []interface{}, collation string) ([]string, error) {
	if limitRange == "" {
		limitRange = "TRUE"
	}
	if collation != "" {
		collation = fmt.Sprintf(" COLLATE \"%s\"", collation)
	}
	query := fmt.Sprintf("SELECT %[1]s FROM (SELECT %[1]s, rand() rand_value FROM %[2]s WHERE %[3]s ORDER BY rand_value LIMIT %[4]d)rand_tmp ORDER BY %[1]s%[5]s",
		ColumnName(column), TableName(schemaName, table), limitRange, num, collation)
	log.Debug("get random values", zap.String("sql", query), zap.Reflect("args", limitArgs))
	rows, err := db.QueryContext(ctx, query, limitArgs...)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()
	randomValue := make([]string, 0, num)
	for rows.Next() {
		var value sql.NullString
		err = rows.Scan(&value)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if value.Valid {
			randomValue = append(randomValue, value.String)
		}
	}
	return randomValue, errors.Trace(rows.Err())
}
func GetMinMaxValue(ctx context.Context, db QueryExecutor, schema, table, column string, limitRange string, limitArgs []interface{}, collation string) (string, string, error) {
	if limitRange == "" {
		limitRange = "TRUE"
	}
	if collation != "" {
		collation = fmt.Sprintf(" COLLATE \"%s\"", collation)
	}
	query := fmt.Sprintf("SELECT /*!40001 SQL_NO_CACHE */ MIN(%s%s) as MIN, MAX(%s%s) as MAX FROM %s WHERE %s",
		ColumnName(column), collation, ColumnName(column), collation, TableName(schema, table), limitRange)
	log.Debug("GetMinMaxValue", zap.String("sql", query), zap.Reflect("args", limitArgs))
	var min, max sql.NullString
	rows, err := db.QueryContext(ctx, query, limitArgs...)
	if err != nil {
		return "", "", errors.Trace(err)
	}
	defer rows.Close()
	for rows.Next() {
		err = rows.Scan(&min, &max)
		if err != nil {
			return "", "", errors.Trace(err)
		}
	}
	if !min.Valid || !max.Valid {
		return "", "", ErrNoData
	}
	return min.String, max.String, errors.Trace(rows.Err())
}
func GetTimeZoneOffset(ctx context.Context, db QueryExecutor) (time.Duration, error) {
	var timeStr string
	err := db.QueryRowContext(ctx, "SELECT cast(TIMEDIFF(NOW(6), UTC_TIMESTAMP(6)) as time);").Scan(&timeStr)
	if err != nil {
		return 0, errors.Trace(err)
	}
	factor := time.Duration(1)
	if timeStr[0] == '-' || timeStr[0] == '+' {
		if timeStr[0] == '-' {
			factor *= -1
		}
		timeStr = timeStr[1:]
	}
	t, err := time.Parse("15:04:05", timeStr)
	if err != nil {
		return 0, errors.Trace(err)
	}
	if t.IsZero() {
		return 0, nil
	}
	hour, minute, second := t.Clock()
	d := time.Duration(hour*3600+minute*60+second) * time.Second * factor
	return d, nil
}
func FormatTimeZoneOffset(offset time.Duration) string {
	prefix := "+"
	if offset < 0 {
		prefix = "-"
		offset *= -1
	}
	hours := offset / time.Hour
	minutes := (offset % time.Hour) / time.Minute
	return fmt.Sprintf("%s%02d:%02d", prefix, hours, minutes)
}
func queryTables(ctx context.Context, db QueryExecutor, q string) (tables []string, err error) {
	log.Debug("query tables", zap.String("query", q))
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()
	tables = make([]string, 0, 8)
	for rows.Next() {
		var table, tType sql.NullString
		err = rows.Scan(&table, &tType)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !table.Valid || !tType.Valid {
			continue
		}
		tables = append(tables, table.String)
	}
	return tables, errors.Trace(rows.Err())
}
func GetTables(ctx context.Context, db QueryExecutor, schemaName string) (tables []string, err error) {
	query := fmt.Sprintf("SHOW FULL TABLES IN `%s` WHERE Table_Type != 'VIEW';", escapeName(schemaName))
	return queryTables(ctx, db, query)
}
func GetViews(ctx context.Context, db QueryExecutor, schemaName string) (tables []string, err error) {
	query := fmt.Sprintf("SHOW FULL TABLES IN `%s` WHERE Table_Type = 'VIEW';", escapeName(schemaName))
	return queryTables(ctx, db, query)
}
func GetSchemas(ctx context.Context, db QueryExecutor) ([]string, error) {
	query := "SHOW DATABASES"
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()
	schemas := make([]string, 0, 10)
	for rows.Next() {
		var schema string
		err = rows.Scan(&schema)
		if err != nil {
			return nil, errors.Trace(err)
		}
		schemas = append(schemas, schema)
	}
	return schemas, errors.Trace(rows.Err())
}
func GetCRC32Checksum(ctx context.Context, db QueryExecutor, schemaName, tableName string, tbInfo *model.TableInfo, limitRange string, args []interface{}) (int64, error) {
	columnNames := make([]string, 0, len(tbInfo.Columns))
	columnIsNull := make([]string, 0, len(tbInfo.Columns))
	for _, col := range tbInfo.Columns {
		columnNames = append(columnNames, ColumnName(col.Name.O))
		columnIsNull = append(columnIsNull, fmt.Sprintf("ISNULL(%s)", ColumnName(col.Name.O)))
	}
	query := fmt.Sprintf("SELECT BIT_XOR(CAST(CRC32(CONCAT_WS(',', %s, CONCAT(%s)))AS UNSIGNED)) AS checksum FROM %s WHERE %s;",
		strings.Join(columnNames, ", "), strings.Join(columnIsNull, ", "), TableName(schemaName, tableName), limitRange)
	log.Debug("checksum", zap.String("sql", query), zap.Reflect("args", args))
	var checksum sql.NullInt64
	err := db.QueryRowContext(ctx, query, args...).Scan(&checksum)
	if err != nil {
		return -1, errors.Trace(err)
	}
	if !checksum.Valid {
		log.Warn("get empty checksum", zap.String("sql", query), zap.Reflect("args", args))
		return 0, nil
	}
	return checksum.Int64, nil
}
type Bucket struct {
	Count      int64
	LowerBound string
	UpperBound string
}
func GetBucketsInfo(ctx context.Context, db QueryExecutor, schema, table string, tableInfo *model.TableInfo) (map[string][]Bucket, error) {
	buckets := make(map[string][]Bucket)
	query := "SHOW STATS_BUCKETS WHERE db_name= ? AND table_name= ?;"
	log.Debug("GetBucketsInfo", zap.String("sql", query), zap.String("schema", schema), zap.String("table", table))
	rows, err := db.QueryContext(ctx, query, schema, table)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()
	cols, err := rows.Columns()
	if err != nil {
		return nil, errors.Trace(err)
	}
	for rows.Next() {
		var dbName, tableName, partitionName, columnName, lowerBound, upperBound sql.NullString
		var isIndex, bucketID, count, repeats, ndv sql.NullInt64
		switch len(cols) {
		case 9:
			err = rows.Scan(&dbName, &tableName, &columnName, &isIndex, &bucketID, &count, &repeats, &lowerBound, &upperBound)
		case 10:
			err = rows.Scan(&dbName, &tableName, &partitionName, &columnName, &isIndex, &bucketID, &count, &repeats, &lowerBound, &upperBound)
		case 11:
			err = rows.Scan(&dbName, &tableName, &partitionName, &columnName, &isIndex, &bucketID, &count, &repeats, &lowerBound, &upperBound, &ndv)
		default:
			return nil, errors.New("Unknown struct for buckets info")
		}
		if err != nil {
			return nil, errors.Trace(err)
		}
		if _, ok := buckets[columnName.String]; !ok {
			buckets[columnName.String] = make([]Bucket, 0, 100)
		}
		buckets[columnName.String] = append(buckets[columnName.String], Bucket{
			Count:      count.Int64,
			LowerBound: lowerBound.String,
			UpperBound: upperBound.String,
		})
	}
	indices := FindAllIndex(tableInfo)
	for _, index := range indices {
		if index.Name.O != "PRIMARY" {
			continue
		}
		_, ok := buckets[index.Name.O]
		if !ok && len(index.Columns) == 1 {
			if _, ok := buckets[index.Columns[0].Name.O]; !ok {
				return nil, errors.NotFoundf("primary key on %s in buckets info", index.Columns[0].Name.O)
			}
			buckets[index.Name.O] = buckets[index.Columns[0].Name.O]
			delete(buckets, index.Columns[0].Name.O)
		}
	}
	return buckets, errors.Trace(rows.Err())
}
func AnalyzeValuesFromBuckets(valueString string, cols []*model.ColumnInfo) ([]string, error) {
	vStr := strings.Trim(valueString, "()")
	values := strings.Split(vStr, ", ")
	if len(values) != len(cols) {
		return nil, errors.Errorf("analyze value %s failed", valueString)
	}
	for i, col := range cols {
		if IsTimeTypeAndNeedDecode(col.Tp) {
			sc := &stmtctx.StatementContext{TimeZone: time.UTC}
			_, err := types.ParseTime(sc, values[i], col.Tp, types.MinFsp)
			if err == nil {
				continue
			}
			value, err := DecodeTimeInBucket(values[i])
			if err != nil {
				log.Error("analyze values from buckets", zap.String("column", col.Name.O), zap.String("value", values[i]), zap.Error(err))
				return nil, errors.Trace(err)
			}
			values[i] = value
		}
	}
	return values, nil
}
func DecodeTimeInBucket(packedStr string) (string, error) {
	packed, err := strconv.ParseUint(packedStr, 10, 64)
	if err != nil {
		return "", err
	}
	if packed == 0 {
		return "", nil
	}
	t := new(types.Time)
	err = t.FromPackedUint(packed)
	if err != nil {
		return "", err
	}
	return t.String(), nil
}
func GetTidbLatestTSO(ctx context.Context, db QueryExecutor) (int64, error) {
	rows, err := db.QueryContext(ctx, "SHOW MASTER STATUS")
	if err != nil {
		return 0, errors.Trace(err)
	}
	defer rows.Close()
	for rows.Next() {
		fields, err1 := ScanRow(rows)
		if err1 != nil {
			return 0, errors.Trace(err1)
		}
		ts, err1 := strconv.ParseInt(string(fields["Position"].Data), 10, 64)
		if err1 != nil {
			return 0, errors.Trace(err1)
		}
		return ts, nil
	}
	return 0, errors.New("get secondary cluster's ts failed")
}
func GetDBVersion(ctx context.Context, db QueryExecutor) (string, error) {
	query := "SELECT version()"
	result, err := db.QueryContext(ctx, query)
	if err != nil {
		return "", errors.Trace(err)
	}
	defer result.Close()
	var version sql.NullString
	for result.Next() {
		err := result.Scan(&version)
		if err != nil {
			return "", errors.Trace(err)
		}
		break
	}
	if version.Valid {
		return version.String, nil
	}
	return "", ErrVersionNotFound
}
func GetSessionVariable(ctx context.Context, db QueryExecutor, variable string) (value string, err error) {
	query := fmt.Sprintf("SHOW VARIABLES LIKE '%s'", variable)
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		return "", errors.Trace(err)
	}
	defer rows.Close()
	for rows.Next() {
		err = rows.Scan(&variable, &value)
		if err != nil {
			return "", errors.Trace(err)
		}
	}
	if rows.Err() != nil {
		return "", errors.Trace(err)
	}
	return value, nil
}
func GetSQLMode(ctx context.Context, db QueryExecutor) (tmysql.SQLMode, error) {
	sqlMode, err := GetSessionVariable(ctx, db, "sql_mode")
	if err != nil {
		return tmysql.ModeNone, err
	}
	mode, err := tmysql.GetSQLMode(sqlMode)
	return mode, errors.Trace(err)
}
func IsTiDB(ctx context.Context, db QueryExecutor) (bool, error) {
	version, err := GetDBVersion(ctx, db)
	if err != nil {
		log.Error("get database's version failed", zap.Error(err))
		return false, errors.Trace(err)
	}
	return strings.Contains(strings.ToLower(version), "tidb"), nil
}
func TableName(schema, table string) string {
	return fmt.Sprintf("`%s`.`%s`", escapeName(schema), escapeName(table))
}
func ColumnName(column string) string {
	return fmt.Sprintf("`%s`", escapeName(column))
}
func escapeName(name string) string {
	return strings.Replace(name, "`", "``", -1)
}
func ReplacePlaceholder(str string, args []string) string {
	newStr := strings.Replace(str, "?", "'%s'", -1)
	return fmt.Sprintf(newStr, utils.StringsToInterfaces(args)...)
}