From 248770e4b6fe857492cd5c94e86d9f905a076360 Mon Sep 17 00:00:00 2001
From: xianganjie <xianganjie@jd.com>
Date: Tue, 20 Apr 2021 13:11:33 +0800
Subject: [PATCH] decimal data type(#313)

---
 pkg/mysql/utils.go |  3 +++
 pkg/utils/db.go    | 18 ++++++++++--------
 2 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/pkg/mysql/utils.go b/pkg/mysql/utils.go
index 8d4ecb8b..e2921cbf 100644
--- a/pkg/mysql/utils.go
+++ b/pkg/mysql/utils.go
@@ -2,6 +2,7 @@ package mysql
 
 import (
 	"database/sql"
+	"github.com/shopspring/decimal"
 	"reflect"
 	"time"
 
@@ -106,6 +107,8 @@ func SQLDataPtrs2Val(dataPtrs []interface{}, columnTypes []*sql.ColumnType) map[
 			} else {
 				ret[columnName] = v.Time
 			}
+		case decimal.Decimal:
+			ret[columnName] = v
 		default:
 			log.Fatalf("failed to catch columnName: %v, type: %v", columnName, reflect.TypeOf(columnData))
 		}
diff --git a/pkg/utils/db.go b/pkg/utils/db.go
index f59d3e66..75a94756 100644
--- a/pkg/utils/db.go
+++ b/pkg/utils/db.go
@@ -3,6 +3,7 @@ package utils
 import (
 	"database/sql"
 	"fmt"
+	"github.com/shopspring/decimal"
 	"math"
 	"math/rand"
 	"net/url"
@@ -149,7 +150,7 @@ func GetScanType(columnType *sql.ColumnType) reflect.Type {
 	if IsColumnString(columnType) {
 		return reflect.TypeOf(sql.NullString{})
 	} else if IsColumnFloat(columnType) {
-		return reflect.TypeOf(sql.NullFloat64{})
+		return reflect.TypeOf(decimal.Decimal{})
 	} else {
 		return columnType.ScanType()
 	}
@@ -610,13 +611,14 @@ func SQLWithAnnotation(annotation string, sql string) string {
 
 func NewBinlogSyncer(serverID uint32, dbConfig *config.DBConfig, heartbeatPeriod time.Duration) *replication.BinlogSyncer {
 	syncerConfig := replication.BinlogSyncerConfig{
-		ServerID:  serverID,
-		Flavor:    "mysql",
-		Host:      dbConfig.Host,
-		Port:      uint16(dbConfig.Port),
-		User:      dbConfig.Username,
-		Password:  dbConfig.Password,
-		ParseTime: true,
+		ServerID:   serverID,
+		Flavor:     "mysql",
+		Host:       dbConfig.Host,
+		Port:       uint16(dbConfig.Port),
+		User:       dbConfig.Username,
+		Password:   dbConfig.Password,
+		ParseTime:  true,
+		UseDecimal: true,
 	}
 
 	if heartbeatPeriod > 0 {