// 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 (
	"bytes"
	"encoding/json"
	"fmt"
	"math"
	"math/rand"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

func init() {
	TestFieldType = []int{
		KindTINYINT,
		KindSMALLINT,
		KindMEDIUMINT,
		KindInt32,
		KindBigInt,

		//KindBit,

		// Float in TiDB and MySQL is different.
		//KindFloat,
		//KindDouble,

		KindDECIMAL,

		KindChar,
		KindVarChar,

		KindBLOB,
		KindTEXT,

		KindBool,

		KindDATE,
		KindTIME,
		KindDATETIME,
		KindTIMESTAMP,
		KindYEAR,

		//KindJSON, // have `admin check table when index is virtual generated column` bug unfix
		KindEnum,
		KindSet,
	}

}

type RandSQLGen struct {
	tables           map[string]*TableInfo
	schemas          map[string]*SchemaInfo
	views            map[string]*ViewInfo
	tablesLock       sync.RWMutex
	stop             int32
	lastDDLID        int
	charsets         []string
	charsetsCollates map[string][]string
}

func NewRandSQLGen() *RandSQLGen {
	return &RandSQLGen{
		tables:  make(map[string]*TableInfo),
		schemas: make(map[string]*SchemaInfo),
		views:   make(map[string]*ViewInfo),
		stop:    0,
	}
}

type ErrorConflict struct{}

func (err ErrorConflict) Error() string {
	return "Conflict operation"
}

func (c *RandSQLGen) StopTest() {
	atomic.StoreInt32(&c.stop, 1)
}

func (c *RandSQLGen) IsStop() bool {
	return atomic.LoadInt32(&c.stop) == 1
}

// schema type, this might be modified to support other operations, but it's not clear
// currently for me.
type SchemaInfo struct {
	name    string
	deleted bool
	charset string
	collate string
}

func (c *RandSQLGen) isSchemaDeleted(schema *SchemaInfo) bool {
	if _, ok := c.schemas[schema.name]; ok {
		return false
	}
	return true
}

func (schema *SchemaInfo) setDeleted() {
	schema.deleted = true
}

func (schema *SchemaInfo) isDeleted() bool {
	return schema.deleted
}

// pickupRandomSchema picks a schema randomly from `c.schemas`.
func (c *RandSQLGen) pickupRandomSchema() *SchemaInfo {
	schemaLen := len(c.schemas)
	if schemaLen == 0 {
		return nil
	}
	loc := rand.Intn(schemaLen)
	for _, schema := range c.schemas {
		if loc == 0 {
			return schema
		}
		loc--
	}
	return nil
}

// pickupRandomTables picks a table randomly. The callee should ensure that
// during this function call the table list is not modified.
//
// Normally the DML op callee should acquire a lock before calling this function
// because the table list may be modified by another parallel DDL op. However
// the DDL op callee doesn't need to acquire a lock because no one will modify the
// table list in parallel ---- DDL ops are executed one by one.
func (c *RandSQLGen) pickupRandomTable() *TableInfo {
	tableNames := make([]string, 0)
	for name, table := range c.tables {
		if table.isDeleted() {
			continue
		}
		tableNames = append(tableNames, name)
	}
	if len(tableNames) == 0 {
		return nil
	}
	name := tableNames[rand.Intn(len(tableNames))]
	return c.tables[name]
}

func (c *RandSQLGen) pickupRandomPartitionTable() *TableInfo {
	tableNames := make([]string, 0)
	for name, table := range c.tables {
		if table.isDeleted() {
			continue
		}
		if !table.isPartitionTable() {
			continue
		}
		tableNames = append(tableNames, name)
	}
	if len(tableNames) == 0 {
		return nil
	}
	name := tableNames[rand.Intn(len(tableNames))]
	return c.tables[name]

}

func (c *RandSQLGen) pickupRandomRangePartitionTable() *TableInfo {
	tableNames := make([]string, 0)
	for name, table := range c.tables {
		if table.isDeleted() {
			continue
		}
		if !table.isPartitionTable() || table.partition.tp != partitionTypeRange {
			continue
		}
		tableNames = append(tableNames, name)
	}
	if len(tableNames) == 0 {
		return nil
	}
	name := tableNames[rand.Intn(len(tableNames))]
	return c.tables[name]

}

