// Add a column into table
func (d *ddl) addColumn(ctx context.Context, schema model.CIStr, tbl table.Table, spec *AlterSpecification) error {
	// Find position
	cols := tbl.Cols()
	position := len(cols)
	name := spec.Column.Name
	// Check column name duplicate
	dc := column.FindCol(cols, name)
	if dc != nil {
		return errors.Errorf("Try to add a column with the same name of an already exists column.")
	}
	if spec.Position.Type == ColumnPositionFirst {
		position = 0
	} else if spec.Position.Type == ColumnPositionAfter {
		// Find the mentioned column
		c := column.FindCol(cols, spec.Position.RelativeColumn)
		if c == nil {
			return errors.Errorf("No such column: %v", name)
		}
		// insert position is after the mentioned column
		position = c.Offset + 1
	}
	// TODO: Set constraint
	col, _, err := d.buildColumnAndConstraint(position, spec.Column)
	if err != nil {
		return errors.Trace(err)
	}
	// insert col into the right place of the column list
	newCols := make([]*column.Col, 0, len(cols)+1)
	newCols = append(newCols, cols[:position]...)
	newCols = append(newCols, col)
	newCols = append(newCols, cols[position:]...)
	// adjust position
	if position != len(cols) {
		offsetChange := make(map[int]int)
		for i := position + 1; i < len(newCols); i++ {
			offsetChange[newCols[i].Offset] = i
			newCols[i].Offset = i
		}
		// Update index offset info
		for _, idx := range tbl.Indices() {
			for _, c := range idx.Columns {
				newOffset, ok := offsetChange[c.Offset]
				if ok {
					c.Offset = newOffset
				}
			}
		}
	}
	tb := tbl.(*tables.Table)
	tb.Columns = newCols
	// TODO: update index
	// TODO: update default value
	// update infomation schema
	err = d.updateInfoSchema(ctx, schema, tb.Meta())
	return errors.Trace(err)
}

// DropColumn will drop a column from the table, now we don't support drop the column with index covered.
func (d *ddl) DropColumn(ctx context.Context, ti table.Ident, colName model.CIStr) error {
	is := d.infoHandle.Get()
	schema, ok := is.SchemaByName(ti.Schema)
	if !ok {
		return errors.Trace(terror.DatabaseNotExists)
	}

	t, err := is.TableByName(ti.Schema, ti.Name)
	if err != nil {
		return errors.Trace(ErrNotExists)
	}

	// Check whether dropped column has existed.
	col := column.FindCol(t.Cols(), colName.L)
	if col == nil {
		return errors.Errorf("column %s doesn’t exist", colName.L)
	}

	job := &model.Job{
		SchemaID: schema.ID,
		TableID:  t.Meta().ID,
		Type:     model.ActionDropColumn,
		Args:     []interface{}{colName},
	}

	err = d.startJob(ctx, job)
	err = d.hook.OnChanged(err)
	return errors.Trace(err)
}

