// Copyright 2019 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package randsql

import (
	"database/sql"
	"fmt"
	"math"
	"math/rand"
	"regexp"
	"sort"
	"strings"
	"unsafe"

	. "github.com/pingcap/schrodinger-test/logic/ddl"
)

func (c *RandSQLGen) GenDDL(tp DDLKind) *DDLJobTask {
	switch tp {
	case DDLCreateSchema:
		return c.GenCreateSchema()
	case DDLDropSchema:
		return c.GenDropSchema()
	case DDLCreateTable:
		return c.GenCreateTable()
	case DDLDropTable:

	}
	return nil
}

const (
	DDLCreateTable DDLKind = iota
	DDLAddIndex
	DDLAddColumn
	DDLCreateSchema
	DDLCreateView

	DDLDropTable
	DDLDropIndex
	DDLDropColumn
	DDLDropSchema

	DDLDropRangePartition
	DDLAddRangePartition

	DDLRenameTable
	DDLRenameIndex
	DDLTruncateTable
	DDLTruncateTablePartition
	DDLShardRowID
	DDLRebaseAutoID
	DDLSetDefaultValue
	DDLModifyColumn
	DDLModifyTableComment
	DDLModifyTableCharsetAndCollate

	ddlKindNil
)

var mapOfDDLKind = map[string]DDLKind{
	"create schema": DDLCreateSchema,
	"create table":  DDLCreateTable,
	"add index":     DDLAddIndex,
	"add column":    DDLAddColumn,

	"drop Schema": DDLDropSchema,
	"drop table":  DDLDropTable,
	"drop index":  DDLDropIndex,
	"drop column": DDLDropColumn,

	"create view": DDLCreateView,

	"rename Table":                     DDLRenameTable,
	"rename Index":                     DDLRenameIndex,
	"truncate table":                   DDLTruncateTable,
	"shard row ID":                     DDLShardRowID,
	"rebase auto_increment ID":         DDLRebaseAutoID,
	"set default value":                DDLSetDefaultValue,
	"modify table comment":             DDLModifyTableComment,
	"modify table charset and collate": DDLModifyTableCharsetAndCollate,

	"modify column": DDLModifyColumn,
}

var mapOfDDLKindToString = map[DDLKind]string{
	DDLCreateSchema: "create schema",
	DDLCreateTable:  "create table",
	DDLAddIndex:     "add index",
	DDLAddColumn:    "add column",

	DDLDropSchema: "drop schema",
	DDLDropTable:  "drop table",
	DDLDropIndex:  "drop index",
	DDLDropColumn: "drop column",

	DDLCreateView: "create view",

	DDLRenameTable:                  "rename Table",
	DDLRenameIndex:                  "rename Index",
	DDLTruncateTable:                "truncate table",
	DDLShardRowID:                   "shard row ID",
	DDLRebaseAutoID:                 "rebase auto_increment ID",
	DDLSetDefaultValue:              "set default value",
	DDLModifyTableComment:           "modify table comment",
	DDLModifyTableCharsetAndCollate: "modify table charset and collate",
	DDLModifyColumn:                 "modify column",
}

// mapOfDDLKindProbability use to control every kind of ddl request execute probability.
var mapOfDDLKindProbability = map[DDLKind]float64{
	DDLCreateTable: 0.15,
	DDLDropTable:   0.15,

	DDLAddIndex:  0.8,
	DDLDropIndex: 0.5,

	DDLAddColumn:    0.8,
	DDLModifyColumn: 0.5,
	DDLDropColumn:   0.5,

	DDLCreateView: 0.30,

	DDLCreateSchema:                 0.10,
	DDLDropSchema:                   0.10,
	DDLRenameTable:                  0.50,
	DDLRenameIndex:                  0.50,
	DDLTruncateTable:                0.50,
	DDLShardRowID:                   0.30,
	DDLRebaseAutoID:                 0.15,
	DDLSetDefaultValue:              0.30,
	DDLModifyTableComment:           0.30,
	DDLModifyTableCharsetAndCollate: 0.30,
}

type ddlJobArg unsafe.Pointer

type DDLJobTask struct {
	k          DDLKind
	tblInfo    *TableInfo
	schemaInfo *SchemaInfo
	viewInfo   *ViewInfo
	sql        string
	arg        ddlJobArg
}

func (task *DDLJobTask) SQL() string {
	return task.sql
}

func (task *DDLJobTask) Table() *TableInfo {
	return task.tblInfo
}

func (c *RandSQLGen) DoSQLJobInLocal(task SQLTask, db *sql.DB) error {
	switch v := task.(type) {
	case *DDLJobTask:
		return c.DoDDLJob(v, db)
	case *DMLJobTask:
		return c.DoDMLJobInLocal(v)
	}
	return fmt.Errorf("unknow task  %#v", task)
}

func (c *RandSQLGen) DoDDLJob(task *DDLJobTask, db *sql.DB) error {
	switch task.k {
	case DDLCreateSchema:
		return c.createSchemaJob(task)
	case DDLDropSchema:
		return c.dropSchemaJob(task)
	case DDLCreateTable:
		return c.addTableInfo(task)
	case DDLRenameTable:
		return c.renameTableJob(task)
	case DDLTruncateTable:
		return c.truncateTableJob(task)
	case DDLTruncateTablePartition:
		// temporary do nothing now.
		return nil
	case DDLModifyTableComment:
		return c.modifyTableCommentJob(task)
	case DDLModifyTableCharsetAndCollate:
		return c.modifyTableCharsetAndCollateJob(task)
	case DDLShardRowID:
		return c.shardRowIDJob(task)
	case DDLRebaseAutoID:
		return c.rebaseAutoIDJob(task, db)
	case DDLDropTable:
		return c.dropTableJob(task)
	case DDLCreateView:
		return c.createViewJob(task)
	case DDLAddIndex:
		return c.addIndexJob(task)
	case DDLRenameIndex:
		return c.renameIndexJob(task)
	case DDLDropIndex:
		return c.dropIndexJob(task)
	case DDLAddColumn:
		return c.addColumnJob(task)
	case DDLModifyColumn:
		return c.modifyColumnJob(task)
	case DDLDropColumn:
		return c.dropColumnJob(task)
	case DDLSetDefaultValue:
		return c.setDefaultValueJob(task)
	case DDLDropRangePartition:
		return c.dropRangePartition(task)
	case DDLAddRangePartition:
		return c.addRangePartition(task)
	}
	return fmt.Errorf("unknow ddl task , %v", *task)
}