func (c *RandSQLGen) pickupRandomCharsetAndCollate() (string, string) {
	charset := c.charsets[rand.Intn(len(c.charsets))]
	collations := c.charsetsCollates[charset]
	collation := collations[rand.Intn(len(collations))]
	return charset, collation
}

func (c *RandSQLGen) isTableDeleted(table *TableInfo) bool {
	if _, ok := c.tables[table.name]; ok {
		return false
	}
	return true
}

type TableInfo struct {
	deleted           int32
	name              string
	id                string // table_id , get from admin show ddl jobs
	columns           []*ColumnInfo
	indexes           []*IndexInfo
	numberOfRows      int
	shardRowId        int64 // shard_row_id_bits
	autoIncID         int64
	comment           string // table comment
	charset           string
	collate           string
	partition         *partitionInfo
	partitionColsName []string
	lock              sync.RWMutex
	idAlloc           *allocator
}

type partitionType int

const (
	partitionTypeHash partitionType = iota
	partitionTypeRange
)

type partitionInfo struct {
	tp      partitionType
	num     int   // partition number.
	values  []int // range partition values
	dropped []bool
	column  *ColumnInfo
}

func (table *TableInfo) isPartitionTable() bool {
	return table.partition != nil
}

func (table *TableInfo) setPartitionColumns(cols []*ColumnInfo) {
	for _, col := range cols {
		table.partitionColsName = append(table.partitionColsName, col.name)
	}
}

func (table *TableInfo) isPartitionColumn(colName string) bool {
	for _, name := range table.partitionColsName {
		if name == colName {
			return true
		}
	}
	return false
}

func (table *TableInfo) pickRandPartitionNames() string {
	// tidb not support truncate multi partition now.
	//num := rand.Intn(table.partitionNum) + 1
	num := 1
	s := ""
	for i := 0; i < num; i++ {
		if i > 0 {
			s += ", "
		}
		s += fmt.Sprintf("p%v", table.pickRandPartition())
	}
	return s
}

func (table *TableInfo) pickRandPartition() int {
	for i := 0; i < table.partition.num; i++ {
		p := rand.Intn(table.partition.num)
		if table.partition.tp == partitionTypeRange && table.partition.dropped[p] {
			continue
		}
		return p
	}
	for i := 0; i < table.partition.num; i++ {
		if table.partition.dropped[i] {
			continue
		}
		return i
	}
	// should never run to here.
	return -1
}

func (table *TableInfo) genAddRangePartitionValue() (int, bool) {
	for i := len(table.partition.dropped) - 1; i >= 0; i-- {
		if table.partition.dropped[i] {
			continue
		}
		if table.partition.values[i] == math.MaxInt64 {
			return 0, false
		}
		return genRangeValue(table.partition.column, table.partition.values[i]), true
	}
	return 0, false
}

func (table *TableInfo) isDeleted() bool {
	return atomic.LoadInt32(&table.deleted) != 0
}

func (table *TableInfo) setDeleted() {
	atomic.StoreInt32(&table.deleted, 1)
}

func (table *TableInfo) filterColumns(predicate func(*ColumnInfo) bool) []*ColumnInfo {
	retColumns := make([]*ColumnInfo, 0)
	for index, col := range table.columns {
		if predicate(col) && !col.isDeleted() {
			retColumns = append(retColumns, table.columns[index])
		}
	}
	return retColumns
}

func (table *TableInfo) predicateAll(col *ColumnInfo) bool {
	return true
}

func (table *TableInfo) predicateNotGenerated(col *ColumnInfo) bool {
	return col.notGenerated()
}

func (table *TableInfo) predicatePrimaryKey(col *ColumnInfo) bool {
	return col.isPrimaryKey
}

func (table *TableInfo) predicateNonPrimaryKey(col *ColumnInfo) bool {
	return !col.isPrimaryKey
}

func (table *TableInfo) predicateNonPrimaryKeyAndCanBeWhere(col *ColumnInfo) bool {
	return !col.isPrimaryKey && col.canBeWhere()
}

func (table *TableInfo) predicateNonPrimaryKeyAndNotGen(col *ColumnInfo) bool {
	return !col.isPrimaryKey && col.notGenerated()
}

// pickupRandomColumns picks columns from `table.columns` randomly and return them.
func (table *TableInfo) pickupRandomColumns() []*ColumnInfo {
	columns := table.filterColumns(table.predicateAll)
	pickedCols := make([]*ColumnInfo, 0)
	for _, col := range columns {
		if rand.Float64() > 0.6 {
			pickedCols = append(pickedCols, col)
		}
	}
	return pickedCols
}

