ddl: fix ddl modify column from null to not null bug: check null value again before change to no null

ddl/column.go

@@ -384,10 +384,18 @@ func (w *worker) doModifyColumn(t *meta.Meta, job *model.Job, newCol *model.Colu
 		}
 	})
 
-	if !mysql.HasNotNullFlag(oldCol.Flag) && mysql.HasNotNullFlag(newCol.Flag) && !mysql.HasPreventNullInsertFlag(oldCol.Flag) {
+	// Column from null to not null.
+	if !mysql.HasNotNullFlag(oldCol.Flag) && mysql.HasNotNullFlag(newCol.Flag) {
+		noPreventNullFlag := !mysql.HasPreventNullInsertFlag(oldCol.Flag)
 		// Introduce the `mysql.HasPreventNullInsertFlag` flag to prevent users from inserting or updating null values.
 		ver, err = modifyColumnFromNull2NotNull(w, t, dbInfo, tblInfo, job, oldCol, newCol)
-		return ver, errors.Trace(err)
+		if err != nil {
+			return ver, errors.Trace(err)
+		}
+		// The column should get into prevent null status first.
+		if noPreventNullFlag {
+			return ver, nil
+		}
 	}
 
 	// We need the latest column's offset and state. This information can be obtained from the store.

ddl/db_test.go

@@ -2470,50 +2470,73 @@ LOOP:
 
 func (s *testDBSuite1) TestModifyColumnNullToNotNull(c *C) {
 	s.tk = testkit.NewTestKit(c, s.store)
+	tk2 := testkit.NewTestKit(c, s.store)
+	tk2.MustExec("use test_db")
 	s.mustExec(c, "use test_db")
 	s.mustExec(c, "drop table if exists t1")
 	s.mustExec(c, "create table t1 (c1 int, c2 int);")
 
 	tbl := s.testGetTable(c, "t1")
-	var insertErr error
-	hook := &ddl.TestDDLCallback{}
-	hook.OnJobRunBeforeExported = func(job *model.Job) {
-		if tbl.Meta().ID != job.TableID {
-			return
-		}
-		var c2 *table.Column
+	getModifyColumn := func() *table.Column {
 		t := s.testGetTable(c, "t1")
 		for _, col := range t.Cols() {
 			if col.Name.L == "c2" {
-				c2 = col
+				return col
 			}
 		}
-		if mysql.HasPreventNullInsertFlag(c2.Flag) {
-			_, insertErr = s.tk.Exec("insert into t1 values ();")
-		}
+		return nil
 	}
 
 	originalHook := s.dom.DDL().GetHook()
+	defer s.dom.DDL().(ddl.DDLForTest).SetHook(originalHook)
+
+	// Check insert null before job first update.
+	times := 0
+	hook := &ddl.TestDDLCallback{}
+	s.tk.MustExec("delete from t1")
+	hook.OnJobUpdatedExported = func(job *model.Job) {
+		if tbl.Meta().ID != job.TableID {
+			return
+		}
+		if job.State != model.JobStateRunning {
+			return
+		}
+		if times == 0 {
+			tk2.MustExec("insert into t1 values ();")
+		}
+		times++
+	}
 	s.dom.DDL().(ddl.DDLForTest).SetHook(hook)
-	done := make(chan error, 1)
-	go backgroundExec(s.store, "alter table t1 change c2 c2 bigint not null;", done)
-	err := <-done
-	c.Assert(err, IsNil)
+	_, err := s.tk.Exec("alter table t1 change c2 c2 int not null;")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[ddl:1138]Invalid use of NULL value")
+	s.tk.MustQuery("select * from t1").Check(testkit.Rows("<nil> <nil>"))
 
-	var c2 *table.Column
-	t := s.testGetTable(c, "t1")
-	for _, col := range t.Cols() {
-		if col.Name.L == "c2" {
-			c2 = col
+	// Check insert error when column has prevent null flag.
+	s.tk.MustExec("delete from t1")
+	hook.OnJobUpdatedExported = nil
+	hook.OnJobRunBeforeExported = func(job *model.Job) {
+		if tbl.Meta().ID != job.TableID {
+			return
+		}
+		if job.State != model.JobStateRunning {
+			return
+		}
+		c2 := getModifyColumn()
+		if mysql.HasPreventNullInsertFlag(c2.Flag) {
+			_, err := s.tk.Exec("insert into t1 values ();")
+			c.Assert(err.Error(), Equals, "[table:1048]Column 'c2' cannot be null")
 		}
 	}
+
+	s.dom.DDL().(ddl.DDLForTest).SetHook(hook)
+	s.tk.MustExec("alter table t1 change c2 c2 bigint not null;")
+
+	c2 := getModifyColumn()
 	c.Assert(mysql.HasNotNullFlag(c2.Flag), IsTrue)
 	c.Assert(mysql.HasPreventNullInsertFlag(c2.Flag), IsFalse)
-	c.Assert(insertErr.Error(), Equals, "[table:1048]Column 'c2' cannot be null")
-	_, insertErr = s.tk.Exec("insert into t1 values ();")
-	c.Assert(insertErr.Error(), Equals, "[table:1364]Field 'c2' doesn't have a default value")
-	s.dom.DDL().(ddl.DDLForTest).SetHook(originalHook)
-	s.mustExec(c, "drop table t1")
+	_, err = s.tk.Exec("insert into t1 values ();")
+	c.Assert(err.Error(), Equals, "[table:1364]Field 'c2' doesn't have a default value")
 }
 
 func (s *testDBSuite2) TestTransactionOnAddDropColumn(c *C) {