fix: converter lose precision

internal/converter/converter.go

@@ -33,7 +33,7 @@ import (
 
 func init() {
 	modules.RegisterConverter(message.FormatJson, func(_ api.StreamContext, _ string, schema map[string]*ast.JsonStreamField, props map[string]any) (message.Converter, error) {
-		return json.NewFastJsonConverter(schema), nil
+		return json.NewFastJsonConverter(schema, props), nil
 	})
 	modules.RegisterConverter(message.FormatXML, func(ctx api.StreamContext, schemaId string, logicalSchema map[string]*ast.JsonStreamField, props map[string]any) (message.Converter, error) {
 		return xml.NewXMLConverter(), nil

internal/converter/json/convert_bench_test.go

@@ -15,9 +15,11 @@
 package json
 
 import (
+	"encoding/json"
 	"os"
 	"testing"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
@@ -106,9 +108,41 @@ func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.Json
 	if err != nil {
 		b.Fatal(err)
 	}
-	f := NewFastJsonConverter(schema)
+	f := NewFastJsonConverter(schema, nil)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		f.Decode(ctx, payload)
 	}
 }
+
+func BenchmarkNativeFloatParse(b *testing.B) {
+	conf.IsTesting = false
+	m := make(map[string]interface{})
+	data := `{"id":1.2}`
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		json.Unmarshal([]byte(data), &m)
+	}
+}
+
+func BenchmarkFloatParse(b *testing.B) {
+	conf.IsTesting = false
+	ctx := mockContext.NewMockContext("test", "test")
+	f := NewFastJsonConverter(nil, map[string]any{"use_int64": true})
+	data := `{"id":1.2}`
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		f.Decode(ctx, []byte(data))
+	}
+}
+
+func BenchmarkIntParse(b *testing.B) {
+	conf.IsTesting = false
+	ctx := mockContext.NewMockContext("test", "test")
+	f := NewFastJsonConverter(nil, map[string]any{"use_int64": true})
+	data := `{"id":1}`
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		f.Decode(ctx, []byte(data))
+	}
+}

internal/converter/json/converter.go

@@ -17,6 +17,7 @@
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
@@ -29,16 +30,31 @@
 
 type FastJsonConverter struct {
 	sync.RWMutex
-	schema map[string]*ast.JsonStreamField
+	schema   map[string]*ast.JsonStreamField
+	useInt64 bool
 }
 
-func NewFastJsonConverter(schema map[string]*ast.JsonStreamField) *FastJsonConverter {
+func NewFastJsonConverter(schema map[string]*ast.JsonStreamField, props map[string]any) *FastJsonConverter {
 	f := &FastJsonConverter{
 		schema: schema,
 	}
+	f.setupProps(props)
 	return f
 }
 
+func (f *FastJsonConverter) setupProps(props map[string]any) {
+	if props == nil {
+		return
+	}
+	v, ok := props["use_int64"]
+	if ok {
+		b, ok := v.(bool)
+		if ok {
+			f.useInt64 = b
+		}
+	}
+}
+
 func (f *FastJsonConverter) ResetSchema(schema map[string]*ast.JsonStreamField) {
 	f.Lock()
 	defer f.Unlock()
@@ -57,11 +73,6 @@
 	}()
 	f.RLock()
 	defer f.RUnlock()
-	if f.schema == nil {
-		var r any
-		err = json.Unmarshal(b, &r)
-		return r, err
-	}
 	return f.decodeWithSchema(b, f.schema)
 }
 
@@ -85,7 +96,7 @@
 		case fastjson.TypeString:
 			return vv.String(), nil
 		case fastjson.TypeNumber:
-			return vv.Float64()
+			return f.extractNumber(vv)
 		case fastjson.TypeTrue, fastjson.TypeFalse:
 			return vv.Bool()
 		}
@@ -340,11 +351,7 @@
 
 func (f *FastJsonConverter) extractNumberValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
 	if field == nil {
-		f64, err := v.Float64()
-		if err != nil {
-			return nil, err
-		}
-		return f64, nil
+		return f.extractNumber(v)
 	}
 	switch {
 	case field.Type == "float", field.Type == "datetime":
@@ -454,3 +461,29 @@
 		return t.Type
 	}
 }
