fix: converter lose precision

internal/converter/json/converter.go

@@ -17,11 +17,13 @@ package json
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/valyala/fastjson"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
@@ -57,11 +59,6 @@ func (f *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err
 	}()
 	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 +82,7 @@ func (f *FastJsonConverter) DecodeField(_ api.StreamContext, b []byte, field str
 		case fastjson.TypeString:
 			return vv.String(), nil
 		case fastjson.TypeNumber:
-			return vv.Float64()
+			return extractNumber(vv)
 		case fastjson.TypeTrue, fastjson.TypeFalse:
 			return vv.Bool()
 		}
@@ -340,11 +337,7 @@ func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.
 
 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 extractNumber(v)
 	}
 	switch {
 	case field.Type == "float", field.Type == "datetime":
@@ -454,3 +447,22 @@ func getType(t *ast.JsonStreamField) string {
 		return t.Type
 	}
 }
+
+func extractNumber(v *fastjson.Value) (any, error) {
+	if !isFloat64(v.String()) && !conf.IsTesting {
+		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 string) bool {
+	return strings.Contains(v, ".")
+}

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"
@@ -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)
+	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/io/memory/lookupsource_test.go

@@ -131,6 +131,7 @@ func TestUpdateLookup(t *testing.T) {
 }
 
 func TestLookup(t *testing.T) {
+	conf.IsTesting = true
 	contextLogger := conf.Log.WithField("rule", "test2")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	ls := GetLookupSource().(api.LookupSource)

internal/topo/node/decode_op_test.go

@@ -24,6 +24,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/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/ast"
@@ -392,6 +393,7 @@ func TestPayloadDecodeWithSchema(t *testing.T) {
 }
 
 func TestPayloadBatchDecodeWithSchema(t *testing.T) {
+	conf.IsTesting = true
 	tests := []struct {
 		name   string
 		input  any

internal/topo/node/lookup_node_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/lookup"
 	"github.com/lf-edge/ekuiper/v2/internal/xsql"
@@ -341,6 +342,7 @@ func TestLookup(t *testing.T) {
 }
 
 func TestLookupInner(t *testing.T) {
+	conf.IsTesting = true
 	tests := []struct {
 		name   string
 		input  any
@@ -608,6 +610,7 @@ func TestLookupInner(t *testing.T) {
 }
 
 func TestLookupPayload(t *testing.T) {
+	conf.IsTesting = true
 	tests := []struct {
 		name   string
 		input  any

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)