Repository: lf-edge/ekuiper
Pull request #1534 (2022-11-24 12:38:13)

fea(sql): add when condition for analytic function


Patch:
diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index 1c53c514..d8689351 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -186,7 +186,7 @@ jobs:
           - sources/random
           - sources/zmq
           - sources/sql
-          - sources/rtsp
+          - sources/video
           - functions/accumulateWordCount
           - functions/countPlusOne
           - functions/echo
diff --git a/Makefile b/Makefile
index 08414fd4..af08ffe5 100644
--- a/Makefile
+++ b/Makefile
@@ -90,7 +90,7 @@ PLUGINS := sinks/file \
 	sources/random \
 	sources/zmq \
 	sources/sql \
-	sources/rtsp \
+	sources/video \
 	sinks/tdengine \
 	functions/accumulateWordCount \
 	functions/countPlusOne \
diff --git a/docs/directory.json b/docs/directory.json
index 05e25e0a..907d9b4b 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -234,8 +234,8 @@
 									"path": "rules/sources/plugin/random"
 								},
 								{
-									"title": "RTSP 视频源",
-									"path": "rules/sources/plugin/rtsp"
+									"title": "视频源",
+									"path": "rules/sources/plugin/video"
 								}
 							]
 						}
@@ -819,8 +819,8 @@
 									"path": "rules/sources/plugin/random"
 								},
 								{
-									"title": "RTSP source",
-									"path": "rules/sources/plugin/rtsp"
+									"title": "Video source",
+									"path": "rules/sources/plugin/video"
 								}
 							]
 						}
diff --git a/docs/en_US/rules/sources/plugin/rtsp.md b/docs/en_US/rules/sources/plugin/video.md
similarity index 57%
rename from docs/en_US/rules/sources/plugin/rtsp.md
rename to docs/en_US/rules/sources/plugin/video.md
index b2524e0d..1f217eb2 100644
--- a/docs/en_US/rules/sources/plugin/rtsp.md
+++ b/docs/en_US/rules/sources/plugin/video.md
@@ -1,25 +1,25 @@
-# RTSP Source
+# Video Source
 
 <span style="background:green;color:white;">stream source</span>
 <span style="background:green;color:white">scan table source</span>
 
-The source will query rtsp video streams by `ffmpeg` command to get images.
+The source will query video streams such as RTSP encoded stream by `ffmpeg` command to get images.
 
 ## Compile & deploy plugin
 
 ```shell
 # cd $eKuiper_src
-# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Rtsp.so extensions/sources/rtsp/rtsp.go
-# cp plugins/sources/Rtsp.so $eKuiper_install/plugins/sources
-# cp plugins/sources/rtsp.json $eKuiper_install/etc/sources
-# cp plugins/sources/rtsp.yaml $eKuiper_install/etc/sources
+# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Video.so extensions/sources/video/video.go
+# cp plugins/sources/Video.so $eKuiper_install/plugins/sources
+# cp plugins/sources/video.json $eKuiper_install/etc/sources
+# cp plugins/sources/video.yaml $eKuiper_install/etc/sources
 ```
 
 Restart the eKuiper server to activate the plugin.
 
 ## Configuration
 
-The configuration for this source is `$ekuiper/etc/sources/rtsp.yaml`. The format is as below:
+The configuration for this source is `$ekuiper/etc/sources/video.yaml`. The format is as below:
 
 ```yaml
 default:
@@ -35,11 +35,11 @@ dedup:
 ```
 ### Global configurations
 
-Use can specify the global rtsp source settings here. The configuration items specified in `default` section will be taken as default settings for the source when running this source.
+Use can specify the global video source settings here. The configuration items specified in `default` section will be taken as default settings for the source when running this source.
 
 ### url
 
-The url address for the video streaming
+The url address for the video streaming.
 
 ### interval
 
@@ -55,7 +55,7 @@ If you have a specific connection that need to overwrite the default settings, y
 ```
 demo (
 		...
-	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="rtsp");
+	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="video");
 ```
 
 The configuration keys "ext" will be used.
diff --git a/docs/en_US/sqls/built-in_functions.md b/docs/en_US/sqls/built-in_functions.md
index b7d3bf07..66fe8cd3 100644
--- a/docs/en_US/sqls/built-in_functions.md
+++ b/docs/en_US/sqls/built-in_functions.md
@@ -164,6 +164,12 @@ When casting to datetime type, the supported column type and casting rule are:
 
 Analytic functions always use state to do analytic jobs. In streaming processing, analytic functions are evaluated first so that they are not affected by predicates in WHERE clause.
 
+Analytic function call format is, over is optional
+
+```
+AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>] [WHEN <Expression>])
+```
+
 Analytic function computations are performed over all the input events of the current query input, optionally you can limit analytic function to only consider events that match the partition_by_clause.
 
 The syntax is like:
@@ -172,6 +178,15 @@ The syntax is like:
 AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>])
 ```
 
+The analysis function can use the WHEN clause to determine whether the current event is a valid event based on whether the condition is met.
+When it is a valid event, calculate the result and update the state according to the analysis function semantics. When it is an invalid event, ignore the event value and reuse the saved state value.
+
+```
+AnalyticFuncName(<arguments>...) OVER ([WHEN <Expression>])
+```
+
+
+
 | Function    | Example                              | Description                                                                                                                                                                                                                |
 |-------------|--------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | lag         | lag(expr, [offset], [default value]) | Return the former result of expression at offset, if not found, return the default value specified , if default value not set, return nil. if offset and default value not specified, offset is 1 and default value is nil |
@@ -191,6 +206,12 @@ Example function call to get the previous temperature value with the same device
 lag(temperature) OVER (PARTITION BY deviceId)
 ```
 
+Example function call to calculate duration of events:  ts is timestamp, and statusCode1 and statusCode2 are device status in the same event
+
+```text
+ts - lag(ts, 1, ts) OVER (WHEN statusCode1 != statusCode2)
+```
+
 ## Other Functions
 | Function     | Example                              | Description                                                                                                                                                                                                                                                                                                                                                                                       |
 |--------------|--------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
diff --git a/docs/zh_CN/rules/sources/plugin/rtsp.md b/docs/zh_CN/rules/sources/plugin/video.md
similarity index 68%
rename from docs/zh_CN/rules/sources/plugin/rtsp.md
rename to docs/zh_CN/rules/sources/plugin/video.md
index 882352e3..1e098f9a 100644
--- a/docs/zh_CN/rules/sources/plugin/rtsp.md
+++ b/docs/zh_CN/rules/sources/plugin/video.md
@@ -1,25 +1,25 @@
-# RTSP 源
+# 视频源
 
 <span style="background:green;color:white;">stream source</span>
 <span style="background:green;color:white">scan table source</span>
 
-RTSP 源会通过 `ffmpeg` 命令查询 RTSP 视频流获取图片
+视频源会通过 `ffmpeg` 命令查询视频流，例如 RTSP 视频流获取图片
 
 ## 编译和部署插件
 
 ```shell
 # cd $eKuiper_src
-# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Rtsp.so extensions/sources/rtsp/rtsp.go
-# cp plugins/sources/Rtsp.so $eKuiper_install/plugins/sources
-# cp plugins/sources/rtsp.json $eKuiper_install/etc/sources
-# cp plugins/sources/rtsp.yaml $eKuiper_install/etc/sources
+# go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sources/Video.so extensions/sources/video/video.go
+# cp plugins/sources/Video.so $eKuiper_install/plugins/sources
+# cp plugins/sources/video.json $eKuiper_install/etc/sources
+# cp plugins/sources/video.yaml $eKuiper_install/etc/sources
 ```
 
 重新启动 eKuiper 服务器以激活插件。
 
 ## 配置
 
