drainer/* Improve performance of dest-type=kafka (#423) (#428)

note binlog.String() may be pretty slow
only 1k+ op/second of the benchmark

Author: WangXiangUSTC

drainer/executor/bench_kafka_test.go

@@ -8,6 +8,48 @@ import (
 	obinlog "github.com/pingcap/tidb-tools/tidb-binlog/slave_binlog_proto/go-binlog"
 )
 
+var bytes = make([]byte, 5*(1<<10))
+var table = &obinlog.Table{
+	SchemaName: proto.String("test"),
+	TableName:  proto.String("test"),
+	ColumnInfo: []*obinlog.ColumnInfo{
+		&obinlog.ColumnInfo{Name: "id", MysqlType: "int"},
+		&obinlog.ColumnInfo{Name: "a1", MysqlType: "blob"},
+	},
+	Mutations: []*obinlog.TableMutation{
+		&obinlog.TableMutation{
+			Type: obinlog.MutationType_Insert.Enum(),
+			Row: &obinlog.Row{
+				Columns: []*obinlog.Column{
+					&obinlog.Column{
+						Int64Value: proto.Int64(1),
+					},
+					&obinlog.Column{
+						BytesValue: bytes,
+					},
+				},
+			},
+		},
+	},
+}
+
+// with bytes = 5KB
+// BenchmarkBinlogMarshal-4          100000            573941 ns/op
+// means only 1742 op/second
+func BenchmarkBinlogMarshal(b *testing.B) {
+	binlog := &obinlog.Binlog{
+		DmlData: &obinlog.DMLData{
+			Tables: []*obinlog.Table{table},
+		},
+	}
+	for i := 0; i < b.N; i++ {
+		binlog.String()
+	}
+}
+
+// with bytes = 5KB
+// BenchmarkKafka-4         1000000             42384 ns/op
+// means 23593 op/second
 func BenchmarkKafka(b *testing.B) {
 	log.SetLevelByString("error")
 
@@ -22,31 +64,6 @@ func BenchmarkKafka(b *testing.B) {
 		b.Fatal(err)
 	}
 
-	bytes := make([]byte, 128)
-	table := &obinlog.Table{
-		SchemaName: proto.String("test"),
-		TableName:  proto.String("test"),
-		ColumnInfo: []*obinlog.ColumnInfo{
-			{Name: "id", MysqlType: "int"},
-			{Name: "a1", MysqlType: "blob"},
-		},
-		Mutations: []*obinlog.TableMutation{
-			{
-				Type: obinlog.MutationType_Insert.Enum(),
-				Row: &obinlog.Row{
-					Columns: []*obinlog.Column{
-						{
-							Int64Value: proto.Int64(1),
-						},
-						{
-							BytesValue: bytes,
-						},
-					},
-				},
-			},
-		},
-	}
-
 	b.ResetTimer()
 	var arg = []interface{}{table}
 	var args = [][]interface{}{arg}

drainer/executor/kafka.go

@@ -139,7 +139,7 @@ func (p *kafkaExecutor) Close() error {
 }
 
 func (p *kafkaExecutor) saveBinlog(binlog *obinlog.Binlog) error {
-	log.Debug("save binlog: ", binlog.String())
+	// log.Debug("save binlog: ", binlog.String())
 	data, err := binlog.Marshal()
 	if err != nil {
 		return errors.Trace(err)

drainer/translator/kafka.go

@@ -148,7 +148,7 @@ func insertRowToRow(tableInfo *model.TableInfo, raw []byte) (row *obinlog.Row, e
 		return
 	}
 
-	log.Debugf("decodeRow: %+v\n", columnValues)
+	// log.Debugf("decodeRow: %+v\n", columnValues)
 	// maybe only the pk column value
 	if columnValues == nil {
 		columnValues = make(map[int64]types.Datum)