From 5940bb2ceb7de833839b09341356ef6b0fe38cb7 Mon Sep 17 00:00:00 2001
From: crazycs <crazycs520@gmail.com>
Date: Tue, 17 Dec 2019 16:52:16 +0800
Subject: [PATCH] ddl: check foreign key constraint when drop,modify,change
 column (#14043)

---
 ddl/column.go  | 11 +++++++++++
 ddl/db_test.go | 15 ++++++++++++++-
 ddl/ddl.go     |  4 ++++
 ddl/ddl_api.go |  8 ++++++++
 4 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/ddl/column.go b/ddl/column.go
index 743f742231..a4eb9287ee 100644
--- a/ddl/column.go
+++ b/ddl/column.go
@@ -535,6 +535,17 @@ func isColumnWithIndex(colName string, indices []*model.IndexInfo) bool {
 	return false
 }
 
+func getColumnForeignKeyInfo(colName string, fkInfos []*model.FKInfo) *model.FKInfo {
+	for _, fkInfo := range fkInfos {
+		for _, col := range fkInfo.Cols {
+			if col.L == colName {
+				return fkInfo
+			}
+		}
+	}
+	return nil
+}
+
 func allocateColumnID(tblInfo *model.TableInfo) int64 {
 	tblInfo.MaxColumnID++
 	return tblInfo.MaxColumnID
diff --git a/ddl/db_test.go b/ddl/db_test.go
index 52ed09b91c..596271bbe2 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -2306,7 +2306,20 @@ func (s *testDBSuite2) TestTableForeignKey(c *C) {
 	// test oreign key not match error
 	failSQL = "alter table t1 add foreign key (a) REFERENCES t3(a, b);"
 	s.tk.MustGetErrCode(failSQL, mysql.ErrWrongFkDef)
-	s.tk.MustExec("drop table if exists t1,t2,t3;")
+	// Test drop column with foreign key.
+	s.tk.MustExec("create table t4 (c int,d int,foreign key (d) references t1 (b));")
+	failSQL = "alter table t4 drop column d"
+	s.tk.MustGetErrCode(failSQL, mysql.ErrFkColumnCannotDrop)
+	// Test change column with foreign key.
+	failSQL = "alter table t4 change column d e bigint;"
+	s.tk.MustGetErrCode(failSQL, mysql.ErrFKIncompatibleColumns)
+	// Test modify column with foreign key.
+	failSQL = "alter table t4 modify column d bigint;"
+	s.tk.MustGetErrCode(failSQL, mysql.ErrFKIncompatibleColumns)
+	s.tk.MustQuery("select count(*) from information_schema.KEY_COLUMN_USAGE;")
+	s.tk.MustExec("alter table t4 drop foreign key d")
+	s.tk.MustExec("alter table t4 modify column d bigint;")
+	s.tk.MustExec("drop table if exists t1,t2,t3,t4;")
 }
 
 func (s *testDBSuite3) TestFKOnGeneratedColumns(c *C) {
diff --git a/ddl/ddl.go b/ddl/ddl.go
index ca13d058e8..b51b5c130c 100644
--- a/ddl/ddl.go
+++ b/ddl/ddl.go
@@ -107,6 +107,8 @@ var (
 	errTooManyFields            = terror.ClassDDL.New(mysql.ErrTooManyFields, mysql.MySQLErrName[mysql.ErrTooManyFields])
 	errInvalidSplitRegionRanges = terror.ClassDDL.New(mysql.ErrInvalidSplitRegionRanges, mysql.MySQLErrName[mysql.ErrInvalidSplitRegionRanges])
 	errReorgPanic               = terror.ClassDDL.New(mysql.ErrReorgPanic, mysql.MySQLErrName[mysql.ErrReorgPanic])
+	errFkColumnCannotDrop       = terror.ClassDDL.New(mysql.ErrFkColumnCannotDrop, mysql.MySQLErrName[mysql.ErrFkColumnCannotDrop])
+	errReferencedForeignKey     = terror.ClassDDL.New(mysql.ErrFKIncompatibleColumns, mysql.MySQLErrName[mysql.ErrFKIncompatibleColumns])
 
 	errOnlyOnRangeListPartition = terror.ClassDDL.New(mysql.ErrOnlyOnRangeListPartition, mysql.MySQLErrName[mysql.ErrOnlyOnRangeListPartition])
 	// errWrongKeyColumn is for table column cannot be indexed.
@@ -728,6 +730,8 @@ func init() {
 		mysql.ErrWrongTableName:                       mysql.ErrWrongTableName,
 		mysql.ErrWrongTypeColumnValue:                 mysql.ErrWrongTypeColumnValue,
 		mysql.WarnDataTruncated:                       mysql.WarnDataTruncated,
+		mysql.ErrFkColumnCannotDrop:                   mysql.ErrFkColumnCannotDrop,
+		mysql.ErrFKIncompatibleColumns:                mysql.ErrFKIncompatibleColumns,
 	}
 	terror.ErrClassToMySQLCodes[terror.ClassDDL] = ddlMySQLErrCodes
 }
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
index aa77ef1286..218b0f3a04 100644
--- a/ddl/ddl_api.go
+++ b/ddl/ddl_api.go
@@ -2638,6 +2638,10 @@ func (d *ddl) getModifiableColumnJob(ctx sessionctx.Context, ident ast.Ident, or
 			return nil, infoschema.ErrColumnExists.GenWithStackByArgs(newColName)
 		}
 	}
+	// Check the column with foreign key.
+	if fkInfo := getColumnForeignKeyInfo(originalColName.L, t.Meta().ForeignKeys); fkInfo != nil {
+		return nil, errReferencedForeignKey.GenWithStackByArgs(originalColName, fkInfo.Name)
+	}
 
 	// Constraints in the new column means adding new constraints. Errors should thrown,
 	// which will be done by `processColumnOptions` later.
@@ -3654,6 +3658,10 @@ func isDroppableColumn(tblInfo *model.TableInfo, colName model.CIStr) error {
 	if isColumnWithIndex(colName.L, tblInfo.Indices) {
 		return errCantDropColWithIndex.GenWithStack("can't drop column %s with index covered now", colName)
 	}
+	// Check the column with foreign key.
+	if fkInfo := getColumnForeignKeyInfo(colName.L, tblInfo.ForeignKeys); fkInfo != nil {
+		return errFkColumnCannotDrop.GenWithStackByArgs(colName, fkInfo.Name)
+	}
 	return nil
 }