package consistency_test

import (
	"context"
	"database/sql"
	"fmt"
	"math/rand"
	"strconv"
	"time"

	. "github.com/onsi/ginkgo"
	"github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/matcher/metrics"
	"github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/tiflash/pkg/checker"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"

	"github.com/pingcap/endless/pkg/client/tidb"
	"github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/util/schedule"
	"github.com/pingcap/endless/testcase/tiflash/pkg"
	"github.com/pingcap/endless/testcase/tiflash/pkg/chaos"

	"github.com/pingcap/endless/pkg/datagen"
	"github.com/pingcap/endless/testcase/tiflash/consistency/utils"
)

const prepareRows = 1000 * 1000 * 100

// TODO add param to control the chaos type
var _ = Describe("TiFlash", func() {
	Context("when execute dml currently", func() {
		ctx := suiteTestCtx
		var (
			tc         resource.TiDBCluster
			dbName     = "test"
			replicas   int
			duration   time.Duration
			ti         *datagen.TableInfo
			db         *sql.DB
			chaosIdRec resource.ChaosIdentifier
			faultTypes []resource.ExperimentKind
		)
		BeforeEach(func() {
			duration = matcher.Try(time.ParseDuration(ctx.Param("duration", "4h"))).(time.Duration)
			replicas = matcher.Try(strconv.Atoi(ctx.Param("replicas", "2"))).(int)
			tc = ctx.Resource("tc").(resource.TiDBCluster)
			db = matcher.Try(pkg.OpenDB(tc, dbName)).(*sql.DB)
			tidbCli := tidb.NewClient(db)
			ti = datagen.NewTableInfo(datagen.AutoIncrWideTable, db, false, true)
			// TODO: 目前 test-infra 还不支持 IODelay 的 chaos 注入，待补充
			// resource.ExperimentIODelayKind
			faultTypes = []resource.ExperimentKind{resource.ExperimentFailureKind}
			ti = datagen.NewTableInfo(datagen.AutoIncrWideTable, db, false, true)
			matcher.Try(tidbCli.AlterBaseTablesSetTiFlashReplica(dbName, []string{ti.GetTableName()}, replicas, ""))
			gomega.Eventually(func() bool {
				return matcher.Try(tidbCli.IsTiFlashReplicaAvailable(dbName, []string{ti.GetTableName()})).(bool)
			}, 3*time.Minute, 3*time.Second).Should(gomega.BeTrue())
			count := matcher.Try(pkg.Query(suiteTestCtx, db, fmt.Sprintf("select count(*) from %s", ti.GetTableName()), false)).([]string)
			if matcher.Try(strconv.Atoi(count[0])).(int) > 0 {
				log.Info("skip prepare data", zap.String("rows count", count[0]))
				return
			}
			log.Info("prepare data begin", zap.Int("prepare rows", prepareRows))
			matcher.Try(ti.Run(&datagen.Request{
				Ctx: context.Background(),
				PrepareRequest: &datagen.PrepareRequest{
					TotalRows:   prepareRows,
					InsertBatch: 10,
					Threads:     10,
				}}))
			log.Info("prepare data done", zap.Int("prepare rows", prepareRows))
		})
		It("should always consistency and not crash when hotpot update #001#", func() {
			pkGen := func() func() string {
				return func() string {
					return fmt.Sprintf("id = '%d'", datagen.RandUint32()%10+1)
				}
			}
			ctx, _ := context.WithTimeout(suiteTestCtx, duration)
			g, c := errgroup.WithContext(ctx)
			g.Go(util.WithGinkgoRecover(func() {
				matcher.Try(ti.Run(&datagen.Request{
					Ctx: c,
					DmlReqs: []*datagen.DMLRequest{
						{
							DmlType:          datagen.Update,
							Threads:          10,
							Limit:            1,
							ErrorHandler:     pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate),
							ConditionHandler: pkGen(),
						},
					}}))
			}))
			checkConsistency(c, ti, g, tc)
			g.Go(util.WithGinkgoRecover(func() {
				matcher.ConsistentlyWithMetrics(tc).
					ShouldNot(metrics.TiFlashRestart().WithCtx(c))
			}))
			matcher.Try(g.Wait())
		})
		It("should always consistency #002#", func() {
			ctx, _ := context.WithTimeout(suiteTestCtx, duration)
			// TODO abstract it
			op1 := matcher.Try(chaos.NewHotReloadTiFlashConfig(tc)).(*chaos.HotReloadTiFlashConfig)
			matcher.Try(schedule.RunPeriodically(ctx, "@every 20m", 10*time.Minute, op1, "randomHotReloadTiFlashConfig"))
			op2 := matcher.Try(chaos.NewPdScheduler(tc, "shuffle-region-scheduler", 60*10)).(*chaos.PdScheduler)
			matcher.Try(schedule.RunPeriodically(ctx, "@every 20m", 10*time.Minute, op2, "randomPauseAndResumeScheduler"))
			op3 := matcher.Try(chaos.NewCompactTiFlashReplica(tc)).(*chaos.CompactTiFlashReplica)
			matcher.Try(schedule.RunPeriodically(ctx, "@every 20m", 10*time.Minute, op3, "randomCompactTiFlashReplica"))
			g, c := errgroup.WithContext(ctx)
			g.Go(util.WithGinkgoRecover(func() {
				matcher.Try(ti.Run(&datagen.Request{
					Ctx: c,
					DmlReqs: []*datagen.DMLRequest{
						{
							DmlType:      datagen.Insert,
							Threads:      10,
							InsertBatch:  10,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate),
						},
						{
							DmlType:      datagen.Update,
							Threads:      5,
							Limit:        1,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate),
						},
						{
							DmlType:      datagen.Delete,
							Threads:      1,
							Limit:        1,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate),
						},
					}}))
			}))
			checkConsistency(c, ti, g, tc)
			g.Go(util.WithGinkgoRecover(func() {
				for {
					select {
					case <-ctx.Done():
						return
					default:
					}
					// query run normal before chaos
					randomFaultType := faultTypes[rand.Intn(len(faultTypes))]
					log.Info("inject chaos", zap.String("chaos type", fmt.Sprintf("%v", randomFaultType)))
					chaosIdRec = matcher.Try(chaos.InjectFault(ctx, tc, resource.TiFlash, randomFaultType)).(resource.ChaosIdentifier)
					pkg.Sleep(ctx, 10*time.Minute)
					log.Info("delete chaos", zap.String("chaos info", chaosIdRec.HumanReadableName()))
					matcher.Try(chaos.DeleteChaos(tc, chaosIdRec))
					pkg.Sleep(ctx, 10*time.Minute)
				}
			}))
			matcher.Try(g.Wait())
		})
	})

	// alter pk
	// 一旦一张表创建成功，其主键就不能再改变。如果需要添加/删除主键，需要在 TiDB 配置文件中将 alter-primary-key 设置为 true，并重启 TiDB 实例使之生效。
	// 当开启添加/删除主键功能以后，TiDB 允许对表添加/删除主键。
	// 但需要注意的是，如果开启该功能前所创建表带有整数类型的主键，即使开启添加/删除主键功能，也不能删除其主键约束。
	// 1. TiDB 不支持在建表之后添加或删除聚簇索引
	// 2. TiDB 支持在建表之后添加或删除非聚簇索引
	// 3. 不支持修改主键列上需要 Reorg-Data 的类型，但是支持修改 Meta-Only 的类型.
	// alter table t add primary key(a)
	// ALTER TABLE t MODIFY COLUMN a bigint; int-> bigint 支持
	// alter table t1 rename column id to d
	Context("when dml and alter pk with consistency", func() {
		It("should always consistency #005#", func() {
			duration := matcher.Try(time.ParseDuration(suiteTestCtx.Param("duration", "5h"))).(time.Duration)
			ctx, cancel := context.WithTimeout(suiteTestCtx, duration)
			defer cancel()
			tc := suiteTestCtx.Resource("tc").(resource.TiDBCluster)
			tables := []string{
				utils.OtherHandle1, utils.OtherHandle5, utils.OtherHandle6,
			}
			tableInfos := genTableInfos(tc, tables, 2, false)

			ddls := []string{
				"alter table %s drop primary key",
				"alter table %s add primary key(c)",
				"alter table %s drop primary key",
				"alter table %s add primary key(a,c)",
				"alter table %s drop primary key",
				"alter table %s add primary key(a)",
			}
			g1, c1 := errgroup.WithContext(ctx)
			for i := 0; i < len(tables); i++ {
				index := i
				ti := tableInfos[index]
				log.Info("begin prepare data", zap.String("database name", ti.GetDBName()), zap.String("table name", ti.GetTableName()), zap.Int("prepare rows", 1000*1000))
				g1.Go(util.WithGinkgoRecover(func() {
					matcher.Try(ti.Run(&datagen.Request{
						Ctx: c1,
						PrepareRequest: &datagen.PrepareRequest{
							TotalRows:    1000 * 1,
							InsertBatch:  100,
							Threads:      2,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, utils.IgnoreDupEntry),
						}}))
				}))
			}
			log.Info("end prepare data")
			matcher.Try(g1.Wait())

			g2, c2 := errgroup.WithContext(ctx)
			for i := 0; i < len(tables); i++ {
				index := i
				ti := tableInfos[index]
				checkConsistency(c2, ti, g2, tc)
				g2.Go(util.WithGinkgoRecover(func() {
					for {
						for _, ddl := range ddls {
							ddlS := fmt.Sprintf(ddl, ti.GetTableName())
							log.Info("ddl sequence", zap.String("sql", ddlS))
							_, err := ti.GetDB().ExecContext(c2, ddlS)
							if err != nil {
								log.Error("ddl execute error", zap.String("sql", ddlS), zap.Error(err))
								if pkg.CtxDoneHandler(err) == nil {
									return
								}
							}
							time.Sleep(5 * time.Minute)
						}
					}
				}))
				g2.Go(util.WithGinkgoRecover(func() {
					matcher.Try(ti.Run(&datagen.Request{
						Ctx: c2,
						DmlReqs: []*datagen.DMLRequest{
							{
								DmlType:      datagen.Insert,
								Threads:      2,
								InsertBatch:  10,
								ErrorHandler: pkg.SkipErrorHandler,
								// ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate, utils.IgnoreDupEntry),
							},
							{
								DmlType:      datagen.Update,
								Threads:      1,
								Limit:        1,
								ErrorHandler: pkg.SkipErrorHandler,
								// ErrorHandler:     pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate),
								ConditionHandler: func() string { return "true" },
							},
							{
								DmlType:      datagen.Delete,
								Threads:      1,
								Limit:        1,
								ErrorHandler: pkg.SkipErrorHandler,
								// ErrorHandler:     pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate),
								ConditionHandler: func() string { return "true" },
							},
						}}))
				}))
			}
			matcher.Try(g2.Wait())
		})
	})

	Context("when dml with check consistency", func() {
		var (
			duration time.Duration
			replicas int
			currency int
			tc       resource.TiDBCluster
		)
		BeforeEach(func() {
			duration = matcher.Try(time.ParseDuration(suiteTestCtx.Param("duration", "5h"))).(time.Duration)
			replicas = matcher.Try(strconv.Atoi(suiteTestCtx.Param("replicas", "2"))).(int)
			currency = matcher.Try(strconv.Atoi(suiteTestCtx.Param("currency", "10"))).(int)
			tc = suiteTestCtx.Resource("tc").(resource.TiDBCluster)
		})
		// 多并发创建表，insert data，对比一致性
		It("should always consistency #003#", func() {
			log.Info("begein insert and check consistency", zap.Int("insert rows", prepareRows), zap.Duration("duration", duration))
			var tableInfos = genTableInfos(tc, utils.Tables, replicas, false)
			ctx, cancel := context.WithTimeout(suiteTestCtx, duration)
			defer cancel()
			g, c := errgroup.WithContext(ctx)
			for i := 0; i < currency; i++ {
				ti := tableInfos[i]
				g.Go(util.WithGinkgoRecover(func() {
					matcher.Try(ti.Run(&datagen.Request{
						Ctx: c,
						PrepareRequest: &datagen.PrepareRequest{
							TotalRows:    prepareRows,
							InsertBatch:  10,
							Threads:      5,
							ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, utils.IgnoreDupEntry),
						}}))
				}))
				checkConsistency(c, ti, g, tc)
			}
			matcher.Try(g.Wait())
		})

		// 1. 多并发 create table，load data, delete all 或者部分数据（比如连续区间、取模随机、正态分布等），
		// 即大事务更新。然后在继续 insert ，并且循环。 Delete limit 数据量从 1w ~ 500w。表行数 10w ~ 5kw。
		// 增加数据量 by Step 100W。
		// 2. 上一个 case 的基础上同时带有 insert（随机、append only）、update、delete 等 workload（小事务）

		// 3. 多个并发反复 create table（多张表）、load data 、delete all 或者 drop table，表行数 10w ~ 5kw。
		// 4. 上一个 case 的基础上，增加少量 update (load data 完成后)。

		// 1. small=false,action_after_insert=delete_some
		// 2. small=true,action_after_insert=delete_some
		// 3. small=false,action_after_insert=delete_all
		// 5. small=false,action_after_insert=update_and_delete_all
		// TODO update pk
		It("should always consistency #004#", func() {
			const (
				deleteSomeAction   = "delete_some"
				deleteAllAction    = "delete_all"
				updateAndDeleteAll = "update_and_delete_all"
			)
			var withSmallTransaction = matcher.Try(strconv.ParseBool(suiteTestCtx.Param("small", "false"))).(bool)
			var actionAfterInsert = suiteTestCtx.Param("action_after_insert", "update_and_delete_all")
			log.Info("param", zap.String("action after insert", actionAfterInsert), zap.Bool("small transaction", withSmallTransaction))
			var loops = 20
			var tableInfos = genTableInfos(tc, utils.Tables, replicas, false)
			log.Info("begin dml and check consistency", zap.Int("insert rows", prepareRows), zap.Duration("duration", duration))
			ctx, cancel := context.WithTimeout(suiteTestCtx, duration)
			defer cancel()

			g, c := errgroup.WithContext(ctx)
			op1 := matcher.Try(chaos.NewPdScheduler(tc, "shuffle-region-scheduler", 60*10)).(*chaos.PdScheduler)
			matcher.Try(schedule.RunPeriodically(c, "@every 20m", 10*time.Minute, op1, "randomPauseAndResumeScheduler"))
			op2 := matcher.Try(chaos.NewCompactTiFlashReplica(tc)).(*chaos.CompactTiFlashReplica)
			matcher.Try(schedule.RunPeriodically(c, "@every 20m", 10*time.Minute, op2, "randomCompactTiFlashReplica"))
			for i := 0; i < currency; i++ {
				ti := tableInfos[i]
				checkConsistency(c, ti, g, tc)
				if withSmallTransaction {
					g.Go(util.WithGinkgoRecover(func() {
						matcher.Try(ti.Run(&datagen.Request{
							Ctx: c,
							DmlReqs: []*datagen.DMLRequest{
								{
									DmlType:      datagen.Insert,
									Threads:      2,
									InsertBatch:  10,
									ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate, utils.IgnoreDupEntry, utils.IgnorePessimisticLock),
								},
								{
									DmlType:          datagen.Update,
									Threads:          1,
									Limit:            1,
									ErrorHandler:     pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate, utils.IgnoreDupEntry, utils.IgnorePessimisticLock),
									ConditionHandler: func() string { return "true" },
								},
								{
									DmlType:          datagen.Delete,
									Threads:          1,
									Limit:            1,
									ErrorHandler:     pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate, utils.IgnoreDupEntry, utils.IgnorePessimisticLock, pkg.LockWaitTimeoutHandler),
									ConditionHandler: func() string { return "true" },
								},
							}}))
					}))
				}
			}

			g.Go(util.WithGinkgoRecover(func() {
				insertRowCount := 1000 * 100
				// insertStep := 1000 * 1000
				insertStep := 1000 * 1000
				deleteRowCount := 1000 * 10
				// deleteRowStep := 1000 * 100
				deleteRowStep := 1000 * 1
				for j := 0; j < loops; j++ {
					// insert insertRowCount ~ insertRowCount + insertStep*(currency-1) rows data
					g1, c1 := errgroup.WithContext(ctx)
					for i := 0; i < currency; i++ {
						index := i
						ti := tableInfos[index]
						log.Info("begin prepare data", zap.Int("loop", j), zap.String("database name", ti.GetDBName()), zap.String("table name", ti.GetTableName()), zap.Int("prepare rows", insertRowCount+index*insertStep))
						g1.Go(util.WithGinkgoRecover(func() {
							matcher.Try(ti.Run(&datagen.Request{
								Ctx: c1,
								PrepareRequest: &datagen.PrepareRequest{
									TotalRows:    insertRowCount + index*insertStep,
									InsertBatch:  1000,
									Threads:      5,
									ErrorHandler: pkg.ErrorHandler(pkg.CtxDoneHandler, utils.IgnoreDupEntry, utils.IgnorePessimisticLock, utils.IgnoreLockWait),
								}}))
						}))
					}
					matcher.Try(g1.Wait())
					log.Info("end prepare data", zap.Int("loop", j))

					switch actionAfterInsert {
					case deleteAllAction, updateAndDeleteAll:
						if actionAfterInsert == updateAndDeleteAll {
							ctxTmp2, _ := context.WithTimeout(ctx, 10*time.Minute)
							g3, c3 := errgroup.WithContext(ctxTmp2)
							for i := 0; i < currency; i++ {
								index := i
								ti := tableInfos[index]
								log.Info("begin update data", zap.Int("loop", j), zap.String("database name", ti.GetDBName()), zap.String("table name", ti.GetTableName()), zap.Int("update limit", 1000))
								g3.Go(util.WithGinkgoRecover(func() {
									matcher.Try(ti.Run(&datagen.Request{
										Ctx: c3,
										DmlReqs: []*datagen.DMLRequest{
											{
												DmlType:          datagen.Update,
												Threads:          1,
												Limit:            1000,
												ErrorHandler:     pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate, utils.IgnoreDupEntry, utils.IgnorePessimisticLock, utils.IgnoreLockWait),
												ConditionHandler: func() string { return "true" },
											},
										}}))
								}))
							}
							matcher.Try(g3.Wait())
							log.Info("end update data", zap.Int("loop", j))
						}
						for i, ti := range tableInfos {
							var qs []string
							switch i % 3 {
							case 0:
								qs = []string{fmt.Sprintf("delete from %s", ti.GetTableName())}
							case 1:
								qs = []string{fmt.Sprintf("truncate table %s", ti.GetTableName())}
							case 2:
								qs = []string{fmt.Sprintf("drop table %s", ti.GetTableName()),
									ti.GetCreateTableSql(),
									fmt.Sprintf("alter table %s set tiflash replica %d", ti.GetTableName(), replicas)}
							}
							log.Info("delete/truncate/drop data", zap.Int("loop", j), zap.Strings("sqls", qs))
							for _, q := range qs {
								_, err := ti.GetDB().Exec(q)
								matcher.Try(pkg.ErrorHandler(pkg.CtxDoneHandler, utils.IgnorePessimisticLock, utils.IgnoreLockWait)(err))
							}
						}
						time.Sleep(2 * time.Minute)
						log.Info("end delete/truncate/drop data", zap.Int("loop", j))
					case deleteSomeAction:
						// delete deleteRowCount ~ deleteRowStep*(currency-1) rows data
						ctxTmp, _ := context.WithTimeout(ctx, 15*time.Minute)
						g2, c2 := errgroup.WithContext(ctxTmp)
						for i := 0; i < currency; i++ {
							index := i
							ti := tableInfos[index]
							log.Info("begin delete data", zap.Int("loop", j), zap.String("database name", ti.GetDBName()), zap.String("table name", ti.GetTableName()), zap.Int("delete rows", deleteRowCount+index*deleteRowStep))
							g2.Go(util.WithGinkgoRecover(func() {
								matcher.Try(ti.Run(&datagen.Request{
									Ctx: c2,
									DmlReqs: []*datagen.DMLRequest{
										{
											DmlType:          datagen.Delete,
											Threads:          1,
											Limit:            deleteRowCount + index*deleteRowStep,
											ErrorHandler:     pkg.ErrorHandler(pkg.CtxDoneHandler, pkg.InvalidConnectionHandler, pkg.TiDBInformationSchemaOutDate, utils.IgnorePessimisticLock, utils.IgnoreLockWait, pkg.LockWaitTimeoutHandler),
											ConditionHandler: func() string { return "true" },
										},
									}}))
							}))
						}
						matcher.Try(g2.Wait())
						time.Sleep(2 * time.Minute)
						log.Info("end delete data", zap.Int("loop", j))
					}
				}
			}))
			checker.CommonCheckWithoutLatencyAndRestart(ctx, tc, g, duration)
			matcher.Try(g.Wait())
		})
	})
})

