mounter(ticdc): decode bytes-level checksum and encode columns-level checksum (#10706)

cdc/entry/mounter.go

@@ -20,6 +20,7 @@
 	"encoding/json"
 	"fmt"
 	"math"
+	"reflect"
 	"sort"
 	"time"
 	"unsafe"
@@ -29,7 +30,6 @@
 	"github.com/pingcap/tidb/pkg/kv"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
-	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/pkg/table"
 	"github.com/pingcap/tidb/pkg/tablecodec"
 	"github.com/pingcap/tidb/pkg/types"
@@ -100,11 +100,6 @@
 	// they should not be nil after decode at least one event in the row format v2.
 	decoder    *rowcodec.DatumMapDecoder
 	preDecoder *rowcodec.DatumMapDecoder
-
-	// encoder is used to calculate the checksum.
-	encoder *rowcodec.Encoder
-	// sctx hold some information can be used by the encoder to calculate the checksum.
-	sctx *stmtctx.StatementContext
 }
 
 // NewMounter creates a mounter
@@ -124,9 +119,6 @@
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		tz:        tz,
 		integrity: integrity,
-
-		encoder: &rowcodec.Encoder{},
-		sctx:    stmtctx.NewStmtCtxWithTimeZone(tz),
 	}
 }
 
@@ -156,6 +148,8 @@
 	if !bytes.HasPrefix(raw.Key, tablePrefix) {
 		return nil, nil
 	}