func IgnoreDMLError(err error) bool {
	if err == nil {
		return true
	}
	errStr := err.Error()
	if strings.Contains(errStr, "Information schema is changed") {
		return true
	}
	if strings.Contains(errStr, "try again later") {
		return true
	}
	// Sometimes, there might be duplicated entry error caused by concurrent.
	// So we ignore here.
	if strings.Contains(errStr, "Duplicate entry") {
		return true
	}
	// Sometimes, a insert to a table might generate an error caused by exceeding maximum auto increment id,
	// we ignore this error here.
	if strings.Contains(errStr, "Failed to read auto-increment value from storage engine") {
		return true
	}
	if strings.Contains(errStr, "invalid connection") {
		return true
	}
	return false
}

func IgnoreDDLError(err error) bool {
	if err == nil {
		return true
	}
	errStr := err.Error()
	if strings.Contains(errStr, "Information schema is changed") {
		return true
	}
	// Sometimes, set shard row id bits to a large value might cause global auto ID overflow error.
	// We ignore this error here.
	if match, _ := regexp.MatchString(`cause next global auto ID( \d+ | )overflow`, errStr); match {
		return true
	}
	if strings.Contains(errStr, "invalid connection") {
		return true
	}
	return false
}

var dbSchemaSyntax = [...]string{"DATABASE", "SCHEMA"}

func (c *RandSQLGen) GenCreateSchema() *DDLJobTask {
	charset, collate := c.pickupRandomCharsetAndCollate()
	schema := SchemaInfo{
		name:    genDBName(),
		deleted: false,
		charset: charset,
		collate: collate,
	}
	sql := fmt.Sprintf("CREATE %s `%s` CHARACTER SET '%s' COLLATE '%s'", dbSchemaSyntax[rand.Intn(len(dbSchemaSyntax))], schema.name,
		charset, collate)
	task := &DDLJobTask{
		k:          DDLCreateSchema,
		sql:        sql,
		schemaInfo: &schema,
	}
	return task
}

func (c *RandSQLGen) createSchemaJob(task *DDLJobTask) error {
	c.schemas[task.schemaInfo.name] = task.schemaInfo
	return nil
}

func (c *RandSQLGen) GenDropSchema() *DDLJobTask {
	schema := c.pickupRandomSchema()
	if schema == nil {
		return nil
	}
	schema.setDeleted()
	sql := fmt.Sprintf("DROP %s `%s`", dbSchemaSyntax[rand.Intn(len(dbSchemaSyntax))], schema.name)
	task := &DDLJobTask{
		k:          DDLDropSchema,
		sql:        sql,
		schemaInfo: schema,
	}
	return task
}

func (c *RandSQLGen) dropSchemaJob(task *DDLJobTask) error {
	if c.isSchemaDeleted(task.schemaInfo) {
		return fmt.Errorf("schema %s doesn't exist", task.schemaInfo.name)
	}
	delete(c.schemas, task.schemaInfo.name)
	putIDToAlloc(task.schemaInfo.name, globalIDAllocator)
	return nil
}

func (c *RandSQLGen) GenRandCreateTable() *DDLJobTask {
	p := rand.Float64()
	if p > 0.66 {
		return c.GenCreateTable()
	}
	if p > 0.33 {
		return c.GenCreateHashPartitionTable()
	}
	return c.GenCreateRangePartitionTable()
}

const NumColumn int = 10

func generateTableInfo() TableInfo {
	tableInfo := TableInfo{
		name:         genTableName(),
		indexes:      make([]*IndexInfo, 0),
		numberOfRows: 0,
		deleted:      0,
		comment:      RandSeq(rand.Intn(20) + 2),
		idAlloc:      newAllocator(),
	}

	columnCount := rand.Intn(NumColumn) + 2
	tableColumns := make([]*ColumnInfo, 0, columnCount)
	for i := 0; i < columnCount; i++ {
		columns := getRandDDLTestColumns(&tableInfo)
		tableColumns = append(tableColumns, columns...)
	}
	tableInfo.columns = tableColumns
	return tableInfo
}

func (c *RandSQLGen) GenCreateTable() *DDLJobTask {
	tableInfo := generateTableInfo()
	tableColumns := tableInfo.columns

	// Generate primary key with [0, 3) size
	primaryKeyFields := rand.Intn(3)
	primaryKeys := make([]int, 0)
	if primaryKeyFields > 0 {
		// Random elections column as primary key, but also check the column whether can be primary key.
		perm := rand.Perm(len(tableColumns))[0:primaryKeyFields]
		for _, columnIndex := range perm {
			if tableColumns[columnIndex].canBePrimary() {
				tableColumns[columnIndex].isPrimaryKey = true
				primaryKeys = append(primaryKeys, columnIndex)
			}
		}
		primaryKeyFields = len(primaryKeys)
	}

	tableInfo.columns = tableColumns

	sql := fmt.Sprintf("CREATE TABLE `%s` (", tableInfo.name)
	for i := 0; i < len(tableInfo.columns); i++ {
		if i > 0 {
			sql += ", "
		}
		sql += fmt.Sprintf("`%s` %s", tableColumns[i].name, tableColumns[i].getDefinition())
	}
	if primaryKeyFields > 0 {
		sql += ", PRIMARY KEY ("
		for i, columnIndex := range primaryKeys {
			if i > 0 {
				sql += ", "
			}
			sql += fmt.Sprintf("`%s`", tableColumns[columnIndex].name)
		}
		sql += ")"
	}
	sql += ")"

	task := &DDLJobTask{
		k:       DDLCreateTable,
		sql:     sql,
		tblInfo: &tableInfo,
	}
	return task
}

