package oltppkg

import (
	"bufio"
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"path/filepath"
	"regexp"
	"runtime"
	"strconv"
	"strings"
	"time"

	gk "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

type byteRow struct {
	data [][]byte
}
type byteRows struct {
	cols []string
	data []byteRow
}

type ConfigRow struct {
	Type     string
	Instance string
	Name     string
	Value    string
}

func StrToInt(s string) int {
	return Try(strconv.Atoi(s)).(int)
}
func LogAndBy(lb string) {
	logWithTime := fmt.Sprintf("%s %s", time.Now().Format("2006/01/02 15:04:05"), lb)
	log.Info(lb, zap.String("callBy", PrintCallerName()))
	gk.By(logWithTime)
}
func SleepWithCtx(ctx context.Context, timeout time.Duration) {
	select {
	case <-time.After(timeout):
		return
	case <-ctx.Done():
		log.Info("ctx done,cancel the sleep")
		return
	}
}
func WaitFor(ctx context.Context, reason string, duration time.Duration) {
	log.Info(reason, zap.Duration("wait", duration), zap.String("callBy", PrintCallerName()))
	SleepWithCtx(ctx, duration)
}

func DbExecContextWithLog(ctx context.Context, db *sql.DB, query string) error {
	log.Info(query)
	_, err := db.ExecContext(ctx, query)
	return err
}

func ShowClusterVersions(tc resource.TiDBCluster) {
	for _, comp := range []resource.TiDBCompType{resource.TiKV, resource.PD, resource.TiDB} {
		ShowCompReleaseInfo(tc, comp)
	}
}
func ShowCompReleaseInfo(tc resource.TiDBCluster, comp resource.TiDBCompType) {
	var server string
	switch comp {
	case resource.TiKV:
		server = "tikv-server"
	case resource.PD:
		server = "pd-server"
	case resource.TiDB:
		server = "tidb-server"
	default:
		err := errors.New("invalid server")
		Expect(err).Should(Succeed())
	}
	cmd := fmt.Sprintf("./%s -V", server)
	stdout, stderr, exitCode, err := tc.Exec(resource.TiDBClusterExecOptions{Target: comp, Command: cmd})
	if stdout != "" {
		fmt.Printf("%v\n %v\n", cmd, stdout)
	}
	if err != nil {
		log.Info(fmt.Sprintf("stdout:%s\n stderr:%s\n exitCode:%d\n", stdout, stderr, exitCode))
	}
}

func ShowBrReleaseInfo(br resource.BR) {
	cmd := "./br -V"
	stdout, stderr, exitCode, err := br.Exec(resource.WorkloadNodeExecOptions{Command: cmd})
	if stdout != "" {
		fmt.Printf("%v\n %v\n", cmd, stdout)
	}
	if err != nil {
		log.Info(fmt.Sprintf("stdout:%s\n stderr:%s\n exitCode:%d\n", stdout, stderr, exitCode))
	}
}

func GetAllTablesNames(ctx context.Context, tc resource.MySQLCompatible, dbName string) []string {
	var tableName string
	var tables []string
	db := Try(tc.GetDB(dbName)).(*sql.DB)
	defer db.Close()
	rows, err := db.Query("show tables;")
	Expect(err).Should(Succeed())
	for rows.Next() {
		rows.Scan(&tableName)
		tables = append(tables, tableName)
	}
	defer rows.Close()
	log.Info("tables info", zap.Strings("tables", tables), zap.Int("tableNum", len(tables)))
	return tables
}

func CancelDDLTaskById(ctx context.Context, tc resource.MySQLCompatible, jobId string) {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	var jobID string
	cmd := fmt.Sprintf("select job_id from information_schema.ddl_jobs where STATE='running' and JOB_ID=%s;", jobID)
	Try(db.QueryRow(cmd).Scan(&jobID))
	Try(DbExecContextWithLog(ctx, db, fmt.Sprintf("admin cancel ddl jobs %s;", jobID)))
}

func CancelAllDDLTasks(ctx context.Context, tc resource.MySQLCompatible) {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	var jobID string
	jobIDRows, err := db.Query("select job_id from information_schema.ddl_jobs where STATE='running';")
	Expect(err).Should(Succeed())
	for jobIDRows.Next() {
		jobIDRows.Scan(&jobID)
		Try(DbExecContextWithLog(ctx, db, fmt.Sprintf("admin cancel ddl jobs %s;", jobID)))
	}
	defer jobIDRows.Close()
}

func GetInstanceNum(tc resource.MySQLCompatible, compType string) uint32 {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	tidbClient := tidb.NewClient(db)
	num := Try(tidbClient.GetInstanceNum(compType)).(uint32)
	return num
}

func OutPutCheck(ctx context.Context, pr *io.PipeReader, duration time.Duration, ifCheck bool, skipList []string, errThreshHold int) error {

	log.Info("Begin to check the output.")
	defer func() {
		log.Info("Out put check finished.")
		//pr.Close()
	}()
	r := bufio.NewReader(pr)
	timer := time.NewTimer(duration)
	errOccurTimes := 0
Loop:
	for {
		select {
		case <-ctx.Done():
			log.Info("Check output context done.")
			break Loop
		case <-timer.C:
			log.Info("Check output timeout.")
			break Loop
		default:
		}
		line, err := r.ReadString('\n')
		if err != nil {
			if err == io.EOF {
				break
			}
			return err
		}
		newline := strings.Replace(line, "\n", "", -1)
		if newline != "" {
			log.Info(line)
		}
		if ifCheck {
			if strings.Contains(strings.ToLower(line), strings.ToLower("error")) ||
				//strings.Contains(strings.ToLower(line), strings.ToLower("fail")) ||
				strings.Contains(strings.ToLower(line), strings.ToLower("fatal")) {
				if len(skipList) == 0 {
					if errOccurTimes < errThreshHold {
						log.Info("err occur ", zap.Int("errOccurTimes", errOccurTimes))
						errOccurTimes++
					} else {
						log.Error("can't accept this error,and error throw more than threshhold")
						return fmt.Errorf("can't accept this error")
					}

				} else {
					match := false
					for _, c := range skipList {
						if strings.Contains(strings.ToLower(line), strings.ToLower(c)) {
							log.Warn("Skip ", zap.String("msg", c))
							match = true
							break
						}
					}
					if !match {
						if errOccurTimes < errThreshHold {
							log.Info("err occur ", zap.Int("errOccurTimes", errOccurTimes))
							errOccurTimes++
						} else {
							log.Error("can't accept this error,and error throw more than threshhold")
							return fmt.Errorf("can't accept this error")
						}
					}
				}
			}
		}
	}
	return nil
}

func GetResourceTypeByString(comType string) resource.TiDBCompType {
	var target resource.TiDBCompType
	if comType == "tikv" {
		target = resource.TiKV
	} else if comType == "tidb" {
		target = resource.TiDB
	} else if comType == "pd" {
		target = resource.PD
	}
	return target
}

func GetCompVersion(ctx context.Context, tc resource.MySQLCompatible, comType string) string {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	var count int
	var version string
	Try(db.QueryRow(fmt.Sprintf(`select count(distinct(version)) from information_schema.cluster_info where type="%s"`, comType)).Scan(&count))
	if count != 1 {
		log.Warn("Instance version is not consistent.")
		rows, _ := db.QueryContext(ctx, fmt.Sprintf(`select instance,version from information_schema.cluster_info where type="%s"`, comType))
		defer rows.Close()
		for rows.Next() {
			var version string
			var instance string
			Try(rows.Scan(&instance, &version))
			log.Info("Inconsistent version detail", zap.String(instance, version))
		}
		return "Invalid"
	}
	Try(db.QueryRow(fmt.Sprintf(`select distinct(version) from information_schema.cluster_info where type="%s"`, comType)).Scan(&version))
	log.Info("Current version", zap.String(comType, version))
	return version
}
func CheckComponentVersion(ctx context.Context, tc resource.TiDBCluster, targetVersion string, comType string) {
	shortVersion := targetVersion
	shortVersion = strings.TrimPrefix(targetVersion, "v")
	if strings.Contains(targetVersion, "-") {
		shortVersion = strings.Split(shortVersion, "-")[0]
	}
	curVersion := GetCompVersion(ctx, tc, comType)
	log.Info("Check version", zap.String("current version", curVersion), zap.String("target version", shortVersion))

	//Maybe hit issue https://github.com/pingcap/tidb/issues/30907
	//Expect(strings.Contains(curVersion, shortVersion)).Should(BeTrue())
	if !strings.Contains(curVersion, shortVersion) {
		log.Warn("Maybe hit issue https://github.com/pingcap/tidb/issues/30907")
	}

}
func CheckClusterVersion(ctx context.Context, tc resource.TiDBCluster, targetVersion string, masterMap map[string]string) {
	for _, comType := range []string{"pd", "tidb", "tikv"} {
		if targetVersion == "master" {
			CheckComponentVersion(ctx, tc, masterMap[comType], comType)
		} else {
			CheckComponentVersion(ctx, tc, targetVersion, comType)
		}

	}
}

func RunInsertAndDelete(ctx context.Context, tc resource.TiDBCluster, sysBench resource.WorkloadNode) {
	dbURL, _ := tc.ServiceURL(resource.DBAddr)
Loop:
	for {
		select {
		case <-ctx.Done():
			break Loop
		default:
		}
		// Catch the panic throw by create database or run workload during chaos inject.
		defer func() {
			r := recover()
			if _, ok := r.(error); ok {
				log.Warn("Recovered")
			}
		}()
		RecreateDatabase(ctx, tc, "todrop")
		sysbench := NewSysbench("oltp_read_write", "prepare")
		sysbench.Db("todrop").Host(dbURL.Hostname()).Port(dbURL.Port()).TableSize(100000).Threads(100).Tables(10)
		ExecWorkloadCmd(ctx, sysBench, sysbench.Build(), true, nil)
		SleepWithCtx(ctx, 10*time.Second)
	}
}

func DumpToByteRows(rows *sql.Rows) (*byteRows, error) {
	cols, err := rows.Columns()
	if err != nil {
		return nil, err
	}

	data := make([]byteRow, 0, 8)
	args := make([]interface{}, len(cols))
	for rows.Next() {
		tmp := make([][]byte, len(cols))
		for i := 0; i < len(args); i++ {
			args[i] = &tmp[i]
		}
		err := rows.Scan(args...)
		if err != nil {
			return nil, err
		}

		data = append(data, byteRow{tmp})
	}
	err = rows.Err()
	if err != nil {
		return nil, err
	}

	return &byteRows{cols: cols, data: data}, nil
}

func (rows *byteRows) ConvertToString(str string) string {
	switch str {
	case "vertical":
		var buf bytes.Buffer
		for i, row := range rows.data {
			buf.WriteString("# ROW:" + strconv.Itoa(i+1) + "\n")
			for j, col := range rows.cols {
				buf.WriteString(col)
				buf.WriteByte('\t')
				buf.Write(row.data[j])
				buf.WriteByte('\n')
			}
		}
		return buf.String()
	default:
		if rows == nil || rows.cols == nil {
			log.Info(fmt.Sprintf("Something is wrong %v", rows))
			return ""
		}
		res := strings.Join(rows.cols, "\t")
		for _, row := range rows.data {
			line := ""
			for _, data := range row.data {
				col := string(data)
				if data == nil {
					col = "NULL"
				}
				if len(line) > 0 {
					line = line + "\t"
				}
				line = line + col
			}
			res = res + "\n" + line
		}
		return res + "\n"
	}
}

func GetTableAnalyzeInfoOnce(ctx context.Context, tc resource.MySQLCompatible, dbName string, tableName string) {
	db := Try(tc.GetDB(dbName)).(*sql.DB)
	defer db.Close()
	beginTime := time.Now()
	rows, err := db.Query(fmt.Sprintf("explain analyze select count(*) from %s", tableName))
	if err != nil {
		return
	}
	timeCost := time.Since(beginTime)
	log.Info("Explain analyze take: ", zap.String("tableName", tableName), zap.Duration("timeCost", timeCost))
	var (
		id             string
		estRows        float64
		actRows        float64
		task           string
		access_object  string
		execution_info string
		operator_info  string
		memory         string
		disk           string
	)
	i := 0
	var total float64
	var skip float64

	for rows.Next() {
		i++
		if err := rows.Scan(&id, &estRows, &actRows, &task, &access_object, &execution_info, &operator_info, &memory, &disk); err != nil {
			return
		}
		if strings.Contains(id, "StreamAgg_") && i == 3 {
			log.Info(execution_info)
			explainRegexp := regexp.MustCompile(`.*total_process_keys:(.*?),.*key_skipped_count:(.*?),.*`)
			params := explainRegexp.FindStringSubmatch(execution_info)
			if len(params) == 3 {
				total, err = strconv.ParseFloat(strings.TrimSpace(params[1]), 64)
				Expect(err).Should(BeNil())
				skip, err = strconv.ParseFloat(strings.TrimSpace(params[2]), 64)
				Expect(err).Should(BeNil())
				log.Info(fmt.Sprintf("skip/total ratio: %f", skip/total))
			} else {
				log.Info("no scan detail this time.")
			}
		}
	}
}
func GetTableAnalyzeInfo(ctx context.Context, tc resource.MySQLCompatible, dbName string, tableName string) {
Loop:
	for {
		select {
		case <-ctx.Done():
			break Loop
		default:
		}
		GetTableAnalyzeInfoOnce(ctx, tc, dbName, tableName)
		time.Sleep(10 * time.Minute)
	}
}

func GetRandDuration(unit time.Duration, scope int) time.Duration {
	rand.Seed(time.Now().UnixNano())
	randValue := rand.Intn(scope) + 1
	return time.Duration(randValue) * unit
}
func GetRandInt(min int, max int) int {
	rand.Seed(time.Now().UnixNano())
	randInt := rand.Intn(max-min) + min
	return randInt
}
func GetRandNum(min interface{}, max interface{}) int64 {
	var min64 int64
	var max64 int64
	if value, ok := min.(int64); ok {
		min64 = value
	}
	if value, ok := max.(int64); ok {
		max64 = value
	}
	if value, ok := min.(int); ok {
		min64 = int64(value)
	}
	if value, ok := max.(int); ok {
		max64 = int64(value)
	}
	rand.Seed(time.Now().UnixNano())
	if max64 <= min64 {
		return 0
	}
	randInt64 := rand.Int63n(max64-min64) + min64
	return randInt64
}

func GetRandIntStr(min int, max int) string {
	rand.Seed(time.Now().UnixNano())
	randInt := rand.Intn(max-min) + min
	return strconv.Itoa(randInt)
}

func GetRandInstance(tc resource.TiDBCluster, comp resource.TiDBCompType) (int, resource.TiDBComponentNode) {
	nodes := Try(tc.GetNodes(comp)).([]resource.TiDBComponentNode)
	nodeNum := len(nodes)
	rand.Seed(time.Now().UnixNano())
	instanceId := rand.Intn(nodeNum)
	return instanceId, nodes[instanceId]
}

func GetInstanceIdList(tc resource.TiDBCluster, comp resource.TiDBCompType) []int {
	nodes := Try(tc.GetNodes(comp)).([]resource.TiDBComponentNode)
	nodeNum := len(nodes)
	var instanceIdList []int
	for i := 0; i < nodeNum; i++ {
		instanceIdList = append(instanceIdList, i)
	}
	return instanceIdList
}

type StrSlice []string

func (list StrSlice) Has(a string) bool {
	for _, b := range list {
		if b == a {
			return true
		}
	}
	return false
}
func FindMinAndMax(a []float64) (min float64, max float64) {
	min = a[0]
	max = a[0]
	for _, value := range a {
		if value < min {
			min = value
		}
		if value > max {
			max = value
		}
	}
	return min, max
}
func CheckRegionCountBalance(ctx context.Context, tc resource.MySQLCompatible, dbName string, tableName string) {
	sqlTmp := `select count(s.region_id) cnt,index_name, p.store_id 
	from INFORMATION_SCHEMA.TIKV_REGION_STATUS s join INFORMATION_SCHEMA.tikv_region_peers p on s.region_id = p.region_id 
	where s.table_name = '%s' and p.is_leader = 0 group by index_name, p.store_id order by index_name,cnt desc;`
	db := Try(tc.GetDB(dbName)).(*sql.DB)
	defer db.Close()
	rows := Try(db.Query(fmt.Sprintf(sqlTmp, tableName))).(*sql.Rows)
	defer rows.Close()
	var (
		region_count float64
		index_name   sql.NullString
		store_id     float64
	)
	var regionCountList []float64
	for rows.Next() {
		Try(rows.Scan(&region_count, &index_name, &store_id))
		if index_name.Valid {
			log.Info("Query Result", zap.Float64("region_count", region_count), zap.String("index_name", index_name.String), zap.Float64("store_id", store_id))
		} else {
			log.Info("Query Result", zap.Float64("region_count", region_count), zap.String("index_name", "NULL"), zap.Float64("store_id", store_id))
			regionCountList = append(regionCountList, region_count)
		}
	}
	min, max := FindMinAndMax(regionCountList)
	if max-min >= 15 {
		log.Info("Warning: Max more than Min 15", zap.Float64("max", max), zap.Float64("min", min))
	}
	//Expect(max - min).Should(BeNumerically("<=", 15))
}

//GetDatabaseSize get table size by MB
func GetDatabaseSize(ctx context.Context, tc resource.TiDBCluster, dbName string) float64 {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	sqlTmp := `select
    concat(round(sum(DATA_LENGTH / 1024 / 1024), 2), 'MB') as data
from
    information_schema.TABLES
WHERE
    table_schema='%s';`
	rows := Try(db.Query(fmt.Sprintf(sqlTmp, dbName))).(*sql.Rows)
	var size string
	for rows.Next() {
		Try(rows.Scan(&size))
		log.Info("Query Result", zap.String("size", size))
	}
	return Try(strconv.ParseFloat(strings.TrimSuffix(size, "MB"), 64)).(float64)
}

func GetRandomWorkloadFromList(useRandomWorkload string) (string, error) {
	var workloadList []string
	randList := strings.Split(useRandomWorkload, ",")
	if len(randList) <= 1 {
		workloadType := GetRandomWorkloadType()
		workloadList = []string{workloadType}
	} else {
		workloadList = randList[1:]
		for _, workload := range workloadList {
			if !StrSlice(GetWorkloadTypeList(nil)).Has(workload) {
				log.Info("Invalid workload", zap.String("workload", workload))
				log.Info(fmt.Sprintf("Support workload list %v", GetWorkloadTypeList(nil)))
				return "", fmt.Errorf("invalid workload")
			}
		}
	}
	return workloadList[GetRandInt(0, len(workloadList))], nil
}

//CompareVersion return -1 if v1 < v 2, return 1 if v1 > v2, return 0 if v1==v2
func CompareVersion(version1 string, version2 string) int {
	var res int
	ver1Strs := strings.Split(version1, ".")
	ver2Strs := strings.Split(version2, ".")
	ver1Len := len(ver1Strs)
	ver2Len := len(ver2Strs)
	verLen := ver1Len
	if len(ver1Strs) < len(ver2Strs) {
		verLen = ver2Len
	}
	for i := 0; i < verLen; i++ {
		var ver1Int, ver2Int int
		if i < ver1Len {
			ver1Int, _ = strconv.Atoi(ver1Strs[i])
		}
		if i < ver2Len {
			ver2Int, _ = strconv.Atoi(ver2Strs[i])
		}
		if ver1Int < ver2Int {
			res = -1
			break
		}
		if ver1Int > ver2Int {
			res = 1
			break
		}
	}
	return res
}

//CompareVersionWithCurrent return 1 if current version higher, return -1 if lower, return 0 if equal
func CompareVersionWithCurrent(ctx context.Context, tc resource.TiDBCluster, versionToCompare string) int {
	tidbVersion := GetCompVersion(ctx, tc, "tidb")
	versionShort := strings.TrimSuffix(strings.TrimPrefix(tidbVersion, "v"), "-pre")
	return CompareVersion(versionShort, versionToCompare)
}
func GetTableRows(ctx context.Context, tc resource.MySQLCompatible, dbName string, tableName string) (int64, error) {
	db := Try(tc.GetDB(dbName)).(*sql.DB)
	defer db.Close()
	var rowCount string
	rows := db.QueryRow(fmt.Sprintf("select count(*) from %s", tableName))
	err := rows.Scan(&rowCount)
	if err != nil {
		return 0, err
	}
	count := Try(strconv.ParseInt(rowCount, 10, 64)).(int64)
	return count, nil
}
func GetRowsCount(ctx context.Context, tc resource.MySQLCompatible, dbName string, querySql string) (int64, error) {
	db := Try(tc.GetDB(dbName)).(*sql.DB)
	defer db.Close()
	rows, err := db.QueryContext(ctx, querySql)
	if err != nil {
		return 0, err
	}
	columns, err := rows.Columns()
	if err != nil {
		return 0, err
	}
	count := len(columns)

	values := make([]interface{}, count)
	valuePtr := make([]interface{}, count)

	var rowCount int64
	for rows.Next() {
		rowCount = rowCount + 1
		for i, _ := range columns {
			valuePtr[i] = &values[i]
		}
		err := rows.Scan(valuePtr...)
		if err != nil {
			return 0, err
		}
	}
	return rowCount, nil
}

func GetConfigByName(tc resource.MySQLCompatible, instanceType string, configName string) []ConfigRow {
	var configRow ConfigRow
	var configRows []ConfigRow
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	getConfigCmd := fmt.Sprintf("show config where type = '%s' and name = '%s'", instanceType, configName)
	log.Info(getConfigCmd)
	rows, err := db.Query(getConfigCmd)
	Expect(err).Should(Succeed())
	for rows.Next() {
		err := rows.Scan(&configRow.Type, &configRow.Instance, &configRow.Name, &configRow.Value)
		Expect(err).Should(Succeed())
		log.Info("config ", zap.String("type", configRow.Type), zap.String("instance", configRow.Instance), zap.String("name", configRow.Name), zap.String("Value", configRow.Value))
		configRows = append(configRows, configRow)
	}
	defer rows.Close()
	return configRows
}

func SetConfigOnline(tc resource.MySQLCompatible, instanceType string, configName string, value string) {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	setConfigCmd := fmt.Sprintf("set config %s `%s`='%s';", instanceType, configName, value)
	log.Info(setConfigCmd)
	_, err := db.Query(setConfigCmd)
	Expect(err).Should(Succeed())
}

func SetAndVerifyConfigOnline(tc resource.MySQLCompatible, instanceType string, configName string, value string) {
	SetConfigOnline(tc, instanceType, configName, value)
	configRows := GetConfigByName(tc, instanceType, configName)
	for _, configRow := range configRows {
		currentValue := configRow.Value
		Expect(currentValue).To(Equal(value))
	}
}

func RandomlyRestartInstance(tc resource.TiDBCluster, instanceType resource.TiDBCompType) (int, error) {
	instanceId, instance := GetRandInstance(tc, instanceType)
	log.Info(fmt.Sprintf("begin to restart %s_%d.", instanceType.String(), instanceId))
	err := instance.Restart()
	if err != nil {
		return 0, err
	}
	return instanceId, nil
}
func SplitTablesRepeatly(ctx context.Context, tc resource.MySQLCompatible, dbName string, splitInterval time.Duration) {
Loop:
	for {
		select {
		case <-ctx.Done():
			break Loop
		default:
		}
		tables := GetAllTablesNames(ctx, tc, dbName)
		db := Try(tc.GetDB(dbName)).(*sql.DB)
		defer db.Close()
		for _, tableName := range tables {
			select {
			case <-ctx.Done():
				break Loop
			default:
			}
			regionNum, err := GetRowsCount(ctx, tc, dbName, fmt.Sprintf("show table %s regions", tableName))
			if err != nil {
				break Loop
			}
			tableRows, err := GetTableRows(ctx, tc, dbName, tableName)
			if err != nil {
				break Loop
			}
			if tableRows < 100*1000 {
				log.Info("The table is too small to split.", zap.String("tableName", tableName))
				break
			}
			splitRange := GetRandNum(1000*100, tableRows)
			splitStart := GetRandNum(0, tableRows-splitRange)
			expectRegion := regionNum * splitRange / tableRows

			targetRegion := int64(float64(expectRegion) * 2)
			if targetRegion <= 20 {
				targetRegion = 20
			}
			if targetRegion > splitRange/1000 {
				targetRegion = splitRange/1000 - 1
			}
			splitSql := fmt.Sprintf("SPLIT TABLE %s BETWEEN (%d) AND (%d) REGIONS %d", tableName, splitStart, splitStart+splitRange, targetRegion)
			if targetRegion <= 1000 {
				log.Info("execute sql", zap.String("splitSql", splitSql))
				_, err := db.ExecContext(ctx, splitSql)
				if err != nil {
					log.Info("Split table failed.")
					log.Info(err.Error())
				}
			} else {
				WaitFor(ctx, "wait for merge", splitInterval)
			}
		}
		WaitFor(ctx, "wait for next loop to split table", splitInterval)
	}
}

func GetTikvLables(ctx context.Context, tc resource.TiDBCluster) (string, error) {
	db := Try(tc.GetDB("")).(*sql.DB)
	rows, err := db.QueryContext(ctx, "select LABEL from INFORMATION_SCHEMA.TIKV_STORE_STATUS;")
	if err != nil {
		return "", err
	}
	var labelList []string
	for rows.Next() {
		var label string
		err := rows.Scan(&label)
		if err != nil {
			return "", err
		}
		labelList = append(labelList, label)
	}
	return strings.Join(labelList, ","), nil
}

func PrintFuncName() string {
	pc, _, _, _ := runtime.Caller(1)
	return runtime.FuncForPC(pc).Name()
}
func PrintCallerName() string {
	_, file, line, _ := runtime.Caller(2)
	_, fileName := filepath.Split(file)
	return fmt.Sprintf("%s:%d", fileName, line)
	//return runtime.FuncForPC(pc).Name()
}

func RollingUpdateConfig(tc resource.TiDBCluster, compType resource.TiDBCompType, config string, timeout time.Duration) {
	log.Info("Start to update tc config!!!", zap.Any("option", config))
	start := time.Now()
	err := tc.UpdateConfig(resource.TiDBClusterUpdateConfigOptions{
		Target:  compType,
		Config:  config,
		Timeout: timeout,
	})
	log.Info("update tc config done", zap.Any("option", config), zap.Duration("elapsed", time.Since(start)))
	Expect(err).Should(Succeed())
}

func SetTidbVariables(ctx context.Context, tc resource.MySQLCompatible, caseCmdParms CaseCmdParms) {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	if caseCmdParms.TidbVariables.GlobalVariables != nil {
		for _, variable := range caseCmdParms.TidbVariables.GlobalVariables {
			v := fmt.Sprintf("set global %s", variable)
			log.Info(v)
			_, err := db.ExecContext(ctx, v)
			if err != nil {
				log.Info(err.Error())
			}
		}
	}
}

func GetTidbVariable(ctx context.Context, tc resource.MySQLCompatible, variable string) (string, error) {
	db := Try(tc.GetDB("")).(*sql.DB)
	defer db.Close()
	sql := fmt.Sprintf("select @@global.%s", variable)
	log.Info(sql)
	rows, err := db.QueryContext(ctx, sql)
	if err != nil {
		return "", err
	}
	defer rows.Close()
	var result string
	for rows.Next() {
		err := rows.Scan(&result)
		if err != nil {
			return "", err
		}
		log.Info(fmt.Sprintf("%s: %s", variable, result))
	}
	return result, nil
}