package pkg

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"math/rand"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"sync"
	"time"

	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"

	"github.com/pingcap/endless/pkg/client/pd"
	pdclient "github.com/pingcap/endless/pkg/client/pd"
)

type table struct {
	ReplicaCount int
	Dropped      bool
	RuleCount    int
	// Rule count for each table partition, will increase when truncated.
	PartitionRuleCount *map[int]int
	// Is true when the table partition is drooped.
	PartitionDropped *map[int]bool
}

type Tables struct {
	sync.Mutex

	Ts      map[int]*table
	Replica int
}

type PDClient struct {
	*pd.Client
}

type PeerRoleType string

type Constraints []Constraint
type Constraint struct {
	Key    string       `json:"key,omitempty"`
	Op     ConstraintOp `json:"op,omitempty"`
	Values []string     `json:"values,omitempty"`
}
type ConstraintOp string

// TiFlashRule extends Rule with other necessary fields.
type TiFlashRule struct {
	GroupID        string       `json:"group_id"`
	ID             string       `json:"id"`
	Index          int          `json:"index,omitempty"`
	Override       bool         `json:"override,omitempty"`
	StartKeyHex    string       `json:"start_key"`
	EndKeyHex      string       `json:"end_key"`
	Role           PeerRoleType `json:"role"`
	Count          int          `json:"count"`
	Constraints    Constraints  `json:"label_constraints,omitempty"`
	LocationLabels []string     `json:"location_labels,omitempty"`
	IsolationLevel string       `json:"isolation_level,omitempty"`
}

type ExistTablesInfo struct {
	sync.Mutex
	MaxPartition map[string]int
}

func (t *ExistTablesInfo) updateInfo(ts map[int]*table) {
	for k, v := range ts {
		name := "t" + strconv.Itoa(k)
		if v.Dropped {
			if _, ok := t.MaxPartition[name]; ok {
				delete(t.MaxPartition, name)
			}
		} else {
			if v.PartitionDropped != nil {
				t.MaxPartition[name] = 0
				for ii, vv := range *v.PartitionDropped {
					if vv == false {
						t.MaxPartition[name] = max(t.MaxPartition[name], ii*10)
					}
				}
			} else {
				t.MaxPartition[name] = intMax
			}
		}
	}
}

// getGroupRules to get all placement rule in a certain group.
func (h *PDClient) getGroupRules(group string) ([]TiFlashRule, error) {
	path := fmt.Sprintf("/pd/api/v1/config/rules/group/%s", group)
	// log.Info("initiate get request", zap.String("path", path))
	resp, err := h.Client.Get(path)
	if err != nil {
		return nil, errors.New("fail get")
	}

	if resp.StatusCode != http.StatusOK {
		return nil, errors.New("getGroupRules returns error")
	}

	buf := new(bytes.Buffer)
	_, err = buf.ReadFrom(resp.Body)
	if err != nil {
		return nil, errors.New("fail read")
	}

	defer func() {
		if err = resp.Body.Close(); err != nil {
		}
	}()
	var rules []TiFlashRule
	err = json.Unmarshal(buf.Bytes(), &rules)
	if err != nil {
		return nil, errors.New("fail parse")
	}

	return rules, nil
}

func (h *PDClient) getGroupRulesCount() (int, error) {
	rules, err := h.getGroupRules("tiflash")
	return len(rules), err
}

func (t *Tables) gather() int {
	rules := 0
	for _, v := range t.Ts {
		if v.PartitionRuleCount != nil {
			for _, vv := range *v.PartitionRuleCount {
				rules += vv
			}
		} else {
			rules += v.RuleCount
		}
	}
	return rules
}

func (t *Tables) check(delta int) bool {
	expected := t.gather()
	log.Info(fmt.Sprintf("actual %v expected %v\n", delta, expected))
	return delta >= expected
}

func (t *Tables) logTiflashRules(pd *PDClient) error {
	// expectNum := t.gather()
	// actualNum, err := pd.getGroupRulesCount()
	// log.Info("PD Rules", zap.Int("expectNum", expectNum), zap.Int("actualNum", actualNum))
	return nil
}