// pickupRandomColumn picks a column from `table.columns` randomly and
// returns the column index and the column.
func (table *TableInfo) pickupRandomColumn() (int, *ColumnInfo) {
	columns := table.filterColumns(table.predicateAll)
	if len(columns) == 0 {
		return 0, nil
	}
	index := rand.Intn(len(columns))
	return index, columns[index]
}

// isColumnDeleted checks the col is deleted in this table
// col.isDeleted() will be true before when dropColumnJob(),
// but the col is really deleted after remote TiDB successful execute drop column ddl, and then, the col will be deleted from table.columns.
func (table *TableInfo) isColumnDeleted(col *ColumnInfo) bool {
	for i := range table.columns {
		if col.name == table.columns[i].name {
			return false
		}
	}
	return true
}

func (table *TableInfo) DebugPrintToString() string {
	var buffer bytes.Buffer
	table.lock.RLock()
	buffer.WriteString(fmt.Sprintf("======== DEBUG BEGIN  ========\n"))
	buffer.WriteString(fmt.Sprintf("Dumping expected contents for table `%s`:\n", table.name))
	if table.isDeleted() {
		buffer.WriteString("[WARN] This table is marked as DELETED.\n")
	}
	buffer.WriteString(fmt.Sprintf("Comment: %s\nCharset: %s, Collate: %s\nShardRowId: %d\nAutoID: %d\n",
		table.comment, table.charset, table.collate, table.shardRowId, table.autoIncID))
	buffer.WriteString("## Non-Primary Indexes: \n")
	for i, index := range table.indexes {
		buffer.WriteString(fmt.Sprintf("Index #%d: Name = `%s`, Columnns = [", i, index.name))
		for _, column := range index.columns {
			buffer.WriteString(fmt.Sprintf("`%s`, ", column.name))
		}
		buffer.WriteString("]\n")
	}
	buffer.WriteString("## Columns: \n")
	for i, column := range table.columns {
		buffer.WriteString(fmt.Sprintf("Column #%d", i))
		if column.isDeleted() {
			buffer.WriteString(" [DELETED]")
		}
		buffer.WriteString(fmt.Sprintf(": Name = `%s`, Definition = %s, isPrimaryKey = %v, used in %d indexes\n",
			column.name, column.getDefinition(), column.isPrimaryKey, column.indexReferences))
	}
	buffer.WriteString(fmt.Sprintf("## Values (number of rows = %d): \n", table.numberOfRows))
	for i := 0; i < table.numberOfRows; i++ {
		buffer.WriteString("#")
		buffer.WriteString(PadRight(fmt.Sprintf("%d", i), " ", 14))
		buffer.WriteString(": ")
		for _, col := range table.columns {
			if len(col.rows) <= i {
				continue
			}
			buffer.WriteString(PadLeft(fmt.Sprintf("%v", col.rows[i]), " ", 11))
			buffer.WriteString(", ")
		}
		buffer.WriteString("\n")
	}
	if table.isPartitionTable() {
		buffer.WriteString("======== partition info ========\n")
		buffer.WriteString(fmt.Sprintf("partition num: %v, type: %v, column: %v\n", table.partition.num, table.partition.tp, table.partition.column.name))
		buffer.WriteString(fmt.Sprintf("partition value: %v\n", table.partition.values))
		buffer.WriteString(fmt.Sprintf("partition droped: %v\n", table.partition.dropped))

	}
	buffer.WriteString("======== DEBUG END ========\n")
	table.lock.RUnlock()
	return buffer.String()
}

type ViewInfo struct {
	name    string
	columns []*ColumnInfo
	table   *TableInfo // the table that this view references.
}

type ColumnDescriptor struct {
	column *ColumnInfo
	value  interface{}
}

func (ddlt *ColumnDescriptor) getValueString() string {
	// make bit data visible
	if ddlt.column.k == KindBit {
		return fmt.Sprintf("b'%v'", ddlt.value)
	} else {
		return fmt.Sprintf("'%v'", ddlt.value)
	}
}

