batch float with trailing zero and add a useFloatWithTrailingZero flag to enable

Author: chenhuiyeh

go.mod

@@ -5,6 +5,7 @@ go 1.22
 toolchain go1.23.1
 
 require (
+	filippo.io/edwards25519 v1.1.0
 	github.com/BurntSushi/toml v1.3.2
 	github.com/Masterminds/semver v1.5.0
 	github.com/go-sql-driver/mysql v1.7.1
@@ -19,7 +20,6 @@ require (
 )
 
 require (
-	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect

replication/binlogsyncer.go

@@ -76,6 +76,9 @@ type BinlogSyncerConfig struct {
 	// Use decimal.Decimal structure for decimals.
 	UseDecimal bool
 
+	// Use FloatWithTrailingZero structure for floats.
+	UseFloatWithTrailingZero bool
+
 	// RecvBufferSize sets the size in bytes of the operating system's receive buffer associated with the connection.
 	RecvBufferSize int
 
@@ -197,6 +200,7 @@ func NewBinlogSyncer(cfg BinlogSyncerConfig) *BinlogSyncer {
 	b.parser.SetParseTime(b.cfg.ParseTime)
 	b.parser.SetTimestampStringLocation(b.cfg.TimestampStringLocation)
 	b.parser.SetUseDecimal(b.cfg.UseDecimal)
+	b.parser.SetUseFloatWithTrailingZero(b.cfg.UseFloatWithTrailingZero)
 	b.parser.SetVerifyChecksum(b.cfg.VerifyChecksum)
 	b.parser.SetRowsEventDecodeFunc(b.cfg.RowsEventDecodeFunc)
 	b.parser.SetTableMapOptionalMetaDecodeFunc(b.cfg.TableMapOptionalMetaDecodeFunc)

replication/json_binary.go

@@ -3,6 +3,7 @@ package replication
 import (
 	"fmt"
 	"math"
+	"strconv"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/utils"
@@ -52,6 +53,8 @@ type (
 	JsonDiffOperation byte
 )
 
+type FloatWithTrailingZero float64
+
 const (
 	// The JSON value in the given path is replaced with a new value.
 	//
@@ -96,6 +99,14 @@ func (jd *JsonDiff) String() string {
 	return fmt.Sprintf("json_diff(op:%s path:%s value:%s)", jd.Op, jd.Path, jd.Value)
 }
 
+func (f FloatWithTrailingZero) MarshalJSON() ([]byte, error) {
+	if float64(f) == float64(int(f)) {
+		return []byte(strconv.FormatFloat(float64(f), 'f', 1, 64)), nil
+	}
+
+	return []byte(strconv.FormatFloat(float64(f), 'f', -1, 64)), nil
+}
+
 func jsonbGetOffsetSize(isSmall bool) int {
 	if isSmall {
 		return jsonbSmallOffsetSize
@@ -125,6 +136,7 @@ func jsonbGetValueEntrySize(isSmall bool) int {
 func (e *RowsEvent) decodeJsonBinary(data []byte) ([]byte, error) {
 	d := jsonBinaryDecoder{
 		useDecimal:      e.useDecimal,
+		useFloatWithTrailingZero: e.useFloatWithTrailingZero,
 		ignoreDecodeErr: e.ignoreJSONDecodeErr,
 	}
 
@@ -141,9 +153,10 @@ func (e *RowsEvent) decodeJsonBinary(data []byte) ([]byte, error) {
 }
 
 type jsonBinaryDecoder struct {
-	useDecimal      bool
-	ignoreDecodeErr bool
-	err             error
+	useDecimal      			bool
+	useFloatWithTrailingZero 	bool
+	ignoreDecodeErr 			bool
+	err             			error
 }
 
 func (d *jsonBinaryDecoder) decodeValue(tp byte, data []byte) interface{} {
@@ -175,6 +188,9 @@ func (d *jsonBinaryDecoder) decodeValue(tp byte, data []byte) interface{} {
 	case JSONB_UINT64:
 		return d.decodeUint64(data)
 	case JSONB_DOUBLE:
+		if d.useFloatWithTrailingZero {
+			return d.decodeDoubleWithTrailingZero(data)
+		}
 		return d.decodeDouble(data)
 	case JSONB_STRING:
 		return d.decodeString(data)
@@ -395,6 +411,11 @@ func (d *jsonBinaryDecoder) decodeDouble(data []byte) float64 {
 	return v
 }
 
+func (d *jsonBinaryDecoder) decodeDoubleWithTrailingZero(data []byte) FloatWithTrailingZero {
+	v := d.decodeDouble(data)
+	return FloatWithTrailingZero(v)
+}
+
 func (d *jsonBinaryDecoder) decodeString(data []byte) string {
 	if d.err != nil {
 		return ""

replication/parser.go

@@ -36,6 +36,7 @@ type BinlogParser struct {
 	stopProcessing uint32
 
 	useDecimal          bool
+	useFloatWithTrailingZero bool
 	ignoreJSONDecodeErr bool
 	verifyChecksum      bool
 
@@ -202,6 +203,10 @@ func (p *BinlogParser) SetUseDecimal(useDecimal bool) {
 	p.useDecimal = useDecimal
 }
 
+func (p *BinlogParser) SetUseFloatWithTrailingZero(useFloatWithTrailingZero bool) {
+	p.useFloatWithTrailingZero = useFloatWithTrailingZero
+}
+
 func (p *BinlogParser) SetIgnoreJSONDecodeError(ignoreJSONDecodeErr bool) {
 	p.ignoreJSONDecodeErr = ignoreJSONDecodeErr
 }
@@ -410,6 +415,7 @@ func (p *BinlogParser) newRowsEvent(h *EventHeader) *RowsEvent {
 	e.parseTime = p.parseTime
 	e.timestampStringLocation = p.timestampStringLocation
 	e.useDecimal = p.useDecimal
+	e.useFloatWithTrailingZero = p.useFloatWithTrailingZero
 	e.ignoreJSONDecodeErr = p.ignoreJSONDecodeErr
 
 	switch h.EventType {

replication/row_event.go

@@ -945,10 +945,11 @@ type RowsEvent struct {
 	Rows           [][]interface{}
 	SkippedColumns [][]int
 
-	parseTime               bool
-	timestampStringLocation *time.Location
-	useDecimal              bool
-	ignoreJSONDecodeErr     bool
+	parseTime               	bool
+	timestampStringLocation 	*time.Location
+	useDecimal              	bool
+	useFloatWithTrailingZero 	bool
+	ignoreJSONDecodeErr     	bool
 }
 
 // EnumRowsEventType is an abridged type describing the operation which triggered the given RowsEvent.