fix(parser): assign funcId in parser

Otherwise, the state may interfere

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

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}

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}

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) {

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},

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

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

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,
 		}

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

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",

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
 			}
 		}

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()    {}