Repository: lf-edge/ekuiper
Pull request #1247 (2022-04-25 01:49:24)

Add had_changed function


Patch:
diff --git a/docs/en_US/sqls/built-in_functions.md b/docs/en_US/sqls/built-in_functions.md
index 60b6e9d7..70e4790b 100644
--- a/docs/en_US/sqls/built-in_functions.md
+++ b/docs/en_US/sqls/built-in_functions.md
@@ -161,17 +161,18 @@ When casting to datetime type, the supported column type and casting rule are:
 **Please refer to [json path functions](./json_expr.md#json-path-functions) for how to compose a json path.**  
 
 ## Other Functions
-| Function     | Example                | Description                                                                                                                                                                                                                                                                                                                                                                                       |
-|--------------|------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| isNull       | isNull(col1)           | Returns true if the argument is the Null value.                                                                                                                                                                                                                                                                                                                                                   |
-| cardinality  | cardinality(col1)      | The number of members in the group. The null value is 0.                                                                                                                                                                                                                                                                                                                                          |
-| newuuid      | newuuid()              | Returns a random 16-byte UUID.                                                                                                                                                                                                                                                                                                                                                                    |
-| tstamp       | tstamp()               | Returns the current timestamp in milliseconds from 00:00:00 Coordinated Universal Time (UTC), Thursday, 1 January 1970                                                                                                                                                                                                                                                                            |
-| mqtt         | mqtt(topic)            | Returns the MQTT meta-data of specified key. The current supported keys<br />- topic: return the topic of message.  If there are multiple stream source, then specify the source name in parameter. Such as `mqtt(src1.topic)`<br />- messageid: return the message id of message. If there are multiple stream source, then specify the source name in parameter. Such as `mqtt(src2.messageid)` |
-| meta         | meta(topic)            | Returns the meta-data of specified key. The key could be:<br/> - a standalone key if there is only one source in the from clause, such as `meta(device)`<br />- A qualified key to specify the stream, such as `meta(src1.device)` <br />- A key with arrow for multi level meta data, such as `meta(src1.reading->device->name)` This assumes reading is a map structure meta data.              |
-| window_start | window_start()         | Return the window start timestamp in int64 format. If there is no time window, it returns 0. The window time is aligned with the timestamp notion of the rule. If the rule is using processing time, then the window start timestamp is the processing timestamp. If the rule is using event time, then the window start timestamp is the event timestamp.                                        |
-| window_end   | window_end()           | Return the window end timestamp in int64 format. If there is no time window, it returns 0. The window time is aligned with the timestamp notion of the rule. If the rule is using processing time, then the window start timestamp is the processing timestamp. If the rule is using event time, then the window start timestamp is the event timestamp.                                          |
-| changed_col  | changed_col(true, col) | Return the column value if it has changed from the last execution.                                                                                                                                                                                                                                                                                                                                |
+| Function     | Example                              | Description                                                                                                                                                                                                                                                                                                                                                                                       |
+|--------------|--------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| isNull       | isNull(col1)                         | Returns true if the argument is the Null value.                                                                                                                                                                                                                                                                                                                                                   |
+| cardinality  | cardinality(col1)                    | The number of members in the group. The null value is 0.                                                                                                                                                                                                                                                                                                                                          |
+| newuuid      | newuuid()                            | Returns a random 16-byte UUID.                                                                                                                                                                                                                                                                                                                                                                    |
+| tstamp       | tstamp()                             | Returns the current timestamp in milliseconds from 00:00:00 Coordinated Universal Time (UTC), Thursday, 1 January 1970                                                                                                                                                                                                                                                                            |
+| mqtt         | mqtt(topic)                          | Returns the MQTT meta-data of specified key. The current supported keys<br />- topic: return the topic of message.  If there are multiple stream source, then specify the source name in parameter. Such as `mqtt(src1.topic)`<br />- messageid: return the message id of message. If there are multiple stream source, then specify the source name in parameter. Such as `mqtt(src2.messageid)` |
+| meta         | meta(topic)                          | Returns the meta-data of specified key. The key could be:<br/> - a standalone key if there is only one source in the from clause, such as `meta(device)`<br />- A qualified key to specify the stream, such as `meta(src1.device)` <br />- A key with arrow for multi level meta data, such as `meta(src1.reading->device->name)` This assumes reading is a map structure meta data.              |
+| window_start | window_start()                       | Return the window start timestamp in int64 format. If there is no time window, it returns 0. The window time is aligned with the timestamp notion of the rule. If the rule is using processing time, then the window start timestamp is the processing timestamp. If the rule is using event time, then the window start timestamp is the event timestamp.                                        |
+| window_end   | window_end()                         | Return the window end timestamp in int64 format. If there is no time window, it returns 0. The window time is aligned with the timestamp notion of the rule. If the rule is using processing time, then the window start timestamp is the processing timestamp. If the rule is using event time, then the window start timestamp is the event timestamp.                                          |
+| changed_col  | changed_col(true, col)               | Return the column value if it has changed from the last execution.                                                                                                                                                                                                                                                                                                                                |
+| had_changed  | had_changed(true, expr1, expr2, ...) | Return if any of the columns had changed since the last run. The expression could be * to easily detect the change status of all columns.                                                                                                                                                                                                                                                         |
 
 ## Multiple Column Functions
 
@@ -233,6 +234,22 @@ The multiple column outputs can only be used in the select clause. Even the sele
 
 For multiple column outputs, the alias can only be set  generally with the prefix. To set alias for each column separately, try to call the changed function for each column respectively and use as to set alias. 
 
+### Had_changed function
+
+This function is a scalar function with one or more arguments.
+
+```HAD_CHANGED (<ignoreNull>, <expr> [,...,<exprN>])```
+
+**Arguments**
+
+**ignoreNull**: whether to ignore null values when detecting changes. If true, the null value won’t trigger a change.
+
+**expr**: An expression to be monitored for the changed status. Allow any expression that can be used in select clause. The expression can be a `*` to detect changes of all columns easily.
+
+**Returns**
+
+Return a bool value to indicate the changed status if any of the arguments had changed since the last run. The multiple arguments' version is a handy way to check HAD_CHANGED(expr1) OR HAD_CHANGED(expr2) ... OR HAD_CHANGED(exprN). To detect other relationship, just use separate HAD_CHANGED functions. For example, to check if all expressions are changed HAD_CHANGED(expr1) AND HAD_CHANGED(expr2) ... AND HAD_CHANGED(exprN).
+
 ### Examples
 
 Create a stream demo and have below inputs
@@ -293,7 +310,7 @@ Rule to get the events when temperature or humidity changed:
 
 ```text
 SQL: SELECT id, temperature, humidity FROM demo
-WHERE ISNULL(CHANGED_COL(true, temperature)) = false OR ISNULL(CHANGED_COL(humidity)) = false
+WHERE HAD_CHANGED(true, temperature, humidity) = true
 _________________________________________________________
 {"ts":1,temperature":23,"humidity":88}
 {"ts":4,temperature":25,"humidity":88}
@@ -305,7 +322,7 @@ Rule to get the events when temperature has changed but humidity has NOT changed
 
 ```text
 SQL: SELECT id, temperature, humidity FROM demo 
-WHERE ISNULL(CHANGED_COL(true, temperature)) = false AND ISNULL(CHANGED_COL(true, humidity))
+WHERE HAD_CHANGED(true, temperature) = true AND HAD_CHANGED(true, humidity) = false
 _________________________________________________________
 {"ts":1,temperature":23,"humidity":88}
 {"ts":4,temperature":25,"humidity":88}
diff --git a/docs/zh_CN/sqls/built-in_functions.md b/docs/zh_CN/sqls/built-in_functions.md
index b1eec02a..b1ba354d 100755
--- a/docs/zh_CN/sqls/built-in_functions.md
+++ b/docs/zh_CN/sqls/built-in_functions.md
@@ -162,17 +162,18 @@ eKuiper 具有许多内置函数，可以对数据执行计算。
 **请参阅 [json 路径函数](./json_expr.md#Json-路径函数) 了解如何编写json路径。**
 
 ## 其它函数
-| 函数           | 示例                     | 说明                                                                                                                                                                                |
-|--------------|------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| isNull       | isNull(col1)           | 如果参数为空值，则返回 true。                                                                                                                                                                 |
-| cardinality  | cardinality(col1)      | 组中成员的数量。空值为0。                                                                                                                                                                     |
-| newuuid      | newuuid()              | 返回一个随机的16字节 UUID。                                                                                                                                                                 |
-| tstamp       | tstamp()               | 返回当前时间戳，以1970年1月1日星期四00:00:00协调世界时（UTC）为单位。                                                                                                                                       |
-| mqtt         | mqtt(topic)            | 返回指定键的 MQTT 元数据。 当前支持的键包括<br />-topic：返回消息的主题。 如果有多个流源，则在参数中指定源名称。 如 `mqtt(src1.topic)`<br />- messageid：返回消息的消息ID。 如果有多个流源，则在参数中指定源名称。 如 `mqtt(src2.messageid)`                  |
-| meta         | meta(topic)            | 返回指定键的元数据。 键可能是：<br/>-如果 from 子句中只有一个来源，则为独立键，例如`meta(device)`<br />-用于指定流的合格键，例如 `meta(src1.device)` <br />-用于多级元数据的带有箭头的键，例如 `meta(src1.reading->device->name)`。这里假定读取是地图结构元数据。 |
-| window_start | window_start()         | 返回窗口的开始时间戳，格式为 int64。若运行时没有时间窗口，则返回默认值0。窗口的时间与规则所用的时间系统相同。若规则采用处理时间，则窗口的时间也为处理时间；若规则采用事件事件，则窗口的时间也为事件时间。                                                                          |
-| window_end   | window_end()           | 返回窗口的结束时间戳，格式为 int64。若运行时没有时间窗口，则返回默认值0。窗口的时间与规则所用的时间系统相同。若规则采用处理时间，则窗口的时间也为处理时间；若规则采用事件事件，则窗口的时间也为事件时间。                                                                          |
-| changed_col  | changed_col(true, col) | 返回列的相比上次执行后的变化值。若未变化则返回 null 。                                                                                                                                                    |
+| 函数           | 示例                                   | 说明                                                                                                                                                                                |
+|--------------|--------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| isNull       | isNull(col1)                         | 如果参数为空值，则返回 true。                                                                                                                                                                 |
+| cardinality  | cardinality(col1)                    | 组中成员的数量。空值为0。                                                                                                                                                                     |
+| newuuid      | newuuid()                            | 返回一个随机的16字节 UUID。                                                                                                                                                                 |
+| tstamp       | tstamp()                             | 返回当前时间戳，以1970年1月1日星期四00:00:00协调世界时（UTC）为单位。                                                                                                                                       |
+| mqtt         | mqtt(topic)                          | 返回指定键的 MQTT 元数据。 当前支持的键包括<br />-topic：返回消息的主题。 如果有多个流源，则在参数中指定源名称。 如 `mqtt(src1.topic)`<br />- messageid：返回消息的消息ID。 如果有多个流源，则在参数中指定源名称。 如 `mqtt(src2.messageid)`                  |
+| meta         | meta(topic)                          | 返回指定键的元数据。 键可能是：<br/>-如果 from 子句中只有一个来源，则为独立键，例如`meta(device)`<br />-用于指定流的合格键，例如 `meta(src1.device)` <br />-用于多级元数据的带有箭头的键，例如 `meta(src1.reading->device->name)`。这里假定读取是地图结构元数据。 |
+| window_start | window_start()                       | 返回窗口的开始时间戳，格式为 int64。若运行时没有时间窗口，则返回默认值0。窗口的时间与规则所用的时间系统相同。若规则采用处理时间，则窗口的时间也为处理时间；若规则采用事件事件，则窗口的时间也为事件时间。                                                                          |
+| window_end   | window_end()                         | 返回窗口的结束时间戳，格式为 int64。若运行时没有时间窗口，则返回默认值0。窗口的时间与规则所用的时间系统相同。若规则采用处理时间，则窗口的时间也为处理时间；若规则采用事件事件，则窗口的时间也为事件时间。                                                                          |
+| changed_col  | changed_col(true, col)               | 返回列的相比上次执行后的变化值。若未变化则返回 null 。                                                                                                                                                    |
+| had_changed  | had_changed(true, expr1, expr2, ...) | 返回是否上次运行后列的值有变化。 其参数可以为 * 以方便地监测所有列。                                                                                                                                              |
 
 ## 多列函数
 
@@ -234,6 +235,22 @@ eKuiper 具有许多内置函数，可以对数据执行计算。
 
 函数返回的列命别名仅能通过 prefix 参数做全局的设置。若需要给每个列设置单独的别名，则需要使用 CHANGED_COL 函数。
 
+### Had_changed 函数
+
+该函数为向量函数，支持不定长度参数。
+
+```HAD_CHANGED (<ignoreNull>, <expr> [,...,<exprN>])```
+
+**参数**
+
+**ignoreNull**: 判断变化时是否忽略 null 值。若为 true，则收到 null 值或未收到值不会触发变化。
+
+**expr**: 用来监控变化状态和输出变化值的表达式。可以为任何可在 SELECT 子句中使用的表达式。若表达式为 `*` 则监测所有列的变化。
+
+**返回值**
+
+返回一个 bool 值，表示上次运行后的变化状态。多参数版本与用或连接使用单个参数的版本相同，即 HAD_CHANGED(expr1) OR HAD_CHANGED(expr2) ... OR HAD_CHANGED(exprN) 。若需要监测别的关系，可单独使用此函数。例如，监测是否所有值都有变化，可使用 HAD_CHANGED(expr1) AND HAD_CHANGED(expr2) ... AND HAD_CHANGED(exprN) 。
+
 ### 范例
 
 创建流 demo，并给与如下输入。
@@ -294,7 +311,7 @@ _________________________________________________________________
 
 ```text
 SQL: SELECT id, temperature, humidity FROM demo
-WHERE ISNULL(CHANGED_COL(true, temperature)) = false OR ISNULL(true, CHANGED_COL(humidity)) = false
+WHERE HAD_CHANGED(true, temperature, humidity) = true
 _________________________________________________________
 {"ts":1,temperature":23,"humidity":88}
 {"ts":4,temperature":25,"humidity":88}
@@ -306,7 +323,7 @@ _________________________________________________________
 
 ```text
 SQL: SELECT id, temperature, humidity FROM demo 
-WHERE ISNULL(CHANGED_COL(true, temperature)) = false AND ISNULL(true, CHANGED_COL(humidity)) = false
+WHERE HAD_CHANGED(true, temperature) = true AND HAD_CHANGED(true, humidity) = false
 _________________________________________________________
 {"ts":1,temperature":23,"humidity":88}
 {"ts":4,temperature":25,"humidity":88}
diff --git a/internal/binder/function/funcs_misc.go b/internal/binder/function/funcs_misc.go
index 49ccdc0d..c4abd685 100644
--- a/internal/binder/function/funcs_misc.go
+++ b/internal/binder/function/funcs_misc.go
@@ -512,6 +512,80 @@ func registerMiscFunc() {
 			return nil
 		},
 	}
+	builtins["changed_col"] = builtinFunc{
+		fType: FuncTypeScalar,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			ignoreNull, ok := args[0].(bool)
+			if !ok {
+				return fmt.Errorf("first arg is not a bool but got %v", args[0]), false
+			}
+			if ignoreNull && args[1] == nil {
+				return nil, true
+			}
+			lv, err := ctx.GetState("self")
+			if err != nil {
+				return err, false
+			}
+			if !reflect.DeepEqual(args[1], lv) {
+				err := ctx.PutState("self", args[1])
+				if err != nil {
+					return err, false
+				}
+				return args[1], true
+			}
+			return nil, true
+		},
+		val: func(_ api.FunctionContext, args []ast.Expr) error {
+			if err := ValidateLen(2, len(args)); err != nil {
+				return err
+			}
+			if ast.IsNumericArg(args[0]) || ast.IsTimeArg(args[0]) || ast.IsStringArg(args[0]) {
+				return ProduceErrInfo(0, "boolean")
+			}
+			return nil
+		},
+	}
+	builtins["had_changed"] = builtinFunc{
+		fType: FuncTypeScalar,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			if len(args) <= 1 {
+				return fmt.Errorf("expect more than one arg but got %d", len(args)), false
+			}
+			ignoreNull, ok := args[0].(bool)
+			if !ok {
+				return fmt.Errorf("first arg is not a bool but got %v", args[0]), false
+			}
+			result := false
+			for i := 1; i < len(args); i++ {
+				v := args[i]
+				k := strconv.Itoa(i)
+				if ignoreNull && v == nil {
+					continue
+				}
+				lv, err := ctx.GetState(k)
+				if err != nil {
+					return fmt.Errorf("error getting state for %s: %v", k, err), false
+				}
+				if !reflect.DeepEqual(v, lv) {
+					result = true
+					err := ctx.PutState(k, v)
+					if err != nil {
+						return fmt.Errorf("error setting state for %s: %v", k, err), false
+					}
+				}
+			}
+			return result, true
+		},
+		val: func(_ api.FunctionContext, args []ast.Expr) error {
+			if len(args) <= 1 {
+				return fmt.Errorf("expect more than one arg but got %d", len(args))
+			}
+			if ast.IsNumericArg(args[0]) || ast.IsTimeArg(args[0]) || ast.IsStringArg(args[0]) {
+				return ProduceErrInfo(0, "bool")
+			}
+			return nil
+		},
+	}
 	builtins["object_construct"] = builtinFunc{
 		fType: FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {


Test patch:
diff --git a/internal/binder/function/funcs_misc_test.go b/internal/binder/function/funcs_misc_test.go
index 5498ec3a..02ea4212 100644
--- a/internal/binder/function/funcs_misc_test.go
+++ b/internal/binder/function/funcs_misc_test.go
@@ -179,3 +179,229 @@ func TestToMap(t *testing.T) {
 		}
 	}
 }
+
+func TestHadChangedValidation(t *testing.T) {
+	f, ok := builtins["had_changed"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	var tests = []struct {
+		args []ast.Expr
+		err  error
+	}{
+		{
+			args: []ast.Expr{
+				&ast.StringLiteral{Val: "foo"},
+			},
+			err: fmt.Errorf("expect more than one arg but got 1"),
+		}, {
+			args: []ast.Expr{
+				&ast.StringLiteral{Val: "foo"},
+				&ast.StringLiteral{Val: "bar"},
+				&ast.StringLiteral{Val: "baz"},
+			},
+			err: fmt.Errorf("Expect bool type for parameter 1"),
+		}, {
+			args: []ast.Expr{
+				&ast.IntegerLiteral{Val: 20},
+				&ast.BooleanLiteral{Val: true},
+				&ast.StringLiteral{Val: "baz"},
+			},
+			err: fmt.Errorf("Expect bool type for parameter 1"),
+		}, {
+			args: []ast.Expr{
+				&ast.FieldRef{
+					StreamName: "demo",
+					Name:       "a",
+					AliasRef:   nil,
+				},
+				&ast.BooleanLiteral{Val: true},
+				&ast.StringLiteral{Val: "baz"},
+			},
+			err: nil,
+		},
+	}
+	for i, tt := range tests {
+		err := f.val(nil, tt.args)
+		if !reflect.DeepEqual(err, tt.err) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, err, tt.err)
+		}
+	}
+}
+
+func TestHadChangedExec(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",
+			},
+			result: fmt.Errorf("first arg is not a bool but got foo"),
+		}, { // 1
+			args: []interface{}{
+				"foo",
+				"bar",
+			},
+			result: fmt.Errorf("first arg is not a bool but got foo"),
+		}, { // 2
+			args: []interface{}{
+				true,
+				"bar",
+				20,
+			},
+			result: true,
+		}, { // 3
+			args: []interface{}{
+				true,
+				"baz",
+				44,
+			},
+			result: true,
+		}, { // 4
+			args: []interface{}{
+				true,
+				"baz",
+				44,
+			},
+			result: false,
+		}, { // 5
+			args: []interface{}{
+				true,
+				"foo",
+				44,
+			},
+			result: true,
+		}, { // 6
+			args: []interface{}{
+				true,
+				"foo",
+				nil,
+			},
+			result: false,
+		}, { // 7
+			args: []interface{}{
+				true,
+				"foo",
+				44,
+			},
+			result: false,
+		}, { // 8
+			args: []interface{}{
+				true,
+				"baz",
+				44,
+			},
+			result: true,
+		},
+	}
+	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 TestHadChangedExecAllowNull(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",
+			},
+			result: fmt.Errorf("first arg is not a bool but got foo"),
+		}, { // 1
+			args: []interface{}{
+				"foo",
+				"bar",
+			},
+			result: fmt.Errorf("first arg is not a bool but got foo"),
+		}, { // 2
+			args: []interface{}{
+				false,
+				"bar",
+				20,
+			},
+			result: true,
+		}, { // 3
+			args: []interface{}{
+				false,
+				"baz",
+				nil,
+			},
+			result: true,
+		}, { // 4
+			args: []interface{}{
+				false,
+				"baz",
+				44,
+			},
+			result: true,
+		}, { // 5
+			args: []interface{}{
+				false,
+				nil,
+				44,
+			},
+			result: true,
+		}, { // 6
+			args: []interface{}{
+				false,
+				"baz",
+				44,
+			},
+			result: true,
+		}, { // 7
+			args: []interface{}{
+				false,
+				"baz",
+				44,
+			},
+			result: false,
+		}, { // 8
+			args: []interface{}{
+				false,
+				nil,
+				nil,
+			},
+			result: true,
+		}, { // 9
+			args: []interface{}{
+				false,
+				"baz",
+				44,
+			},
+			result: true,
+		},
+	}
+	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)
+		}
+	}
+}