Repository: lf-edge/ekuiper
Pull request #1591 (2023-01-05 02:36:42)

fea(function): add stats function for built-in function
Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>


Patch:
diff --git a/Makefile b/Makefile
index e9e421f8..280bb369 100644
--- a/Makefile
+++ b/Makefile
@@ -56,6 +56,14 @@ build_with_edgex: build_prepare
 	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
 	@echo "Build successfully"
 
+.PHONY: build_with_edgex_and_script
+build_with_edgex_and_script: build_prepare
+	GO111MODULE=on CGO_ENABLED=1 go build -ldflags="-s -w -X main.Version=$(VERSION) -X main.LoadFileType=relative" -tags "edgex include_nats_messaging" -o kuiper cmd/kuiper/main.go
+	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X main.Version=$(VERSION) -X main.LoadFileType=relative" -tags "edgex include_nats_messaging script" -o kuiperd cmd/kuiperd/main.go
+	@if [ ! -z $$(which upx) ]; then upx ./kuiper; upx ./kuiperd; fi
+	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
+	@echo "Build successfully"
+
 .PHONY: build_core
 build_core: build_prepare
 	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X main.Version=$(VERSION) -X main.LoadFileType=relative" -tags core -o kuiperd cmd/kuiperd/main.go
diff --git a/deploy/docker/Dockerfile-dev b/deploy/docker/Dockerfile-dev
index d1b62671..cfeb87a8 100644
--- a/deploy/docker/Dockerfile-dev
+++ b/deploy/docker/Dockerfile-dev
@@ -22,7 +22,7 @@ WORKDIR /go/kuiper
 RUN apt-get clean \
     && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
 
-RUN make build_with_edgex \
+RUN make build_with_edgex_and_script \
     && ln -s /go/kuiper/_build/kuiper-$(git describe --tags --always)-$(go env GOOS)-$(go env GOARCH) /kuiper
 
 RUN ln -s /go/kuiper/deploy/docker/docker-entrypoint.sh /usr/bin/docker-entrypoint.sh
diff --git a/deploy/docker/Dockerfile-slim b/deploy/docker/Dockerfile-slim
index bae98338..9542b417 100644
--- a/deploy/docker/Dockerfile-slim
+++ b/deploy/docker/Dockerfile-slim
@@ -19,7 +19,7 @@ COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN make build_with_edgex
+RUN make build_with_edgex_and_script
 
 FROM debian:buster-20211011 
 
diff --git a/deploy/docker/Dockerfile-slim-python b/deploy/docker/Dockerfile-slim-python
index 90ed91bc..9175d9c3 100644
--- a/deploy/docker/Dockerfile-slim-python
+++ b/deploy/docker/Dockerfile-slim-python
@@ -19,7 +19,7 @@ COPY . /go/kuiper
 
 WORKDIR /go/kuiper
 
-RUN make build_with_edgex
+RUN make build_with_edgex_and_script
 
 FROM python:3.8.12-slim-bullseye
 