func (ddlt *ColumnDescriptor) buildConditionSQL() string {
	var sql string
	if ddlt.value == NullValue || ddlt.value == nil {
		sql += fmt.Sprintf("`%s` IS NULL", ddlt.column.name)
	} else {
		switch ddlt.column.k {
		case KindFloat:
			sql += fmt.Sprintf("abs(`%s` - %v) < 0.0000001", ddlt.column.name, ddlt.getValueString())
		case KindDouble:
			sql += fmt.Sprintf("abs(`%s` - %v) < 0.0000000000000001", ddlt.column.name, ddlt.getValueString())
		default:
			sql += fmt.Sprintf("`%s` = %v", ddlt.column.name, ddlt.getValueString())
		}
	}
	return sql
}

type ColumnInfo struct {
	k         int
	deleted   int32
	name      string
	fieldType string

	filedTypeM      int //such as:  VARCHAR(10) ,    filedTypeM = 10
	filedTypeD      int //such as:  DECIMAL(10,5) ,  filedTypeD = 5
	filedPrecision  int
	defaultValue    interface{}
	isPrimaryKey    bool
	isPartitionKey  bool
	rows            []interface{}
	indexReferences int

	dependenciedCols []*ColumnInfo
	dependency       *ColumnInfo
	mValue           map[string]interface{}
	nameOfGen        string

	setValue []string //for enum , set data type
}

func (col *ColumnInfo) isDeleted() bool {
	return atomic.LoadInt32(&col.deleted) != 0
}

func (col *ColumnInfo) setDeleted() {
	atomic.StoreInt32(&col.deleted, 1)
}

func (col *ColumnInfo) setDeletedRecover() {
	atomic.StoreInt32(&col.deleted, 0)
}

func (col *ColumnInfo) getMatchedColumnDescriptor(descriptors []*ColumnDescriptor) *ColumnDescriptor {
	for _, d := range descriptors {
		if d.column == col {
			return d
		}
	}
	return nil
}

func (col *ColumnInfo) getDefinition() string {
	if col.isPrimaryKey {
		return col.fieldType
	}

	if col.isGenerated() {
		return fmt.Sprintf("%s AS (JSON_EXTRACT(`%s`,'$.%s'))", col.fieldType, col.dependency.name, col.nameOfGen)
	}

	if col.canHaveDefaultValue() {
		return fmt.Sprintf("%s NULL DEFAULT %v", col.fieldType, getDefaultValueString(col.k, col.defaultValue))
	} else {
		return fmt.Sprintf("%s NULL", col.fieldType)
	}

}

func (col *ColumnInfo) getSelectName() string {
	if col.k == KindBit {
		return fmt.Sprintf("bin(`%s`)", col.name)
	} else {
		return fmt.Sprintf("`%s`", col.name)
	}
}

// getDefaultValueString returns a string representation of `defaultValue` according
// to the type `k` of column.
func getDefaultValueString(k int, defaultValue interface{}) string {
	if k == KindBit {
		return fmt.Sprintf("b'%v'", defaultValue)
	} else {
		return fmt.Sprintf("'%v'", defaultValue)
	}
}

func (col *ColumnInfo) isEqual(r int, str string) bool {
	vstr := fmt.Sprintf("%v", col.rows[r])
	return strings.Compare(vstr, str) == 0
}

func (col *ColumnInfo) getDependenciedColsValue(genCol *ColumnInfo) interface{} {
	if col.mValue == nil {
		return nil
	}
	v := col.mValue[genCol.nameOfGen]
	switch genCol.k {
	case KindChar, KindVarChar, KindTEXT, KindBLOB:
		v = fmt.Sprintf("\"%v\"", v)
	}
	return v
}

// canBeModified returns whether this column can be changed by a SQL query `change column` or `modify column`.
// Only a few type columns are supported. See https://pingcap.com/docs/sql/ddl/ for more detail.
func (col *ColumnInfo) canBeModified() bool {
	typeSupported := false
	switch col.k {
	case KindTINYINT, KindSMALLINT, KindMEDIUMINT, KindInt32, KindBigInt:
		typeSupported = true
	case KindDECIMAL:
		typeSupported = true
	case KindChar, KindVarChar:
		typeSupported = true
	case KindTEXT, KindBLOB:
		typeSupported = true
	}
	return typeSupported
}