-该源的配置为 `$ekuiper/etc/sources/rtsp.yaml`。格式如下：
+该源的配置为 `$ekuiper/etc/sources/video.yaml`。格式如下：
 
 ```yaml
 default:
@@ -38,7 +38,7 @@ dedup:
 
 ### url
 
-RTSP 视频源地址
+视频源地址
 
 ### interval
 
@@ -54,7 +54,7 @@ RTSP 视频源地址
 ```
 demo (
 		...
-	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="rtsp");
+	) WITH (FORMAT="JSON", CONF_KEY="ext", TYPE="video");
 ```
 
 配置键 "ext" 将被使用。
diff --git a/docs/zh_CN/sqls/built-in_functions.md b/docs/zh_CN/sqls/built-in_functions.md
index 0a6efb12..7312f294 100755
--- a/docs/zh_CN/sqls/built-in_functions.md
+++ b/docs/zh_CN/sqls/built-in_functions.md
@@ -165,14 +165,26 @@ eKuiper 具有许多内置函数，可以对数据执行计算。
 
 分析函数会保持状态来做分析工作。在流式处理规则中，分析函数会首先被执行，这样它们就不会受到 WHERE 子句的影响而必不更新状态。
 
-分析函数的计算是在当前查询输入的所有输入事件上进行的，可以选择限制分析函数只考虑符合 PARTITION BY 子句的事件。
+分析函数完整使用格式为, over 参数可选
+
+```
+AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>] [WHEN <Expression>])
+```
 
+分析函数的计算是在当前查询输入的所有输入事件上进行的，可以选择限制分析函数只考虑符合 PARTITION BY 子句的事件。
 分析函数可以使用 PARTITION BY 子句，语法如下：
 
 ```
 AnalyticFuncName(<arguments>...) OVER ([PARTITION BY <partition key>])
 ```
 
+分析函数可以使用 WHEN 条件判断子句，根据是否满足条件来确定当前事件是否为有效事件。
+当为有效事件时，根据分析函数语意计算结果并更新状态。当为无效事件时，忽略事件值，复用保存的状态值。
+
+```
+AnalyticFuncName(<arguments>...) OVER ([WHEN <Expression>])
+```
+
 | Function    | Example                              | Description                                                                                        |
 |-------------|--------------------------------------|----------------------------------------------------------------------------------------------------|
 | lag         | lag(expr, [offset], [default value]) | 返回表达式前一个值在偏移 offset 处的结果，如果没有找到，则返回默认值，如果没有指定默认值则返回 nil。如果除 expression 外其余参数均未指定，偏移量默认为 1，默认值为 nil |
@@ -192,6 +204,12 @@ lag(temperature)
 lag(temperature) OVER (PARTITION BY deviceId)
 ```
 
+示例3：ts为时间戳，获取设备状态 statusCode1 和 statusCode2 不相等持续时间
+
+```text
+ts - lag(ts, 1, ts) OVER (WHEN statusCode1 != statusCode2)
+```
+
 ## 其它函数
 | 函数           | 示例                                   | 说明                                                                                                                                                                                |
 |--------------|--------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
diff --git a/extensions/sources/rtsp/install.sh b/extensions/sources/video/install.sh
similarity index 97%
rename from extensions/sources/rtsp/install.sh
rename to extensions/sources/video/install.sh
index b108b9a5..ff786eea 100644
--- a/extensions/sources/rtsp/install.sh
+++ b/extensions/sources/video/install.sh
@@ -1,6 +1,6 @@
 #!/bin/sh
 #