diff --git a/etc/ops/script.json b/etc/ops/script.json
index b652a024..081024b1 100644
--- a/etc/ops/script.json
+++ b/etc/ops/script.json
@@ -1,5 +1,5 @@
 {
-  "name": "function",
+  "name": "script",
   "about": {
     "trial": false,
     "author": {
diff --git a/go.mod b/go.mod
index ab5855fd..828db96d 100644
--- a/go.mod
+++ b/go.mod
@@ -24,6 +24,7 @@ require (
 	github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible
 	github.com/mattn/go-sqlite3 v1.14.12
 	github.com/mitchellh/mapstructure v1.5.0
+	github.com/montanaflynn/stats v0.6.6
 	github.com/msgpack-rpc/msgpack-rpc-go v0.0.0-20131026060856-c76397e1782b
 	github.com/pebbe/zmq4 v1.2.7
 	github.com/prometheus/client_golang v1.11.0
diff --git a/go.sum b/go.sum
index c6751649..42fec415 100644
--- a/go.sum
+++ b/go.sum
@@ -211,6 +211,8 @@ github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJ
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/montanaflynn/stats v0.6.6 h1:Duep6KMIDpY4Yo11iFsvyqJDyfzLF9+sndUKT+v64GQ=
+github.com/montanaflynn/stats v0.6.6/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
 github.com/msgpack-rpc/msgpack-rpc-go v0.0.0-20131026060856-c76397e1782b h1:6XGbs6qqORYTSLxOWS7JeWnO79Xvucus6RKHicqMoF4=
 github.com/msgpack-rpc/msgpack-rpc-go v0.0.0-20131026060856-c76397e1782b/go.mod h1:YUWGR1lA7NRnzRjOmYpSgNVip20OCp6NMBwiVFPPhZ8=
 github.com/msgpack/msgpack-go v0.0.0-20130625150338-8224460e6fa3 h1:6pY2f1fJC+u27cqhH0sPkXRquVmGF0VOkLKqraRMYfg=
diff --git a/internal/binder/function/funcs_agg.go b/internal/binder/function/funcs_agg.go
index 8e6bb351..0baa0666 100644
--- a/internal/binder/function/funcs_agg.go
+++ b/internal/binder/function/funcs_agg.go
@@ -19,6 +19,7 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/montanaflynn/stats"
 )
 
 func registerAggFunc() {
@@ -207,6 +208,136 @@ func registerAggFunc() {
 			return nil
 		},
 	}
+	builtins["stddev"] = builtinFunc{
+		fType: ast.FuncTypeAgg,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			arg0 := args[0].([]interface{})
+			if len(arg0) > 0 {
+				float64Slice, err := cast.ToFloat64Slice(arg0, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("requires float64 slice but found %[1]T(%[1]v)", arg0), false
+				}
+				deviation, err := stats.StandardDeviation(float64Slice)
+				if err != nil {
+					return fmt.Errorf("StandardDeviation exec with error: %v", err), false
+				}
+				return deviation, true
+			}
+			return fmt.Errorf("run stddev function error: empty data"), false
+		},
+		val: ValidateOneNumberArg,
+	}
+	builtins["stddevs"] = builtinFunc{
+		fType: ast.FuncTypeAgg,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			arg0 := args[0].([]interface{})
+			if len(arg0) > 0 {
+				float64Slice, err := cast.ToFloat64Slice(arg0, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("requires float64 slice but found %[1]T(%[1]v)", arg0), false
+				}
+				deviation, err := stats.StandardDeviationSample(float64Slice)
+				if err != nil {
+					return fmt.Errorf("StandardDeviationSample exec with error: %v", err), false
+				}
+				return deviation, true
+			}
+			return fmt.Errorf("run stddevs function error: empty data"), false
+		},
+		val: ValidateOneNumberArg,
+	}
+	builtins["var"] = builtinFunc{
+		fType: ast.FuncTypeAgg,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			arg0 := args[0].([]interface{})
+			if len(arg0) > 0 {
+				float64Slice, err := cast.ToFloat64Slice(arg0, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("requires float64 slice but found %[1]T(%[1]v)", arg0), false
+				}
+				deviation, err := stats.Variance(float64Slice)
+				if err != nil {
+					return fmt.Errorf("PopulationVariance exec with error: %v", err), false
+				}
+				return deviation, true
+			}
+			return fmt.Errorf("run var function error: empty data"), false
+		},
+		val: ValidateOneNumberArg,
+	}
+	builtins["vars"] = builtinFunc{
+		fType: ast.FuncTypeAgg,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			arg0 := args[0].([]interface{})
+			if len(arg0) > 0 {
+				float64Slice, err := cast.ToFloat64Slice(arg0, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("requires float64 slice but found %[1]T(%[1]v)", arg0), false
+				}
+				deviation, err := stats.SampleVariance(float64Slice)
+				if err != nil {
+					return fmt.Errorf("SampleVariance exec with error: %v", err), false
+				}
+				return deviation, true
+			}
+			return fmt.Errorf("run vars function error: empty data"), false
+		},
+		val: ValidateOneNumberArg,
+	}
+	builtins["percentile_cont"] = builtinFunc{
+		fType: ast.FuncTypeAgg,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			if err := ValidateLen(2, len(args)); err != nil {
+				return err, false
+			}
+			arg0 := args[0].([]interface{})
+			arg1 := args[1]
+			arg1Float64, err := cast.ToFloat64(arg1, cast.CONVERT_SAMEKIND)
+			if err != nil {
+				return fmt.Errorf("the second parameter requires float64 but found %[1]T(%[1]v)", arg1), false
+			}
+			if len(arg0) > 0 {
+				float64Slice, err := cast.ToFloat64Slice(arg0, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("requires float64 slice but found %[1]T(%[1]v)", arg0), false
+				}
+				deviation, err := stats.Percentile(float64Slice, arg1Float64*100)
+				if err != nil {
+					return fmt.Errorf("percentile exec with error: %v", err), false
+				}
+				return deviation, true
+			}
+			return fmt.Errorf("run percentile_cont function error: empty data"), false
+		},
+		val: ValidateTwoNumberArg,
+	}
+	builtins["percentile_disc"] = builtinFunc{
+		fType: ast.FuncTypeAgg,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			if err := ValidateLen(2, len(args)); err != nil {
+				return err, false
+			}
+			arg0 := args[0].([]interface{})
+			arg1 := args[1]
+			arg1Float64, err := cast.ToFloat64(arg1, cast.CONVERT_SAMEKIND)
+			if err != nil {
+				return fmt.Errorf("the second parameter requires float64 but found %[1]T(%[1]v)", arg1), false
+			}
+			if len(arg0) > 0 {
+				float64Slice, err := cast.ToFloat64Slice(arg0, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("requires float64 slice but found %[1]T(%[1]v)", arg0), false
+				}
+				deviation, err := stats.PercentileNearestRank(float64Slice, arg1Float64*100)
+				if err != nil {
+					return fmt.Errorf("PopulationVariance exec with error: %v", err), false
+				}
+				return deviation, true
+			}
+			return fmt.Errorf("run percentile_cont function error: empty data"), false
+		},
+		val: ValidateTwoNumberArg,
+	}
 }
 
 func getCount(s []interface{}) int {
diff --git a/internal/topo/graph/node.go b/internal/topo/graph/node.go
index 96871c82..a8f0c5f1 100644
--- a/internal/topo/graph/node.go
+++ b/internal/topo/graph/node.go
@@ -60,4 +60,5 @@ type Switch struct {
 
 type Script struct {
 	Script string `json:"script"`
+	IsAgg  bool   `json:"isAgg"`
 }
diff --git a/internal/topo/operator/script_operator.go b/internal/topo/operator/script_operator.go
index eedd02f0..136c20c1 100644
--- a/internal/topo/operator/script_operator.go
+++ b/internal/topo/operator/script_operator.go
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:build script
+
 package operator
 
 import (
@@ -24,9 +26,10 @@ import (
 type ScriptOp struct {
 	vm     *goja.Runtime
 	jsfunc goja.Callable
+	isAgg  bool
 }
 
-func NewScriptOp(script string) (*ScriptOp, error) {
+func NewScriptOp(script string, isAgg bool) (*ScriptOp, error) {
 	vm := goja.New()
 	_, err := vm.RunString(script)
 	if err != nil {
@@ -39,6 +42,7 @@ func NewScriptOp(script string) (*ScriptOp, error) {
 	n := &ScriptOp{
 		vm:     vm,
 		jsfunc: exec,
+		isAgg:  isAgg,
 	}
 	return n, nil
 }
@@ -60,8 +64,34 @@ func (p *ScriptOp) Apply(ctx api.StreamContext, data interface{}, _ *xsql.Functi
 				return &xsql.Tuple{Message: nm, Metadata: input.Metadata, Emitter: input.Emitter, Timestamp: input.Timestamp}
 			}
 		}
+	case xsql.Collection:
+		val, err := p.jsfunc(goja.Undefined(), p.vm.ToValue(input.ToMaps()))
+		if err != nil {
+			return fmt.Errorf("failed to execute script: %v", err)
+		} else {
+			switch nm := val.Export().(type) {
+			case map[string]interface{}:
+				if !p.isAgg {
+					return fmt.Errorf("script node is not aggregate but exec result is aggregated: %v", val.Export())
+
+				}
+				return &xsql.Tuple{Message: nm}
+			case []map[string]interface{}:
+				if p.isAgg {
+					return fmt.Errorf("script node is aggregate but exec result is not aggreagated: %v", val.Export())
+				}
+				w := &xsql.WindowTuples{}
+				for _, v := range nm {
+					if v != nil {
+						w.Content = append(w.Content, &xsql.Tuple{Message: v})
+					}
+				}
+				return w
+			default:
+				return fmt.Errorf("script exec result is not a map or array of map: %v", val.Export())
+			}
+		}
 	default:
 		return fmt.Errorf("run script op invalid input allow tuple only but got %[1]T(%[1]v)", input)
 	}
-	return data
 }
diff --git a/internal/topo/planner/ext_graph_node.go b/internal/topo/planner/ext_graph_node.go
new file mode 100644
index 00000000..d37324ed
--- /dev/null
+++ b/internal/topo/planner/ext_graph_node.go
@@ -0,0 +1,48 @@
+// Copyright 2023 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.
+
+//go:build script
+
+package planner
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/topo/graph"
+	"github.com/lf-edge/ekuiper/internal/topo/operator"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+)
+
+func init() {
+	extNodes["script"] = func(name string, props map[string]interface{}, options *api.RuleOption) (api.TopNode, error) {
+		sop, err := parseScript(props)
+		if err != nil {
+			return nil, err
+		}
+		op := Transform(sop, name, options)
+		return op, nil
+	}
+}
+
+func parseScript(props map[string]interface{}) (*operator.ScriptOp, error) {
+	n := &graph.Script{}
+	err := cast.MapToStruct(props, n)
+	if err != nil {
+		return nil, err
+	}
+	if n.Script == "" {
+		return nil, fmt.Errorf("script node must have script")
+	}
+	return operator.NewScriptOp(n.Script, n.IsAgg)
+}
diff --git a/internal/topo/planner/planner_graph.go b/internal/topo/planner/planner_graph.go
index 737372d2..e83c6457 100644
--- a/internal/topo/planner/planner_graph.go
+++ b/internal/topo/planner/planner_graph.go
@@ -30,6 +30,10 @@ import (
 	"strings"
 )
 
+type genNodeFunc func(name string, props map[string]interface{}, options *api.RuleOption) (api.TopNode, error)
+
+var extNodes = map[string]genNodeFunc{}
+
 // PlanByGraph returns a topo.Topo object by a graph
 func PlanByGraph(rule *api.Rule) (*topo.Topo, error) {
 	ruleGraph := rule.Graph
@@ -93,7 +97,8 @@ func PlanByGraph(rule *api.Rule) (*topo.Topo, error) {
 			if _, ok := ruleGraph.Topo.Edges[nodeName]; !ok {
 				return nil, fmt.Errorf("no edge defined for operator node %s", nodeName)
 			}
-			switch strings.ToLower(gn.NodeType) {
+			nt := strings.ToLower(gn.NodeType)
+			switch nt {
 			case "function":
 				fop, err := parseFunc(gn.Props)
 				if err != nil {
@@ -164,15 +169,16 @@ func PlanByGraph(rule *api.Rule) (*topo.Topo, error) {
 					return nil, fmt.Errorf("create switch %s error: %v", nodeName, err)
 				}
 				nodeMap[nodeName] = op
-			case "script":
-				sop, err := parseScript(gn.Props)
+			default:
+				gnf, ok := extNodes[nt]
+				if !ok {
+					return nil, fmt.Errorf("unknown operator type %s", gn.NodeType)
+				}
+				op, err := gnf(nodeName, gn.Props, rule.Options)
 				if err != nil {
 					return nil, err
 				}
-				op := Transform(sop, nodeName, rule.Options)
 				nodeMap[nodeName] = op
-			default: // TODO other node type
-				return nil, fmt.Errorf("unknown operator type %s", gn.NodeType)
 			}
 		default:
 			return nil, fmt.Errorf("unknown node type %s", gn.Type)
@@ -597,15 +603,3 @@ func parseSwitch(props map[string]interface{}) (*node.SwitchConfig, error) {
 		StopAtFirstMatch: n.StopAtFirstMatch,
 	}, nil
 }
-
-func parseScript(props map[string]interface{}) (*operator.ScriptOp, error) {
-	n := &graph.Script{}
-	err := cast.MapToStruct(props, n)
-	if err != nil {
-		return nil, err
-	}
-	if n.Script == "" {
-		return nil, fmt.Errorf("script node must have script")
-	}
-	return operator.NewScriptOp(n.Script)
-}


Test patch:
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 3bcfe593..d74a90bf 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -59,7 +59,7 @@ jobs:
         cd ../../../../
         cp -r sdk/python/example/pysam plugins/portable/pysam
         cp -r sdk/python/ekuiper plugins/portable/pysam/
-        go test --tags="edgex test" ./...
+        go test --tags="edgex script test" ./...
     - name: run plugins test case
       run: |
         mkdir -p data/test/uploads
diff --git a/internal/binder/function/funcs_agg_test.go b/internal/binder/function/funcs_agg_test.go
index 8f14be89..228cd125 100644
--- a/internal/binder/function/funcs_agg_test.go
+++ b/internal/binder/function/funcs_agg_test.go
@@ -37,15 +37,35 @@ func TestAggExec(t *testing.T) {
 	if !ok {
 		t.Fatal("builtin not found")
 	}
+	fStddev, ok := builtins["stddev"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	fStddevs, ok := builtins["stddevs"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	fVar, ok := builtins["var"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	fVars, ok := builtins["vars"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
 	contextLogger := conf.Log.WithField("rule", "testExec")
 	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
 	tempStore, _ := state.CreateStore("mockRule0", api.AtMostOnce)
 	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("mockRule0", "test", tempStore), 2)
 	var tests = []struct {
-		args []interface{}
-		avg  interface{}
-		max  interface{}
-		min  interface{}
+		args    []interface{}
+		avg     interface{}
+		max     interface{}
+		min     interface{}
+		stddev  interface{}
+		stddevs interface{}
+		var1    interface{}
+		vars    interface{}
 	}{
 		{ // 0
 			args: []interface{}{
@@ -55,9 +75,13 @@ func TestAggExec(t *testing.T) {
 					"self",
 				},
 			},
-			avg: fmt.Errorf("run avg function error: found invalid arg string(foo)"),
-			max: "self",
-			min: "bar",
+			avg:     fmt.Errorf("run avg function error: found invalid arg string(foo)"),
+			max:     "self",
+			min:     "bar",
+			stddev:  fmt.Errorf("requires float64 slice but found []interface {}([foo bar self])"),
+			stddevs: fmt.Errorf("requires float64 slice but found []interface {}([foo bar self])"),
+			var1:    fmt.Errorf("requires float64 slice but found []interface {}([foo bar self])"),
+			vars:    fmt.Errorf("requires float64 slice but found []interface {}([foo bar self])"),
 		}, { // 1
 			args: []interface{}{
 				[]interface{}{
@@ -66,9 +90,13 @@ func TestAggExec(t *testing.T) {
 					int64(200),
 				},
 			},
-			avg: int64(150),
-			max: int64(200),
-			min: int64(100),
+			avg:     int64(150),
+			max:     int64(200),
+			min:     int64(100),
+			stddev:  40.824829046386306,
+			stddevs: float64(50),
+			var1:    1666.6666666666667,
+			vars:    float64(2500),
 		}, { // 2
 			args: []interface{}{
 				[]interface{}{
@@ -77,18 +105,26 @@ func TestAggExec(t *testing.T) {
 					float64(200),
 				},
 			},
-			avg: float64(150),
-			max: float64(200),
-			min: float64(100),
+			avg:     float64(150),
+			max:     float64(200),
+			min:     float64(100),
+			stddev:  40.824829046386306,
+			stddevs: float64(50),
+			var1:    1666.6666666666667,
+			vars:    float64(2500),
 		}, { // 3
 			args: []interface{}{
 				[]interface{}{
 					100, 150, 200,
 				},
 			},
-			avg: int64(150),
-			max: int64(200),
-			min: int64(100),
+			avg:     int64(150),
+			max:     int64(200),
+			min:     int64(100),
+			stddev:  40.824829046386306,
+			stddevs: float64(50),
+			var1:    1666.6666666666667,
+			vars:    float64(2500),
 		},
 	}
 	for i, tt := range tests {
@@ -104,5 +140,105 @@ func TestAggExec(t *testing.T) {
 		if !reflect.DeepEqual(rMin, tt.min) {
 			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rMin, tt.min)
 		}
+		rStddev, _ := fStddev.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rStddev, tt.stddev) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rStddev, tt.stddev)
+		}
+		rStddevs, _ := fStddevs.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rStddevs, tt.stddevs) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rStddevs, tt.stddevs)
+		}
+		rVar, _ := fVar.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rVar, tt.var1) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rVar, tt.var1)
+		}
+		rVars, _ := fVars.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rVars, tt.vars) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rVars, tt.vars)
+		}
+	}
+}
+
+func TestPercentileExec(t *testing.T) {
+	pCont, ok := builtins["percentile_cont"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	pDisc, ok := builtins["percentile_disc"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	contextLogger := conf.Log.WithField("rule", "testExec")
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	tempStore, _ := state.CreateStore("mockRule0", api.AtMostOnce)
+	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("mockRule0", "test", tempStore), 2)
+	var tests = []struct {
+		args  []interface{}
+		pCont interface{}
+		pDisc interface{}
+	}{
+		{ // 0
+			args: []interface{}{
+				[]interface{}{
+					"foo",
+					"bar",
+					"self",
+				},
+				0.25,
+			},
+			pCont: fmt.Errorf("requires float64 slice but found []interface {}([foo bar self])"),
+			pDisc: fmt.Errorf("requires float64 slice but found []interface {}([foo bar self])"),
+		}, { // 1
+			args: []interface{}{
+				[]interface{}{
+					int64(100),
+					int64(150),
+					int64(200),
+				},
+			},
+			pCont: fmt.Errorf("Expect 2 arguments but found 1."),
+			pDisc: fmt.Errorf("Expect 2 arguments but found 1."),
+		}, { // 2
+			args: []interface{}{
+				[]interface{}{
+					int64(100),
+					int64(150),
+					int64(200),
+				},
+				0.5,
+			},
+			pCont: float64(125),
+			pDisc: float64(150),
+		}, { // 3
+			args: []interface{}{
+				[]interface{}{
+					float64(100),
+					float64(150),
+					float64(200),
+				},
+				0.5,
+			},
+			pCont: float64(125),
+			pDisc: float64(150),
+		}, { // 4
+			args: []interface{}{
+				[]interface{}{
+					100, 150, 200,
+				},
+				0.5,
+			},
+			pCont: float64(125),
+			pDisc: float64(150),
+		},
+	}
+	for i, tt := range tests {
+		rCont, _ := pCont.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rCont, tt.pCont) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rCont, tt.pCont)
+		}
+		rDisc, _ := pDisc.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rDisc, tt.pDisc) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rDisc, tt.pCont)
+		}
 	}
 }