+
+func (f *FastJsonConverter) extractNumber(v *fastjson.Value) (any, error) {
+	if f.useInt64 {
+		if isFloat64(v) {
+			f64, err := v.Float64()
+			if err != nil {
+				return nil, err
+			}
+			return f64, nil
+		}
+		i64, err := v.Int64()
+		if err != nil {
+			return nil, err
+		}
+		return i64, nil
+	}
+	f64, err := v.Float64()
+	if err != nil {
+		return nil, err
+	}
+	return f64, nil
+}
+
+func isFloat64(v *fastjson.Value) bool {
+	return strings.Contains(v.String(), ".")
+}

internal/converter/json/converter_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
@@ -162,7 +163,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter(tc.schema)
+		f := NewFastJsonConverter(tc.schema, nil)
 		v, err := f.Decode(ctx, tc.payload)
 		require.NoError(t, err)
 		require.Equal(t, v, tc.require)
@@ -173,7 +174,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter(tc.schema)
+		f := NewFastJsonConverter(tc.schema, nil)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -333,7 +334,7 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter(tc.schema)
+		f := NewFastJsonConverter(tc.schema, nil)
 		_, err := f.Decode(ctx, tc.payload)
 		require.Error(t, err)
 		require.Equal(t, err.Error(), tc.err.Error())