func genTableInfos(tc resource.TiDBCluster, tables []string, replicas int, recordInsert bool) []*datagen.TableInfo {
	var tableInfos []*datagen.TableInfo
	var defaultDB = matcher.Try(tc.GetDB("")).(*sql.DB)
	defer defaultDB.Close()
	for i, t := range tables {
		// database name db{i}
		dbName := fmt.Sprintf("db%d", i)
		log.Info("drop and create database", zap.String("database name", dbName))
		matcher.Try(defaultDB.Exec(fmt.Sprintf("drop database if exists %s", dbName)))
		matcher.Try(defaultDB.Exec(fmt.Sprintf("create database %s", dbName)))
		db := matcher.Try(pkg.OpenDB(tc, dbName, "tidb_mem_quota_query=32212254720")).(*sql.DB)
		ti := datagen.NewTableInfo(t, db, true, recordInsert)
		tableInfos = append(tableInfos, ti)
		tidbCli := tidb.NewClient(db)
		log.Info("create table", zap.String("database name", dbName), zap.String("table name", ti.GetTableName()))
		matcher.Try(tidbCli.AlterBaseTablesSetTiFlashReplica(dbName, []string{ti.GetTableName()}, replicas, ""))
		gomega.Eventually(func() bool {
			return matcher.Try(tidbCli.IsTiFlashReplicaAvailable(dbName, []string{ti.GetTableName()})).(bool)
		}, 10*time.Minute, 3*time.Second).Should(gomega.BeTrue())
	}
	return tableInfos
}