func waitAllTableOK(db *sql.DB, dbn string, tableAvailableTime int, tag string, noReplica int) bool {
	tick := 0
	log.Info("check weather all table ok")
	for {
		select {
		case <-time.After(1 * time.Second):
			var x int
			s := fmt.Sprintf("SELECT count(*) FROM information_schema.tiflash_replica where progress = 0 and table_schema = '%v';", dbn)
			row := db.QueryRow(s)
			if err := row.Scan(&x); err != nil {
				panic(err)
			}
			tick += 1
			if x == 0 {
				log.Info(fmt.Sprintf("OK Check db %v tag %v retry %v noReplica %v count %v", dbn, tag, tick, noReplica, x))
				return true
			}
			if tick >= tableAvailableTime {
				log.Info(fmt.Sprintf("Fail db %v count %v tag %v retry %v noReplica %v", dbn, x, tag, tick, noReplica))
				return false
			}
		}
	}
}

func addPartitionHelper(tb *table) int {
	nn := len(*(tb.PartitionRuleCount))
	(*(tb.PartitionRuleCount))[nn] = 1
	(*(tb.PartitionDropped))[nn] = false
	return nn
}

func addGivenPartitionHelper(tb *table, i int) {
	(*(tb.PartitionRuleCount))[i] = 1
	(*(tb.PartitionDropped))[i] = false
}

func (t *Tables) addTable(pd *PDClient, db *sql.DB, partition bool, setReplica bool, tableInfo *ExistTablesInfo) error {
	t.Lock()
	tableInfo.Lock()
	defer func() {
		t.Unlock()
		tableInfo.Unlock()
	}()
	n := len(t.Ts)
	if partition {
		m := make(map[int]int)
		m[0] = 1
		m2 := make(map[int]bool)
		m2[0] = false
		t.Ts[n] = &table{
			Dropped:            false,
			RuleCount:          0,
			PartitionRuleCount: &m,
			PartitionDropped:   &m2,
			ReplicaCount:       0,
		}
		ss := []string{fmt.Sprintf("create table test98.t%v (z int) partition by range (z) (partition p0 values less than (0))", n)}
		if setReplica {
			t.Ts[n].ReplicaCount = t.Replica
			addGivenPartitionHelper(t.Ts[n], 0)
			ss = append(ss, fmt.Sprintf("alter table test98.t%v set tiflash replica %v", n, t.Replica))
		}
		for _, e := range ss {
			log.Info("ddl - add table", zap.String("sql", e))
			_, err := db.Exec(e)
			if err != nil {
				return err
			}
		}
		tableInfo.updateInfo(t.Ts)
		err := t.logTiflashRules(pd)
		return err
	} else {
		t.Ts[n] = &table{
			Dropped:            false,
			RuleCount:          0,
			PartitionRuleCount: nil,
			PartitionDropped:   nil,
			ReplicaCount:       0,
		}
		ss := []string{fmt.Sprintf("create table test98.t%v (z int)", n)}
		if setReplica {
			t.Ts[n].ReplicaCount = t.Replica
			t.Ts[n].RuleCount = 1
			ss = append(ss, fmt.Sprintf("alter table test98.t%v set tiflash replica %v", n, t.Replica))
		}
		for _, e := range ss {
			log.Info("ddl - add table", zap.String("sql", e))
			_, err := db.Exec(e)
			if err != nil {
				return err
			}
		}
		tableInfo.updateInfo(t.Ts)
		err := t.logTiflashRules(pd)
		return err
	}
}

func (t *Tables) addPartition(pd *PDClient, db *sql.DB, tableInfo *ExistTablesInfo) error {
	t.Lock()
	tableInfo.Lock()
	defer func() {
		t.Unlock()
		tableInfo.Unlock()
	}()
	for k, v := range t.Ts {
		if v.PartitionRuleCount != nil && !v.Dropped && v.ReplicaCount != 0 {
			n := len(*(v.PartitionRuleCount))
			(*(v.PartitionRuleCount))[n] = 1
			(*(v.PartitionDropped))[n] = false
			s := fmt.Sprintf("alter table test98.t%v add partition (partition p%v values less than (%v))", k, n, n*10)
			log.Info("ddl - add partition", zap.String("sql", s))
			_, err := db.Exec(s)
			tableInfo.updateInfo(t.Ts)
			if err != nil {
				return err
			}
			err = t.logTiflashRules(pd)
			return err
		}
	}
	return nil
}

