diff --git a/dm/pkg/binlog/event/util.go b/dm/pkg/binlog/event/util.go
index dd660e285fd..3d902ea861f 100644
--- a/dm/pkg/binlog/event/util.go
+++ b/dm/pkg/binlog/event/util.go
@@ -356,7 +356,7 @@ func GetParserForStatusVars(statusVars []byte) (*parser.Parser, error) {
 }
 
 // GetServerCollationByStatusVars gets server collation by binlog statusVars.
-func GetServerCollationByStatusVars(statusVars []byte) (string, error) {
+func GetServerCollationByStatusVars(statusVars []byte, idAndCollationMap map[int]string) (string, error) {
 	vars, err := statusVarsToKV(statusVars)
 	b, ok := vars[QCharsetCode]
 
@@ -365,12 +365,15 @@ func GetServerCollationByStatusVars(statusVars []byte) (string, error) {
 			// only happen when this is a dummy event generated by DM
 			err = fmt.Errorf("Q_CHARSET_CODE not found in status_vars %v", statusVars)
 		}
-		// mysql default 'latin1_swedish_ci'
+		// mysql 5.7.22 default 'latin1_swedish_ci'
 		return "latin1_swedish_ci", err
 	}
+
 	// QCharsetCode 2-byte character_set_client + 2-byte collation_connection + 2-byte collation_server
-	// collation is less than 255 and we use the first byte.
-	return mysql.Collations[b[4]], err
+	r := bytes.NewReader(b[4:])
+	var v uint16
+	_ = binary.Read(r, binary.LittleEndian, &v)
+	return idAndCollationMap[int(v)], err
 }
 
 // if returned error is `io.EOF`, it means UnexpectedEOF because we handled expected `io.EOF` as success
diff --git a/dm/pkg/utils/db.go b/dm/pkg/utils/db.go
index fa9d6337d5d..06bc1c6da97 100644
--- a/dm/pkg/utils/db.go
+++ b/dm/pkg/utils/db.go
@@ -391,51 +391,6 @@ func GetServerUnixTS(ctx context.Context, db *sql.DB) (int64, error) {
 	return ts, err
 }
 
