encoder (CDC) shrink the bytes.Buffer to reduce memory usage

cdc/model/schema_storage.go

@@ -281,8 +281,8 @@ func IsColCDCVisible(col *model.ColumnInfo) bool {
 	return true
 }
 
-// ExistTableUniqueColumn returns whether the table has a unique column
-func (ti *TableInfo) ExistTableUniqueColumn() bool {
+// HasUniqueColumn returns whether the table has a unique column
+func (ti *TableInfo) HasUniqueColumn() bool {
 	return ti.hasUniqueColumn
 }
 
@@ -299,7 +299,7 @@ func (ti *TableInfo) IsEligible(forceReplicate bool) bool {
 	if ti.IsView() {
 		return true
 	}
-	return ti.ExistTableUniqueColumn()
+	return ti.HasUniqueColumn()
 }
 
 // IsIndexUnique returns whether the index is unique

cdc/model/schema_storage_test.go

@@ -190,7 +190,7 @@ func TestTableInfoGetterFuncs(t *testing.T) {
 	require.Equal(t, 3, len(fts))
 	require.Equal(t, 3, len(colInfos))
 
-	require.True(t, info.ExistTableUniqueColumn())
+	require.True(t, info.HasUniqueColumn())
 
 	// check IsEligible
 	require.True(t, info.IsEligible(false))

pkg/sink/codec/canal/canal_json_txn_event_encoder.go

@@ -84,7 +84,11 @@ func (j *JSONTxnEventEncoder) Build() []*common.Message {
 		j.valueBuf.Bytes(), j.txnCommitTs, model.MessageTypeRow, j.txnSchema, j.txnTable)
 	ret.SetRowsCount(j.batchSize)
 	ret.Callback = j.callback
-	j.valueBuf.Reset()
+	if j.valueBuf.Cap() > codec.MemBufShrinkThreshold {
+		j.valueBuf = &bytes.Buffer{}
+	} else {
+		j.valueBuf.Reset()
+	}
 	j.callback = nil
 	j.batchSize = 0
 	j.txnCommitTs = 0

pkg/sink/codec/csv/csv_encoder.go

@@ -57,7 +57,11 @@ func (b *BatchEncoder) Build() (messages []*common.Message) {
 		b.valueBuf.Bytes(), 0, model.MessageTypeRow, nil, nil)
 	ret.SetRowsCount(b.batchSize)
 	ret.Callback = b.callback
-	b.valueBuf.Reset()
+	if b.valueBuf.Cap() > codec.MemBufShrinkThreshold {
+		b.valueBuf = &bytes.Buffer{}
+	} else {
+		b.valueBuf.Reset()
+	}
 	b.callback = nil
 	b.batchSize = 0
 

pkg/sink/codec/encoder.go

@@ -24,6 +24,9 @@ import (
 const (
 	// BatchVersion1 represents the version of batch format
 	BatchVersion1 uint64 = 1
+
+	// MemBufShrinkThreshold represents the threshold of shrinking the buffer.
+	MemBufShrinkThreshold = 1024 * 1024
 )
 
 // DDLEventBatchEncoder is an abstraction for DDL event encoder.