ddl: validate table info before doing ddl job to avoid load infoschema error

ddl/column.go

@@ -242,6 +242,11 @@ func onDropColumn(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 		colInfo.State = model.StateWriteOnly
 		// Set this column's offset to the last and reset all following columns' offsets.
 		adjustColumnInfoInDropColumn(tblInfo, colInfo.Offset)
+		err = checkTableInfoValid(tblInfo)
+		if err != nil {
+			job.State = model.JobStateCancelled
+			return ver, errors.Trace(err)
+		}
 		// When the dropping column has not-null flag and it hasn't the default value, we can backfill the column value like "add column".
 		// NOTE: If the state of StateWriteOnly can be rollbacked, we'd better reconsider the original default value.
 		// And we need consider the column without not-null flag.

ddl/db_test.go

@@ -1526,6 +1526,13 @@ func (s *testDBSuite2) TestDropColumn(c *C) {
 		}
 	}
 
+	// Test for drop partition table column.
+	s.tk.MustExec("drop table if exists t1")
+	s.tk.MustExec("create table t1 (a int,b int) partition by hash(a) partitions 4;")
+	_, err := s.tk.Exec("alter table t1 drop column a")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[expression:1054]Unknown column 'a' in 'expression'")
+
 	s.tk.MustExec("drop database drop_col_db")
 }
 
@@ -1803,6 +1810,16 @@ func (s *testDBSuite1) TestCreateTable(c *C) {
 	assertErrorCode(c, s.tk, failSQL, tmysql.ErrDuplicatedValueInType)
 	_, err = s.tk.Exec("create table t_enum (a enum('B','b'));")
 	c.Assert(err.Error(), Equals, "[types:1291]Column 'a' has duplicated value 'B' in ENUM")
+
+	// Test for invalid expression.
+	s.tk.MustExec("drop table if exists t1")
+	_, err = s.tk.Exec(`CREATE TABLE t1 (
+		c0 int(11) ,
+  		c1 int(11),
+    	c2 decimal(16,4) GENERATED ALWAYS AS ((case when (c0 = 0) then 0 when (c0 > 0) then (c1 / c0) end))
+	);`)
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[parser:1064]You have an error in your SQL syntax; check the manual that corresponds to your TiDB version for the right syntax to use line 1 column 55 near \"THEN (`c1` / `c0`) END)\" ")
 }
 
 func (s *testDBSuite2) TestTableForeignKey(c *C) {

ddl/ddl_api.go

@@ -1057,6 +1057,11 @@ func (d *ddl) CreateTableWithLike(ctx sessionctx.Context, ident, referIdent ast.
 	return errors.Trace(err)
 }
 
+func checkTableInfoValid(tblInfo *model.TableInfo) error {
+	_, err := tables.TableFromMeta(nil, tblInfo)
+	return err
+}
+
 func buildTableInfoWithLike(ident ast.Ident, referTblInfo *model.TableInfo) model.TableInfo {
 	tblInfo := *referTblInfo
 	// Check non-public column and adjust column offset.
@@ -1191,6 +1196,12 @@ func (d *ddl) CreateTable(ctx sessionctx.Context, s *ast.CreateTableStmt) (err e
 	if err != nil {
 		return errors.Trace(err)
 	}
+	tbInfo.State = model.StatePublic
+	err = checkTableInfoValid(tbInfo)
+	if err != nil {
+		return err
+	}
+	tbInfo.State = model.StateNone
 
 	job := &model.Job{
 		SchemaID:   schema.ID,
@@ -2153,6 +2164,19 @@ func (d *ddl) DropColumn(ctx sessionctx.Context, ti ast.Ident, colName model.CIS
 		return errUnsupportedPKHandle
 	}
 
+	// Clone table info to avoid change memory schema.
+	checkTblInfo := tblInfo.Clone()
+	adjustColumnInfoInDropColumn(checkTblInfo, col.Offset)
+	checkCol := model.FindColumnInfo(checkTblInfo.Columns, colName.L)
+	if checkCol == nil {
+		return ErrCantDropFieldOrKey.GenWithStack("column %s doesn't exist", colName)
+	}
+	checkCol.State = model.StateWriteOnly
+	err = checkTableInfoValid(checkTblInfo)
+	if err != nil {
+		return err
+	}
+
 	job := &model.Job{
 		SchemaID:   schema.ID,
 		TableID:    t.Meta().ID,