From 89ac601f8648640ec964a29f8d65b2d35c2cfdd1 Mon Sep 17 00:00:00 2001
From: Lynn <zimu_xia@126.com>
Date: Wed, 18 Sep 2019 20:11:30 +0800
Subject: [PATCH] ddl: check set default value is string type

---
 ddl/db_test.go | 21 ++++++++++++++----
 ddl/ddl_api.go | 58 +++++++++++++++++++++++++++++++++++++++++---------
 2 files changed, 65 insertions(+), 14 deletions(-)

diff --git a/ddl/db_test.go b/ddl/db_test.go
index be8de9985bef6..49371d0e8d840 100644
--- a/ddl/db_test.go
+++ b/ddl/db_test.go
@@ -1839,16 +1839,29 @@ func (s *testDBSuite2) TestCreateTableWithSetCol(c *C) {
 		"  `a` int(11) DEFAULT NULL,\n" +
 		"  `b` set('e') DEFAULT ''\n" +
 		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+	s.tk.MustExec("drop table t_set")
+	s.tk.MustExec("create table t_set (a set('a', 'b', 'c', 'd') default 'a,C,c');")
+	s.tk.MustQuery("show create table t_set").Check(testkit.Rows("t_set CREATE TABLE `t_set` (\n" +
+		"  `a` set('a','b','c','d') DEFAULT 'a,c'\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
 
-	// The type of default value is int.
-	// for failure cases
+	// It's for failure cases.
+	// The type of default value is string.
 	s.tk.MustExec("drop table t_set")
-	failedSQL := "create table t_set (a set('1', '4', '10') default 0);"
+	failedSQL := "create table t_set (a set('1', '4', '10') default '3');"
+	s.tk.MustGetErrCode(failedSQL, tmysql.ErrInvalidDefault)
+	failedSQL = "create table t_set (a set('1', '4', '10') default '1,4,11');"
+	s.tk.MustGetErrCode(failedSQL, tmysql.ErrInvalidDefault)
+	failedSQL = "create table t_set (a set('1', '4', '10') default '1 ,4');"
+	s.tk.MustGetErrCode(failedSQL, tmysql.ErrInvalidDefault)
+	// The type of default value is int.
+	failedSQL = "create table t_set (a set('1', '4', '10') default 0);"
 	s.tk.MustGetErrCode(failedSQL, tmysql.ErrInvalidDefault)
 	failedSQL = "create table t_set (a set('1', '4', '10') default 8);"
 	s.tk.MustGetErrCode(failedSQL, tmysql.ErrInvalidDefault)
 
-	// for successful cases
+	// The type of default value is int.
+	// It's for successful cases
 	s.tk.MustExec("create table t_set (a set('1', '4', '10', '21') default 1);")
 	s.tk.MustQuery("show create table t_set").Check(testkit.Rows("t_set CREATE TABLE `t_set` (\n" +
 		"  `a` set('1','4','10','21') DEFAULT '1'\n" +
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
index 3a149d5534b0e..fd6fa22a9d449 100644
--- a/ddl/ddl_api.go
+++ b/ddl/ddl_api.go
@@ -680,7 +680,26 @@ func getDefaultValue(ctx sessionctx.Context, col *table.Column, c *ast.ColumnOpt
 		}
 		return strconv.FormatUint(value, 10), nil
 	}
-	if tp == mysql.TypeSet && v.Kind() == types.KindInt64 {
+
+	switch tp {
+	case mysql.TypeSet:
+		return setSetDefaultValue(v, col)
+	case mysql.TypeDuration:
+		if v, err = v.ConvertTo(ctx.GetSessionVars().StmtCtx, &col.FieldType); err != nil {
+			return "", errors.Trace(err)
+		}
+	case mysql.TypeBit:
+		if v.Kind() == types.KindInt64 || v.Kind() == types.KindUint64 {
+			// For BIT fields, convert int into BinaryLiteral.
+			return types.NewBinaryLiteralFromUint(v.GetUint64(), -1).ToString(), nil
+		}
+	}
+
+	return v.ToString()
+}
+
+func setSetDefaultValue(v types.Datum, col *table.Column) (string, error) {
+	if v.Kind() == types.KindInt64 {
 		setCnt := len(col.Elems)
 		maxLimit := int64(1<<uint(setCnt) - 1)
 		val := v.GetInt64()
@@ -692,21 +711,40 @@ func getDefaultValue(ctx sessionctx.Context, col *table.Column, c *ast.ColumnOpt
 			return "", errors.Trace(err)
 		}
 		v.SetMysqlSet(setVal)
+		return v.ToString()
 	}
 
-	if tp == mysql.TypeDuration {
-		var err error
-		if v, err = v.ConvertTo(ctx.GetSessionVars().StmtCtx, &col.FieldType); err != nil {
-			return "", errors.Trace(err)
-		}
+	str, err := v.ToString()
+	if err != nil {
+		return "", errors.Trace(err)
+	}
+	if str == "" {
+		return str, nil
 	}
 
-	if tp == mysql.TypeBit {
-		if v.Kind() == types.KindInt64 || v.Kind() == types.KindUint64 {
-			// For BIT fields, convert int into BinaryLiteral.
-			return types.NewBinaryLiteralFromUint(v.GetUint64(), -1).ToString(), nil
+	valMap := make(map[string]struct{}, len(col.Elems))
+	dVals := strings.SplitN(strings.ToLower(str), ",", -1)
+	for _, dv := range dVals {
+		valMap[dv] = struct{}{}
+	}
+	var existCnt int
+	for dv := range valMap {
+		for i := range col.Elems {
+			e := strings.ToLower(col.Elems[i])
+			if e == dv {
+				existCnt++
+				break
+			}
 		}
 	}
+	if existCnt != len(valMap) {
+		return "", ErrInvalidDefaultValue.GenWithStackByArgs(col.Name.O)
+	}
+	setVal, err := types.ParseSetName(col.Elems, str)
+	if err != nil {
+		return "", ErrInvalidDefaultValue.GenWithStackByArgs(col.Name.O)
+	}
+	v.SetMysqlSet(setVal)
 
 	return v.ToString()
 }