func (t *Tables) truncatePartition(pd *PDClient, db *sql.DB, tableInfo *ExistTablesInfo) error {
	t.Lock()
	tableInfo.Lock()
	defer func() {
		t.Unlock()
		tableInfo.Unlock()
	}()
	for k, v := range t.Ts {
		if v.PartitionRuleCount != nil && !v.Dropped && v.ReplicaCount != 0 {
			for kk, vv := range *v.PartitionDropped {
				if !vv {
					(*v.PartitionRuleCount)[kk] += 1
					s := fmt.Sprintf("alter table test98.t%v truncate partition p%v", k, kk)
					tableInfo.updateInfo(t.Ts)
					log.Info("ddl - truncate partition", zap.String("sql", s))
					_, err := db.Exec(s)
					if err != nil {
						return err
					}
					err = t.logTiflashRules(pd)
					return err
				}
			}
		}
	}
	return nil
}

func (t *Tables) truncateTable(pd *PDClient, db *sql.DB, tableInfo *ExistTablesInfo) error {
	t.Lock()
	tableInfo.Lock()
	defer func() {
		t.Unlock()
		tableInfo.Unlock()
	}()
	for k, v := range t.Ts {
		if !v.Dropped && v.ReplicaCount != 0 {
			if v.PartitionRuleCount != nil {
				for kk, vv := range *v.PartitionDropped {
					if !vv {
						(*v.PartitionRuleCount)[kk] += 1
					}
				}
			} else {
				v.RuleCount += 1
			}
			s := fmt.Sprintf("truncate table test98.t%v", k)
			tableInfo.updateInfo(t.Ts)
			log.Info("ddl - truncate table", zap.String("sql", s))
			_, err := db.Exec(s)
			err = t.logTiflashRules(pd)
			return err
		}
	}
	return nil
}

func (t *Tables) setTiFlashReplica(pd *PDClient, db *sql.DB) error {
	t.Lock()
	defer t.Unlock()
	for k, v := range t.Ts {
		if !v.Dropped && v.ReplicaCount == 0 {
			t.Ts[k].ReplicaCount = t.Replica
			if v.PartitionRuleCount != nil {
				for pk, _ := range *(v.PartitionRuleCount) {
					if !(*(v.PartitionDropped))[pk] {
						(*(v.PartitionRuleCount))[pk] += 1
					}
				}
			} else {
				v.RuleCount += 1
			}
			s := fmt.Sprintf("alter table test98.t%v set tiflash replica %v", k, t.Replica)
			log.Info("ddl - set TiFlash replica", zap.String("sql", s))
			_, err := db.Exec(s)
			if err != nil {
				return err
			}
			err = t.logTiflashRules(pd)
			return err
		}
	}
	return nil
}

func (t *Tables) removeTiFlashReplica(pd *PDClient, db *sql.DB) error {
	t.Lock()
	defer t.Unlock()
	for k, v := range t.Ts {
		if !v.Dropped && v.ReplicaCount != 0 {
			t.Ts[k].ReplicaCount = 0
			if v.PartitionRuleCount != nil {
				for pk, _ := range *(v.PartitionRuleCount) {
					if !(*(v.PartitionDropped))[pk] {
						(*(v.PartitionRuleCount))[pk] -= 1
					}
				}
			} else {
				v.RuleCount -= 1
			}
			s := fmt.Sprintf("alter table test98.t%v set tiflash replica 0", k)
			log.Info("ddl - remove TiFlash replica", zap.String("sql", s))
			_, err := db.Exec(s)
			if err != nil {
				return err
			}
			err = t.logTiflashRules(pd)
			return err
		}
	}
	return nil
}

