drainer: Bugfix, should handle ActionModifySchemaCharsetAndCollate correctly (pingcap#769)

Author: suzaku

drainer/schema.go

@@ -289,6 +289,18 @@ func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string,
 		s.currentVersion = job.BinlogInfo.SchemaVersion
 		schemaName = schema.Name.O
 
+	case model.ActionModifySchemaCharsetAndCollate:
+		db := job.BinlogInfo.DBInfo
+		if _, ok := s.schemas[db.ID]; !ok {
+			return "", "", "", errors.NotFoundf("schema %s(%d)", db.Name, db.ID)
+		}
+
+		s.schemas[db.ID] = db
+		s.schemaNameToID[db.Name.O] = db.ID
+		s.version2SchemaTable[job.BinlogInfo.SchemaVersion] = TableName{Schema: db.Name.O, Table: ""}
+		s.currentVersion = job.BinlogInfo.SchemaVersion
+		schemaName = db.Name.O
+
 	case model.ActionDropSchema:
 		schemaName, err = s.DropSchema(job.SchemaID)
 		if err != nil {

drainer/schema_test.go

@@ -327,6 +327,7 @@ func (t *schemaSuite) TestHandleDDL(c *C) {
 		tableName   string
 	}{
 		{name: "createSchema", jobID: 3, schemaID: 2, tableID: 0, jobType: model.ActionCreateSchema, binlogInfo: &model.HistoryInfo{SchemaVersion: 1, DBInfo: dbInfo, TableInfo: nil, FinishedTS: 123}, query: "create database Test", resultQuery: "create database Test", schemaName: dbInfo.Name.O, tableName: ""},
+		{name: "updateSchema", jobID: 4, schemaID: 2, tableID: 0, jobType: model.ActionModifySchemaCharsetAndCollate, binlogInfo: &model.HistoryInfo{SchemaVersion: 8, DBInfo: dbInfo, TableInfo: nil, FinishedTS: 123}, query: "ALTER DATABASE Test CHARACTER SET utf8mb4;", resultQuery: "ALTER DATABASE Test CHARACTER SET utf8mb4;", schemaName: dbInfo.Name.O},
 		{name: "createTable", jobID: 7, schemaID: 2, tableID: 6, jobType: model.ActionCreateTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 3, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "create table T(id int);", resultQuery: "create table T(id int);", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},
 		{name: "addColumn", jobID: 9, schemaID: 2, tableID: 6, jobType: model.ActionAddColumn, binlogInfo: &model.HistoryInfo{SchemaVersion: 4, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "alter table T add a varchar(45);", resultQuery: "alter table T add a varchar(45);", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},
 		{name: "truncateTable", jobID: 10, schemaID: 2, tableID: 6, jobType: model.ActionTruncateTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 5, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "truncate table T;", resultQuery: "truncate table T;", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},

tests/dailytest/case.go

@@ -164,6 +164,16 @@ var caseRecoverAndInsertClean = []string{`
 `,
 }
 
+var (
+	caseAlterDatabase = []string{
+		`CREATE DATABASE to_be_altered CHARACTER SET utf8;`,
+		`ALTER DATABASE to_be_altered CHARACTER SET utf8mb4;`,
+	}
+	caseAlterDatabaseClean = []string{
+		`DROP DATABASE to_be_altered;`,
+	}
+)
+
 type testRunner struct {
 	src    *sql.DB
 	dst    *sql.DB
@@ -195,6 +205,9 @@ func RunCase(src *sql.DB, dst *sql.DB, schema string) {
 	tr.execSQLs(caseUKWithNoPK)
 	tr.execSQLs(caseUKWithNoPKClean)
 
+	tr.execSQLs(caseAlterDatabase)
+	tr.execSQLs(caseAlterDatabaseClean)
+
 	// run casePKAddDuplicateUK
 	tr.run(func(src *sql.DB) {
 		err := execSQLs(src, casePKAddDuplicateUK)