sink/codec(cdc): refactor Avro codec (pingcap#5339)

ref pingcap#5338

cdc/model/sink.go

@@ -326,13 +326,35 @@ func (r *RowChangedEvent) HandleKeyColumns() []*Column {
 	}
 
 	if len(pkeyCols) == 0 {
-		// TODO redact the message
-		log.Panic("Cannot find handle key columns, bug?", zap.Reflect("event", r))
+		log.Panic("Cannot find handle key columns.", zap.Any("event", r))
 	}
 
 	return pkeyCols
 }
 
+// PrimaryKeyColInfos returns the column(s) and colInfo(s) corresponding to the primary key(s)
+func (r *RowChangedEvent) PrimaryKeyColInfos() ([]*Column, []rowcodec.ColInfo) {
+	pkeyCols := make([]*Column, 0)
+	pkeyColInfos := make([]rowcodec.ColInfo, 0)
+
+	var cols []*Column
+	if r.IsDelete() {
+		cols = r.PreColumns
+	} else {
+		cols = r.Columns
+	}
+
+	for i, col := range cols {
+		if col != nil && col.Flag.IsPrimaryKey() {
+			pkeyCols = append(pkeyCols, col)
+			pkeyColInfos = append(pkeyColInfos, r.ColInfos[i])
+		}
+	}
+
+	// It is okay not to have primary keys, so the empty array is an acceptable result
+	return pkeyCols, pkeyColInfos
+}
+
 // WithHandlePrimaryFlag set `HandleKeyFlag` and `PrimaryKeyFlag`
 func (r *RowChangedEvent) WithHandlePrimaryFlag(colNames map[string]struct{}) {
 	for _, col := range r.Columns {