Fix numeric

flow/connectors/postgres/qvalue_convert.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math"
 	"math/big"
 	"strings"
 	"time"
@@ -19,6 +18,8 @@ import (
 	geom "github.com/twpayne/go-geos"
 )
 
+var big10 = big.NewInt(10)
+
 func postgresOIDToQValueKind(recvOID uint32) qvalue.QValueKind {
 	switch recvOID {
 	case pgtype.BoolOID:
@@ -380,13 +381,14 @@ func numericToRat(numVal *pgtype.Numeric) (*big.Rat, error) {
 		case pgtype.NegativeInfinity, pgtype.Infinity:
 			return nil, errors.New("numeric value is infinity")
 		}
-
-		rat := new(big.Rat)
-
-		rat.SetInt(numVal.Int)
-		divisor := new(big.Rat).SetFloat64(math.Pow10(int(-numVal.Exp)))
-		rat.Quo(rat, divisor)
-
+		rat := new(big.Rat).SetInt(numVal.Int)
+		if numVal.Exp > 0 {
+			mul := new(big.Int).Exp(big10, big.NewInt(int64(numVal.Exp)), nil)
+			rat.Mul(rat, new(big.Rat).SetInt(mul))
+		} else if numVal.Exp < 0 {
+			mul := new(big.Int).Exp(big10, big.NewInt(int64(-numVal.Exp)), nil)
+			rat.Quo(rat, new(big.Rat).SetInt(mul))
+		}
 		return rat, nil
 	}
 

flow/model/model.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/PeerDB-io/peer-flow/generated/protos"
+	"github.com/PeerDB-io/peer-flow/model/numeric"
 	"github.com/PeerDB-io/peer-flow/model/qvalue"
 )
 
@@ -149,7 +150,7 @@ func (r *RecordItems) toMap() (map[string]interface{}, error) {
 			if !ok {
 				return nil, errors.New("expected *big.Rat value")
 			}
-			jsonStruct[col] = bigRat.FloatString(9)
+			jsonStruct[col] = bigRat.FloatString(numeric.PeerDBNumericScale)
 		default:
 			jsonStruct[col] = v.Value
 		}

flow/model/numeric/scale.go

@@ -0,0 +1,11 @@
+package numeric
+
+import "strings"
+
+const PeerDBNumericScale = 9
+
+func StripTrailingZeros(value string) string {
+	value = strings.TrimRight(value, "0")
+	value = strings.TrimSuffix(value, ".")
+	return value
+}

flow/model/qvalue/avro_converter.go

@@ -5,6 +5,7 @@ import (
 	"math/big"
 	"time"
 
+	"github.com/PeerDB-io/peer-flow/model/numeric"
 	"github.com/google/uuid"
 	"github.com/linkedin/goavro/v2"
 	log "github.com/sirupsen/logrus"
@@ -256,6 +257,8 @@ func (c *QValueAvroConverter) processNumeric() (interface{}, error) {
 		return nil, fmt.Errorf("invalid Numeric value: expected *big.Rat, got %T", c.Value.Value)
 	}
 
+	decimalValue := num.FloatString(numeric.PeerDBNumericScale)
+	num.SetString(decimalValue)
 	if c.Nullable {
 		return goavro.Union("bytes.decimal", num), nil
 	}

flow/model/qvalue/qvalue.go

@@ -7,8 +7,10 @@ import (
 	"math/big"
 	"reflect"
 	"strconv"
+	"strings"
 	"time"
 
+	"github.com/PeerDB-io/peer-flow/model/numeric"
 	"github.com/google/uuid"
 )
 
@@ -188,9 +190,24 @@ func compareNumeric(value1, value2 interface{}) bool {
 		return false
 	}
 
-	// check if the difference is less than 1e-9
-	diff := new(big.Rat).Sub(rat1, rat2)
-	return diff.Abs(diff).Cmp(big.NewRat(1, 1000000000)) < 0
+	str1 := numeric.StripTrailingZeros(rat1.FloatString(10))
+	str2 := numeric.StripTrailingZeros(rat2.FloatString(10))
+	if len(str1) > 2 && len(str2) > 2 {
+		decimals1 := len(str1) - strings.Index(str1, ".") - 1
+		decimals2 := len(str2) - strings.Index(str2, ".") - 1
+		// for cases like 1.12345 or 1.12345678, check exact equality
+		if decimals1 <= 9 && decimals2 <= 9 {
+			return str1 == str2
+		} else {
+			// check if the difference is less than 1e-9
+			diff := new(big.Rat).Sub(rat1, rat2)
+			if diff.Cmp(big.NewRat(1, 1000000000)) < 0 {
+				return true
+			}
+		}
+	}
+
+	return str1 == str2
 }
 
 func compareString(value1, value2 interface{}) bool {