cdc(ddl): ensure strict ordering for multi-table DDLs after split

cdc/entry/schema_storage.go

@@ -519,6 +519,7 @@ func (s *schemaStorage) buildRenameEvents(
 		tableInfo := model.WrapTableInfo(info.NewSchemaID, newSchemaName,
 			job.BinlogInfo.FinishedTS, tableInfo)
 		event.FromJobWithArgs(job, preTableInfo, tableInfo, oldSchemaName, newSchemaName)
+		event.Seq = uint64(i)
 		ddlEvents = append(ddlEvents, event)
 	}
 	return ddlEvents, nil

cdc/model/sink.go

@@ -1056,6 +1056,10 @@ type DDLEvent struct {
 	// the DDL is executed by the primary cluster.
 	BDRRole string        `msg:"-"`
 	SQLMode mysql.SQLMode `msg:"-"`
+	// Seq is used to order the DDLs with the same commit ts
+	// Only used in the splited DDLEvent generated by a multi-table DDL,
+	// we need to keep the order of the original multi-table DDL
+	Seq uint64 `msg:"seq"`
 }
 
 // FromJob fills the values with DDLEvent from DDL job

cdc/owner/ddl_manager.go

@@ -486,7 +486,7 @@ func (m *ddlManager) getNextDDL() *model.DDLEvent {
 			delete(m.pendingDDLs, tb)
 			continue
 		}
-		if res == nil || res.CommitTs > ddls[0].CommitTs {
+		if res == nil || res.CommitTs > ddls[0].CommitTs || (res.CommitTs == ddls[0].CommitTs && res.Seq > ddls[0].Seq) {
 			res = ddls[0]
 		}
 	}