func (t *Tables) dropPartition(pd *PDClient, db *sql.DB, tableInfo *ExistTablesInfo) error {
	t.Lock()
	tableInfo.Lock()
	defer func() {
		t.Unlock()
		tableInfo.Unlock()
	}()
	for k, v := range t.Ts {
		if v.PartitionRuleCount != nil && !v.Dropped {
			for kk, vv := range *v.PartitionDropped {
				if !vv {
					(*v.PartitionDropped)[kk] = true
					s := fmt.Sprintf("alter table test98.t%v drop partition p%v", k, kk)
					log.Info("ddl - drop Partition", zap.String("sql", s))
					tableInfo.updateInfo(t.Ts)
					// Cannot remove all partitions, use DROP TABLE instead
					if _, err := db.Exec(s); err != nil {
						(*v.PartitionDropped)[kk] = false
					}
					err := t.logTiflashRules(pd)
					return err
				}
			}
		}
	}
	return nil
}

func (t *Tables) dropTable(pd *PDClient, db *sql.DB, tableInfo *ExistTablesInfo) error {
	t.Lock()
	tableInfo.Lock()
	defer func() {
		t.Unlock()
		tableInfo.Unlock()
	}()
	for k, v := range t.Ts {
		if !v.Dropped {
			t.Ts[k].Dropped = true
			s := fmt.Sprintf("drop table test98.t%v", k)
			tableInfo.updateInfo(t.Ts)
			log.Info("ddl - drop table", zap.String("sql", s))
			_, err := db.Exec(s)
			if err != nil {
				return err
			}
			err = t.logTiflashRules(pd)
			return err
		}
	}
	return nil
}

func (t *Tables) alterDatabaseSetReplica(db *sql.DB, Replica int, DBName string) error {
	t.Lock()
	defer t.Unlock()

	for _, v := range t.Ts {
		if !v.Dropped {
			if v.ReplicaCount == 0 {
				if v.PartitionRuleCount != nil {
					for pk, _ := range *(v.PartitionRuleCount) {
						if !(*(v.PartitionDropped))[pk] {
							(*(v.PartitionRuleCount))[pk] += 1
						}
					}
				} else {
					v.RuleCount += 1
				}
			}
			v.ReplicaCount = Replica
		}
	}

	s := fmt.Sprintf("alter database %v set tiflash replica %v", DBName, Replica)
	log.Info("ddl - alter database det replica", zap.String("sql", s))
	_, err := db.Exec(s)
	return err
}

func (t *Tables) removeDatabaseSetReplica(db *sql.DB, DBName string) error {
	t.Lock()
	defer t.Unlock()

	for _, v := range t.Ts {
		if !v.Dropped {
			if v.ReplicaCount != 0 {
				if v.PartitionRuleCount != nil {
					// If has partition
					for pk, _ := range *(v.PartitionRuleCount) {
						if !(*(v.PartitionDropped))[pk] {
							// Maybe we have already truncated
							(*(v.PartitionRuleCount))[pk] -= 1
						}
					}
				} else {
					// Maybe we have already truncated
					v.RuleCount -= 1
				}
			}
			v.ReplicaCount = 0
		}
	}

	s := fmt.Sprintf("alter database %v set tiflash replica %v", DBName, 0)
	_, err := db.Exec(s)
	return err
}

func (t *Tables) flashbackTable(pd *PDClient, db *sql.DB, tableInfo *ExistTablesInfo) error {
	t.Lock()
	tableInfo.Lock()
	defer func() {
		t.Unlock()
		tableInfo.Unlock()
	}()
	for k, v := range t.Ts {
		if v.Dropped {
			t.Ts[k].Dropped = false
			s := fmt.Sprintf("flashback table test98.t%v", k)
			log.Info("ddl - flashback table", zap.String("sql", s))
			_, err := db.Exec(s)
			if err != nil {
				return err
			}
			tableInfo.updateInfo(t.Ts)
			err = t.logTiflashRules(pd)
			return err
		}
	}
	return nil
}

