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 not 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

@@ -2465,50 +2465,77 @@ 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
+	getModifyColumn := func() *table.Column {
+		t := s.testGetTable(c, "t1")
+		for _, col := range t.Cols() {
+			if col.Name.L == "c2" {
+				return col
+			}
+		}
+		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)
+	_, 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>"))
+
+	// Check insert error when column has prevent null flag.
+	var insertErr error
+	s.tk.MustExec("delete from t1")
+	hook.OnJobUpdatedExported = nil
 	hook.OnJobRunBeforeExported = func(job *model.Job) {
 		if tbl.Meta().ID != job.TableID {
 			return
 		}
-		var c2 *table.Column
-		t := s.testGetTable(c, "t1")
-		for _, col := range t.Cols() {
-			if col.Name.L == "c2" {
-				c2 = col
-			}
+		if job.State != model.JobStateRunning {
+			return
 		}
+		c2 := getModifyColumn()
 		if mysql.HasPreventNullInsertFlag(c2.Flag) {
 			_, insertErr = s.tk.Exec("insert into t1 values ();")
 		}
 	}
 
-	originalHook := s.dom.DDL().GetHook()
 	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
+	err = <-done
 	c.Assert(err, IsNil)
 
-	var c2 *table.Column
-	t := s.testGetTable(c, "t1")
-	for _, col := range t.Cols() {
-		if col.Name.L == "c2" {
-			c2 = col
-		}
-	}
+	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")
 }
 
 func (s *testDBSuite2) TestTransactionOnAddDropColumn(c *C) {