refactor(valuer): avoid dynamic case cast when valuing message

Keep field name as it is according to stream def.
If no stream defined, keep the case from the source.

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

internal/topo/operator/field_processor.go

@@ -26,7 +26,6 @@ import (
 	"math"
 	"reflect"
 	"strconv"
-	"strings"
 	"time"
 )
 
@@ -204,7 +203,7 @@ func (p *defaultFieldProcessor) addRecField(ft ast.FieldType, r map[string]inter
 			}
 			nextR := make(map[string]interface{})
 			for _, nextF := range st.StreamFields {
-				nextP := strings.ToLower(nextF.Name)
+				nextP := nextF.Name
 				if e := p.addRecField(nextF.FieldType, nextR, nextJ, nextP); e != nil {
 					return e
 				}

internal/topo/operator/preprocessor.go

@@ -19,7 +19,6 @@ import (
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
-	"strings"
 )
 
 type Preprocessor struct {
@@ -75,7 +74,7 @@ func (p *Preprocessor) Apply(ctx api.StreamContext, data interface{}, _ *xsql.Fu
 		newMeta := make(xsql.Metadata)
 		for _, f := range p.metaFields {
 			if m, ok := tuple.Metadata.Value(f); ok {
-				newMeta[strings.ToLower(f)] = m
+				newMeta[f] = m
 			}
 		}
 		tuple.Metadata = newMeta

internal/topo/operator/preprocessor_test.go

@@ -101,13 +101,13 @@ func TestPreprocessor_Apply(t *testing.T) {
 				Name: ast.StreamName("demo"),
 				StreamFields: []ast.StreamField{
 					{Name: "abc", FieldType: &ast.BasicType{Type: ast.FLOAT}},
-					{Name: "def", FieldType: &ast.BasicType{Type: ast.STRINGS}},
+					{Name: "dEf", FieldType: &ast.BasicType{Type: ast.STRINGS}},
 				},
 			},
 			data: []byte(`{"abc": 34, "def" : "hello", "ghi": 50}`),
 			result: &xsql.Tuple{Message: xsql.Message{
 				"abc": float64(34),
-				"def": "hello",
+				"dEf": "hello",
 			},
 			},
 		},

internal/xsql/collections.go

@@ -58,8 +58,7 @@ func (m Message) Value(key string) (interface{}, bool) {
 		conf.Log.Println("Invalid key: " + key + ", expect source.field or field.")
 		return nil, false
 	}
-	key1 := strings.ToLower(colkey)
-	if v, ok := m[key1]; ok {
+	if v, ok := m[colkey]; ok {
 		return v, ok
 	} else {
 		//Only when with 'SELECT * FROM ...'  and 'schemaless', the key in map is not convert to lower case.
@@ -70,9 +69,8 @@ func (m Message) Value(key string) (interface{}, bool) {
 
 func (m Message) getIgnoreCase(key interface{}) (interface{}, bool) {
 	if k, ok := key.(string); ok {
-		key = strings.ToLower(k)
-		for k, v := range m {
-			if strings.ToLower(k) == key {
+		for mk, v := range m {
+			if strings.EqualFold(k, mk) {
 				return v, true
 			}
 		}