func getDDLTestColumn(n int, table *TableInfo) *ColumnInfo {
	column := &ColumnInfo{
		k:         n,
		name:      genColName(table),
		fieldType: ALLFieldType[n],
		rows:      make([]interface{}, 0),
		deleted:   0,
	}
	switch n {
	case KindChar, KindVarChar, KindBLOB, KindTEXT, KindBit:
		maxLen := GetMaxLenByKind(n)
		column.filedTypeM = int(rand.Intn(maxLen))
		for column.filedTypeM == 0 && column.k == KindBit {
			column.filedTypeM = int(rand.Intn(maxLen))
		}

		for column.filedTypeM < 3 && column.k != KindBit { // len('""') = 2
			column.filedTypeM = int(rand.Intn(maxLen))
		}
		column.fieldType = fmt.Sprintf("%s(%d)", ALLFieldType[n], column.filedTypeM)
	case KindTINYBLOB, KindMEDIUMBLOB, KindLONGBLOB, KindTINYTEXT, KindMEDIUMTEXT, KindLONGTEXT:
		column.filedTypeM = GetMaxLenByKind(n)
		column.fieldType = fmt.Sprintf("%s(%d)", ALLFieldType[n], column.filedTypeM)
	case KindDECIMAL:
		column.filedTypeM, column.filedTypeD = RandMD()
		column.fieldType = fmt.Sprintf("%s(%d,%d)", ALLFieldType[n], column.filedTypeM, column.filedTypeD)
	case KindEnum, KindSet:
		maxLen := GetMaxLenByKind(n)
		l := maxLen + 1
		column.setValue = make([]string, l)
		m := make(map[string]struct{})
		column.fieldType += "("
		for i := 0; i < l; i++ {
			column.setValue[i] = RandEnumString(m)
			if i > 0 {
				column.fieldType += ", "
			}
			column.fieldType += fmt.Sprintf("\"%s\"", column.setValue[i])
		}
		column.fieldType += ")"
	}

	if column.canHaveDefaultValue() {
		column.defaultValue = column.randValue()
	}

	return column
}

func getRandDDLTestColumn(table *TableInfo) *ColumnInfo {
	var n int
	for {
		n = RandDataType()
		if n != KindJSON {
			break
		}
	}
	return getDDLTestColumn(n, table)
}

// generateRandModifiedColumn returns a random column to modify column `col`.
// It doesn't change any properties of column `col` and instead, it first
// generates a copy of column `col` and then modifies some properties of the
// generated column randomly. The parameter `renameCol` indicates whether to modify
// column name.
func generateRandModifiedColumn(table *TableInfo, col *ColumnInfo, renameCol bool) *ColumnInfo {
	// Shadow copy of column col.
	modifiedColumn := *col
	if renameCol {
		modifiedColumn.name = genColName(table)
	} else {
		modifiedColumn.name = col.name
	}
	// ModifyColumn operator has some limitations.
	// See https://pingcap.com/docs/sql/ddl/ for more detail.
	// For example, only supports type changes where original type
	// is belong to integer types, string types or Blob types
	// and only supports extending length of the original type.
	// Charset and collate cannot be changed.
	switch col.k {
	case KindTINYINT, KindSMALLINT, KindMEDIUMINT, KindInt32, KindBigInt:
		modifiedColumn.k = rand.Intn(KindBigInt-col.k+1) + col.k
		modifiedColumn.fieldType = ALLFieldType[modifiedColumn.k]
	case KindDECIMAL:
		modifiedColumn.filedTypeM, modifiedColumn.filedTypeD = RandMDN(col.filedTypeM, col.filedTypeD)
		modifiedColumn.fieldType = fmt.Sprintf("%s(%d, %d)", ALLFieldType[modifiedColumn.k],
			modifiedColumn.filedTypeM, modifiedColumn.filedTypeD)
	case KindChar, KindVarChar:
		modifiedColumn.k = rand.Intn(KindVarChar-col.k+1) + col.k
		modifiedColumn.filedTypeM = rand.Intn(GetMaxLenByKind(modifiedColumn.k)-col.filedTypeM+1) + col.filedTypeM
		modifiedColumn.fieldType = fmt.Sprintf("%s(%d)", ALLFieldType[modifiedColumn.k], modifiedColumn.filedTypeM)
	case KindTEXT, KindBLOB:
		// Only types on `testFieldType` are considered.
		// See ./datatype.go for more detail.
		modifiedColumn.filedTypeM = rand.Intn(GetMaxLenByKind(col.k)-col.filedTypeM+1) + col.filedTypeM
		modifiedColumn.fieldType = fmt.Sprintf("%s(%d)", ALLFieldType[col.k], modifiedColumn.filedTypeM)
	}
	if modifiedColumn.canHaveDefaultValue() {
		modifiedColumn.defaultValue = modifiedColumn.randValue()
	}
	return &modifiedColumn
}

