ddl: fix alter table charset bug and compatibility

ddl/db_integration_test.go

@@ -24,6 +24,7 @@ import (
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/ast"
+	"github.com/pingcap/parser/charset"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
 	tmysql "github.com/pingcap/parser/mysql"
@@ -40,6 +41,7 @@ import (
 	"github.com/pingcap/tidb/store/mockstore/mocktikv"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/util/mock"
 	"github.com/pingcap/tidb/util/testkit"
 )
 
@@ -587,6 +589,83 @@ func (s *testIntegrationSuite) TestChangingTableCharset(c *C) {
 
 	rs, err = tk.Exec("alter table t charset utf8mb4 collate utf8mb4_bin")
 	c.Assert(err, NotNil)
+
+	// Test change column charset when change table charset.
+	tk.MustExec("drop table t;")
+	tk.MustExec("create table t(a varchar(10)) charset utf8")
+	tk.MustExec("alter table t convert to charset utf8mb4;")
+	checkCharset := func() {
+		tbl := testGetTableByName(c, s.ctx, "test", "t")
+		c.Assert(tbl, NotNil)
+		c.Assert(tbl.Meta().Charset, Equals, charset.CharsetUTF8MB4)
+		c.Assert(tbl.Meta().Collate, Equals, charset.CollationUTF8MB4)
+		for _, col := range tbl.Meta().Columns {
+			c.Assert(col.Charset, Equals, charset.CharsetUTF8MB4)
+			c.Assert(col.Collate, Equals, charset.CollationUTF8MB4)
+		}
+	}
+	checkCharset()
+
+	// Test when column charset can not convert to the target charset.
+	tk.MustExec("drop table t;")
+	tk.MustExec("create table t(a varchar(10) character set ascii) charset utf8mb4")
+	_, err = tk.Exec("alter table t convert to charset utf8mb4;")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[ddl:210]unsupported modify charset from ascii to utf8mb4")
+
+	// Test when table charset is equal to target charset but column charset is not equal.
+	tk.MustExec("drop table t;")
+	tk.MustExec("create table t(a varchar(10) character set utf8) charset utf8mb4")
+	tk.MustExec("alter table t convert to charset utf8mb4;")
+	checkCharset()
+
+	// Mock table info with charset is "". Old TiDB maybe create table with charset is "".
+	db, ok := domain.GetDomain(s.ctx).InfoSchema().SchemaByName(model.NewCIStr("test"))
+	c.Assert(ok, IsTrue)
+	tbl := testGetTableByName(c, s.ctx, "test", "t")
+	tblInfo := tbl.Meta().Clone()
+	tblInfo.Charset = ""
+	tblInfo.Collate = ""
+	updateTableInfo := func(tblInfo *model.TableInfo) {
+		mockCtx := mock.NewContext()
+		mockCtx.Store = s.store
+		err = mockCtx.NewTxn(context.Background())
+		c.Assert(err, IsNil)
+		txn, err := mockCtx.Txn(true)
+		c.Assert(err, IsNil)
+		mt := meta.NewMeta(txn)
+
+		err = mt.UpdateTable(db.ID, tblInfo)
+		c.Assert(err, IsNil)
+		err = txn.Commit(context.Background())
+		c.Assert(err, IsNil)
+	}
+	updateTableInfo(tblInfo)
+
+	// check table charset is ""
+	tk.MustExec("alter table t add column b varchar(10);") //  load latest schema.
+	tbl = testGetTableByName(c, s.ctx, "test", "t")
+	c.Assert(tbl, NotNil)
+	c.Assert(tbl.Meta().Charset, Equals, "")
+	c.Assert(tbl.Meta().Collate, Equals, "")
+	// Test when table charset is "", this for compatibility.
+	tk.MustExec("alter table t convert to charset utf8mb4;")
+	checkCharset()
+
+	// Test when column charset is "".
+	tbl = testGetTableByName(c, s.ctx, "test", "t")
+	tblInfo = tbl.Meta().Clone()
+	tblInfo.Columns[0].Charset = ""
+	tblInfo.Columns[0].Collate = ""
+	updateTableInfo(tblInfo)
+	// check table charset is ""
+	tk.MustExec("alter table t drop column b;") //  load latest schema.
+	tbl = testGetTableByName(c, s.ctx, "test", "t")
+	c.Assert(tbl, NotNil)
+	c.Assert(tbl.Meta().Columns[0].Charset, Equals, "")
+	c.Assert(tbl.Meta().Columns[0].Collate, Equals, "")
+	tk.MustExec("alter table t convert to charset utf8mb4;")
+	checkCharset()
 }
 
 func (s *testIntegrationSuite) TestCaseInsensitiveCharsetAndCollate(c *C) {

ddl/ddl_api.go

@@ -224,18 +224,27 @@ func ResolveCharsetCollation(tblCharset, dbCharset string) (string, string, erro
 	return charset, collate, nil
 }
 
+func typesNeedCharset(tp byte) bool {
+	switch tp {
+	case mysql.TypeString, mysql.TypeVarchar, mysql.TypeVarString,
+		mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob,
+		mysql.TypeEnum, mysql.TypeSet:
+		return true
+	}
+	return false
+}
+
 func setCharsetCollationFlenDecimal(tp *types.FieldType, tblCharset string, dbCharset string) error {
 	tp.Charset = strings.ToLower(tp.Charset)
 	tp.Collate = strings.ToLower(tp.Collate)
 	if len(tp.Charset) == 0 {
-		switch tp.Tp {
-		case mysql.TypeString, mysql.TypeVarchar, mysql.TypeVarString, mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeEnum, mysql.TypeSet:
+		if typesNeedCharset(tp.Tp) {
 			var err error
 			tp.Charset, tp.Collate, err = ResolveCharsetCollation(tblCharset, dbCharset)
 			if err != nil {
 				return errors.Trace(err)
 			}
-		default:
+		} else {
 			tp.Charset = charset.CharsetBin
 			tp.Collate = charset.CharsetBin
 		}
@@ -2536,12 +2545,9 @@ func (d *ddl) AlterTableCharsetAndCollate(ctx sessionctx.Context, ident ast.Iden
 	if err != nil {
 		return errors.Trace(infoschema.ErrTableNotExists.GenWithStackByArgs(ident.Schema, ident.Name))
 	}
-
-	origCharset := tb.Meta().Charset
-	origCollate := tb.Meta().Collate
 	if toCharset == "" {
 		// charset does not change.
-		toCharset = origCharset
+		toCharset = tb.Meta().Charset
 	}
 
 	if toCollate == "" {
@@ -2552,22 +2558,14 @@ func (d *ddl) AlterTableCharsetAndCollate(ctx sessionctx.Context, ident ast.Iden
 		}
 	}
 
-	if origCharset == toCharset && origCollate == toCollate {
-		// nothing to do.
-		return nil
+	doNothing, err := checkAlterTableCharset(tb.Meta(), schema, toCharset, toCollate)
+	if err != nil {
+		return err
 	}
-
-	if err = modifiableCharsetAndCollation(toCharset, toCollate, origCharset, origCollate); err != nil {
-		return errors.Trace(err)
+	if doNothing {
+		return nil
 	}
 
-	for _, col := range tb.Meta().Cols() {
-		if col.Tp == mysql.TypeVarchar {
-			if err = IsTooBigFieldLength(col.Flen, col.Name.O, toCharset); err != nil {
-				return errors.Trace(err)
-			}
-		}
-	}
 	job := &model.Job{
 		SchemaID:   schema.ID,
 		TableID:    tb.Meta().ID,
@@ -2580,6 +2578,62 @@ func (d *ddl) AlterTableCharsetAndCollate(ctx sessionctx.Context, ident ast.Iden
 	return errors.Trace(err)
 }
 
+// checkAlterTableCharset uses to check is it possible to change the charset of table.
+// This function returns 2 variable:
+// doNothing: if doNothing is true, means no need to change any more, because the target charset is same with the charset of table.
+// err: if err is not nil, means it is not possible to change table charset to target charset.
+func checkAlterTableCharset(tblInfo *model.TableInfo, dbInfo *model.DBInfo, toCharset, toCollate string) (doNothing bool, err error) {
+	origCharset := tblInfo.Charset
+	origCollate := tblInfo.Collate
+	if origCharset == toCharset && origCollate == toCollate {
+		// nothing to do.
+		doNothing = true
+		for _, col := range tblInfo.Columns {
+			if col.Charset == charset.CharsetBin {
+				continue
+			}
+			if col.Charset == toCharset && col.Collate == toCollate {
+				continue
+			}
+			doNothing = false
+		}
+		if doNothing {
+			return doNothing, nil
+		}
+	}
+
+	if len(origCharset) == 0 {
+		// The table charset may be "", if the table is create in old TiDB version.
+		// This DDL will update the table charset to default charset.
+		origCharset, origCollate, err = ResolveCharsetCollation("", dbInfo.Charset)
+		if err != nil {
+			return doNothing, err
+		}
+	}
+
+	if err = modifiableCharsetAndCollation(toCharset, toCollate, origCharset, origCollate); err != nil {
+		return doNothing, err
+	}
+
+	for _, col := range tblInfo.Columns {
+		if col.Tp == mysql.TypeVarchar {
+			if err = IsTooBigFieldLength(col.Flen, col.Name.O, toCharset); err != nil {
+				return doNothing, err
+			}
+		}
+		if col.Charset == charset.CharsetBin {
+			continue
+		}
+		if len(col.Charset) == 0 {
+			continue
+		}
+		if err = modifiableCharsetAndCollation(toCharset, toCollate, col.Charset, col.Collate); err != nil {
+			return doNothing, err
+		}
+	}
+	return doNothing, nil
+}
+
 // RenameIndex renames an index.
 // In TiDB, indexes are case-insensitive (so index 'a' and 'A" are considered the same index),
 // but index names are case-sensitive (we can rename index 'a' to 'A')

ddl/rollingback.go

@@ -193,7 +193,7 @@ func rollingbackDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, err
 }
 
 func rollingbackDropSchema(t *meta.Meta, job *model.Job) error {
-	dbInfo, err := checkDropSchema(t, job)
+	dbInfo, err := checkSchemaExistAndCancelNonExistJob(t, job)
 	if err != nil {
 		return errors.Trace(err)
 	}

ddl/schema.go

@@ -71,7 +71,7 @@ func onCreateSchema(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 }
 
 func onDropSchema(t *meta.Meta, job *model.Job) (ver int64, _ error) {
-	dbInfo, err := checkDropSchema(t, job)
+	dbInfo, err := checkSchemaExistAndCancelNonExistJob(t, job)
 	if err != nil {
 		return ver, errors.Trace(err)
 	}
@@ -120,7 +120,7 @@ func onDropSchema(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	return ver, errors.Trace(err)
 }
 
-func checkDropSchema(t *meta.Meta, job *model.Job) (*model.DBInfo, error) {
+func checkSchemaExistAndCancelNonExistJob(t *meta.Meta, job *model.Job) (*model.DBInfo, error) {
 	dbInfo, err := t.GetDatabase(job.SchemaID)
 	if err != nil {
 		return nil, errors.Trace(err)

ddl/table.go

@@ -20,6 +20,7 @@ import (
 	"sync/atomic"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/charset"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/tidb/ddl/util"
 	"github.com/pingcap/tidb/infoschema"
@@ -585,13 +586,36 @@ func onModifyTableCharsetAndCollate(t *meta.Meta, job *model.Job) (ver int64, _
 		return ver, errors.Trace(err)
 	}
 
+	dbInfo, err := checkSchemaExistAndCancelNonExistJob(t, job)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
 	tblInfo, err := getTableInfoAndCancelFaultJob(t, job, job.SchemaID)
 	if err != nil {
 		return ver, errors.Trace(err)
 	}
 
+	// double check.
+	_, err = checkAlterTableCharset(tblInfo, dbInfo, toCharset, toCollate)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
 	tblInfo.Charset = toCharset
 	tblInfo.Collate = toCollate
+	// update column charset.
+	for _, col := range tblInfo.Columns {
+		if typesNeedCharset(col.Tp) {
+			col.Charset = toCharset
+			col.Collate = toCollate
+		} else {
+			col.Charset = charset.CharsetBin
+			col.Collate = charset.CharsetBin
+		}
+	}
+
 	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
 	if err != nil {
 		return ver, errors.Trace(err)