-// GetCharsetAndDefaultCollation gets charset and default collation map.
-func GetCharsetAndDefaultCollation(ctx context.Context, db *sql.DB) (map[string]string, error) {
-	charsetAndDefaultCollation := make(map[string]string)
-	conn, err := db.Conn(ctx)
-	if err != nil {
-		return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-	}
-	defer conn.Close()
-
-	// Show an example.
-	/*
-		mysql> SHOW CHARACTER SET;
-		+----------+---------------------------------+---------------------+--------+
-		| Charset  | Description                     | Default collation   | Maxlen |
-		+----------+---------------------------------+---------------------+--------+
-		| armscii8 | ARMSCII-8 Armenian              | armscii8_general_ci |      1 |
-		| ascii    | US ASCII                        | ascii_general_ci    |      1 |
-		| big5     | Big5 Traditional Chinese        | big5_chinese_ci     |      2 |
-		| binary   | Binary pseudo charset           | binary              |      1 |
-		| cp1250   | Windows Central European        | cp1250_general_ci   |      1 |
-		| cp1251   | Windows Cyrillic                | cp1251_general_ci   |      1 |
-		+----------+---------------------------------+---------------------+--------+
-	*/
-
-	rows, err := conn.QueryContext(ctx, "SHOW CHARACTER SET")
-	if err != nil {
-		return nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
-	}
-
-	defer rows.Close()
-	for rows.Next() {
-		var charset, description, collation string
-		var maxlen int
-		if scanErr := rows.Scan(&charset, &description, &collation, &maxlen); scanErr != nil {
-			return nil, terror.DBErrorAdapt(scanErr, terror.ErrDBDriverError)
-		}
-		charsetAndDefaultCollation[strings.ToLower(charset)] = collation
-	}
-
-	if err = rows.Close(); err != nil {
-		return nil, terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
-	}
-	return charsetAndDefaultCollation, err
-}
-
 // GetSchemaList gets db schema list with `SHOW DATABASES`.
 func GetSchemaList(ctx context.Context, db *sql.DB) ([]string, error) {
 	schemaList := []string{}
diff --git a/dm/syncer/dbconn/upstream_db.go b/dm/syncer/dbconn/upstream_db.go
index ccbba8d6c4c..8577605d542 100644
--- a/dm/syncer/dbconn/upstream_db.go
+++ b/dm/syncer/dbconn/upstream_db.go
@@ -15,6 +15,7 @@ package dbconn
 
 import (
 	"context"
+	"strings"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/pingcap/failpoint"
@@ -71,9 +72,49 @@ func (conn *UpStreamConn) GetServerUnixTS(ctx context.Context) (int64, error) {
 	return utils.GetServerUnixTS(ctx, conn.BaseDB.DB)
 }
 
-// GetCharsetAndDefaultCollation returns charset and default collation map.
-func (conn *UpStreamConn) GetCharsetAndDefaultCollation(ctx context.Context) (map[string]string, error) {
-	return utils.GetCharsetAndDefaultCollation(ctx, conn.BaseDB.DB)
+// GetCharsetAndDefaultCollation returns charset and collation info.
+func GetCharsetAndCollationInfo(tctx *tcontext.Context, conn *DBConn) (map[string]string, map[int]string, error) {
+	charsetAndDefaultCollation := make(map[string]string)
+	idAndCollationMap := make(map[int]string)
+
+	// Show an example.
+	/*
+		mysql> SELECT COLLATION_NAME,CHARACTER_SET_NAME,ID,IS_DEFAULT from INFORMATION_SCHEMA.COLLATIONS;
+		+----------------------------+--------------------+-----+------------+
+		| COLLATION_NAME             | CHARACTER_SET_NAME | ID  | IS_DEFAULT |
+		+----------------------------+--------------------+-----+------------+
+		| armscii8_general_ci        | armscii8           |  32 | Yes        |
+		| armscii8_bin               | armscii8           |  64 |            |
+		| ascii_general_ci           | ascii              |  11 | Yes        |
+		| ascii_bin                  | ascii              |  65 |            |
+		| big5_chinese_ci            | big5               |   1 | Yes        |
+		| big5_bin                   | big5               |  84 |            |
+		| binary                     | binary             |  63 | Yes        |
+		+----------------------------+--------------------+-----+------------+
+	*/
+
+	rows, err := conn.QuerySQL(tctx, "SELECT COLLATION_NAME,CHARACTER_SET_NAME,ID,IS_DEFAULT from INFORMATION_SCHEMA.COLLATIONS")
+	if err != nil {
+		return nil, nil, terror.DBErrorAdapt(err, terror.ErrDBDriverError)
+	}
+
+	defer rows.Close()
+	for rows.Next() {
+		var collation, charset, isDefault string
+		var id int
+		if scanErr := rows.Scan(&collation, &charset, &id, &isDefault); scanErr != nil {
+			return nil, nil, terror.DBErrorAdapt(scanErr, terror.ErrDBDriverError)
+		}
+		idAndCollationMap[id] = strings.ToLower(collation)
+		if strings.ToLower(isDefault) == "yes" {
+			charsetAndDefaultCollation[strings.ToLower(charset)] = collation
+		}
+	}
+
+	if err = rows.Close(); err != nil {
+		return nil, nil, terror.DBErrorAdapt(rows.Err(), terror.ErrDBDriverError)
+	}
+	return charsetAndDefaultCollation, idAndCollationMap, err
 }
 
 // GetParser returns the parser with correct flag for upstream.
diff --git a/dm/syncer/ddl.go b/dm/syncer/ddl.go
index a55ef451c67..ddcf83f7cd4 100644
--- a/dm/syncer/ddl.go
+++ b/dm/syncer/ddl.go
@@ -127,7 +127,7 @@ func (s *Syncer) genDDLInfo(qec *queryEventContext, sql string) (*ddlInfo, error
 		targetTables: targetTables,
 	}
 
-	adjustCollation(s.tctx, ddlInfo, qec.eventStatusVars, s.charsetAndDefaultCollation)
+	adjustCollation(s.tctx, ddlInfo, qec.eventStatusVars, s.charsetAndDefaultCollation, s.idAndCollationMap)
 	routedDDL, err := parserpkg.RenameDDLTable(ddlInfo.originStmt, ddlInfo.targetTables)
 	ddlInfo.routedDDL = routedDDL
 	return ddlInfo, err
@@ -196,12 +196,13 @@ func (s *Syncer) clearOnlineDDL(tctx *tcontext.Context, targetTable *filter.Tabl
 }
 
 // adjustCollation adds collation for create database and check create table.
-func adjustCollation(tctx *tcontext.Context, ddlInfo *ddlInfo, statusVars []byte, charsetAndDefaultCollationMap map[string]string) {
+func adjustCollation(tctx *tcontext.Context, ddlInfo *ddlInfo, statusVars []byte, charsetAndDefaultCollationMap map[string]string, idAndCollationMap map[int]string) {
 	switch createStmt := ddlInfo.originStmt.(type) {
 	case *ast.CreateTableStmt:
 		if createStmt.ReferTable != nil {
 			return
 		}
+		adjustColumnsCollation(tctx, createStmt, charsetAndDefaultCollationMap)
 		var justCharset string
 		for _, tableOption := range createStmt.Options {
 			// already have 'Collation'
@@ -222,12 +223,13 @@ func adjustCollation(tctx *tcontext.Context, ddlInfo *ddlInfo, statusVars []byte
 			tctx.L().Warn("not found charset default collation.", zap.String("originSQL", ddlInfo.originDDL), zap.String("charset", strings.ToLower(justCharset)))
 			return
 		}
-		tctx.L().Info("detect create table risk which use explicit charset and implicit collation, we will add collation by SHOW CHARACTER SET", zap.String("originSQL", ddlInfo.originDDL), zap.String("collation", collation))
+		tctx.L().Info("detect create table risk which use explicit charset and implicit collation, we will add collation by INFORMATION_SCHEMA.COLLATIONS", zap.String("originSQL", ddlInfo.originDDL), zap.String("collation", collation))
 		createStmt.Options = append(createStmt.Options, &ast.TableOption{Tp: ast.TableOptionCollate, StrValue: collation})
 
 	case *ast.CreateDatabaseStmt:
 		var justCharset, collation string
 		var ok bool
+		var err error
 		for _, createOption := range createStmt.Options {
 			// already have 'Collation'
 			if createOption.Tp == ast.DatabaseOptionCollate {
@@ -245,11 +247,18 @@ func adjustCollation(tctx *tcontext.Context, ddlInfo *ddlInfo, statusVars []byte
 				tctx.L().Warn("not found charset default collation.", zap.String("originSQL", ddlInfo.originDDL), zap.String("charset", strings.ToLower(justCharset)))
 				return
 			}
-			tctx.L().Info("detect create database risk which use explicit charset and implicit collation, we will add collation by SHOW CHARACTER SET", zap.String("originSQL", ddlInfo.originDDL), zap.String("collation", collation))
+			tctx.L().Info("detect create database risk which use explicit charset and implicit collation, we will add collation by INFORMATION_SCHEMA.COLLATIONS", zap.String("originSQL", ddlInfo.originDDL), zap.String("collation", collation))
 		} else {
 			// has no charset and collation
 			// add collation by server collation from binlog statusVars
-			collation, _ = event.GetServerCollationByStatusVars(statusVars)
+			collation, err = event.GetServerCollationByStatusVars(statusVars, idAndCollationMap)
+			if err != nil {
+				tctx.L().Error("can not get charset server collation from binlog statusVars.", zap.Error(err), zap.String("originSQL", ddlInfo.originDDL))
+			}
+			if collation == "" {
+				tctx.L().Error("get server collation from binlog statusVars is nil.", zap.Error(err), zap.String("originSQL", ddlInfo.originDDL))
+				return
+			}
 			// add collation
 			tctx.L().Info("detect create database risk which use implicit charset and collation, we will add collation by binlog status_vars", zap.String("originSQL", ddlInfo.originDDL), zap.String("collation", collation))
 		}
@@ -257,6 +266,32 @@ func adjustCollation(tctx *tcontext.Context, ddlInfo *ddlInfo, statusVars []byte
 	}
 }
 
+// adjustColumnsCollation adds column's collation.
+func adjustColumnsCollation(tctx *tcontext.Context, createStmt *ast.CreateTableStmt, charsetAndDefaultCollationMap map[string]string) {
+	for _, col := range createStmt.Cols {
+		for _, options := range col.Options {
+			// already have 'Collation'
+			if options.Tp == ast.ColumnOptionCollate {
+				continue
+			}
+		}
+		fieldType := col.Tp
+		// already have 'Collation'
+		if fieldType.Collate != "" {
+			continue
+		}
+		if fieldType.Charset != "" {
+			// just have charset
+			collation, ok := charsetAndDefaultCollationMap[strings.ToLower(fieldType.Charset)]
+			if !ok {
+				tctx.L().Warn("not found charset default collation for column.", zap.String("table", createStmt.Table.Name.String()), zap.String("column", col.Name.String()), zap.String("charset", strings.ToLower(fieldType.Charset)))
+				continue
+			}
+			col.Options = append(col.Options, &ast.ColumnOption{Tp: ast.ColumnOptionCollate, StrValue: collation})
+		}
+	}
+}
+
 type ddlInfo struct {
 	originDDL    string
 	routedDDL    string
diff --git a/dm/syncer/ddl_test.go b/dm/syncer/ddl_test.go
index 5cbd0986a5d..088f690c3a8 100644
--- a/dm/syncer/ddl_test.go
+++ b/dm/syncer/ddl_test.go
@@ -234,6 +234,7 @@ func (s *testDDLSuite) TestResolveDDLSQL(c *C) {
 		tctx: tctx,
 	}
 	statusVars := []byte{4, 0, 0, 0, 0, 46, 0}
+	syncer.idAndCollationMap = map[int]string{46: "utf8mb4_bin"}
 	for i, sql := range sqls {
 		qec := &queryEventContext{
 			eventContext:    ec,
@@ -638,17 +639,89 @@ func (s *testDDLSuite) TestClearOnlineDDL(c *C) {
 	c.Assert(mock.toFinish, HasLen, 0)
 }
 
+func (s *testDDLSuite) TestAdjustDatabaseCollation(c *C) {
+	statusVarsArray := [][]byte{
+		{
+			4, 0, 0, 0, 0, 46, 0,
+		},
+		{
+			4, 0, 0, 0, 0, 21, 1,
+		},
+	}
+
+	sqls := []string{
+		"create database if not exists `test`",
+		"create database `test` CHARACTER SET=utf8mb4 COLLATE=utf8mb4_general_ci",
+		"create database `test` CHARACTER SET=utf8mb4",
+		"create database `test` COLLATE=utf8mb4_general_ci",
+	}
+
+	expectedSQLs := [][]string{
+		{
+			"CREATE DATABASE IF NOT EXISTS `test` COLLATE = utf8mb4_bin",
+			"CREATE DATABASE `test` CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci",
+			"CREATE DATABASE `test` CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci",
+			"CREATE DATABASE `test` COLLATE = utf8mb4_general_ci",
+		},
+		{
+			"CREATE DATABASE IF NOT EXISTS `test` COLLATE = utf8mb4_vi_0900_ai_ci",
+			"CREATE DATABASE `test` CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci",
+			"CREATE DATABASE `test` CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci",
+			"CREATE DATABASE `test` COLLATE = utf8mb4_general_ci",
+		},
+	}
+
+	tctx := tcontext.Background().WithLogger(log.With(zap.String("test", "TestAdjustTableCollation")))
+	syncer := NewSyncer(&config.SubTaskConfig{}, nil, nil)
+	syncer.tctx = tctx
+	p := parser.New()
+	tab := &filter.Table{
+		Schema: "test",
+		Name:   "t",
+	}
+
+	charsetAndDefaultCollationMap := map[string]string{"utf8mb4": "utf8mb4_general_ci"}
+	idAndCollationMap := map[int]string{46: "utf8mb4_bin", 277: "utf8mb4_vi_0900_ai_ci"}
+	for i, statusVars := range statusVarsArray {
+		for j, sql := range sqls {
+			ddlInfo := &ddlInfo{
+				originDDL:    sql,
+				routedDDL:    sql,
+				sourceTables: []*filter.Table{tab},
+				targetTables: []*filter.Table{tab},
+			}
+			stmt, err := p.ParseOneStmt(sql, "", "")
+			c.Assert(err, IsNil)
+			c.Assert(stmt, NotNil)
+			ddlInfo.originStmt = stmt
+			adjustCollation(tctx, ddlInfo, statusVars, charsetAndDefaultCollationMap, idAndCollationMap)
+			routedDDL, err := parserpkg.RenameDDLTable(ddlInfo.originStmt, ddlInfo.targetTables)
+			c.Assert(err, IsNil)
+			c.Assert(routedDDL, Equals, expectedSQLs[i][j])
+		}
+	}
+}
+
 func (s *testDDLSuite) TestAdjustCollation(c *C) {
-	// duplicate with pkg/parser
 	sqls := []string{
 		"create table `test`.`t1` (id int) CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci",
 		"create table `test`.`t1` (id int) CHARSET=utf8mb4",
 		"create table `test`.`t1` (id int) COLLATE=utf8mb4_general_ci",
 		"create table `test`.`t1` (id int)",
-		"create database `test` CHARACTER SET=utf8mb4 COLLATE=utf8mb4_general_ci",
-		"create database `test` CHARACTER SET=utf8mb4",
-		"create database `test` COLLATE=utf8mb4_general_ci",
-		"create database if not exists `test`",
+		"create table `test`.`t1` (id int, name varchar(20) CHARACTER SET utf8mb4, work varchar(20))",
+		"create table `test`.`t1` (id int, name varchar(20), work varchar(20))",
+		"create table `test`.`t1` (id int, name varchar(20) COLLATE utf8mb4_general_ci, work varchar(20))",
+		"create table `test`.`t1` (id int, name varchar(20) COLLATE utf8mb4_general_ci, work varchar(20) CHARACTER SET utf8mb4)",
+		"create table `test`.`t1` (id int, name varchar(20) CHARACTER SET utf8mb4, work varchar(20)) COLLATE=utf8mb4_general_ci",
+		"create table `test`.`t1` (id int, name varchar(20) CHARACTER SET utf8mb4, work varchar(20)) COLLATE=latin1_swedish_ci",
+		"create table `test`.`t1` (id int, name varchar(20), work varchar(20)) COLLATE=utf8mb4_general_ci",
+		"create table `test`.`t1` (id int, name varchar(20) COLLATE utf8mb4_general_ci, work varchar(20)) COLLATE=utf8mb4_general_ci",
+		"create table `test`.`t1` (id int, name varchar(20) COLLATE utf8mb4_general_ci, work varchar(20) CHARACTER SET utf8mb4) COLLATE=utf8mb4_general_ci",
+		"create table `test`.`t1` (id int, name varchar(20) CHARACTER SET utf8mb4, work varchar(20)) CHARSET=utf8mb4 ",
+		"create table `test`.`t1` (id int, name varchar(20) CHARACTER SET latin1, work varchar(20)) CHARSET=utf8mb4 ",
+		"create table `test`.`t1` (id int, name varchar(20), work varchar(20)) CHARSET=utf8mb4",
+		"create table `test`.`t1` (id int, name varchar(20) COLLATE utf8mb4_general_ci, work varchar(20)) CHARSET=utf8mb4",
+		"create table `test`.`t1` (id int, name varchar(20) COLLATE utf8mb4_general_ci, work varchar(20) CHARACTER SET utf8mb4) CHARSET=utf8mb4",
 	}
 
 	expectedSQLs := []string{
@@ -656,10 +729,20 @@ func (s *testDDLSuite) TestAdjustCollation(c *C) {
 		"CREATE TABLE `test`.`t` (`id` INT) DEFAULT CHARACTER SET = UTF8MB4 DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
 		"CREATE TABLE `test`.`t` (`id` INT) DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
 		"CREATE TABLE `test`.`t` (`id` INT)",
-		"CREATE DATABASE `test` CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci",
-		"CREATE DATABASE `test` CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci",
-		"CREATE DATABASE `test` COLLATE = utf8mb4_general_ci",
-		"CREATE DATABASE IF NOT EXISTS `test` COLLATE = utf8mb4_bin",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) CHARACTER SET UTF8MB4 COLLATE utf8mb4_general_ci,`work` VARCHAR(20))",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20),`work` VARCHAR(20))",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) COLLATE utf8mb4_general_ci,`work` VARCHAR(20))",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) COLLATE utf8mb4_general_ci,`work` VARCHAR(20) CHARACTER SET UTF8MB4 COLLATE utf8mb4_general_ci)",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) CHARACTER SET UTF8MB4 COLLATE utf8mb4_general_ci,`work` VARCHAR(20)) DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) CHARACTER SET UTF8MB4 COLLATE utf8mb4_general_ci,`work` VARCHAR(20)) DEFAULT COLLATE = LATIN1_SWEDISH_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20),`work` VARCHAR(20)) DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) COLLATE utf8mb4_general_ci,`work` VARCHAR(20)) DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) COLLATE utf8mb4_general_ci,`work` VARCHAR(20) CHARACTER SET UTF8MB4 COLLATE utf8mb4_general_ci) DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) CHARACTER SET UTF8MB4 COLLATE utf8mb4_general_ci,`work` VARCHAR(20)) DEFAULT CHARACTER SET = UTF8MB4 DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) CHARACTER SET LATIN1 COLLATE latin1_swedish_ci,`work` VARCHAR(20)) DEFAULT CHARACTER SET = UTF8MB4 DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20),`work` VARCHAR(20)) DEFAULT CHARACTER SET = UTF8MB4 DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) COLLATE utf8mb4_general_ci,`work` VARCHAR(20)) DEFAULT CHARACTER SET = UTF8MB4 DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
+		"CREATE TABLE `test`.`t` (`id` INT,`name` VARCHAR(20) COLLATE utf8mb4_general_ci,`work` VARCHAR(20) CHARACTER SET UTF8MB4 COLLATE utf8mb4_general_ci) DEFAULT CHARACTER SET = UTF8MB4 DEFAULT COLLATE = UTF8MB4_GENERAL_CI",
 	}
 
 	tctx := tcontext.Background().WithLogger(log.With(zap.String("test", "TestAdjustTableCollation")))
