ddl: check if the column name already exists when we rename the column name

ddl/column.go

@@ -451,6 +451,14 @@ func (d *ddl) doModifyColumn(t *meta.Meta, job *model.Job, newCol *model.ColumnI
 		job.State = model.JobStateCancelled
 		return ver, infoschema.ErrColumnNotExists.GenByArgs(oldName, tblInfo.Name)
 	}
+	// If we want to rename the column name, we need to check whether it already exists.
+	if newCol.Name.L != oldName.L {
+		c := findCol(tblInfo.Columns, newCol.Name.L)
+		if c != nil {
+			job.State = model.JobStateCancelled
+			return ver, infoschema.ErrColumnExists.GenByArgs(newCol.Name)
+		}
+	}
 
 	// We need the latest column's offset and state. This information can be obtained from the store.
 	newCol.Offset = oldCol.Offset

ddl/ddl_api.go

@@ -1350,6 +1350,14 @@ func (d *ddl) getModifiableColumnJob(ctx sessionctx.Context, ident ast.Ident, or
 	if col == nil {
 		return nil, infoschema.ErrColumnNotExists.GenByArgs(originalColName, ident.Name)
 	}
+	newColName := specNewColumn.Name.Name
+	// If we want to rename the column name, we need to check whether it already exists.
+	if newColName.L != originalColName.L {
+		c := table.FindCol(t.Cols(), newColName.L)
+		if c != nil {
+			return nil, infoschema.ErrColumnExists.GenByArgs(newColName)
+		}
+	}
 
 	// Constraints in the new column means adding new constraints. Errors should thrown,
 	// which will be done by `setDefaultAndComment` later.
@@ -1362,7 +1370,7 @@ func (d *ddl) getModifiableColumnJob(ctx sessionctx.Context, ident ast.Ident, or
 		ID:                 col.ID,
 		OriginDefaultValue: col.OriginDefaultValue,
 		FieldType:          *specNewColumn.Tp,
-		Name:               specNewColumn.Name.Name,
+		Name:               newColName,
 	})
 
 	err = setCharsetCollationFlenDecimal(&newCol.FieldType)

ddl/ddl_db_change_test.go

@@ -501,3 +501,74 @@ func (s *testStateChangeSuite) TestParallelDDL(c *C) {
 	callback = &ddl.TestDDLCallback{}
 	d.SetHook(callback)
 }
+
+func (s *testStateChangeSuite) TestParallelChangeColumnName(c *C) {
+	_, err := s.se.Execute(context.Background(), "use test_db_state")
+	c.Assert(err, IsNil)
+	_, err = s.se.Execute(context.Background(), "create table t(a int, b int, c int)")
+	c.Assert(err, IsNil)
+	defer s.se.Execute(context.Background(), "drop table t")
+
+	callback := &ddl.TestDDLCallback{}
+	once := sync.Once{}
+	callback.OnJobUpdatedExported = func(job *model.Job) {
+		// Make sure the both DDL statements have entered the DDL queue before running the DDL jobs.
+		once.Do(func() {
+			var qLen int64
+			var err error
+			for {
+				kv.RunInNewTxn(s.store, false, func(txn kv.Transaction) error {
+					m := meta.NewMeta(txn)
+					qLen, err = m.DDLJobQueueLen()
+					if err != nil {
+						return err
+					}
+					return nil
+				})
+				if qLen == 2 {
+					break
+				}
+				time.Sleep(5 * time.Millisecond)
+			}
+		})
+	}
+	d := s.dom.DDL()
+	d.SetHook(callback)
+
+	// Use two sessions to run DDL statements in parallel.
+	wg := sync.WaitGroup{}
+	var err1 error
+	var err2 error
+	se, err := session.CreateSession(s.store)
+	c.Assert(err, IsNil)
+	_, err = se.Execute(context.Background(), "use test_db_state")
+	c.Assert(err, IsNil)
+	se1, err := session.CreateSession(s.store)
+	c.Assert(err, IsNil)
+	_, err = se1.Execute(context.Background(), "use test_db_state")
+	c.Assert(err, IsNil)
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		_, err1 = se.Execute(context.Background(), "ALTER TABLE t CHANGE a aa int;")
+	}()
+	go func() {
+		defer wg.Done()
+		_, err2 = se1.Execute(context.Background(), "ALTER TABLE t CHANGE b aa int;")
+	}()
+
+	wg.Wait()
+	// Make sure only a DDL encounters the error of 'duplicate column name'.
+	var oneErr error
+	if (err1 != nil && err2 == nil) || (err1 == nil && err2 != nil) {
+		if err1 != nil {
+			oneErr = err1
+		} else {
+			oneErr = err2
+		}
+	}
+	c.Assert(oneErr.Error(), Equals, "[schema:1060]Duplicate column name 'aa'")
+
+	callback = &ddl.TestDDLCallback{}
+	d.SetHook(callback)
+}

ddl/ddl_db_test.go

@@ -1145,6 +1145,10 @@ func (s *testDBSuite) TestChangeColumn(c *C) {
 	s.testErrorCode(c, sql, tmysql.ErrUnknown)
 	sql = "alter table t3 modify en enum('a', 'z', 'b', 'c') not null default 'a'"
 	s.testErrorCode(c, sql, tmysql.ErrUnknown)
+	// Rename to an existing column.
+	s.mustExec(c, "alter table t3 add column a bigint")
+	sql = "alter table t3 change aa a bigint"
+	s.testErrorCode(c, sql, tmysql.ErrDupFieldName)
 
 	s.tk.MustExec("drop table t3")
 }