Repository: lf-edge/ekuiper
Pull request #1253 (2022-04-27 09:18:43)

fix(parser): assign funcId in parser
Otherwise, the state may interfere

Signed-off-by: Jiyong Huang <huangjy@emqx.io>


Patch:
diff --git a/docs/en_US/sqls/built-in_functions.md b/docs/en_US/sqls/built-in_functions.md
index 70e4790b..ef820ed0 100644
--- a/docs/en_US/sqls/built-in_functions.md
+++ b/docs/en_US/sqls/built-in_functions.md
@@ -279,10 +279,14 @@ Rule to get the changed temperature and humidity values, and rename the changed
 ```text
 SQL: SELECT CHANGED_COLS("c_", true, temperature, humidity) FROM demo
 _________________________________________________________
-{"c_temperature":23,"c_humidity":88}
-{"c_temperature":25}
-{"c_humidity":90}
-{"c_humidity":91}
+{"c_ts":1, "c_temperature":23, "c_humidity":88}
+{"c_ts":2}
+{"c_ts":3}
+{"c_ts":4, "c_temperature":25}
+{"c_ts":5, "c_humidity":90}
+{"c_ts":6, "c_humidity":91}
+{"c_ts":7}
+{"c_ts":8}
 ```
 
 Rule to get the changed values of all columns and do not ignore null:
@@ -309,7 +313,7 @@ _________________________________________________________________
 Rule to get the events when temperature or humidity changed:
 
 ```text
-SQL: SELECT id, temperature, humidity FROM demo
+SQL: SELECT ts, temperature, humidity FROM demo
 WHERE HAD_CHANGED(true, temperature, humidity) = true
 _________________________________________________________
 {"ts":1,temperature":23,"humidity":88}
@@ -321,10 +325,9 @@ _________________________________________________________
 Rule to get the events when temperature has changed but humidity has NOT changed:
 
 ```text
-SQL: SELECT id, temperature, humidity FROM demo 
+SQL: SELECT ts, temperature, humidity FROM demo 
 WHERE HAD_CHANGED(true, temperature) = true AND HAD_CHANGED(true, humidity) = false
 _________________________________________________________
