Repository: lf-edge/ekuiper
Pull request #1418 (2022-09-28 08:56:52)

fix(bug): Remove warnings in native plugin manager


Patch:
diff --git a/internal/binder/function/binder.go b/internal/binder/function/binder.go
index 27b5cdee..8ef545ea 100644
--- a/internal/binder/function/binder.go
+++ b/internal/binder/function/binder.go
@@ -17,6 +17,7 @@ package function
 import (
 	"github.com/lf-edge/ekuiper/internal/binder"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 )
 
@@ -25,15 +26,6 @@ var ( // init once and read only
 	funcFactoriesNames []string
 )
 
-type FuncType int
-
-const (
-	FuncTypeUnknown FuncType = iota - 1
-	FuncTypeScalar
-	FuncTypeAgg
-	FuncTypeCols
-)
-
 func init() {
 	f := binder.FactoryEntry{
 		Name:    "built-in",
@@ -92,14 +84,14 @@ func ConvName(name string) (string, bool) {
 }
 
 type multiAggFunc interface {
-	GetFuncType(name string) FuncType
+	GetFuncType(name string) ast.FuncType
 }
 
 func IsAggFunc(funcName string) bool {
 	f, _ := Function(funcName)
 	if f != nil {
 		if mf, ok := f.(multiAggFunc); ok {
-			return mf.GetFuncType(funcName) == FuncTypeAgg
+			return mf.GetFuncType(funcName) == ast.FuncTypeAgg
 		} else {
 			return f.IsAggregate()
 		}
@@ -107,16 +99,16 @@ func IsAggFunc(funcName string) bool {
 	return false
 }
 
-func GetFuncType(funcName string) FuncType {
+func GetFuncType(funcName string) ast.FuncType {
 	f, _ := Function(funcName)
 	if f != nil {
 		if mf, ok := f.(multiAggFunc); ok {
 			return mf.GetFuncType(funcName)
 		}
 		if f.IsAggregate() {
-			return FuncTypeAgg
+			return ast.FuncTypeAgg
 		}
-		return FuncTypeScalar
+		return ast.FuncTypeScalar
 	}
-	return FuncTypeUnknown
+	return ast.FuncTypeUnknown
 }
diff --git a/internal/binder/function/funcs_agg.go b/internal/binder/function/funcs_agg.go
index 9ee1eab7..38172a06 100644
--- a/internal/binder/function/funcs_agg.go
+++ b/internal/binder/function/funcs_agg.go
@@ -22,7 +22,7 @@ import (
 
 func registerAggFunc() {
 	builtins["avg"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			c := getCount(arg0)
@@ -52,7 +52,7 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["count"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			return getCount(arg0), true
@@ -60,7 +60,7 @@ func registerAggFunc() {
 		val: ValidateOneArg,
 	}
 	builtins["max"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			if len(arg0) > 0 {
@@ -101,7 +101,7 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["min"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			if len(arg0) > 0 {
@@ -142,7 +142,7 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sum"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			if len(arg0) > 0 {
@@ -171,14 +171,14 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["collect"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return args[0], true
 		},
 		val: ValidateOneArg,
 	}
 	builtins["deduplicate"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].([]interface{})
 			v2, ok2 := args[1].([]interface{})
diff --git a/internal/binder/function/funcs_cols.go b/internal/binder/function/funcs_cols.go
index 055d9edc..4826c821 100644
--- a/internal/binder/function/funcs_cols.go
+++ b/internal/binder/function/funcs_cols.go
@@ -42,7 +42,7 @@ func wrapColFunc(colFunc ColFunc) funcExe {
 
 func registerColsFunc() {
 	builtins["changed_cols"] = builtinFunc{
-		fType: FuncTypeCols,
+		fType: ast.FuncTypeCols,
 		exec:  wrapColFunc(changedFunc),
 		val: func(_ api.FunctionContext, args []ast.Expr) error {
 			if len(args) <= 2 {
diff --git a/internal/binder/function/funcs_math.go b/internal/binder/function/funcs_math.go
index 0fe831c9..15693ab5 100644
--- a/internal/binder/function/funcs_math.go
+++ b/internal/binder/function/funcs_math.go
@@ -24,7 +24,7 @@ import (
 
 func registerMathFunc() {
 	builtins["abs"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[0].(int); ok {
 				t := float64(v)
@@ -39,7 +39,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["acos"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Acos(v), true
@@ -50,7 +50,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["asin"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Asin(v), true
@@ -61,7 +61,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["atan"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Atan(v), true
@@ -72,7 +72,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["atan2"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v1, e := toF64(args[0]); e == nil {
 				if v2, e1 := toF64(args[1]); e1 == nil {
@@ -87,7 +87,7 @@ func registerMathFunc() {
 		val: ValidateTwoNumberArg,
 	}
 	builtins["bitand"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			v2, ok2 := args[1].(int)
@@ -100,7 +100,7 @@ func registerMathFunc() {
 		val: ValidateTwoIntArg,
 	}
 	builtins["bitor"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			v2, ok2 := args[1].(int)
@@ -113,7 +113,7 @@ func registerMathFunc() {
 		val: ValidateTwoIntArg,
 	}
 	builtins["bitxor"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			v2, ok2 := args[1].(int)
@@ -126,7 +126,7 @@ func registerMathFunc() {
 		val: ValidateTwoIntArg,
 	}
 	builtins["bitnot"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			if ok1 {
@@ -146,7 +146,7 @@ func registerMathFunc() {
 		},
 	}
 	builtins["ceil"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Ceil(v), true
@@ -157,7 +157,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["cos"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Cos(v), true
@@ -168,7 +168,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["cosh"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Cosh(v), true
@@ -179,7 +179,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["exp"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Exp(v), true
@@ -190,7 +190,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["ln"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Log2(v), true
@@ -201,7 +201,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["log"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Log10(v), true
@@ -212,7 +212,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["mod"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				if v1, e1 := toF64(args[1]); e == nil {
@@ -227,7 +227,7 @@ func registerMathFunc() {
 		val: ValidateTwoNumberArg,
 	}
 	builtins["power"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v1, e := toF64(args[0]); e == nil {
 				if v2, e2 := toF64(args[1]); e2 == nil {
@@ -242,14 +242,14 @@ func registerMathFunc() {
 		val: ValidateTwoNumberArg,
 	}
 	builtins["rand"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return rand.Float64(), true
 		},
 		val: ValidateOneArg,
 	}
 	builtins["round"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Round(v), true
@@ -260,7 +260,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sign"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				if v > 0 {
@@ -277,7 +277,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sin"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Sin(v), true
@@ -288,7 +288,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sinh"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Sinh(v), true
@@ -299,7 +299,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sqrt"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Sqrt(v), true
@@ -310,7 +310,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["tan"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Tan(v), true
@@ -321,7 +321,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["tanh"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Tanh(v), true
diff --git a/internal/binder/function/funcs_misc.go b/internal/binder/function/funcs_misc.go
index f9011f76..74fcf178 100644
--- a/internal/binder/function/funcs_misc.go
+++ b/internal/binder/function/funcs_misc.go
@@ -35,7 +35,7 @@ import (
 
 func registerMiscFunc() {
 	builtins["cast"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[1].(string); ok {
 				switch v {
@@ -150,7 +150,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["chr"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[0].(int); ok {
 				return rune(v), true
@@ -178,7 +178,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["encode"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[1].(string); ok {
 				if strings.EqualFold(v, "base64") {
@@ -215,7 +215,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["trunc"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			var v0 float64
 			if v1, ok := args[0].(int); ok {
@@ -247,7 +247,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["md5"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -263,7 +263,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha1"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -279,7 +279,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha256"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -295,7 +295,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha384"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -311,7 +311,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha512"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -327,7 +327,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["isnull"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return true, true
@@ -344,7 +344,7 @@ func registerMiscFunc() {
 		val: ValidateOneArg,
 	}
 	builtins["newuuid"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if newUUID, err := uuid.NewUUID(); err != nil {
 				return err, false
@@ -355,14 +355,14 @@ func registerMiscFunc() {
 		val: ValidateNoArg,
 	}
 	builtins["tstamp"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return conf.GetNowInMilli(), true
 		},
 		val: ValidateNoArg,
 	}
 	builtins["mqtt"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[0].(string); ok {
 				return v, true
@@ -386,7 +386,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["meta"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return args[0], true
 		},
@@ -412,7 +412,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["cardinality"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			val := reflect.ValueOf(args[0])
 			if val.Kind() == reflect.Slice {
@@ -423,7 +423,7 @@ func registerMiscFunc() {
 		val: ValidateOneArg,
 	}
 	builtins["json_path_query"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result, err := jsonCall(ctx, args)
 			if err != nil {
@@ -434,7 +434,7 @@ func registerMiscFunc() {
 		val: ValidateJsonFunc,
 	}
 	builtins["json_path_query_first"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result, err := jsonCall(ctx, args)
 			if err != nil {
@@ -449,7 +449,7 @@ func registerMiscFunc() {
 		val: ValidateJsonFunc,
 	}
 	builtins["json_path_exists"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result, err := jsonCall(ctx, args)
 			if err != nil {
@@ -470,17 +470,17 @@ func registerMiscFunc() {
 		val: ValidateJsonFunc,
 	}
 	builtins["window_start"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec:  nil, // directly return in the valuer
 		val:   ValidateNoArg,
 	}
 	builtins["window_end"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec:  nil, // directly return in the valuer
 		val:   ValidateNoArg,
 	}
 	builtins["changed_col"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			ignoreNull, ok := args[0].(bool)
 			if !ok {
@@ -513,7 +513,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["had_changed"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if len(args) <= 1 {
 				return fmt.Errorf("expect more than one arg but got %d", len(args)), false
@@ -555,7 +555,7 @@ func registerMiscFunc() {
 	}
 
 	builtins["lag"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if len(args) != 1 && len(args) != 2 && len(args) != 3 {
 				return fmt.Errorf("expect one two or three args but got %d", len(args)), false
@@ -625,7 +625,7 @@ func registerMiscFunc() {
 	}
 
 	builtins["object_construct"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result := make(map[string]interface{})
 			for i := 0; i < len(args); i += 2 {
diff --git a/internal/binder/function/funcs_str.go b/internal/binder/function/funcs_str.go
index b5d83764..9b74dcb6 100644
--- a/internal/binder/function/funcs_str.go
+++ b/internal/binder/function/funcs_str.go
@@ -28,7 +28,7 @@ import (
 
 func registerStrFunc() {
 	builtins["concat"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			var b bytes.Buffer
 			for _, arg := range args {
@@ -49,7 +49,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["endswith"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return false, true
@@ -60,7 +60,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["indexof"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return -1, true
@@ -71,7 +71,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["length"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := cast.ToStringAlways(args[0])
 			return utf8.RuneCountInString(arg0), true
@@ -79,7 +79,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["lower"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -90,7 +90,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["lpad"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -105,7 +105,7 @@ func registerStrFunc() {
 		val: ValidateOneStrOneInt,
 	}
 	builtins["ltrim"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -116,7 +116,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["numbytes"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := cast.ToStringAlways(args[0])
 			return len(arg0), true
@@ -124,7 +124,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["format_time"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -155,7 +155,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["regexp_matches"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return false, true
@@ -170,7 +170,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["regexp_replace"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil || args[2] == nil {
 				return nil, true
@@ -195,7 +195,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["regexp_substr"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return nil, true
@@ -210,7 +210,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["rpad"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -225,7 +225,7 @@ func registerStrFunc() {
 		val: ValidateOneStrOneInt,
 	}
 	builtins["rtrim"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -236,7 +236,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["substring"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -306,7 +306,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["startswith"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return false, true
@@ -317,7 +317,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["split_value"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return nil, true
@@ -354,7 +354,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["trim"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -365,7 +365,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["upper"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
diff --git a/internal/binder/function/function.go b/internal/binder/function/function.go
index 71a66474..7499eea2 100644
--- a/internal/binder/function/function.go
+++ b/internal/binder/function/function.go
@@ -25,7 +25,7 @@ type funcExe func(ctx api.FunctionContext, args []interface{}) (interface{}, boo
 type funcVal func(ctx api.FunctionContext, args []ast.Expr) error
 
 type builtinFunc struct {
-	fType FuncType
+	fType ast.FuncType
 	exec  funcExe
 	val   funcVal
 }
@@ -87,10 +87,10 @@ func (f *funcExecutor) IsAggregate() bool {
 	return false
 }
 
-func (f *funcExecutor) GetFuncType(name string) FuncType {
+func (f *funcExecutor) GetFuncType(name string) ast.FuncType {
 	fs, ok := builtins[name]
 	if !ok {
-		return FuncTypeUnknown
+		return ast.FuncTypeUnknown
 	}
 	return fs.fType
 }
diff --git a/internal/plugin/native/manager.go b/internal/plugin/native/manager.go
index a98067d2..9632bb89 100644
--- a/internal/plugin/native/manager.go
+++ b/internal/plugin/native/manager.go
@@ -583,8 +583,8 @@ func (rr *Manager) HasFunctionSet(name string) bool {
 }
 
 func (rr *Manager) ConvName(name string) (string, bool) {
-	_, err := rr.Function(name)
-	if err == nil {
+	_, ok := rr.GetPluginBySymbol(plugin2.FUNCTION, name)
+	if ok {
 		return name, true
 	}
 	return name, false
@@ -606,7 +606,6 @@ func (rr *Manager) loadRuntime(t plugin2.PluginType, name, soFilepath string) (p
 		} else {
 			mod, err := rr.getSoFilePath(t, name, false)
 			if err != nil {
-				conf.Log.Warnf(fmt.Sprintf("cannot find the native plugin %s in path: %v", name, err))
 				return nil, nil
 			}
 			soPath = mod
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index 87634e6b..e3ec4a9e 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.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.
@@ -344,6 +344,11 @@ func (m *Manager) Delete(name string) error {
 	if !ok {
 		return fmt.Errorf("portable plugin %s is not found", name)
 	}
+	pm := runtime.GetPluginInsManager()
+	err := pm.Kill(name)
+	if err != nil {
+		conf.Log.Errorf("fail to kill portable plugin %s process, please try to kill it manually", name)
+	}
 	// unregister the plugin
 	m.reg.Delete(name)
 	// delete files and uninstall metas
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager.go b/internal/plugin/portable/runtime/plugin_ins_manager.go
index 191da6cb..3c210cf3 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager.go
@@ -178,16 +178,22 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 		return nil, fmt.Errorf("invalid conf: %v", pconf)
 	}
 	var cmd *exec.Cmd
-	switch pluginMeta.Language {
-	case "go":
-		conf.Log.Printf("starting go plugin executable %s", pluginMeta.Executable)
-		cmd = exec.Command(pluginMeta.Executable, string(jsonArg))
+	err = infra.SafeRun(func() error {
+		switch pluginMeta.Language {
+		case "go":
+			conf.Log.Printf("starting go plugin executable %s", pluginMeta.Executable)
+			cmd = exec.Command(pluginMeta.Executable, string(jsonArg))
 
-	case "python":
-		conf.Log.Printf("starting python plugin executable %s with script %s\n", conf.Config.Portable.PythonBin, pluginMeta.Executable)
-		cmd = exec.Command(conf.Config.Portable.PythonBin, pluginMeta.Executable, string(jsonArg))
-	default:
-		return nil, fmt.Errorf("unsupported language: %s", pluginMeta.Language)
+		case "python":
+			conf.Log.Printf("starting python plugin executable %s with script %s\n", conf.Config.Portable.PythonBin, pluginMeta.Executable)
+			cmd = exec.Command(conf.Config.Portable.PythonBin, pluginMeta.Executable, string(jsonArg))
+		default:
+			return fmt.Errorf("unsupported language: %s", pluginMeta.Language)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("fail to start plugin %s: %v", pluginMeta.Name, err)
 	}
 	cmd.Stdout = conf.Log.Out
 	cmd.Stderr = conf.Log.Out
diff --git a/internal/topo/node/dynamic_channel_buffer.go b/internal/topo/node/dynamic_channel_buffer.go
index ec2a668a..dc98b45f 100644
--- a/internal/topo/node/dynamic_channel_buffer.go
+++ b/internal/topo/node/dynamic_channel_buffer.go
@@ -29,7 +29,7 @@ type DynamicChannelBuffer struct {
 
 func NewDynamicChannelBuffer() *DynamicChannelBuffer {
 	buffer := &DynamicChannelBuffer{
-		In:     make(chan api.SourceTuple, 10),
+		In:     make(chan api.SourceTuple, 1024),
 		Out:    make(chan api.SourceTuple),
 		buffer: make([]api.SourceTuple, 0),
 		limit:  102400,
@@ -58,8 +58,10 @@ func (b *DynamicChannelBuffer) run() {
 		} else if l > 0 {
 			select {
 			case b.Out <- b.buffer[0]:
+				//fmt.Println("out loud")
 				b.buffer = b.buffer[1:]
 			case value := <-b.In:
+				//fmt.Printf("in loud with length %d\n", len(b.In))
 				b.buffer = append(b.buffer, value)
 			case <-b.done:
 				return
@@ -67,6 +69,7 @@ func (b *DynamicChannelBuffer) run() {
 		} else {
 			select {
 			case value := <-b.In:
+				//fmt.Printf("in quiet with length %d \n", len(b.In))
 				b.buffer = append(b.buffer, value)
 			case <-b.done:
 				return
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 0c560404..34f296c9 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -823,7 +823,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 	p.inFunc = name
 	defer func() { p.inFunc = "" }()
 	ft := function.GetFuncType(name)
-	if ft == function.FuncTypeCols && p.clause != "select" {
+	if ft == ast.FuncTypeCols && p.clause != "select" {
 		return nil, fmt.Errorf("function %s can only be used inside the select clause", n)
 	}
 	var args []ast.Expr
@@ -832,7 +832,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 			if valErr := validateFuncs(name, nil); valErr != nil {
 				return nil, valErr
 			}
-			c := &ast.Call{Name: name, Args: args, FuncId: p.fn}
+			c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
 			p.fn += 1
 			return c, nil
 		} else {
@@ -842,7 +842,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 		if exp, err := p.ParseExpr(); err != nil {
 			return nil, err
 		} else {
-			if ft == function.FuncTypeCols {
+			if ft == ast.FuncTypeCols {
 				field := &ast.ColFuncField{Expr: exp, Name: nameExpr(exp)}
 				args = append(args, field)
 			} else {
@@ -867,7 +867,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 		if name == "deduplicate" {
 			args = append([]ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}}, args...)
 		}
-		c := &ast.Call{Name: name, Args: args, FuncId: p.fn}
+		c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
 		p.fn += 1
 		return c, nil
 	} else {
diff --git a/internal/xsql/valuer.go b/internal/xsql/valuer.go
index 8ec00626..0eaf90a7 100644
--- a/internal/xsql/valuer.go
+++ b/internal/xsql/valuer.go
@@ -281,11 +281,11 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 			if valuer, ok := v.Valuer.(CallValuer); ok {
 				var (
 					args []interface{}
-					ft   = function.GetFuncType(expr.Name)
+					ft   = expr.FuncType
 				)
 				if len(expr.Args) > 0 {
 					switch ft {
-					case function.FuncTypeAgg:
+					case ast.FuncTypeAgg:
 						args = make([]interface{}, len(expr.Args))
 						for i, arg := range expr.Args {
 							if aggreValuer, ok := valuer.(AggregateCallValuer); ok {
@@ -297,7 +297,7 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 								}
 							}
 						}
-					case function.FuncTypeScalar:
+					case ast.FuncTypeScalar:
 						args = make([]interface{}, len(expr.Args))
 						for i, arg := range expr.Args {
 							args[i] = v.Eval(arg)
@@ -305,7 +305,7 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 								return args[i]
 							}
 						}
-					case function.FuncTypeCols:
+					case ast.FuncTypeCols:
 						var keys []string
 						for _, arg := range expr.Args { // In the parser, the col func arguments must be ColField
 							cf, ok := arg.(*ast.ColFuncField)
diff --git a/pkg/ast/expr.go b/pkg/ast/expr.go
index 3ca9d981..689b13db 100644
--- a/pkg/ast/expr.go
+++ b/pkg/ast/expr.go
@@ -122,10 +122,20 @@ func (sl *StringLiteral) expr()    {}
 func (sl *StringLiteral) literal() {}
 func (sl *StringLiteral) node()    {}
 
+type FuncType int
+
+const (
+	FuncTypeUnknown FuncType = iota - 1
+	FuncTypeScalar
+	FuncTypeAgg
+	FuncTypeCols
+)
+
 type Call struct {
-	Name   string
-	FuncId int
-	Args   []Expr
+	Name     string
+	FuncId   int
+	FuncType FuncType
+	Args     []Expr
 }
 
 func (c *Call) expr()    {}


Test patch:
diff --git a/internal/topo/node/dynamic_channel_buffer_test.go b/internal/topo/node/dynamic_channel_buffer_test.go
new file mode 100644
index 00000000..3de4ee7f
--- /dev/null
+++ b/internal/topo/node/dynamic_channel_buffer_test.go
@@ -0,0 +1,49 @@
+// Copyright 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package node
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"testing"
+	"time"
+)
+
+func TestBuffer(t *testing.T) {
+	b := NewDynamicChannelBuffer()
+	b.SetLimit(100)
+	stopSign := make(chan struct{})
+	go func(done chan struct{}) {
+		for i := 0; i < 100; i++ {
+			select {
+			case b.In <- api.NewDefaultSourceTuple(map[string]interface{}{"a": 5}, nil):
+				fmt.Printf("feed in %d\n", i)
+			default:
+				t.Errorf("message %d dropped, should not drop message", i)
+			}
+		}
+		close(done)
+	}(stopSign)
+	for i := 0; i < 50; i++ {
+		_ = <-b.Out
+		fmt.Printf("eaten %d\n", i)
+		time.Sleep(10 * time.Millisecond)
+	}
+	<-stopSign
+	l := b.GetLength()
+	if l != 50 {
+		t.Errorf("Expect buffer length 50, but got %d", l)
+	}
+}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 6f6387fb..cf87bd02 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -406,7 +406,7 @@ func Test_createLogicalPlan(t *testing.T) {
 							condition: &ast.BinaryExpr{
 								LHS: &ast.Call{Name: "count", FuncId: 0, Args: []ast.Expr{&ast.Wildcard{
 									Token: ast.ASTERISK,
-								}}},
+								}}, FuncType: ast.FuncTypeAgg},
 								OP:  ast.GT,
 								RHS: &ast.IntegerLiteral{Val: 2},
 							},
@@ -1430,7 +1430,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 							condition: &ast.BinaryExpr{
 								LHS: &ast.Call{Name: "count", FuncId: 0, Args: []ast.Expr{&ast.Wildcard{
 									Token: ast.ASTERISK,
-								}}},
+								}}, FuncType: ast.FuncTypeAgg},
 								OP:  ast.GT,
 								RHS: &ast.IntegerLiteral{Val: 2},
 							},
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index 5c03bf92..6dd5bcd7 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -431,8 +431,9 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: "",
 						Name:  "count",
 						Expr: &ast.Call{
-							Name: "count",
-							Args: []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}},
+							Name:     "count",
+							Args:     []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}},
+							FuncType: ast.FuncTypeAgg,
 						},
 					},
 				},
@@ -483,8 +484,9 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: "",
 						Name:  "deduplicate",
 						Expr: &ast.Call{
-							Name: "deduplicate",
-							Args: []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}, &ast.FieldRef{Name: "temperature", StreamName: ast.DefaultStream}, &ast.BooleanLiteral{Val: false}},
+							Name:     "deduplicate",
+							Args:     []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}, &ast.FieldRef{Name: "temperature", StreamName: ast.DefaultStream}, &ast.BooleanLiteral{Val: false}},
+							FuncType: ast.FuncTypeAgg,
 						},
 					},
 				},
@@ -1160,7 +1162,7 @@ func TestParser_ParseStatement(t *testing.T) {
 				Sources:    []ast.Source{&ast.Table{Name: "topic/sensor1"}},
 				Condition:  &ast.BinaryExpr{LHS: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}, OP: ast.EQ, RHS: &ast.StringLiteral{Val: "dname"}},
 				Dimensions: ast.Dimensions{ast.Dimension{Expr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
-				Having:     &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
+				Having:     &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}, FuncType: ast.FuncTypeAgg}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
 			},
 		},
 
@@ -1173,7 +1175,7 @@ func TestParser_ParseStatement(t *testing.T) {
 				},
 				Sources:   []ast.Source{&ast.Table{Name: "topic/sensor1"}},
 				Condition: &ast.BinaryExpr{LHS: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}, OP: ast.EQ, RHS: &ast.StringLiteral{Val: "dname"}},
-				Having:    &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
+				Having:    &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}, FuncType: ast.FuncTypeAgg}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
 			},
 		},
 
@@ -1657,6 +1659,7 @@ func TestParser_ParseStatement(t *testing.T) {
 								Args: []ast.Expr{
 									&ast.Wildcard{Token: ast.ASTERISK},
 								},
+								FuncType: ast.FuncTypeAgg,
 							},
 							RHS: &ast.IntegerLiteral{Val: 10},
 						},
@@ -1721,6 +1724,7 @@ func TestParser_ParseStatement(t *testing.T) {
 									Name:       "c",
 								}},
 							},
+							FuncType: ast.FuncTypeCols,
 						},
 					},
 				},
@@ -1756,6 +1760,7 @@ func TestParser_ParseStatement(t *testing.T) {
 									Name:       "c",
 								}},
 							},
+							FuncType: ast.FuncTypeCols,
 						},
 					},
 				},