func (c *RandSQLGen) GenCreateHashPartitionTable() *DDLJobTask {
	tableInfo := generateTableInfo()
	tableColumns := tableInfo.columns

	// Generate primary key with [0, 3) size
	primaryKeyFields := rand.Intn(3)
	primaryKeys := make([]int, 0)
	hashKey := -1
	if primaryKeyFields > 0 {
		// Random elections column as primary key, but also check the column whether can be primary key.
		perm := rand.Perm(len(tableColumns))[0:primaryKeyFields]
		for _, columnIndex := range perm {
			if tableColumns[columnIndex].canBePrimary() {
				tableColumns[columnIndex].isPrimaryKey = true
				primaryKeys = append(primaryKeys, columnIndex)
				if tableColumns[columnIndex].canBePartitionType() && hashKey == -1 {
					hashKey = columnIndex
					tableColumns[columnIndex].isPartitionKey = true
				}
			}
		}
		primaryKeyFields = len(primaryKeys)
	}
	if hashKey == -1 {
		tp := AllowPartitionType[rand.Intn(len(AllowPartitionType))]
		tableColumns = append(tableColumns, getDDLTestColumn(tp, &tableInfo))
		columnIndex := len(tableColumns) - 1
		hashKey = columnIndex
		if primaryKeyFields > 0 {
			tableColumns[columnIndex].isPrimaryKey = true
			primaryKeys = append(primaryKeys, columnIndex)
			primaryKeyFields = len(primaryKeys)
		}
		tableColumns[columnIndex].isPartitionKey = true
	}

	tableInfo.columns = tableColumns

	sql := fmt.Sprintf("CREATE TABLE `%s` (", tableInfo.name)
	for i := 0; i < len(tableInfo.columns); i++ {
		if i > 0 {
			sql += ", "
		}
		sql += fmt.Sprintf("`%s` %s", tableColumns[i].name, tableColumns[i].getDefinition())
	}
	if primaryKeyFields > 0 {
		sql += ", PRIMARY KEY ("
		for i, columnIndex := range primaryKeys {
			if i > 0 {
				sql += ", "
			}
			sql += fmt.Sprintf("`%s`", tableColumns[columnIndex].name)
		}
		sql += ")"
	}
	sql += ")"
	if hashKey != -1 {
		tableInfo.partition = &partitionInfo{
			num:    rand.Intn(100) + 1,
			tp:     partitionTypeHash,
			column: tableColumns[hashKey],
		}
		tableInfo.setPartitionColumns([]*ColumnInfo{tableColumns[hashKey]})
		sql += fmt.Sprintf(" partition by hash(`%s`) partitions %v", tableColumns[hashKey].name, tableInfo.partition.num)
	}

	task := &DDLJobTask{
		k:       DDLCreateTable,
		sql:     sql,
		tblInfo: &tableInfo,
	}
	return task
}

func (c *RandSQLGen) GenCreateRangePartitionTable() *DDLJobTask {
	tableInfo := generateTableInfo()
	tableColumns := tableInfo.columns

	// Generate primary key with [0, 3) size
	primaryKeyFields := rand.Intn(3)
	primaryKeys := make([]int, 0)
	rangeKey := -1
	if primaryKeyFields > 0 {
		// Random elections column as primary key, but also check the column whether can be primary key.
		perm := rand.Perm(len(tableColumns))[0:primaryKeyFields]
		for _, columnIndex := range perm {
			if tableColumns[columnIndex].canBePrimary() {
				tableColumns[columnIndex].isPrimaryKey = true
				primaryKeys = append(primaryKeys, columnIndex)
				if tableColumns[columnIndex].canBePartitionType() && rangeKey == -1 {
					rangeKey = columnIndex
					tableColumns[columnIndex].isPartitionKey = true
				}
			}
		}
		primaryKeyFields = len(primaryKeys)
	}
	if rangeKey == -1 {
		tp := AllowPartitionType[rand.Intn(len(AllowPartitionType))]
		tableColumns = append(tableColumns, getDDLTestColumn(tp, &tableInfo))
		columnIndex := len(tableColumns) - 1
		rangeKey = columnIndex
		if primaryKeyFields > 0 {
			tableColumns[columnIndex].isPrimaryKey = true
			primaryKeys = append(primaryKeys, columnIndex)
			primaryKeyFields = len(primaryKeys)
		}
		tableColumns[columnIndex].isPartitionKey = true
	}

	tableInfo.columns = tableColumns

	sql := fmt.Sprintf("CREATE TABLE `%s` (", tableInfo.name)
	for i := 0; i < len(tableInfo.columns); i++ {
		if i > 0 {
			sql += ", "
		}
		sql += fmt.Sprintf("`%s` %s", tableColumns[i].name, tableColumns[i].getDefinition())
	}
	if primaryKeyFields > 0 {
		sql += ", PRIMARY KEY ("
		for i, columnIndex := range primaryKeys {
			if i > 0 {
				sql += ", "
			}
			sql += fmt.Sprintf("`%s`", tableColumns[columnIndex].name)
		}
		sql += ")"
	}
	sql += ")"
	if rangeKey != -1 {
		sql += fmt.Sprintf(" partition by range(`%s`) (", tableColumns[rangeKey].name)
		values := genRangeValues(tableColumns[rangeKey])
		tableInfo.partition = &partitionInfo{
			num:     len(values),
			tp:      partitionTypeRange,
			values:  values,
			dropped: make([]bool, len(values)),
			column:  tableColumns[rangeKey],
		}
		tableInfo.setPartitionColumns([]*ColumnInfo{tableColumns[rangeKey]})
		for i, v := range values {
			if i > 0 {
				sql += ", "
			}
			if v < math.MaxInt64 {
				sql += fmt.Sprintf("partition p%v values less than (%v)", i, v)
				continue
			}
			sql += fmt.Sprintf("partition p%v values less than (MAXVALUE)", i)
		}

		sql += ")"
	}

	return &DDLJobTask{
		k:       DDLCreateTable,
		sql:     sql,
		tblInfo: &tableInfo,
	}
}