func getRandDDLTestColumnForJson(table *TableInfo) *ColumnInfo {
	var n int
	for {
		n = RandDataType()
		if n != KindJSON && n != KindBit && n != KindSet && n != KindEnum {
			break
		}
	}
	return getDDLTestColumn(n, table)
}

func getRandDDLTestColumns(table *TableInfo) []*ColumnInfo {
	n := RandDataType()
	cols := make([]*ColumnInfo, 0)

	if n == KindJSON {
		cols = getRandJsonCol(table)
	} else {
		column := getDDLTestColumn(n, table)
		cols = append(cols, column)
	}
	return cols
}

func getRandJsonCol(table *TableInfo) []*ColumnInfo {
	fieldNum := rand.Intn(JsonFieldNum) + 1

	cols := make([]*ColumnInfo, 0, fieldNum+1)

	column := &ColumnInfo{
		k:         KindJSON,
		name:      genColName(table),
		fieldType: ALLFieldType[KindJSON],
		rows:      make([]interface{}, 0),
		deleted:   0,

		dependenciedCols: make([]*ColumnInfo, 0, fieldNum),
	}

	m := make(map[string]interface{}, 0)
	for i := 0; i < fieldNum; i++ {
		col := getRandDDLTestColumnForJson(table)
		col.nameOfGen = RandFieldName(m)
		m[col.nameOfGen] = col.randValue()
		col.dependency = column

		column.dependenciedCols = append(column.dependenciedCols, col)
		cols = append(cols, col)
	}
	column.mValue = m
	cols = append(cols, column)
	return cols
}

func (col *ColumnInfo) isGenerated() bool {
	return col.dependency != nil
}

func (col *ColumnInfo) notGenerated() bool {
	return col.dependency == nil
}

func (col *ColumnInfo) hasGenerateCol() bool {
	return len(col.dependenciedCols) > 0
}

// randValue return a rand value of the column
func (col *ColumnInfo) randValue() interface{} {
	switch col.k {
	case KindTINYINT:
		return rand.Int31n(1<<8) - 1<<7
	case KindSMALLINT:
		return rand.Int31n(1<<16) - 1<<15
	case KindMEDIUMINT:
		return rand.Int31n(1<<24) - 1<<23
	case KindInt32:
		return rand.Int63n(1<<32) - 1<<31
	case KindBigInt:
		if rand.Intn(2) == 1 {
			return rand.Int63()
		}
		return -1 - rand.Int63()
	case KindBit:
		if col.filedTypeM >= 64 {
			return fmt.Sprintf("%b", rand.Uint64())
		} else {
			m := col.filedTypeM
			if col.filedTypeM > 7 { // it is a bug
				m = m - 1
			}
			n := (int64)((1 << (uint)(m)) - 1)
			return fmt.Sprintf("%b", rand.Int63n(n))
		}
	case KindFloat:
		return rand.Float32() + 1
	case KindDouble:
		return rand.Float64() + 1
	case KindDECIMAL:
		return RandDecimal(col.filedTypeM, col.filedTypeD)
	case KindChar, KindVarChar, KindBLOB, KindTINYBLOB, KindMEDIUMBLOB, KindLONGBLOB, KindTEXT, KindTINYTEXT, KindMEDIUMTEXT, KindLONGTEXT:
		if col.filedTypeM == 0 {
			return ""
		} else {
			if col.isGenerated() {
				if col.filedTypeM <= 2 {
					return ""
				}
				return RandSeq(rand.Intn(col.filedTypeM - 2))
			}
			return RandSeq(rand.Intn(col.filedTypeM))
		}
	case KindBool:
		return rand.Intn(2)
	case KindDATE:
		randTime := time.Unix(MinDATETIME.Unix()+rand.Int63n(GapDATETIMEUnix), 0)
		return randTime.Format(TimeFormatForDATE)
	case KindTIME:
		randTime := time.Unix(MinTIMESTAMP.Unix()+rand.Int63n(GapTIMESTAMPUnix), 0)
		return randTime.Format(TimeFormatForTIME)
	case KindDATETIME:
		randTime := randTime(MinDATETIME, GapDATETIMEUnix)
		return randTime.Format(TimeFormat)
	case KindTIMESTAMP:
		randTime := randTime(MinTIMESTAMP, GapTIMESTAMPUnix)
		return randTime.Format(TimeFormat)
	case KindYEAR:
		return rand.Intn(254) + 1901 //1901 ~ 2155
	case KindJSON:
		return col.randJsonValue()
	case KindEnum:
		i := rand.Intn(len(col.setValue))
		return col.setValue[i]
	case KindSet:
		var l int
		for l == 0 {
			l = rand.Intn(len(col.setValue))
		}
		idxs := make([]int, l)
		m := make(map[int]struct{})
		for i := 0; i < l; i++ {
			idx := rand.Intn(len(col.setValue))
			_, ok := m[idx]
			for ok {
				idx = rand.Intn(len(col.setValue))
				_, ok = m[idx]
			}
			m[idx] = struct{}{}
			idxs[i] = idx
		}
		sort.Ints(idxs)
		s := ""
		for i := range idxs {
			if i > 0 {
				s += ","
			}
			s += col.setValue[idxs[i]]
		}
		return s
	default:
		return nil
	}
}

