ddl: add support first/after for modify/change column (pingcap#3215)

ddl/column.go

@@ -427,13 +427,98 @@ func (d *ddl) onSetDefaultValue(t *meta.Meta, job *model.Job) error {
 func (d *ddl) onModifyColumn(t *meta.Meta, job *model.Job) error {
 	newCol := &model.ColumnInfo{}
 	oldColName := &model.CIStr{}
-	err := job.DecodeArgs(newCol, oldColName)
+	pos := &ast.ColumnPosition{}
+	err := job.DecodeArgs(newCol, oldColName, pos)
 	if err != nil {
 		job.State = model.JobCancelled
 		return errors.Trace(err)
 	}
 
-	return errors.Trace(d.updateColumn(t, job, newCol, oldColName))
+	return errors.Trace(d.doModifyColumn(t, job, newCol, oldColName, pos))
+}
+
+// doModifyColumn updates the column information and reorders all columns.
+func (d *ddl) doModifyColumn(t *meta.Meta, job *model.Job, col *model.ColumnInfo, oldName *model.CIStr, pos *ast.ColumnPosition) error {
+	tblInfo, err := getTableInfo(t, job, job.SchemaID)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	oldCol := findCol(tblInfo.Columns, oldName.L)
+	if oldCol == nil || oldCol.State != model.StatePublic {
+		job.State = model.JobCancelled
+		return infoschema.ErrColumnNotExists.GenByArgs(oldName, tblInfo.Name)
+	}
+
+	// Calculate column's new position.
+	oldPos, newPos := oldCol.Offset, oldCol.Offset
+	if pos.Tp == ast.ColumnPositionAfter {
+		if oldName.L == pos.RelativeColumn.Name.L {
+			// `alter table tableName modify column b int after b` will return ErrColumnNotExists.
+			job.State = model.JobCancelled
+			return infoschema.ErrColumnNotExists.GenByArgs(oldName, tblInfo.Name)
+		}
+
+		relative := findCol(tblInfo.Columns, pos.RelativeColumn.Name.L)
+		if relative == nil || relative.State != model.StatePublic {
+			job.State = model.JobCancelled
+			return infoschema.ErrColumnNotExists.GenByArgs(pos.RelativeColumn, tblInfo.Name)
+		}
+
+		if relative.Offset < oldPos {
+			newPos = relative.Offset + 1
+		} else {
+			newPos = relative.Offset
+		}
+	} else if pos.Tp == ast.ColumnPositionFirst {
+		newPos = 0
+	}
+
+	columnChanged := make(map[string]*model.ColumnInfo)
+	columnChanged[oldName.L] = col
+
+	if newPos == oldPos {
+		tblInfo.Columns[newPos] = col
+	} else {
+		cols := tblInfo.Columns
+
+		// Reorder columns in place.
+		if newPos < oldPos {
+			copy(cols[newPos+1:], cols[newPos:oldPos])
+		} else {
+			copy(cols[oldPos:], cols[oldPos+1:newPos+1])
+		}
+		cols[newPos] = col
+
+		for i, col := range tblInfo.Columns {
+			if col.Offset != i {
+				columnChanged[col.Name.L] = col
+				col.Offset = i
+			}
+		}
+	}
+
+	// Change offset and name in indices.
+	for _, idx := range tblInfo.Indices {
+		for _, c := range idx.Columns {
+			if newCol, ok := columnChanged[c.Name.L]; ok {
+				c.Name = newCol.Name
+				c.Offset = newCol.Offset
+			}
+		}
+	}
+
+	originalState := job.SchemaState
+	job.SchemaState = model.StatePublic
+	ver, err := updateTableInfo(t, job, tblInfo, originalState)
+	if err != nil {
+		job.State = model.JobCancelled
+		return errors.Trace(err)
+	}
+
+	job.State = model.JobDone
+	job.BinlogInfo.AddTableInfo(ver, tblInfo)
+	return nil
 }
 
 func (d *ddl) updateColumn(t *meta.Meta, job *model.Job, newCol *model.ColumnInfo, oldColName *model.CIStr) error {

ddl/ddl_api.go

@@ -1037,8 +1037,7 @@ func (d *ddl) getModifiableColumnJob(ctx context.Context, ident ast.Ident, origi
 	if col == nil {
 		return nil, infoschema.ErrColumnNotExists.GenByArgs(originalColName, ident.Name)
 	}
-	if spec.Constraint != nil || (spec.Position != nil && spec.Position.Tp != ast.ColumnPositionNone) ||
-		spec.NewColumn.Tp == nil {
+	if spec.Constraint != nil || spec.NewColumn.Tp == nil {
 		// Make sure the column definition is simple field type.
 		return nil, errors.Trace(errUnsupportedModifyColumn)
 	}
@@ -1072,7 +1071,7 @@ func (d *ddl) getModifiableColumnJob(ctx context.Context, ident ast.Ident, origi
 		TableID:    t.Meta().ID,
 		Type:       model.ActionModifyColumn,
 		BinlogInfo: &model.HistoryInfo{},
-		Args:       []interface{}{&newCol, originalColName},
+		Args:       []interface{}{&newCol, originalColName, spec.Position},
 	}
 	return job, nil
 }

ddl/ddl_db_test.go

@@ -1114,3 +1114,60 @@ func (s *testDBSuite) TestIssue2858And2717(c *C) {
 	s.tk.MustQuery("select a from t_issue_2858_hex").Check(testkit.Rows("291", "123", "801"))
 	s.tk.MustExec(`alter table t_issue_2858_hex alter column a set default 0x321`)
 }
+
+func (s *testDBSuite) TestChangeColumnPosition(c *C) {
+	defer testleak.AfterTest(c)()
+	s.tk = testkit.NewTestKit(c, s.store)
+	s.tk.MustExec("use " + s.schemaName)
+
+	s.tk.MustExec("create table position (a int default 1, b int default 2)")
+	s.tk.MustExec("insert into position value ()")
+	s.tk.MustExec("insert into position values (3,4)")
+	s.tk.MustQuery("select * from position").Check(testkit.Rows("1 2", "3 4"))
+	s.tk.MustExec("alter table position modify column b int first")
+	s.tk.MustQuery("select * from position").Check(testkit.Rows("2 1", "4 3"))
+	s.tk.MustExec("insert into position value ()")
+	s.tk.MustQuery("select * from position").Check(testkit.Rows("2 1", "4 3", "<nil> 1"))
+
+	s.tk.MustExec("drop table position")
+	s.tk.MustExec("create table position (a int, b int, c double, d varchar(5))")
+	s.tk.MustExec(`insert into position value (1, 2, 3.14, 'TiDB')`)
+	s.tk.MustExec("alter table position modify column d varchar(5) after a")
+	s.tk.MustQuery("select * from position").Check(testkit.Rows("1 TiDB 2 3.14"))
+	s.tk.MustExec("alter table position modify column a int after c")
+	s.tk.MustQuery("select * from position").Check(testkit.Rows("TiDB 2 3.14 1"))
+	s.tk.MustExec("alter table position modify column c double first")
+	s.tk.MustQuery("select * from position").Check(testkit.Rows("3.14 TiDB 2 1"))
+	s.testErrorCode(c, "alter table position modify column b int after b", tmysql.ErrBadField)
+
+	s.tk.MustExec("drop table position")
+	s.tk.MustExec("create table position (a int, b int)")
+	s.tk.MustExec("alter table position add index t(a, b)")
+	s.tk.MustExec("alter table position modify column b int first")
+	s.tk.MustExec("insert into position value (3, 5)")
+	s.tk.MustQuery("select a from position where a = 3").Check(testkit.Rows())
+
+	s.tk.MustExec("alter table position change column b c int first")
+	s.tk.MustQuery("select * from position where c = 3").Check(testkit.Rows("3 5"))
+	s.testErrorCode(c, "alter table position change column c b int after c", tmysql.ErrBadField)
+
+	s.tk.MustExec("drop table position")
+	s.tk.MustExec("create table position (a int default 2)")
+	s.tk.MustExec("alter table position modify column a int default 5 first")
+	s.tk.MustExec("insert into position value ()")
+	s.tk.MustQuery("select * from position").Check(testkit.Rows("5"))
+
+	s.tk.MustExec("drop table position")
+	s.tk.MustExec("create table position (a int, b int)")
+	s.tk.MustExec("alter table position add index t(b)")
+	s.tk.MustExec("alter table position change column b c int first")
+	createSQL := s.tk.MustQuery("show create table position").Rows()[0][1]
+	exceptedSQL := []string{
+		"CREATE TABLE `position` (",
+		"  `c` int(11) DEFAULT NULL,",
+		"  `a` int(11) DEFAULT NULL,",
+		"  KEY `t` (`c`)",
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin",
+	}
+	c.Assert(createSQL, Equals, strings.Join(exceptedSQL, "\n"))
+}

parser/parser.y

@@ -992,12 +992,13 @@ AlterTableSpec:
 			Position:	$4.(*ast.ColumnPosition),
 		}
 	}
-|	"CHANGE" ColumnKeywordOpt ColumnName ColumnDef
+|	"CHANGE" ColumnKeywordOpt ColumnName ColumnDef ColumnPosition
 	{
 		$$ = &ast.AlterTableSpec{
 			Tp:    		ast.AlterTableChangeColumn,
 			OldColumnName:	$3.(*ast.ColumnName),
 			NewColumn: 	$4.(*ast.ColumnDef),
+			Position:	$5.(*ast.ColumnPosition),
 		}
 	}
 |	"ALTER" ColumnKeywordOpt ColumnName "SET" "DEFAULT" SignedLiteral

parser/parser_test.go

@@ -1259,6 +1259,7 @@ func (s *testParserSuite) TestDDL(c *C) {
 		{"ALTER TABLE t ENABLE KEYS", true},
 		{"ALTER TABLE t MODIFY COLUMN a varchar(255)", true},
 		{"ALTER TABLE t CHANGE COLUMN a b varchar(255)", true},
+		{"ALTER TABLE t CHANGE COLUMN a b varchar(255) FIRST", true},
 		{"ALTER TABLE db.t RENAME to db1.t1", true},
 		{"ALTER TABLE t RENAME as t1", true},
 		{"ALTER TABLE t ALTER COLUMN a SET DEFAULT 1", true},