br: compare collate check privilege

br/pkg/restore/BUILD.bazel

@@ -122,6 +122,7 @@ go_test(
         "search_test.go",
         "split_test.go",
         "stream_metas_test.go",
+        "systable_restore_test.go",
         "util_test.go",
     ],
     embed = [":restore"],

br/pkg/restore/client.go

@@ -206,6 +206,8 @@ type Client struct {
 
 	// checkpoint information for log restore
 	useCheckpoint bool
+
+	privilegeTableRowsCollateCompatibility bool
 }
 
 // NewRestoreClient returns a new RestoreClient.
@@ -1253,7 +1255,13 @@ func (rc *Client) CheckSysTableCompatibility(dom *domain.Domain, tables []*metau
 					table.Info.Name.O,
 					col.Name, col.FieldType.String())
 			}
-			if !utils.IsTypeCompatible(backupCol.FieldType, col.FieldType) {
+			typeEq, collateEq := utils.IsTypeCompatible(backupCol.FieldType, col.FieldType)
+			collateCompatible := collateEq
+			if typeEq && !collateEq {
+				rc.privilegeTableRowsCollateCompatibility = true
+				collateCompatible = checkSysTableColumnCollateCompatibility(mysql.SystemDB, table.Info.Name.L, col.Name.L, backupCol.GetCollate(), col.GetCollate())
+			}
+			if !(typeEq && collateCompatible) {
 				log.Error("incompatible column",
 					zap.Stringer("table", table.Info.Name),
 					zap.String("col in cluster", fmt.Sprintf("%s %s", col.Name, col.FieldType.String())),

br/pkg/restore/client_test.go

@@ -278,7 +278,44 @@ func TestCheckSysTableCompatibility(t *testing.T) {
 
 	// other system tables in cluster have more columns(failed)
 	mockedDBTI := dbTI.Clone()
-	dbTI.Columns = append(dbTI.Columns, &model.ColumnInfo{Name: model.NewCIStr("new-name")})
+	mockedDBTI.Columns = append(dbTI.Columns, &model.ColumnInfo{Name: model.NewCIStr("new-name")})
+	err = client.CheckSysTableCompatibility(cluster.Domain, []*metautil.Table{{
+		DB:   tmpSysDB,
+		Info: mockedDBTI,
+	}})
+	require.True(t, berrors.ErrRestoreIncompatibleSys.Equal(err))
+
+	// skip check collate
+	mockedDBTI = dbTI.Clone()
+	mockedDBTI.Columns[1].SetCollate("utf8mb4_bin")
+	err = client.CheckSysTableCompatibility(cluster.Domain, []*metautil.Table{{
+		DB:   tmpSysDB,
+		Info: mockedDBTI,
+	}})
+	require.NoError(t, err)
+
+	// skip check collate but type mismatch
+	mockedDBTI = dbTI.Clone()
+	mockedDBTI.Columns[1].SetCollate("utf8mb4_bin")
+	mockedDBTI.Columns[1].FieldType.SetFlen(2000) // Columns[1] is `DB` char(64)
+	err = client.CheckSysTableCompatibility(cluster.Domain, []*metautil.Table{{
+		DB:   tmpSysDB,
+		Info: mockedDBTI,
+	}})
+	require.True(t, berrors.ErrRestoreIncompatibleSys.Equal(err))
+
+	// another column collate mismatch
+	mockedDBTI = dbTI.Clone()
+	mockedDBTI.Columns[0].SetCollate("utf8mb4_general_ci")
+	err = client.CheckSysTableCompatibility(cluster.Domain, []*metautil.Table{{
+		DB:   tmpSysDB,
+		Info: mockedDBTI,
+	}})
+	require.True(t, berrors.ErrRestoreIncompatibleSys.Equal(err))
+
+	// another column collate mismatch
+	mockedDBTI = dbTI.Clone()
+	mockedDBTI.Columns[1].SetCollate("utf8mb4_unicode_ci")
 	err = client.CheckSysTableCompatibility(cluster.Domain, []*metautil.Table{{
 		DB:   tmpSysDB,
 		Info: mockedDBTI,

br/pkg/restore/systable_restore.go

@@ -13,6 +13,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/logutil"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/pkg/bindinfo"
+	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	filter "github.com/pingcap/tidb/pkg/util/table-filter"
@@ -60,6 +61,32 @@ var unRecoverableTable = map[string]map[string]struct{}{
 	},
 }
 
+type checkPrivilegeTableRowsCollateCompatibilitySQLPair struct {
+	upstreamCollateSQL   string
+	downstreamCollateSQL string
+	columns              map[string]struct{}
+}
+
+var collateCompatibilityTables = map[string]map[string]checkPrivilegeTableRowsCollateCompatibilitySQLPair{
+	"mysql": {
+		"db": {
+			upstreamCollateSQL:   "SELECT COUNT(1) FROM __TiDB_BR_Temporary_mysql.db",
+			downstreamCollateSQL: "SELECT COUNT(1) FROM (SELECT Host, DB COLLATE utf8mb4_general_ci, User FROM __TiDB_BR_Temporary_mysql.db GROUP BY Host, DB COLLATE utf8mb4_general_ci, User) as a",
+			columns:              map[string]struct{}{"db": {}},
+		},
+		"tables_priv": {
+			upstreamCollateSQL:   "SELECT COUNT(1) FROM __TiDB_BR_Temporary_mysql.tables_priv",
+			downstreamCollateSQL: "SELECT COUNT(1) FROM (SELECT Host, DB COLLATE utf8mb4_general_ci, User, Table_name COLLATE utf8mb4_general_ci FROM __TiDB_BR_Temporary_mysql.tables_priv GROUP BY Host, DB COLLATE utf8mb4_general_ci, User, Table_name COLLATE utf8mb4_general_ci) as a",
+			columns:              map[string]struct{}{"db": {}, "table_name": {}},
+		},
+		"columns_priv": {
+			upstreamCollateSQL:   "SELECT COUNT(1) FROM __TiDB_BR_Temporary_mysql.columns_priv",
+			downstreamCollateSQL: "SELECT COUNT(1) FROM (SELECT Host, DB COLLATE utf8mb4_general_ci, User, Table_name COLLATE utf8mb4_general_ci, Column_name COLLATE utf8mb4_general_ci FROM __TiDB_BR_Temporary_mysql.columns_priv GROUP BY Host, DB COLLATE utf8mb4_general_ci, User, Table_name COLLATE utf8mb4_general_ci, Column_name COLLATE utf8mb4_general_ci) as a",
+			columns:              map[string]struct{}{"db": {}, "table_name": {}, "column_name": {}},
+		},
+	},
+}
+
 func isUnrecoverableTable(schemaName string, tableName string) bool {
 	tableMap, ok := unRecoverableTable[schemaName]
 	if !ok {
@@ -252,6 +279,11 @@ func (rc *Client) replaceTemporaryTableToSystable(ctx context.Context, ti *model
 		log.Info("replace into existing table",
 			zap.String("table", tableName),
 			zap.Stringer("schema", db.Name))
+		if rc.privilegeTableRowsCollateCompatibility {
+			if err := rc.checkPrivilegeTableRowsCollateCompatibility(ctx, dbName, tableName, ti, db.ExistingTables[tableName]); err != nil {
+				return err
+			}
+		}
 		// target column order may different with source cluster
 		columnNames := make([]string, 0, len(ti.Columns))
 		for _, col := range ti.Columns {
@@ -283,3 +315,97 @@ func (rc *Client) cleanTemporaryDatabase(ctx context.Context, originDB string) {
 		)
 	}
 }
+
+func checkSysTableColumnCollateCompatibility(dbNameL, tableNameL, columnNameL, upstreamCollate, downstreamCollate string) bool {
+	if upstreamCollate != "utf8mb4_bin" || downstreamCollate != "utf8mb4_general_ci" {
+		return false
+	}
+	collateCompatibilityTableMap, exists := collateCompatibilityTables[dbNameL]
+	if !exists {
+		return false
+	}
+	collateCompatibilityColumnMap, exists := collateCompatibilityTableMap[tableNameL]
+	if !exists {
+		return false
+	}
+	_, exists = collateCompatibilityColumnMap.columns[columnNameL]
+	return exists
+}
+
+func (rc *Client) checkPrivilegeTableRowsCollateCompatibility(
+	ctx context.Context,
+	dbNameL, tableNameL string,
+	upstreamTable, downstreamTable *model.TableInfo,
+) error {
+	collateCompatibilityTableMap, exists := collateCompatibilityTables[dbNameL]
+	if !exists {
+		return nil
+	}
+	collateCompatibilityColumnMap, exists := collateCompatibilityTableMap[tableNameL]
+	if !exists {
+		return nil
+	}
+	colCount := 0
+	for _, col := range upstreamTable.Columns {
+		if _, exists := collateCompatibilityColumnMap.columns[col.Name.L]; exists {
+			if col.GetCollate() != "utf8mb4_bin" && col.GetCollate() != "utf8mb4_general_ci" {
+				return errors.Annotatef(berrors.ErrRestoreIncompatibleSys,
+					"incompatible column collate, upstream table %s.%s column %s collate is %s but should be utf8mb4_bin or utf8mb4_general_ci",
+					dbNameL, tableNameL, col.Name.L, col.GetCollate())
+			}
+			colCount += 1
+		}
+	}
+	if colCount != len(collateCompatibilityColumnMap.columns) {
+		return errors.Annotatef(berrors.ErrRestoreIncompatibleSys,
+			"incompatible column collate, upstream table %s.%s has only %d compatible columns",
+			dbNameL, tableNameL, colCount)
+	}
+	colCount = 0
+	for _, col := range downstreamTable.Columns {
+		if _, exists := collateCompatibilityColumnMap.columns[col.Name.L]; exists {
+			if col.GetCollate() != "utf8mb4_general_ci" {
+				return errors.Annotatef(berrors.ErrRestoreIncompatibleSys,
+					"incompatible column collate, downstream table %s.%s column %s collate is %s but should be utf8mb4_general_ci",
+					dbNameL, tableNameL, col.Name.L, col.GetCollate())
+			}
+			colCount += 1
+		}
+	}
+	if colCount != len(collateCompatibilityColumnMap.columns) {
+		return errors.Annotatef(berrors.ErrRestoreIncompatibleSys,
+			"incompatible column collate, downstream table %s.%s has only %d compatible columns",
+			dbNameL, tableNameL, colCount)
+	}
+	ectx := rc.db.se.GetSessionCtx().GetRestrictedSQLExecutor()
+	rows, _, err := ectx.ExecRestrictedSQL(
+		kv.WithInternalSourceType(ctx, kv.InternalTxnBR),
+		nil,
+		collateCompatibilityColumnMap.upstreamCollateSQL,
+	)
+	if err != nil {
+		return errors.Annotatef(err, "failed to get the count of privilege rows")
+	}
+	if len(rows) == 0 {
+		return errors.Errorf("failed to get the count of privilege rows")
+	}
+	upstreamCount := rows[0].GetInt64(0)
+	rows, _, err = ectx.ExecRestrictedSQL(
+		kv.WithInternalSourceType(ctx, kv.InternalTxnBR),
+		nil,
+		collateCompatibilityColumnMap.downstreamCollateSQL,
+	)
+	if err != nil {
+		return errors.Annotatef(err, "failed to get the count of privilege rows")
+	}
+	if len(rows) == 0 {
+		return errors.Errorf("failed to get the count of privilege rows")
+	}
+	downstreamCount := rows[0].GetInt64(0)
+	if upstreamCount != downstreamCount {
+		return errors.Annotatef(berrors.ErrRestoreIncompatibleSys,
+			"there are duplicated privilege rows with collate utf8mb4_general_ci [upstream count %d != downstream count %d]",
+			upstreamCount, downstreamCount)
+	}
+	return nil
+}