Wrong behavior when executing alter table t add index c (c(3)); (pi…

…ngcap#2432) (pingcap#2441)

ddl/ddl_api.go

@@ -225,7 +225,7 @@ func columnDefToCol(ctx context.Context, offset int, colDef *ast.ColumnDef) (*ta
 	if colDef.Options != nil {
 		len := types.UnspecifiedLength
 
-		if types.IsTypeSpecifiable(colDef.Tp.Tp) {
+		if types.IsTypePrefixable(colDef.Tp.Tp) {
 			len = colDef.Tp.Flen
 		}
 

ddl/index.go

@@ -55,11 +55,18 @@ func buildIndexColumns(columns []*model.ColumnInfo, idxColNames []*ast.IndexColN
 			return nil, errors.Trace(errBlobKeyWithoutLength)
 		}
 
+		// Length can only be specified for specifiable types.
+		if ic.Length != types.UnspecifiedLength && !types.IsTypePrefixable(col.FieldType.Tp) {
+			return nil, errors.Trace(errIncorrectPrefixKey)
+		}
+
+		// Key length must be shorter or equal to the column length.
 		if ic.Length != types.UnspecifiedLength &&
-			!types.IsTypeSpecifiable(col.FieldType.Tp) {
+			types.IsTypeChar(col.FieldType.Tp) && col.Flen < ic.Length {
 			return nil, errors.Trace(errIncorrectPrefixKey)
 		}
 
+		// Specified length must be shorter than the max length for prefix.
 		if ic.Length > maxPrefixLength {
 			return nil, errors.Trace(errTooLongKey)
 		}
@@ -84,10 +91,8 @@ func buildIndexColumns(columns []*model.ColumnInfo, idxColNames []*ast.IndexColN
 				}
 
 				// Special case for time fraction.
-				if (col.FieldType.Tp == mysql.TypeDatetime ||
-					col.FieldType.Tp == mysql.TypeDuration ||
-					col.FieldType.Tp == mysql.TypeTimestamp) && col.FieldType.Decimal != -1 {
-
+				if types.IsTypeFractionable(col.FieldType.Tp) &&
+					col.FieldType.Decimal != types.UnspecifiedLength {
 					if len, ok := mysql.DefaultLengthOfTimeFraction[col.FieldType.Decimal]; ok {
 						sumLength += len
 					} else {
@@ -97,6 +102,7 @@ func buildIndexColumns(columns []*model.ColumnInfo, idxColNames []*ast.IndexColN
 			}
 		}
 
+		// The sum of all lengths must be shorter than the max length for prefix.
 		if sumLength > maxPrefixLength {
 			return nil, errors.Trace(errTooLongKey)
 		}

session_test.go

@@ -2259,7 +2259,35 @@ func (s *testSessionSuite) TestSpecifyIndexPrefixLength(c *C) {
 	se := newSession(c, store, s.dbName)
 	mustExecSQL(c, se, "drop table if exists t;")
 
-	_, err := exec(se, "create table t (c1 int, c2 blob, c3 varchar(64), index(c2));")
+	_, err := exec(se, "create table t (c1 char, index(c1(3)));")
+	// ERROR 1089 (HY000): 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
+	c.Assert(err, NotNil)
+
+	_, err = exec(se, "create table t (c1 int, index(c1(3)));")
+	// ERROR 1089 (HY000): 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
+	c.Assert(err, NotNil)
+
+	_, err = exec(se, "create table t (c1 bit(10), index(c1(3)));")
+	// ERROR 1089 (HY000): 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
+	c.Assert(err, NotNil)
+
+	mustExecSQL(c, se, "create table t (c1 char, c2 int, c3 bit(10));")
+
+	_, err = exec(se, "create index idx_c1 on t (c1(3));")
+	// ERROR 1089 (HY000): 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
+	c.Assert(err, NotNil)
+
+	_, err = exec(se, "create index idx_c1 on t (c2(3));")
+	// ERROR 1089 (HY000): 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
+	c.Assert(err, NotNil)
+
+	_, err = exec(se, "create index idx_c1 on t (c3(3));")
+	// ERROR 1089 (HY000): 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
+	c.Assert(err, NotNil)
+
+	mustExecSQL(c, se, "drop table if exists t;")
+
+	_, err = exec(se, "create table t (c1 int, c2 blob, c3 varchar(64), index(c2));")
 	// ERROR 1170 (42000): BLOB/TEXT column 'c2' used in key specification without a key length
 	c.Assert(err, NotNil)
 

util/types/etc.go

@@ -49,10 +49,16 @@ func IsTypeChar(tp byte) bool {
 	}
 }
 
-//IsTypeSpecifiable returns a boolean indicating
-// whether the tp is a specifiable type.
-func IsTypeSpecifiable(tp byte) bool {
-	return IsTypeBlob(tp) || IsTypeChar(tp) || tp == mysql.TypeBit
+// IsTypePrefixable returns a boolean indicating
+// whether an index on a column with the tp can be defined with a prefix.
+func IsTypePrefixable(tp byte) bool {
+	return IsTypeBlob(tp) || IsTypeChar(tp)
+}
+
+// IsTypeFractionable returns a boolean indicating
+// whether the tp can has time fraction.
+func IsTypeFractionable(tp byte) bool {
+	return tp == mysql.TypeDatetime || tp == mysql.TypeDuration || tp == mysql.TypeTimestamp
 }
 
 var type2Str = map[byte]string{