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
}
getTableName extracts the table name from *ast.TableName table, and returns
the raw extracted name (the MySQL table name).
*ast.TableName is used to represent table names. It consists of two components:
Schema: schemas in MySQL db often unspecified;
Name: name of the table
We build a table name from these components as follows:
a) nil components are dropped.
b) if more than one component is specified, they are joined using "."
(Note that Spanner doesn't allow "." in table names, so this
will eventually get re-mapped when we construct the Spanner table name).
c) return error if Table is nil or "".
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))
	}
}