func changeGCSafePoint(db *sql.DB, t time.Time, enable string, lifeTime string) error {
	log.Info("ddl - Change GC Safe Point")
	gcTimeFormat := "20060102-15:04:05 -0700 MST"
	lastSafePoint := t.Format(gcTimeFormat)
	s := `INSERT HIGH_PRIORITY INTO mysql.tidb VALUES ('tikv_gc_safe_point', '%[1]s', '') ON DUPLICATE KEY UPDATE variable_value = '%[1]s'`

	s = fmt.Sprintf(s, lastSafePoint)
	log.Info("ddl -  exec sql", zap.String("sql", s))
	_, err := db.Exec(s)
	if err != nil {
		return err
	}

	s = `INSERT HIGH_PRIORITY INTO mysql.tidb VALUES ('tikv_gc_enable','%[1]s','') ON DUPLICATE KEY UPDATE variable_value = '%[1]s'`
	s = fmt.Sprintf(s, enable)
	log.Info("ddl - exec sql", zap.String("sql", s))
	_, err = db.Exec(s)
	if err != nil {
		return err
	}

	s = `INSERT HIGH_PRIORITY INTO mysql.tidb VALUES ('tikv_gc_life_time','%[1]s','') ON DUPLICATE KEY UPDATE variable_value = '%[1]s'`
	s = fmt.Sprintf(s, lifeTime)
	log.Info("ddl - exec sql", zap.String("sql", s))
	_, err = db.Exec(s)
	if err != nil {
		return err
	}
	return nil
}

func newPDClient(tc resource.TiDBCluster) (*PDClient, error) {
	log.Info("New PDClient")
	pdClient, err := pdclient.NewPDClient(tc)
	return &PDClient{pdClient}, err
}

func (h *PDClient) clearAllRules(group string) error {
	log.Info("Clear All PD Rules")
	rules, err := h.getGroupRules("tiflash")
	if err != nil {
		return err
	}
	for _, r := range rules {
		err := h.deletePlacementRule(group, r.ID)
		if err != nil {
			return err
		}
	}
	return nil
}

func (h *PDClient) deletePlacementRule(group string, ruleID string) error {
	deleteURL := fmt.Sprintf("%v/pd/api/v1/config/rule/%v/%v", h.Client.GetPDAddr(), url.QueryEscape(group), url.QueryEscape(ruleID))
	log.Info("Delete placement rule by http request", zap.String("url", deleteURL))
	err := h.Client.Delete(deleteURL)
	return err
}

func createDatabase(db *sql.DB) error {
	log.Info("Create Database")
	var cmds = []string{"drop database if exists test98", "create database test98", "drop database if exists test97", "create database test97"}
	for _, cmd := range cmds {
		log.Info("db exec sql", zap.String("sql", cmd))
		_, err := db.Exec(cmd)
		if err != nil {
			return err
		}
	}
	return nil
}

