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,56 @@ 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) character set ascii) 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 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 ascii) 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 "".
+	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)
+	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 = ""
+	err = mt.UpdateTable(db.ID, tblInfo)
+	c.Assert(err, IsNil)
+	err = txn.Commit(context.Background())
+	c.Assert(err, IsNil)
+
+	// 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()
 }
 
 func (s *testIntegrationSuite) TestCaseInsensitiveCharsetAndCollate(c *C) {

ddl/ddl_api.go

@@ -2426,12 +2426,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 == "" {
@@ -2442,22 +2439,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(), 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,
@@ -2470,6 +2459,49 @@ func (d *ddl) AlterTableCharsetAndCollate(ctx sessionctx.Context, ident ast.Iden
 	return errors.Trace(err)
 }
 
+func checkAlterTableCharset(tblInfo *model.TableInfo, toCharset, toCollate string) (bool, error) {
+	var err error
+	origCharset := tblInfo.Charset
+	origCollate := tblInfo.Collate
+	if origCharset == toCharset && origCollate == toCollate {
+		// nothing to do.
+		var 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 true, 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.
+		// Use default charset or database charset?
+		// show create table is uses the default charset.
+		origCharset, origCollate = charset.GetDefaultCharsetAndCollate()
+	}
+
+	if err = modifiableCharsetAndCollation(toCharset, toCollate, origCharset, origCollate); err != nil {
+		return false, errors.Trace(err)
+	}
+
+	for _, col := range tblInfo.Columns {
+		if col.Tp == mysql.TypeVarchar {
+			if err = IsTooBigFieldLength(col.Flen, col.Name.O, toCharset); err != nil {
+				return false, errors.Trace(err)
+			}
+		}
+	}
+	return false, 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/table.go

@@ -20,7 +20,9 @@ import (
 	"sync/atomic"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/charset"
 	"github.com/pingcap/parser/model"
+	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tidb/ddl/util"
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
@@ -589,8 +591,27 @@ func onModifyTableCharsetAndCollate(t *meta.Meta, job *model.Job) (ver int64, _
 		return ver, errors.Trace(err)
 	}
 
+	// double check.
+	_, err = checkAlterTableCharset(tblInfo, 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 {
+		switch col.Tp {
+		case mysql.TypeString, mysql.TypeVarchar, mysql.TypeVarString, mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeEnum, mysql.TypeSet:
+			col.Charset = toCharset
+			col.Collate = toCollate
+		default:
+			col.Charset = charset.CharsetBin
+			col.Collate = charset.CharsetBin
+		}
+	}
+
 	ver, err = updateVersionAndTableInfo(t, job, tblInfo, true)
 	if err != nil {
 		return ver, errors.Trace(err)