Repository: lf-edge/ekuiper
Pull request #1494 (2022-11-17 06:10:42)

fix(meta): clean up for resource manager


Patch:
diff --git a/docs/en_US/rules/sinks/builtin/memory.md b/docs/en_US/rules/sinks/builtin/memory.md
index bc128068..834f30f5 100644
--- a/docs/en_US/rules/sinks/builtin/memory.md
+++ b/docs/en_US/rules/sinks/builtin/memory.md
@@ -4,9 +4,10 @@
 
 The action is used to flush the result into an in-memory topic so that it can be consumed by the [memory source](../../sources/builtin/memory.md). The topic is like pubsub topic such as mqtt, so that there could be multiple memory sinks which publish to the same topic and multiple memory sources which subscribe to the same topic. The typical usage for memory action is to form [rule pipelines](../../rule_pipeline.md).
 
-| Property name | Optional | Description                                    |
-|---------------|----------|------------------------------------------------|
-| topic         | false    | The in-memory topic, such as `analysis/result` |
+| Property name | Optional | Description                                                                                                        |
+|---------------|----------|--------------------------------------------------------------------------------------------------------------------|
+| topic         | false    | The in-memory topic, such as `analysis/result`                                                                     |
+| rowkindField  | true     | Specify which field represents the action like insert or update. If not specified, all rows are default to insert. |
 
 Below is a sample memory action configuration:
 
diff --git a/docs/en_US/rules/sinks/builtin/redis.md b/docs/en_US/rules/sinks/builtin/redis.md
index 1a3d653c..fb776b11 100644
--- a/docs/en_US/rules/sinks/builtin/redis.md
+++ b/docs/en_US/rules/sinks/builtin/redis.md
@@ -26,9 +26,10 @@ Restart the eKuiper server to activate the plugin.
 | field         | true     | This field must exist and be of type string. Otherwise, use the field character as the key. Note: Do not use a data template to configure this value                                                 |
 | dataType      | false    | The default Redis data type is string. Note that the original key must be deleted after the Redis data type is changed. Otherwise, the modification is invalid. now only support "list" and "string" |
 | expiration    | false    | Timeout duration of Redis data. This parameter is valid only for string data in seconds. The default value is -1                                                                                     |
+| rowkindField  | true     | Specify which field represents the action like insert or update. If not specified, all rows are default to insert.                                                                                   |
 ## Sample usage
 
