ddl: add check when create table with foreign key.

ddl/ddl.go

@@ -96,7 +96,7 @@ var (
 	errIncorrectPrefixKey   = terror.ClassDDL.New(codeIncorrectPrefixKey, "Incorrect prefix key; the used key part isn't a string, the used length is longer than the key part, or the storage engine doesn't support unique prefix keys")
 	errTooLongKey           = terror.ClassDDL.New(codeTooLongKey,
 		fmt.Sprintf("Specified key was too long; max key length is %d bytes", maxPrefixLength))
-	errKeyColumnDoesNotExits    = terror.ClassDDL.New(codeKeyColumnDoesNotExits, "this key column doesn't exist in table")
+	errKeyColumnDoesNotExits    = terror.ClassDDL.New(codeKeyColumnDoesNotExits, mysql.MySQLErrName[mysql.ErrKeyColumnDoesNotExits])
 	errUnknownTypeLength        = terror.ClassDDL.New(codeUnknownTypeLength, "Unknown length for type tp %d")
 	errUnknownFractionLength    = terror.ClassDDL.New(codeUnknownFractionLength, "Unknown Length for type tp %d and fraction %d")
 	errInvalidJobVersion        = terror.ClassDDL.New(codeInvalidJobVersion, "DDL job with version %d greater than current %d")
@@ -583,7 +583,7 @@ const (
 	codeTooLongIdent                           = 1059
 	codeDupKeyName                             = 1061
 	codeTooLongKey                             = 1071
-	codeKeyColumnDoesNotExits                  = 1072
+	codeKeyColumnDoesNotExits                  = mysql.ErrKeyColumnDoesNotExits
 	codeIncorrectPrefixKey                     = 1089
 	codeCantRemoveAllFields                    = 1090
 	codeCantDropFieldOrKey                     = 1091

ddl/ddl_api.go

@@ -727,6 +727,9 @@ func buildTableInfo(ctx sessionctx.Context, d *ddl, tableName model.CIStr, cols
 			fk.RefTable = constr.Refer.Table.Name
 			fk.State = model.StatePublic
 			for _, key := range constr.Keys {
+				if !table.FindColInfo(tbInfo.Columns, key.Column.Name.O) {
+					return nil, errKeyColumnDoesNotExits.GenWithStackByArgs(key.Column.Name)
+				}
 				fk.Cols = append(fk.Cols, key.Column.Name)
 			}
 			for _, key := range constr.Refer.IndexColNames {
@@ -749,7 +752,7 @@ func buildTableInfo(ctx sessionctx.Context, d *ddl, tableName model.CIStr, cols
 			for _, key := range constr.Keys {
 				col = table.FindCol(cols, key.Column.Name.O)
 				if col == nil {
-					return nil, errKeyColumnDoesNotExits.GenWithStack("key column %s doesn't exist in table", key.Column.Name)
+					return nil, errKeyColumnDoesNotExits.GenWithStackByArgs(key.Column.Name)
 				}
 				// Virtual columns cannot be used in primary key.
 				if col.IsGenerated() && !col.GeneratedStored {

executor/executor_test.go

@@ -303,6 +303,15 @@ func (s *testSuite) TestAdmin(c *C) {
 	tk.MustExec("admin check table t1;")
 }
 
+func (s *testSuite) TestTableForeignKey(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1 (a int, b int);")
+	_, err := tk.Exec("create table t2 (c int, foreign key (a) references t1(a));")
+	c.Assert(err.Error(), Equals, "[ddl:1072]Key column 'a' doesn't exist in table")
+	tk.MustExec("drop table if exists t1,t2;")
+}
+
 func (s *testSuite) fillData(tk *testkit.TestKit, table string) {
 	tk.MustExec("use test")
 	tk.MustExec(fmt.Sprintf("create table %s(id int not null default 1, name varchar(255), PRIMARY KEY(id));", table))

table/column.go

@@ -70,6 +70,16 @@ func FindCol(cols []*Column, name string) *Column {
 	return nil
 }
 
+// FindColInfo finds column in cols by name.
+func FindColInfo(cols []*model.ColumnInfo, name string) bool {
+	for _, col := range cols {
+		if strings.EqualFold(col.Name.O, name) {
+			return true
+		}
+	}
+	return false
+}
+
 // ToColumn converts a *model.ColumnInfo to *Column.
 func ToColumn(col *model.ColumnInfo) *Column {
 	return &Column{