txn: bump checksum version (#57205) (#57219)

close #57206

pkg/ddl/column_type_change_test.go

@@ -260,7 +260,7 @@ func TestRowFormatWithChecksums(t *testing.T) {
 	data, err := h.GetMvccByEncodedKey(encodedKey)
 	require.NoError(t, err)
 	// row value with checksums
-	expected := []byte{0x80, 0x2, 0x3, 0x0, 0x0, 0x0, 0x1, 0x2, 0x3, 0x1, 0x0, 0x4, 0x0, 0x7, 0x0, 0x1, 0x31, 0x32, 0x33, 0x31, 0x32, 0x33, 0x1, 0x5b, 0x6a, 0x78, 0x7c}
+	expected := []byte{0x80, 0x2, 0x3, 0x0, 0x0, 0x0, 0x1, 0x2, 0x3, 0x1, 0x0, 0x4, 0x0, 0x7, 0x0, 0x1, 0x31, 0x32, 0x33, 0x31, 0x32, 0x33, 0x2, 0x9e, 0x56, 0xf5, 0x45}
 	require.Equal(t, expected, data.Info.Writes[0].ShortValue)
 	tk.MustExec("drop table if exists t")
 }
@@ -284,7 +284,7 @@ func TestRowLevelChecksumWithMultiSchemaChange(t *testing.T) {
 	data, err := h.GetMvccByEncodedKey(encodedKey)
 	require.NoError(t, err)
 	// checksum skipped and with a null col vv
-	expected := []byte{0x80, 0x2, 0x3, 0x0, 0x1, 0x0, 0x1, 0x2, 0x4, 0x3, 0x1, 0x0, 0x4, 0x0, 0x7, 0x0, 0x1, 0x31, 0x32, 0x33, 0x31, 0x32, 0x33, 0x1, 0xc5, 0x73, 0x5f, 0x1f}
+	expected := []byte{0x80, 0x2, 0x3, 0x0, 0x1, 0x0, 0x1, 0x2, 0x4, 0x3, 0x1, 0x0, 0x4, 0x0, 0x7, 0x0, 0x1, 0x31, 0x32, 0x33, 0x31, 0x32, 0x33, 0x2, 0x0, 0x4f, 0xd2, 0x26}
 	require.Equal(t, expected, data.Info.Writes[0].ShortValue)
 	tk.MustExec("drop table if exists t")
 }

pkg/util/rowcodec/encoder.go

@@ -239,7 +239,14 @@ func (NoChecksum) encode(encoder *Encoder, buf []byte) ([]byte, error) {
 	return encoder.toBytes(buf), nil
 }
 
-const checksumVersionRaw byte = 1
+// introduced since v7.1.0
+const checksumVersionColumn byte = 0
+
+// introduced since v8.3.0
+const checksumVersionRawKey byte = 1
+
+// introduced since v8.4.0
+const checksumVersionRawHandle byte = 2
 
 // RawChecksum indicates encode the raw bytes checksum and append it to the raw bytes.
 type RawChecksum struct {
@@ -248,9 +255,9 @@ type RawChecksum struct {
 
 func (c RawChecksum) encode(encoder *Encoder, buf []byte) ([]byte, error) {
 	encoder.flags |= rowFlagChecksum
-	encoder.checksumHeader &^= checksumFlagExtra   // revert extra checksum flag
-	encoder.checksumHeader &^= checksumMaskVersion // revert checksum version
-	encoder.checksumHeader |= checksumVersionRaw   // set checksum version
+	encoder.checksumHeader &^= checksumFlagExtra       // revert extra checksum flag
+	encoder.checksumHeader &^= checksumMaskVersion     // revert checksum version
+	encoder.checksumHeader |= checksumVersionRawHandle // set checksum version
 	valueBytes := encoder.toBytes(buf)
 	valueBytes = append(valueBytes, encoder.checksumHeader)
 	encoder.checksum1 = crc32.Checksum(valueBytes, crc32.IEEETable)

pkg/util/rowcodec/row.go

@@ -152,7 +152,9 @@ func (r *row) fromBytes(rowData []byte) error {
 		r.checksumHeader = rowData[cursor]
 		checksumVersion := r.ChecksumVersion()
 		// make sure it can be read previous version checksum to support backward compatibility.
-		if checksumVersion != 0 && checksumVersion != 1 {
+		switch checksumVersion {
+		case 0, 1, 2:
+		default:
 			return errInvalidChecksumVer
 		}
 		cursor++
@@ -303,12 +305,8 @@ func (r *row) initOffsets32() {
 // CalculateRawChecksum calculates the bytes-level checksum by using the given elements.
 // this is mainly used by the TiCDC to implement E2E checksum functionality.
 func (r *row) CalculateRawChecksum(
-	loc *time.Location, colIDs []int64, values []*types.Datum, handle kv.Handle, buf []byte,
+	loc *time.Location, colIDs []int64, values []*types.Datum, key kv.Key, handle kv.Handle, buf []byte,
 ) (uint32, error) {
-	r.flags |= rowFlagChecksum
-	r.checksumHeader &^= checksumFlagExtra   // revert extra checksum flag
-	r.checksumHeader &^= checksumMaskVersion // revert checksum version
-	r.checksumHeader |= checksumVersionRaw   // set checksum version
 	for idx, colID := range colIDs {
 		data, err := encodeValueDatum(loc, values[idx], nil)
 		if err != nil {
@@ -325,6 +323,11 @@ func (r *row) CalculateRawChecksum(
 	buf = r.toBytes(buf)
 	buf = append(buf, r.checksumHeader)
 	rawChecksum := crc32.Checksum(buf, crc32.IEEETable)
-	rawChecksum = crc32.Update(rawChecksum, crc32.IEEETable, handle.Encoded())
+	// keep backward compatibility to v8.3.0
+	if r.ChecksumVersion() == int(checksumVersionRawKey) {
+		rawChecksum = crc32.Update(rawChecksum, crc32.IEEETable, key)
+	} else {
+		rawChecksum = crc32.Update(rawChecksum, crc32.IEEETable, handle.Encoded())
+	}
 	return rawChecksum, nil
 }

pkg/util/rowcodec/rowcodec_test.go

@@ -1226,7 +1226,7 @@ func TestEncodeDecodeRowWithChecksum(t *testing.T) {
 	require.Equal(t, expected, checksum)
 
 	version := dec.ChecksumVersion()
-	require.Equal(t, 1, version)
+	require.Equal(t, 2, version)
 }
 
 var (