func randTime(minTime time.Time, gap int64) time.Time {
	// https://github.com/chronotope/chrono-tz/issues/23
	// see all invalid time: https://timezonedb.com/time-zones/Asia/Shanghai
	var randTime time.Time
	for {
		randTime = time.Unix(minTime.Unix()+rand.Int63n(gap), 0).In(Local)
		if NotAmbiguousTime(randTime) {
			break
		}
	}
	return randTime
}

func (col *ColumnInfo) randJsonValue() string {
	for _, dCol := range col.dependenciedCols {
		col.mValue[dCol.nameOfGen] = dCol.randValue()
	}
	jsonRow, _ := json.Marshal(col.mValue)
	return string(jsonRow)
}

// randValueUnique use for primary key column to get unique value
func (col *ColumnInfo) randValueUnique(PreValue []interface{}) (interface{}, bool) {
	// retry times
	for i := 0; i < 10; i++ {
		v := col.randValue()
		flag := true
		for _, pv := range PreValue {
			if v == pv {
				flag = false
				break
			}
		}
		if flag {
			return v, true
		}
	}
	return nil, false
}

func (col *ColumnInfo) canBePrimary() bool {
	return col.canBeIndex() && col.notGenerated()
}

func (col *ColumnInfo) canBePartitionType() bool {
	switch col.k {
	case KindTINYINT, KindSMALLINT, KindMEDIUMINT, KindInt32, KindBigInt, KindBool:
		return true
	default:
		return false
	}
}

func (col *ColumnInfo) canBeIndex() bool {
	switch col.k {
	case KindChar, KindVarChar:
		if col.filedTypeM == 0 {
			return false
		} else {
			return true
		}
	case KindBLOB, KindTINYBLOB, KindMEDIUMBLOB, KindLONGBLOB, KindTEXT, KindTINYTEXT, KindMEDIUMTEXT, KindLONGTEXT, KindJSON:
		return false
	default:
		return true
	}
}

func (col *ColumnInfo) canBeSet() bool {
	return col.notGenerated()
}

func (col *ColumnInfo) canBeWhere() bool {
	switch col.k {
	case KindJSON:
		return false
	default:
		return true
	}
}

//BLOB, TEXT, GEOMETRY or JSON column 'b' can't have a default value")
func (col *ColumnInfo) canHaveDefaultValue() bool {
	switch col.k {
	case KindBLOB, KindTINYBLOB, KindMEDIUMBLOB, KindLONGBLOB, KindTEXT, KindTINYTEXT, KindMEDIUMTEXT, KindLONGTEXT, KindJSON:
		return false
	default:
		return true
	}
}

type IndexInfo struct {
	name      string
	signature string
	columns   []*ColumnInfo
	lens      []int
}

func (idx *IndexInfo) getIndexSignature() string {
	if idx.signature != "" {
		return idx.signature
	}
	s := ""
	for i, column := range idx.columns {
		if i > 0 {
			s += ", "
		}
		s += fmt.Sprintf("`%s`", column.name)
		if idx.lens[i] > 0 {
			s += fmt.Sprintf("(%d)", idx.lens[i])
		}
	}
	idx.signature = s
	return s
}