func genRangeValues(col *ColumnInfo) []int {
	rangeLen := rand.Intn(10) + 1
	values := make([]int, 0, rangeLen)
	uniqueMap := make(map[int]struct{})
	value := 0
	for i := 0; i < rangeLen; i++ {
		v := col.randValue()
		switch x := v.(type) {
		case int:
			value = int(x)
		case int32:
			value = int(x)
		case int64:
			value = int(x)
		default:
			value = int(rand.Int63())
		}
		if _, ok := uniqueMap[value]; ok {
			continue
		}
		uniqueMap[value] = struct{}{}
		values = append(values, value)
	}
	sort.Ints(values)
	if rand.Float64() > 0.5 {
		values[len(values)-1] = math.MaxInt64
	}
	return values
}

func genRangeValue(col *ColumnInfo, largeThanValue int) int {
	var value int
	for i := 0; i < 10; i++ {
		v := col.randValue()
		switch x := v.(type) {
		case int:
			value = int(x)
		case int32:
			value = int(x)
		case int64:
			value = int(x)
		default:
			value = int(rand.Int63())
		}
		if value > largeThanValue {
			return value
		}
	}
	return math.MaxInt64
}

func (c *RandSQLGen) addTableInfo(task *DDLJobTask) error {
	c.tablesLock.Lock()
	defer c.tablesLock.Unlock()
	c.tables[task.tblInfo.name] = task.tblInfo
	return nil
}

func (c *RandSQLGen) GenDropRangePartition() *DDLJobTask {
	tableToTruncate := c.pickupRandomRangePartitionTable()
	if tableToTruncate == nil {
		return nil
	}
	partitionIdx := tableToTruncate.pickRandPartition()
	sql := fmt.Sprintf("ALTER TABLE `%s` drop partition p%v", tableToTruncate.name, partitionIdx)
	task := &DDLJobTask{
		k:       DDLDropRangePartition,
		sql:     sql,
		tblInfo: tableToTruncate,
		arg:     ddlJobArg(&dropRangePartitionArg{partitionIdx: partitionIdx}),
	}
	return task
}

type dropRangePartitionArg struct {
	partitionIdx int
}

func (c *RandSQLGen) dropRangePartition(task *DDLJobTask) error {
	c.tablesLock.Lock()
	defer c.tablesLock.Unlock()
	arg := (*dropRangePartitionArg)(task.arg)
	task.tblInfo.partition.dropped[arg.partitionIdx] = true
	return nil
}

func (c *RandSQLGen) GenAddRangePartition() *DDLJobTask {
	table := c.pickupRandomRangePartitionTable()
	if table == nil {
		return nil
	}
	value, ok := table.genAddRangePartitionValue()
	if !ok {
		return nil
	}
	sql := fmt.Sprintf("ALTER TABLE `%s` add partition ( ", table.name)
	if value < math.MaxInt64 {
		sql += fmt.Sprintf("partition p%v values less than (%v) )", table.partition.num, value)
	} else {
		sql += fmt.Sprintf("partition p%v values less than (MAXVALUE) )", table.partition.num)
	}
	task := &DDLJobTask{
		k:       DDLAddRangePartition,
		sql:     sql,
		tblInfo: table,
		arg:     ddlJobArg(&addRangePartitionArg{value: value}),
	}
	return task
}

type addRangePartitionArg struct {
	value int
}

func (c *RandSQLGen) addRangePartition(task *DDLJobTask) error {
	c.tablesLock.Lock()
	defer c.tablesLock.Unlock()
	arg := (*addRangePartitionArg)(task.arg)
	task.tblInfo.partition.dropped = append(task.tblInfo.partition.dropped, false)
	task.tblInfo.partition.values = append(task.tblInfo.partition.values, arg.value)
	task.tblInfo.partition.num++
	return nil
}

var toAsSyntax = [...]string{"TO", "AS"}

func (c *RandSQLGen) prepareRenameTable() (*DDLJobTask, error) {
	c.tablesLock.Lock()
	defer c.tablesLock.Unlock()
	table := c.pickupRandomTable()
	if table == nil {
		return nil, nil
	}
	// Shadow copy the table.
	newTbl := *table
	table.setDeleted()
	newTbl.name = genTableName()
	sql := fmt.Sprintf("ALTER TABLE `%s` RENAME %s `%s`", table.name,
		toAsSyntax[rand.Intn(len(toAsSyntax))], newTbl.name)
	task := &DDLJobTask{
		k:       DDLRenameTable,
		sql:     sql,
		tblInfo: table,
		arg:     ddlJobArg(&newTbl),
	}
	return task, nil
}

func (c *RandSQLGen) renameTableJob(task *DDLJobTask) error {
	c.tablesLock.Lock()
	defer c.tablesLock.Unlock()
	delete(c.tables, task.tblInfo.name)
	newTbl := (*TableInfo)(task.arg)
	c.tables[newTbl.name] = newTbl
	return nil
}

func (c *RandSQLGen) GenTruncateTable() *DDLJobTask {
	tableToTruncate := c.pickupRandomTable()
	if tableToTruncate == nil {
		return nil
	}
	sql := fmt.Sprintf("TRUNCATE TABLE `%s`", tableToTruncate.name)
	task := &DDLJobTask{
		k:       DDLTruncateTable,
		sql:     sql,
		tblInfo: tableToTruncate,
	}
	return task
}

func (c *RandSQLGen) truncateTableJob(task *DDLJobTask) error {
	table := task.tblInfo
	table.numberOfRows = 0
	for _, column := range table.columns {
		if !column.isGenerated() {
			column.rows = column.rows[0:0]
		}
	}
	return nil
}

func (c *RandSQLGen) GenTruncateTablePartition() *DDLJobTask {
	tableToTruncate := c.pickupRandomPartitionTable()
	if tableToTruncate == nil {
		return nil
	}
	sql := fmt.Sprintf("ALTER TABLE `%s` truncate partition %v", tableToTruncate.name, tableToTruncate.pickRandPartitionNames())
	task := &DDLJobTask{
		k:       DDLTruncateTablePartition,
		sql:     sql,
		tblInfo: tableToTruncate,
	}
	return task
}