// TestMultiSessionRandomDDL random ddl exec
func TestMultiSessionRandomDDL(ctx context.Context, tc resource.TiDBCluster, goroutineNum int, replica int, randomTimes int, tablesInfo *ExistTablesInfo, tableAvailableTime int) error {
	// Need configure - store - limit
	// https://docs.pingcap.com/zh/tidb/stable/configure - store - limit/
	var err error
	var db, db2 *sql.DB
	db, err = tc.GetDB("")
	defer db.Close()
	if err != nil {
		return err
	}

	err = createDatabase(db)
	if err != nil {
		return err
	}
	err = changeGCSafePoint(db, time.Now().Add(0-24*time.Hour), "false", "1000m")
	if err != nil {
		return err
	}
	time.Sleep(2 * time.Second)

	pdClient, err := newPDClient(tc)
	if err != nil {
		return err
	}

	err = pdClient.clearAllRules("tiflash")
	if err != nil {
		return err
	}

	origin, err := pdClient.getGroupRulesCount()
	if err != nil {
		return err
	}

	tables := Tables{
		Ts:      make(map[int]*table),
		Replica: replica,
	}

	var wg sync.WaitGroup
	errChan := make(chan error, goroutineNum)
	for t := 0; t < goroutineNum; t++ {
		wg.Add(1)
		y := t
		go func(index int) {
			defer func() {
				wg.Done()
			}()
			db2, err = tc.GetDB("")
			if err != nil {
				errChan <- err
				return
			}
			err = tables.addTable(pdClient, db2, false, true, tablesInfo) // 1
			if err != nil {
				errChan <- err
				return
			}
			err = tables.addTable(pdClient, db2, true, true, tablesInfo) // 1
			if err != nil {
				errChan <- err
				return
			}
			err = tables.addPartition(pdClient, db2, tablesInfo) // 1
			if err != nil {
				errChan <- err
				return
			}

			for i := 0; i < randomTimes; i++ {
				select {
				case <-ctx.Done():
					return
				default:
				}
				// in := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11}
				in := []int{-1, -2, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11}
				randomIndex := rand.Intn(len(in))
				pick := in[randomIndex]
				if pick == 0 {
					err = tables.addTable(pdClient, db2, false, true, tablesInfo)
				} else if pick == 1 {
					err = tables.addTable(pdClient, db2, true, true, tablesInfo)
				} else if pick == 2 {
					err = tables.addPartition(pdClient, db2, tablesInfo)
				} else if pick == 3 {
					err = tables.dropPartition(pdClient, db2, tablesInfo)
				} else if pick == 4 {
					err = tables.dropTable(pdClient, db2, tablesInfo)
				} else if pick == 5 {
					err = tables.truncatePartition(pdClient, db2, tablesInfo)
				} else if pick == 6 {
					err = tables.truncateTable(pdClient, db2, tablesInfo)
				} else if pick == 7 {
					err = tables.flashbackTable(pdClient, db2, tablesInfo)
				} else if pick == -1 {
					err = tables.alterDatabaseSetReplica(db2, replica, "test98")
				} else if pick == -2 {
					err = tables.removeDatabaseSetReplica(db2, "test98")
				} else if pick == 8 {
					err = tables.addTable(pdClient, db2, false, false, tablesInfo)
				} else if pick == 9 {
					err = tables.addTable(pdClient, db2, false, false, tablesInfo)
				} else if pick == 10 {
					err = tables.setTiFlashReplica(pdClient, db2)
				} else if pick == 11 {
					err = tables.removeTiFlashReplica(pdClient, db2)
				}
				if err != nil {
					errChan <- err
					return
				}
			}
		}(y)
	}
	wg.Wait()
	select {
	case <-ctx.Done():
		return nil
	default:
	}
	if len(errChan) != 0 {
		return <-errChan
	}
	noReplica := 0
	for i, t := range tables.Ts {
		if !t.Dropped && t.ReplicaCount == 0 {
			log.Info(fmt.Sprintf("table %v has 0 Replica\n", i))
			noReplica += 1
		}
	}

	log.Info("check table weather ok")
	db2, err = tc.GetDB("")
	if err != nil {
		return err
	}
	if ok := waitAllTableOK(db2, "test98", tableAvailableTime, "all", noReplica); !ok {
		return errors.New("some table not ready")
	}

	log.Info("check pd rules")
	later, err := pdClient.getGroupRulesCount()
	if err != nil {
		return err
	}
	// pd rule 会存在多余的情况，但是不影响实际调度性能，因此只校验 rule 数量不比预期少
	// 相关 issue : https://github.com/pingcap/tidb/issues/31540
	_ = tables.check(later - origin)

	z, err := pdClient.getGroupRules("tiflash")
	if err != nil {
		return err
	}
	for _, x := range z {
		log.Info(fmt.Sprintf("%v %v %v", x.ID, x.GroupID, x.Constraints))
	}
	return nil
}

func (t *ExistTablesInfo) getExistTableName() (string, bool) {
	var tables []string
	for k := range t.MaxPartition {
		tables = append(tables, k)
	}
	if len(tables) == 0 {
		return "", false
	}
	randomIndex := rand.Intn(len(tables))
	return tables[randomIndex], true
}

func max(a, b int) int {
	if a > b {
		return a
	} else {
		return b
	}
}

const (
	intMax = 2147483647
	intMin = -2147483648
)

func (t *ExistTablesInfo) RandomDQL(ctx context.Context, db *sql.DB) error {
	t.Lock()
	defer t.Unlock()
	name, ok := t.getExistTableName()
	if !ok {
		return nil
	}
	q := fmt.Sprintf("select * from test98.%s", name)
	err := insertBaseData(db, name)
	if err != nil {
		log.Info("errrrrr")
		return err
	}
	err = diffTiFlashTiKV(ctx, db, q, false)
	if err != nil {
		log.Error("execute query failed", zap.String("sql", q), zap.Error(err))
		if CtxDoneHandler(err) == nil {
			return nil
		}
		if InconsistentHandler(err) != nil {
			return err
		}
	}
	return nil
}

