(redo)ticdc: fix the event orderliness in redo log

cdc/redo/reader/reader.go

@@ -346,6 +346,7 @@
 }
 
 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
@@ -363,10 +364,22 @@
 		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 true
+		} else if h[i].data.RedoRow.Row.StartTs > h[j].data.RedoRow.Row.StartTs {
+			return false
+		} else {
+			// in the same txn, we need to sort by delete/update/insert order
+			if h[i].data.RedoRow.Row.ToRowChangedEvent().IsDelete() {
+				return true
+			} else if h[i].data.RedoRow.Row.ToRowChangedEvent().IsUpdate() {
+				return !h[j].data.RedoRow.Row.ToRowChangedEvent().IsDelete()
+			}
+			return false
+		}
 	}
+
 	return h[i].data.RedoRow.Row.CommitTs < h[j].data.RedoRow.Row.CommitTs
 }