Repository: lf-edge/ekuiper
Pull request #1513 (2022-11-18 09:14:22)

fea(plug): add general tensorflow function
Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>


Patch:
diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index d8689351..737a85cc 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -193,6 +193,7 @@ jobs:
           - functions/image
           - functions/geohash
           - functions/labelImage
+          - functions/tfLite
         arch:
           - linux/amd64
           - linux/arm64
diff --git a/Makefile b/Makefile
index af08ffe5..e9e421f8 100644
--- a/Makefile
+++ b/Makefile
@@ -97,7 +97,8 @@ PLUGINS := sinks/file \
 	functions/image \
 	functions/geohash \
 	functions/echo \
-	functions/labelImage
+	functions/labelImage \
+	functions/tfLite
 
 .PHONY: plugins $(PLUGINS)
 plugins: $(PLUGINS)
diff --git a/build-plugins.sh b/build-plugins.sh
index 1a5d9803..162b8f6d 100755
--- a/build-plugins.sh
+++ b/build-plugins.sh
@@ -44,15 +44,29 @@ build(){
             go build -trimpath -modfile extensions.mod --buildmode=plugin -tags plugins -o extensions/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME@$VERSION.so extensions/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME.go
             ;;
         labelImage )
-            git clone -b v2.2.0-rc3 --depth 1 https://github.com/tensorflow/tensorflow.git /tmp/tensorflow;
+            if [ ! -d "/tmp/tensorflow" ];then
+                git clone -b v2.2.0-rc3 --depth 1 https://github.com/tensorflow/tensorflow.git /tmp/tensorflow;
+            fi;
             if [ "$(uname -m)" = "x86_64" ]; then