-Below is a sample for selecting temperature great than 50 degree, and some profiles only for your reference.
+Below is a sample for selecting temperature greater than 50 degree, and some profiles only for your reference.
 
 ### /tmp/redisRule.txt
 ```json
diff --git a/docs/en_US/rules/sinks/plugin/sql.md b/docs/en_US/rules/sinks/plugin/sql.md
index 7d73c6b3..0c95f4b1 100644
--- a/docs/en_US/rules/sinks/plugin/sql.md
+++ b/docs/en_US/rules/sinks/plugin/sql.md
@@ -33,6 +33,7 @@ for example, if he only wants mysql, then he can build with build tag `mysql`.
 | table          | false    | The table name of the result                                                                                                                                  |
 | fields         | true     | The fields to be inserted to. The result map and the database should both have these fields. If not specified, all fields in the result map will be inserted. |
 | tableDataField | true     | Write the nested values of the tableDataField into database.                                                                                                  |
+| rowkindField   | true     | Specify which field represents the action like insert or update. If not specified, all rows are default to insert.                                            |
 
 ## Sample usage
 
diff --git a/docs/zh_CN/rules/sinks/builtin/memory.md b/docs/zh_CN/rules/sinks/builtin/memory.md
index 33bd8858..7b91e610 100644
--- a/docs/zh_CN/rules/sinks/builtin/memory.md
+++ b/docs/zh_CN/rules/sinks/builtin/memory.md
@@ -4,9 +4,10 @@
 
 该动作用于将结果刷新到内存中的主题中，以便 [内存源](../../sources/builtin/memory.md) 可以使用它。 该主题类似于 pubsub 主题，例如 mqtt，因此可能有多个内存目标发布到同一主题，也可能有多个内存源订阅同一主题。 内存动作的典型用途是形成[规则管道](../../rule_pipeline.md)。
 
-| 属性名称  | 是否可选 | 描述                                  |
-|-------|------|-------------------------------------|
-| topic | 否    | 内存中的主题，例如 `analysis/result`, 支持动态属性 |
+| 属性名称         | 是否可选 | 描述                                     |
+|--------------|------|----------------------------------------|
+| topic        | 否    | 内存中的主题，例如 `analysis/result`, 支持动态属性    |
+| rowkindField | 是    | 指定哪个字段表示操作，例如插入或更新。如果不指定，默认所有的数据都是插入操作 |
 
 下面是一个内存动作配置示例：
 
diff --git a/docs/zh_CN/rules/sinks/builtin/redis.md b/docs/zh_CN/rules/sinks/builtin/redis.md
index 1e91b491..3a003c17 100644
--- a/docs/zh_CN/rules/sinks/builtin/redis.md
+++ b/docs/zh_CN/rules/sinks/builtin/redis.md
@@ -17,15 +17,16 @@ redis 源代码在 extensions 目录中，但是需要在 eKuiper 根目录编
 
 ## 属性
 
-| 属性名称       | 是否必填 | 说明                                                                                                                                                                                                                                       |
-|------------|------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| addr       | 是    | Redis 的地址, 例如: 10.122.48.17:6379                                                                                                                                                                                                         |
-| password   | 否    | Redis 登陆密码                                                                                                                                                                                                                               |
-| db         | 是    | Redis 的数据库,例如0                                                                                                                                                                                                                           |
-| key        | 是    | Redis 数据的 Key， key 与 field 选择其中一个, 优先 field                                                                                                                                                                                              |
-| field      | 否    | json 数据某一个属性，配置它作为 redis 数据的 key 值, 例如 deviceName, 该字段必须存在且为 string 类型，否则以 field 字符作为 key。比如 field 属性为 "deviceName", 收到 {“deviceName":"abc"}, 那么存入redis用的key是 "abc"; 收到 {“deviceName": 2}, 那么存入redis用的key是 "deviceName"。 注意:配置该值不要使用数据模板 |
-| dataType   | 是    | Redis 数据的类型, 默认是 string, 注意修改类型之后，需在redis中删除原有 key，否则修改无效。目前只支持 "list" 和 "string"                                                                                                                                                        |
-| expiration | 是    | 超时时间                                                                                                                                                                                                                                     |
+| 属性名称         | 是否必填 | 说明                                                                                                                                                                                                                                       |
+|--------------|------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| addr         | 是    | Redis 的地址, 例如: 10.122.48.17:6379                                                                                                                                                                                                         |
+| password     | 否    | Redis 登陆密码                                                                                                                                                                                                                               |
+| db           | 是    | Redis 的数据库,例如0                                                                                                                                                                                                                           |
+| key          | 是    | Redis 数据的 Key， key 与 field 选择其中一个, 优先 field                                                                                                                                                                                              |
+| field        | 否    | json 数据某一个属性，配置它作为 redis 数据的 key 值, 例如 deviceName, 该字段必须存在且为 string 类型，否则以 field 字符作为 key。比如 field 属性为 "deviceName", 收到 {“deviceName":"abc"}, 那么存入redis用的key是 "abc"; 收到 {“deviceName": 2}, 那么存入redis用的key是 "deviceName"。 注意:配置该值不要使用数据模板 |
+| dataType     | 是    | Redis 数据的类型, 默认是 string, 注意修改类型之后，需在redis中删除原有 key，否则修改无效。目前只支持 "list" 和 "string"                                                                                                                                                        |
+| expiration   | 是    | 超时时间                                                                                                                                                                                                                                     |
+| rowkindField | 是    | 指定哪个字段表示操作，例如插入或更新。如果不指定，默认所有的数据都是插入操作                                                                                                                                                                                                   |
 
 ## 示例用法
 
diff --git a/docs/zh_CN/rules/sinks/plugin/sql.md b/docs/zh_CN/rules/sinks/plugin/sql.md
index 2a5ee1d2..ce176586 100644
--- a/docs/zh_CN/rules/sinks/plugin/sql.md
+++ b/docs/zh_CN/rules/sinks/plugin/sql.md
@@ -30,6 +30,7 @@
 | table          | false    | 结果的表名                                       |
 | fields         | true     | 要插入的字段。结果映射和数据库都应该有这些字段。如果未指定，将插入结果映射中的所有字段 |
 | tableDataField | true     | 将 tableDataField 的嵌套值写入数据库。                 |
+| rowkindField   | true     | 指定哪个字段表示操作，例如插入或更新。如果不指定，默认所有的数据都是插入操作      |
 
 ## 使用样例
 
diff --git a/etc/mqtt_source.json b/etc/mqtt_source.json
index 151dff9d..0f48597c 100644
--- a/etc/mqtt_source.json
+++ b/etc/mqtt_source.json
@@ -80,17 +80,21 @@
 			}
 		}, {
 			"name": "protocolVersion",
-			"default": "3.1",
+			"default": "3.1.1",
 			"optional": true,
 			"connection_related": true,
-			"control": "text",
+			"control": "select",
+			"values": [
+				"3.1",
+				"3.1.1"
+			],
 			"type": "string",
 			"hint": {
 				"en_US": "MQTT protocol version. 3.1 (also refer as MQTT 3) or 3.1.1 (also refer as MQTT 4). If not specified, the default value is 3.1.",
 				"zh_CN": "MQTT 协议版本。3.1 (也被称为 MQTT 3) 或者 3.1.1 (也被称为 MQTT 4)。 如果未指定，缺省值为 3.1。 "
 			},
 			"label": {
-				"en_US": "protocolVersion",
+				"en_US": "MQTT Protocol Version",
 				"zh_CN": "MQTT 协议版本"
 			}
 		}, {
@@ -105,7 +109,7 @@
 				"zh_CN": "MQTT 连接的客户端 ID。 如果未指定，将使用一个 uuid。"
 			},
 			"label": {
-				"en_US": "clientid",
+				"en_US": "MQTT Client ID",
 				"zh_CN": "客户端 ID"
 			}
 		}, {
diff --git a/etc/sinks/edgex.json b/etc/sinks/edgex.json
index 96379975..dad2d3f8 100644
--- a/etc/sinks/edgex.json
+++ b/etc/sinks/edgex.json
@@ -43,7 +43,7 @@
         "tcp",
         "redis"
       ],
-      "type": "list_string",
+      "type": "string",
       "hint": {
         "en_US": "The protocol. If it's not specified, then use default value 'redis'.",
         "zh_CN": "协议，如未指定，使用缺省值 redis。"
@@ -128,7 +128,7 @@
         "nats-jetstream",
         "nats-core"
       ],
-      "type": "list_string",
+      "type": "string",
       "hint": {
         "en_US": "The message bus type, two types of message buses are supported,'redis', 'zero' or 'mqtt', and 'redis' is the default value.",
         "zh_CN": "消息总线类型，目前支持两种类型的消息总线，'redis', 'zero' 或者 'mqtt'，其中 'redis' 为缺省类型。"
@@ -147,7 +147,7 @@
         "event",
         "request"
       ],
-      "type": "list_string",
+      "type": "string",
       "hint": {
         "en_US": "The EdgeX message model type. To publish the message as an event like EdgeX application service, use `event`. Otherwise, to publish the message as an event request like EdgeX device service or core data service, use `request`. If not specified, then use the default value `event`.",
         "zh_CN": "EdgeX 消息模型类型。若要将消息发送为类似 apllication service 的 event 类型，则应设置为 `event`。否则，若要将消息发送为类似 device service 或者 core data service 的 event request 类型，则应设置为 `request`。如未指定，使用缺省值 `event` 。"
diff --git a/etc/sinks/memory.json b/etc/sinks/memory.json
index 90191083..d02b4114 100644
--- a/etc/sinks/memory.json
+++ b/etc/sinks/memory.json
@@ -31,6 +31,36 @@
         "en_US": "Topic",
         "zh_CN": "主题"
       }
+    },
+    {
+      "name": "rowkindField",
+      "default": "",
+      "optional": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "Specify which field represents the action like insert or update. If not specified, all rows are default to insert",
+        "zh_CN": "指定哪个字段表示操作，例如插入或更新。如果不指定，默认所有的数据都是插入操作"
+      },
+      "label": {
+        "en_US": "Rowkind Field",
+        "zh_CN": "动作字段"
+      }
+    },
+    {
+      "name": "keyField",
+      "default": "",
+      "optional": true,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "Specify which field represents the key of the message. It is required when rowkindField is specified.",
+        "zh_CN": "指定哪个字段表示消息的key。当指定了动作字段时，该字段是必须的。"
+      },
+      "label": {
+        "en_US": "Key Field",
+        "zh_CN": "Key字段"
+      }
     }
   ],
   "node": {
diff --git a/etc/sinks/mqtt.json b/etc/sinks/mqtt.json
index 6bca64f3..a4f494a1 100644
--- a/etc/sinks/mqtt.json
+++ b/etc/sinks/mqtt.json
@@ -82,7 +82,7 @@
     },
     {
       "name": "protocolVersion",
-      "default": "3.1",
+      "default": "3.1.1",
       "optional": true,
       "control": "select",
       "values": [
diff --git a/etc/sinks/neuron.json b/etc/sinks/neuron.json
index 9b8fb6d1..cd85319a 100644
--- a/etc/sinks/neuron.json
+++ b/etc/sinks/neuron.json
@@ -19,7 +19,7 @@
   "properties": [
     {
       "name": "nodeName",
-      "optional": false,
+      "optional": true,
       "control": "text",
       "default": "",
       "type": "string",
@@ -34,7 +34,7 @@
     },
     {
       "name": "groupName",
-      "optional": false,
+      "optional": true,
       "control": "text",
       "default": "",
       "type": "string",
@@ -50,7 +50,7 @@
     {
       "name": "tags",
       "default": [],
-      "optional": false,
+      "optional": true,
       "control": "list",
       "type": "list_string",
       "hint": {
diff --git a/etc/sinks/redis.json b/etc/sinks/redis.json
index 72b6b820..07cdafde 100644
--- a/etc/sinks/redis.json
+++ b/etc/sinks/redis.json
@@ -31,7 +31,7 @@
 				"zh_CN": "Redis的地址, 例如: 10.122.48.17:6379"
 			},
 			"label": {
-				"en_US": "Addr",
+				"en_US": "Address",
 				"zh_CN": "地址"
 			}
 		},
@@ -68,7 +68,7 @@
 		{
 			"name": "key",
 			"default": "key",
-			"optional": false,
+			"optional": true,
 			"control": "text",
 			"type": "string",
 			"hint": {
@@ -91,8 +91,8 @@
 				"zh_CN": "json数据某一个属性，配置它作为redis数据的key值, 例如 deviceName,该字段必须存在且为string类型，否则以field字符作为key,注意:配置该值不要使用数据模板"
 			},
 			"label": {
-				"en_US": "field",
-				"zh_CN": "field"
+				"en_US": "Key Field",
+				"zh_CN": "Key 字段"
 			}
 		},
 		{
@@ -110,7 +110,7 @@
 				"zh_CN": "Redis 数据的类型, 默认是 string, 注意修改类型之后，需在redis中删除原有key，否则修改无效。"
 			},
 			"label": {
-				"en_US": "data type",
+				"en_US": "Data type",
 				"zh_CN": "数据类型"
 			}
 		},
@@ -125,9 +125,32 @@
 				"zh_CN": "Redis数据的超时时间，仅在string类型数据有效,单位是秒,默认是永久保存-1 "
 			},
 			"label": {
-				"en_US": "expiration",
+				"en_US": "Expiration",
 				"zh_CN": "超时时间"
 			}
+		},
+		{
+			"name": "rowkindField",
+			"default": "",
+			"optional": true,
+			"control": "text",
+			"type": "string",
+			"hint": {
+				"en_US": "Specify which field represents the action like insert or update. If not specified, all rows are default to insert",
+				"zh_CN": "指定哪个字段表示操作，例如插入或更新。如果不指定，默认所有的数据都是插入操作"
+			},
+			"label": {
+				"en_US": "Rowkind Field",
+				"zh_CN": "动作字段"
+			}
+		}
+	],
+	"node": {
+		"category": "sink",
+		"icon": "iconPath",
+		"label": {
+			"en": "Redis",
+			"zh": "Redis"
 		}
-	]
+	}
 }
diff --git a/etc/sinks/rest.json b/etc/sinks/rest.json
index 0a197862..556fdd0c 100644
--- a/etc/sinks/rest.json
+++ b/etc/sinks/rest.json
@@ -37,7 +37,7 @@
       "optional": true,
       "control": "select",
       "default": "GET",
-      "type": "list_string",
+      "type": "string",
       "values": [
         "GET",
         "POST",
@@ -59,7 +59,7 @@
       "optional": true,
       "control": "select",
       "default": "none",
-      "type": "list_string",
+      "type": "string",
       "values": [
         "none",
         "json",
diff --git a/etc/sources/edgex.json b/etc/sources/edgex.json
index 22805995..0e33d6aa 100644
--- a/etc/sources/edgex.json
+++ b/etc/sources/edgex.json
@@ -114,7 +114,7 @@
 					"nats-jetstream",
 					"nats-core"
 				],
-				"type": "list_string",
+				"type": "string",
 				"hint": {
 					"en_US": "The EdgeX message bus type, currently two types of message buses are supported. If specified other values, then will use the default redis value.",
 					"zh_CN": "EdgeX 消息总线类型，目前支持两种消息总线。如果指定了错误的消息总线类型，那么会使用缺省 redis 类型。"
@@ -133,7 +133,7 @@
 					"event",
 					"request"
 				],
-				"type": "list_string",
+				"type": "string",
 				"hint": {
 					"en_US": "The EdgeX message model type. If connected to the topic of EdgeX application service, the message model is an event. Otherwise, if connected to the topic of EdgeX message bus directly to receive the message from device service or core data, the message is request.",
 					"zh_CN": "EdgeX 消息模型类型。如果连接到 EdgeX application service 的 topic, 则消息为 event 类型。否则，如果直接连接到消息总线的 topic，接收到 device service 或者 core data 发出的数据，则消息类型为 request。"
diff --git a/etc/sources/httppush.json b/etc/sources/httppush.json
index 7d1fdd97..a10b7064 100644
--- a/etc/sources/httppush.json
+++ b/etc/sources/httppush.json
@@ -22,7 +22,7 @@
 			"name": "method",
 			"default": "POST",
 			"optional": true,
-			"control": "text",
+			"control": "select",
 			"type": "string",
 			"values": ["POST","PUT"],
 			"hint": {
diff --git a/extensions/sinks/image/image.json b/extensions/sinks/image/image.json
index 22b50392..680120d8 100644
--- a/extensions/sinks/image/image.json
+++ b/extensions/sinks/image/image.json
@@ -41,7 +41,7 @@
 			"jpeg",
 			"png"
 		],
-		"type": "list_string",
+		"type": "string",
 		"hint": {
 			"en_US": "File format, support jpeg and png.",
 			"zh_CN": "文件格式，支持 jpeg 和 png。"
diff --git a/extensions/sinks/sql/sql.json b/extensions/sinks/sql/sql.json
index 78acb756..ce7a0b0b 100644
--- a/extensions/sinks/sql/sql.json
+++ b/extensions/sinks/sql/sql.json
@@ -74,6 +74,21 @@
 			"en_US": "Table field",
 			"zh_CN": "表字段"
 		}
+	},
+	{
+		"name": "rowkindField",
+		"default": "",
+		"optional": true,
+		"control": "text",
+		"type": "string",
+		"hint": {
+			"en_US": "Specify which field represents the action like insert or update. If not specified, all rows are default to insert",
+			"zh_CN": "指定哪个字段表示操作，例如插入或更新。如果不指定，默认所有的数据都是插入操作"
+		},
+		"label": {
+			"en_US": "Rowkind Field",
+			"zh_CN": "动作字段"
+		}
 	}
  ],
 	"node": {
diff --git a/extensions/sources/sql/sql.json b/extensions/sources/sql/sql.json
index 89347e04..d26f8bea 100644
--- a/extensions/sources/sql/sql.json
+++ b/extensions/sources/sql/sql.json
@@ -30,14 +30,14 @@
           "zh_CN": "数据库服务器的 URL"
         },
         "label": {
-          "en_US": "server address",
+          "en_US": "Server address",
           "zh_CN": "数据库地址"
         }
       },
       {
         "name": "interval",
         "default": 1000,
-        "optional": false,
+        "optional": true,
         "control": "text",
         "type": "int",
         "hint": {
@@ -63,7 +63,7 @@
               "zh_CN": "指定查询的数据库表名"
             },
             "label": {
-              "en_US": "tableName",
+              "en_US": "Table Name",
               "zh_CN": "表名"
             }
           },
@@ -78,7 +78,7 @@
               "zh_CN": "索引字段名"
             },
             "label": {
-              "en_US": "indexField",
+              "en_US": "Index Field",
               "zh_CN": "索引字段名"
             }
           },
@@ -93,7 +93,7 @@
               "zh_CN": "索引字段初始值"
             },
             "label": {
-              "en_US": "index init value",
+              "en_US": "Index init value",
               "zh_CN": "索引字段初始值"
             }
           },
@@ -118,14 +118,14 @@
             "optional": true,
             "control": "select",
             "type": "string",
-            "values": ["DATETIME"],
+            "values": ["DATETIME", "INT", "FLOAT", "STRING"],
             "hint": {
               "en_US": "is the index datetime type",
               "zh_CN": "是否为时间格式"
             },
             "label": {
-              "en_US": "indexFieldType",
-              "zh_CN": "indexFieldType"
+              "en_US": "Index Field Type",
+              "zh_CN": "索引字段格式"
             }
           },
           {
@@ -139,8 +139,8 @@
               "zh_CN": "dateTimeFormat"
             },
             "label": {
-              "en_US": "dateTimeFormat",
-              "zh_CN": "dateTimeFormat"
+              "en_US": "DateTime Format",
+              "zh_CN": "时间格式"
             }
           }
         ],
@@ -170,7 +170,7 @@
               "zh_CN": "查询语句模版"
             },
             "label": {
-              "en_US": "query template",
+              "en_US": "Query template",
               "zh_CN": "查询语句模版"
             }
           },
@@ -185,7 +185,7 @@
               "zh_CN": "索引字段名"
             },
             "label": {
-              "en_US": "indexField",
+              "en_US": "Index Field",
               "zh_CN": "索引字段名"
             }
           },
@@ -200,7 +200,7 @@
               "zh_CN": "索引字段初始值"
             },
             "label": {
-              "en_US": "index init value",
+              "en_US": "Index init value",
               "zh_CN": "索引字段初始值"
             }
           },
@@ -210,14 +210,14 @@
             "optional": true,
             "control": "select",
             "type": "string",
-            "values": ["DATETIME"],
+            "values": ["DATETIME", "INT", "FLOAT", "STRING"],
             "hint": {
               "en_US": "is the index datetime type",
               "zh_CN": "是否为时间格式"
             },
             "label": {
-              "en_US": "indexFieldType",
-              "zh_CN": "indexFieldType"
+              "en_US": "Index Field Type",
+              "zh_CN": "索引字段格式"
             }
           },
           {
@@ -231,8 +231,8 @@
               "zh_CN": "dateTimeFormat"
             },
             "label": {
-              "en_US": "dateTimeFormat",
-              "zh_CN": "dateTimeFormat"
+              "en_US": "DateTime Format",
+              "zh_CN": "时间格式"
             }
           }
         ],
@@ -260,7 +260,7 @@
     }
   ],
   "node": {
-    "category": ["Protocol","Protocol1"],
+    "category": "source",
     "icon": "iconPath",
     "label": {
       "en_US": "SQL",
diff --git a/internal/binder/function/funcs_agg.go b/internal/binder/function/funcs_agg.go
index 38172a06..8e6bb351 100644
--- a/internal/binder/function/funcs_agg.go
+++ b/internal/binder/function/funcs_agg.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
+	"github.com/lf-edge/ekuiper/pkg/cast"
 )
 
 func registerAggFunc() {
@@ -33,7 +34,7 @@ func registerAggFunc() {
 					if r, err := sliceIntTotal(arg0); err != nil {
 						return err, false
 					} else {
-						return r / c, true
+						return r / int64(c), true
 					}
 				case float64:
 					if r, err := sliceFloatTotal(arg0); err != nil {
@@ -67,13 +68,13 @@ func registerAggFunc() {
 				v := getFirstValidArg(arg0)
 				switch t := v.(type) {
 				case int:
-					if r, err := sliceIntMax(arg0, t); err != nil {
+					if r, err := sliceIntMax(arg0, int64(t)); err != nil {
 						return err, false
 					} else {
 						return r, true
 					}
 				case int64:
-					if r, err := sliceIntMax(arg0, int(t)); err != nil {
+					if r, err := sliceIntMax(arg0, t); err != nil {
 						return err, false
 					} else {
 						return r, true
@@ -108,13 +109,13 @@ func registerAggFunc() {
 				v := getFirstValidArg(arg0)
 				switch t := v.(type) {
 				case int:
-					if r, err := sliceIntMin(arg0, t); err != nil {
+					if r, err := sliceIntMin(arg0, int64(t)); err != nil {
 						return err, false
 					} else {
 						return r, true
 					}
 				case int64:
-					if r, err := sliceIntMin(arg0, int(t)); err != nil {
+					if r, err := sliceIntMin(arg0, t); err != nil {
 						return err, false
 					} else {
 						return r, true
@@ -227,10 +228,11 @@ func getFirstValidArg(s []interface{}) interface{} {
 	return nil
 }
 
-func sliceIntTotal(s []interface{}) (int, error) {
-	var total int
+func sliceIntTotal(s []interface{}) (int64, error) {
+	var total int64
 	for _, v := range s {
-		if vi, ok := v.(int); ok {
+		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
+		if err == nil {
 			total += vi
 		} else if v != nil {
 			return 0, fmt.Errorf("requires int but found %[1]T(%[1]v)", v)
@@ -250,14 +252,15 @@ func sliceFloatTotal(s []interface{}) (float64, error) {
 	}
 	return total, nil
 }
-func sliceIntMax(s []interface{}, max int) (int, error) {
+func sliceIntMax(s []interface{}, max int64) (int64, error) {
 	for _, v := range s {
-		if vi, ok := v.(int); ok {
-			if max < vi {
+		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
+		if err == nil {
+			if vi > max {
 				max = vi
 			}
 		} else if v != nil {
-			return 0, fmt.Errorf("requires int but found %[1]T(%[1]v)", v)
+			return 0, fmt.Errorf("requires int64 but found %[1]T(%[1]v)", v)
 		}
 	}
 	return max, nil
@@ -287,14 +290,15 @@ func sliceStringMax(s []interface{}, max string) (string, error) {
 	}
 	return max, nil
 }
-func sliceIntMin(s []interface{}, min int) (int, error) {
+func sliceIntMin(s []interface{}, min int64) (int64, error) {
 	for _, v := range s {
-		if vi, ok := v.(int); ok {
-			if min > vi {
+		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
+		if err == nil {
+			if vi < min {
 				min = vi
 			}
 		} else if v != nil {
-			return 0, fmt.Errorf("requires int but found %[1]T(%[1]v)", v)
+			return 0, fmt.Errorf("requires int64 but found %[1]T(%[1]v)", v)
 		}
 	}
 	return min, nil
@@ -315,7 +319,7 @@ func sliceFloatMin(s []interface{}, min float64) (float64, error) {
 func sliceStringMin(s []interface{}, min string) (string, error) {
 	for _, v := range s {
 		if vs, ok := v.(string); ok {
-			if min < vs {
+			if vs < min {
 				min = vs
 			}
 		} else if v != nil {
diff --git a/internal/pkg/httpx/http.go b/internal/pkg/httpx/http.go
index 6171953f..0844ef6c 100644
--- a/internal/pkg/httpx/http.go
+++ b/internal/pkg/httpx/http.go
@@ -220,3 +220,17 @@ func DownloadFile(filepath string, uri string) error {
 	_, err = io.Copy(out, src)
 	return err
 }
+
+func IsHttpUrl(str string) error {
+	url, err := url.ParseRequestURI(str)
+	if err != nil {
+		return err
+	}
+	if url.Scheme != "http" && url.Scheme != "https" {
+		return fmt.Errorf("Invalid scheme %s", url.Scheme)
+	}
+	if url.Host == "" {
+		return fmt.Errorf("Invalid url, host not found")
+	}
+	return nil
+}
diff --git a/internal/topo/neuron/sink.go b/internal/topo/neuron/sink.go
index beed3bc8..bc34d070 100644
--- a/internal/topo/neuron/sink.go
+++ b/internal/topo/neuron/sink.go
@@ -48,14 +48,20 @@ type neuronTemplate struct {
 func (s *sink) Configure(props map[string]interface{}) error {
 	s.url = NeuronUrl
 	cc := &c{
-		NodeName:  "unknown",
-		GroupName: "unknown",
-		Raw:       false,
+		Raw: false,
 	}
 	err := cast.MapToStruct(props, cc)
 	if err != nil {
 		return err
 	}
+	if !cc.Raw {
+		if cc.NodeName == "" {
+			return fmt.Errorf("node name is required if raw is not set")
+		}
+		if cc.GroupName == "" {
+			return fmt.Errorf("group name is required if raw is not set")
+		}
+	}
 	s.c = cc
 	return nil
 }
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index 184b1d4a..0bc4d7a5 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -188,7 +188,7 @@ func SinkOpen(sinkType string, config map[string]interface{}) error {
 }
 
 func SourceOpen(sourceType string, config map[string]interface{}) error {
-	dataSource := "testSource"
+	dataSource := "/$$TEST_CONNECTION$$"
 	if v, ok := config["DATASOURCE"]; ok {
 		dataSource = v.(string)
 	}
diff --git a/internal/topo/redis/sink.go b/internal/topo/redis/sink.go
index 38f895f3..009bf209 100644
--- a/internal/topo/redis/sink.go
+++ b/internal/topo/redis/sink.go
@@ -78,8 +78,8 @@ func (r *RedisSink) Open(ctx api.StreamContext) (err error) {
 		Password: r.c.Password,
 		DB:       r.c.Db, // use default DB
 	})
-
-	return nil
+	_, err = r.cli.Ping().Result()
+	return err
 }
 
 func (r *RedisSink) Collect(ctx api.StreamContext, data interface{}) error {
diff --git a/internal/topo/sink/rest_sink.go b/internal/topo/sink/rest_sink.go
index 72971743..0615a891 100644
--- a/internal/topo/sink/rest_sink.go
+++ b/internal/topo/sink/rest_sink.go
@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"net/url"
 	"strings"
 	"time"
 
@@ -201,7 +200,7 @@ func (ms *RestSink) Open(ctx api.StreamContext) error {
 		Timeout:   time.Duration(ms.timeout) * time.Millisecond}
 	logger.Infof("open rest sink with configuration: {method: %s, url: %s, bodyType: %s, timeout: %d, header: %v, sendSingle: %v, tls cfg: %v", ms.method, ms.url, ms.bodyType, ms.timeout, ms.headers, ms.sendSingle, tlsOpts)
 
-	if _, err := url.Parse(ms.url); err != nil {
+	if err := httpx.IsHttpUrl(ms.url); err != nil {
 		return err
 	}
 	return nil
diff --git a/internal/topo/source/file_source.go b/internal/topo/source/file_source.go
index 7b6113e8..af716f8d 100644
--- a/internal/topo/source/file_source.go
+++ b/internal/topo/source/file_source.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -80,13 +80,15 @@ func (fs *FileSource) Configure(fileName string, props map[string]interface{}) e
 			return fmt.Errorf("invalid path %s", cfg.Path)
 		}
 	}
-	fs.file = path.Join(cfg.Path, fileName)
+	if fileName != "/$$TEST_CONNECTION$$" {
+		fs.file = path.Join(cfg.Path, fileName)
 
-	if fi, err := os.Stat(fs.file); err != nil {
-		if os.IsNotExist(err) {
-			return fmt.Errorf("file %s not exist", fs.file)
-		} else if !fi.Mode().IsRegular() {
-			return fmt.Errorf("file %s is not a regular file", fs.file)
+		if fi, err := os.Stat(fs.file); err != nil {
+			if os.IsNotExist(err) {
+				return fmt.Errorf("file %s not exist", fs.file)
+			} else if !fi.Mode().IsRegular() {
+				return fmt.Errorf("file %s is not a regular file", fs.file)
+			}
 		}
 	}
 	fs.config = cfg
diff --git a/internal/topo/source/httppull_source.go b/internal/topo/source/httppull_source.go
index a131b3a3..bc2f7e6e 100644
--- a/internal/topo/source/httppull_source.go
+++ b/internal/topo/source/httppull_source.go
@@ -21,7 +21,6 @@ import (
 	"github.com/lf-edge/ekuiper/internal/pkg/cert"
 	"io"
 	"net/http"
-	"net/url"
 	"strings"
 	"time"
 
@@ -167,7 +166,7 @@ func (hps *HTTPPullSource) Configure(device string, props map[string]interface{}
 		}
 	}
 
-	_, err := url.Parse(hps.url)
+	err := httpx.IsHttpUrl(hps.url)
 	if err != nil {
 		return err
 	}
diff --git a/internal/xsql/row.go b/internal/xsql/row.go
index 1e687bca..5d5fb827 100644
--- a/internal/xsql/row.go
+++ b/internal/xsql/row.go
@@ -220,7 +220,7 @@ type Tuple struct {
 	Metadata  Metadata // immutable
 
 	AffiliateRow
-
+	lock      sync.Mutex             // lock for the cachedMap, because it is possible to access by multiple sinks
 	cachedMap map[string]interface{} // clone of the row and cached for performance
 }
 
@@ -230,6 +230,7 @@ var _ TupleRow = &Tuple{}
 type JoinTuple struct {
 	Tuples []TupleRow // The content is immutable, but the slice may be add or removed
 	AffiliateRow
+	lock      sync.Mutex
 	cachedMap map[string]interface{} // clone of the row and cached for performance of toMap
 }
 
@@ -244,6 +245,7 @@ type GroupedTuples struct {
 	Content []TupleRow
 	*WindowRange
 	AffiliateRow
+	lock      sync.Mutex
 	cachedMap map[string]interface{} // clone of the row and cached for performance of toMap
 }
 
@@ -339,6 +341,8 @@ func (t *Tuple) Clone() TupleRow {
 
 // ToMap should only use in sink.
 func (t *Tuple) ToMap() map[string]interface{} {
+	t.lock.Lock()
+	defer t.lock.Unlock()
 	if t.AffiliateRow.IsEmpty() {
 		return t.Message
 	}
@@ -487,6 +491,8 @@ func (jt *JoinTuple) Clone() TupleRow {
 }
 
 func (jt *JoinTuple) ToMap() map[string]interface{} {
+	jt.lock.Lock()
+	defer jt.lock.Unlock()
 	if jt.cachedMap == nil { // clone the message
 		m := make(map[string]interface{})
 		for i := len(jt.Tuples) - 1; i >= 0; i-- {
@@ -544,6 +550,8 @@ func (s *GroupedTuples) All(_ string) (Message, bool) {
 }
 
 func (s *GroupedTuples) ToMap() map[string]interface{} {
+	s.lock.Lock()
+	defer s.lock.Unlock()
 	if s.cachedMap == nil {
 		m := make(map[string]interface{})
 		for k, v := range s.Content[0].ToMap() {


Test patch:
diff --git a/internal/binder/function/funcs_agg_test.go b/internal/binder/function/funcs_agg_test.go
new file mode 100644
index 00000000..8f14be89
--- /dev/null
+++ b/internal/binder/function/funcs_agg_test.go
@@ -0,0 +1,108 @@
+// 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 function
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/state"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"reflect"
+	"testing"
+)
+
+func TestAggExec(t *testing.T) {
+	fAvg, ok := builtins["avg"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	fMax, ok := builtins["max"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	fMin, ok := builtins["min"]
+	if !ok {
+		t.Fatal("builtin not found")
+	}
+	contextLogger := conf.Log.WithField("rule", "testExec")
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	tempStore, _ := state.CreateStore("mockRule0", api.AtMostOnce)
+	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("mockRule0", "test", tempStore), 2)
+	var tests = []struct {
+		args []interface{}
+		avg  interface{}
+		max  interface{}
+		min  interface{}
+	}{
+		{ // 0
+			args: []interface{}{
+				[]interface{}{
+					"foo",
+					"bar",
+					"self",
+				},
+			},
+			avg: fmt.Errorf("run avg function error: found invalid arg string(foo)"),
+			max: "self",
+			min: "bar",
+		}, { // 1
+			args: []interface{}{
+				[]interface{}{
+					int64(100),
+					int64(150),
+					int64(200),
+				},
+			},
+			avg: int64(150),
+			max: int64(200),
+			min: int64(100),
+		}, { // 2
+			args: []interface{}{
+				[]interface{}{
+					float64(100),
+					float64(150),
+					float64(200),
+				},
+			},
+			avg: float64(150),
+			max: float64(200),
+			min: float64(100),
+		}, { // 3
+			args: []interface{}{
+				[]interface{}{
+					100, 150, 200,
+				},
+			},
+			avg: int64(150),
+			max: int64(200),
+			min: int64(100),
+		},
+	}
+	for i, tt := range tests {
+		rAvg, _ := fAvg.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rAvg, tt.avg) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rAvg, tt.avg)
+		}
+		rMax, _ := fMax.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rMax, tt.max) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rMax, tt.max)
+		}
+		rMin, _ := fMin.exec(fctx, tt.args)
+		if !reflect.DeepEqual(rMin, tt.min) {
+			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, rMin, tt.min)
+		}
+	}
+}
diff --git a/internal/pkg/httpx/http_test.go b/internal/pkg/httpx/http_test.go
new file mode 100644
index 00000000..ad0226b6
--- /dev/null
+++ b/internal/pkg/httpx/http_test.go
@@ -0,0 +1,40 @@
+// 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 httpx
+
+import "testing"
+
+func TestIsUrl(t *testing.T) {
+	urls := []string{
+		"http://localhost:8080/abc",
+		"https://localhost:8080/abc",
+		"http://122.122.122:8080/abc",
+	}
+	for _, u := range urls {
+		if err := IsHttpUrl(u); err != nil {
+			t.Errorf("expect %s is url but got %v", u, err)
+		}
+	}
+	badUrls := []string{
+		"ws://localhost:8080/abc",
+		"http:/baidu.com:8080/abc",
+		"localhost:8080/abc",
+	}
+	for _, u := range badUrls {
+		if err := IsHttpUrl(u); err == nil {
+			t.Errorf("expect %s is not url but passed", u)
+		}
+	}
+}
diff --git a/internal/topo/operator/project_test.go b/internal/topo/operator/project_test.go
index a65e7f9a..20d9b2c1 100644
--- a/internal/topo/operator/project_test.go
+++ b/internal/topo/operator/project_test.go
@@ -1714,7 +1714,7 @@ func TestProjectPlan_AggFuncs(t *testing.T) {
 				WindowRange: xsql.NewWindowRange(1541152486013, 1541152487013),
 			},
 			result: []map[string]interface{}{{
-				"sum":        123203,
+				"sum":        int64(123203),
 				"ws":         int64(1541152486013),
 				"window_end": int64(1541152487013),
 			}},
@@ -1735,7 +1735,7 @@ func TestProjectPlan_AggFuncs(t *testing.T) {
 			},
 
 			result: []map[string]interface{}{{
-				"s": 123203,
+				"s": int64(123203),
 			}},
 		},
 		//8
@@ -1753,7 +1753,7 @@ func TestProjectPlan_AggFuncs(t *testing.T) {
 				},
 			},
 			result: []map[string]interface{}{{
-				"sum": 123203,
+				"sum": int64(123203),
 			}},
 		},
 		//9
@@ -1773,10 +1773,10 @@ func TestProjectPlan_AggFuncs(t *testing.T) {
 			result: []map[string]interface{}{{
 				"all": 3,
 				"c":   2,
-				"a":   40,
-				"s":   80,
-				"min": 27,
-				"max": 53,
+				"a":   int64(40),
+				"s":   int64(80),
+				"min": int64(27),
+				"max": int64(53),
 			}},
 		},
 		//10
@@ -2158,8 +2158,8 @@ func TestProjectPlan_AggFuncs(t *testing.T) {
 			},
 			result: []map[string]interface{}{{
 				"var2": "moduleB topic",
-				"max2": 1,
-				"max3": 100,
+				"max2": int64(1),
+				"max3": int64(100),
 			}},
 		},
 	}