ddl: Fix partition value type check (#21016)

ddl/db_partition_test.go

@@ -1883,3 +1883,94 @@ func (s *testIntegrationSuite3) TestCommitWhenSchemaChange(c *C) {
 	tk.MustExec("admin check table schema_change")
 	tk.MustQuery("select * from schema_change").Check(testkit.Rows())
 }
+
+func (s *testIntegrationSuite7) TestCreatePartitionTableWithWrongType(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	var err error
+	_, err = tk.Exec(`create table t(
+	b int(10)
+	) partition by range columns (b) (
+		partition p0 values less than (0x10),
+		partition p3 values less than (0x20)
+	)`)
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec(`create table t(
+	b int(10)
+	) partition by range columns (b) (
+		partition p0 values less than ('g'),
+		partition p3 values less than ('k')
+	)`)
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec(`create table t(
+	b char(10)
+	) partition by range columns (b) (
+		partition p0 values less than (30),
+		partition p3 values less than (60)
+	)`)
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec(`create table t(
+	b datetime
+	) partition by range columns (b) (
+		partition p0 values less than ('g'),
+		partition p3 values less than ('m')
+	)`)
+	c.Assert(err, NotNil)
+}
+
+func (s *testIntegrationSuite7) TestAddPartitionForTableWithWrongType(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop tables if exists t_int, t_char, t_date")
+	tk.MustExec(`create table t_int(b int(10))
+	partition by range columns (b) (
+		partition p0 values less than (10)
+	)`)
+
+	tk.MustExec(`create table t_char(b char(10))
+	partition by range columns (b) (
+		partition p0 values less than ('a')
+	)`)
+
+	tk.MustExec(`create table t_date(b datetime)
+	partition by range columns (b) (
+		partition p0 values less than ('2020-09-01')
+	)`)
+
+	var err error
+
+	_, err = tk.Exec("alter table t_int add partition (partition p1 values less than ('g'))")
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec("alter table t_int add partition (partition p1 values less than (0x20))")
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec("alter table t_char add partition (partition p1 values less than (0x20))")
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec("alter table t_char add partition (partition p1 values less than (10))")
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec("alter table t_date add partition (partition p1 values less than ('m'))")
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec("alter table t_date add partition (partition p1 values less than (0x20))")
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+
+	_, err = tk.Exec("alter table t_date add partition (partition p1 values less than (20))")
+	c.Assert(err, NotNil)
+	c.Assert(ddl.ErrWrongTypeColumnValue.Equal(err), IsTrue)
+}

ddl/ddl_api.go

@@ -4346,10 +4346,32 @@ func checkRangeColumnsTypeAndValuesMatch(ctx sessionctx.Context, meta *model.Tab
 		}
 
 		// Check val.ConvertTo(colType) doesn't work, so we need this case by case check.
+		vkind := val.Kind()
 		switch colType.Tp {
 		case mysql.TypeDate, mysql.TypeDatetime:
-			switch val.Kind() {
+			switch vkind {
 			case types.KindString, types.KindBytes:
+				if _, err := val.ConvertTo(ctx.GetSessionVars().StmtCtx, colType); err != nil {
+					return ErrWrongTypeColumnValue.GenWithStackByArgs()
+				}
+			default:
+				return ErrWrongTypeColumnValue.GenWithStackByArgs()
+			}
+		case mysql.TypeTiny, mysql.TypeShort, mysql.TypeInt24, mysql.TypeLong, mysql.TypeLonglong:
+			switch vkind {
+			case types.KindInt64, types.KindUint64, types.KindNull:
+			default:
+				return ErrWrongTypeColumnValue.GenWithStackByArgs()
+			}
+		case mysql.TypeFloat, mysql.TypeDouble:
+			switch vkind {
+			case types.KindFloat32, types.KindFloat64, types.KindNull:
+			default:
+				return ErrWrongTypeColumnValue.GenWithStackByArgs()
+			}
+		case mysql.TypeString, mysql.TypeVarString:
+			switch vkind {
+			case types.KindString, types.KindBytes, types.KindNull:
 			default:
 				return ErrWrongTypeColumnValue.GenWithStackByArgs()
 			}