(redo)ticdc: fix the event orderliness in redo log (#11117) (#11181)

close #11096

cdc/redo/reader/reader.go

@@ -364,6 +364,7 @@ func (h logHeap) Len() int {
 }
 
 func (h logHeap) Less(i, j int) bool {
+	// we separate ddl and dml, so we only need to compare dml with dml, and ddl with ddl.
 	if h[i].data.Type == model.RedoLogTypeDDL {
 		if h[i].data.RedoDDL.DDL == nil {
 			return true
@@ -381,10 +382,19 @@ func (h logHeap) Less(i, j int) bool {
 		return false
 	}
 
-	if h[i].data.RedoRow.Row.CommitTs == h[j].data.RedoRow.Row.CommitTs &&
-		h[i].data.RedoRow.Row.StartTs < h[j].data.RedoRow.Row.StartTs {
-		return true
+	if h[i].data.RedoRow.Row.CommitTs == h[j].data.RedoRow.Row.CommitTs {
+		if h[i].data.RedoRow.Row.StartTs != h[j].data.RedoRow.Row.StartTs {
+			return h[i].data.RedoRow.Row.StartTs < h[j].data.RedoRow.Row.StartTs
+		}
+		// in the same txn, we need to sort by delete/update/insert order
+		if h[i].data.RedoRow.Row.IsDelete() {
+			return true
+		} else if h[i].data.RedoRow.Row.IsUpdate() {
+			return !h[j].data.RedoRow.Row.IsDelete()
+		}
+		return false
 	}
+
 	return h[i].data.RedoRow.Row.CommitTs < h[j].data.RedoRow.Row.CommitTs
 }