@@ -671,7 +754,8 @@ func (s *testDDLSuite) TestAdjustCollation(c *C) {
 		Name:   "t",
 	}
 	statusVars := []byte{4, 0, 0, 0, 0, 46, 0}
-	charsetAndDefaultCollationMap := map[string]string{"utf8mb4": "utf8mb4_general_ci"}
+	charsetAndDefaultCollationMap := map[string]string{"utf8mb4": "utf8mb4_general_ci", "latin1": "latin1_swedish_ci"}
+	idAndCollationMap := map[int]string{46: "utf8mb4_bin"}
 	for i, sql := range sqls {
 		ddlInfo := &ddlInfo{
 			originDDL:    sql,
@@ -683,7 +767,7 @@ func (s *testDDLSuite) TestAdjustCollation(c *C) {
 		c.Assert(err, IsNil)
 		c.Assert(stmt, NotNil)
 		ddlInfo.originStmt = stmt
-		adjustCollation(tctx, ddlInfo, statusVars, charsetAndDefaultCollationMap)
+		adjustCollation(tctx, ddlInfo, statusVars, charsetAndDefaultCollationMap, idAndCollationMap)
 		routedDDL, err := parserpkg.RenameDDLTable(ddlInfo.originStmt, ddlInfo.targetTables)
 		c.Assert(err, IsNil)
 		c.Assert(routedDDL, Equals, expectedSQLs[i])
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index 1979478d296..cb3afe758fa 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -141,7 +141,8 @@ type Syncer struct {
 
 	schemaTracker *schema.Tracker
 
-	fromDB *dbconn.UpStreamConn
+	fromDB   *dbconn.UpStreamConn
+	fromConn *dbconn.DBConn
 
 	toDB                *conn.BaseDB
 	toDBConns           []*dbconn.DBConn
@@ -231,6 +232,7 @@ type Syncer struct {
 
 	relay                      relay.Process
 	charsetAndDefaultCollation map[string]string
+	idAndCollationMap          map[int]string
 }
 
 // NewSyncer creates a new Syncer.
@@ -332,7 +334,7 @@ func (s *Syncer) Init(ctx context.Context) (err error) {
 		return terror.ErrSchemaTrackerInit.Delegate(err)
 	}
 
-	s.charsetAndDefaultCollation, err = s.fromDB.GetCharsetAndDefaultCollation(ctx)
+	s.charsetAndDefaultCollation, s.idAndCollationMap, err = dbconn.GetCharsetAndCollationInfo(tctx, s.fromConn)
 	if err != nil {
 		return err
 	}
@@ -3035,10 +3037,12 @@ func (s *Syncer) createDBs(ctx context.Context) error {
 	var err error
 	dbCfg := s.cfg.From
 	dbCfg.RawDBCfg = config.DefaultRawDBConfig().SetReadTimeout(maxDMLConnectionTimeout)
-	s.fromDB, err = dbconn.NewUpStreamConn(&dbCfg)
+	fromDB, fromConns, err := dbconn.CreateConns(s.tctx, s.cfg, &dbCfg, 1)
 	if err != nil {
 		return err
 	}
+	s.fromDB = &dbconn.UpStreamConn{BaseDB: fromDB}
+	s.fromConn = fromConns[0]
 	conn, err := s.fromDB.BaseDB.GetBaseConn(ctx)
 	if err != nil {
 		return err
diff --git a/dm/tests/others_integration_1.txt b/dm/tests/others_integration_1.txt
index 06230c842a0..2b538f41477 100644
--- a/dm/tests/others_integration_1.txt
+++ b/dm/tests/others_integration_1.txt
@@ -10,3 +10,4 @@ checkpoint_transaction
 lightning_mode
 downstream_diff_index
 slow_relay_writer
+sync_collation
diff --git a/dm/tests/sync_collation/conf/dm-master.toml b/dm/tests/sync_collation/conf/dm-master.toml
new file mode 100644
index 00000000000..7cecf59ad86
--- /dev/null
+++ b/dm/tests/sync_collation/conf/dm-master.toml
@@ -0,0 +1,4 @@
+# Master Configuration.
+master-addr = ":8261"
+advertise-addr = "127.0.0.1:8261"
+auto-compaction-retention = "3s"
diff --git a/dm/tests/sync_collation/conf/dm-task.yaml b/dm/tests/sync_collation/conf/dm-task.yaml
new file mode 100644
index 00000000000..cfdedf6b30c
--- /dev/null
+++ b/dm/tests/sync_collation/conf/dm-task.yaml
@@ -0,0 +1,45 @@
+---
+name: sync_collation
+task-mode: all
+is-sharding: false
+meta-schema: "dm_meta"
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  - source-id: "mysql-replica-01"
+    block-allow-list:  "instance"
+    mydumper-config-name: "global"
+    loader-config-name: "global"
+    syncer-config-name: "global"
+
+  - source-id: "mysql-replica-02"
+    block-allow-list:  "instance"
+    mydumper-config-name: "global"
+    loader-config-name: "global"
+    syncer-config-name: "global"
+
+block-allow-list:
+  instance:
+    do-dbs: ["sync_collation*"]
+
+mydumpers:
+  global:
+    threads: 4
+    chunk-filesize: 64
+    skip-tz-utc: true
+    extra-args: ""
+
+loaders:
+  global:
+    pool-size: 16
+    dir: "./dumped_data"
+
+syncers:
+  global:
+    worker-count: 16
+    batch: 100
diff --git a/dm/tests/sync_collation/conf/dm-worker1.toml b/dm/tests/sync_collation/conf/dm-worker1.toml
new file mode 100644
index 00000000000..7a72ea72bf8
--- /dev/null
+++ b/dm/tests/sync_collation/conf/dm-worker1.toml
@@ -0,0 +1,2 @@
+name = "worker1"
+join = "127.0.0.1:8261"
diff --git a/dm/tests/sync_collation/conf/dm-worker2.toml b/dm/tests/sync_collation/conf/dm-worker2.toml
new file mode 100644
index 00000000000..010e21c73eb
--- /dev/null
+++ b/dm/tests/sync_collation/conf/dm-worker2.toml
@@ -0,0 +1,2 @@
+name = "worker2"
+join = "127.0.0.1:8261"
diff --git a/dm/tests/sync_collation/conf/source1.yaml b/dm/tests/sync_collation/conf/source1.yaml
new file mode 100644
index 00000000000..679a2f4db7c
--- /dev/null
+++ b/dm/tests/sync_collation/conf/source1.yaml
@@ -0,0 +1,13 @@
+source-id: mysql-replica-01
+flavor: 'mysql'
+enable-gtid: true
+relay-binlog-name: ''
+relay-binlog-gtid: ''
+enable-relay: false
+from:
+  host: 127.0.0.1
+  user: root
+  password: /Q7B9DizNLLTTfiZHv9WoEAKamfpIUs=
+  port: 3306
+checker:
+  check-enable: false
diff --git a/dm/tests/sync_collation/conf/source2.yaml b/dm/tests/sync_collation/conf/source2.yaml
new file mode 100644
index 00000000000..1d1d87b9d0c
--- /dev/null
+++ b/dm/tests/sync_collation/conf/source2.yaml
@@ -0,0 +1,13 @@
+source-id: mysql-replica-02
+flavor: 'mysql'
+enable-gtid: true
+relay-binlog-name: ''
+relay-binlog-gtid: ''
+enable-relay: false
+from:
+  host: 127.0.0.1
+  user: root
+  password: /Q7B9DizNLLTTfiZHv9WoEAKamfpIUs=
+  port: 3307
+checker:
+  check-enable: false
diff --git a/dm/tests/sync_collation/data/clean_data.sql b/dm/tests/sync_collation/data/clean_data.sql
new file mode 100644
index 00000000000..2dcf97a00d3
--- /dev/null
+++ b/dm/tests/sync_collation/data/clean_data.sql
@@ -0,0 +1,6 @@
+drop database if exists `sync_collation`;
+drop database if exists `sync_collation2`;
+drop database if exists `sync_collation_increment`;
+drop database if exists `sync_collation_increment2`;
+drop database if exists `sync_collation_server`;
+drop database if exists `sync_collation_server2`;
diff --git a/dm/tests/sync_collation/data/db1.increment.sql b/dm/tests/sync_collation/data/db1.increment.sql
new file mode 100644
index 00000000000..dffe9f0ea94
--- /dev/null
+++ b/dm/tests/sync_collation/data/db1.increment.sql
@@ -0,0 +1,13 @@
+drop database if exists `sync_collation_increment`;
+create database `sync_collation_increment` character set utf8;
+use `sync_collation_increment`;
+create table t1 (id int, name varchar(20), primary key(`id`)) character set utf8;
+insert into t1 (id, name) values (1, 'Aa'), (2, 'aA');
+create table t2 (id int, name varchar(20) character set utf8, primary key(`id`)) character set latin1 collate latin1_bin;
+insert into t2 (id, name) values (1, 'Aa'), (2, 'aA');
+set collation_server = utf8_general_ci;
+drop database if exists `sync_collation_server`;
+create database `sync_collation_server`;
+use `sync_collation_server`;
+create table t1 (id int, name varchar(20), primary key(`id`));
+insert into t1 (id, name) values (1, 'Aa'), (2, 'aA');
\ No newline at end of file
diff --git a/dm/tests/sync_collation/data/db1.increment_err.sql b/dm/tests/sync_collation/data/db1.increment_err.sql
new file mode 100644
index 00000000000..b02217031ba
--- /dev/null
+++ b/dm/tests/sync_collation/data/db1.increment_err.sql
@@ -0,0 +1,3 @@
+set collation_server = latin1_swedish_ci;
+drop database if exists `sync_collation_increment`;
+create database `sync_collation_increment`;
\ No newline at end of file
diff --git a/dm/tests/sync_collation/data/db1.prepare.sql b/dm/tests/sync_collation/data/db1.prepare.sql
new file mode 100644
index 00000000000..4ce27c05f52
--- /dev/null
+++ b/dm/tests/sync_collation/data/db1.prepare.sql
@@ -0,0 +1,7 @@
+drop database if exists `sync_collation`;
+create database `sync_collation` character set utf8;
+use `sync_collation`;
+create table t1 (id int, name varchar(20), primary key(`id`)) character set utf8;
+insert into t1 (id, name) values (1, 'Aa'), (2, 'aA');
+create table t2 (id int, name varchar(20) character set utf8, primary key(`id`)) character set latin1 collate latin1_bin;
+insert into t2 (id, name) values (1, 'Aa'), (2, 'aA');
diff --git a/dm/tests/sync_collation/data/db1.prepare_err.sql b/dm/tests/sync_collation/data/db1.prepare_err.sql
new file mode 100644
index 00000000000..cddcc949396
--- /dev/null
+++ b/dm/tests/sync_collation/data/db1.prepare_err.sql
@@ -0,0 +1,4 @@
+set collation_server = latin1_swedish_ci;
+drop database if exists `sync_collation`;
+create database `sync_collation`;
+use `sync_collation`;
diff --git a/dm/tests/sync_collation/data/db2.increment.sql b/dm/tests/sync_collation/data/db2.increment.sql
new file mode 100644
index 00000000000..d31ff16947c
--- /dev/null
+++ b/dm/tests/sync_collation/data/db2.increment.sql
@@ -0,0 +1,13 @@
+drop database if exists `sync_collation_increment2`;
+create database `sync_collation_increment2` character set utf8;
+use `sync_collation_increment2`;
+create table t1 (id int, name varchar(20), primary key(`id`)) character set utf8;
+insert into t1 (id, name) values (1, 'Aa'), (2, 'aA');
+create table t2 (id int, name varchar(20) character set utf8, primary key(`id`)) character set latin1 collate latin1_bin;
+insert into t2 (id, name) values (1, 'Aa'), (2, 'aA');
+set collation_server = utf8_general_ci;
+drop database if exists `sync_collation_server2`;
+create database `sync_collation_server2`;
+use `sync_collation_server2`;
+create table t1 (id int, name varchar(20), primary key(`id`));
+insert into t1 (id, name) values (1, 'Aa'), (2, 'aA');
diff --git a/dm/tests/sync_collation/data/db2.increment_err.sql b/dm/tests/sync_collation/data/db2.increment_err.sql
new file mode 100644
index 00000000000..fbd44e57747
--- /dev/null
+++ b/dm/tests/sync_collation/data/db2.increment_err.sql
@@ -0,0 +1,3 @@
+set collation_server = utf8mb4_0900_ai_ci;
+drop database if exists `sync_collation_increment2`;
+create database `sync_collation_increment2`;
diff --git a/dm/tests/sync_collation/data/db2.prepare.sql b/dm/tests/sync_collation/data/db2.prepare.sql
new file mode 100644
index 00000000000..e67b60de547
--- /dev/null
+++ b/dm/tests/sync_collation/data/db2.prepare.sql
@@ -0,0 +1,7 @@
+drop database if exists `sync_collation2`;
+create database `sync_collation2` character set utf8;
+use `sync_collation2`;
+create table t1 (id int, name varchar(20), primary key(`id`)) character set utf8;
+insert into t1 (id, name) values (1, 'Aa'), (2, 'aA');
+create table t2 (id int, name varchar(20) character set utf8, primary key(`id`)) character set latin1 collate latin1_bin;
+insert into t2 (id, name) values (1, 'Aa'), (2, 'aA');
diff --git a/dm/tests/sync_collation/data/db2.prepare_err.sql b/dm/tests/sync_collation/data/db2.prepare_err.sql
new file mode 100644
index 00000000000..0de2d569389
--- /dev/null
+++ b/dm/tests/sync_collation/data/db2.prepare_err.sql
@@ -0,0 +1,4 @@
+set collation_server = utf8mb4_0900_ai_ci;
+drop database if exists `sync_collation2`;
+create database `sync_collation2`;
+use `sync_collation2`;
diff --git a/dm/tests/sync_collation/data/tidb.checktable.increment.prepare.sql b/dm/tests/sync_collation/data/tidb.checktable.increment.prepare.sql
new file mode 100644
index 00000000000..fde2a87a791
--- /dev/null
+++ b/dm/tests/sync_collation/data/tidb.checktable.increment.prepare.sql
@@ -0,0 +1,6 @@
+use `sync_collation_increment`;
+create table t_check (id int, name varchar(20), primary key (`id`));
+insert into t_check (id, name) values (1, 'Aa'), (2, 'aA');
+use `sync_collation_increment2`;
+create table t_check (id int, name varchar(20), primary key (`id`));
+insert into t_check (id, name) values (1, 'Aa'), (2, 'aA');
\ No newline at end of file
diff --git a/dm/tests/sync_collation/data/tidb.checktable.prepare.sql b/dm/tests/sync_collation/data/tidb.checktable.prepare.sql
new file mode 100644
index 00000000000..ef1fd468038
--- /dev/null
+++ b/dm/tests/sync_collation/data/tidb.checktable.prepare.sql
@@ -0,0 +1,6 @@
+use `sync_collation`;
+create table t_check (id int, name varchar(20), primary key (`id`));
+insert into t_check (id, name) values (1, 'Aa'), (2, 'aA');
+use `sync_collation2`;
+create table t_check (id int, name varchar(20), primary key (`id`));
+insert into t_check (id, name) values (1, 'Aa'), (2, 'aA');
diff --git a/dm/tests/sync_collation/run.sh b/dm/tests/sync_collation/run.sh
new file mode 100755
index 00000000000..5eadd952fe6
--- /dev/null
+++ b/dm/tests/sync_collation/run.sh
@@ -0,0 +1,130 @@
+#!/bin/bash
+
+set -eu
+
+cur=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $cur/../_utils/test_prepare
+WORK_DIR=$TEST_DIR/$TEST_NAME
+TASK_NAME="sync_collation"
+db="sync_collation"
+db_increment="sync_collation_increment"
+db_server_collation="sync_collation_server"
+db2="sync_collation2"
+db_increment2="sync_collation_increment2"
+db_server_collation2="sync_collation_server2"
+tb="t1"
+tb2="t2"
+tb_check="t_check"
+
+function run() {
+	run_dm_master $WORK_DIR/master $MASTER_PORT $cur/conf/dm-master.toml
+	check_rpc_alive $cur/../bin/check_master_online 127.0.0.1:$MASTER_PORT
+	run_dm_worker $WORK_DIR/worker1 $WORKER1_PORT $cur/conf/dm-worker1.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER1_PORT
+	run_dm_worker $WORK_DIR/worker2 $WORKER2_PORT $cur/conf/dm-worker2.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER2_PORT
+
+	# operate mysql config to worker
+	cp $cur/conf/source1.yaml $WORK_DIR/source1.yaml
+	cp $cur/conf/source2.yaml $WORK_DIR/source2.yaml
+	dmctl_operate_source create $WORK_DIR/source1.yaml $SOURCE_ID1
+	dmctl_operate_source create $WORK_DIR/source2.yaml $SOURCE_ID2
+
+	echo "prepare data"
+	run_sql_file $cur/data/clean_data.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/clean_data.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+	run_sql_file $cur/data/clean_data.sql $TIDB_HOST $TIDB_PORT $TIDB_PASSWORD
+	run_sql_file $cur/data/db1.prepare.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/db2.prepare.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+
+	echo "start task"
+	cat $cur/conf/dm-task.yaml >$WORK_DIR/dm-task.yaml
+	dmctl_start_task $WORK_DIR/dm-task.yaml "--remove-meta"
+
+	echo "check full phase"
+	# wait
+	run_sql_tidb_with_retry " select count(1) from information_schema.tables where TABLE_SCHEMA='${db}' and TABLE_NAME = '${tb}';" "count(1): 1"
+	run_sql_tidb_with_retry " select count(1) from information_schema.tables where TABLE_SCHEMA='${db2}' and TABLE_NAME = '${tb}';" "count(1): 1"
+
+	# check table
+	run_sql_tidb_with_retry "select count(1) from ${db}.${tb} where name = 'aa';" "count(1): 2"
+	run_sql_tidb_with_retry "select count(1) from ${db2}.${tb} where name = 'aa';" "count(1): 2"
+
+	# check column
+	run_sql_tidb_with_retry "select count(1) from ${db}.${tb2} where name = 'aa';" "count(1): 2"
+	run_sql_tidb_with_retry "select count(1) from ${db2}.${tb2} where name = 'aa';" "count(1): 2"
+
+	# check database by create table
+	run_sql_file $cur/data/tidb.checktable.prepare.sql $TIDB_HOST $TIDB_PORT $TIDB_PASSWORD
+	run_sql_tidb_with_retry "select count(1) from ${db}.${tb_check} where name = 'aa';" "count(1): 2"
+	run_sql_tidb_with_retry "select count(1) from ${db2}.${tb_check} where name = 'aa';" "count(1): 2"
+
+	echo "prepare incremental data"
+	run_sql_file $cur/data/db1.increment.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/db2.increment.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+
+	echo "check incremental phase"
+	# wait
+	run_sql_tidb_with_retry " select count(1) from information_schema.tables where TABLE_SCHEMA='${db_increment}' and TABLE_NAME = '${tb}';" "count(1): 1"
+	run_sql_tidb_with_retry " select count(1) from information_schema.tables where TABLE_SCHEMA='${db_increment2}' and TABLE_NAME = '${tb}';" "count(1): 1"
+
+	# check table
+	run_sql_tidb_with_retry "select count(1) from ${db_increment}.${tb} where name = 'aa';" "count(1): 2"
+	run_sql_tidb_with_retry "select count(1) from ${db_increment2}.${tb} where name = 'aa';" "count(1): 2"
+
+	# check column
+	run_sql_tidb_with_retry "select count(1) from ${db_increment}.${tb2} where name = 'aa';" "count(1): 2"
+	run_sql_tidb_with_retry "select count(1) from ${db_increment2}.${tb2} where name = 'aa';" "count(1): 2"
+
+	# check database by create table
+	run_sql_file $cur/data/tidb.checktable.increment.prepare.sql $TIDB_HOST $TIDB_PORT $TIDB_PASSWORD
+	run_sql_tidb_with_retry "select count(1) from ${db_increment}.${tb_check} where name = 'aa';" "count(1): 2"
+	run_sql_tidb_with_retry "select count(1) from ${db_increment2}.${tb_check} where name = 'aa';" "count(1): 2"
+
+	# check set server collation
+	run_sql_tidb_with_retry "select count(1) from ${db_server_collation}.${tb} where name = 'aa';" "count(1): 2"
+	run_sql_tidb_with_retry "select count(1) from ${db_server_collation2}.${tb} where name = 'aa';" "count(1): 2"
+
+	dmctl_stop_task $TASK_NAME $MASTER_PORT
+
+	echo "prepare data for full phase error test"
+	run_sql_file $cur/data/clean_data.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/clean_data.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+	run_sql_file $cur/data/clean_data.sql $TIDB_HOST $TIDB_PORT $TIDB_PASSWORD
+	run_sql_file $cur/data/db1.prepare_err.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/db2.prepare_err.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+
+	dmctl_start_task $WORK_DIR/dm-task.yaml "--remove-meta"
+
+	echo "check full phase error"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status ${TASK_NAME}" \
+		"Error 1273: Unsupported collation when new collation is enabled: 'latin1_swedish_ci'" 1 \
+		"Error 1273: Unsupported collation when new collation is enabled: 'utf8mb4_0900_ai_ci'" 1
+
+	dmctl_stop_task $TASK_NAME $MASTER_PORT
+
+	echo "prepare data for incremental phase error test"
+	run_sql_file $cur/data/clean_data.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/clean_data.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+	run_sql_file $cur/data/clean_data.sql $TIDB_HOST $TIDB_PORT $TIDB_PASSWORD
+
+	dmctl_start_task $WORK_DIR/dm-task.yaml "--remove-meta"
+
+	run_sql_file $cur/data/db1.increment_err.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	run_sql_file $cur/data/db2.increment_err.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+
+	echo "check incremental phase error"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status ${TASK_NAME}" \
+		"Error 1273: Unsupported collation when new collation is enabled: 'latin1_swedish_ci'" 1 \
+		"Error 1273: Unsupported collation when new collation is enabled: 'utf8mb4_0900_ai_ci'" 1
+}
+
+cleanup_data $TEST_NAME
+# also cleanup dm processes in case of last run failed
+cleanup_process $*
+run $*
+cleanup_process $*
+
+echo "[$(date)] <<<<<< test case $TEST_NAME success! >>>>>>"
diff --git a/go.mod b/go.mod
index 6883519d34b..47a668e87e2 100644
--- a/go.mod
+++ b/go.mod
@@ -54,11 +54,11 @@ require (
 	github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712
 	github.com/pingcap/errors v0.11.5-0.20211009033009-93128226aaa3
 	github.com/pingcap/failpoint v0.0.0-20210316064728-7acb0f0a3dfd
-	github.com/pingcap/kvproto v0.0.0-20211122024046-03abd340988f
-	github.com/pingcap/log v0.0.0-20211207084639-71a2e5860834
-	github.com/pingcap/tidb v1.1.0-beta.0.20211208001616-f900cfafda19
+	github.com/pingcap/kvproto v0.0.0-20211202065422-a412f7a319c3
+	github.com/pingcap/log v0.0.0-20210906054005-afc726e70354
+	github.com/pingcap/tidb v1.1.0-beta.0.20211209055157-9f744cdf8266
 	github.com/pingcap/tidb-tools v5.2.3-0.20211105044302-2dabb6641a6e+incompatible
-	github.com/pingcap/tidb/parser v0.0.0-20211208001616-f900cfafda19
+	github.com/pingcap/tidb/parser v0.0.0-20211209055157-9f744cdf8266
 	github.com/prometheus/client_golang v1.7.1
 	github.com/prometheus/client_model v0.2.0
 	github.com/r3labs/diff v1.1.0
diff --git a/go.sum b/go.sum
index 443b3e4d0f6..c9aed6bf381 100644
--- a/go.sum
+++ b/go.sum
@@ -816,16 +816,16 @@ github.com/pingcap/kvproto v0.0.0-20210805052247-76981389e818/go.mod h1:IOdRDPLy
 github.com/pingcap/kvproto v0.0.0-20210819164333-bd5706b9d9f2/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20210915062418-0f5764a128ad/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20211109071446-a8b4d34474bc/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
-github.com/pingcap/kvproto v0.0.0-20211122024046-03abd340988f h1:hjInxK1Ie6CYx7Jy2pYnBdEnWI8jIfr423l9Yh6LRy8=
 github.com/pingcap/kvproto v0.0.0-20211122024046-03abd340988f/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
+github.com/pingcap/kvproto v0.0.0-20211202065422-a412f7a319c3 h1:PZMewlB2b5JC0pMteDmWNBtAgmJv2ih1+z+WifdVcEk=
+github.com/pingcap/kvproto v0.0.0-20211202065422-a412f7a319c3/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20200511115504-543df19646ad/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20201112100606-8f1e84a3abc8/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20210317133921-96f4fcab92a4/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM=
+github.com/pingcap/log v0.0.0-20210906054005-afc726e70354 h1:SvWCbCPh1YeHd9yQLksvJYAgft6wLTY1aNG81tpyscQ=
 github.com/pingcap/log v0.0.0-20210906054005-afc726e70354/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
-github.com/pingcap/log v0.0.0-20211207084639-71a2e5860834 h1:/wKEagMl3KDpt7758/4Z+31yY3uxQCygOZ1xSbRXCP0=
-github.com/pingcap/log v0.0.0-20211207084639-71a2e5860834/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/parser v0.0.0-20210415081931-48e7f467fd74/go.mod h1:xZC8I7bug4GJ5KtHhgAikjTfU4kBv1Sbo3Pf1MZ6lVw=
 github.com/pingcap/parser v0.0.0-20210525032559-c37778aff307/go.mod h1:xZC8I7bug4GJ5KtHhgAikjTfU4kBv1Sbo3Pf1MZ6lVw=
 github.com/pingcap/sysutil v0.0.0-20200206130906-2bfa6dc40bcd/go.mod h1:EB/852NMQ+aRKioCpToQ94Wl7fktV+FNnxf3CX/TTXI=
@@ -833,8 +833,8 @@ github.com/pingcap/sysutil v0.0.0-20210315073920-cc0985d983a3/go.mod h1:tckvA041
 github.com/pingcap/sysutil v0.0.0-20210730114356-fcd8a63f68c5 h1:7rvAtZe/ZUzOKzgriNPQoBNvleJXBk4z7L3Z47+tS98=
 github.com/pingcap/sysutil v0.0.0-20210730114356-fcd8a63f68c5/go.mod h1:XsOaV712rUk63aOEKYP9PhXTIE3FMNHmC2r1wX5wElY=
 github.com/pingcap/tidb v1.1.0-beta.0.20211023132847-efa94595c071/go.mod h1:Ci7ABF58a4jn6YtaHi7655jP409edqC2JxWWFRqOubg=
-github.com/pingcap/tidb v1.1.0-beta.0.20211208001616-f900cfafda19 h1:pGLR1DOev6mLv16SkvqRXOl+KEs1QuaL3FWBr4nuzRg=
-github.com/pingcap/tidb v1.1.0-beta.0.20211208001616-f900cfafda19/go.mod h1:R2E0ZlYqy7GLnRUmW5uqbi+jM3pRhtseEPqBuQZrmmA=
+github.com/pingcap/tidb v1.1.0-beta.0.20211209055157-9f744cdf8266 h1:tfSd06ZEBlJephWmWGtuZRInvETtGA5Tby3QcDsZ6VA=
+github.com/pingcap/tidb v1.1.0-beta.0.20211209055157-9f744cdf8266/go.mod h1:R2E0ZlYqy7GLnRUmW5uqbi+jM3pRhtseEPqBuQZrmmA=
 github.com/pingcap/tidb-dashboard v0.0.0-20210312062513-eef5d6404638/go.mod h1:OzFN8H0EDMMqeulPhPMw2i2JaiZWOKFQ7zdRPhENNgo=
 github.com/pingcap/tidb-dashboard v0.0.0-20210716172320-2226872e3296/go.mod h1:OCXbZTBTIMRcIt0jFsuCakZP+goYRv6IjawKbwLS2TQ=
 github.com/pingcap/tidb-dashboard v0.0.0-20211008050453-a25c25809529/go.mod h1:OCXbZTBTIMRcIt0jFsuCakZP+goYRv6IjawKbwLS2TQ=
@@ -845,8 +845,8 @@ github.com/pingcap/tidb-tools v5.2.3-0.20211105044302-2dabb6641a6e+incompatible
 github.com/pingcap/tidb-tools v5.2.3-0.20211105044302-2dabb6641a6e+incompatible/go.mod h1:XGdcy9+yqlDSEMTpOXnwf3hiTeqrV6MN/u1se9N8yIM=
 github.com/pingcap/tidb/parser v0.0.0-20211011031125-9b13dc409c5e/go.mod h1:e1MGCA9Sg3T8jid8PKAEq5eYVuMMCq4n8gJ+Kqp4Plg=
 github.com/pingcap/tidb/parser v0.0.0-20211023132847-efa94595c071/go.mod h1:e1MGCA9Sg3T8jid8PKAEq5eYVuMMCq4n8gJ+Kqp4Plg=
-github.com/pingcap/tidb/parser v0.0.0-20211208001616-f900cfafda19 h1:bohoMvyPQaPoDy8gYmMh2gTGNbUkdP2XZuv1pWrAaD8=
-github.com/pingcap/tidb/parser v0.0.0-20211208001616-f900cfafda19/go.mod h1:ElJiub4lRy6UZDb+0JHDkGEdr6aOli+ykhyej7VCLoI=
+github.com/pingcap/tidb/parser v0.0.0-20211209055157-9f744cdf8266 h1:6FuBP6k1AXX4pLiwBLkEqBb26ncwZmxUqoJEpc+WXGM=
+github.com/pingcap/tidb/parser v0.0.0-20211209055157-9f744cdf8266/go.mod h1:ElJiub4lRy6UZDb+0JHDkGEdr6aOli+ykhyej7VCLoI=
 github.com/pingcap/tipb v0.0.0-20211008080435-3fd327dfce0e/go.mod h1:A7mrd7WHBl1o63LE2bIBGEJMTNWXqhgmYiOvMLxozfs=
 github.com/pingcap/tipb v0.0.0-20211201080053-bd104bb270ba h1:Tt5W/maVBUbG+wxg2nfc88Cqj/HiWYb0TJQ2Rfi0UOQ=
 github.com/pingcap/tipb v0.0.0-20211201080053-bd104bb270ba/go.mod h1:A7mrd7WHBl1o63LE2bIBGEJMTNWXqhgmYiOvMLxozfs=