*: fix utf8 charset upgrade compatibility (#9820)

config/config.go

@@ -73,8 +73,11 @@ type Config struct {
 	TiKVClient          TiKVClient        `toml:"tikv-client" json:"tikv-client"`
 	Binlog              Binlog            `toml:"binlog" json:"binlog"`
 	CompatibleKillQuery bool              `toml:"compatible-kill-query" json:"compatible-kill-query"`
-	CheckMb4ValueInUtf8 bool              `toml:"check-mb4-value-in-utf8" json:"check-mb4-value-in-utf8"`
-	Plugin              Plugin            `toml:"plugin" json:"plugin"`
+	CheckMb4ValueInUTF8 bool              `toml:"check-mb4-value-in-utf8" json:"check-mb4-value-in-utf8"`
+	// TreatOldVersionUTF8AsUTF8MB4 is use to treat old version table/column UTF8 charset as UTF8MB4. This is for compatibility.
+	// Currently not support dynamic modify, because this need to reload all old version schema.
+	TreatOldVersionUTF8AsUTF8MB4 bool   `toml:"treat-old-version-utf8-as-utf8mb4" json:"treat-old-version-utf8-as-utf8mb4"`
+	Plugin                       Plugin `toml:"plugin" json:"plugin"`
 }
 
 // Log is the log section of config.
@@ -259,19 +262,20 @@ type Binlog struct {
 }
 
 var defaultConf = Config{
-	Host:                "0.0.0.0",
-	AdvertiseAddress:    "",
-	Port:                4000,
-	Store:               "mocktikv",
-	Path:                "/tmp/tidb",
-	RunDDL:              true,
-	SplitTable:          true,
-	Lease:               "45s",
-	TokenLimit:          1000,
-	OOMAction:           "log",
-	MemQuotaQuery:       32 << 30,
-	EnableStreaming:     false,
-	CheckMb4ValueInUtf8: true,
+	Host:                         "0.0.0.0",
+	AdvertiseAddress:             "",
+	Port:                         4000,
+	Store:                        "mocktikv",
+	Path:                         "/tmp/tidb",
+	RunDDL:                       true,
+	SplitTable:                   true,
+	Lease:                        "45s",
+	TokenLimit:                   1000,
+	OOMAction:                    "log",
+	MemQuotaQuery:                32 << 30,
+	EnableStreaming:              false,
+	CheckMb4ValueInUTF8:          true,
+	TreatOldVersionUTF8AsUTF8MB4: true,
 	TxnLocalLatches: TxnLocalLatches{
 		Enabled:  false,
 		Capacity: 10240000,

config/config.toml.example

@@ -51,6 +51,9 @@ compatible-kill-query = false
 # check mb4 value in utf8 is used to control whether to check the mb4 characters when the charset is utf8.
 check-mb4-value-in-utf8 = true
 
+# treat-old-version-utf8-as-utf8mb4 use for upgrade compatibility. Set to true will treat old version table/column UTF8 charset as UTF8MB4.
+treat-old-version-utf8-as-utf8mb4 = true
+
 [log]
 # Log level: debug, info, warn, error, fatal.
 level = "info"

config/config_test.go

@@ -37,7 +37,7 @@ func (s *testConfigSuite) TestConfig(c *C) {
 	conf.Binlog.IgnoreError = true
 	conf.Binlog.BinlogSocket = "/tmp/socket"
 	conf.TiKVClient.CommitTimeout = "10s"
-	conf.CheckMb4ValueInUtf8 = true
+	conf.CheckMb4ValueInUTF8 = true
 	configFile := "config.toml"
 	_, localFile, _, _ := runtime.Caller(0)
 	configFile = path.Join(path.Dir(localFile), configFile)

ddl/column.go

@@ -472,10 +472,6 @@ func generateOriginDefaultValue(col *model.ColumnInfo) (interface{}, error) {
 	if odValue == strings.ToUpper(ast.CurrentTimestamp) {
 		if col.Tp == mysql.TypeTimestamp {
 			odValue = time.Now().UTC().Format(types.TimeFormat)
-			// Version = 1: For OriginDefaultValue and DefaultValue of timestamp column will stores the default time in UTC time zone.
-			//              This will fix bug in version 0.
-			// TODO: remove this version field after there is no old version 0.
-			col.Version = model.ColumnInfoVersion1
 		} else if col.Tp == mysql.TypeDatetime {
 			odValue = time.Now().Format(types.TimeFormat)
 		}

ddl/db_integration_test.go

@@ -14,20 +14,25 @@
 package ddl_test
 
 import (
+	"context"
 	"fmt"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/charset"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/terror"
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/session"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/store/mockstore"
+	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/mock"
 	"github.com/pingcap/tidb/util/testkit"
@@ -314,3 +319,147 @@ func (s *testIntegrationSuite) TestResolveCharset(c *C) {
 	c.Assert(tbl.Cols()[0].Charset, Equals, "binary")
 	c.Assert(tbl.Meta().Charset, Equals, "binary")
 }
+
+func (s *testIntegrationSuite) TestIgnoreColumnUTF8Charset(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	defer tk.MustExec("drop table if exists t")
+
+	tk.MustExec("create table t (a varchar(10) character set utf8, b varchar(10) character set ascii) charset=utf8mb4;")
+	_, err := tk.Exec("insert into t set a= x'f09f8c80'")
+	c.Assert(err, NotNil)
+	c.Assert(terror.ErrorEqual(err, table.ErrTruncatedWrongValueForField), IsTrue, Commentf("err %v", err))
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(10) CHARSET utf8 COLLATE utf8_bin DEFAULT NULL,\n" +
+		"  `b` varchar(10) CHARSET ascii COLLATE ascii_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+
+	// Mock old version table info with column charset is utf8.
+	db, ok := domain.GetDomain(tk.Se).InfoSchema().SchemaByName(model.NewCIStr("test"))
+	tbl := testGetTableByName(c, tk.Se, "test", "t")
+	tblInfo := tbl.Meta().Clone()
+	tblInfo.Version = model.TableInfoVersion0
+	tblInfo.Columns[0].Version = model.ColumnInfoVersion0
+	updateTableInfo := func(tblInfo *model.TableInfo) {
+		mockCtx := mock.NewContext()
+		mockCtx.Store = s.store
+		err := mockCtx.NewTxn()
+		c.Assert(err, IsNil)
+		txn, err := mockCtx.Txn(true)
+		c.Assert(err, IsNil)
+		mt := meta.NewMeta(txn)
+		c.Assert(ok, IsTrue)
+		err = mt.UpdateTable(db.ID, tblInfo)
+		c.Assert(err, IsNil)
+		err = txn.Commit(context.Background())
+		c.Assert(err, IsNil)
+	}
+	updateTableInfo(tblInfo)
+	tk.MustExec("alter table t add column c varchar(10) character set utf8;") // load latest schema.
+	c.Assert(config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4, IsTrue)
+	tk.MustExec("insert into t set a= x'f09f8c80'")
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(10) DEFAULT NULL,\n" +
+		"  `b` varchar(10) CHARSET ascii COLLATE ascii_bin DEFAULT NULL,\n" +
+		"  `c` varchar(10) CHARSET utf8 COLLATE utf8_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+
+	config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 = false
+	tk.MustExec("alter table t drop column c;") //  reload schema.
+	_, err = tk.Exec("insert into t set a= x'f09f8c80'")
+	c.Assert(err, NotNil)
+	c.Assert(terror.ErrorEqual(err, table.ErrTruncatedWrongValueForField), IsTrue, Commentf("err %v", err))
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(10) CHARSET utf8 COLLATE utf8_bin DEFAULT NULL,\n" +
+		"  `b` varchar(10) CHARSET ascii COLLATE ascii_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+
+	// Mock old version table info with table and column charset is utf8.
+	tbl = testGetTableByName(c, tk.Se, "test", "t")
+	tblInfo = tbl.Meta().Clone()
+	tblInfo.Charset = charset.CharsetUTF8
+	tblInfo.Collate = charset.CollationUTF8
+	tblInfo.Version = model.TableInfoVersion0
+	tblInfo.Columns[0].Version = model.ColumnInfoVersion0
+	updateTableInfo(tblInfo)
+
+	config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 = true
+	tk.MustExec("alter table t add column c varchar(10);") //  load latest schema.
+	tk.MustExec("insert into t set a= x'f09f8c80'")
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(10) DEFAULT NULL,\n" +
+		"  `b` varchar(10) CHARSET ascii COLLATE ascii_bin DEFAULT NULL,\n" +
+		"  `c` varchar(10) DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+
+	config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 = false
+	tk.MustExec("alter table t drop column c;") //  reload schema.
+	_, err = tk.Exec("insert into t set a= x'f09f8c80'")
+	c.Assert(err, NotNil)
+	c.Assert(terror.ErrorEqual(err, table.ErrTruncatedWrongValueForField), IsTrue, Commentf("err %v", err))
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(10) DEFAULT NULL,\n" +
+		"  `b` varchar(10) CHARSET ascii COLLATE ascii_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin"))
+
+	// Test modify column charset.
+	config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 = true
+	tk.MustExec("alter table t modify column a varchar(10) character set utf8mb4") //  change column charset.
+	tbl = testGetTableByName(c, tk.Se, "test", "t")
+	c.Assert(tbl.Meta().Columns[0].Charset, Equals, charset.CharsetUTF8MB4)
+	c.Assert(tbl.Meta().Columns[0].Collate, Equals, charset.CollationUTF8MB4)
+	c.Assert(tbl.Meta().Columns[0].Version, Equals, model.ColumnInfoVersion0)
+	tk.MustExec("insert into t set a= x'f09f8c80'")
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(10) DEFAULT NULL,\n" +
+		"  `b` varchar(10) CHARSET ascii COLLATE ascii_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+	// Test for change column should not modify the column version.
+	tk.MustExec("alter table t change column a a varchar(20)") //  change column.
+	tbl = testGetTableByName(c, tk.Se, "test", "t")
+	c.Assert(tbl.Meta().Columns[0].Charset, Equals, charset.CharsetUTF8MB4)
+	c.Assert(tbl.Meta().Columns[0].Collate, Equals, charset.CollationUTF8MB4)
+	c.Assert(tbl.Meta().Columns[0].Version, Equals, model.ColumnInfoVersion0)
+
+	// Test for v2.1.5 and v2.1.6 that table version is 1 but column version is 0.
+	tbl = testGetTableByName(c, tk.Se, "test", "t")
+	tblInfo = tbl.Meta().Clone()
+	tblInfo.Charset = charset.CharsetUTF8
+	tblInfo.Collate = charset.CollationUTF8
+	tblInfo.Version = model.TableInfoVersion1
+	tblInfo.Columns[0].Version = model.ColumnInfoVersion0
+	tblInfo.Columns[0].Charset = charset.CharsetUTF8
+	tblInfo.Columns[0].Collate = charset.CollationUTF8
+	updateTableInfo(tblInfo)
+	c.Assert(config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4, IsTrue)
+	tk.MustExec("alter table t change column b b varchar(20) character set ascii") // reload schema.
+	tk.MustExec("insert into t set a= x'f09f8c80'")
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(20) DEFAULT NULL,\n" +
+		"  `b` varchar(20) CHARSET ascii COLLATE ascii_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+
+	config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 = false
+	tk.MustExec("alter table t change column b b varchar(30) character set ascii") // reload schema.
+	_, err = tk.Exec("insert into t set a= x'f09f8c80'")
+	c.Assert(err, NotNil)
+	c.Assert(terror.ErrorEqual(err, table.ErrTruncatedWrongValueForField), IsTrue, Commentf("err %v", err))
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(20) DEFAULT NULL,\n" +
+		"  `b` varchar(30) CHARSET ascii COLLATE ascii_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin"))
+
+	// Test for alter table convert charset
+	config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 = true
+	tk.MustExec("alter table t change column b b varchar(40) character set ascii") // reload schema.
+	tk.MustExec("alter table t convert to charset utf8mb4;")
+
+	config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 = false
+	tk.MustExec("alter table t change column b b varchar(50) CHARSET ascii") // reload schema.
+	// TODO: fix  this after PR 9790.
+	tk.MustQuery("show create table t").Check(testkit.Rows("t CREATE TABLE `t` (\n" +
+		"  `a` varchar(20) CHARSET utf8 COLLATE utf8_bin DEFAULT NULL,\n" +
+		"  `b` varchar(50) CHARSET ascii COLLATE ascii_bin DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+}

ddl/db_test.go

@@ -304,7 +304,11 @@ func (s *testDBSuite) TestRenameIndex(c *C) {
 }
 
 func (s *testDBSuite) testGetTableByName(c *C, db, table string) table.Table {
-	ctx := s.s.(sessionctx.Context)
+	return testGetTableByName(c, s.s, db, table)
+}
+
+func testGetTableByName(c *C, se sessionctx.Context, db, table string) table.Table {
+	ctx := se.(sessionctx.Context)
 	dom := domain.GetDomain(ctx)
 	// Make sure the table schema is the new schema.
 	err := dom.Reload()

ddl/ddl_api.go

@@ -331,10 +331,6 @@ func convertTimestampDefaultValToUTC(ctx sessionctx.Context, defaultVal interfac
 				return defaultVal, errors.Trace(err)
 			}
 			defaultVal = t.String()
-			// Version = 1: For OriginDefaultValue and DefaultValue of timestamp column will stores the default time in UTC time zone.
-			//              This will fix bug in version 0.
-			// TODO: remove this version field after there is no old version 0.
-			col.Version = model.ColumnInfoVersion1
 		}
 	}
 	return defaultVal, nil
@@ -356,6 +352,8 @@ func columnDefToCol(ctx sessionctx.Context, offset int, colDef *ast.ColumnDef, o
 		Offset:    offset,
 		Name:      colDef.Name.Name,
 		FieldType: *colDef.Tp,
+		// TODO: remove this version field after there is no old version.
+		Version: model.CurrLatestColumnInfoVersion,
 	})
 
 	if !isExplicitTimeStamp() {
@@ -1866,6 +1864,7 @@ func (d *ddl) getModifiableColumnJob(ctx sessionctx.Context, ident ast.Ident, or
 		OriginDefaultValue: col.OriginDefaultValue,
 		FieldType:          *specNewColumn.Tp,
 		Name:               newColName,
+		Version:            col.Version,
 	})
 
 	// TODO: Remove it when all table versions are greater than or equal to TableInfoVersion1.
@@ -2082,8 +2081,9 @@ func (d *ddl) AlterTableCharsetAndCollate(ctx sessionctx.Context, ident ast.Iden
 			return errors.Trace(err)
 		}
 	}
-
-	if origCharset == toCharset && origCollate == toCollate {
+	// Old version schema charset maybe modified when load schema if TreatOldVersionUTF8AsUTF8MB4 was enable.
+	// So even if the origCharset equal toCharset, we still need to do the ddl for old version schema.
+	if origCharset == toCharset && origCollate == toCollate && tb.Meta().Version >= model.TableInfoVersion2 {
 		// nothing to do.
 		return nil
 	}

domain/domain.go

@@ -175,6 +175,12 @@ func (do *Domain) fetchSchemasWithTables(schemas []*model.DBInfo, m *meta.Meta,
 			done <- err
 			return
 		}
+		// If TreatOldVersionUTF8AsUTF8MB4 was enable, need to convert the old version schema UTF8 charset to UTF8MB4.
+		if config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 {
+			for _, tbInfo := range tables {
+				infoschema.ConvertOldVersionUTF8ToUTF8MB4IfNeed(tbInfo)
+			}
+		}
 		di.Tables = make([]*model.TableInfo, 0, len(tables))
 		for _, tbl := range tables {
 			if tbl.State != model.StatePublic {

executor/statement_context_test.go

@@ -102,9 +102,9 @@ func (s *testSuite) TestStatementContext(c *C) {
 	_, err = tk.Exec("insert t1 values (unhex('F0A48BAE'))")
 	c.Assert(err, NotNil)
 	c.Assert(terror.ErrorEqual(err, table.ErrTruncateWrongValue), IsTrue, Commentf("err %v", err))
-	config.GetGlobalConfig().CheckMb4ValueInUtf8 = false
+	config.GetGlobalConfig().CheckMb4ValueInUTF8 = false
 	tk.MustExec("insert t1 values (unhex('f09f8c80'))")
-	config.GetGlobalConfig().CheckMb4ValueInUtf8 = true
+	config.GetGlobalConfig().CheckMb4ValueInUTF8 = true
 	_, err = tk.Exec("insert t1 values (unhex('F0A48BAE'))")
 	c.Assert(err, NotNil)
 }

go.mod

@@ -48,7 +48,7 @@ require (
 	github.com/pingcap/gofail v0.0.0-20181217135706-6a951c1e42c3
 	github.com/pingcap/goleveldb v0.0.0-20171020084629-8d44bfdf1030
 	github.com/pingcap/kvproto v0.0.0-20190226063853-f6c0b7ffff11
-	github.com/pingcap/parser v0.0.0-20190326030543-750ee1d923e5
+	github.com/pingcap/parser v0.0.0-20190326083041-06045580997a
 	github.com/pingcap/pd v2.1.0-rc.4+incompatible
 	github.com/pingcap/tidb-tools v2.1.3-0.20190116051332-34c808eef588+incompatible
 	github.com/pingcap/tipb v0.0.0-20180910045846-371b48b15d93

go.sum

@@ -92,8 +92,8 @@ github.com/pingcap/goleveldb v0.0.0-20171020084629-8d44bfdf1030 h1:XJLuW0lsP7vAt
 github.com/pingcap/goleveldb v0.0.0-20171020084629-8d44bfdf1030/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw=
 github.com/pingcap/kvproto v0.0.0-20190226063853-f6c0b7ffff11 h1:iGNfAHgK0VHJobW4bPTlFmdnt3YWsEHdSTIcjut6ffk=
 github.com/pingcap/kvproto v0.0.0-20190226063853-f6c0b7ffff11/go.mod h1:0gwbe1F2iBIjuQ9AH0DbQhL+Dpr5GofU8fgYyXk+ykk=
-github.com/pingcap/parser v0.0.0-20190326030543-750ee1d923e5 h1:a7/kE/0gRUzO9ZpmantcRbTILozM8O5a1FCBg43eUQk=
-github.com/pingcap/parser v0.0.0-20190326030543-750ee1d923e5/go.mod h1:1FNvfp9+J0wvc4kl8eGNh7Rqrxveg15jJoWo/a0uHwA=
+github.com/pingcap/parser v0.0.0-20190326083041-06045580997a h1:4trGl3lvogZnelwJSF8ELqeS/5+RUVZte84bVmKT8+w=
+github.com/pingcap/parser v0.0.0-20190326083041-06045580997a/go.mod h1:1FNvfp9+J0wvc4kl8eGNh7Rqrxveg15jJoWo/a0uHwA=
 github.com/pingcap/pd v2.1.0-rc.4+incompatible h1:/buwGk04aHO5odk/+O8ZOXGs4qkUjYTJ2UpCJXna8NE=
 github.com/pingcap/pd v2.1.0-rc.4+incompatible/go.mod h1:nD3+EoYes4+aNNODO99ES59V83MZSI+dFbhyr667a0E=
 github.com/pingcap/tidb-tools v2.1.3-0.20190116051332-34c808eef588+incompatible h1:e9Gi/LP9181HT3gBfSOeSBA+5JfemuE4aEAhqNgoE4k=

infoschema/builder.go

@@ -18,7 +18,9 @@ import (
 	"sort"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/charset"
 	"github.com/pingcap/parser/model"
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/meta/autoid"
 	"github.com/pingcap/tidb/perfschema"
@@ -171,6 +173,8 @@ func (b *Builder) applyCreateTable(m *meta.Meta, dbInfo *model.DBInfo, tableID i
 			fmt.Sprintf("(Table ID %d)", tableID),
 		)
 	}
+	ConvertOldVersionUTF8ToUTF8MB4IfNeed(tblInfo)
+
 	if alloc == nil {
 		schemaID := dbInfo.ID
 		alloc = autoid.NewAllocator(b.handle.store, tblInfo.GetDBID(schemaID), tblInfo.IsAutoIncColUnsigned())
@@ -194,6 +198,23 @@ func (b *Builder) applyCreateTable(m *meta.Meta, dbInfo *model.DBInfo, tableID i
 	return nil
 }
 
+// ConvertOldVersionUTF8ToUTF8MB4IfNeed convert old version UTF8 to UTF8MB4 if config.TreatOldVersionUTF8AsUTF8MB4 is enable.
+func ConvertOldVersionUTF8ToUTF8MB4IfNeed(tbInfo *model.TableInfo) {
+	if !config.GetGlobalConfig().TreatOldVersionUTF8AsUTF8MB4 || tbInfo.Version >= model.TableInfoVersion2 {
+		return
+	}
+	if tbInfo.Charset == charset.CharsetUTF8 {
+		tbInfo.Charset = charset.CharsetUTF8MB4
+		tbInfo.Collate = charset.CollationUTF8MB4
+	}
+	for _, col := range tbInfo.Columns {
+		if col.Version < model.ColumnInfoVersion2 && col.Charset == charset.CharsetUTF8 {
+			col.Charset = charset.CharsetUTF8MB4
+			col.Collate = charset.CollationUTF8MB4
+		}
+	}
+}
+
 func (b *Builder) applyDropTable(dbInfo *model.DBInfo, tableID int64) {
 	bucketIdx := tableBucketIdx(tableID)
 	sortedTbls := b.is.sortedTablesBuckets[bucketIdx]