func (c *RandSQLGen) GenModifyTableComment() *DDLJobTask {
	table := c.pickupRandomTable()
	if table == nil {
		return nil
	}
	newComm := RandSeq(rand.Intn(20) + 2)
	sql := fmt.Sprintf("ALTER TABLE `%s` COMMENT '%s'", table.name, newComm)
	return &DDLJobTask{
		k:       DDLModifyTableComment,
		tblInfo: table,
		sql:     sql,
		arg:     ddlJobArg(&newComm),
	}
}

func (c *RandSQLGen) modifyTableCommentJob(task *DDLJobTask) error {
	table := task.tblInfo
	newComm := *((*string)(task.arg))
	table.comment = newComm
	return nil
}

type ddlModifyTableCharsetAndCollateJob struct {
	newCharset string
	newCollate string
}

func (c *RandSQLGen) prepareModifyTableCharsetAndCollate(_ interface{}) (*DDLJobTask, error) {
	table := c.pickupRandomTable()
	if table == nil {
		return nil, nil
	}
	charset, collate := c.pickupRandomCharsetAndCollate()
	// Currently, only utf8 character sets like utf8 and utf8mb4, and binary collate are supported.
	// Besides, It supports converting utf8 to utf8mb4 but shows error when converting character set
	// utf8mb4 to utf8.
	// See https://pingcap.com/docs/sql/mysql-compatibility/ for more detail.
	if table.charset != "utf8" || charset != "utf8mb4" {
		return nil, nil
	}
	sql := fmt.Sprintf("ALTER TABLE `%s` CHARACTER SET '%s' COLLATE '%s'",
		table.name, charset, collate)
	task := &DDLJobTask{
		k:       DDLModifyTableCharsetAndCollate,
		sql:     sql,
		tblInfo: table,
		arg: ddlJobArg(&ddlModifyTableCharsetAndCollateJob{
			newCharset: charset,
			newCollate: collate,
		}),
	}
	return task, nil
}

func (c *RandSQLGen) modifyTableCharsetAndCollateJob(task *DDLJobTask) error {
	table := task.tblInfo
	arg := (*ddlModifyTableCharsetAndCollateJob)(task.arg)
	table.charset = arg.newCharset
	table.collate = arg.newCollate
	return nil
}

// OverflowShardBits checks whether the rowID overflow `1<<(64-shardRowIDBits-1) -1`.
func OverflowShardBits(rowID int64, shardRowIDBits uint64) bool {
	mask := (1<<shardRowIDBits - 1) << (64 - shardRowIDBits - 1)
	return rowID&int64(mask) > 0
}

func (c *RandSQLGen) prepareShardRowID() (*DDLJobTask, error) {
	table := c.pickupRandomTable()
	if table == nil {
		return nil, nil
	}
	// Don't make shard row bits too large.
	shardRowId := rand.Intn(MaxShardRowIDBits)
	if OverflowShardBits(table.autoIncID+30000, uint64(shardRowId)) {
		return nil, nil
	}
	sql := fmt.Sprintf("ALTER TABLE `%s` SHARD_ROW_ID_BITS = %d", table.name, shardRowId)
	task := &DDLJobTask{
		k:       DDLShardRowID,
		tblInfo: table,
		sql:     sql,
		arg:     ddlJobArg(&shardRowId),
	}
	return task, nil
}

func (c *RandSQLGen) shardRowIDJob(task *DDLJobTask) error {
	table := task.tblInfo
	shardRowId := *((*int)(task.arg))
	table.shardRowId = int64(shardRowId)
	return nil
}

func (c *RandSQLGen) prepareRebaseAutoID() (*DDLJobTask, error) {
	table := c.pickupRandomTable()
	newAutoID := rand.Int63n(1 << 48)
	sql := fmt.Sprintf("alter table `%s` auto_increment=%d", table.name, newAutoID)
	task := &DDLJobTask{
		k:       DDLRebaseAutoID,
		sql:     sql,
		tblInfo: table,
	}
	return task, nil
}

func (c *RandSQLGen) rebaseAutoIDJob(task *DDLJobTask, db *sql.DB) error {
	// The autoID might be different from what we specified in task, so instead, do
	// a simple query to fetch the AutoID.
	table := task.tblInfo
	sql := fmt.Sprintf("select auto_increment from information_schema.tables "+
		"where table_schema='test' and table_name='%s'", table.name)
	// Ignore check error, it doesn't matter.
	db.QueryRow(sql).Scan(&table.autoIncID)
	return nil
}

func (c *RandSQLGen) GenDropTable() *DDLJobTask {
	c.tablesLock.Lock()
	defer c.tablesLock.Unlock()
	tableToDrop := c.pickupRandomTable()
	if len(c.tables) <= 1 || tableToDrop == nil {
		return nil
	}
	tableToDrop.setDeleted()
	sql := fmt.Sprintf("DROP TABLE `%s`", tableToDrop.name)

	task := &DDLJobTask{
		k:       DDLDropTable,
		sql:     sql,
		tblInfo: tableToDrop,
	}
	return task
}

func (c *RandSQLGen) dropTableJob(task *DDLJobTask) error {
	c.tablesLock.Lock()
	defer c.tablesLock.Unlock()
	if c.isTableDeleted(task.tblInfo) {
		return fmt.Errorf("table %s is not exists", task.tblInfo.name)
	}
	delete(c.tables, task.tblInfo.name)
	putIDToAlloc(task.tblInfo.name, globalIDAllocator)
	return nil
}

func (c *RandSQLGen) GenCreateView() (*DDLJobTask, error) {
	table := c.pickupRandomTable()
	if table == nil {
		return nil, nil
	}
	columns := table.pickupRandomColumns()
	if len(columns) == 0 {
		return nil, nil
	}
	view := &ViewInfo{
		name:    genViewName(),
		columns: columns,
		table:   table,
	}
	sql := fmt.Sprintf("create view `%s` as select ", view.name)
	var i = 0
	for ; i < len(columns)-1; i++ {
		sql += fmt.Sprintf("`%s`, ", columns[i].name)
	}
	sql += fmt.Sprintf("`%s` from `%s`", columns[i].name, table.name)
	task := &DDLJobTask{
		k:        DDLCreateView,
		sql:      sql,
		viewInfo: view,
	}
	return task, nil
}