// Check if column scope privilege entry exists in mysql.Columns_priv.
// If unexists, insert a new one.
func (s *GrantStmt) checkAndInitColumnPriv(ctx context.Context, user string, host string, cols []string) error {
	db, tbl, err := s.getTargetSchemaAndTable(ctx)
	if err != nil {
		return errors.Trace(err)
	}
	for _, c := range cols {
		col := column.FindCol(tbl.Cols(), c)
		if col == nil {
			return errors.Errorf("Unknown column: %s", c)
		}
		ok, err := columnPrivEntryExists(ctx, user, host, db.Name.O, tbl.TableName().O, col.Name.O)
		if err != nil {
			return errors.Trace(err)
		}
		if ok {
			continue
		}
		// Entry does not exists for user-host-db-tbl-col. Insert a new entry.
		err = initColumnPrivEntry(ctx, user, host, db.Name.O, tbl.TableName().O, col.Name.O)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func getUpdateColumns(t table.Table, assignList []expression.Assignment, isMultipleTable bool, tblAliasMap map[string]string) ([]*column.Col, []expression.Assignment, error) {
	// TODO: We should check the validate if assignList in somewhere else. Maybe in building plan.
	// TODO: We should use field.GetFieldIndex to replace this function.
	tcols := make([]*column.Col, 0, len(assignList))
	tAsgns := make([]expression.Assignment, 0, len(assignList))
	tname := t.TableName()
	for _, asgn := range assignList {
		if isMultipleTable {
			if tblAliasMap != nil {
				if alias, ok := tblAliasMap[asgn.TableName]; ok {
					if !strings.EqualFold(tname.O, alias) {
						continue
					}
				}
			} else if !strings.EqualFold(tname.O, asgn.TableName) {
				continue
			}
		}
		col := column.FindCol(t.Cols(), asgn.ColName)
		if col == nil {
			if isMultipleTable {
				continue
			}
			return nil, nil, errors.Errorf("UPDATE: unknown column %s", asgn.ColName)
		}
		tcols = append(tcols, col)
		tAsgns = append(tAsgns, asgn)
	}
	return tcols, tAsgns, nil
}

func (r *TableDefaultPlan) filterIsNull(ctx context.Context, x *expression.IsNull) (plan.Plan, bool, error) {
	if _, ok := x.Expr.(*expression.Ident); !ok {
		// if expression is not Ident expression, we cannot use index
		// e.g, "(x > null) is not null", (x > null) is a binary expression, we must evaluate it first
		return r, false, nil
	}

	cns := expression.MentionedColumns(x.Expr)
	if len(cns) == 0 {
		return r, false, nil
	}

	cn := cns[0]
	t := r.T
	ix := t.FindIndexByColName(cn)
	if ix == nil { // Column cn has no index.
		return r, false, nil
	}
	col := column.FindCol(t.Cols(), cn)
	var spans []*indexSpan
	if x.Not {
		spans = toSpans(opcode.GE, minNotNullVal, nil)
	} else {
		spans = toSpans(opcode.EQ, nil, nil)
	}
	return &indexPlan{
		src:     t,
		col:     col,
		unique:  ix.Unique,
		idxName: ix.Name.L,
		idx:     ix.X,
		spans:   spans,
	}, true, nil
}

// Check if column scope privilege entry exists in mysql.Columns_priv.
// If unexists, insert a new one.
func (e *GrantExec) checkAndInitColumnPriv(user string, host string, cols []*ast.ColumnName) error {
	db, tbl, err := e.getTargetSchemaAndTable()
	if err != nil {
		return errors.Trace(err)
	}
	for _, c := range cols {
		col := column.FindCol(tbl.Cols(), c.Name.L)
		if col == nil {
			return errors.Errorf("Unknown column: %s", c.Name.O)
		}
		ok, err := columnPrivEntryExists(e.ctx, user, host, db.Name.O, tbl.Meta().Name.O, col.Name.O)
		if err != nil {
			return errors.Trace(err)
		}
		if ok {
			continue
		}
		// Entry does not exist for user-host-db-tbl-col. Insert a new entry.
		err = initColumnPrivEntry(e.ctx, user, host, db.Name.O, tbl.Meta().Name.O, col.Name.O)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (s *testColumnSuite) testGetColumn(c *C, t table.Table, name string, isExist bool) {
	col := column.FindCol(t.Cols(), name)
	if isExist {
		c.Assert(col, NotNil)
	} else {
		c.Assert(col, IsNil)
	}
}

func (r *TableDefaultPlan) filterBinOp(ctx context.Context, x *expression.BinaryOperation) (plan.Plan, bool, error) {
	ok, name, rval, err := x.IsIdentCompareVal()
	if err != nil {
		return r, false, errors.Trace(err)
	}
	if !ok {
		return r, false, nil
	}
	if rval == nil {
		// if nil, any <, <=, >, >=, =, != operator will do nothing
		// any value compared null returns null
		// TODO: if we support <=> later, we must handle null
		return &NullPlan{r.GetFields()}, true, nil
	}

	_, tn, cn := field.SplitQualifiedName(name)
	t := r.T
	if tn != "" && tn != t.TableName().L {
		return r, false, nil
	}
	c := column.FindCol(t.Cols(), cn)
	if c == nil {
		return nil, false, errors.Errorf("No such column: %s", cn)
	}
	var seekVal interface{}
	if seekVal, err = types.Convert(rval, &c.FieldType); err != nil {
		return nil, false, errors.Trace(err)
	}
	spans := toSpans(x.Op, rval, seekVal)
	if c.IsPKHandleColumn(r.T.Meta()) {
		if r.rangeScan {
			spans = filterSpans(r.spans, spans)
		}
		return &TableDefaultPlan{
			T:         r.T,
			Fields:    r.Fields,
			rangeScan: true,
			spans:     spans,
		}, true, nil
	} else if r.rangeScan {
		// Already filtered on PK handle column, should not switch to index plan.
		return r, false, nil
	}

	ix := t.FindIndexByColName(cn)
	if ix == nil { // Column cn has no index.
		return r, false, nil
	}
	return &indexPlan{
		src:     t,
		col:     c,
		unique:  ix.Unique,
		idxName: ix.Name.O,
		idx:     ix.X,
		spans:   spans,
	}, true, nil
}

func getUpdateColumns(t table.Table, assignList []expressions.Assignment) ([]*column.Col, error) {
	tcols := make([]*column.Col, len(assignList))
	for i, asgn := range assignList {
		col := column.FindCol(t.Cols(), asgn.ColName)
		if col == nil {
			return nil, errors.Errorf("UPDATE: unknown column %s", asgn.ColName)
		}
		tcols[i] = col
	}

	return tcols, nil
}

func findColumnByName(t table.Table, name string) (*column.Col, error) {
	_, tableName, colName := field.SplitQualifiedName(name)
	if len(tableName) > 0 && tableName != t.TableName().O {
		return nil, errors.Errorf("unknown field %s.%s", tableName, colName)
	}

	c := column.FindCol(t.Cols(), colName)
	if c == nil {
		return nil, errors.Errorf("unknown field %s", colName)
	}
	return c, nil
}

func (d *ddl) CreateIndex(ctx context.Context, ti table.Ident, unique bool, indexName model.CIStr, idxColNames []*coldef.IndexColName) error {
	is := d.infoHandle.Get()
	t, err := is.TableByName(ti.Schema, ti.Name)
	if err != nil {
		return errors.Trace(err)
	}
	if _, ok := is.IndexByName(ti.Schema, ti.Name, indexName); ok {
		return errors.Errorf("CREATE INDEX: index already exist %s", indexName)
	}
	if is.ColumnExists(ti.Schema, ti.Name, indexName) {
		return errors.Errorf("CREATE INDEX: index name collision with existing column: %s", indexName)
	}

	tbInfo := t.Meta()

	// build offsets
	idxColumns := make([]*model.IndexColumn, 0, len(idxColNames))
	for i, ic := range idxColNames {
		col := column.FindCol(t.Cols(), ic.ColumnName)
		if col == nil {
			return errors.Errorf("CREATE INDEX: column does not exist: %s", ic.ColumnName)
		}
		idxColumns = append(idxColumns, &model.IndexColumn{
			Name:   col.Name,
			Offset: col.Offset,
			Length: ic.Length,
		})
		// Set ColumnInfo flag
		if i == 0 {
			if unique && len(idxColNames) == 1 {
				tbInfo.Columns[col.Offset].Flag |= mysql.UniqueKeyFlag
			} else {
				tbInfo.Columns[col.Offset].Flag |= mysql.MultipleKeyFlag
			}
		}
	}
	// create index info
	idxInfo := &model.IndexInfo{
		Name:    indexName,
		Columns: idxColumns,
		Unique:  unique,
	}
	tbInfo.Indices = append(tbInfo.Indices, idxInfo)

	// build index
	err = d.buildIndex(ctx, t, idxInfo, unique)
	if err != nil {
		return errors.Trace(err)
	}

	// update InfoSchema
	return d.updateInfoSchema(ctx, ti.Schema, tbInfo)
}

// Filter implements plan.Plan Filter interface.
// Filter merges BinaryOperations, and determines the lower and upper bound.
func (r *indexPlan) Filter(ctx context.Context, expr expression.Expression) (plan.Plan, bool, error) {
	switch x := expr.(type) {
	case *expressions.BinaryOperation:
		ok, cname, val, err := x.IsIdentRelOpVal()
		if err != nil {
			return nil, false, err
		}

		if !ok || r.colName != cname {
			break
		}

		col := column.FindCol(r.src.Cols(), cname)
		if col == nil {
			break
		}

		if val, err = col.CastValue(ctx, val); err != nil {
			return nil, false, err
		}
		r.spans = filterSpans(r.spans, toSpans(x.Op, val))
		return r, true, nil
	case *expressions.Ident:
		if r.colName != x.L {
			break
		}
		r.spans = filterSpans(r.spans, toSpans(opcode.GE, minNotNullVal))
		return r, true, nil
	case *expressions.UnaryOperation:
		if x.Op != '!' {
			break
		}

		operand, ok := x.V.(*expressions.Ident)
		if !ok {
			break
		}

		cname := operand.L
		if r.colName != cname {
			break
		}
		r.spans = filterSpans(r.spans, toSpans(opcode.EQ, nil))
		return r, true, nil
	}

	return r, false, nil
}

func (d *ddl) buildTableInfo(tableName model.CIStr, cols []*column.Col, constraints []*coldef.TableConstraint) (tbInfo *model.TableInfo, err error) {
	tbInfo = &model.TableInfo{
		Name: tableName,
	}
	tbInfo.ID, err = d.genGlobalID()
	if err != nil {
		return nil, errors.Trace(err)
	}
	for _, v := range cols {
		tbInfo.Columns = append(tbInfo.Columns, &v.ColumnInfo)
	}
	for _, constr := range constraints {
		// 1. check if the column is exists
		// 2. add index
		indexColumns := make([]*model.IndexColumn, 0, len(constr.Keys))
		for _, key := range constr.Keys {
			col := column.FindCol(cols, key.ColumnName)
			if col == nil {
				return nil, errors.Errorf("No such column: %v", key)
			}
			indexColumns = append(indexColumns, &model.IndexColumn{
				Name:   model.NewCIStr(key.ColumnName),
				Offset: col.Offset,
				Length: key.Length,
			})
		}
		idxInfo := &model.IndexInfo{
			Name:    model.NewCIStr(constr.ConstrName),
			Columns: indexColumns,
			State:   model.StatePublic,
		}
		switch constr.Tp {
		case coldef.ConstrPrimaryKey:
			idxInfo.Unique = true
			idxInfo.Primary = true
			idxInfo.Name = model.NewCIStr(column.PrimaryKeyName)
		case coldef.ConstrUniq, coldef.ConstrUniqKey, coldef.ConstrUniqIndex:
			idxInfo.Unique = true
		}
		idxInfo.ID, err = d.genGlobalID()
		if err != nil {
			return nil, errors.Trace(err)
		}
		tbInfo.Indices = append(tbInfo.Indices, idxInfo)
	}
	return
}

// AddColumn will add a new column to the table.
func (d *ddl) AddColumn(ctx context.Context, ti table.Ident, spec *AlterSpecification) error {
	// Check whether the added column constraints are supported.
	err := checkColumnConstraint(spec.Column.Constraints)
	if err != nil {
		return errors.Trace(err)
	}

	is := d.infoHandle.Get()
	schema, ok := is.SchemaByName(ti.Schema)
	if !ok {
		return errors.Trace(terror.DatabaseNotExists)
	}

	t, err := is.TableByName(ti.Schema, ti.Name)
	if err != nil {
		return errors.Trace(ErrNotExists)
	}

	// Check whether added column has existed.
	colName := spec.Column.Name
	col := column.FindCol(t.Cols(), colName)
	if col != nil {
		return errors.Errorf("column %s already exists", colName)
	}

	// ingore table constraints now, maybe return error later
	// we use length(t.Cols()) as the default offset first, later we will change the
	// column's offset later.
	col, _, err = d.buildColumnAndConstraint(len(t.Cols()), spec.Column)
	if err != nil {
		return errors.Trace(err)
	}

	job := &model.Job{
		SchemaID: schema.ID,
		TableID:  t.Meta().ID,
		Type:     model.ActionAddColumn,
		Args:     []interface{}{&col.ColumnInfo, spec.Position, 0},
	}

	err = d.startJob(ctx, job)
	err = d.hook.OnChanged(err)
	return errors.Trace(err)
}

func (r *TableDefaultPlan) filterBinOp(ctx context.Context, x *expression.BinaryOperation) (plan.Plan, bool, error) {
	ok, name, rval, err := x.IsIdentCompareVal()
	if err != nil {
		return r, false, err
	}
	if !ok {
		return r, false, nil
	}
	if rval == nil {
		// if nil, any <, <=, >, >=, =, != operator will do nothing
		// any value compared null returns null
		// TODO: if we support <=> later, we must handle null
		return &NullPlan{r.GetFields()}, true, nil
	}

	_, tn, cn := field.SplitQualifiedName(name)
	t := r.T
	if tn != "" && tn != t.TableName().L {
		return r, false, nil
	}
	c := column.FindCol(t.Cols(), cn)
	if c == nil {
		return nil, false, errors.Errorf("No such column: %s", cn)
	}

	ix := t.FindIndexByColName(cn)
	if ix == nil { // Column cn has no index.
		return r, false, nil
	}

	var seekVal interface{}
	if seekVal, err = types.Convert(rval, &c.FieldType); err != nil {
		return nil, false, err
	}
	return &indexPlan{
		src:     t,
		col:     c,
		idxName: ix.Name.O,
		idx:     ix.X,
		spans:   toSpans(x.Op, rval, seekVal),
	}, true, nil
}

func getUpdateColumns(t table.Table, assignList []expressions.Assignment, multipleTable bool) ([]*column.Col, error) {
	// TODO: We should check the validate if assignList in somewhere else. Maybe in building plan.
	tcols := make([]*column.Col, 0, len(assignList))
	tname := t.TableName()
	for _, asgn := range assignList {
		if multipleTable {
			if !strings.EqualFold(tname.O, asgn.TableName) {
				continue
			}
		}
		col := column.FindCol(t.Cols(), asgn.ColName)
		if col == nil {
			if multipleTable {
				continue
			}
			return nil, errors.Errorf("UPDATE: unknown column %s", asgn.ColName)
		}
		tcols = append(tcols, col)
	}
	return tcols, nil
}

func (r *TableDefaultPlan) filterBinOp(ctx context.Context, x *expressions.BinaryOperation) (plan.Plan, bool, error) {
	ok, cn, rval, err := x.IsIdentRelOpVal()
	if err != nil {
		return r, false, err
	}
	if !ok {
		return r, false, nil
	}

	t := r.T
	c := column.FindCol(t.Cols(), cn)
	if c == nil {
		return nil, false, errors.Errorf("No such column: %s", cn)
	}

	ix := t.FindIndexByColName(cn)
	if ix == nil { // Column cn has no index.
		return r, false, nil
	}

	if rval, err = c.CastValue(ctx, rval); err != nil {
		return nil, false, err
	}

	if rval == nil {
		// if nil, any <, <=, >, >=, =, != operator will do nothing
		// any value compared null returns null
		// TODO: if we support <=> later, we must handle null
		return &NullPlan{r.GetFields()}, true, nil
	}
	return &indexPlan{
		src:     t,
		colName: cn,
		idxName: ix.Name.O,
		idx:     ix.X,
		spans:   toSpans(x.Op, rval),
	}, true, nil
}

// Manipulate mysql.tables_priv table.
func (s *GrantStmt) grantColumnPriv(ctx context.Context, priv *coldef.PrivElem, user *coldef.UserSpecification) error {
	db, tbl, err := s.getTargetSchemaAndTable(ctx)
	if err != nil {
		return errors.Trace(err)
	}
	userName, host := parseUser(user.User)
	for _, c := range priv.Cols {
		col := column.FindCol(tbl.Cols(), c)
		if col == nil {
			return errors.Errorf("Unknown column: %s", c)
		}
		asgns, err := composeColumnPrivUpdate(ctx, priv.Priv, userName, host, db.Name.O, tbl.TableName().O, col.Name.O)
		if err != nil {
			return errors.Trace(err)
		}
		sql := fmt.Sprintf(`UPDATE %s.%s SET %s WHERE User="%s" AND Host="%s" AND DB="%s" AND Table_name="%s" AND Column_name="%s";`, mysql.SystemDB, mysql.ColumnPrivTable, asgns, userName, host, db.Name.O, tbl.TableName().O, col.Name.O)
		_, err = ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(ctx, sql)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (d *ddl) CreateIndex(ctx context.Context, ti table.Ident, unique bool, indexName model.CIStr, idxColNames []*coldef.IndexColName) error {
	is := d.infoHandle.Get()
	schema, ok := is.SchemaByName(ti.Schema)
	if !ok {
		return errors.Trace(qerror.ErrDatabaseNotExist)
	}

	t, err := is.TableByName(ti.Schema, ti.Name)
	if err != nil {
		return errors.Trace(err)
	}
	if _, ok := is.IndexByName(ti.Schema, ti.Name, indexName); ok {
		return errors.Errorf("CREATE INDEX: index already exist %s", indexName)
	}
	if is.ColumnExists(ti.Schema, ti.Name, indexName) {
		return errors.Errorf("CREATE INDEX: index name collision with existing column: %s", indexName)
	}

	tbInfo := t.Meta()

	// build offsets
	idxColumns := make([]*model.IndexColumn, 0, len(idxColNames))
	for i, ic := range idxColNames {
		col := column.FindCol(t.Cols(), ic.ColumnName)
		if col == nil {
			return errors.Errorf("CREATE INDEX: column does not exist: %s", ic.ColumnName)
		}
		idxColumns = append(idxColumns, &model.IndexColumn{
			Name:   col.Name,
			Offset: col.Offset,
			Length: ic.Length,
		})
		// Set ColumnInfo flag
		if i == 0 {
			if unique && len(idxColNames) == 1 {
				tbInfo.Columns[col.Offset].Flag |= mysql.UniqueKeyFlag
			} else {
				tbInfo.Columns[col.Offset].Flag |= mysql.MultipleKeyFlag
			}
		}
	}
	// create index info
	idxInfo := &model.IndexInfo{
		Name:    indexName,
		Columns: idxColumns,
		Unique:  unique,
	}
	tbInfo.Indices = append(tbInfo.Indices, idxInfo)

	// build index
	err = d.buildIndex(ctx, t, idxInfo, unique)
	if err != nil {
		return errors.Trace(err)
	}

	// update InfoSchema
	err = kv.RunInNewTxn(d.store, false, func(txn kv.Transaction) error {
		t := meta.NewMeta(txn)
		err := d.verifySchemaMetaVersion(t, is.SchemaMetaVersion())
		if err != nil {
			return errors.Trace(err)
		}

		err = t.UpdateTable(schema.ID, tbInfo)
		return errors.Trace(err)
	})
	if d.onDDLChange != nil {
		err = d.onDDLChange(err)
	}
	return errors.Trace(err)
}

// Add a column into table
func (d *ddl) addColumn(ctx context.Context, schema *model.DBInfo, tbl table.Table, spec *AlterSpecification, schemaMetaVersion int64) error {
	// Find position
	cols := tbl.Cols()
	position := len(cols)
	name := spec.Column.Name
	// Check column name duplicate.
	dc := column.FindCol(cols, name)
	if dc != nil {
		return errors.Errorf("Try to add a column with the same name of an already exists column.")
	}
	if spec.Position.Type == ColumnPositionFirst {
		position = 0
	} else if spec.Position.Type == ColumnPositionAfter {
		// Find the mentioned column.
		c := column.FindCol(cols, spec.Position.RelativeColumn)
		if c == nil {
			return errors.Errorf("No such column: %v", name)
		}
		// Insert position is after the mentioned column.
		position = c.Offset + 1
	}
	// TODO: set constraint
	col, _, err := d.buildColumnAndConstraint(position, spec.Column)
	if err != nil {
		return errors.Trace(err)
	}
	// insert col into the right place of the column list
	newCols := make([]*column.Col, 0, len(cols)+1)
	newCols = append(newCols, cols[:position]...)
	newCols = append(newCols, col)
	newCols = append(newCols, cols[position:]...)
	// adjust position
	if position != len(cols) {
		offsetChange := make(map[int]int)
		for i := position + 1; i < len(newCols); i++ {
			offsetChange[newCols[i].Offset] = i
			newCols[i].Offset = i
		}
		// Update index offset info
		for _, idx := range tbl.Indices() {
			for _, c := range idx.Columns {
				newOffset, ok := offsetChange[c.Offset]
				if ok {
					c.Offset = newOffset
				}
			}
		}
	}
	tb := tbl.(*tables.Table)
	tb.Columns = newCols

	// TODO: update index
	if err = updateOldRows(ctx, tb, col); err != nil {
		return errors.Trace(err)
	}

	// update infomation schema
	err = kv.RunInNewTxn(d.store, false, func(txn kv.Transaction) error {
		t := meta.NewMeta(txn)
		err := d.verifySchemaMetaVersion(t, schemaMetaVersion)
		if err != nil {
			return errors.Trace(err)
		}

		err = t.UpdateTable(schema.ID, tb.Meta())
		return errors.Trace(err)
	})
	if d.onDDLChange != nil {
		err = d.onDDLChange(err)
	}
	return errors.Trace(err)
}