func checkConsistency(ctx context.Context, ti *datagen.TableInfo, g *errgroup.Group, tc resource.TiDBCluster) {
	g.Go(util.WithGinkgoRecover(func() {
		db2 := matcher.Try(pkg.OpenDB(tc, ti.GetDBName())).(*sql.DB)
		cli := tidb.NewClient(db2)
		for {
			tbls := matcher.Try(cli.ShowTableNames(ti.GetDBName())).([]string)
			log.Info("convert tiflash mode", zap.String("db name", ti.GetDBName()), zap.Strings("table names", tbls), zap.String("mode", "normal"))
			matcher.Try(cli.AlterBaseTablesSetTiFlashMode(ti.GetDBName(), tbls, "normal"))
			err := pkg.DiffTiFlashTiKV(ctx, db2, fmt.Sprintf("select count(*) from %s", ti.GetTableName()), false, true)
			if err != nil {
				log.Error("check consistence error", zap.String("db name", ti.GetDBName()), zap.String("sql", fmt.Sprintf("select count(*) from %s", ti.GetTableName())), zap.Error(err))
				if pkg.CtxDoneHandler(err) == nil {
					return
				}
				matcher.Try(pkg.InconsistentHandler(err))
			}
			time.Sleep(10 * time.Second)
			log.Info("convert tiflash mode", zap.String("db name", ti.GetDBName()), zap.Strings("table names", tbls), zap.String("mode", "fast"))
			matcher.Try(cli.AlterBaseTablesSetTiFlashMode(ti.GetDBName(), tbls, "fast"))
		}
	}))
}