-                mv $(pwd)/extensions/functions/labelImage/dependencies/amd64/*.so $(pwd)/extensions/functions/labelImage/lib
+                cp $(pwd)/extensions/functions/dependencies/tensorflow/amd64/*.so $(pwd)/extensions/functions/labelImage/lib
             fi;
             if [ "$(uname -m)" = "aarch64" ]; then
-                mv $(pwd)/extensions/functions/labelImage/dependencies/arm64/*.so $(pwd)/extensions/functions/labelImage/lib
+                cp $(pwd)/extensions/functions/dependencies/tensorflow/arm64/*.so $(pwd)/extensions/functions/labelImage/lib
             fi;
             CGO_CFLAGS=-I/tmp/tensorflow CGO_LDFLAGS=-L$(pwd)/extensions/functions/labelImage/lib go build -trimpath -modfile extensions.mod --buildmode=plugin -o extensions/functions/labelImage/labelImage.so extensions/functions/labelImage/*.go
             ;;
+        tfLite )
+            if [ ! -d "/tmp/tensorflow" ];then
+                git clone -b v2.2.0-rc3 --depth 1 https://github.com/tensorflow/tensorflow.git /tmp/tensorflow;
+            fi;
+            if [ "$(uname -m)" = "x86_64" ]; then
+                cp $(pwd)/extensions/functions/dependencies/tensorflow/amd64/*.so $(pwd)/extensions/functions/tfLite/lib
+            fi;
+            if [ "$(uname -m)" = "aarch64" ]; then
+                cp $(pwd)/extensions/functions/dependencies/tensorflow/arm64/*.so $(pwd)/extensions/functions/tfLite/lib
+            fi;
+                CGO_CFLAGS=-I/tmp/tensorflow CGO_LDFLAGS=-L$(pwd)/extensions/functions/tfLite/lib go build -trimpath -modfile extensions.mod --buildmode=plugin -o extensions/functions/tfLite/tfLite.so extensions/functions/tfLite/*.go
+            ;;
         * )
             go build -trimpath -modfile extensions.mod --buildmode=plugin -o extensions/$PLUGIN_TYPE/$PLUGIN_NAME/$PLUGIN_NAME@$VERSION.so extensions/$PLUGIN_TYPE/$PLUGIN_NAME/*.go
           ;;
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/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/functions/labelImage/dependencies/amd64/libtensorflowlite.so b/extensions/functions/dependencies/tensorflow/amd64/libtensorflowlite.so
similarity index 100%
rename from extensions/functions/labelImage/dependencies/amd64/libtensorflowlite.so
rename to extensions/functions/dependencies/tensorflow/amd64/libtensorflowlite.so
diff --git a/extensions/functions/labelImage/dependencies/amd64/libtensorflowlite_c.so b/extensions/functions/dependencies/tensorflow/amd64/libtensorflowlite_c.so
similarity index 100%
rename from extensions/functions/labelImage/dependencies/amd64/libtensorflowlite_c.so
rename to extensions/functions/dependencies/tensorflow/amd64/libtensorflowlite_c.so
diff --git a/extensions/functions/labelImage/dependencies/arm64/libtensorflowlite.so b/extensions/functions/dependencies/tensorflow/arm64/libtensorflowlite.so
similarity index 100%
rename from extensions/functions/labelImage/dependencies/arm64/libtensorflowlite.so
rename to extensions/functions/dependencies/tensorflow/arm64/libtensorflowlite.so
diff --git a/extensions/functions/labelImage/dependencies/arm64/libtensorflowlite_c.so b/extensions/functions/dependencies/tensorflow/arm64/libtensorflowlite_c.so
similarity index 100%
rename from extensions/functions/labelImage/dependencies/arm64/libtensorflowlite_c.so
rename to extensions/functions/dependencies/tensorflow/arm64/libtensorflowlite_c.so
diff --git a/extensions/functions/geohash/geohash.json b/extensions/functions/geohash/geohash.json
index af2462ae..5b9fca2f 100644
--- a/extensions/functions/geohash/geohash.json
+++ b/extensions/functions/geohash/geohash.json
@@ -17,7 +17,7 @@
 		}
 	},
 	"libs": ["github.com/mmcloughlin/geohash@master"],
-  "name":"geohash",
+	"name":"geohash",
 	"functions": [{
 		"name": "geohashEncode",
 		"example": "geohashEncode(la,lo)",
diff --git a/extensions/functions/labelImage/labelImage.go b/extensions/functions/labelImage/labelImage.go
index 7b37c2a4..23412925 100644
--- a/extensions/functions/labelImage/labelImage.go
+++ b/extensions/functions/labelImage/labelImage.go
@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build tflite
-// +build tflite
-
 package main
 
 import (
diff --git a/extensions/functions/tfLite/install.sh b/extensions/functions/tfLite/install.sh
new file mode 100644
index 00000000..f78e7081
--- /dev/null
+++ b/extensions/functions/tfLite/install.sh
@@ -0,0 +1,40 @@
+#!/bin/sh
+#
+# Copyright 2021 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.
+#
+
+dir=/usr/local/tflite
+cur=$(dirname "$0")
+echo "Base path $cur" 
+if [ -d "$dir" ]; then
+    echo "SDK path $dir exists." 
+else
+    echo "Creating SDK path $dir"
+    mkdir -p $dir
+    echo "Created SDK path $dir"
+    echo "Moving libs"
+    cp -R $cur/lib $dir
+    echo "Moved libs"
+fi
+
+if [ -f "/etc/ld.so.conf.d/tflite.conf" ]; then
+    echo "/etc/ld.so.conf.d/tflite.conf exists"
+else
+    echo "Copy conf file"
+    cp $cur/tflite.conf /etc/ld.so.conf.d/
+    echo "Copied conf file"
+fi
+ldconfig
+echo "Done"
\ No newline at end of file
diff --git a/extensions/functions/tfLite/interpreters.go b/extensions/functions/tfLite/interpreters.go
new file mode 100644
index 00000000..b21456f6
--- /dev/null
+++ b/extensions/functions/tfLite/interpreters.go
@@ -0,0 +1,70 @@
+// 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 main
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/mattn/go-tflite"
+	"path/filepath"
+	"sync"
+)
+
+var ipManager *interpreterManager
+
+func init() {
+	path, err := conf.GetDataLoc()
+	if err != nil {
+		panic(err)
+	}
+	ipManager = &interpreterManager{
+		registry: make(map[string]*tflite.Interpreter),
+		path:     filepath.Join(path, "etc"),
+	}
+}
+
+type interpreterManager struct {
+	sync.Mutex
+	registry map[string]*tflite.Interpreter
+	path     string
+}
+
+func (m *interpreterManager) GetOrCreate(name string) (*tflite.Interpreter, error) {
+	m.Lock()
+	defer m.Unlock()
+	ip, ok := m.registry[name]
+	if !ok {
+		mf := filepath.Join(m.path, name+".tflite")
+		model := tflite.NewModelFromFile(mf)
+		if model == nil {
+			return nil, fmt.Errorf("fail to load model: %s", mf)
+		}
+		defer model.Delete()
+		options := tflite.NewInterpreterOptions()
+		options.SetNumThread(4)
+		options.SetErrorReporter(func(msg string, user_data interface{}) {
+			fmt.Println(msg)
+		}, nil)
+		defer options.Delete()
+		ip = tflite.NewInterpreter(model, options)
+		status := ip.AllocateTensors()
+		if status != tflite.OK {
+			ip.Delete()
+			return nil, fmt.Errorf("allocate failed: %v", status)
+		}
+		m.registry[name] = ip
+	}
+	return ip, nil
+}
diff --git a/extensions/functions/tfLite/lib/Readme.md b/extensions/functions/tfLite/lib/Readme.md
new file mode 100644
index 00000000..e19d7b73
--- /dev/null
+++ b/extensions/functions/tfLite/lib/Readme.md
@@ -0,0 +1,25 @@
+# Tensorflow Lite C API library
+
+This is the prebuilt tensorflow lite c library for debian 10. It can be used directly in eKuiper docker image of tags x.x.x or x.x.x-slim.
+
+To use in other environment, you need to build the library from source.
+
+## Build from source
+
+Here are the steps to build from source in debian. 
+
+1. Install [Python](https://www.tensorflow.org/install/pip#1.-install-the-python-development-environment-on-your-system)
+
+2. Install required python lib: `pip3 install -r requirements.txt`. The requirements are from `tensorflow/tensorflow/tools/pip_package/setup.py` of the corresponding tensorflow version.
+
+3. Install [Bazel](https://docs.bazel.build/versions/4.0.0/install-ubuntu.html)
+
+4. Clone [tensorflow](https://github.com/tensorflow/tensorflow)，switch to `git checkout v2.2.0-rc3 -b mybranch`
+
+5. Build the so files, the outputs are in ./bazel-bin
+
+   ```bash
+   $ cd $tensorflowSrc
+   $ bazel build --config monolithic -c opt //tensorflow/lite:libtensorflowlite.so
+   $ bazel build --config monolithic -c opt //tensorflow/lite/c:libtensorflowlite_c.so
+   ```
diff --git a/extensions/functions/tfLite/tfLite.go b/extensions/functions/tfLite/tfLite.go
new file mode 100644
index 00000000..e93a0967
--- /dev/null
+++ b/extensions/functions/tfLite/tfLite.go
@@ -0,0 +1,262 @@
+// 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 main
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/mattn/go-tflite"
+	"strconv"
+)
+
+type Tffunc struct {
+}
+
+// Validate the arguments.
+// args[0]: string, model name which maps to a path
+// args[1 to n]: tensors
+func (f *Tffunc) Validate(args []interface{}) error {
+	if len(args) < 2 {
+		return fmt.Errorf("tensorflow function must have at least 2 parameters but got %d", len(args))
+	}
+	return nil
+}
+
+func (f *Tffunc) IsAggregate() bool {
+	return false
+}
+
+func (f *Tffunc) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
+	model, ok := args[0].(string)
+	if !ok {
+		return fmt.Errorf("tensorflow function first parameter must be a string, but got %[1]T(%[1]v)", args[0]), false
+	}
+	interpreter, err := ipManager.GetOrCreate(model)
+	if err != nil {
+		return err, false
+	}
+	inputCount := interpreter.GetInputTensorCount()
+	if len(args)-1 != inputCount {
+		return fmt.Errorf("tensorflow function requires %d tensors but got %d", inputCount, len(args)-1), false
+	}
+
+	ctx.GetLogger().Warnf("tensorflow function %s with %d tensors", model, inputCount)
+	// Set input tensors
+	for i := 1; i < len(args); i++ {
+		input := interpreter.GetInputTensor(i - 1)
+		dims := "("
+		for j := 1; j < input.NumDims(); j++ {
+			dims += strconv.Itoa(input.Dim(j)) + ","
+		}
+		dims += ")"
+		ctx.GetLogger().Warnf("tensorflow function %s input %d shape %s", model, i, dims)
+		var arg []interface{}
+		switch v := args[i].(type) {
+		case []byte:
+			if int(input.ByteSize()) != len(v) {
+				return fmt.Errorf("tensorflow function input tensor %d has %d bytes but got %d", i-1, input.ByteSize(), len(v)), false
+			}
+			input.CopyFromBuffer(v)
+			continue
+		case []interface{}:
+			arg = v
+		default:
+			return fmt.Errorf("tensorflow function parameter %d must be a bytea or array of bytea, but got %[1]T(%[1]v)", i), false
+		}
+		t := input.Type()
+		ctx.GetLogger().Warnf("tensor %d input dims %d type %s", i-1, input.NumDims(), t)
+		for j := 0; j < input.NumDims(); j++ {
+			ctx.GetLogger().Warnf("tensor %d input dim %d %d", i-1, j, input.Dim(j))
+		}
+		switch input.NumDims() {
+		case 0, 1:
+			return fmt.Errorf("tensorflow function input tensor %d must have at least 2 dimensions but got 1", i-1), false
+		case 2:
+			if input.Dim(1) != len(arg) {
+				return fmt.Errorf("tensorflow function input tensor %d must have %d elements but got %d", i-1, input.Dim(1), len(arg)), false
+			}
+			switch t {
+			case tflite.Float32:
+				v, err := cast.ToFloat32Slice(arg, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect float32 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetFloat32s(v)
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int64:
+				v, err := cast.ToInt64Slice(arg, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int64 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt64s(v)
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int32:
+				v, err := cast.ToTypedSlice(args, func(input interface{}, sn cast.Strictness) (interface{}, error) {
+					return cast.ToInt32(input, sn)
+				}, "int32", cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int32 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt32s(v.([]int32))
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int16:
+				v, err := cast.ToTypedSlice(args, func(input interface{}, sn cast.Strictness) (interface{}, error) {
+					return cast.ToInt16(input, sn)
+				}, "int16", cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int16 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt16s(v.([]int16))
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int8:
+				v, err := cast.ToTypedSlice(args, func(input interface{}, sn cast.Strictness) (interface{}, error) {
+					return cast.ToInt8(input, sn)
+				}, "int8", cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int8 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt8s(v.([]int8))
+				if err != nil {
+					return nil, false
+				}
+			case tflite.UInt8:
+				v, err := cast.ToBytes(args, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect uint8 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetUint8s(v)
+				if err != nil {
+					return nil, false
+				}
+			default:
+				return fmt.Errorf("invalid %d parameter, unsupported type %v in the model", i, t), false
+			}
+		default:
+			// TODO support multiple dimensions. Here assume user passes a 1D array.
+			//if input.Dim(1)*input.Dim(2) != len(arg) {
+			//	return fmt.Errorf("tensorflow function input tensor %d must have %d elements but got %d", i-1, input.Dim(1), len(arg)), false
+			//}
+			switch t {
+			case tflite.Float32:
+				v, err := cast.ToFloat32Slice(args[i], cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect float32 but got %[2]T(%[2]v)", i, args[i]), false
+				}
+				err = input.SetFloat32s(v)
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int64:
+				v, err := cast.ToInt64Slice(arg, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int64 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt64s(v)
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int32:
+				v, err := cast.ToTypedSlice(args, func(input interface{}, sn cast.Strictness) (interface{}, error) {
+					return cast.ToInt32(input, sn)
+				}, "int32", cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int32 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt32s(v.([]int32))
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int16:
+				v, err := cast.ToTypedSlice(args, func(input interface{}, sn cast.Strictness) (interface{}, error) {
+					return cast.ToInt16(input, sn)
+				}, "int16", cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int16 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt16s(v.([]int16))
+				if err != nil {
+					return nil, false
+				}
+			case tflite.Int8:
+				v, err := cast.ToTypedSlice(args, func(input interface{}, sn cast.Strictness) (interface{}, error) {
+					return cast.ToInt8(input, sn)
+				}, "int8", cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect int8 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetInt8s(v.([]int8))
+				if err != nil {
+					return nil, false
+				}
+			case tflite.UInt8:
+				v, err := cast.ToBytes(args, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return fmt.Errorf("invalid %d parameter, expect uint8 but got %[2]T(%[2]v) with err %v", i, args[i], err), false
+				}
+				err = input.SetUint8s(v)
+				if err != nil {
+					return nil, false
+				}
+			default:
+				return fmt.Errorf("invalid %d parameter, unsupported type %v in the model", i, t), false
+			}
+		}
+	}
+	status := interpreter.Invoke()
+	if status != tflite.OK {
+		return fmt.Errorf("invoke failed"), false
+	}
+	outputCount := interpreter.GetOutputTensorCount()
+	results := make([]interface{}, outputCount)
+	for i := 0; i < outputCount; i++ {
+		output := interpreter.GetOutputTensor(i)
+		//outputSize := output.Dim(output.NumDims() - 1)
+		//b := make([]byte, outputSize)
+		//status = output.CopyToBuffer(&b[0])
+		//if status != tflite.OK {
+		//	return fmt.Errorf("output failed"), false
+		//}
+		//results[i] = b
+		t := output.Type()
+		switch t {
+		case tflite.Float32:
+			results[i] = output.Float32s()
+		case tflite.Int64:
+			results[i] = output.Int64s()
+		case tflite.Int32:
+			results[i] = output.Int32s()
+		case tflite.Int16:
+			results[i] = output.Int16s()
+		case tflite.Int8:
+			results[i] = output.Int8s()
+		case tflite.UInt8:
+			results[i] = output.UInt8s()
+		default:
+			return fmt.Errorf("invalid %d parameter, unsupported type %v in the model", i, t), false
+		}
+	}
+	return results, true
+}
+
+var TfLite Tffunc
diff --git a/extensions/functions/tfLite/tfLite.json b/extensions/functions/tfLite/tfLite.json
new file mode 100644
index 00000000..935acc27
--- /dev/null
+++ b/extensions/functions/tfLite/tfLite.json
@@ -0,0 +1,91 @@
+{
+	"about": {
+		"trial": false,
+		"author": {
+			"name": "EMQ",
+			"email": "contact@emqx.io",
+			"company": "EMQ Technologies Co., Ltd",
+			"website": "https://www.emqx.io"
+		},
+		"helpUrl": {
+			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/sqls/custom_functions.md",
+			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/sqls/custom_functions.md"
+		},
+		"description": {
+			"en_US": "",
+			"zh_CN": ""
+		}
+	},
+	"libs": [],
+	"name": "tfLite",
+	"functions": [{
+		"name": "tflite",
+		"example": "tflite(model,para1, para2,...)",
+		"hint": {
+			"en_US": "Creates a scaled image with new dimensions (width, height) .If either width or height is set to 0, it will be set to an aspect ratio preserving value.",
+			"zh_CN": "创建具有新尺寸（宽度，高度）的缩放图像。如果width或height设置为0，则将其设置为长宽比保留值。"
+		},
+		"args": [
+			{
+				"name": "model",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入模型"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入模型"
+				}
+			},{
+				"name": "para1",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据1"
+				},
+				"label": {
+					"en_US": "Input data",
+					"zh_CN": "输入数据1"
+				}
+			},{
+				"name": "para2",
+				"hidden": true,
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Input height",
+					"zh_CN": "输入数据2"
+				},
+				"label": {
+					"en_US": "Input height",
+					"zh_CN": "输入数据2"
+				}
+			}
+		],
+		"outputs": [
+			{
+				"label": {
+					"en_US": "Output",
+					"zh_CN": "输出"
+				},
+				"value": "tflite"
+			}
+		],
+		"node": {
+			"category": "abc",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "tflite",
+				"zh_CN": "tflite"
+			}
+		}
+	}]
+}
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/server/plugin_init.go b/internal/server/plugin_init.go
index 57132aa2..ef18f6c6 100644
--- a/internal/server/plugin_init.go
+++ b/internal/server/plugin_init.go
@@ -234,9 +234,9 @@ func prebuildPluginsHandler(w http.ResponseWriter, _ *http.Request, t plugin.Plu
 	}
 }
 
-var NativeSourcePlugin = []string{"random", "zmq", "sql"}
+var NativeSourcePlugin = []string{"random", "zmq", "sql", "video"}
 var NativeSinkPlugin = []string{"file", "image", "influx", "tdengine", "zmq", "sql"}
-var NativeFunctionPlugin = []string{"accumulateWordCount", "countPlusOne", "echo", "geohash", "image", "labelImage"}
+var NativeFunctionPlugin = []string{"accumulateWordCount", "countPlusOne", "echo", "geohash", "image", "labelImage", "tfLite"}
 
 func fetchPluginList(t plugin.PluginType, hosts, os, arch string) (err error, result map[string]string) {
 	ptype := "sources"
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/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)
 
diff --git a/pkg/cast/cast.go b/pkg/cast/cast.go
index 6bcbbcc5..b0246324 100644
--- a/pkg/cast/cast.go
+++ b/pkg/cast/cast.go
@@ -897,6 +897,22 @@ func ToFloat64Slice(input interface{}, sn Strictness) ([]float64, error) {
 	return result, nil
 }
 
+func ToFloat32Slice(input interface{}, sn Strictness) ([]float32, error) {
+	s := reflect.ValueOf(input)
+	if s.Kind() != reflect.Slice {
+		return nil, fmt.Errorf("cannot convert %[1]T(%[1]v) to float slice)", input)
+	}
+	var result []float32
+	for i := 0; i < s.Len(); i++ {
+		ele, err := ToFloat32(s.Index(i).Interface(), sn)
+		if err != nil {
+			return nil, fmt.Errorf("cannot convert %[1]T(%[1]v) to float slice for the %d element: %v", input, i, err)
+		}
+		result = append(result, ele)
+	}
+	return result, nil
+}
+
 func ToBoolSlice(input interface{}, sn Strictness) ([]bool, error) {
 	s := reflect.ValueOf(input)
 	if s.Kind() != reflect.Slice {


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