func (c *RandSQLGen) createViewJob(task *DDLJobTask) error {
	c.views[task.viewInfo.name] = task.viewInfo
	return nil
}

type ddlTestIndexStrategy = int

const (
	ddlTestIndexStrategyBegin ddlTestIndexStrategy = iota
	ddlTestIndexStrategySingleColumnAtBeginning
	ddlTestIndexStrategySingleColumnAtEnd
	ddlTestIndexStrategySingleColumnRandom
	ddlTestIndexStrategyMultipleColumnRandom
	ddlTestIndexStrategyEnd
)

type ddlIndexJobArg struct {
	index *IndexInfo
}

func (c *RandSQLGen) GenAddIndex() *DDLJobTask {
	table := c.pickupRandomTable()
	if table == nil {
		return nil
	}
	strategy := rand.Intn(ddlTestIndexStrategyMultipleColumnRandom) + ddlTestIndexStrategySingleColumnAtBeginning
	// build index definition
	index := IndexInfo{
		name:      genIdxName(table),
		signature: "",
		columns:   make([]*ColumnInfo, 0),
	}

	switch strategy {
	case ddlTestIndexStrategySingleColumnAtBeginning:
		if !table.columns[0].canBeIndex() {
			return nil
		}
		index.columns = append(index.columns, table.columns[0])
	case ddlTestIndexStrategySingleColumnAtEnd:
		if !table.columns[len(table.columns)-1].canBeIndex() {
			return nil
		}
		index.columns = append(index.columns, table.columns[len(table.columns)-1])
	case ddlTestIndexStrategySingleColumnRandom:
		col := table.columns[rand.Intn(len(table.columns))]
		if !col.canBeIndex() {
			return nil
		}
		index.columns = append(index.columns, col)
	case ddlTestIndexStrategyMultipleColumnRandom:
		numberOfColumns := rand.Intn(len(table.columns)) + 1
		// Multiple columns of one index should no more than 16.
		if numberOfColumns > 10 {
			numberOfColumns = 10
		}
		perm := rand.Perm(len(table.columns))[:numberOfColumns]
		for _, idx := range perm {
			if table.columns[idx].canBeIndex() {
				index.columns = append(index.columns, table.columns[idx])
			}
		}
	}

	if len(index.columns) == 0 {
		return nil
	}

	index.lens = make([]int, len(index.columns))
	for i, col := range index.columns {
		if col.k == KindChar || col.k == KindVarChar {
			index.lens[i] = rand.Intn(col.filedTypeM) + 1
		}
	}

	index.signature = index.getIndexSignature()

	// check whether index duplicates
	for _, idx := range table.indexes {
		if idx.signature == index.signature {
			return nil
		}
	}

	sql := fmt.Sprintf("ALTER TABLE `%s` ADD INDEX `%s` (", table.name, index.name)
	sql += index.signature
	sql += ")"

	arg := &ddlIndexJobArg{index: &index}
	task := &DDLJobTask{
		k:       DDLAddIndex,
		sql:     sql,
		tblInfo: table,
		arg:     ddlJobArg(arg),
	}
	return task
}

func (c *RandSQLGen) addIndexJob(task *DDLJobTask) error {
	jobArg := (*ddlIndexJobArg)(task.arg)
	tblInfo := task.tblInfo

	if c.isTableDeleted(tblInfo) {
		return fmt.Errorf("table %s is not exists", tblInfo.name)
	}

	for _, column := range jobArg.index.columns {
		if tblInfo.isColumnDeleted(column) {
			return fmt.Errorf("local Execute add index %s on column %s error , column is deleted", jobArg.index.name, column.name)
		}
	}
	tblInfo.indexes = append(tblInfo.indexes, jobArg.index)
	for _, column := range jobArg.index.columns {
		column.indexReferences++
	}
	return nil
}

type ddlRenameIndexArg struct {
	preIndex int
	newIndex string
}

func (c *RandSQLGen) GenRenameIndex() (*DDLJobTask, error) {
	table := c.pickupRandomTable()
	if table == nil || len(table.indexes) == 0 {
		return nil, nil
	}
	loc := rand.Intn(len(table.indexes))
	index := table.indexes[loc]
	newIndex := genIdxName(table)
	sql := fmt.Sprintf("ALTER TABLE `%s` RENAME INDEX `%s` to `%s`",
		table.name, index.name, newIndex)
	task := &DDLJobTask{
		k:       DDLRenameIndex,
		sql:     sql,
		tblInfo: table,
		arg:     ddlJobArg(&ddlRenameIndexArg{loc, newIndex}),
	}
	return task, nil
}

func (c *RandSQLGen) renameIndexJob(task *DDLJobTask) error {
	table := task.tblInfo
	arg := (*ddlRenameIndexArg)(task.arg)
	table.indexes[arg.preIndex].name = arg.newIndex
	return nil
}

func (c *RandSQLGen) GenDropIndex() *DDLJobTask {
	table := c.pickupRandomTable()
	if table == nil {
		return nil
	}
	if len(table.indexes) == 0 {
		return nil
	}
	indexToDropIndex := rand.Intn(len(table.indexes))
	indexToDrop := table.indexes[indexToDropIndex]
	sql := fmt.Sprintf("ALTER TABLE `%s` DROP INDEX `%s`", table.name, indexToDrop.name)

	arg := &ddlIndexJobArg{index: indexToDrop}
	task := &DDLJobTask{
		k:       DDLDropIndex,
		sql:     sql,
		tblInfo: table,
		arg:     ddlJobArg(arg),
	}
	return task
}

func (c *RandSQLGen) dropIndexJob(task *DDLJobTask) error {
	jobArg := (*ddlIndexJobArg)(task.arg)
	tblInfo := task.tblInfo

	if c.isTableDeleted(tblInfo) {
		return fmt.Errorf("table %s is not exists", tblInfo.name)
	}

	iOfDropIndex := -1
	for i := range tblInfo.indexes {
		if jobArg.index.name == tblInfo.indexes[i].name {
			iOfDropIndex = i
			break
		}
	}
	if iOfDropIndex == -1 {
		return fmt.Errorf("table %s , index %s is not exists", tblInfo.name, jobArg.index.name)
	}

	for _, column := range jobArg.index.columns {
		column.indexReferences--
		if column.indexReferences < 0 {
			return fmt.Errorf("drop index, index.column %s Unexpected index reference", column.name)
		}
	}
	tblInfo.indexes = append(tblInfo.indexes[:iOfDropIndex], tblInfo.indexes[iOfDropIndex+1:]...)
	putIDToAlloc(jobArg.index.name, tblInfo.idAlloc)
	return nil
}

