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

ref #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 {