*: chose one index to output delete events

cdc/entry/mounter.go

@@ -18,9 +18,7 @@ import (
 	"context"
 	"encoding/binary"
 	"encoding/json"
-	"fmt"
 	"math/rand"
-	"strconv"
 	"strings"
 	"time"
 
@@ -378,11 +376,7 @@ func datum2Column(tableInfo *model.TableInfo, datums map[int64]types.Datum, fill
 		if !exist {
 			return nil, errors.NotFoundf("column info, colID: %d", index)
 		}
-		// the judge about `fillWithDefaultValue` is tricky
-		// if the `fillWithDefaultValue` is true, the event must be deletion
-		// we should output the generated column in deletion event
-		// this tricky code will be improve after pingcap/ticdc#787 merged
-		if !tableInfo.IsColWritable(colInfo) && fillWithDefaultValue {
+		if !tableInfo.IsColCDCVisible(colInfo) {
 			continue
 		}
 		colName := colInfo.Name.O
@@ -401,24 +395,28 @@ func datum2Column(tableInfo *model.TableInfo, datums map[int64]types.Datum, fill
 		}
 		cols[colName] = col
 	}
-	if !fillWithDefaultValue {
-		return cols, nil
-	}
-	for _, col := range tableInfo.Columns {
-		_, ok := cols[col.Name.O]
-		if !ok && tableInfo.IsColWritable(col) {
-			column := &model.Column{
-				Type:  col.Tp,
-				Value: getDefaultOrZeroValue(col),
-				Flag:  transColumnFlag(col),
-			}
-			if tableInfo.IsColumnUnique(col.ID) {
-				whereHandle := true
-				column.WhereHandle = &whereHandle
+	if fillWithDefaultValue {
+		for _, col := range tableInfo.Columns {
+			_, ok := cols[col.Name.O]
+			if !ok && tableInfo.IsColCDCVisible(col) {
+				column := &model.Column{
+					Type:  col.Tp,
+					Value: getDefaultOrZeroValue(col),
+					Flag:  transColumnFlag(col),
+				}
+				if tableInfo.IsColumnUnique(col.ID) {
+					whereHandle := true
+					column.WhereHandle = &whereHandle
+				}
+				cols[col.Name.O] = column
 			}
-			cols[col.Name.O] = column
 		}
 	}
+
+	err := setHandleKeyFlag(tableInfo, cols)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
 	return cols, nil
 }
 
@@ -480,11 +478,43 @@ func (m *mounterImpl) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntr
 	}, nil
 }
 
+func setHandleKeyFlag(tableInfo *model.TableInfo, colValues map[string]*model.Column) error {
+	switch tableInfo.HandleIndexID {
+	case model.HandleIndexTableIneligible:
+		log.Fatal("this table is not a eligible", zap.Int64("tableID", tableInfo.ID))
+	case model.HandleIndexPKIsHandle:
+		// pk is handle
+		if !tableInfo.PKIsHandle {
+			log.Fatal("the pk of this table is not handle", zap.Int64("tableID", tableInfo.ID))
+		}
+		for _, colInfo := range tableInfo.Columns {
+			if mysql.HasPriKeyFlag(colInfo.Flag) {
+				colValues[colInfo.Name.O].Flag.SetIsHandleKey()
+				break
+			}
+		}
+	default:
+		handleIndexInfo, exist := tableInfo.GetIndexInfo(tableInfo.HandleIndexID)
+		if !exist {
+			return errors.NotFoundf("handle index info(%d) in table(%d)", tableInfo.HandleIndexID, tableInfo.ID)
+		}
+		for _, colInfo := range handleIndexInfo.Columns {
+			colName := tableInfo.Columns[colInfo.Offset].Name.O
+			colValues[colName].Flag.SetIsHandleKey()
+		}
+	}
+	return nil
+}
+
 func (m *mounterImpl) mountIndexKVEntry(tableInfo *model.TableInfo, idx *indexKVEntry) (*model.RowChangedEvent, error) {
 	// skip set index KV
 	if !idx.Delete || m.enableOldValue {
 		return nil, nil
 	}
+	// skip the index which not handle
+	if idx.IndexID != tableInfo.HandleIndexID {
+		return nil, nil
+	}
 
 	indexInfo, exist := tableInfo.GetIndexInfo(idx.IndexID)
 	if !exist {
@@ -508,12 +538,14 @@ func (m *mounterImpl) mountIndexKVEntry(tableInfo *model.TableInfo, idx *indexKV
 			return nil, errors.Trace(err)
 		}
 		whereHandle := true
-		preCols[idxCol.Name.O] = &model.Column{
+		col := &model.Column{
 			Type:        tableInfo.Columns[idxCol.Offset].Tp,
 			WhereHandle: &whereHandle,
 			Value:       value,
 			Flag:        transColumnFlag(tableInfo.Columns[idxCol.Offset]),
 		}
+		col.Flag.SetIsHandleKey()
+		preCols[idxCol.Name.O] = col
 	}
 	return &model.RowChangedEvent{
 		StartTs:  idx.StartTs,
@@ -664,55 +696,14 @@ func genMultipleKeys(ti *timodel.TableInfo, preCols, cols map[string]*model.Colu
 	return multipleKeys
 }
 
-func columnValue(value interface{}) string {
-	var data string
-	switch v := value.(type) {
-	case nil:
-		data = "null"
-	case bool:
-		if v {
-			data = "1"
-		} else {
-			data = "0"
-		}
-	case int:
-		data = strconv.FormatInt(int64(v), 10)
-	case int8:
-		data = strconv.FormatInt(int64(v), 10)
-	case int16:
-		data = strconv.FormatInt(int64(v), 10)
-	case int32:
-		data = strconv.FormatInt(int64(v), 10)
-	case int64:
-		data = strconv.FormatInt(int64(v), 10)
-	case uint8:
-		data = strconv.FormatUint(uint64(v), 10)
-	case uint16:
-		data = strconv.FormatUint(uint64(v), 10)
-	case uint32:
-		data = strconv.FormatUint(uint64(v), 10)
-	case uint64:
-		data = strconv.FormatUint(uint64(v), 10)
-	case float32:
-		data = strconv.FormatFloat(float64(v), 'f', -1, 32)
-	case float64:
-		data = strconv.FormatFloat(float64(v), 'f', -1, 64)
-	case string:
-		data = v
-	case []byte:
-		data = string(v)
-	default:
-		data = fmt.Sprintf("%v", v)
-	}
-
-	return data
-}
-
 func transColumnFlag(col *timodel.ColumnInfo) model.ColumnFlagType {
 	var flag model.ColumnFlagType
 	if col.Charset == "binary" {
 		flag.SetIsBinary()
 	}
+	if col.IsGenerated() {
+		flag.SetIsGeneratedColumn()
+	}
 	return flag
 }
 
@@ -724,7 +715,7 @@ func genKeyList(table string, columns []*timodel.ColumnInfo, values map[string]*
 			log.L().Debug("ignore null value", zap.String("column", col.Name.O), zap.String("table", table))
 			continue // ignore `null` value.
 		}
-		buf.WriteString(columnValue(val.Value))
+		buf.WriteString(model.ColumnValueString(val.Value))
 	}
 	if buf.Len() == 0 {
 		log.L().Debug("all value are nil, no key generated", zap.String("table", table))

cdc/model/schema_storage.go

@@ -21,6 +21,13 @@ import (
 	"github.com/pingcap/tidb/util/rowcodec"
 )
 
+const (
+	// HandleIndexPKIsHandle represents the handle index is pk and the pk is handle
+	HandleIndexPKIsHandle = -1
+	// HandleIndexTableIneligible represents the table is ineligible
+	HandleIndexTableIneligible = -2
+)
+
 // TableInfo provides meta data describing a DB table.
 type TableInfo struct {
 	*model.TableInfo
@@ -30,7 +37,15 @@ type TableInfo struct {
 	columnsOffset    map[int64]int
 	indicesOffset    map[int64]int
 	uniqueColumns    map[int64]struct{}
-	handleColID      int64
+
+	// only for new row format decoder
+	handleColID int64
+
+	// the mounter will chose this index to output delete events
+	// special value:
+	// HandleIndexPKIsHandle(-1) : pk is handle
+	// HandleIndexTableIneligible(-2) : the table is not eligible
+	HandleIndexID int64
 
 	// if the table of this row only has one unique index(includes primary key),
 	// IndieMarkCol will be set to the name of the unique index
@@ -49,6 +64,7 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 		indicesOffset:    make(map[int64]int, len(info.Indices)),
 		uniqueColumns:    make(map[int64]struct{}),
 		handleColID:      -1,
+		HandleIndexID:    HandleIndexTableIneligible,
 		rowColInfos:      make([]rowcodec.ColInfo, len(info.Columns)),
 	}
 
@@ -59,6 +75,7 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 		isPK := (ti.PKIsHandle && mysql.HasPriKeyFlag(col.Flag)) || col.ID == model.ExtraHandleID
 		if isPK {
 			ti.handleColID = col.ID
+			ti.HandleIndexID = HandleIndexPKIsHandle
 			ti.uniqueColumns[col.ID] = struct{}{}
 			uniqueIndexNum++
 		}
@@ -90,10 +107,39 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 			}
 		}
 	}
-
+	ti.findHandleIndex()
 	return ti
 }
 
+func (ti *TableInfo) findHandleIndex() {
+	if ti.HandleIndexID == HandleIndexPKIsHandle {
+		// pk is handle
+		return
+	}
+	handleIndexOffset := -1
+	for i, idx := range ti.Indices {
+		if !ti.IsIndexUnique(idx) {
+			continue
+		}
+		if idx.Primary {
+			handleIndexOffset = i
+			break
+		}
+		if handleIndexOffset < 0 {
+			handleIndexOffset = i
+		} else {
+			if len(ti.Indices[handleIndexOffset].Columns) > len(ti.Indices[i].Columns) ||
+				(len(ti.Indices[handleIndexOffset].Columns) == len(ti.Indices[i].Columns) &&
+					ti.Indices[handleIndexOffset].ID > ti.Indices[i].ID) {
+				handleIndexOffset = i
+			}
+		}
+	}
+	if handleIndexOffset >= 0 {
+		ti.HandleIndexID = ti.Indices[handleIndexOffset].ID
+	}
+}
+
 // GetColumnInfo returns the column info by ID
 func (ti *TableInfo) GetColumnInfo(colID int64) (info *model.ColumnInfo, exist bool) {
 	colOffset, exist := ti.columnsOffset[colID]
@@ -121,9 +167,13 @@ func (ti *TableInfo) GetRowColInfos() (int64, []rowcodec.ColInfo) {
 	return ti.handleColID, ti.rowColInfos
 }
 
-// IsColWritable returns is the col is writeable
-func (ti *TableInfo) IsColWritable(col *model.ColumnInfo) bool {
-	return col.State == model.StatePublic && !col.IsGenerated()
+// IsColCDCVisible returns is the col is visible for CDC
+func (ti *TableInfo) IsColCDCVisible(col *model.ColumnInfo) bool {
+	// this column is a virtual generated column
+	if col.IsGenerated() && !col.GeneratedStored {
+		return false
+	}
+	return col.State == model.StatePublic
 }
 
 // GetUniqueKeys returns all unique keys of the table as a slice of column names
@@ -180,7 +230,12 @@ func (ti *TableInfo) IsIndexUnique(indexInfo *model.IndexInfo) bool {
 	}
 	if indexInfo.Unique {
 		for _, col := range indexInfo.Columns {
-			if !mysql.HasNotNullFlag(ti.Columns[col.Offset].Flag) {
+			colInfo := ti.Columns[col.Offset]
+			if !mysql.HasNotNullFlag(colInfo.Flag) {
+				return false
+			}
+			// this column is a virtual generated column
+			if colInfo.IsGenerated() && !colInfo.GeneratedStored {
 				return false
 			}
 		}