@@ -344,7 +345,7 @@ func TestFastJsonEncode(t *testing.T) {
 	a := make(map[string]int)
 	a["a"] = 1
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter(nil)
+	f := NewFastJsonConverter(nil, nil)
 	v, err := f.Encode(ctx, a)
 	require.NoError(t, err)
 	require.Equal(t, v, []byte(`{"a":1}`))
@@ -377,7 +378,7 @@ func TestArrayWithArray(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter(schema)
+	f := NewFastJsonConverter(schema, nil)
 	v, err := f.Decode(ctx, payload)
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -558,7 +559,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter(tc.schema)
+		f := NewFastJsonConverter(tc.schema, nil)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -568,7 +569,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter(tc.schema)
+		f := NewFastJsonConverter(tc.schema, nil)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -585,7 +586,7 @@ func TestConvertBytea(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter(schema)
+	f := NewFastJsonConverter(schema, nil)
 	v, err := f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -601,7 +602,7 @@ func TestConvertBytea(t *testing.T) {
 			},
 		},
 	}
-	f = NewFastJsonConverter(schema)
+	f = NewFastJsonConverter(schema, nil)
 	v, err = f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -613,7 +614,7 @@ func TestSchemaless(t *testing.T) {
 	originSchema := map[string]*ast.JsonStreamField{
 		"a": nil,
 	}
-	f := NewFastJsonConverter(originSchema)
+	f := NewFastJsonConverter(originSchema, nil)
 	testcases := []struct {
 		data   map[string]interface{}
 		expect map[string]interface{}
@@ -664,7 +665,7 @@ func TestIssue(t *testing.T) {
 	originSchema := map[string]*ast.JsonStreamField{
 		"results": nil,
 	}
-	f := NewFastJsonConverter(originSchema)
+	f := NewFastJsonConverter(originSchema, nil)
 	data := `{
     "results": [
         {
@@ -704,7 +705,7 @@ func TestIssue(t *testing.T) {
 	schmema2 := map[string]*ast.JsonStreamField{
 		"others": nil,
 	}
-	f2 := NewFastJsonConverter(schmema2)
+	f2 := NewFastJsonConverter(schmema2, nil)
 	m, err = f2.Decode(context.Background(), []byte(data))
 	require.NoError(t, err)
 	require.Len(t, m, 0)
@@ -754,7 +755,7 @@ func TestDecodeField(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter(nil)
+	f := NewFastJsonConverter(nil, nil)
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
 			field, err := f.DecodeField(ctx, tc.payload, "id")
@@ -767,3 +768,21 @@ func TestDecodeField(t *testing.T) {
 		})
 	}
 }
+
+func TestIssue3441(t *testing.T) {
+	conf.IsTesting = false
+	originSchema := map[string]*ast.JsonStreamField{
+		"id": nil,
+	}
+	f := NewFastJsonConverter(originSchema, map[string]any{"use_int64": true})
+	data := `{"id":1795292668348461056}`
+	ctx := mockContext.NewMockContext("test", "op1")
+	m, err := f.Decode(ctx, []byte(data))
+	require.NoError(t, err)
+	require.Equal(t, map[string]interface{}{"id": int64(1795292668348461056)}, m)
+
+	data = `{"id":17952926683484.44}`
+	m, err = f.Decode(ctx, []byte(data))
+	require.NoError(t, err)
+	require.Equal(t, map[string]interface{}{"id": 17952926683484.44}, m)
+}

internal/topo/node/conf/source.go

@@ -85,6 +85,7 @@ func GetSourceConf(sourceType string, options *ast.Options) map[string]interface
 	props["strictValidation"] = options.STRICT_VALIDATION
 	props["timestamp"] = options.TIMESTAMP
 	props["timestampFormat"] = options.TIMESTAMP_FORMAT
+	props["use_int64"] = options.USE_INT64
 	conf.Log.Infof("get conf for %s with conf key %s: %v", sourceType, confkey, printable(props))
 	return props
 }

internal/topo/node/conf/source_test.go

@@ -68,6 +68,7 @@ func TestGetSourceConf(t *testing.T) {
 				"strictValidation":   false,
 				"timestamp":          "",
 				"timestampFormat":    "",
+				"use_int64":          false,
 			},
 		},
 		{
@@ -93,6 +94,7 @@ func TestGetSourceConf(t *testing.T) {
 				"strictValidation":   false,
 				"timestamp":          "",
 				"timestampFormat":    "",
+				"use_int64":          false,
 			},
 		},
 		{
@@ -120,6 +122,7 @@ func TestGetSourceConf(t *testing.T) {
 				"timestampFormat":    "",
 				"connectionSelector": "test11",
 				"a":                  1,
+				"use_int64":          false,
 			},
 		},
 	}

internal/topo/node/decode_op_test.go

@@ -69,9 +69,9 @@ func TestJSON(t *testing.T) {
 					&xsql.Tuple{Emitter: "test", Message: map[string]interface{}{"a": 3.0, "b": 4.0, "sourceConf": "hello"}, Timestamp: time.UnixMilli(111), Metadata: map[string]any{"topic": "demo", "qos": 1}},
 				},
 				{errors.New("go through error")},
-				{errors.New("invalid character ':' after top-level value")},
-				{errors.New("only map[string]any inside a list is supported but got: hello")},
-				{errors.New("unsupported decode result: hello")},
+				{errors.New(`unexpected tail: ":1,\"b\":2},{\"a\":3,\"b\":4,\"sourceConf\":\"hello\"}]"`)},
+				{errors.New("value doesn't contain object; it contains string")},
+				{errors.New("only map[string]interface{} and []map[string]interface{} is supported")},
 				{errors.New("unsupported data received: invalid")},
 			}
 			timex.Add(2 * time.Second)

internal/topo/node/window_inc_agg_op_test.go

@@ -298,8 +298,9 @@ func TestIncAggTumblingWindow(t *testing.T) {
 	errCh := make(chan error, 10)
 	ctx, cancel := mockContext.NewMockContext("1", "2").WithCancel()
 	op.Exec(ctx, errCh)
-	time.Sleep(10 * time.Millisecond)
+	waitExecute()
 	input <- &xsql.Tuple{Message: map[string]any{"a": int64(1)}}
+	waitExecute()
 	timex.Add(1100 * time.Millisecond)
 	got := <-output
 	wt, ok := got.(*xsql.WindowTuples)

internal/xsql/parser.go

@@ -1551,6 +1551,12 @@
 						case ast.KIND:
 							val := strings.ToLower(lit3)
 							opts.KIND = val
+						case ast.USE_INT64:
+							if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
+								return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, lit1)
+							} else {
+								opts.USE_INT64 = val == "TRUE"
+							}
 						default:
 							f := v.Elem().FieldByName(lit1)
 							if f.IsValid() {
@@ -1573,7 +1579,7 @@
 				}
 				return nil, fmt.Errorf("Parenthesis is not matched in options definition.")
 			} else {
-				return nil, fmt.Errorf("found %q, unknown option keys(DATASOURCE|FORMAT|KEY|CONF_KEY|SHARED|STRICT_VALIDATION|TYPE|TIMESTAMP|TIMESTAMP_FORMAT|RETAIN_SIZE|SCHEMAID).", lit1)
+				return nil, fmt.Errorf("found %q, unknown option keys(DATASOURCE|FORMAT|KEY|CONF_KEY|SHARED|STRICT_VALIDATION|TYPE|TIMESTAMP|TIMESTAMP_FORMAT|RETAIN_SIZE|SCHEMAID|USE_INT64).", lit1)
 			}
 		}
 	} else {