diff --git a/cdc/schema.go b/cdc/schema.go
index d4f8becdf05..90d138242df 100644
--- a/cdc/schema.go
+++ b/cdc/schema.go
@@ -307,6 +307,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 {
diff --git a/cdc/schema_test.go b/cdc/schema_test.go
index 966dd0b460c..b46460d6a98 100644
--- a/cdc/schema_test.go
+++ b/cdc/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},