diff --git a/internal/topo/operator/script_test.go b/internal/topo/operator/script_test.go
index 6b030a95..4d68f2a4 100644
--- a/internal/topo/operator/script_test.go
+++ b/internal/topo/operator/script_test.go
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:build script
+
 package operator
 
 import (
@@ -26,6 +28,7 @@ import (
 func TestScriptOp(t *testing.T) {
 	var tests = []struct {
 		script string
+		isAgg  bool
 		data   interface{}
 		result interface{}
 	}{
@@ -44,13 +47,102 @@ func TestScriptOp(t *testing.T) {
 				},
 			},
 		},
+		{
+			script: `function exec(msgs) {
+					for (let i = 0; i < msgs.length; i++) {
+  						msgs[i].value = msgs[i].value + 1;
+					} 
+					return msgs
+				}`,
+			data: &xsql.WindowTuples{
+				Content: []xsql.TupleRow{
+					&xsql.Tuple{
+						Emitter: "tbl",
+						Message: xsql.Message{
+							"value": int64(6),
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "tbl",
+						Message: xsql.Message{
+							"value": 8.5,
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "tbl",
+						Message: xsql.Message{
+							"value": 10.2,
+						},
+					},
+				},
+			},
+			result: &xsql.WindowTuples{
+				Content: []xsql.TupleRow{
+					&xsql.Tuple{
+						Emitter: "",
+						Message: xsql.Message{
+							"value": int64(7),
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "",
+						Message: xsql.Message{
+							"value": 9.5,
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "",
+						Message: xsql.Message{
+							"value": 11.2,
+						},
+					},
+				},
+			},
+		},
+		{
+			script: `function exec(msgs) {
+					agg = {value:0}
+					for (let i = 0; i < msgs.length; i++) {
+						agg.value = agg.value + msgs[i].value;
+					} 
+					return agg
+				}`,
+			isAgg: true,
+			data: &xsql.WindowTuples{
+				Content: []xsql.TupleRow{
+					&xsql.Tuple{
+						Emitter: "tbl",
+						Message: xsql.Message{
+							"value": int64(6),
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "tbl",
+						Message: xsql.Message{
+							"value": 8.5,
+						},
+					},
+					&xsql.Tuple{
+						Emitter: "tbl",
+						Message: xsql.Message{
+							"value": 10.2,
+						},
+					},
+				},
+			},
+			result: &xsql.Tuple{
+				Message: xsql.Message{
+					"value": 24.7,
+				},
+			},
+		},
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	contextLogger := conf.Log.WithField("rule", "TestScriptOp_Apply")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	for i, tt := range tests {
-		pp, err := NewScriptOp(tt.script)
+		pp, err := NewScriptOp(tt.script, tt.isAgg)
 		if err != nil {
 			t.Errorf("NewScriptOp error: %v", err)
 			continue