type ddlTestAddDropColumnStrategy = int

const (
	ddlTestAddDropColumnStrategyBegin ddlTestAddDropColumnStrategy = iota
	ddlTestAddDropColumnStrategyAtBeginning
	ddlTestAddDropColumnStrategyAtEnd
	ddlTestAddDropColumnStrategyAtRandom
	ddlTestAddDropColumnStrategyEnd
)

type ddlTestAddDropColumnConfig struct {
	strategy ddlTestAddDropColumnStrategy
}

type ddlColumnJobArg struct {
	origColumnIndex   int // the column index we want to modify, only used for modifyColumnJob
	column            *ColumnInfo
	strategy          ddlTestAddDropColumnStrategy
	insertAfterColumn *ColumnInfo
}

func (c *RandSQLGen) GenAddColumn() *DDLJobTask {
	table := c.pickupRandomTable()
	if table == nil {
		return nil
	}
	strategy := rand.Intn(ddlTestAddDropColumnStrategyAtRandom) + ddlTestAddDropColumnStrategyAtBeginning
	newColumn := getRandDDLTestColumn(table)
	insertAfterPosition := -1
	// build SQL
	sql := fmt.Sprintf("ALTER TABLE `%s` ADD COLUMN `%s` %s", table.name, newColumn.name, newColumn.getDefinition())
	switch strategy {
	case ddlTestAddDropColumnStrategyAtBeginning:
		sql += " FIRST"
	case ddlTestAddDropColumnStrategyAtEnd:
		// do nothing
	case ddlTestAddDropColumnStrategyAtRandom:
		insertAfterPosition = rand.Intn(len(table.columns))
		sql += fmt.Sprintf(" AFTER `%s`", table.columns[insertAfterPosition].name)
	}

	arg := &ddlColumnJobArg{
		column:   newColumn,
		strategy: strategy,
	}
	if insertAfterPosition != -1 {
		arg.insertAfterColumn = table.columns[insertAfterPosition]
	}
	return &DDLJobTask{
		k:       DDLAddColumn,
		sql:     sql,
		tblInfo: table,
		arg:     ddlJobArg(arg),
	}
}

func (c *RandSQLGen) addColumnJob(task *DDLJobTask) error {
	jobArg := (*ddlColumnJobArg)(task.arg)
	table := task.tblInfo

	if c.isTableDeleted(table) {
		return fmt.Errorf("table %s is not exists", table.name)
	}
	newColumn := jobArg.column
	strategy := jobArg.strategy

	newColumn.rows = make([]interface{}, table.numberOfRows)
	for i := 0; i < table.numberOfRows; i++ {
		newColumn.rows[i] = newColumn.defaultValue
	}

	switch strategy {
	case ddlTestAddDropColumnStrategyAtBeginning:
		table.columns = append([]*ColumnInfo{newColumn}, table.columns...)
	case ddlTestAddDropColumnStrategyAtEnd:
		table.columns = append(table.columns, newColumn)
	case ddlTestAddDropColumnStrategyAtRandom:
		insertAfterPosition := -1
		for i := range table.columns {
			if jobArg.insertAfterColumn.name == table.columns[i].name {
				insertAfterPosition = i
				break
			}
		}
		if insertAfterPosition == -1 {
			return fmt.Errorf("table %s ,insert column %s after column, column %s is not exists ", table.name, newColumn.name, jobArg.insertAfterColumn.name)
		}
		table.columns = append(table.columns[:insertAfterPosition+1], append([]*ColumnInfo{newColumn}, table.columns[insertAfterPosition+1:]...)...)
	}
	return nil
}

func (c *RandSQLGen) GenModifyColumn() *DDLJobTask {
	table := c.pickupRandomTable()
	if table == nil {
		return nil
	}
	origColIndex, origColumn := table.pickupRandomColumn()
	if origColumn == nil || !origColumn.canBeModified() || origColumn.isGenerated() || origColumn.isPrimaryKey {
		return nil
	}
	var modifiedColumn *ColumnInfo
	var sql string
	if rand.Float64() > 0.5 {
		// If a column has dependency, it cannot be renamed.
		if origColumn.hasGenerateCol() {
			return nil
		}
		renameCol := true
		if table.isPartitionColumn(origColumn.name) {
			renameCol = false
		}
		modifiedColumn = generateRandModifiedColumn(table, origColumn, renameCol)
		sql = fmt.Sprintf("alter table `%s` change column `%s` `%s` %s", table.name,
			origColumn.name, modifiedColumn.name, modifiedColumn.getDefinition())
	} else {
		modifiedColumn = generateRandModifiedColumn(table, origColumn, false)
		sql = fmt.Sprintf("alter table `%s` modify column `%s` %s", table.name,
			origColumn.name, modifiedColumn.getDefinition())
	}
	if origColumn.k == KindDECIMAL {
		return nil
	}
	modifiedColumn.rows = origColumn.rows
	modifiedColumn.indexReferences = origColumn.indexReferences
	origColumn.setDeleted()
	strategy := rand.Intn(ddlTestAddDropColumnStrategyAtRandom) + ddlTestAddDropColumnStrategyAtBeginning
	var insertAfterColumn *ColumnInfo = nil
	switch strategy {
	case ddlTestAddDropColumnStrategyAtBeginning:
		sql += " FIRST"
	case ddlTestAddDropColumnStrategyAtEnd:
		endColumn := table.columns[len(table.columns)-1]
		if endColumn.name != origColumn.name {
			sql += fmt.Sprintf(" AFTER `%s`", endColumn.name)
		}
	case ddlTestAddDropColumnStrategyAtRandom:
		insertPosition := rand.Intn(len(table.columns))
		insertAfterColumn = table.columns[insertPosition]
		if insertPosition != origColIndex {
			sql += fmt.Sprintf(" AFTER `%s`", insertAfterColumn.name)
		}
	}
	return &DDLJobTask{
		k:       DDLModifyColumn,
		tblInfo: table,
		sql:     sql,
		arg: ddlJobArg(&ddlColumnJobArg{
			origColumnIndex:   origColIndex,
			column:            modifiedColumn,
			strategy:          strategy,
			insertAfterColumn: insertAfterColumn,
		}),
	}
}