+	// checksumKey is only used to calculate raw checksum if necessary.
+	checksumKey := raw.Key
 	key, physicalTableID, err := decodeTableID(raw.Key)
 	if err != nil {
 		return nil, err
@@ -207,7 +201,7 @@
 			if rowKV == nil {
 				return nil, nil
 			}
-			row, rawRow, err := m.mountRowKVEntry(tableInfo, rowKV, raw.ApproximateDataSize())
+			row, rawRow, err := m.mountRowKVEntry(tableInfo, rowKV, checksumKey, raw.ApproximateDataSize())
 			if err != nil {
 				return nil, err
 			}
@@ -297,7 +291,6 @@
 	if err != nil {
 		return nil, false, errors.Trace(err)
 	}
-
 	datums, err = tablecodec.DecodeHandleToDatumMap(
 		recordID, handleColIDs, handleColFt, m.tz, datums)
 	if err != nil {
@@ -455,8 +448,8 @@
 	return cols, rawCols, columnInfos, nil
 }
 
-func (m *mounter) calculateChecksum(
-	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum,
+func calculateColumnChecksum(
+	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum, tz *time.Location,
 ) (uint32, error) {
 	columns := make([]rowcodec.ColData, 0, len(rawColumns))
 	for idx, col := range columnInfos {
@@ -475,37 +468,24 @@
 		Data: make([]byte, 0),
 	}
 
-	checksum, err := calculator.Checksum(m.tz)
+	checksum, err := calculator.Checksum(tz)
 	if err != nil {
 		return 0, errors.Trace(err)
 	}
 	return checksum, nil
 }
 
-// return error when calculate the checksum failed.
-// return false if the checksum is not matched
-func (m *mounter) verifyChecksum(
-	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum, isPreRow bool,
+func verifyColumnChecksum(
+	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum, decoder *rowcodec.DatumMapDecoder, tz *time.Location,
 ) (uint32, bool, error) {
-	if !m.integrity.Enabled() {
-		return 0, true, nil
-	}
-
-	var decoder *rowcodec.DatumMapDecoder
-	if isPreRow {
-		decoder = m.preDecoder
-	} else {
-		decoder = m.decoder
-	}
-
 	// if the checksum cannot be found, which means the upstream TiDB checksum is not enabled,
 	// so return matched as true to skip check the event.
 	first, ok := decoder.GetChecksum()
 	if !ok {
 		return 0, true, nil
 	}
 
-	checksum, err := m.calculateChecksum(columnInfos, rawColumns)
+	checksum, err := calculateColumnChecksum(columnInfos, rawColumns, tz)
 	if err != nil {
 		log.Error("failed to calculate the checksum", zap.Uint32("first", first), zap.Error(err))
 		return 0, false, err
@@ -540,7 +520,168 @@
 	return checksum, false, nil
 }
 
-func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, dataSize int64) (*model.RowChangedEvent, model.RowChangedDatums, error) {
+// todo: do we really need this? how about the datum.ConvertTo ?
+func newDatum(value interface{}, ft types.FieldType) (types.Datum, error) {
+	if value == nil {
+		return types.NewDatum(nil), nil
+	}
+	switch ft.GetType() {
+	case mysql.TypeTiny, mysql.TypeShort, mysql.TypeLong, mysql.TypeLonglong, mysql.TypeInt24, mysql.TypeYear:
+		switch v := value.(type) {
+		case uint64:
+			return types.NewUintDatum(v), nil
+		case int64:
+			return types.NewIntDatum(v), nil
+		}
+	case mysql.TypeDate, mysql.TypeDatetime, mysql.TypeNewDate, mysql.TypeTimestamp:
+		// todo: DefaultStmtNoWarningContext timezone is set to UTC, is it correct?
+		t, err := types.ParseTime(types.DefaultStmtNoWarningContext, value.(string), ft.GetType(), ft.GetDecimal())
+		if err != nil {
+			return types.Datum{}, errors.Trace(err)
+		}
+		return types.NewTimeDatum(t), nil
+	case mysql.TypeDuration:
+		d, _, err := types.ParseDuration(types.StrictContext, value.(string), ft.GetDecimal())
+		if err != nil {
+			return types.Datum{}, errors.Trace(err)
+		}
+		return types.NewDurationDatum(d), nil
+	case mysql.TypeJSON:
+		bj, err := types.ParseBinaryJSONFromString(value.(string))
+		if err != nil {
+			return types.Datum{}, errors.Trace(err)
+		}
+		return types.NewJSONDatum(bj), nil
+	case mysql.TypeNewDecimal:
+		mysqlDecimal := new(types.MyDecimal)
+		err := mysqlDecimal.FromString([]byte(value.(string)))
+		if err != nil {
+			return types.Datum{}, errors.Trace(err)
+		}
+		datum := types.NewDecimalDatum(mysqlDecimal)
+		datum.SetLength(ft.GetFlen())
+		datum.SetFrac(ft.GetDecimal())
+		return datum, nil
+	case mysql.TypeEnum:
+		enum, err := types.ParseEnumValue(ft.GetElems(), value.(uint64))
+		if err != nil {
+			return types.Datum{}, errors.Trace(err)
+		}
+		return types.NewMysqlEnumDatum(enum), nil
+	case mysql.TypeSet:
+		set, err := types.ParseSetValue(ft.GetElems(), value.(uint64))
+		if err != nil {
+			return types.Datum{}, errors.Trace(err)
+		}
+		return types.NewMysqlSetDatum(set, ft.GetCollate()), nil
+	case mysql.TypeBit:
+		byteSize := (ft.GetFlen() + 7) >> 3
+		binaryLiteral := types.NewBinaryLiteralFromUint(value.(uint64), byteSize)
+		return types.NewMysqlBitDatum(binaryLiteral), nil
+	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar,
+		mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob:
+		switch v := value.(type) {
+		case []byte:
+			return types.NewBytesDatum(v), nil
+		case string:
+			return types.NewBytesDatum([]byte(v)), nil
+		}
+		log.Panic("unknown data type when build datum",
+			zap.Any("type", ft.GetType()), zap.Any("value", value), zap.Reflect("type", reflect.TypeOf(value)))
+	case mysql.TypeFloat:
+		return types.NewFloat32Datum(value.(float32)), nil
+	case mysql.TypeDouble:
+		return types.NewFloat64Datum(value.(float64)), nil
+	default:
+		log.Panic("unexpected mysql type found", zap.Any("type", ft.GetType()))
+	}
+	return types.Datum{}, nil
+}
+
+func verifyRawBytesChecksum(
+	tableInfo *model.TableInfo, columns []*model.ColumnData, decoder *rowcodec.DatumMapDecoder,
+	key kv.Key, tz *time.Location,
+) (uint32, bool, error) {
+	expected, ok := decoder.GetChecksum()
+	if !ok {
+		return 0, true, nil
+	}
+	var (
+		columnIDs []int64
+		datums    []*types.Datum
+	)
+	for _, col := range columns {
+		// TiDB does not encode null value into the bytes, so just ignore it.
+		if col.Value == nil {
+			continue
+		}
+		columnID := col.ColumnID
+		columnInfo := tableInfo.ForceGetColumnInfo(columnID)
+		datum, err := newDatum(col.Value, columnInfo.FieldType)
+		if err != nil {
+			return 0, false, errors.Trace(err)
+		}
+		datums = append(datums, &datum)
+		columnIDs = append(columnIDs, columnID)
+	}
+	obtained, err := decoder.CalculateRawChecksum(tz, columnIDs, datums, key, nil)
+	if err != nil {
+		return 0, false, errors.Trace(err)
+	}
+	if obtained == expected {
+		return expected, true, nil
+	}
+
+	log.Error("raw bytes checksum mismatch",
+		zap.Uint32("expected", expected), zap.Uint32("obtained", obtained))
+
+	return expected, false, nil
+}
+
+// return error when calculate the checksum.
+// return false if the checksum is not matched
+func (m *mounter) verifyChecksum(
+	tableInfo *model.TableInfo, columnInfos []*timodel.ColumnInfo,
+	columns []*model.ColumnData, rawColumns []types.Datum,
+	key kv.Key, isPreRow bool,
+) (uint32, bool, error) {
+	if !m.integrity.Enabled() {
+		return 0, true, nil
+	}
+
+	var decoder *rowcodec.DatumMapDecoder
+	if isPreRow {
+		decoder = m.preDecoder
+	} else {
+		decoder = m.decoder
+	}
+
+	version := decoder.ChecksumVersion()
+	switch version {
+	case 0:
+		return verifyColumnChecksum(columnInfos, rawColumns, decoder, m.tz)
+	case 1:
+		expected, matched, err := verifyRawBytesChecksum(tableInfo, columns, decoder, key, m.tz)
+		if err != nil {
+			return 0, false, errors.Trace(err)
+		}
+		if !matched {
+			return expected, matched, err
+		}
+		columnChecksum, err := calculateColumnChecksum(columnInfos, rawColumns, m.tz)
+		if err != nil {
+			log.Error("failed to calculate column-level checksum, after raw checksum verification passed", zap.Error(err))
+			return 0, false, errors.Trace(err)
+		}
+		return columnChecksum, true, nil
+	default:
+	}
+	return 0, false, errors.Errorf("unknown checksum version %d", version)
+}
+
+func (m *mounter) mountRowKVEntry(
+	tableInfo *model.TableInfo, row *rowKVEntry, key kv.Key, dataSize int64,
+) (*model.RowChangedEvent, model.RowChangedDatums, error) {
 	var (
 		rawRow      model.RowChangedDatums
 		columnInfos []*timodel.ColumnInfo
@@ -573,7 +714,7 @@
 			return nil, rawRow, errors.Trace(err)
 		}
 
-		preChecksum, matched, err = m.verifyChecksum(columnInfos, preRawCols, true)
+		preChecksum, matched, err = m.verifyChecksum(tableInfo, columnInfos, preCols, preRawCols, key, true)
 		if err != nil {
 			log.Error("calculate the previous columns checksum failed",
 				zap.Any("tableInfo", tableInfo),
@@ -605,7 +746,7 @@
 			return nil, rawRow, errors.Trace(err)
 		}
 
-		currentChecksum, matched, err = m.verifyChecksum(columnInfos, rawCols, false)
+		currentChecksum, matched, err = m.verifyChecksum(tableInfo, columnInfos, cols, rawCols, key, false)
 		if err != nil {
 			log.Error("calculate the current columns checksum failed",
 				zap.Any("tableInfo", tableInfo),