func getArrayBounds(ft string, elem []string) []int64 {
	if strings.HasPrefix(ft, "set") {
		return []int64{int64(len(elem))}
	}
	return nil
}
getArrayBounds calculate array bound for only set data type
and we do not expect multidimensional array.
package mysql
import (
	"fmt"
	"reflect"
	"regexp"
	"strconv"
	"strings"
	"github.com/cloudspannerecosystem/harbourbridge/internal"
	"github.com/cloudspannerecosystem/harbourbridge/schema"
	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
	"github.com/pingcap/parser"
	"github.com/pingcap/parser/ast"
	"github.com/pingcap/parser/opcode"
	"github.com/pingcap/tidb/types"
	driver "github.com/pingcap/tidb/types/parser_driver"
)
var valuesRegexp = regexp.MustCompile("\\((.*?)\\)")
var insertRegexp = regexp.MustCompile("INSERT\\sINTO\\s(.*?)\\sVALUES\\s")
var unsupportedRegexp = regexp.MustCompile("function|procedure|trigger")
var MysqlSpatialDataTypes = []string{"geometrycollection", "multipoint", "multilinestring", "multipolygon", "point", "linestring", "polygon", "geometry"}
var spatialRegexps = func() []*regexp.Regexp {
	l := make([]*regexp.Regexp, len(MysqlSpatialDataTypes))
	for i, spatial := range MysqlSpatialDataTypes {
		l[i] = regexp.MustCompile("(?i)" + " " + spatial)
	}
	return l
}()
var spatialIndexRegex = regexp.MustCompile("(?i)\\sSPATIAL\\s")
var spatialSridRegex = regexp.MustCompile("(?i)\\sSRID\\s\\d*")
type DbDumpImpl struct{}
func (ddi DbDumpImpl) GetToDdl() common.ToDdl {
	return ToDdlImpl{}
}
func (ddi DbDumpImpl) ProcessDump(conv *internal.Conv, r *internal.Reader) error {
	return processMySQLDump(conv, r)
}
func processMySQLDump(conv *internal.Conv, r *internal.Reader) error {
	for {
		startLine := r.LineNumber
		startOffset := r.Offset
		b, stmts, err := readAndParseChunk(conv, r)
		if err != nil {
			return err
		}
		for _, stmt := range stmts {
			isInsert := processStatement(conv, stmt)
			internal.VerbosePrintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) Insert Statement=%v\n", startLine, startOffset, 1, r.LineNumber-startLine, len(b), isInsert)
		}
		if r.EOF {
			break
		}
	}
	return nil
}
func readAndParseChunk(conv *internal.Conv, r *internal.Reader) ([]byte, []ast.StmtNode, error) {
	var l [][]byte
	for {
		b := r.ReadLine()
		l = append(l, b)
		if strings.Contains(string(b), ";") || r.EOF {
			n := 0
			for i := range l {
				n += len(l[i])
			}
			s := make([]byte, n)
			n = 0
			for i := range l {
				n += copy(s[n:], l[i])
			}
			chunk := string(s)
			tree, _, err := parser.New().Parse(chunk, "", "")
			if err == nil {
				return s, tree, nil
			}
			newTree, ok := handleParseError(conv, chunk, err, l)
			if ok {
				return s, newTree, nil
			}
			conv.Stats.Reparsed++
		}
		if r.EOF {
			return nil, nil, fmt.Errorf("Error parsing last %d line(s) of input", len(l))
		}
	}
}
func processStatement(conv *internal.Conv, stmt ast.StmtNode) bool {
	switch s := stmt.(type) {
	case *ast.CreateTableStmt:
		if conv.SchemaMode() {
			processCreateTable(conv, s)
		}
	case *ast.AlterTableStmt:
		if conv.SchemaMode() {
			processAlterTable(conv, s)
		}
	case *ast.SetStmt:
		if conv.SchemaMode() {
			processSetStmt(conv, s)
		}
	case *ast.InsertStmt:
		processInsertStmt(conv, s)
		return true
	case *ast.CreateIndexStmt:
		if conv.SchemaMode() {
			processCreateIndex(conv, s)
		}
	default:
		conv.SkipStatement(NodeType(stmt))
	}
	return false
}
func processCreateIndex(conv *internal.Conv, stmt *ast.CreateIndexStmt) {
	if stmt.Table == nil {
		logStmtError(conv, stmt, fmt.Errorf("cannot process index statement with nil table."))
		return
	}
	tableName, err := getTableName(stmt.Table)
	if err != nil {
		logStmtError(conv, stmt, fmt.Errorf("can't get table name: %w", err))
		return
	}
	if _, ok := conv.SrcSchema[tableName]; ok {
		ctable := conv.SrcSchema[tableName]
		ctable.Indexes = append(ctable.Indexes, schema.Index{
			Name:   stmt.IndexName,
			Unique: (stmt.KeyType == ast.IndexKeyTypeUnique),
			Keys:   toSchemaKeys(stmt.IndexPartSpecifications),
		})
		conv.SrcSchema[tableName] = ctable
	} else {
		conv.Unexpected(fmt.Sprintf("Table %s not found while processing index statement", tableName))
		conv.SkipStatement(NodeType(stmt))
	}
}
func processSetStmt(conv *internal.Conv, stmt *ast.SetStmt) {
	if stmt.Variables != nil && len(stmt.Variables) > 0 {
		for _, variable := range stmt.Variables {
			if variable.Name == "TIME_ZONE" {
				value := variable.Value
				switch val := value.(type) {
				case *driver.ValueExpr:
					if val.GetValue() == nil {
						logStmtError(conv, stmt, fmt.Errorf("found nil value in 'SET TIME_ZONE' statement"))
						return
					}
					conv.TimezoneOffset = fmt.Sprintf("%v", val.GetValue())
				default:
					return
				}
			}
		}
	}
}
func processCreateTable(conv *internal.Conv, stmt *ast.CreateTableStmt) {
	if stmt.Table == nil {
		logStmtError(conv, stmt, fmt.Errorf("table is nil"))
		return
	}
	tableName, err := getTableName(stmt.Table)
	internal.VerbosePrintf("processing create table elem=%s stmt=%v\n", tableName, stmt)
	if err != nil {
		logStmtError(conv, stmt, fmt.Errorf("can't get table name: %w", err))
		return
	}
	var colNames []string
	colDef := make(map[string]schema.Column)
	var keys []schema.Key
	var fkeys []schema.ForeignKey
	var index []schema.Index
	for _, element := range stmt.Cols {
		colname, col, constraint, err := processColumn(conv, tableName, element)
		if err != nil {
			logStmtError(conv, stmt, err)
			return
		}
		colNames = append(colNames, colname)
		colDef[colname] = col
		if constraint.isPk {
			keys = append(keys, schema.Key{Column: colname})
		}
		if constraint.fk.Columns != nil {
			fkeys = append(fkeys, constraint.fk)
		}
		if constraint.isUniqueKey {
			index = append(index, schema.Index{Name: "", Unique: true, Keys: []schema.Key{schema.Key{Column: colname, Desc: false}}})
		}
	}
	conv.SchemaStatement(NodeType(stmt))
	conv.SrcSchema[tableName] = schema.Table{
		Name:        tableName,
		ColNames:    colNames,
		ColDefs:     colDef,
		PrimaryKeys: keys,
		ForeignKeys: fkeys,
		Indexes:     index}
	for _, constraint := range stmt.Constraints {
		processConstraint(conv, tableName, constraint, "CREATE TABLE")
	}
}
func processConstraint(conv *internal.Conv, table string, constraint *ast.Constraint, stmtType string) {
	st := conv.SrcSchema[table]
	switch ct := constraint.Tp; ct {
	case ast.ConstraintPrimaryKey:
		checkEmpty(conv, st.PrimaryKeys, stmtType)  
		st.PrimaryKeys = toSchemaKeys(constraint.Keys)
		updateCols(conv, ast.ConstraintPrimaryKey, constraint.Keys, st.ColDefs, table)
	case ast.ConstraintForeignKey:
		st.ForeignKeys = append(st.ForeignKeys, toForeignKeys(conv, constraint))
	case ast.ConstraintIndex:
		st.Indexes = append(st.Indexes, schema.Index{Name: constraint.Name, Keys: toSchemaKeys(constraint.Keys)})
	case ast.ConstraintUniq:
		st.Indexes = append(st.Indexes, schema.Index{Name: constraint.Name, Unique: true, Keys: toSchemaKeys(constraint.Keys)})
	default:
		updateCols(conv, ct, constraint.Keys, st.ColDefs, table)
	}
	conv.SrcSchema[table] = st
}
func toSchemaKeys(columns []*ast.IndexPartSpecification) (keys []schema.Key) {
	for _, colname := range columns {
		keys = append(keys, schema.Key{Column: colname.Column.Name.String()})
	}
	return keys
}
func toForeignKeys(conv *internal.Conv, fk *ast.Constraint) (fkey schema.ForeignKey) {
	columns := fk.Keys
	referTable, err := getTableName(fk.Refer.Table)
	if err != nil {
		conv.Unexpected(err.Error())
		return schema.ForeignKey{}
	}
	referColumns := fk.Refer.IndexPartSpecifications
	var colNames, referColNames []string
	for i, column := range columns {
		colNames = append(colNames, column.Column.Name.String())
		referColNames = append(referColNames, referColumns[i].Column.Name.String())
	}
	fkey = schema.ForeignKey{
		Name:         fk.Name,
		Columns:      colNames,
		ReferTable:   referTable,
		ReferColumns: referColNames,
		OnDelete:     fk.Refer.OnDelete.ReferOpt.String(),
		OnUpdate:     fk.Refer.OnUpdate.ReferOpt.String()}
	return fkey
}
func updateCols(conv *internal.Conv, ct ast.ConstraintType, colNames []*ast.IndexPartSpecification, colDef map[string]schema.Column, tableName string) {
	for _, column := range colNames {
		colName := column.Column.OrigColName()
		cd := colDef[colName]
		switch ct {
		case ast.ConstraintCheck:
			cd.Ignored.Check = true
		case ast.ConstraintPrimaryKey:
			cd.NotNull = true
		}
		colDef[colName] = cd
	}
}
func processAlterTable(conv *internal.Conv, stmt *ast.AlterTableStmt) {
	if stmt.Table == nil {
		logStmtError(conv, stmt, fmt.Errorf("table is nil"))
		return
	}
	tableName, err := getTableName(stmt.Table)
	if err != nil {
		logStmtError(conv, stmt, fmt.Errorf("can't get table name: %w", err))
		return
	}
	if _, ok := conv.SrcSchema[tableName]; ok {
		for _, item := range stmt.Specs {
			switch alterType := item.Tp; alterType {
			case ast.AlterTableAddConstraint:
				processConstraint(conv, tableName, item.Constraint, "ALTER TABLE")
				conv.SchemaStatement(NodeType(stmt))
			case ast.AlterTableModifyColumn:
				colname, col, constraint, err := processColumn(conv, tableName, item.NewColumns[0])
				if err != nil {
					logStmtError(conv, stmt, err)
					return
				}
				conv.SrcSchema[tableName].ColDefs[colname] = col
				if constraint.isPk {
					ctable := conv.SrcSchema[tableName]
					checkEmpty(conv, ctable.PrimaryKeys, "ALTER TABLE")
					ctable.PrimaryKeys = []schema.Key{{Column: colname}}
					conv.SrcSchema[tableName] = ctable
				}
				if constraint.fk.Columns != nil {
					ctable := conv.SrcSchema[tableName]
					ctable.ForeignKeys = append(ctable.ForeignKeys, constraint.fk)
					conv.SrcSchema[tableName] = ctable
				}
				if constraint.isUniqueKey {
					ctable := conv.SrcSchema[tableName]
					ctable.Indexes = append(ctable.Indexes, schema.Index{Name: "", Unique: true, Keys: []schema.Key{schema.Key{Column: colname, Desc: false}}})
					conv.SrcSchema[tableName] = ctable
				}
				conv.SchemaStatement(NodeType(stmt))
			default:
				conv.SkipStatement(NodeType(stmt))
			}
		}
	} else {
		conv.SkipStatement(NodeType(stmt))
	}
}
func getTableName(table *ast.TableName) (string, error) {
	var l []string
	if table.Schema.String() != "" {
		l = append(l, table.Schema.String())
	}
	if table.Name.String() == "" {
		return "", fmt.Errorf("tablename is empty: can't build table name")
	}
	l = append(l, table.Name.String())
	return strings.Join(l, "."), nil
}
func processColumn(conv *internal.Conv, tableName string, col *ast.ColumnDef) (string, schema.Column, columnConstraint, error) {
	if col.Name == nil {
		return "", schema.Column{}, columnConstraint{}, fmt.Errorf("column name is nil")
	}
	name := col.Name.OrigColName()
	if col.Tp == nil {
		return "", schema.Column{}, columnConstraint{}, fmt.Errorf("can't get column type for %s: %w", name, fmt.Errorf("found nil *ast.ColumnDef.Tp"))
	}
	tid, mods := getTypeModsAndID(conv, col.Tp.String())
	ty := schema.Type{
		Name:        tid,
		Mods:        mods,
		ArrayBounds: getArrayBounds(col.Tp.String(), col.Tp.Elems)}
	column := schema.Column{Name: name, Type: ty}
	return name, column, updateColsByOption(conv, tableName, col, &column), nil
}
type columnConstraint struct {
	isPk        bool
	isUniqueKey bool
	fk          schema.ForeignKey
}
func updateColsByOption(conv *internal.Conv, tableName string, col *ast.ColumnDef, column *schema.Column) columnConstraint {
	var cc columnConstraint
	for _, elem := range col.Options {
		switch op := elem.Tp; op {
		case ast.ColumnOptionPrimaryKey:
			column.NotNull = true
			cc.isPk = true
		case ast.ColumnOptionNotNull:
			column.NotNull = true
		case ast.ColumnOptionAutoIncrement:
			column.Ignored.AutoIncrement = true
		case ast.ColumnOptionDefaultValue:
			v, ok := elem.Expr.(*driver.ValueExpr)
			nullDefault := ok && v.GetValue() == nil
			if !nullDefault {
				column.Ignored.Default = true
			}
		case ast.ColumnOptionUniqKey:
			cc.isUniqueKey = true
		case ast.ColumnOptionCheck:
			column.Ignored.Check = true
		case ast.ColumnOptionReference:
			column := col.Name.String()
			referTable, err := getTableName(elem.Refer.Table)
			if err != nil {
				conv.Unexpected(err.Error())
				continue
			}
			referColumn := elem.Refer.IndexPartSpecifications[0].Column.Name.String()
			fkey := schema.ForeignKey{
				Columns:      []string{column},
				ReferTable:   referTable,
				ReferColumns: []string{referColumn},
				OnDelete:     elem.Refer.OnDelete.ReferOpt.String(),
				OnUpdate:     elem.Refer.OnUpdate.ReferOpt.String()}
			cc.fk = fkey
		}
	}
	return cc
}
func getTypeModsAndID(conv *internal.Conv, columnType string) (string, []int64) {
	id := columnType
	var mods []int64
	if strings.Contains(columnType, "(") {
		id = strings.Split(columnType, "(")[0]
		if id == "set" || id == "enum" {
			return id, nil
		}
		values := valuesRegexp.FindString(columnType)
		strMods := strings.Split(values[1:len(values)-1], ",")
		for _, i := range strMods {
			j, err := strconv.ParseInt(i, 10, 64)
			if err != nil {
				conv.Unexpected(fmt.Sprintf("Unable to get modifiers for `%s` datatype.", id))
				return id, nil
			}
			mods = append(mods, j)
		}
	}
	if strings.Contains(id, " ") {
		id = strings.TrimSuffix(columnType, " BINARY")
	}
	return id, mods
}
func handleParseError(conv *internal.Conv, chunk string, err error, l [][]byte) ([]ast.StmtNode, bool) {
	errMsg := strings.ToLower(err.Error())
	if unsupportedRegexp.MatchString(errMsg) || strings.Contains(errMsg, "delimiter") {
		if strings.Count(strings.ToLower(chunk), "delimiter") == 1 {
			return nil, false
		}
		return nil, skipUnsupported(conv, strings.ToLower(chunk))
	}
	insertStmtPrefix := insertRegexp.FindString(chunk)
	if insertStmtPrefix != "" {
		valuesChunk := insertRegexp.Split(chunk, 2)[1]  
		return handleInsertStatement(conv, valuesChunk, insertStmtPrefix)
	}
	for _, spatial := range MysqlSpatialDataTypes {
		if strings.Contains(errMsg, `near "`+spatial) {
			if conv.SchemaMode() {
				conv.Unexpected(fmt.Sprintf("Unsupported datatype '%s' encountered while parsing following statement at line number %d : \n%s", spatial, len(l), chunk))
				internal.VerbosePrintf("Converting datatype '%s' to 'Text' and retrying to parse the statement\n", spatial)
			}
			return handleSpatialDatatype(conv, chunk, l)
		}
	}
	return nil, false
}
// handleInsertStatement handles error in parsing the insert statement.
// Likely causes of failing to parse Insert statement:
// 	a) Due to some invalid value.
// 	b) chunk size is more than what pingcap parser could handle (more than 40MB in size).
// We deal with this cases by extracting all rows and creating
// extended insert statements. Then we parse one Insert statement
// at a time, ensuring no size issue and skipping only invalid entries.
func handleInsertStatement(conv *internal.Conv, chunk, insertStmtPrefix string) ([]ast.StmtNode, bool) {
	var stmts []ast.StmtNode
	values := valuesRegexp.FindAllString(chunk, -1)
	if len(values) == 0 {
		return nil, false
	}
	for _, value := range values {
		chunk = insertStmtPrefix + value + ";"
		newTree, _, err := parser.New().Parse(chunk, "", "")
		if err != nil {
			if conv.SchemaMode() {
				conv.Unexpected(fmt.Sprintf("Either unsupported value is encountered or syntax is incorrect for following statement : \n%s", chunk))
			}
			conv.SkipStatement("InsertStmt")
			continue
		}
		stmts = append(stmts, newTree[0])
	}
	return stmts, true
}
// handleSpatialDatatype handles error in parsing spatial datatype.
// We parse chunk again after taking these actions:
// a) Replace spatial datatype with 'text'.
// b) Remove 'SPATIAL' keyword from Index/Key.
// c) Remove SRID(spatial reference identifier) attribute.
func handleSpatialDatatype(conv *internal.Conv, chunk string, l [][]byte) ([]ast.StmtNode, bool) {
	if !conv.SchemaMode() {
		return nil, true
	}
	for _, spatialRegexp := range spatialRegexps {
		chunk = spatialRegexp.ReplaceAllString(chunk, " text")
	}
	chunk = spatialIndexRegex.ReplaceAllString(chunk, "")
	chunk = spatialSridRegex.ReplaceAllString(chunk, "")
	newTree, _, err := parser.New().Parse(chunk, "", "")
	if err != nil {
		return nil, false
	}
	return newTree, true
}
// skipUnsupported skips the stored programs that are not supported
// by pingcap parser.
func skipUnsupported(conv *internal.Conv, chunk string) bool {
	createOrdrop := "Create"
	if strings.Contains(chunk, "drop") {
		createOrdrop = "Drop"
	}
	switch {
	case strings.Contains(chunk, "trigger"):
		conv.SkipStatement(createOrdrop + "TrigStmt")
	case strings.Contains(chunk, "procedure"):
		conv.SkipStatement(createOrdrop + "ProcedureStmt")
	case strings.Contains(chunk, "function"):
		conv.SkipStatement(createOrdrop + "FunctionStmt")
	default:
		return false
	}
	return true
}