mounter(ticdc): reuse buf to reduce memory allocation when checksum enabled

cdc/entry/mounter.go

@@ -94,14 +94,12 @@
 	metricTotalRows              prometheus.Gauge
 	metricIgnoredDMLEventCounter prometheus.Counter
 
-	integrity *integrity.Config
+	verifier *checksumVerifier
 
 	// decoder and preDecoder are used to decode the raw value, also used to extract checksum,
 	// they should not be nil after decode at least one event in the row format v2.
 	decoder    *rowcodec.DatumMapDecoder
 	preDecoder *rowcodec.DatumMapDecoder
-
-	lastSkipOldValueTime time.Time
 }
 
 // NewMounter creates a mounter
@@ -119,8 +117,8 @@
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		metricIgnoredDMLEventCounter: ignoredDMLEventCounter.
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
-		tz:        tz,
-		integrity: integrity,
+		tz:       tz,
+		verifier: newChecksumVerifier(changefeedID, integrity, tz),
 	}
 }
 
@@ -450,74 +448,76 @@
 	return cols, rawCols, columnInfos, nil
 }
 
-func calculateColumnChecksum(
-	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum, tz *time.Location,
+func (v *checksumVerifier) calculateColumnChecksum(
+	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum,
 ) (uint32, error) {
-	columns := make([]rowcodec.ColData, 0, len(rawColumns))
+	v.buf = v.buf[:0]
+	v.columns = v.columns[:0]
 	for idx, col := range columnInfos {
 		column := rowcodec.ColData{
 			ColumnInfo: col,
 			Datum:      &rawColumns[idx],
 		}
-		columns = append(columns, column)
+		v.columns = append(v.columns, column)
 	}
-	sort.Slice(columns, func(i, j int) bool {
-		return columns[i].ID < columns[j].ID
+	sort.Slice(v.columns, func(i, j int) bool {
+		return v.columns[i].ID < v.columns[j].ID
 	})
 
 	calculator := rowcodec.RowData{
-		Cols: columns,
-		Data: make([]byte, 0),
+		Cols: v.columns,
+		Data: v.buf,
 	}
 
-	checksum, err := calculator.Checksum(tz)
-	if err != nil {
-		return 0, errors.Trace(err)
-	}
-	return checksum, nil
+	checksum, err := calculator.Checksum(v.tz)
+	return checksum, errors.Trace(err)
 }
 
-func (m *mounter) verifyColumnChecksum(
+// return error if calculate checksum failed, this should not happen.
+// return true if the checksum matched.
+// return the matched checksum.
+func (v *checksumVerifier) verifyColumnChecksum(
 	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum,
 	decoder *rowcodec.DatumMapDecoder, skipFail bool,
-) (uint32, bool, error) {
+) (checksum uint32, corrupted bool, err error) {
 	// 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
+		return 0, false, nil
 	}
 
-	checksum, err := calculateColumnChecksum(columnInfos, rawColumns, m.tz)
+	checksum, err = v.calculateColumnChecksum(columnInfos, rawColumns)
 	if err != nil {
-		log.Error("failed to calculate the checksum", zap.Uint32("first", first), zap.Error(err))
-		return 0, false, err
+		log.Error("failed to calculate the checksum",
+			zap.Uint32("first", first), zap.Uint32("checksum", checksum), zap.Error(err))
+		return checksum, false, err
 	}
 
 	// the first checksum matched, it hits in the most case.
 	if checksum == first {
 		log.Debug("checksum matched", zap.Uint32("checksum", checksum), zap.Uint32("first", first))
-		return checksum, true, nil
+		return checksum, false, nil
 	}
 
 	extra, ok := decoder.GetExtraChecksum()
 	if ok && checksum == extra {
 		log.Debug("extra checksum matched, this may happen the upstream TiDB is during the DDL execution phase",
 			zap.Uint32("checksum", checksum), zap.Uint32("extra", extra))
-		return checksum, true, nil
+		return checksum, false, nil
 	}
 
 	if !skipFail {
-		log.Error("cannot found the extra checksum, the first checksum mismatched",
+		log.Error("verify the previous checksum v1 failed",
 			zap.Uint32("checksum", checksum), zap.Uint32("first", first), zap.Uint32("extra", extra))
 		return checksum, false, nil
 	}
 
-	if time.Since(m.lastSkipOldValueTime) > time.Minute {
+	if time.Since(v.lastSkipOldValueTime) > time.Minute {
 		log.Warn("checksum mismatch on the old value, "+
 			"this may caused by Add Column / Drop Column executed, skip verification",
 			zap.Uint32("checksum", checksum), zap.Uint32("first", first), zap.Uint32("extra", extra))
-		m.lastSkipOldValueTime = time.Now()
+		v.lastSkipOldValueTime = time.Now()
 	}
 	return checksum, true, nil
 }
@@ -598,18 +598,41 @@
 	return types.Datum{}, nil
 }
 
-func verifyRawBytesChecksum(
+type checksumVerifier struct {
+	changefeed model.ChangeFeedID
+	integrity  *integrity.Config
+
+	tz      *time.Location
+	buf     []byte
+	columns []rowcodec.ColData
+
+	// used for the checksum v2.
+	datums    []*types.Datum
+	columnIDs []int64
+
+	lastSkipOldValueTime time.Time
+}
+
+func newChecksumVerifier(changefeed model.ChangeFeedID, integrity *integrity.Config, tz *time.Location) *checksumVerifier {
+	return &checksumVerifier{
+		changefeed: changefeed,
+		integrity:  integrity,
+		tz:         tz,
+		buf:        make([]byte, 0, 1024),
+	}
+}
+
+func (v *checksumVerifier) verifyRawBytesChecksum(
 	tableInfo *model.TableInfo, columns []*model.ColumnData, decoder *rowcodec.DatumMapDecoder,
-	key kv.Key, tz *time.Location,
-) (uint32, bool, error) {
+	key kv.Key,
+) (corrupted bool, err error) {
 	expected, ok := decoder.GetChecksum()
 	if !ok {
-		return 0, true, nil
+		return false, nil
 	}
-	var (
-		columnIDs []int64
-		datums    []*types.Datum
-	)
+	v.buf = v.buf[:0]
+	v.datums = v.datums[:0]
+	v.columnIDs = v.columnIDs[:0]
 	for _, col := range columns {
 		// TiDB does not encode null value into the bytes, so just ignore it.
 		if col.Value == nil {
@@ -619,41 +642,33 @@
 		columnInfo := tableInfo.ForceGetColumnInfo(columnID)
 		datum, err := newDatum(col.Value, columnInfo.FieldType)
 		if err != nil {
-			return 0, false, errors.Trace(err)
+			return false, errors.Trace(err)
 		}
-		datums = append(datums, &datum)
-		columnIDs = append(columnIDs, columnID)
+		v.datums = append(v.datums, &datum)
+		v.columnIDs = append(v.columnIDs, columnID)
 	}
-	obtained, err := decoder.CalculateRawChecksum(tz, columnIDs, datums, key, nil)
+	obtained, err := decoder.CalculateRawChecksum(v.tz, v.columnIDs, v.datums, key, v.buf)
 	if err != nil {
-		return 0, false, errors.Trace(err)
+		return false, errors.Trace(err)
 	}
 	if obtained == expected {
-		return expected, true, nil
+		return false, nil
 	}
-
 	log.Error("raw bytes checksum mismatch",
 		zap.Uint32("expected", expected), zap.Uint32("obtained", obtained))
 
-	return expected, false, nil
+	return true, nil
 }
 
 // return error when calculate the checksum.
 // return false if the checksum is not matched
-func (m *mounter) verifyChecksum(
+func (v *checksumVerifier) 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
+	key kv.Key, decoder *rowcodec.DatumMapDecoder, isPreRow bool,
+) (checksum uint32, corrupted bool, err error) {
+	if !v.integrity.Enabled() {
+		return 0, false, nil
 	}
 
 	version := decoder.ChecksumVersion()
@@ -662,24 +677,37 @@
 		// skip old value checksum verification for the checksum v1, since it cannot handle
 		// Update / Delete event correctly, after Add Column / Drop column DDL,
 		// since the table schema does not contain complete column information.
-		return m.verifyColumnChecksum(columnInfos, rawColumns, decoder, isPreRow)
-	case 1:
-		expected, matched, err := verifyRawBytesChecksum(tableInfo, columns, decoder, key, m.tz)
+		checksum, corrupted, err = v.verifyColumnChecksum(columnInfos, rawColumns, decoder, isPreRow)
 		if err != nil {
 			return 0, false, errors.Trace(err)
 		}
-		if !matched {
-			return expected, matched, err
-		}
-		columnChecksum, err := calculateColumnChecksum(columnInfos, rawColumns, m.tz)
+	case 1:
+		corrupted, err = v.verifyRawBytesChecksum(tableInfo, columns, decoder, key)
 		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
+		if !corrupted {
+			checksum, err = v.calculateColumnChecksum(columnInfos, rawColumns)
+			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)
+			}
+		}
 	default:
+		return 0, false, errors.Errorf("unknown checksum version %d", version)
+	}
+
+	if corrupted {
+		log.Error("columns checksum mismatch",
+			zap.Uint32("checksum", checksum),
+			zap.Any("columnInfos", columnInfos),
+			zap.Any("rawColumns", rawColumns))
+		if v.integrity.ErrorHandle() {
+			return 0, true, cerror.ErrCorruptedDataMutation.
+				GenWithStackByArgs(v.changefeed.Namespace, v.changefeed.ID)
+		}
 	}
-	return 0, false, errors.Errorf("unknown checksum version %d", version)
+	return checksum, corrupted, nil
 }
 
 func (m *mounter) mountRowKVEntry(
@@ -688,10 +716,8 @@
 	var (
 		rawRow      model.RowChangedDatums
 		columnInfos []*timodel.ColumnInfo
-		matched     bool
 		err         error
-
-		checksum *integrity.Checksum
+		checksum    *integrity.Checksum
 
 		checksumVersion int
 		corrupted       bool
@@ -717,56 +743,34 @@
 			return nil, rawRow, errors.Trace(err)
 		}
 
-		preChecksum, matched, err = m.verifyChecksum(tableInfo, columnInfos, preCols, preRawCols, key, true)
+		preChecksum, corrupted, err = m.verifier.verifyChecksum(tableInfo, columnInfos, preCols, preRawCols, key, m.preDecoder, true)
 		if err != nil {
-			log.Error("calculate the previous columns checksum failed",
+			log.Error("verify the previous columns checksum failed",
 				zap.Any("tableInfo", tableInfo),
 				zap.Any("rawCols", preRawCols))
 			return nil, rawRow, errors.Trace(err)
 		}
-
-		if !matched {
-			log.Error("previous columns checksum mismatch",
-				zap.Uint32("checksum", preChecksum),
-				zap.Any("tableInfo", tableInfo),
-				zap.Any("rawCols", preRawCols))
-			if m.integrity.ErrorHandle() {
-				return nil, rawRow, cerror.ErrCorruptedDataMutation.
-					GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID)
-			}
-			corrupted = true
-		}
 	}
 
 	var (
-		cols            []*model.ColumnData
-		rawCols         []types.Datum
-		currentChecksum uint32
+		cols             []*model.ColumnData
+		rawCols          []types.Datum
+		currentChecksum  uint32
+		currentCorrupted bool
 	)
 	if row.RowExist {
 		cols, rawCols, columnInfos, err = datum2Column(tableInfo, row.Row, m.tz)
 		if err != nil {
 			return nil, rawRow, errors.Trace(err)
 		}
-
-		currentChecksum, matched, err = m.verifyChecksum(tableInfo, columnInfos, cols, rawCols, key, false)
+		currentChecksum, currentCorrupted, err = m.verifier.verifyChecksum(tableInfo, columnInfos, cols, rawCols, key, m.decoder, false)
 		if err != nil {
-			log.Error("calculate the current columns checksum failed",
+			log.Error("verify the current columns checksum failed",
 				zap.Any("tableInfo", tableInfo),
 				zap.Any("rawCols", rawCols))
 			return nil, rawRow, errors.Trace(err)
 		}
-		if !matched {
-			log.Error("current columns checksum mismatch",
-				zap.Uint32("checksum", currentChecksum),
-				zap.Any("tableInfo", tableInfo),
-				zap.Any("rawCols", rawCols))
-			if m.integrity.ErrorHandle() {
-				return nil, rawRow, cerror.ErrCorruptedDataMutation.
-					GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID)
-			}
-			corrupted = true
-		}
+		corrupted = corrupted || currentCorrupted
 	}
 
 	var intRowID int64

cdc/owner/ddl_sink.go

@@ -238,7 +238,8 @@ func (s *ddlSinkImpl) writeDDLEvent(ctx context.Context, ddl *model.DDLEvent) er
 	log.Info("begin emit ddl event",
 		zap.String("namespace", s.changefeedID.Namespace),
 		zap.String("changefeed", s.changefeedID.ID),
-		zap.Any("DDL", ddl))
+		zap.Uint64("commitTs", ddl.CommitTs),
+		zap.String("DDL", ddl.Query))
 
 	doWrite := func() (err error) {
 		if err = s.makeSinkReady(ctx); err == nil {
@@ -251,14 +252,16 @@ func (s *ddlSinkImpl) writeDDLEvent(ctx context.Context, ddl *model.DDLEvent) er
 			log.Error("Execute DDL failed",
 				zap.String("namespace", s.changefeedID.Namespace),
 				zap.String("changefeed", s.changefeedID.ID),
-				zap.Any("DDL", ddl),
+				zap.Uint64("commitTs", ddl.CommitTs),
+				zap.String("DDL", ddl.Query),
 				zap.Error(err))
 		} else {
 			ddl.Done.Store(true)
 			log.Info("Execute DDL succeeded",
 				zap.String("namespace", s.changefeedID.Namespace),
 				zap.String("changefeed", s.changefeedID.ID),
-				zap.Any("DDL", ddl))
+				zap.Uint64("commitTs", ddl.CommitTs),
+				zap.String("DDL", ddl.Query))
 		}
 		return
 	}

cdc/puller/ddl_puller.go

@@ -329,8 +329,10 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 			zap.String("table", job.TableName),
 			zap.Uint64("startTs", job.StartTS),
 			zap.Uint64("finishedTs", job.BinlogInfo.FinishedTS),
-			zap.String("query", job.Query),
-			zap.Uint64("pullerResolvedTs", p.getResolvedTs()))
+			zap.Uint64("pullerResolvedTs", p.getResolvedTs()),
+			zap.Int64("jobSchemaVersion", job.BinlogInfo.SchemaVersion),
+			zap.Int64("schemaVersion", p.schemaVersion),
+			zap.String("query", job.Query))
 		return true, nil
 	}