func diffTiFlashTiKV(ctx context.Context, db *sql.DB, q string, isSort bool) (err error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() {
		if err != nil && InconsistentHandler(err) == nil {
			err = tx.Rollback()
			if err == sql.ErrTxDone {
				err = nil
			}
		}
	}()
	var tso uint64
	if err := tx.QueryRow("select @@tidb_current_ts").Scan(&tso); err != nil {
		return err
	}
	if _, err = tx.Exec("set @@tidb_isolation_read_engines='tiflash';"); err != nil {
		return err
	}
	rowValuesTiFlash, _, err := Query(ctx, tx, q, false)
	if err != nil {
		return err
	}
	if _, err = tx.Exec("set @@tidb_isolation_read_engines='tikv';"); err != nil {
		return err
	}
	rowValuesTiKV, _, err := Query(ctx, tx, q, false)
	if err != nil {
		return err
	}
	if isSort {
		sort.Strings(rowValuesTiFlash)
		sort.Strings(rowValuesTiKV)
	}
	if !testEq(rowValuesTiKV, rowValuesTiFlash) {
		err = fmt.Errorf("%w tso %d, tikv: %v, tiflash %v", InconsistentErr, tso, rowValuesTiKV, rowValuesTiFlash)
	} else {
		log.Info("dql - consistency check success", zap.String("count result", strconv.Itoa(len(rowValuesTiFlash))))
	}
	err = tx.Commit()
	return
}

func (t *ExistTablesInfo) RandomDML(ctx context.Context, db *sql.DB) error {
	t.Lock()
	defer t.Unlock()
	tableName, ok := t.getExistTableName()
	if !ok {
		return nil
	}

	maxInsert, ok := t.MaxPartition[tableName]
	if !ok {
		maxInsert = intMax
	}
	for i := maxInsert - 10; i < maxInsert && i > intMin; i++ {
		q := fmt.Sprintf(`insert into test98.%s value(%d)`, tableName, i)
		_, _, err := Query(ctx, db, q, false)
		if err != nil {
			log.Error("dml - sql failed", zap.String("sql", q), zap.Error(err))
			if CtxDoneHandler(err) == nil {
				return nil
			} else {
				return err
			}
		}

		q = fmt.Sprintf(`UPDATE test98.%s SET z = %d WHERE z = %d;`, tableName, i-1000, i)
		_, _, err = Query(ctx, db, q, false)
		if err != nil {
			log.Error("dml - sql failed", zap.String("sql", q), zap.Error(err))
			if CtxDoneHandler(err) == nil {
				return nil
			} else {
				return err
			}
		}

		q = fmt.Sprintf(`DELETE FROM test98.%s WHERE z = %d`, tableName, i-1000)
		_, _, err = Query(ctx, db, q, false)
		if err != nil {
			log.Error("dml - sql failed", zap.String("sql", q), zap.Error(err))
			if CtxDoneHandler(err) == nil {
				return nil
			} else {
				return err
			}
		}

		q = fmt.Sprintf(`insert into test98.%s value(%d)`, tableName, i)
		_, _, err = Query(ctx, db, q, false)
		if err != nil {
			log.Error("dml - sql failed", zap.String("sql", q), zap.Error(err))
			if CtxDoneHandler(err) == nil {
				return nil
			} else {
				return err
			}
		}
		t.MaxPartition[tableName] = t.MaxPartition[tableName] - 10
	}

	log.Info(fmt.Sprintf("dml - success insert into test98.%s, value from %d to %d", tableName, maxInsert-10, maxInsert-1))
	return nil
}

func insertBaseData(db *sql.DB, table string) error {
	for i := intMin; i < intMin+10; i++ {
		q := fmt.Sprintf(`insert into test98.%s value(%d)`, table, i)
		_, err := db.Exec(q)
		if err != nil {
			log.Error("dml - sql failed", zap.String("sql", q), zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil
			}
			return err
		}
	}
	return nil
}