Repository: lf-edge/ekuiper
Pull request #2823 (2024-04-28 08:29:19)

feat: remove useless flag in conveter


Patch:
diff --git a/internal/converter/converter.go b/internal/converter/converter.go
index eb5c7f91..02eccb4d 100644
--- a/internal/converter/converter.go
+++ b/internal/converter/converter.go
@@ -55,7 +55,10 @@ func GetOrCreateConverter(options *ast.Options) (c message.Converter, err error)
 		if options.RuleID == "" || options.StreamName == "" {
 			return json.GetConverter()
 		}
-		return json.NewFastJsonConverter(options.Schema, options.IsWildCard, options.IsSchemaLess), nil
+		if options.IsWildCard {
+			return json.NewFastJsonConverter(nil), nil
+		}
+		return json.NewFastJsonConverter(options.Schema), nil
 	}
 
 	schemaFile := ""
diff --git a/internal/converter/json/converter.go b/internal/converter/json/converter.go
index 36854eb3..82114233 100644
--- a/internal/converter/json/converter.go
+++ b/internal/converter/json/converter.go
@@ -56,25 +56,20 @@ func (c *Converter) Decode(ctx api.StreamContext, b []byte) (m any, err error) {
 
 type FastJsonConverter struct {
 	sync.RWMutex
-	isSchemaLess bool
-	isWildcard   bool
-	schema       map[string]*ast.JsonStreamField
+	schema map[string]*ast.JsonStreamField
 }
 
-func NewFastJsonConverter(schema map[string]*ast.JsonStreamField, isWildcard, isSchemaLess bool) *FastJsonConverter {
+func NewFastJsonConverter(schema map[string]*ast.JsonStreamField) *FastJsonConverter {
 	f := &FastJsonConverter{
-		schema:       schema,
-		isSchemaLess: isSchemaLess,
-		isWildcard:   isWildcard,
+		schema: schema,
 	}
 	return f
 }
 
-func (c *FastJsonConverter) ResetSchema(schema map[string]*ast.JsonStreamField, isWildcard bool) {
+func (c *FastJsonConverter) ResetSchema(schema map[string]*ast.JsonStreamField) {
 	c.Lock()
 	defer c.Unlock()
 	c.schema = schema
-	c.isWildcard = isWildcard
 }
 
 func (c *FastJsonConverter) Encode(ctx api.StreamContext, d any) (b []byte, err error) {
@@ -89,7 +84,7 @@ func (c *FastJsonConverter) Decode(ctx api.StreamContext, b []byte) (m any, err
 	}()
 	c.RLock()
 	defer c.RUnlock()
-	if c.isWildcard {
+	if c.schema == nil {
 		return converter.Decode(ctx, b)
 	}
 	return c.decodeWithSchema(b, c.schema)
@@ -135,9 +130,6 @@ func (f *FastJsonConverter) decodeWithSchema(b []byte, schema map[string]*ast.Js
 }
 
 func (f *FastJsonConverter) decodeArray(array []*fastjson.Value, field *ast.JsonStreamField) ([]interface{}, error) {
-	if !f.isSchemaLess && field == nil {
-		return nil, nil
-	}
 	vs := make([]interface{}, len(array))
 	for i, item := range array {
 		switch item.Type() {
@@ -221,54 +213,50 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 		case fastjson.TypeNull:
 			m[key] = nil
 		case fastjson.TypeObject:
-			add, valid := f.checkSchema(key, "struct", schema)
+			valid := f.checkSchema(key, "struct", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
-			if add {
-				childObj, err2 := v.Object()
-				if err2 != nil {
-					err = err2
-					return
-				}
-				var props map[string]*ast.JsonStreamField
-				if schema != nil && schema[key] != nil {
-					props = schema[key].Properties
-				}
-				childMap, err2 := f.decodeObject(childObj, props)
-				if err2 != nil {
-					err = err2
-					return
-				}
-				if childMap != nil {
-					m[key] = childMap
-				}
+			childObj, err2 := v.Object()
+			if err2 != nil {
+				err = err2
+				return
+			}
+			var props map[string]*ast.JsonStreamField
+			if schema != nil && schema[key] != nil {
+				props = schema[key].Properties
+			}
+			childMap, err2 := f.decodeObject(childObj, props)
+			if err2 != nil {
+				err = err2
+				return
+			}
+			if childMap != nil {
+				m[key] = childMap
 			}
 		case fastjson.TypeArray:
-			add, valid := f.checkSchema(key, "array", schema)
+			valid := f.checkSchema(key, "array", schema)
 			if !valid {
 				err = fmt.Errorf("%v has wrong type:%v, expect:%v", key, v.Type().String(), getType(schema[key]))
 				return
 			}
-			if add {
-				childArray, err2 := v.Array()
-				if err2 != nil {
-					err = err2
-					return
-				}
-				var items *ast.JsonStreamField
-				if schema != nil && schema[key] != nil {
-					items = schema[key].Items
-				}
-				subList, err2 := f.decodeArray(childArray, items)
-				if err2 != nil {
-					err = err2
-					return
-				}
-				if subList != nil {
-					m[key] = subList
-				}
+			childArray, err2 := v.Array()
+			if err2 != nil {
+				err = err2
+				return
+			}
+			var items *ast.JsonStreamField
+			if schema != nil && schema[key] != nil {
+				items = schema[key].Items
+			}
+			subList, err2 := f.decodeArray(childArray, items)
+			if err2 != nil {
+				err = err2
+				return
+			}
+			if subList != nil {
+				m[key] = subList
 			}
 		case fastjson.TypeString:
 			if schema != nil {
@@ -323,110 +311,92 @@ func (f *FastJsonConverter) decodeObject(obj *fastjson.Object, schema map[string
 	return m, nil
 }
 
-func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (add bool, valid bool) {
-	if f.isSchemaLess {
-		_, ok := schema[key]
-		return ok, true
-	}
-	if !f.isSchemaLess && schema[key] != nil && schema[key].Type == typ {
-		return true, true
+func (f *FastJsonConverter) checkSchema(key, typ string, schema map[string]*ast.JsonStreamField) (valid bool) {
+	v := schema[key]
+	if v == nil {
+		return true
 	}
-	return false, false
+	return v.Type == typ
 }
 
 func (f *FastJsonConverter) extractNumberValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
-	if f.isSchemaLess && field == nil {
+	if field == nil {
 		f64, err := v.Float64()
 		if err != nil {
 			return nil, err
 		}
 		return f64, nil
 	}
-	if !f.isSchemaLess {
-		if field == nil {
-			return nil, nil
+	switch {
+	case field.Type == "float", field.Type == "datetime":
+		f64, err := v.Float64()
+		if err != nil {
+			return nil, err
 		}
-		switch {
-		case field.Type == "float", field.Type == "datetime":
-			f64, err := v.Float64()
-			if err != nil {
-				return nil, err
-			}
-			return f64, nil
-		case field.Type == "bigint":
-			i64, err := v.Int64()
-			if err != nil {
-				return nil, err
-			}
-			return i64, nil
-		case field.Type == "string":
-			f64, err := v.Float64()
-			if err != nil {
-				return nil, err
-			}
-			return cast.ToStringAlways(f64), nil
-		case field.Type == "boolean":
-			bv, err := getBooleanFromValue(v)
-			if err != nil {
-				return nil, err
-			}
-			return bv, nil
+		return f64, nil
+	case field.Type == "bigint":
+		i64, err := v.Int64()
+		if err != nil {
+			return nil, err
+		}
+		return i64, nil
+	case field.Type == "string":
+		f64, err := v.Float64()
+		if err != nil {
+			return nil, err
+		}
+		return cast.ToStringAlways(f64), nil
+	case field.Type == "boolean":
+		bv, err := getBooleanFromValue(v)
+		if err != nil {
+			return nil, err
 		}
+		return bv, nil
 	}
 	return nil, fmt.Errorf("%v has wrong type:%v, expect:%v", name, fastjson.TypeNumber.String(), getType(field))
 }
 
 func (f *FastJsonConverter) extractStringValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
-	if f.isSchemaLess && field == nil {
+	if field == nil {
 		bs, err := v.StringBytes()
 		if err != nil {
 			return nil, err
 		}
 		return string(bs), nil
 	}
-	if !f.isSchemaLess {
-		if field == nil {
-			return nil, nil
+	switch {
+	case field.Type == "string", field.Type == "datetime":
+		bs, err := v.StringBytes()
+		if err != nil {
+			return nil, err
 		}
-		switch {
-		case field.Type == "string", field.Type == "datetime":
-			bs, err := v.StringBytes()
-			if err != nil {
-				return nil, err
-			}
-			return string(bs), nil
-		case field.Type == "bytea":
-			s, err := v.StringBytes()
-			if err != nil {
-				return nil, err
-			}
-			return cast.ToByteA(string(s), cast.CONVERT_ALL)
-		case field.Type == "boolean":
-			return getBooleanFromValue(v)
+		return string(bs), nil
+	case field.Type == "bytea":
+		s, err := v.StringBytes()
+		if err != nil {
+			return nil, err
 		}
+		return cast.ToByteA(string(s), cast.CONVERT_ALL)
+	case field.Type == "boolean":
+		return getBooleanFromValue(v)
 	}
 	return nil, fmt.Errorf("%v has wrong type:%v, expect:%v", name, fastjson.TypeString.String(), getType(field))
 }
 
 func (f *FastJsonConverter) extractBooleanFromValue(name string, v *fastjson.Value, field *ast.JsonStreamField) (interface{}, error) {
-	if f.isSchemaLess && field == nil {
+	if field == nil {
 		s, err := v.Bool()
 		if err != nil {
 			return nil, err
 		}
 		return s, nil
 	}
-	if !f.isSchemaLess {
-		if field == nil {
-			return nil, nil
-		}
-		if field.Type == "boolean" {
-			s, err := v.Bool()
-			if err != nil {
-				return nil, err
-			}
-			return s, nil
+	if field.Type == "boolean" {
+		s, err := v.Bool()
+		if err != nil {
+			return nil, err
 		}
+		return s, nil
 	}
 	return nil, fmt.Errorf("%v has wrong type:%v, expect:%v", name, v.Type().String(), getType(field))
 }
@@ -465,11 +435,3 @@ func getType(t *ast.JsonStreamField) string {
 		return t.Type
 	}
 }
-
-func getType2(t *ast.JsonStreamField) string {
-	if t == nil {
-		return "null"
-	} else {
-		return t.Type
-	}
-}
diff --git a/internal/converter/schema/schema.go b/internal/converter/schema/schema.go
index 95fec429..4fba89d1 100644
--- a/internal/converter/schema/schema.go
+++ b/internal/converter/schema/schema.go
@@ -105,13 +105,13 @@ func (s *SchemaLayer) DetachSchema(ruleID string) error {
 	return nil
 }
 
-func (s *SchemaLayer) GetSchema() (map[string]*ast.JsonStreamField, bool) {
+func (s *SchemaLayer) GetSchema() map[string]*ast.JsonStreamField {
 	s.RLock()
 	defer s.RUnlock()
 	if len(s.wildcardMap) > 0 {
-		return nil, true
+		return nil
 	}
-	return s.schema, false
+	return s.schema
 }
 
 func mergeSchema(originSchema, newSchema map[string]*ast.JsonStreamField) (map[string]*ast.JsonStreamField, error) {
diff --git a/pkg/message/artifacts.go b/pkg/message/artifacts.go
index c602ddf6..bb9a127f 100644
--- a/pkg/message/artifacts.go
+++ b/pkg/message/artifacts.go
@@ -41,7 +41,7 @@ type Converter interface {
 type ConverterProvider func(schemaFileName string, SchemaMessageName string, delimiter string) (Converter, error)
 
 type SchemaResetAbleConverter interface {
-	ResetSchema(schema map[string]*ast.JsonStreamField, isWildcard bool)
+	ResetSchema(schema map[string]*ast.JsonStreamField)
 }
 
 type ColumnSetter interface {


Test patch:
diff --git a/internal/converter/json/convert_bench_test.go b/internal/converter/json/convert_bench_test.go
index 4735469f..934717bd 100644
--- a/internal/converter/json/convert_bench_test.go
+++ b/internal/converter/json/convert_bench_test.go
@@ -107,7 +107,7 @@ func benchmarkByFiles(filePath string, b *testing.B, schema map[string]*ast.Json
 		b.Fatalf(err.Error())
 	}
 	if schema != nil {
-		f := NewFastJsonConverter(schema, false, false)
+		f := NewFastJsonConverter(schema)
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
 			f.Decode(ctx, payload)
diff --git a/internal/converter/json/converter_test.go b/internal/converter/json/converter_test.go
index f86e1bcd..5964d375 100644
--- a/internal/converter/json/converter_test.go
+++ b/internal/converter/json/converter_test.go
@@ -216,7 +216,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter(tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, tc.payload)
 		require.NoError(t, err)
 		require.Equal(t, v, tc.require)
@@ -227,7 +227,7 @@ func TestFastJsonConverterWithSchema(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter(tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -387,7 +387,7 @@ func TestFastJsonConverterWithSchemaError(t *testing.T) {
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
 	for _, tc := range testcases {
-		f := NewFastJsonConverter(tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		_, err := f.Decode(ctx, tc.payload)
 		require.Error(t, err)
 		require.Equal(t, err.Error(), tc.err.Error())
@@ -398,7 +398,7 @@ func TestFastJsonEncode(t *testing.T) {
 	a := make(map[string]int)
 	a["a"] = 1
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter(nil, false, false)
+	f := NewFastJsonConverter(nil)
 	v, err := f.Encode(ctx, a)
 	require.NoError(t, err)
 	require.Equal(t, v, []byte(`{"a":1}`))
@@ -431,7 +431,7 @@ func TestArrayWithArray(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter(schema, false, false)
+	f := NewFastJsonConverter(schema)
 	v, err := f.Decode(ctx, payload)
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -612,7 +612,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter(tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -622,7 +622,7 @@ func TestTypeNull(t *testing.T) {
 		arrayRequire := []map[string]interface{}{
 			tc.require,
 		}
-		f := NewFastJsonConverter(tc.schema, false, false)
+		f := NewFastJsonConverter(tc.schema)
 		v, err := f.Decode(ctx, arrayPayload)
 		require.NoError(t, err)
 		require.Equal(t, v, arrayRequire)
@@ -639,7 +639,7 @@ func TestConvertBytea(t *testing.T) {
 		},
 	}
 	ctx := mockContext.NewMockContext("test", "op1")
-	f := NewFastJsonConverter(schema, false, false)
+	f := NewFastJsonConverter(schema)
 	v, err := f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -655,7 +655,7 @@ func TestConvertBytea(t *testing.T) {
 			},
 		},
 	}
-	f = NewFastJsonConverter(schema, false, false)
+	f = NewFastJsonConverter(schema)
 	v, err = f.Decode(ctx, []byte(payload))
 	require.NoError(t, err)
 	require.Equal(t, v, map[string]interface{}{
@@ -667,7 +667,7 @@ func TestSchemaless(t *testing.T) {
 	originSchema := map[string]*ast.JsonStreamField{
 		"a": nil,
 	}
-	f := NewFastJsonConverter(originSchema, false, true)
+	f := NewFastJsonConverter(originSchema)
 	testcases := []struct {
 		data   map[string]interface{}
 		expect map[string]interface{}
@@ -722,7 +722,7 @@ func TestJsonError(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, errorx.CovnerterErr, errWithCode.Code())
 	// fastjson
-	c := NewFastJsonConverter(nil, false, true)
+	c := NewFastJsonConverter(nil)
 	_, err = c.Decode(ctx, nil)
 	require.Error(t, err)
 	errWithCode, ok = err.(errorx.ErrorWithCode)
diff --git a/internal/converter/schema/schema_test.go b/internal/converter/schema/schema_test.go
index bb7b736b..ecc2181f 100644
--- a/internal/converter/schema/schema_test.go
+++ b/internal/converter/schema/schema_test.go
@@ -206,9 +206,8 @@ func TestMergeWildcardSchema(t *testing.T) {
 		},
 	}
 	require.NoError(t, f.MergeSchema("3", "", newSchema, false))
-	s, isWildcard := f.GetSchema()
+	s := f.GetSchema()
 	require.Nil(t, s)
-	require.True(t, isWildcard)
 }
 
 func TestAttachDetachSchema(t *testing.T) {