From eb1465bffa95cee7bd3efa4f26c0031a4b1f5aeb Mon Sep 17 00:00:00 2001
From: Fritz Larco <flarco@live.com>
Date: Thu, 18 Jan 2024 22:43:24 -0300
Subject: [PATCH] improve clickhouse decimal logic [nt]

---
 database/database.go                           |  2 +-
 database/database_clickhouse.go                | 10 ++++++++++
 database/templates/types_general_to_native.tsv |  2 +-
 iop/stream_processor.go                        |  2 +-
 4 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/database/database.go b/database/database.go
index 85ba116..3a46ec6 100755
--- a/database/database.go
+++ b/database/database.go
@@ -2289,7 +2289,7 @@ func (conn *BaseConn) GetNativeType(col iop.Column) (nativeType string, err erro
 				fmt.Sprintf("(%d)", length),
 			)
 		}
-	} else if strings.HasSuffix(nativeType, "(,)") {
+	} else if strings.Contains(nativeType, "(,)") {
 
 		scale := lo.Ternary(col.DbScale < ddlMinDecScale, ddlMinDecScale, col.DbScale)
 		scale = lo.Ternary(scale < col.Stats.MaxDecLen, col.Stats.MaxDecLen, scale)
diff --git a/database/database_clickhouse.go b/database/database_clickhouse.go
index 4af71d8..34070ac 100755
--- a/database/database_clickhouse.go
+++ b/database/database_clickhouse.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/flarco/dbio"
 	"github.com/flarco/dbio/iop"
+	"github.com/shopspring/decimal"
 	"github.com/spf13/cast"
 
 	"github.com/flarco/g"
@@ -146,6 +147,15 @@ func (conn *ClickhouseConn) BulkImportStream(tableFName string, ds *iop.Datastre
 			}
 
 			for row := range batch.Rows {
+				// set decimals correctly
+				for i, col := range batch.Columns {
+					if col.Type.IsDecimal() {
+						if val, err := decimal.NewFromString(cast.ToString(row[i])); err == nil {
+							row[i] = val
+						}
+					}
+				}
+
 				count++
 				// Do insert
 				ds.Context.Lock()
diff --git a/database/templates/types_general_to_native.tsv b/database/templates/types_general_to_native.tsv
index 0c37a11..a20aab8 100755
--- a/database/templates/types_general_to_native.tsv
+++ b/database/templates/types_general_to_native.tsv
@@ -4,7 +4,7 @@ binary	varbinary()	bytes	varbinary	varbinary	varbinary	varbinary	varchar(65535)
 bool	varchar(5)	bool	char(5)	varchar(5)	varchar(5)	varchar(5)	bool	boolean	boolean	bool	Nullable(String)	bool	bool
 date	timestamp(9)	timestamp	datetime(6)	datetime2	datetime2	datetime2	timestamp	timestamp	text	timestamp	Nullable(Date)	date	date
 datetime	timestamp(9)	timestamp	datetime(6)	datetime2	datetime2	datetime2	timestamp	timestamp	text	timestamp	Nullable(DateTime)	datetime	datetime
-decimal	number(,)	numeric	decimal(,)	decimal(,)	decimal(,)	decimal(,)	decimal(,)	decimal(,)	real	numeric	Nullable(Float64)	decimal(,)	decimal(,)
+decimal	number(,)	numeric	decimal(,)	decimal(,)	decimal(,)	decimal(,)	decimal(,)	decimal(,)	real	numeric	Nullable(Decimal(,))	decimal(,)	decimal(,)
 integer	number(10)	integer	integer	integer	integer	integer	integer	integer	integer	int64	Nullable(Int64)	integer	integer
 json	clob	jsonb	json	nvarchar(max)	nvarchar(65535)	nvarchar(max)	varchar(65535)	variant	text	json	Nullable(String)	json	json
 smallint	number(5)	smallint	smallint	smallint	smallint	smallint	smallint	smallint	integer	int64	Nullable(Int32)	smallint	smallint
diff --git a/iop/stream_processor.go b/iop/stream_processor.go
index 6b9f71d..ba83e08 100644
--- a/iop/stream_processor.go
+++ b/iop/stream_processor.go
@@ -560,7 +560,7 @@ func (sp *StreamProcessor) CastVal(i int, val interface{}, col *Column) interfac
 
 		// max 9 decimals for bigquery compatibility
 		if sp.config.MaxDecimals > -1 {
-			nVal = math.Round(fVal*sp.config.MaxDecimals) / sp.config.MaxDecimals
+			nVal = cast.ToString(math.Round(fVal*sp.config.MaxDecimals) / sp.config.MaxDecimals)
 		} else {
 			nVal = val // use string to keep accuracy
 		}