func (c *RandSQLGen) modifyColumnJob(task *DDLJobTask) error {
	table := task.tblInfo
	arg := (*ddlColumnJobArg)(task.arg)
	table.columns = append(table.columns[:arg.origColumnIndex], table.columns[arg.origColumnIndex+1:]...)
	switch arg.strategy {
	case ddlTestAddDropColumnStrategyAtBeginning:
		table.columns = append([]*ColumnInfo{arg.column}, table.columns...)
	case ddlTestAddDropColumnStrategyAtEnd:
		table.columns = append(table.columns, arg.column)
	case ddlTestAddDropColumnStrategyAtRandom:
		insertPosition := arg.origColumnIndex - 1
		for i, col := range table.columns {
			if col.name == arg.insertAfterColumn.name {
				insertPosition = i
				break
			}
		}
		table.columns = append(table.columns[:insertPosition+1],
			append([]*ColumnInfo{arg.column}, table.columns[insertPosition+1:]...)...)
	}
	return nil
}

func (c *RandSQLGen) GenDropColumn() *DDLJobTask {
	table := c.pickupRandomTable()
	if table == nil {
		return nil
	}

	columnsSnapshot := table.filterColumns(table.predicateAll)
	if len(columnsSnapshot) <= 1 {
		return nil
	}

	strategy := rand.Intn(ddlTestAddDropColumnStrategyAtRandom) + ddlTestAddDropColumnStrategyAtBeginning
	columnToDropIndex := -1
	switch strategy {
	case ddlTestAddDropColumnStrategyAtBeginning:
		columnToDropIndex = 0
	case ddlTestAddDropColumnStrategyAtEnd:
		columnToDropIndex = len(table.columns) - 1
	case ddlTestAddDropColumnStrategyAtRandom:
		columnToDropIndex = rand.Intn(len(table.columns))
	}

	columnToDrop := table.columns[columnToDropIndex]

	// Primary key columns cannot be dropped
	if columnToDrop.isPrimaryKey {
		return nil
	}

	// partition key columns cannot be dropped.
	if columnToDrop.isPartitionKey {
		return nil
	}

	// Column cannot be dropped if the column has generated column dependency
	if columnToDrop.hasGenerateCol() {
		return nil
	}

	// We does not support dropping a column with index
	if columnToDrop.indexReferences > 0 {
		return nil
	}
	columnToDrop.setDeleted()
	sql := fmt.Sprintf("ALTER TABLE `%s` DROP COLUMN `%s`", table.name, columnToDrop.name)

	arg := &ddlColumnJobArg{
		column: columnToDrop,
	}
	return &DDLJobTask{
		k:       DDLDropColumn,
		sql:     sql,
		tblInfo: table,
		arg:     ddlJobArg(arg),
	}
}

func (c *RandSQLGen) dropColumnJob(task *DDLJobTask) error {
	jobArg := (*ddlColumnJobArg)(task.arg)
	table := task.tblInfo
	if c.isTableDeleted(table) {
		return fmt.Errorf("table %s is not exists", table.name)
	}
	columnToDrop := jobArg.column
	if columnToDrop.indexReferences > 0 {
		columnToDrop.setDeletedRecover()
		return fmt.Errorf("local Execute drop column %s on table %s error , column has index reference", jobArg.column.name, table.name)
	}
	dropColumnPosition := -1
	for i := range table.columns {
		if columnToDrop.name == table.columns[i].name {
			dropColumnPosition = i
			break
		}
	}
	if dropColumnPosition == -1 {
		return fmt.Errorf("table %s ,drop column , column %s is not exists ", table.name, columnToDrop.name)
	}
	// update table definitions
	table.columns = append(table.columns[:dropColumnPosition], table.columns[dropColumnPosition+1:]...)
	// if the drop column is a generated column , we should update the dependency column
	if columnToDrop.isGenerated() {
		col := columnToDrop.dependency
		i := 0
		for i = range col.dependenciedCols {
			if col.dependenciedCols[i].name == columnToDrop.name {
				break
			}
		}
		col.dependenciedCols = append(col.dependenciedCols[:i], col.dependenciedCols[i+1:]...)
	}
	putIDToAlloc(columnToDrop.name, table.idAlloc)
	return nil
}

type ddlSetDefaultValueArg struct {
	column          *ColumnInfo
	newDefaultValue interface{}
}

func (c *RandSQLGen) GenSetDefaultValue() *DDLJobTask {
	table := c.pickupRandomTable()
	if table == nil {
		return nil
	}
	columns := table.filterColumns(table.predicateAll)
	if len(columns) == 0 {
		return nil
	}
	loc := rand.Intn(len(columns))
	column := columns[loc]
	// If the chosen column cannot have default value, just return nil.
	if !column.canHaveDefaultValue() {
		return nil
	}
	newDefaultValue := column.randValue()
	sql := fmt.Sprintf("ALTER TABLE `%s` ALTER `%s` SET DEFAULT %s", table.name,
		column.name, getDefaultValueString(column.k, newDefaultValue))
	return &DDLJobTask{
		k:       DDLSetDefaultValue,
		sql:     sql,
		tblInfo: table,
		arg: ddlJobArg(&ddlSetDefaultValueArg{
			column:          column,
			newDefaultValue: newDefaultValue,
		}),
	}
}

func (c *RandSQLGen) setDefaultValueJob(task *DDLJobTask) error {
	arg := (*ddlSetDefaultValueArg)(task.arg)
	arg.column.defaultValue = arg.newDefaultValue
	return nil
}