-{"ts":1,temperature":23,"humidity":88}
 {"ts":4,temperature":25,"humidity":88}
 ```
 
@@ -342,7 +345,7 @@ _________________________________________________________
 Rule to get the changed values when the temperature had changed to value bigger than 24:
 
 ```text
-SQL: SELECT id, temperature, humidity FROM demo 
+SQL: SELECT ts, temperature, humidity FROM demo 
 WHERE CHANGED_COL(true, temperature) > 24
 _________________________________________________________
 {"ts":4,temperature":25,"humidity":88}
diff --git a/docs/zh_CN/sqls/built-in_functions.md b/docs/zh_CN/sqls/built-in_functions.md
index b1ba354d..4f748b06 100755
--- a/docs/zh_CN/sqls/built-in_functions.md
+++ b/docs/zh_CN/sqls/built-in_functions.md
@@ -291,10 +291,14 @@ _________________________________________________________
 ```text
 SQL: SELECT CHANGED_COLS("c_", false, *) FROM demo
 _________________________________________________________
-{"c_temperature":23,"c_humidity":88}
-{"c_temperature":25}
-{"c_humidity":90}
-{"c_humidity":91}
+{"c_ts":1, "c_temperature":23, "c_humidity":88}
+{"c_ts":2}
+{"c_ts":3}
+{"c_ts":4, "c_temperature":25}
+{"c_ts":5, "c_humidity":90}
+{"c_ts":6, "c_humidity":91}
+{"c_ts":7}
+{"c_ts":8}
 ```
 
 获取窗口中平均值变化的规则:
@@ -310,7 +314,7 @@ _________________________________________________________________
 当 temperature 或者 humidity 变化时获取数据:
 
 ```text
-SQL: SELECT id, temperature, humidity FROM demo
+SQL: SELECT ts, temperature, humidity FROM demo
 WHERE HAD_CHANGED(true, temperature, humidity) = true
 _________________________________________________________
 {"ts":1,temperature":23,"humidity":88}
@@ -322,10 +326,9 @@ _________________________________________________________
 当 temperature 变化且 humidity 未变化时获取数据:
 
 ```text
-SQL: SELECT id, temperature, humidity FROM demo 
+SQL: SELECT ts, temperature, humidity FROM demo 
 WHERE HAD_CHANGED(true, temperature) = true AND HAD_CHANGED(true, humidity) = false
 _________________________________________________________
-{"ts":1,temperature":23,"humidity":88}
 {"ts":4,temperature":25,"humidity":88}
 ```
 
@@ -343,7 +346,7 @@ _________________________________________________________
 当 temperature 值变化后大于 24 时获取数据:
 
 ```text
-SQL: SELECT id, temperature, humidity FROM demo 
+SQL: SELECT ts, temperature, humidity FROM demo 
 WHERE CHANGED_COL(true, temperature) > 24
 _________________________________________________________
 {"ts":4,temperature":25,"humidity":88}
diff --git a/etc/sources/sql.yaml b/etc/sources/sql.yaml
index 5f216951..6a5093be 100644
--- a/etc/sources/sql.yaml
+++ b/etc/sources/sql.yaml
@@ -31,10 +31,10 @@ mysql_config:
     indexFieldType: "DATETIME"
     dateTimeFormat: "YYYY-MM-dd HH:mm:ss"
 
-sqltemplate_config:
+template_config:
   templateSqlQueryCfg:
-    TemplateSql: "select * from table where entry_data > {{.entry_data}}"
-    indexField: entry_data
+    TemplateSql: "select * from table where entry_date > '{{.entry_date}}'"
+    indexField: entry_date
     indexValue: "2022-04-13 06:22:32.233"
     indexFieldType: "DATETIME"
     dateTimeFormat: "YYYY-MM-dd HH:mm:ssSSS"
\ No newline at end of file
diff --git a/extensions/sqldatabase/sqlgen/sqlgen.go b/extensions/sqldatabase/sqlgen/sqlgen.go
index c7e45c1f..cba180e5 100644
--- a/extensions/sqldatabase/sqlgen/sqlgen.go
+++ b/extensions/sqldatabase/sqlgen/sqlgen.go
@@ -80,6 +80,25 @@ func (cfg *sqlConfig) Init(props map[string]interface{}) error {
 	if cfg.TemplateSqlQueryCfg == nil && cfg.InternalSqlQueryCfg == nil {
 		return fmt.Errorf("read properties %v fail with error: %v", props, err)
 	}
+
+	if cfg.TemplateSqlQueryCfg != nil && cfg.TemplateSqlQueryCfg.IndexFieldType == DATETIME_TYPE && cfg.TemplateSqlQueryCfg.DateTimeFormat != "" {
+		t, err := cast.InterfaceToTime(cfg.TemplateSqlQueryCfg.IndexValue, cfg.TemplateSqlQueryCfg.DateTimeFormat)
+		if err != nil {
+			err = fmt.Errorf("TemplateSqlQueryCfg InterfaceToTime datetime convert got error %v", err)
+			return err
+		}
+		cfg.TemplateSqlQueryCfg.IndexValue = t
+	}
+
+	if cfg.InternalSqlQueryCfg != nil && cfg.InternalSqlQueryCfg.IndexFieldType == DATETIME_TYPE && cfg.InternalSqlQueryCfg.DateTimeFormat != "" {
+		t, err := cast.InterfaceToTime(cfg.InternalSqlQueryCfg.IndexValue, cfg.InternalSqlQueryCfg.DateTimeFormat)
+		if err != nil {
+			err = fmt.Errorf("InternalSqlQueryCfg InterfaceToTime datetime convert got error %v", err)
+			return err
+		}
+		cfg.InternalSqlQueryCfg.IndexValue = t
+	}
+
 	return nil
 }
 
diff --git a/internal/binder/function/funcs_misc.go b/internal/binder/function/funcs_misc.go
index c4abd685..be06c7ed 100644
--- a/internal/binder/function/funcs_misc.go
+++ b/internal/binder/function/funcs_misc.go
@@ -512,39 +512,6 @@ func registerMiscFunc() {
 			return nil
 		},
 	}
-	builtins["changed_col"] = builtinFunc{
-		fType: FuncTypeScalar,
-		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
-			ignoreNull, ok := args[0].(bool)
-			if !ok {
-				return fmt.Errorf("first arg is not a bool but got %v", args[0]), false
-			}
-			if ignoreNull && args[1] == nil {
-				return nil, true
-			}
-			lv, err := ctx.GetState("self")
-			if err != nil {
-				return err, false
-			}
-			if !reflect.DeepEqual(args[1], lv) {
-				err := ctx.PutState("self", args[1])
-				if err != nil {
-					return err, false
-				}
-				return args[1], true
-			}
-			return nil, true
-		},
-		val: func(_ api.FunctionContext, args []ast.Expr) error {
-			if err := ValidateLen(2, len(args)); err != nil {
-				return err
-			}
-			if ast.IsNumericArg(args[0]) || ast.IsTimeArg(args[0]) || ast.IsStringArg(args[0]) {
-				return ProduceErrInfo(0, "boolean")
-			}
-			return nil
-		},
-	}
 	builtins["had_changed"] = builtinFunc{
 		fType: FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
diff --git a/internal/xsql/funcValuer.go b/internal/xsql/funcValuer.go
index 3026eabd..19aca9a1 100644
--- a/internal/xsql/funcValuer.go
+++ b/internal/xsql/funcValuer.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -47,8 +47,8 @@ func (*FunctionValuer) AliasValue(string) (interface{}, bool) {
 	return nil, false
 }
 
-func (fv *FunctionValuer) Call(name string, args []interface{}) (interface{}, bool) {
-	nf, fctx, err := fv.runtime.Get(name)
+func (fv *FunctionValuer) Call(name string, funcId int, args []interface{}) (interface{}, bool) {
+	nf, fctx, err := fv.runtime.Get(name, funcId)
 	switch err {
 	case errorx.NotFoundErr:
 		return nil, false
diff --git a/internal/xsql/funcsAggregate.go b/internal/xsql/funcsAggregate.go
index 670d9088..a6295c09 100644
--- a/internal/xsql/funcsAggregate.go
+++ b/internal/xsql/funcsAggregate.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -68,8 +68,8 @@ func (v *AggregateFunctionValuer) AliasValue(_ string) (interface{}, bool) {
 	return nil, false
 }
 
-func (v *AggregateFunctionValuer) Call(name string, args []interface{}) (interface{}, bool) {
-	nf, fctx, err := v.fv.runtime.Get(name)
+func (v *AggregateFunctionValuer) Call(name string, funcId int, args []interface{}) (interface{}, bool) {
+	nf, fctx, err := v.fv.runtime.Get(name, funcId)
 	switch err {
 	case errorx.NotFoundErr:
 		return nil, false
diff --git a/internal/xsql/functionRuntime.go b/internal/xsql/functionRuntime.go
index 81fd9850..1ba1e51e 100644
--- a/internal/xsql/functionRuntime.go
+++ b/internal/xsql/functionRuntime.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -26,7 +26,7 @@ import (
 //Each operator has a single instance of this to hold the context
 type funcRuntime struct {
 	sync.Mutex
-	regs      map[string]*funcReg
+	regs      []*funcReg
 	parentCtx api.StreamContext
 }
 
@@ -41,13 +41,15 @@ func NewFuncRuntime(ctx api.StreamContext) *funcRuntime {
 	}
 }
 
-func (fp *funcRuntime) Get(name string) (api.Function, api.FunctionContext, error) {
+func (fp *funcRuntime) Get(name string, funcId int) (api.Function, api.FunctionContext, error) {
 	fp.Lock()
 	defer fp.Unlock()
-	if fp.regs == nil {
-		fp.regs = make(map[string]*funcReg)
+	if len(fp.regs) <= funcId {
+		for i := len(fp.regs); i <= funcId; i++ {
+			fp.regs = append(fp.regs, nil)
+		}
 	}
-	if reg, ok := fp.regs[name]; !ok {
+	if reg := fp.regs[funcId]; reg == nil {
 		var (
 			nf  api.Function
 			err error
@@ -61,8 +63,8 @@ func (fp *funcRuntime) Get(name string) (api.Function, api.FunctionContext, erro
 				return nil, nil, err
 			}
 		}
-		fctx := context.NewDefaultFuncContext(fp.parentCtx, len(fp.regs))
-		fp.regs[name] = &funcReg{
+		fctx := context.NewDefaultFuncContext(fp.parentCtx, funcId)
+		fp.regs[funcId] = &funcReg{
 			ins: nf,
 			ctx: fctx,
 		}
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 1359de34..6bddd051 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -38,6 +38,7 @@ type Parser struct {
 	}
 	inFunc string // currently parsing function name
 	f      int    // anonymous field index number
+	fn     int    // function index number
 	clause string
 }
 
@@ -700,7 +701,9 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 			if valErr := validateFuncs(name, nil); valErr != nil {
 				return nil, valErr
 			}
-			return &ast.Call{Name: name, Args: args}, nil
+			c := &ast.Call{Name: name, Args: args, FuncId: p.fn}
+			p.fn += 1
+			return c, nil
 		} else {
 			p.unscan()
 		}
@@ -733,7 +736,9 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 		if name == "deduplicate" {
 			args = append([]ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}}, args...)
 		}
-		return &ast.Call{Name: name, Args: args}, nil
+		c := &ast.Call{Name: name, Args: args, FuncId: p.fn}
+		p.fn += 1
+		return c, nil
 	} else {
 		if err != nil {
 			return nil, err
diff --git a/internal/xsql/valuer.go b/internal/xsql/valuer.go
index c8dbcdf6..ad4ac8fc 100644
--- a/internal/xsql/valuer.go
+++ b/internal/xsql/valuer.go
@@ -44,7 +44,7 @@ type CallValuer interface {
 	Valuer
 
 	// Call is invoked to evaluate a function call (if possible).
-	Call(name string, args []interface{}) (interface{}, bool)
+	Call(name string, funcId int, args []interface{}) (interface{}, bool)
 }
 
 // FuncValuer can calculate function type value like window_start and window_end
@@ -299,10 +299,10 @@ func (a multiValuer) FuncValue(key string) (interface{}, bool) {
 	return nil, false
 }
 
-func (a multiValuer) Call(name string, args []interface{}) (interface{}, bool) {
+func (a multiValuer) Call(name string, funcId int, args []interface{}) (interface{}, bool) {
 	for _, valuer := range a {
 		if valuer, ok := valuer.(CallValuer); ok {
-			if v, ok := valuer.Call(name, args); ok {
+			if v, ok := valuer.Call(name, funcId, args); ok {
 				return v, true
 			} else {
 				return fmt.Errorf("call func %s error: %v", name, v), false
@@ -326,18 +326,18 @@ func MultiAggregateValuer(data AggregateData, singleCallValuer CallValuer, value
 	}
 }
 
-func (a *multiAggregateValuer) Call(name string, args []interface{}) (interface{}, bool) {
+func (a *multiAggregateValuer) Call(name string, funcId int, args []interface{}) (interface{}, bool) {
 	// assume the aggFuncMap already cache the custom agg funcs in IsAggFunc()
 	isAgg := function.IsAggFunc(name)
 	for _, valuer := range a.multiValuer {
 		if a, ok := valuer.(AggregateCallValuer); ok && isAgg {
-			if v, ok := a.Call(name, args); ok {
+			if v, ok := a.Call(name, funcId, args); ok {
 				return v, true
 			} else {
 				return fmt.Errorf("call func %s error: %v", name, v), false
 			}
 		} else if c, ok := valuer.(CallValuer); ok && !isAgg {
-			if v, ok := c.Call(name, args); ok {
+			if v, ok := c.Call(name, funcId, args); ok {
 				return v, true
 			}
 		}
@@ -467,7 +467,7 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 					}
 
 				}
-				val, _ := valuer.Call(expr.Name, args)
+				val, _ := valuer.Call(expr.Name, expr.FuncId, args)
 				return val
 			}
 		}
diff --git a/pkg/ast/expr.go b/pkg/ast/expr.go
index dc268efb..bc29722f 100644
--- a/pkg/ast/expr.go
+++ b/pkg/ast/expr.go
@@ -121,8 +121,9 @@ func (sl *StringLiteral) literal() {}
 func (sl *StringLiteral) node()    {}
 
 type Call struct {
-	Name string
-	Args []Expr
+	Name   string
+	FuncId int
+	Args   []Expr
 }
 
 func (c *Call) expr()    {}


Test patch:
diff --git a/extensions/sqldatabase/sqlgen/templateSqlDialect_test.go b/extensions/sqldatabase/sqlgen/templateSqlDialect_test.go
index 934b9613..4f4c3fe2 100644
--- a/extensions/sqldatabase/sqlgen/templateSqlDialect_test.go
+++ b/extensions/sqldatabase/sqlgen/templateSqlDialect_test.go
@@ -122,3 +122,53 @@ func TestTemplateQuery(t *testing.T) {
 		t.Errorf("SqlQueryStatement() = %v, want %v", nextSqlStr, want)
 	}
 }
+
+func TestTemplateQuery_DateTime(t *testing.T) {
+	tempCfg := map[string]interface{}{
+		"templateSql":    "select * from table where responseTime > `{{.responseTime}}`",
+		"indexField":     "responseTime",
+		"indexValue":     "2008-10-25 14:56:59.123",
+		"indexFieldType": DATETIME_TYPE,
+		"dateTimeFormat": "YYYY-MM-dd HH:mm:ssSSS",
+	}
+
+	props := map[string]interface{}{
+		"templateSqlQueryCfg": tempCfg,
+	}
+
+	sqlcfg := &sqlConfig{}
+	_ = sqlcfg.Init(props)
+
+	s, _ := NewTemplateSqlQuery(sqlcfg.TemplateSqlQueryCfg)
+
+	//query
+	firstSqlStr, _ := s.SqlQueryStatement()
+
+	want := "select * from table where responseTime > `2008-10-25 14:56:59.123`"
+
+	if !reflect.DeepEqual(firstSqlStr, want) {
+		t.Errorf("SqlQueryStatement() = %v, want %v", firstSqlStr, want)
+	}
+
+	//query result
+	s.UpdateMaxIndexValue(map[string]interface{}{
+		"responseTime": getDatetimeFromstring("2008-10-29 14:56:59"),
+	})
+	s.UpdateMaxIndexValue(map[string]interface{}{
+		"responseTime": getDatetimeFromstring("2008-11-11 11:12:01"),
+	})
+	s.UpdateMaxIndexValue(map[string]interface{}{
+		"responseTime": getDatetimeFromstring("2008-11-09 15:45:21"),
+	})
+	s.UpdateMaxIndexValue(map[string]interface{}{
+		"responseTime": getDatetimeFromstring("2008-11-11 13:23:44"),
+	})
+
+	nextSqlStr, _ := s.SqlQueryStatement()
+
+	want = "select * from table where responseTime > `2008-11-11 13:23:44.000`"
+
+	if !reflect.DeepEqual(nextSqlStr, want) {
+		t.Errorf("SqlQueryStatement() = %v, want %v", nextSqlStr, want)
+	}
+}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index b996b0bf..0db5b040 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -404,7 +404,7 @@ func Test_createLogicalPlan(t *testing.T) {
 								},
 							},
 							condition: &ast.BinaryExpr{
-								LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.Wildcard{
+								LHS: &ast.Call{Name: "count", FuncId: 0, Args: []ast.Expr{&ast.Wildcard{
 									Token: ast.ASTERISK,
 								}}},
 								OP:  ast.GT,
@@ -883,7 +883,8 @@ func Test_createLogicalPlan(t *testing.T) {
 							},
 							condition: &ast.BinaryExpr{
 								LHS: &ast.Call{
-									Name: "meta",
+									Name:   "meta",
+									FuncId: 2,
 									Args: []ast.Expr{&ast.MetaRef{
 										Name:       "device",
 										StreamName: ast.DefaultStream,
@@ -904,7 +905,7 @@ func Test_createLogicalPlan(t *testing.T) {
 						AName: "",
 					}, {
 						Expr: &ast.FieldRef{Name: "eid", StreamName: ast.AliasStream, AliasRef: ast.MockAliasRef(
-							&ast.Call{Name: "meta", Args: []ast.Expr{&ast.MetaRef{
+							&ast.Call{Name: "meta", FuncId: 0, Args: []ast.Expr{&ast.MetaRef{
 								Name:       "id",
 								StreamName: ast.DefaultStream,
 							}}},
@@ -915,7 +916,7 @@ func Test_createLogicalPlan(t *testing.T) {
 						AName: "eid",
 					}, {
 						Expr: &ast.FieldRef{Name: "hdevice", StreamName: ast.AliasStream, AliasRef: ast.MockAliasRef(
-							&ast.Call{Name: "meta", Args: []ast.Expr{
+							&ast.Call{Name: "meta", FuncId: 1, Args: []ast.Expr{
 								&ast.BinaryExpr{
 									OP:  ast.ARROW,
 									LHS: &ast.MetaRef{Name: "Humidity", StreamName: ast.DefaultStream},
@@ -1072,7 +1073,8 @@ func Test_createLogicalPlan(t *testing.T) {
 							},
 							condition: &ast.BinaryExpr{
 								LHS: &ast.Call{
-									Name: "meta",
+									Name:   "meta",
+									FuncId: 1,
 									Args: []ast.Expr{&ast.MetaRef{
 										Name:       "device",
 										StreamName: ast.DefaultStream,
@@ -1093,7 +1095,7 @@ func Test_createLogicalPlan(t *testing.T) {
 						AName: "",
 					}, {
 						Expr: &ast.FieldRef{Name: "m", StreamName: ast.AliasStream, AliasRef: ast.MockAliasRef(
-							&ast.Call{Name: "meta", Args: []ast.Expr{&ast.MetaRef{
+							&ast.Call{Name: "meta", FuncId: 0, Args: []ast.Expr{&ast.MetaRef{
 								Name:       "*",
 								StreamName: ast.DefaultStream,
 							}}},
@@ -1434,7 +1436,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 								},
 							},
 							condition: &ast.BinaryExpr{
-								LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.Wildcard{
+								LHS: &ast.Call{Name: "count", FuncId: 0, Args: []ast.Expr{&ast.Wildcard{
 									Token: ast.ASTERISK,
 								}}},
 								OP:  ast.GT,
@@ -1863,7 +1865,8 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 							},
 							condition: &ast.BinaryExpr{
 								LHS: &ast.Call{
-									Name: "meta",
+									Name:   "meta",
+									FuncId: 2,
 									Args: []ast.Expr{&ast.MetaRef{
 										Name:       "device",
 										StreamName: ast.DefaultStream,
@@ -1884,7 +1887,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 						AName: "",
 					}, {
 						Expr: &ast.FieldRef{Name: "eid", StreamName: ast.AliasStream, AliasRef: ast.MockAliasRef(
-							&ast.Call{Name: "meta", Args: []ast.Expr{&ast.MetaRef{
+							&ast.Call{Name: "meta", FuncId: 0, Args: []ast.Expr{&ast.MetaRef{
 								Name:       "id",
 								StreamName: ast.DefaultStream,
 							}}},
@@ -1895,7 +1898,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 						AName: "eid",
 					}, {
 						Expr: &ast.FieldRef{Name: "hdevice", StreamName: ast.AliasStream, AliasRef: ast.MockAliasRef(
-							&ast.Call{Name: "meta", Args: []ast.Expr{
+							&ast.Call{Name: "meta", FuncId: 1, Args: []ast.Expr{
 								&ast.BinaryExpr{
 									OP:  ast.ARROW,
 									LHS: &ast.MetaRef{Name: "Humidity", StreamName: ast.DefaultStream},
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index c9ac54a3..5f5204f2 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -350,7 +350,8 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: "",
 						Name:  "lpad",
 						Expr: &ast.Call{
-							Name: "lpad",
+							Name:   "lpad",
+							FuncId: 1,
 							Args: []ast.Expr{
 								&ast.Call{
 									Name: "lower",
@@ -375,7 +376,8 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: "field1",
 						Name:  "lpad",
 						Expr: &ast.Call{
-							Name: "lpad",
+							Name:   "lpad",
+							FuncId: 1,
 							Args: []ast.Expr{
 								&ast.Call{
 									Name: "lower",
@@ -400,7 +402,8 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: "",
 						Name:  "length",
 						Expr: &ast.Call{
-							Name: "length",
+							Name:   "length",
+							FuncId: 1,
 							Args: []ast.Expr{
 								&ast.Call{
 									Name: "lower",