-# 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.
diff --git a/extensions/sources/rtsp/rtsp.go b/extensions/sources/video/video.go
similarity index 79%
rename from extensions/sources/rtsp/rtsp.go
rename to extensions/sources/video/video.go
index e7cd43fa..e24847ea 100644
--- a/extensions/sources/rtsp/rtsp.go
+++ b/extensions/sources/video/video.go
@@ -27,12 +27,12 @@ import (
 const RTSP_DEFAULT_INTERVAL = 10000
 const FRAMENUMBER = 5
 
-type RTSPPullSource struct {
+type VideoPullSource struct {
 	url      string
 	interval int
 }
 
-func (rps *RTSPPullSource) Configure(_ string, props map[string]interface{}) error {
+func (rps *VideoPullSource) Configure(_ string, props map[string]interface{}) error {
 
 	if u, ok := props["url"]; ok {
 		if p, ok := u.(string); ok {
@@ -53,18 +53,18 @@ func (rps *RTSPPullSource) Configure(_ string, props map[string]interface{}) err
 	return nil
 }
 
-func (rps *RTSPPullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
+func (rps *VideoPullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
 	rps.initTimerPull(ctx, consumer, errCh)
 }
 
-func (rps *RTSPPullSource) Close(ctx api.StreamContext) error {
+func (rps *VideoPullSource) Close(ctx api.StreamContext) error {
 	logger := ctx.GetLogger()
-	logger.Infof("Closing rtsp pull source")
+	logger.Infof("Closing video pull source")
 
 	return nil
 }
 
-func (rps *RTSPPullSource) initTimerPull(ctx api.StreamContext, consumer chan<- api.SourceTuple, _ chan<- error) {
+func (rps *VideoPullSource) initTimerPull(ctx api.StreamContext, consumer chan<- api.SourceTuple, _ chan<- error) {
 	ticker := time.NewTicker(time.Millisecond * time.Duration(rps.interval))
 	logger := ctx.GetLogger()
 	defer ticker.Stop()
@@ -91,7 +91,7 @@ func (rps *RTSPPullSource) initTimerPull(ctx api.StreamContext, consumer chan<-
 	}
 }
 
-func (rps *RTSPPullSource) readFrameAsJpeg(ctx api.StreamContext) *bytes.Buffer {
+func (rps *VideoPullSource) readFrameAsJpeg(ctx api.StreamContext) *bytes.Buffer {
 	logger := ctx.GetLogger()
 	buf := bytes.NewBuffer(nil)
 	err := ffmpeg.Input(rps.url).
@@ -106,6 +106,6 @@ func (rps *RTSPPullSource) readFrameAsJpeg(ctx api.StreamContext) *bytes.Buffer
 	return buf
 }
 
-func Rtsp() api.Source {
-	return &RTSPPullSource{}
+func Video() api.Source {
+	return &VideoPullSource{}
 }
diff --git a/extensions/sources/rtsp/rtsp.json b/extensions/sources/video/video.json
similarity index 75%
rename from extensions/sources/rtsp/rtsp.json
rename to extensions/sources/video/video.json
index 4a8fdf64..567b5c5b 100644
--- a/extensions/sources/rtsp/rtsp.json
+++ b/extensions/sources/video/video.json
@@ -9,12 +9,12 @@
 			"website": "https://www.emqx.io"
 		},
 		"helpUrl": {
-			"en_US": "https://github.com/lf-edge/ekuiper/tree/master/docs/en_US/rules/sources/plugin/rtsp.md",
-			"zh_CN": "https://github.com/lf-edge/ekuiper/tree/master/docs/zh_CN/rules/sources/plugin/rtsp.md"
+			"en_US": "https://github.com/lf-edge/ekuiper/tree/master/docs/en_US/rules/sources/plugin/video.md",
+			"zh_CN": "https://github.com/lf-edge/ekuiper/tree/master/docs/zh_CN/rules/sources/plugin/video.md"
 		},
 		"description": {
-			"en_US": "eKuiper provides plug in support for RTSP source stream, which can pull images from RTSP server",
-			"zh_CN": "eKuiper 为提取 RTSP 源中的图片流提供了插件支持，该支持可从 RTSP 服务器提取图片消息并输入 eKuiper 处理管道。"
+			"en_US": "eKuiper provides plug in support for video stream, which can pull images from a video stream",
+			"zh_CN": "eKuiper 为提取视频源中的图片流提供了插件支持，该支持可从视频流提取图片消息并输入 eKuiper 处理管道。"
 		}
 	},
 	"properties": {
@@ -61,8 +61,8 @@
 		"category": "source",
 		"icon": "iconPath",
 		"label": {
-			"en_US": "RTSP PULL",
-			"zh_CN": "RTSP PULL"
+			"en_US": "Video",
+			"zh_CN": "视频"
 		}
 	}
 }
diff --git a/extensions/sources/rtsp/rtsp.yaml b/extensions/sources/video/video.yaml
similarity index 74%
rename from extensions/sources/rtsp/rtsp.yaml
rename to extensions/sources/video/video.yaml
index 359810dc..3485f860 100644
--- a/extensions/sources/rtsp/rtsp.yaml
+++ b/extensions/sources/video/video.yaml
@@ -1,4 +1,4 @@
-#Global RTSP configurations
+#Global video source configurations
 default:
   # url of the request server address
   # username password, can be this format: rtsp://xxxxxx:xxxxxx@192.168.0.64:554/h265/ch1/main/av_stream
diff --git a/internal/binder/function/funcs_analytic.go b/internal/binder/function/funcs_analytic.go
index 5017dc0d..d6450c13 100644
--- a/internal/binder/function/funcs_analytic.go
+++ b/internal/binder/function/funcs_analytic.go
@@ -35,6 +35,13 @@ func registerAnalyticFunc() {
 			if ignoreNull && args[1] == nil {
 				return nil, true
 			}
+			validData, ok := args[len(args)-2].(bool)
+			if !ok {
+				return fmt.Errorf("when arg is not a bool but got %v", args[len(args)-2]), false
+			}
+			if !validData {
+				return nil, true
+			}
 			key := args[len(args)-1].(string)
 			lv, err := ctx.GetState(key)
 			if err != nil {
@@ -62,17 +69,25 @@ func registerAnalyticFunc() {
 	builtins["had_changed"] = builtinFunc{
 		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
-			l := len(args) - 1
+			l := len(args) - 2
 			if l <= 1 {
 				return fmt.Errorf("expect more than one arg but got %d", len(args)), false
 			}
+			validData, ok := args[len(args)-2].(bool)
+			if !ok {
+				return fmt.Errorf("when arg is not a bool but got %v", args[len(args)-2]), false
+			}
+			if !validData {
+				return false, true
+			}
 			ignoreNull, ok := args[0].(bool)
 			if !ok {
 				return fmt.Errorf("first arg is not a bool but got %v", args[0]), false
 			}
-			key := args[l].(string)
+			key := args[len(args)-1].(string)
+			paraLen := len(args) - 2
 			result := false
-			for i := 1; i < l; i++ {
+			for i := 1; i < paraLen; i++ {
 				v := args[i]
 				k := key + strconv.Itoa(i)
 				if ignoreNull && v == nil {
@@ -106,23 +121,31 @@ func registerAnalyticFunc() {
 	builtins["lag"] = builtinFunc{
 		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
-			l := len(args) - 1
-			key := args[l].(string)
+			l := len(args) - 2
 			if l != 1 && l != 2 && l != 3 {
 				return fmt.Errorf("expect one two or three args but got %d", l), false
 			}
+			key := args[len(args)-1].(string)
 			v, err := ctx.GetState(key)
 			if err != nil {
 				return fmt.Errorf("error getting state for %s: %v", key, err), false
 			}
+			validData, ok := args[len(args)-2].(bool)
+			if !ok {
+				return fmt.Errorf("when arg is not a bool but got %v", args[len(args)-2]), false
+			}
+			paraLen := len(args) - 2
+			var rq *ringqueue = nil
+			var rtnVal interface{} = nil
+
+			// first time call, need create state for lag
 			if v == nil {
 				size := 0
 				var dftVal interface{} = nil
-				if l == 3 {
+				if paraLen == 3 {
 					dftVal = args[2]
 				}
-				// first time call, need create state for lag
-				if l == 1 {
+				if paraLen == 1 {
 					size = 1
 				} else {
 					siz, ok := args[1].(int)
@@ -131,30 +154,27 @@ func registerAnalyticFunc() {
 					}
 					size = siz
 				}
-
-				rq := newRingqueue(size)
+				rq = newRingqueue(size)
 				rq.fill(dftVal)
-
-				rtnVal, _ := rq.fetch()
-				rq.append(args[0])
 				err := ctx.PutState(key, rq)
 				if err != nil {
 					return fmt.Errorf("error setting state for %s: %v", key, err), false
 				}
-				return rtnVal, true
 			} else {
-				rq, ok := v.(*ringqueue)
-				if !ok {
-					return fmt.Errorf("error getting state for %s: %v", key, err), false
-				}
-				rtnVal, _ := rq.fetch()
+				rq, _ = v.(*ringqueue)
+			}
+
+			if validData {
+				rtnVal, _ = rq.fetch()
 				rq.append(args[0])
 				err := ctx.PutState(key, rq)
 				if err != nil {
 					return fmt.Errorf("error setting state for %s: %v", key, err), false
 				}
-				return rtnVal, true
+			} else {
+				rtnVal, _ = rq.peek()
 			}
+			return rtnVal, true
 		},
 		val: func(_ api.FunctionContext, args []ast.Expr) error {
 			l := len(args)
@@ -178,18 +198,24 @@ func registerAnalyticFunc() {
 	builtins["latest"] = builtinFunc{
 		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
-			l := len(args) - 1
-			key := args[l].(string)
+			l := len(args) - 2
 			if l != 1 && l != 2 {
 				return fmt.Errorf("expect one or two args but got %d", l), false
 			}
+			paraLen := len(args) - 2
+			key := args[len(args)-1].(string)
+			validData, ok := args[len(args)-2].(bool)
+			if !ok {
+				return fmt.Errorf("when arg is not a bool but got %v", args[len(args)-2]), false
+			}
+
 			if args[0] == nil {
 				v, err := ctx.GetState(key)
 				if err != nil {
 					return fmt.Errorf("error getting state for %s: %v", key, err), false
 				}
 				if v == nil {
-					if l == 2 {
+					if paraLen == 2 {
 						return args[1], true
 					} else {
 						return nil, true
@@ -198,7 +224,9 @@ func registerAnalyticFunc() {
 					return v, true
 				}
 			} else {
-				ctx.PutState(key, args[0])
+				if validData {
+					ctx.PutState(key, args[0])
+				}
 				return args[0], true
 			}
 		},
diff --git a/internal/binder/function/funcs_misc.go b/internal/binder/function/funcs_misc.go
index bac35685..253996eb 100644
--- a/internal/binder/function/funcs_misc.go
+++ b/internal/binder/function/funcs_misc.go
@@ -569,7 +569,7 @@ func (p *ringqueue) append(item interface{}) bool {
 	return true
 }
 
-// fetch get the first item in the cache
+// fetch get the first item in the cache and remove
 func (p *ringqueue) fetch() (interface{}, bool) {
 	if p.l == 0 {
 		return nil, false
@@ -583,6 +583,15 @@ func (p *ringqueue) fetch() (interface{}, bool) {
 	return result, true
 }
 
+// peek get the first item in the cache but keep it
+func (p *ringqueue) peek() (interface{}, bool) {
+	if p.l == 0 {
+		return nil, false
+	}
+	result := p.data[p.h]
+	return result, true
+}
+
 func (p *ringqueue) isFull() bool {
 	return p.l == p.size
 }
diff --git a/internal/topo/planner/analyzer.go b/internal/topo/planner/analyzer.go
index 379d2e42..9d0b79f5 100644
--- a/internal/topo/planner/analyzer.go
+++ b/internal/topo/planner/analyzer.go
@@ -133,6 +133,7 @@ func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt,
 					Args:        f.Args,
 					CachedField: f.CachedField,
 					Partition:   f.Partition,
+					WhenExpr:    f.WhenExpr,
 				})
 			}
 		}
@@ -156,6 +157,7 @@ func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt,
 					Args:        f.Args,
 					CachedField: f.CachedField,
 					Partition:   f.Partition,
+					WhenExpr:    f.WhenExpr,
 				})
 			}
 		}
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 54fd0a9a..adee758a 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -860,7 +860,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 		}
 		c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
 		p.fn += 1
-		e := p.parseOverPartition(c)
+		e := p.parseOver(c)
 		return c, e
 	} else {
 		if err != nil {
@@ -1494,7 +1494,7 @@ func (p *Parser) inmeta() bool {
 	return p.inFunc == "meta" || p.inFunc == "mqtt"
 }
 
-func (p *Parser) parseOverPartition(c *ast.Call) error {
+func (p *Parser) parseOver(c *ast.Call) error {
 	if tok, _ := p.scanIgnoreWhitespace(); tok != ast.OVER {
 		p.unscan()
 		return nil
@@ -1516,20 +1516,36 @@ func (p *Parser) parseOverPartition(c *ast.Call) error {
 							break
 						}
 					}
-					if ttt, _ := p.scanIgnoreWhitespace(); ttt != ast.RPAREN {
-						return fmt.Errorf("found %q, expect right parentheses after PARTITION BY", ttt)
-					}
 					if len(pe.Exprs) == 0 {
 						return fmt.Errorf("PARTITION BY must have at least one expression.")
 					}
 					c.Partition = pe
-					return nil
 				} else {
 					return fmt.Errorf("found %q, expected by after partition.", l1)
 				}
 			} else {
-				return fmt.Errorf("Found %q after OVER (, expect partition by.", tok1)
+				p.unscan()
+			}
+
+			if t, _ := p.scanIgnoreWhitespace(); t == ast.WHEN {
+				if exp, err := p.ParseExpr(); err != nil {
+					return err
+				} else {
+					c.WhenExpr = exp
+				}
+			} else {
+				p.unscan()
+			}
+			if c.Partition != nil || c.WhenExpr != nil {
+				if ttt, _ := p.scanIgnoreWhitespace(); ttt != ast.RPAREN {
+					return fmt.Errorf("Found %q, expect right parentheses after OVER ", ttt)
+				}
+			}
+			if c.Partition == nil && c.WhenExpr == nil {
+				ttt, _ := p.scanIgnoreWhitespace()
+				return fmt.Errorf("Found %q after OVER (, expect partition by or when.", ttt)
 			}
+			return nil
 		} else {
 			return fmt.Errorf("Found %q after OVER, expect parentheses.", tok1)
 		}
diff --git a/internal/xsql/row.go b/internal/xsql/row.go
index 57ab8ad2..a3e5521f 100644
--- a/internal/xsql/row.go
+++ b/internal/xsql/row.go
@@ -154,7 +154,10 @@ func (d *AffiliateRow) MergeMap(cachedMap map[string]interface{}) {
 	d.lock.RLock()
 	defer d.lock.RUnlock()
 	for k, v := range d.CalCols {
-		cachedMap[k] = v
+		// Do not write out the internal fields
+		if !strings.HasPrefix(k, "$$") {
+			cachedMap[k] = v
+		}
 	}
 	for k, v := range d.AliasMap {
 		cachedMap[k] = v
diff --git a/internal/xsql/valuer.go b/internal/xsql/valuer.go
index 45300786..faa22a3d 100644
--- a/internal/xsql/valuer.go
+++ b/internal/xsql/valuer.go
@@ -347,7 +347,22 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 						return fmt.Errorf("unknown function type")
 					}
 				}
-				if function.IsAnalyticFunc(expr.Name) { // analytic func must put the partition key into the args
+				if function.IsAnalyticFunc(expr.Name) {
+					// this data should be recorded or not ? default answer is yes
+					if expr.WhenExpr != nil {
+						validData := true
+						temp := v.Eval(expr.WhenExpr)
+						whenExprVal, ok := temp.(bool)
+						if ok {
+							validData = whenExprVal
+						}
+
+						args = append(args, validData)
+					} else {
+						args = append(args, true)
+					}
+
+					// analytic func must put the partition key into the args
 					if expr.Partition != nil && len(expr.Partition.Exprs) > 0 {
 						pk := ""
 						for _, pe := range expr.Partition.Exprs {
diff --git a/pkg/ast/expr.go b/pkg/ast/expr.go
index b5aae5d0..4b8346d2 100644
--- a/pkg/ast/expr.go
+++ b/pkg/ast/expr.go
@@ -142,6 +142,7 @@ type Call struct {
 	CachedField string
 	Cached      bool
 	Partition   *PartitionExpr
+	WhenExpr    Expr
 }
 
 func (c *Call) expr()    {}
@@ -152,6 +153,9 @@ type PartitionExpr struct {
 	Exprs []Expr
 }
 
+func (pe *PartitionExpr) expr() {}
+func (pe *PartitionExpr) node() {}
+
 type BinaryExpr struct {
 	OP  Token
 	LHS Expr
diff --git a/pkg/ast/visitor.go b/pkg/ast/visitor.go
index 4711eca8..e013ca45 100644
--- a/pkg/ast/visitor.go
+++ b/pkg/ast/visitor.go
@@ -94,6 +94,16 @@ func Walk(v Visitor, node Node) {
 			Walk(v, expr)
 		}
 
+		if n.Partition != nil {
+			for _, expr := range n.Partition.Exprs {
+				Walk(v, expr)
+			}
+		}
+
+		if n.WhenExpr != nil {
+			Walk(v, n.WhenExpr)
+		}
+
 	case *ParenExpr:
 		Walk(v, n.Expr)
 


Test patch:
diff --git a/internal/binder/function/funcs_analytic_test.go b/internal/binder/function/funcs_analytic_test.go
index a575519f..c5fedbda 100644
--- a/internal/binder/function/funcs_analytic_test.go
+++ b/internal/binder/function/funcs_analytic_test.go
@@ -80,17 +80,11 @@ func TestChangedColExec(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
-			args: []interface{}{
-				"foo",
-				"bar",
-				"self",
-			},
-			result: fmt.Errorf("first arg is not a bool but got foo"),
-		}, { // 1
+		{ // 1
 			args: []interface{}{
 				true,
 				"bar",
+				true,
 				"self",
 			},
 			result: "bar",
@@ -98,6 +92,7 @@ func TestChangedColExec(t *testing.T) {
 			args: []interface{}{
 				true,
 				"bar",
+				true,
 				"self",
 			},
 			result: nil,
@@ -105,6 +100,7 @@ func TestChangedColExec(t *testing.T) {
 			args: []interface{}{
 				true,
 				"baz",
+				true,
 				"self",
 			},
 			result: "baz",
@@ -112,6 +108,7 @@ func TestChangedColExec(t *testing.T) {
 			args: []interface{}{
 				false,
 				nil,
+				true,
 				"self",
 			},
 			result: nil,
@@ -119,6 +116,7 @@ func TestChangedColExec(t *testing.T) {
 			args: []interface{}{
 				false,
 				"baz",
+				true,
 				"self",
 			},
 			result: "baz",
@@ -126,6 +124,7 @@ func TestChangedColExec(t *testing.T) {
 			args: []interface{}{
 				true,
 				"foo",
+				true,
 				"self",
 			},
 			result: "foo",
@@ -152,17 +151,82 @@ func TestChangedColPartition(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
+		{ // 1
 			args: []interface{}{
-				"foo",
+				true,
 				"bar",
+				true,
+				"2",
+			},
+			result: "bar",
+		}, { // 2
+			args: []interface{}{
+				true,
+				"bar",
+				true,
 				"1",
 			},
-			result: fmt.Errorf("first arg is not a bool but got foo"),
-		}, { // 1
+			result: "bar",
+		}, { // 3
+			args: []interface{}{
+				true,
+				"baz",
+				true,
+				"2",
+			},
+			result: "baz",
+		}, { // 4
+			args: []interface{}{
+				false,
+				nil,
+				true,
+				"1",
+			},
+			result: nil,
+		}, { // 5
+			args: []interface{}{
+				false,
+				"baz",
+				true,
+				"2",
+			},
+			result: nil,
+		}, { // 6
+			args: []interface{}{
+				true,
+				"foo",
+				true,
+				"1",
+			},
+			result: "foo",
+		},
+	}
+	for i, tt := range tests {
+		result, _ := f.exec(fctx, tt.args)
+		if !reflect.DeepEqual(result, tt.result) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
+		}
+	}
+}
+
+func TestChangedColPartitionWithWhen(t *testing.T) {
+	f, ok := builtins["changed_col"]
+	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{}
+		result interface{}
+	}{
+		{ // 1
 			args: []interface{}{
 				true,
 				"bar",
+				true,
 				"2",
 			},
 			result: "bar",
@@ -170,6 +234,7 @@ func TestChangedColPartition(t *testing.T) {
 			args: []interface{}{
 				true,
 				"bar",
+				true,
 				"1",
 			},
 			result: "bar",
@@ -177,13 +242,23 @@ func TestChangedColPartition(t *testing.T) {
 			args: []interface{}{
 				true,
 				"baz",
+				true,
 				"2",
 			},
 			result: "baz",
+		}, { // 3.1 copy of 3 with baz changed to bar and when condition false
+			args: []interface{}{
+				true,
+				"bar",
+				false,
+				"2",
+			},
+			result: nil,
 		}, { // 4
 			args: []interface{}{
 				false,
 				nil,
+				true,
 				"1",
 			},
 			result: nil,
@@ -191,6 +266,7 @@ func TestChangedColPartition(t *testing.T) {
 			args: []interface{}{
 				false,
 				"baz",
+				true,
 				"2",
 			},
 			result: nil,
@@ -198,9 +274,18 @@ func TestChangedColPartition(t *testing.T) {
 			args: []interface{}{
 				true,
 				"foo",
+				true,
 				"1",
 			},
 			result: "foo",
+		}, { // 7
+			args: []interface{}{
+				true,
+				"bar",
+				false,
+				"1",
+			},
+			result: nil,
 		},
 	}
 	for i, tt := range tests {
@@ -278,6 +363,7 @@ func TestHadChangedExec(t *testing.T) {
 				"foo",
 				"bar",
 				"baz",
+				true,
 				"self",
 			},
 			result: fmt.Errorf("first arg is not a bool but got foo"),
@@ -285,6 +371,7 @@ func TestHadChangedExec(t *testing.T) {
 			args: []interface{}{
 				"foo",
 				"bar",
+				true,
 				"self",
 			},
 			result: fmt.Errorf("first arg is not a bool but got foo"),
@@ -293,6 +380,7 @@ func TestHadChangedExec(t *testing.T) {
 				true,
 				"bar",
 				20,
+				true,
 				"self",
 			},
 			result: true,
@@ -301,6 +389,7 @@ func TestHadChangedExec(t *testing.T) {
 				true,
 				"baz",
 				44,
+				true,
 				"self",
 			},
 			result: true,
@@ -309,6 +398,7 @@ func TestHadChangedExec(t *testing.T) {
 				true,
 				"baz",
 				44,
+				true,
 				"self",
 			},
 			result: false,
@@ -317,6 +407,7 @@ func TestHadChangedExec(t *testing.T) {
 				true,
 				"foo",
 				44,
+				true,
 				"self",
 			},
 			result: true,
@@ -325,6 +416,7 @@ func TestHadChangedExec(t *testing.T) {
 				true,
 				"foo",
 				nil,
+				true,
 				"self",
 			},
 			result: false,
@@ -333,6 +425,7 @@ func TestHadChangedExec(t *testing.T) {
 				true,
 				"foo",
 				44,
+				true,
 				"self",
 			},
 			result: false,
@@ -341,6 +434,7 @@ func TestHadChangedExec(t *testing.T) {
 				true,
 				"baz",
 				44,
+				true,
 				"self",
 			},
 			result: true,
@@ -372,6 +466,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				"foo",
 				"bar",
 				"baz",
+				true,
 				"self",
 			},
 			result: fmt.Errorf("first arg is not a bool but got foo"),
@@ -379,6 +474,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 			args: []interface{}{
 				"foo",
 				"bar",
+				true,
 				"self",
 			},
 			result: fmt.Errorf("first arg is not a bool but got foo"),
@@ -387,6 +483,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				"bar",
 				20,
+				true,
 				"self",
 			},
 			result: true,
@@ -395,6 +492,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				"baz",
 				nil,
+				true,
 				"self",
 			},
 			result: true,
@@ -403,6 +501,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				"baz",
 				44,
+				true,
 				"self",
 			},
 			result: true,
@@ -411,6 +510,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				nil,
 				44,
+				true,
 				"self",
 			},
 			result: true,
@@ -419,6 +519,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				"baz",
 				44,
+				true,
 				"self",
 			},
 			result: true,
@@ -427,6 +528,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				"baz",
 				44,
+				true,
 				"self",
 			},
 			result: false,
@@ -435,6 +537,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				nil,
 				nil,
+				true,
 				"self",
 			},
 			result: true,
@@ -443,6 +546,7 @@ func TestHadChangedExecAllowNull(t *testing.T) {
 				false,
 				"baz",
 				44,
+				true,
 				"self",
 			},
 			result: true,
@@ -474,6 +578,7 @@ func TestHadChangedPartition(t *testing.T) {
 				"foo",
 				"bar",
 				"baz",
+				true,
 				"1",
 			},
 			result: fmt.Errorf("first arg is not a bool but got foo"),
@@ -481,6 +586,7 @@ func TestHadChangedPartition(t *testing.T) {
 			args: []interface{}{
 				"foo",
 				"bar",
+				true,
 				"1",
 			},
 			result: fmt.Errorf("first arg is not a bool but got foo"),
@@ -489,6 +595,7 @@ func TestHadChangedPartition(t *testing.T) {
 				true,
 				"bar",
 				20,
+				true,
 				"3",
 			},
 			result: true,
@@ -497,6 +604,7 @@ func TestHadChangedPartition(t *testing.T) {
 				true,
 				"baz",
 				44,
+				true,
 				"2",
 			},
 			result: true,
@@ -505,6 +613,7 @@ func TestHadChangedPartition(t *testing.T) {
 				true,
 				"baz",
 				44,
+				true,
 				"2",
 			},
 			result: false,
@@ -513,6 +622,7 @@ func TestHadChangedPartition(t *testing.T) {
 				true,
 				"foo",
 				44,
+				true,
 				"3",
 			},
 			result: true,
@@ -521,6 +631,7 @@ func TestHadChangedPartition(t *testing.T) {
 				true,
 				"foo",
 				nil,
+				true,
 				"1",
 			},
 			result: true,
@@ -529,6 +640,7 @@ func TestHadChangedPartition(t *testing.T) {
 				true,
 				"foo",
 				44,
+				true,
 				"2",
 			},
 			result: true,
@@ -537,6 +649,7 @@ func TestHadChangedPartition(t *testing.T) {
 				true,
 				"baz",
 				44,
+				true,
 				"3",
 			},
 			result: true,
@@ -550,6 +663,118 @@ func TestHadChangedPartition(t *testing.T) {
 	}
 }
 
+func TestHadChangedPartitionWithWhen(t *testing.T) {
+	f, ok := builtins["had_changed"]
+	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), 1)
+	var tests = []struct {
+		args   []interface{}
+		result interface{}
+	}{
+		{ // 0
+			args: []interface{}{
+				"foo",
+				"bar",
+				"baz",
+				true,
+				"1",
+			},
+			result: fmt.Errorf("first arg is not a bool but got foo"),
+		}, { // 1
+			args: []interface{}{
+				"foo",
+				"bar",
+				true,
+				"1",
+			},
+			result: fmt.Errorf("first arg is not a bool but got foo"),
+		}, { // 2
+			args: []interface{}{
+				true,
+				"bar",
+				20,
+				true,
+				"3",
+			},
+			result: true,
+		}, { // 3
+			args: []interface{}{
+				true,
+				"baz",
+				44,
+				true,
+				"2",
+			},
+			result: true,
+		}, { // 4
+			args: []interface{}{
+				true,
+				"baz",
+				44,
+				true,
+				"2",
+			},
+			result: false,
+		}, { // 5
+			args: []interface{}{
+				true,
+				"baz",
+				44,
+				true,
+				"2",
+			},
+			result: false,
+		}, { // 6
+			args: []interface{}{
+				true,
+				"foo",
+				45,
+				false,
+				"2",
+			},
+			result: false,
+		}, { // 7
+			args: []interface{}{
+				true,
+				"foo",
+				nil,
+				true,
+				"1",
+			},
+			result: true,
+		}, { // 8
+			args: []interface{}{
+				true,
+				"foo",
+				44,
+				true,
+				"2",
+			},
+			result: true,
+		}, { // 9
+			args: []interface{}{
+				true,
+				"baz",
+				44,
+				false,
+				"3",
+			},
+			result: false,
+		},
+	}
+	for i, tt := range tests {
+		result, _ := f.exec(fctx, tt.args)
+		if !reflect.DeepEqual(result, tt.result) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
+		}
+	}
+}
+
 func TestLagExec(t *testing.T) {
 	f, ok := builtins["lag"]
 	if !ok {
@@ -563,12 +788,10 @@ func TestLagExec(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
-			args:   []interface{}{"self"},
-			result: fmt.Errorf("expect one two or three args but got 0"),
-		}, { // 1
+		{ // 1
 			args: []interface{}{
 				"foo",
+				true,
 				"self",
 			},
 			result: nil,
@@ -576,6 +799,7 @@ func TestLagExec(t *testing.T) {
 		{ // 2
 			args: []interface{}{
 				"bar",
+				true,
 				"self",
 			},
 			result: "foo",
@@ -583,6 +807,7 @@ func TestLagExec(t *testing.T) {
 		{ // 3
 			args: []interface{}{
 				"bar",
+				true,
 				"self",
 			},
 			result: "bar",
@@ -590,6 +815,7 @@ func TestLagExec(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				"foo",
+				true,
 				"self",
 			},
 			result: "bar",
@@ -597,6 +823,7 @@ func TestLagExec(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				"foo",
+				true,
 				"self",
 			},
 			result: "foo",
@@ -623,12 +850,10 @@ func TestLagPartition(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
-			args:   []interface{}{"self"},
-			result: fmt.Errorf("expect one two or three args but got 0"),
-		}, { // 1
+		{ // 1
 			args: []interface{}{
 				"foo",
+				true,
 				"1",
 			},
 			result: nil,
@@ -636,6 +861,7 @@ func TestLagPartition(t *testing.T) {
 		{ // 2
 			args: []interface{}{
 				"bar",
+				true,
 				"1",
 			},
 			result: "foo",
@@ -643,6 +869,7 @@ func TestLagPartition(t *testing.T) {
 		{ // 3
 			args: []interface{}{
 				"bar",
+				true,
 				"2",
 			},
 			result: nil,
@@ -650,6 +877,7 @@ func TestLagPartition(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				"foo",
+				true,
 				"1",
 			},
 			result: "bar",
@@ -657,6 +885,7 @@ func TestLagPartition(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				"foo",
+				true,
 				"2",
 			},
 			result: "bar",
@@ -670,7 +899,7 @@ func TestLagPartition(t *testing.T) {
 	}
 }
 
-func TestLagExecIndexWithDefaultValue(t *testing.T) {
+func TestLagExecWithWhen(t *testing.T) {
 	f, ok := builtins["lag"]
 	if !ok {
 		t.Fatal("builtin not found")
@@ -683,20 +912,136 @@ func TestLagExecIndexWithDefaultValue(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
+		{ // 1
 			args: []interface{}{
 				"foo",
+				true,
+				"self",
+			},
+			result: nil,
+		},
+		{ // 2
+			args: []interface{}{
 				"bar",
-				"baz",
-				"baw",
+				false,
 				"self",
 			},
-			result: fmt.Errorf("expect one two or three args but got 4"),
-		}, { // 1
+			result: "foo",
+		},
+		{ // 3
+			args: []interface{}{
+				"bar",
+				true,
+				"self",
+			},
+			result: "foo",
+		},
+		{ // 4
+			args: []interface{}{
+				"foo",
+				false,
+				"self",
+			},
+			result: "bar",
+		},
+		{ // 4
+			args: []interface{}{
+				"foo",
+				true,
+				"self",
+			},
+			result: "bar",
+		},
+	}
+	for i, tt := range tests {
+		result, _ := f.exec(fctx, tt.args)
+		if !reflect.DeepEqual(result, tt.result) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
+		}
+	}
+}
+
+func TestLagPartitionWithWhen(t *testing.T) {
+	f, ok := builtins["lag"]
+	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{}
+		result interface{}
+	}{
+		{ // 1
+			args: []interface{}{
+				"foo",
+				true,
+				"1",
+			},
+			result: nil,
+		},
+		{ // 2
+			args: []interface{}{
+				"bar",
+				false,
+				"1",
+			},
+			result: "foo",
+		},
+		{ // 3
+			args: []interface{}{
+				"bar",
+				true,
+				"2",
+			},
+			result: nil,
+		},
+		{ // 4
+			args: []interface{}{
+				"foo",
+				true,
+				"1",
+			},
+			result: "foo",
+		},
+		{ // 4
+			args: []interface{}{
+				"foo",
+				true,
+				"2",
+			},
+			result: "bar",
+		},
+	}
+	for i, tt := range tests {
+		result, _ := f.exec(fctx, tt.args)
+		if !reflect.DeepEqual(result, tt.result) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
+		}
+	}
+}
+
+func TestLagExecIndexWithDefaultValue(t *testing.T) {
+	f, ok := builtins["lag"]
+	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{}
+		result interface{}
+	}{
+		{ // 1
 			args: []interface{}{
 				"bar",
 				2,
 				"no result",
+				true,
 				"self",
 			},
 			result: "no result",
@@ -706,6 +1051,7 @@ func TestLagExecIndexWithDefaultValue(t *testing.T) {
 				"bar",
 				2,
 				"no result",
+				true,
 				"self",
 			},
 			result: "no result",
@@ -715,6 +1061,7 @@ func TestLagExecIndexWithDefaultValue(t *testing.T) {
 				"foo",
 				2,
 				"no result",
+				true,
 				"self",
 			},
 			result: "bar",
@@ -724,6 +1071,7 @@ func TestLagExecIndexWithDefaultValue(t *testing.T) {
 				"foo",
 				2,
 				"no result",
+				true,
 				"self",
 			},
 			result: "bar",
@@ -733,6 +1081,7 @@ func TestLagExecIndexWithDefaultValue(t *testing.T) {
 				"foo",
 				2,
 				"no result",
+				true,
 				"self",
 			},
 			result: "foo",
@@ -759,19 +1108,11 @@ func TestLagExecIndex(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
-			args: []interface{}{
-				"foo",
-				"bar",
-				"baz",
-				"baw",
-				"self",
-			},
-			result: fmt.Errorf("expect one two or three args but got 4"),
-		}, { // 1
+		{ // 1
 			args: []interface{}{
 				"bar",
 				2,
+				true,
 				"self",
 			},
 			result: nil,
@@ -780,6 +1121,7 @@ func TestLagExecIndex(t *testing.T) {
 			args: []interface{}{
 				"bar",
 				2,
+				true,
 				"self",
 			},
 			result: nil,
@@ -788,6 +1130,7 @@ func TestLagExecIndex(t *testing.T) {
 			args: []interface{}{
 				"foo",
 				2,
+				true,
 				"self",
 			},
 			result: "bar",
@@ -796,6 +1139,7 @@ func TestLagExecIndex(t *testing.T) {
 			args: []interface{}{
 				"foo",
 				2,
+				true,
 				"self",
 			},
 			result: "bar",
@@ -804,6 +1148,7 @@ func TestLagExecIndex(t *testing.T) {
 			args: []interface{}{
 				"foo",
 				2,
+				true,
 				"self",
 			},
 			result: "foo",
@@ -830,12 +1175,10 @@ func TestLatestExec(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
-			args:   []interface{}{"self"},
-			result: fmt.Errorf("expect one or two args but got 0"),
-		}, { // 1
+		{ // 1
 			args: []interface{}{
 				"foo",
+				true,
 				"self",
 			},
 			result: "foo",
@@ -843,6 +1186,7 @@ func TestLatestExec(t *testing.T) {
 		{ // 2
 			args: []interface{}{
 				nil,
+				true,
 				"self",
 			},
 			result: "foo",
@@ -850,6 +1194,7 @@ func TestLatestExec(t *testing.T) {
 		{ // 3
 			args: []interface{}{
 				"bar",
+				true,
 				"self",
 			},
 			result: "bar",
@@ -857,6 +1202,7 @@ func TestLatestExec(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				nil,
+				true,
 				"self",
 			},
 			result: "bar",
@@ -864,6 +1210,69 @@ func TestLatestExec(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				"foo",
+				true,
+				"self",
+			},
+			result: "foo",
+		},
+	}
+	for i, tt := range tests {
+		result, _ := f.exec(fctx, tt.args)
+		if !reflect.DeepEqual(result, tt.result) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
+		}
+	}
+}
+
+func TestLatestExecWithWhen(t *testing.T) {
+	f, ok := builtins["latest"]
+	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{}
+		result interface{}
+	}{
+		{ // 1
+			args: []interface{}{
+				"foo",
+				true,
+				"self",
+			},
+			result: "foo",
+		},
+		{ // 2
+			args: []interface{}{
+				nil,
+				true,
+				"self",
+			},
+			result: "foo",
+		},
+		{ // 3
+			args: []interface{}{
+				"bar",
+				false,
+				"self",
+			},
+			result: "bar",
+		},
+		{ // 4
+			args: []interface{}{
+				nil,
+				true,
+				"self",
+			},
+			result: "foo",
+		},
+		{ // 4
+			args: []interface{}{
+				"foo",
+				true,
 				"self",
 			},
 			result: "foo",
@@ -890,12 +1299,10 @@ func TestLatestPartition(t *testing.T) {
 		args   []interface{}
 		result interface{}
 	}{
-		{ // 0
-			args:   []interface{}{"self"},
-			result: fmt.Errorf("expect one or two args but got 0"),
-		}, { // 1
+		{ // 1
 			args: []interface{}{
 				"foo",
+				true,
 				"2",
 			},
 			result: "foo",
@@ -904,6 +1311,7 @@ func TestLatestPartition(t *testing.T) {
 			args: []interface{}{
 				nil,
 				"dd",
+				true,
 				"1",
 			},
 			result: "dd",
@@ -911,6 +1319,7 @@ func TestLatestPartition(t *testing.T) {
 		{ // 3
 			args: []interface{}{
 				"bar",
+				true,
 				"1",
 			},
 			result: "bar",
@@ -918,6 +1327,7 @@ func TestLatestPartition(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				nil,
+				true,
 				"2",
 			},
 			result: "foo",
@@ -925,6 +1335,7 @@ func TestLatestPartition(t *testing.T) {
 		{ // 4
 			args: []interface{}{
 				"foo",
+				true,
 				"1",
 			},
 			result: "foo",
diff --git a/internal/topo/operator/analyticfuncs_operator_test.go b/internal/topo/operator/analyticfuncs_operator_test.go
index 536db349..ecc969a4 100644
--- a/internal/topo/operator/analyticfuncs_operator_test.go
+++ b/internal/topo/operator/analyticfuncs_operator_test.go
@@ -31,7 +31,7 @@ func TestAnalyticFuncs(t *testing.T) {
 	var tests = []struct {
 		funcs  []*ast.Call
 		data   []interface{}
-		result [][]map[string]interface{}
+		result []map[string]interface{}
 	}{
 		{ // 0 Lag test
 			funcs: []*ast.Call{
@@ -85,17 +85,16 @@ func TestAnalyticFuncs(t *testing.T) {
 					},
 				},
 			},
-			result: [][]map[string]interface{}{{{
+			result: []map[string]interface{}{{
 				"$$a_lag_0": nil,
 				"$$a_lag_1": nil,
-				"a":         "a1", "b": "b1", "c": "c1",
-			}}, {{
-				"$$a_lag_0": "a1", "$$a_lag_1": "b1", "a": "a1", "b": "b2", "c": "c1",
-			}}, {{
-				"$$a_lag_0": "a1", "$$a_lag_1": "b2", "a": "a1", "c": "c1",
-			}}, {{
-				"$$a_lag_0": "a1", "$$a_lag_1": interface{}(nil), "a": "a1", "b": "b2", "c": "c2",
-			}}},
+			}, {
+				"$$a_lag_0": "a1", "$$a_lag_1": "b1",
+			}, {
+				"$$a_lag_0": "a1", "$$a_lag_1": "b2",
+			}, {
+				"$$a_lag_0": "a1", "$$a_lag_1": interface{}(nil),
+			}},
 		},
 		{ // 1 changed test
 			funcs: []*ast.Call{
@@ -157,16 +156,16 @@ func TestAnalyticFuncs(t *testing.T) {
 					},
 				},
 			},
-			result: [][]map[string]interface{}{
-				{{
-					"$$a_changed_col_0": "a1", "$$a_had_changed_0": false, "$$a_lag_1": nil, "a": "a1", "b": "b1",
-				}}, {{
-					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": "b1", "a": "a1", "c": "c1",
-				}}, {{
-					"$$a_changed_col_0": nil, "$$a_had_changed_0": false, "$$a_lag_1": nil, "a": "a1", "c": "c1",
-				}}, {{
-					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": nil, "a": "a1", "b": "b2", "c": "c2",
-				}},
+			result: []map[string]interface{}{
+				{
+					"$$a_changed_col_0": "a1", "$$a_had_changed_0": false, "$$a_lag_1": nil,
+				}, {
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": "b1",
+				}, {
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": false, "$$a_lag_1": nil,
+				}, {
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": nil,
+				},
 			},
 		},
 	}
@@ -179,15 +178,10 @@ func TestAnalyticFuncs(t *testing.T) {
 
 		pp := &AnalyticFuncsOp{Funcs: tt.funcs}
 		fv, afv := xsql.NewFunctionValuersForOp(ctx)
-		r := make([][]map[string]interface{}, 0, len(tt.data))
+		r := make([]map[string]interface{}, 0, len(tt.data))
 		for _, d := range tt.data {
 			opResult := pp.Apply(ctx, d, fv, afv)
-			result, err := parseResult(opResult, false)
-			if err != nil {
-				t.Errorf("parse result error： %s", err)
-				continue
-			}
-			r = append(r, result)
+			r = append(r, opResult.(*xsql.Tuple).CalCols)
 		}
 
 		if !reflect.DeepEqual(tt.result, r) {
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index bec71987..0c2240a4 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -1254,6 +1254,203 @@ func Test_createLogicalPlan(t *testing.T) {
 				sendMeta:    false,
 			}.Init(),
 		},
+		{ // 15 analytic function over partition plan
+			sql: `SELECT latest(lag(name)) OVER (PARTITION BY temp), id1 FROM src1 WHERE lag(temp) > temp`,
+			p: ProjectPlan{
+				baseLogicalPlan: baseLogicalPlan{
+					children: []LogicalPlan{
+						FilterPlan{
+							baseLogicalPlan: baseLogicalPlan{
+								children: []LogicalPlan{
+									AnalyticFuncsPlan{
+										baseLogicalPlan: baseLogicalPlan{
+											children: []LogicalPlan{
+												DataSourcePlan{
+													name: "src1",
+													streamFields: []interface{}{
+														&ast.StreamField{
+															Name:      "id1",
+															FieldType: &ast.BasicType{Type: ast.BIGINT},
+														},
+														&ast.StreamField{
+															Name:      "name",
+															FieldType: &ast.BasicType{Type: ast.STRINGS},
+														},
+														&ast.StreamField{
+															Name:      "temp",
+															FieldType: &ast.BasicType{Type: ast.BIGINT},
+														},
+													},
+													streamStmt: streams["src1"],
+													metaFields: []string{},
+												}.Init(),
+											},
+										},
+										funcs: []*ast.Call{
+											{
+												Name:        "lag",
+												FuncId:      2,
+												CachedField: "$$a_lag_2",
+												Args: []ast.Expr{&ast.FieldRef{
+													Name:       "temp",
+													StreamName: "src1",
+												}},
+											},
+											{
+												Name: "latest", FuncId: 1, CachedField: "$$a_latest_1", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.Call{Name: "lag", FuncId: 0, Cached: true, CachedField: "$$a_lag_0", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}}}}, Partition: &ast.PartitionExpr{Exprs: []ast.Expr{&ast.FieldRef{Name: "temp", StreamName: "src1"}}},
+											},
+											{
+												Name: "lag", FuncId: 0, CachedField: "$$a_lag_0", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}},
+											},
+										},
+									}.Init(),
+								},
+							},
+							condition: &ast.BinaryExpr{
+								LHS: &ast.Call{
+									Name:   "lag",
+									FuncId: 2,
+									Args: []ast.Expr{&ast.FieldRef{
+										Name:       "temp",
+										StreamName: "src1",
+									}},
+									CachedField: "$$a_lag_2",
+									Cached:      true,
+								},
+								OP: ast.GT,
+								RHS: &ast.FieldRef{
+									Name:       "temp",
+									StreamName: "src1",
+								},
+							},
+						}.Init(),
+					},
+				},
+				fields: []ast.Field{
+					{
+						Expr: &ast.Call{
+							Name:        "latest",
+							FuncId:      1,
+							Args:        []ast.Expr{&ast.Call{Name: "lag", FuncId: 0, Cached: true, CachedField: "$$a_lag_0", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}}}},
+							CachedField: "$$a_latest_1",
+							Cached:      true,
+							Partition: &ast.PartitionExpr{
+								Exprs: []ast.Expr{
+									&ast.FieldRef{Name: "temp", StreamName: "src1"},
+								},
+							},
+						},
+						Name: "latest",
+					}, {
+						Expr: &ast.FieldRef{Name: "id1", StreamName: "src1"},
+						Name: "id1",
+					},
+				},
+				isAggregate: false,
+				sendMeta:    false,
+			}.Init(),
+		},
+		{ // 16 analytic function over partition when plan
+			sql: `SELECT latest(lag(name)) OVER (PARTITION BY temp WHEN temp > 12), id1 FROM src1 WHERE lag(temp) > temp`,
+			p: ProjectPlan{
+				baseLogicalPlan: baseLogicalPlan{
+					children: []LogicalPlan{
+						FilterPlan{
+							baseLogicalPlan: baseLogicalPlan{
+								children: []LogicalPlan{
+									AnalyticFuncsPlan{
+										baseLogicalPlan: baseLogicalPlan{
+											children: []LogicalPlan{
+												DataSourcePlan{
+													name: "src1",
+													streamFields: []interface{}{
+														&ast.StreamField{
+															Name:      "id1",
+															FieldType: &ast.BasicType{Type: ast.BIGINT},
+														},
+														&ast.StreamField{
+															Name:      "name",
+															FieldType: &ast.BasicType{Type: ast.STRINGS},
+														},
+														&ast.StreamField{
+															Name:      "temp",
+															FieldType: &ast.BasicType{Type: ast.BIGINT},
+														},
+													},
+													streamStmt: streams["src1"],
+													metaFields: []string{},
+												}.Init(),
+											},
+										},
+										funcs: []*ast.Call{
+											{
+												Name:        "lag",
+												FuncId:      2,
+												CachedField: "$$a_lag_2",
+												Args: []ast.Expr{&ast.FieldRef{
+													Name:       "temp",
+													StreamName: "src1",
+												}},
+											},
+											{
+												Name: "latest", FuncId: 1, CachedField: "$$a_latest_1", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.Call{Name: "lag", FuncId: 0, Cached: true, CachedField: "$$a_lag_0", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}}}}, Partition: &ast.PartitionExpr{Exprs: []ast.Expr{&ast.FieldRef{Name: "temp", StreamName: "src1"}}}, WhenExpr: &ast.BinaryExpr{LHS: &ast.FieldRef{Name: "temp", StreamName: "src1"}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 12}},
+											},
+											{
+												Name: "lag", FuncId: 0, CachedField: "$$a_lag_0", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}},
+											},
+										},
+									}.Init(),
+								},
+							},
+							condition: &ast.BinaryExpr{
+								LHS: &ast.Call{
+									Name:   "lag",
+									FuncId: 2,
+									Args: []ast.Expr{&ast.FieldRef{
+										Name:       "temp",
+										StreamName: "src1",
+									}},
+									CachedField: "$$a_lag_2",
+									Cached:      true,
+								},
+								OP: ast.GT,
+								RHS: &ast.FieldRef{
+									Name:       "temp",
+									StreamName: "src1",
+								},
+							},
+						}.Init(),
+					},
+				},
+				fields: []ast.Field{
+					{
+						Expr: &ast.Call{
+							Name:        "latest",
+							FuncId:      1,
+							Args:        []ast.Expr{&ast.Call{Name: "lag", FuncId: 0, Cached: true, CachedField: "$$a_lag_0", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}}}},
+							CachedField: "$$a_latest_1",
+							Cached:      true,
+							Partition: &ast.PartitionExpr{
+								Exprs: []ast.Expr{
+									&ast.FieldRef{Name: "temp", StreamName: "src1"},
+								},
+							},
+							WhenExpr: &ast.BinaryExpr{
+								LHS: &ast.FieldRef{Name: "temp", StreamName: "src1"},
+								OP:  ast.GT,
+								RHS: &ast.IntegerLiteral{Val: 12},
+							},
+						},
+						Name: "latest",
+					}, {
+						Expr: &ast.FieldRef{Name: "id1", StreamName: "src1"},
+						Name: "id1",
+					},
+				},
+				isAggregate: false,
+				sendMeta:    false,
+			}.Init(),
+		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index fca55248..2edd0910 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -1878,6 +1878,93 @@ func TestParser_ParseStatement(t *testing.T) {
 				Sources: []ast.Source{&ast.Table{Name: "tbl"}},
 			},
 		},
+		{
+			s: `SELECT lag(name) OVER (PARTITION BY device WHEN abc > 12) as ll FROM tbl`,
+			stmt: &ast.SelectStatement{
+				Fields: []ast.Field{
+					{
+						Expr: &ast.Call{
+							Name:   "lag",
+							FuncId: 0,
+							Args: []ast.Expr{
+								&ast.FieldRef{Name: "name", StreamName: ast.DefaultStream},
+							},
+							Partition: &ast.PartitionExpr{
+								Exprs: []ast.Expr{
+									&ast.FieldRef{Name: "device", StreamName: ast.DefaultStream},
+								},
+							},
+							WhenExpr: &ast.BinaryExpr{
+								LHS: &ast.FieldRef{Name: "abc", StreamName: ast.DefaultStream},
+								OP:  ast.GT,
+								RHS: &ast.IntegerLiteral{Val: 12},
+							},
+						},
+						Name:  "lag",
+						AName: "ll"},
+				},
+				Sources: []ast.Source{&ast.Table{Name: "tbl"}},
+			},
+		},
+		{
+			s: `SELECT lag(name) OVER (PARTITION BY device WHEN had_changed(true, StatusCode)) as ll FROM tbl`,
+			stmt: &ast.SelectStatement{
+				Fields: []ast.Field{
+					{
+						Expr: &ast.Call{
+							Name:   "lag",
+							FuncId: 0,
+							Args: []ast.Expr{
+								&ast.FieldRef{Name: "name", StreamName: ast.DefaultStream},
+							},
+							Partition: &ast.PartitionExpr{
+								Exprs: []ast.Expr{
+									&ast.FieldRef{Name: "device", StreamName: ast.DefaultStream},
+								},
+							},
+							WhenExpr: &ast.Call{
+								Name:   "had_changed",
+								FuncId: 1,
+								Args:   []ast.Expr{&ast.BooleanLiteral{Val: true}, &ast.FieldRef{Name: "StatusCode", StreamName: ast.DefaultStream}},
+							},
+						},
+						Name:  "lag",
+						AName: "ll"},
+				},
+				Sources: []ast.Source{&ast.Table{Name: "tbl"}},
+			},
+		},
+		{
+			s: `SELECT lag(name) OVER (WHEN had_changed(true, StatusCode)) as ll FROM tbl`,
+			stmt: &ast.SelectStatement{
+				Fields: []ast.Field{
+					{
+						Expr: &ast.Call{
+							Name:   "lag",
+							FuncId: 0,
+							Args: []ast.Expr{
+								&ast.FieldRef{Name: "name", StreamName: ast.DefaultStream},
+							},
+							WhenExpr: &ast.Call{
+								Name:   "had_changed",
+								FuncId: 1,
+								Args:   []ast.Expr{&ast.BooleanLiteral{Val: true}, &ast.FieldRef{Name: "StatusCode", StreamName: ast.DefaultStream}},
+							},
+						},
+						Name:  "lag",
+						AName: "ll"},
+				},
+				Sources: []ast.Source{&ast.Table{Name: "tbl"}},
+			},
+		},
+		{
+			s:   `SELECT name OVER (WHEN a > b) FROM tbl`,
+			err: `found "OVER", expected FROM.`,
+		},
+		{
+			s:   `SELECT avg(name) OVER (WHEN a > b) FROM tbl`,
+			err: `Found OVER after non analytic function avg`,
+		},
 		{
 			s: `SELECT *, name, lower(name) as ln FROM tbl`,
 			stmt: &ast.SelectStatement{
diff --git a/internal/xsql/row_test.go b/internal/xsql/row_test.go
index 0520b678..67dcb45a 100644
--- a/internal/xsql/row_test.go
+++ b/internal/xsql/row_test.go
@@ -75,12 +75,12 @@ func TestCollectionRow(t *testing.T) {
 			result:   []interface{}{4, "v1", 4, Message{"a": 4, "b": "v1", "d": 4}},
 		}, {
 			rowC: &JoinTuples{Content: []*JoinTuple{{Tuples: []TupleRow{
-				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}},
+				&Tuple{Emitter: "src1", Message: Message{"a": 1, "b": "v1"}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"b": "v2", "$$lag_a": 1}}},
 				&Tuple{Emitter: "src2", Message: Message{"a": 2, "c": "w2"}},
 			}}}, AffiliateRow: AffiliateRow{CalCols: map[string]interface{}{"a": 4, "d": 3}, AliasMap: map[string]interface{}{"d": 4}}},
 			value:    []string{"a", "b", "d"},
 			wildcard: []string{""},
-			result:   []interface{}{4, "v1", 4, Message{"a": 4, "b": "v1", "c": "w2", "d": 4}},
+			result:   []interface{}{4, "v2", 4, Message{"a": 4, "b": "v2", "c": "w2", "d": 4}},
 		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))