Repository: lf-edge/ekuiper
Pull request #1442 (2022-10-18 09:34:09)

fix(fea): support sink config by resource id
Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>


Patch:
diff --git a/.github/workflows/deploy_docs.yaml b/.github/workflows/deploy_docs.yaml
index 655c94d1..d409e541 100644
--- a/.github/workflows/deploy_docs.yaml
+++ b/.github/workflows/deploy_docs.yaml
@@ -92,7 +92,7 @@ jobs:
         tccli configure set secretId ${{ secrets.TENCENT_COS_ID }}
         tccli configure set secretKey ${{ secrets.TENCENT_COS_KEY }}
         tccli configure set region ap-hongkong
-        tccli cdn PurgePathCache --Paths '["https://ekuiper.org/docs/"]' --FlushType delete
+        tccli cdn PurgePathCache --Paths '["https://ekuiper.org/docs/", "https://ekuiper-docs.emqx.net/"]' --FlushType delete
 
     - name: update search index
       uses: Swilder-M/docsearch-scraper-simple@v4
diff --git a/docs/en_US/operation/config/configuration_file.md b/docs/en_US/operation/config/configuration_file.md
index 69cc7854..687adaaa 100644
--- a/docs/en_US/operation/config/configuration_file.md
+++ b/docs/en_US/operation/config/configuration_file.md
@@ -86,7 +86,9 @@ basic:
   prometheus: true
   prometheusPort: 20499
 ```
-For such a default configuration, eKuiper will export metrics and serve prometheus at `http://localhost:20499/metrics`
+For such a default configuration, eKuiper will export metrics and serve prometheus at `http://localhost:20499/metrics`.
+
+The prometheus port can be the same as the eKuiper REST API port. If so, both service will be served on the same server.
 
 ## Pluginhosts Configuration
 
diff --git a/docs/en_US/rules/sinks/overview.md b/docs/en_US/rules/sinks/overview.md
index 217efb23..8a2f35f0 100644
--- a/docs/en_US/rules/sinks/overview.md
+++ b/docs/en_US/rules/sinks/overview.md
@@ -103,4 +103,49 @@ In the following example configuration of the rule, log sink has no cache-relate
     }
   }]
 }
+```
+
+## Resource Reuse
+
+Like sources, actions also support configuration reuse. Users only need to create a yaml file with the same name as the target action in the sinks folder and write the configuration in the same form as the source.
+
+For example, for the MQTT action scenario, the user can create the mqtt.yaml file in the sinks directory and write the following content
+
+```yaml
+test:
+  qos: 1
+  server: "tcp://broker.emqx.io:1883"
+```
+
+When users need MQTT actions, in addition to the traditional configuration method, as shown below
+
+```json
+    {
+      "mqtt": {
+        "server": "tcp://broker.emqx.io:1883",
+        "topic": "devices/demo_001/messages/events/",
+        "protocolVersion": "3.1.1",
+        "qos": 1,
+        "clientId": "demo_001",
+        "username": "xyz.azure-devices.net/demo_001/?api-version=2018-06-30",
+        "password": "SharedAccessSignature sr=*******************",
+        "retained": false
+      }
+    }
+```
+
+Can also use the `resourceId` reference form with the following configuration
+
+```json
+{
+      "mqtt": {
+        "resourceId": "test",
+        "topic": "devices/demo_001/messages/events/",
+        "protocolVersion": "3.1.1",
+        "clientId": "demo_001",
+        "username": "xyz.azure-devices.net/demo_001/?api-version=2018-06-30",
+        "password": "SharedAccessSignature sr=*******************",
+        "retained": false
+      }
+}
 ```
\ No newline at end of file
diff --git a/docs/zh_CN/operation/config/configuration_file.md b/docs/zh_CN/operation/config/configuration_file.md
index c40e9148..05779d5d 100755
--- a/docs/zh_CN/operation/config/configuration_file.md
+++ b/docs/zh_CN/operation/config/configuration_file.md
@@ -87,7 +87,9 @@ basic:
   prometheus: true
   prometheusPort: 20499
 ```
-在如上默认配置中，eKuiper 暴露于 Prometheusd 运行指标可通过 `http://localhost:20499/metrics` 访问。
+在如上默认配置中，eKuiper 暴露于 Prometheus 运行指标可通过 `http://localhost:20499/metrics` 访问。
+
+Prometheus 端口可设置为与 eKuiper 的 REST 服务端口相同。这样设置的话，两个服务将运行在同一个 HTTP 服务中。
 
 ## Pluginhosts 配置
 
diff --git a/docs/zh_CN/rules/sinks/overview.md b/docs/zh_CN/rules/sinks/overview.md
index efe10aa7..de3ebd9f 100644
--- a/docs/zh_CN/rules/sinks/overview.md
+++ b/docs/zh_CN/rules/sinks/overview.md
@@ -103,4 +103,45 @@ Sink 缓存的配置有两个层次。`etc/kuiper.yaml` 中的全局配置，定
     }
   }]
 }
+```
+
+## 资源引用
+ 
+像源一样，动作也支持配置复用，用户只需要在 sinks 文件夹中创建与目标动作同名的 yaml 文件并按照源一样的形式写入配置。
+
+例如，针对 MQTT 动作场景， 用户可以在 sinks 目录下创建 mqtt.yaml 文件，并写入如下内容
+```yaml
+test:
+  qos: 1
+  server: "tcp://broker.emqx.io:1883"
+```
+当用户需要 MQTT 动作时，除了采用传统的配置方式，如下所示
+```json
+    {
+      "mqtt": {
+        "server": "tcp://broker.emqx.io:1883",
+        "topic": "devices/demo_001/messages/events/",
+        "protocolVersion": "3.1.1",
+        "qos": 1,
+        "clientId": "demo_001",
+        "username": "xyz.azure-devices.net/demo_001/?api-version=2018-06-30",
+        "password": "SharedAccessSignature sr=*******************",
+        "retained": false
+      }
+    }
+```
+还可以通过 `resourceId` 引用形式，采用如下的配置
+
+```json
+ {
+      "mqtt": {
+        "resourceId": "test",
+        "topic": "devices/demo_001/messages/events/",
+        "protocolVersion": "3.1.1",
+        "clientId": "demo_001",
+        "username": "xyz.azure-devices.net/demo_001/?api-version=2018-06-30",
+        "password": "SharedAccessSignature sr=*******************",
+        "retained": false
+      }
+}
 ```
\ No newline at end of file
diff --git a/etc/functions/internal.json b/etc/functions/internal.json
index 45052a2e..33aa472e 100644
--- a/etc/functions/internal.json
+++ b/etc/functions/internal.json
@@ -19,51 +19,289 @@
 	"functions": [{
 		"name": "avg",
 		"example": "avg(col1)",
+		"aggregate": true,
 		"hint": {
 			"en_US": "The average of the values in a group. The null values will be ignored.",
 			"zh_CN": "组中的平均值。空值不参与计算。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate the average.",
+					"zh_CN": "用于计算平均值的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Average",
+				"zh_CN": "平均值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Average",
+				"zh_CN": "平均值"
+			}
 		}
 	}, {
 		"name": "count",
 		"example": "count(*)",
+		"aggregate": true,
 		"hint": {
 			"en_US": "The number of items in a group. The null values will be ignored.",
 			"zh_CN": "组中的项目数。空值不参与计算。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "any",
+				"hint": {
+					"en_US": "The field to calculate the count.",
+					"zh_CN": "用于计数的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Count",
+				"zh_CN": "计数值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Count",
+				"zh_CN": "计数"
+			}
 		}
 	}, {
 		"name": "max",
 		"example": "max(col1)",
+		"aggregate": true,
 		"hint": {
 			"en_US": "The maximum value in a group. The null values will be ignored.",
 			"zh_CN": "组中的最大值。空值不参与计算。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate the maximum value.",
+					"zh_CN": "用于计算最大值的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Maximum",
+				"zh_CN": "最大值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Maximum",
+				"zh_CN": "最大值"
+			}
 		}
 	}, {
 		"name": "min",
 		"example": "min(col1)",
+		"aggregate": true,
 		"hint": {
 			"en_US": "The minimum value in a group. The null values will be ignored.",
 			"zh_CN": "组中的最小值。空值不参与计算。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate the minimum value.",
+					"zh_CN": "用于计算最小值的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Minimum",
+				"zh_CN": "最小值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Minimum",
+				"zh_CN": "最小值"
+			}
 		}
 	}, {
 		"name": "sum",
 		"example": "sum(col1)",
+		"aggregate": true,
 		"hint": {
 			"en_US": "The sum of all the values in a group. The null values will be ignored.",
 			"zh_CN": "组中所有值的总和。空值不参与计算。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate the sum.",
+					"zh_CN": "用于计算总和的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Sum",
+				"zh_CN": "总和"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Sum",
+				"zh_CN": "总和"
+			}
 		}
 	}, {
 		"name": "collect",
 		"example": "collect(*), collect(col1)",
+		"aggregate": true,
 		"hint": {
 			"en_US": "Returns an array with all column or the whole record (when the parameter is *) values from the group.",
 			"zh_CN": "返回组中指定的列或整个消息（参数为*时）的值组成的数组。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "any",
+				"hint": {
+					"en_US": "The field to return in the array.",
+					"zh_CN": "要返回列表的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "array",
+			"hint": {
+				"en_US": "Collection",
+				"zh_CN": "集合"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Collect",
+				"zh_CN": "集合"
+			}
 		}
 	}, {
 		"name": "deduplicate",
 		"example": "deduplicate(col, false), deduplicate(col, true)",
+		"aggregate": true,
 		"hint": {
 			"en_US": "Returns the deduplicate results in the group, usually a window. The first argument is the column as the key to deduplicate; the second argument is whether to return all items or just the latest item which is not duplicate. If the latest item is a duplicate, the sink will receive an empty map.",
 			"zh_CN": "返回当前组去重的结果，通常用在窗口中。其中，第一个参数指定用于去重的列；第二个参数指定是否返回全部结果。若为 false ，则仅返回最近的未重复的项；若最近的项有重复，则返回空数组。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "any",
+				"hint": {
+					"en_US": "The field to deduplicate.",
+					"zh_CN": "要去重的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			},
+			{
+				"name": "all",
+				"optional": false,
+				"control": "checkbox",
+				"type": "boolean",
+				"hint": {
+					"en_US": "Whether return all fields",
+					"zh_CN": "是否返回全部字段"
+				},
+				"label": {
+					"en_US": "Return all fields",
+					"zh_CN": "返回全部字段"
+				}
+			}
+		],
+		"return": {
+			"type": "array",
+			"hint": {
+				"en_US": "Deduplicate",
+				"zh_CN": "去重值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "deduplicate",
+				"zh_CN": "去重"
+			}
 		}
 	}, {
 		"name": "abs",
@@ -71,6 +309,37 @@
 		"hint": {
 			"en_US": "The absolute value of a value",
 			"zh_CN": "绝对值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "参与计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Abs",
+				"zh_CN": "绝对值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Absolute",
+				"zh_CN": "绝对值"
+			}
 		}
 	}, {
 		"name": "acos",
@@ -78,6 +347,37 @@
 		"hint": {
 			"en_US": "The inverse cosine of a number in radians",
 			"zh_CN": "弧度数的反余弦值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "参与计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "aCos",
+				"zh_CN": "反余弦值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Inverse Cosine",
+				"zh_CN": "反余弦"
+			}
 		}
 	}, {
 		"name": "asin",
@@ -85,6 +385,37 @@
 		"hint": {
 			"en_US": "The inverse sine of a number in radians",
 			"zh_CN": "弧度数的反正弦值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "参与计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "aSin",
+				"zh_CN": "反正弦值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Inverse Sine",
+				"zh_CN": "反正弦"
+			}
 		}
 	}, {
 		"name": "atan",
@@ -92,6 +423,37 @@
 		"hint": {
 			"en_US": "The inverse tangent of a number in radians",
 			"zh_CN": "弧度数的反正切值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "参与计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "aTan",
+				"zh_CN": "反正切值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Inverse Tangent",
+				"zh_CN": "反正切值"
+			}
 		}
 	}, {
 		"name": "atan2",
@@ -99,6 +461,51 @@
 		"hint": {
 			"en_US": "The angle, in radians, between the positive x-axis and the (x, y) point defined in the two arguments",
 			"zh_CN": "正x轴与两个自变量中定义的（x，y）点之间的弧度角"
+		},
+		"args": [
+			{
+				"name": "field1",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field of the first point to calculate.",
+					"zh_CN": "点X的字段名"
+				},
+				"label": {
+					"en_US": "X Field",
+					"zh_CN": "X 字段"
+				}
+			},
+			{
+				"name": "field2",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field of the second point to calculate.",
+					"zh_CN": "点Y的字段名"
+				},
+				"label": {
+					"en_US": "Y Field",
+					"zh_CN": "Y 字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Angle",
+				"zh_CN": "弧度角"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Angle",
+				"zh_CN": "弧度角"
+			}
 		}
 	}, {
 		"name": "bitand",
@@ -106,6 +513,51 @@
 		"hint": {
 			"en_US": "Performs a bitwise AND on the bit representations of the two Int(-converted) arguments",
 			"zh_CN": "对两个Int（-converted）参数的位表示执行按位“与”运算"
+		},
+		"args": [
+			{
+				"name": "field1",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The first field to calculate.",
+					"zh_CN": "参与计算的第一个字段名"
+				},
+				"label": {
+					"en_US": "Field 1",
+					"zh_CN": "字段1"
+				}
+			},
+			{
+				"name": "field2",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The second field to calculate.",
+					"zh_CN": "参与计算的第二个字段名"
+				},
+				"label": {
+					"en_US": "Field 2",
+					"zh_CN": "字段2"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Bitand",
+				"zh_CN": "按位与"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Bit And",
+				"zh_CN": "按位与"
+			}
 		}
 	}, {
 		"name": "bitor",
@@ -113,6 +565,51 @@
 		"hint": {
 			"en_US": "Performs a bitwise OR of the bit representations of the two arguments",
 			"zh_CN": "对两个参数的位表示进行或运算"
+		},
+		"args": [
+			{
+				"name": "field1",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The first field to calculate.",
+					"zh_CN": "参与计算的第一个字段名"
+				},
+				"label": {
+					"en_US": "Field 1",
+					"zh_CN": "字段1"
+				}
+			},
+			{
+				"name": "field2",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The second field to calculate.",
+					"zh_CN": "参与计算的第二个字段名"
+				},
+				"label": {
+					"en_US": "Field 2",
+					"zh_CN": "字段2"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Bitor",
+				"zh_CN": "按位或"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Bit Or",
+				"zh_CN": "按位或"
+			}
 		}
 	}, {
 		"name": "bitxor",
@@ -120,6 +617,51 @@
 		"hint": {
 			"en_US": "Performs a bitwise XOR on the bit representations of the two Int(-converted) arguments",
 			"zh_CN": "对两个Int（-converted）参数的位表示执行逐位异或运算"
+		},
+		"args": [
+			{
+				"name": "field1",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The first field to calculate.",
+					"zh_CN": "参与计算的第一个字段名"
+				},
+				"label": {
+					"en_US": "Field 1",
+					"zh_CN": "字段1"
+				}
+			},
+			{
+				"name": "field2",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The second field to calculate.",
+					"zh_CN": "参与计算的第二个字段名"
+				},
+				"label": {
+					"en_US": "Field 2",
+					"zh_CN": "字段2"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Bit Xor",
+				"zh_CN": "按位异或值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Bit Xor",
+				"zh_CN": "按位异或"
+			}
 		}
 	}, {
 		"name": "bitnot",
@@ -127,6 +669,37 @@
 		"hint": {
 			"en_US": "Performs a bitwise NOT on the bit representations of the Int(-converted) argument",
 			"zh_CN": "在Int（-converted）参数的位表示形式上执行按位NOT运算"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Bit Not",
+				"zh_CN": "按位非值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Bit Not",
+				"zh_CN": "按位非"
+			}
 		}
 	}, {
 		"name": "ceil",
@@ -134,6 +707,37 @@
 		"hint": {
 			"en_US": "Round a value up to the nearest BIGINT value.",
 			"zh_CN": "将值舍入到最接近的BIGINT值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Ceil",
+				"zh_CN": "向上取整值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Ceil",
+				"zh_CN": "向上取整"
+			}
 		}
 	}, {
 		"name": "cos",
@@ -141,6 +745,37 @@
 		"hint": {
 			"en_US": "Returns the cosine of a number in radians.",
 			"zh_CN": "返回以弧度为单位的数字的余弦值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Cosine",
+				"zh_CN": "余弦值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Cosine",
+				"zh_CN": "余弦"
+			}
 		}
 	}, {
 		"name": "cosh",
@@ -148,6 +783,37 @@
 		"hint": {
 			"en_US": "Returns the hyperbolic cosine of a number in radians.",
 			"zh_CN": "返回弧度数的双曲余弦值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Hyperbolic Cosine",
+				"zh_CN": "双曲余弦值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Hyperbolic Cosine",
+				"zh_CN": "双曲余弦"
+			}
 		}
 	}, {
 		"name": "exp",
@@ -155,6 +821,37 @@
 		"hint": {
 			"en_US": "Returns e raised to the Decimal argument.",
 			"zh_CN": "返回小数点参数的e。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Exponent",
+				"zh_CN": "幂数值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Exponent",
+				"zh_CN": "幂数"
+			}
 		}
 	}, {
 		"name": "ln",
@@ -162,13 +859,75 @@
 		"hint": {
 			"en_US": "Returns the natural logarithm of the argument.",
 			"zh_CN": "返回参数的自然对数。"
-		}
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Natural Logarithm",
+				"zh_CN": "自然对数值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+					"en_US": "Natural Logarithm",
+					"zh_CN": "自然对数"
+				}
+			}
 	}, {
 		"name": "log",
 		"example": "log(col1)",
 		"hint": {
 			"en_US": "Returns the base 10 logarithm of the argument.",
 			"zh_CN": "返回参数的以10为底的对数。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Logarithm",
+				"zh_CN": "对数值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Logarithm",
+				"zh_CN": "对数"
+			}
 		}
 	}, {
 		"name": "mod",
@@ -176,6 +935,51 @@
 		"hint": {
 			"en_US": "Returns the remainder of the division of the first argument by the second argument.",
 			"zh_CN": "返回第一个参数除以第二个参数的余数。"
+		},
+		"args": [
+			{
+				"name": "divided",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to be divided.",
+					"zh_CN": "被除数的字段名"
+				},
+				"label": {
+					"en_US": "Divided Number",
+					"zh_CN": "被除数"
+				}
+			},
+			{
+				"name": "divisor",
+				"optional": false,
+				"control": "text",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to divide.",
+					"zh_CN": "除数的字段名"
+				},
+				"label": {
+					"en_US": "Divisor",
+					"zh_CN": "除数"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Modulo",
+				"zh_CN": "取模值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Modulo",
+				"zh_CN": "取模"
+			}
 		}
 	}, {
 		"name": "power",
@@ -183,6 +987,51 @@
 		"hint": {
 			"en_US": "Pow returns x**y, the base-x exponential of y.",
 			"zh_CN": "返回 x 的 y 次方。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			},
+			{
+				"name": "power",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "The power number",
+					"zh_CN": "乘方数"
+				},
+				"label": {
+					"en_US": "Power",
+					"zh_CN": "乘方"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Power",
+				"zh_CN": "乘方"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Power",
+				"zh_CN": "乘方"
+			}
 		}
 	}, {
 		"name": "rand",
@@ -190,6 +1039,22 @@
 		"hint": {
 			"en_US": "Returns a pseudorandom, uniformly distributed double between 0.0 and 1.0.",
 			"zh_CN": "返回一个伪随机数，其均匀分布在0.0和1.0之间。"
+		},
+		"args": [],
+		"return": {
+			"type": "float",
+			"hint": {
+				"en_US": "Random Number",
+				"zh_CN": "随机数"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Random Number",
+				"zh_CN": "随机数"
+			}
 		}
 	}, {
 		"name": "round",
@@ -197,6 +1062,37 @@
 		"hint": {
 			"en_US": "Round a value to the nearest BIGINT value.",
 			"zh_CN": "将值四舍五入到最接近的 BIGINT 值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Round",
+				"zh_CN": "四舍五入值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Round",
+				"zh_CN": "四舍五入"
+			}
 		}
 	}, {
 		"name": "sign",
@@ -204,6 +1100,37 @@
 		"hint": {
 			"en_US": "Returns the sign of the given number. When the sign of the argument is positive, 1 is returned. When the sign of the argument is negative, -1 is returned. If the argument is 0, 0 is returned.",
 			"zh_CN": "返回给定数字的符号。 当参数的符号为正时，将返回1。 当参数的符号为负数时，返回-1。 如果参数为0，则返回0。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Sign",
+				"zh_CN": "正负号"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Sign",
+				"zh_CN": "正负号"
+			}
 		}
 	}, {
 		"name": "sin",
@@ -211,6 +1138,37 @@
 		"hint": {
 			"en_US": "Returns the sine of a number in radians.",
 			"zh_CN": "返回弧度数的正弦值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Sine",
+				"zh_CN": "正弦值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Sine",
+				"zh_CN": "正弦"
+			}
 		}
 	}, {
 		"name": "sinh",
@@ -218,6 +1176,37 @@
 		"hint": {
 			"en_US": "Returns the hyperbolic sine of a number in radians.",
 			"zh_CN": "返回弧度数的双曲正弦值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Hyperbolic Sine",
+				"zh_CN": "双曲正弦值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Hyperbolic Sine",
+				"zh_CN": "双曲正弦"
+			}
 		}
 	}, {
 		"name": "sqrt",
@@ -225,6 +1214,37 @@
 		"hint": {
 			"en_US": "Returns the square root of a number.",
 			"zh_CN": "返回数字的平方根。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Square Root",
+				"zh_CN": "平方根值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Square Root",
+				"zh_CN": "平方根"
+			}
 		}
 	}, {
 		"name": "tan",
@@ -232,6 +1252,37 @@
 		"hint": {
 			"en_US": "Returns the tangent of a number in radians.",
 			"zh_CN": "返回以弧度表示的数字的正切值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Tangent",
+				"zh_CN": "正切值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Tangent",
+				"zh_CN": "正切"
+			}
 		}
 	}, {
 		"name": "tanh",
@@ -239,6 +1290,37 @@
 		"hint": {
 			"en_US": "Returns the hyperbolic tangent of a number in radians.",
 			"zh_CN": "返回弧度数的双曲正切值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The field to calculate.",
+					"zh_CN": "用于计算的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Hyperbolic Tangent",
+				"zh_CN": "双曲正切值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Hyperbolic Tangent",
+				"zh_CN": "双曲正切"
+			}
 		}
 	}, {
 		"name": "concat",
@@ -246,6 +1328,37 @@
 		"hint": {
 			"en_US": "Concatenates arrays or strings. This function accepts any number of arguments and returns a String or an Array",
 			"zh_CN": "连接数组或字符串。 此函数接受任意数量的参数并返回 String 或 Array"
+		},
+		"args": [
+			{
+				"name": "stringField",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "String or Array",
+					"zh_CN": "数组或字符串"
+				},
+				"label": {
+					"en_US": "String or Array",
+					"zh_CN": "数组或字符串"
+				}
+			}
+		],
+		"return": {
+			"type": "any",
+			"hint": {
+				"en_US": "Concatenated array or string",
+				"zh_CN": "拼接完的数组或字符串"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Concatenate",
+				"zh_CN": "字符串/数组拼接"
+			}
 		}
 	}, {
 		"name": "endswith",
@@ -253,6 +1366,51 @@
 		"hint": {
 			"en_US": "Returns a Boolean indicating whether the first String argument ends with the second String argument.",
 			"zh_CN": "返回一个布尔值，该布尔值指示第一个 String参数是否以第二个 String 参数结尾。"
+		},
+		"args": [
+			{
+				"name": "string1",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "String argument 1",
+					"zh_CN": "String 参数 1"
+				},
+				"label": {
+					"en_US": "String1",
+					"zh_CN": "String1"
+				}
+			},
+			{
+				"name": "string2",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "String argument 2",
+					"zh_CN": "String 参数 2"
+				},
+				"label": {
+					"en_US": "String2",
+					"zh_CN": "String2"
+				}
+			}
+		],
+		"return": {
+			"type": "bool",
+			"hint": {
+				"en_US": "Matched",
+				"zh_CN": "是否命中"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Ends With",
+				"zh_CN": "Ends With"
+			}
 		}
 	}, {
 		"name": "format_time",
@@ -260,6 +1418,51 @@
 		"hint": {
 			"en_US": "Format a datetime to string.",
 			"zh_CN": "将日期时间格式化为字符串。"
+		},
+		"args": [
+			{
+				"name": "time",
+				"optional": false,
+				"control": "field",
+				"type": "datetime",
+				"hint": {
+					"en_US": "The datetime value",
+					"zh_CN": "时间值"
+				},
+				"label": {
+					"en_US": "Datetime Value",
+					"zh_CN": "时间值"
+				}
+			},
+			{
+				"name": "format",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "format value",
+					"zh_CN": "格式值"
+				},
+				"label": {
+					"en_US": "format value",
+					"zh_CN": "格式值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Formatted Time",
+				"zh_CN": "格式化的时间"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Format Time",
+				"zh_CN": "格式化时间"
+			}
 		}
 	}, {
 		"name": "indexof",
@@ -267,6 +1470,51 @@
 		"hint": {
 			"en_US": "Returns the first index (0-based) of the second argument as a substring in the first argument.",
 			"zh_CN": "返回第二个参数的第一个索引（从0开始），作为第一个参数中的子字符串。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "substring",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "Substring to search for",
+					"zh_CN": "子字符串"
+				},
+				"label": {
+					"en_US": "Substring to search for",
+					"zh_CN": "子字符串"
+				}
+			}
+		],
+		"return": {
+			"type": "bool",
+			"hint": {
+				"en_US": "Matched",
+				"zh_CN": "是否命中"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Index Of",
+				"zh_CN": "Index Of"
+			}
 		}
 	}, {
 		"name": "length",
@@ -274,6 +1522,37 @@
 		"hint": {
 			"en_US": "Returns the number of characters in the provided string.",
 			"zh_CN": "返回提供的字符串中的字符数。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			}
+		],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "String Length",
+				"zh_CN": "字符长度"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "String Length",
+				"zh_CN": "字符长度"
+			}
 		}
 	}, {
 		"name": "lower",
@@ -281,6 +1560,37 @@
 		"hint": {
 			"en_US": "Returns the lowercase version of the given String.",
 			"zh_CN": "返回给定 String 的小写版本。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "strng",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Lowercase",
+				"zh_CN": "小写值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Lowercase",
+				"zh_CN": "字符串小写"
+			}
 		}
 	}, {
 		"name": "lpad",
@@ -288,6 +1598,51 @@
 		"hint": {
 			"en_US": "Returns the String argument, padded on the left side with the number of spaces specified by the second argument.",
 			"zh_CN": "返回 String，在左侧用第二个参数指定的空格数填充。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "number",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "number of spaces",
+					"zh_CN": "空格数"
+				},
+				"label": {
+					"en_US": "number of spaces",
+					"zh_CN": "空格数"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Left Padding",
+				"zh_CN": "左侧填充值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Left Padding",
+				"zh_CN": "左侧填充"
+			}
 		}
 	}, {
 		"name": "ltrim",
@@ -295,6 +1650,37 @@
 		"hint": {
 			"en_US": "Removes all leading whitespace (tabs and spaces) from the provided String.",
 			"zh_CN": "从提供的字符串中删除所有前导空格（制表符和空格）。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "text",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Left Trim",
+				"zh_CN": "左侧去空格值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Left Trim",
+				"zh_CN": "左侧去空格"
+			}
 		}
 	}, {
 		"name": "numbytes",
@@ -302,6 +1688,22 @@
 		"hint": {
 			"en_US": "Returns the number of bytes in the UTF-8 encoding of the provided string.",
 			"zh_CN": "以提供的字符串的 UTF-8 编码返回字节数。"
+		},
+		"args": [],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Number of Bytes",
+				"zh_CN": "字节数"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Number of Bytes",
+				"zh_CN": "字节数"
+			}
 		}
 	}, {
 		"name": "regexp_matches",
@@ -309,6 +1711,51 @@
 		"hint": {
 			"en_US": "Returns true if the string (first argument) contains a match for the regular expression.",
 			"zh_CN": "如果字符串（第一个参数）包含正则表达式的匹配项，则返回 true。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "expression",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "regex expression",
+					"zh_CN": "正则表达式"
+				},
+				"label": {
+					"en_US": "regex expression",
+					"zh_CN": "正则表达式"
+				}
+			}
+		],
+		"return": {
+			"type": "bool",
+			"hint": {
+				"en_US": "Matched",
+				"zh_CN": "是否命中"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Regular Expression Match",
+				"zh_CN": "正则匹配"
+			}
 		}
 	}, {
 		"name": "regexp_replace",
@@ -316,6 +1763,65 @@
 		"hint": {
 			"en_US": "Replaces all occurrences of the second argument (regular expression) in the first argument with the third argument.",
 			"zh_CN": "将第一个参数中所有出现的第二个参数（正则表达式）替换为第三个参数。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "expression",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "regex expression",
+					"zh_CN": "正则表达式"
+				},
+				"label": {
+					"en_US": "regex expression",
+					"zh_CN": "正则表达式"
+				}
+			},
+			{
+				"name": "value",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "string value",
+					"zh_CN": "字符串值"
+				},
+				"label": {
+					"en_US": "string value",
+					"zh_CN": "字符串值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Regular Expression Replace",
+				"zh_CN": "正则替换值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Regular Expression Replace",
+				"zh_CN": "正则替换"
+			}
 		}
 	}, {
 		"name": "regexp_substr",
@@ -323,6 +1829,51 @@
 		"hint": {
 			"en_US": "Finds the first match of the 2nd parameter (regex) in the first parameter.",
 			"zh_CN": "在第一个参数中找到第二个参数（regex）的第一个匹配项。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "expression",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "regex expression",
+					"zh_CN": "正则表达式"
+				},
+				"label": {
+					"en_US": "regex expression",
+					"zh_CN": "正则表达式"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Regular Expression First Match",
+				"zh_CN": "正则匹配子串"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Regular Expression First Match",
+				"zh_CN": "正则匹配子串"
+			}
 		}
 	}, {
 		"name": "rpad",
@@ -330,6 +1881,51 @@
 		"hint": {
 			"en_US": "Returns the String argument, padded on the right side with the number of spaces specified by the second argument.",
 			"zh_CN": "返回字符串参数，在右侧填充第二个参数指定的空格数。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "value",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "spaces number",
+					"zh_CN": "空格数"
+				},
+				"label": {
+					"en_US": "spaces number",
+					"zh_CN": "空格数"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Right Padding",
+				"zh_CN": "右侧填充值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Right Padding",
+				"zh_CN": "右侧填充"
+			}
 		}
 	}, {
 		"name": "rtrim",
@@ -337,6 +1933,37 @@
 		"hint": {
 			"en_US": "Removes all trailing whitespace (tabs and spaces) from the provided String.",
 			"zh_CN": "从提供的字符串中删除所有尾随空白（制表符和空格）。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Right Trim",
+				"zh_CN": "右侧去空格值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Right Trim",
+				"zh_CN": "右侧去空格"
+			}
 		}
 	}, {
 		"name": "substring",
@@ -344,6 +1971,65 @@
 		"hint": {
 			"en_US": "returns the substring of the provided String from the provided Int index (0-based, inclusive) to the end of the String.",
 			"zh_CN": "从提供的 Int 索引（从0开始，包括0）到字符串的结尾，返回提供的String的子字符串。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "start",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "start",
+					"zh_CN": "start"
+				},
+				"label": {
+					"en_US": "start",
+					"zh_CN": "start"
+				}
+			},
+			{
+				"name": "end",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "end",
+					"zh_CN": "end"
+				},
+				"label": {
+					"en_US": "end",
+					"zh_CN": "end"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Substring",
+				"zh_CN": "子字符串"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Substring",
+				"zh_CN": "子字符串"
+			}
 		}
 	}, {
 		"name": "startswith",
@@ -351,13 +2037,117 @@
 		"hint": {
 			"en_US": "Returns Boolean, whether the first string argument starts with the second string argument.",
 			"zh_CN": "返回布尔值，是否第一个字符串参数是否以第二个字符串参数开头。"
+		},
+		"args": [
+			{
+				"name": "string1",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "String argument 1",
+					"zh_CN": "String 参数 1"
+				},
+				"label": {
+					"en_US": "String1",
+					"zh_CN": "String1"
+				}
+			},
+			{
+				"name": "string2",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "String argument 2",
+					"zh_CN": "String 参数 2"
+				},
+				"label": {
+					"en_US": "String2",
+					"zh_CN": "String2"
+				}
+			}
+		],
+		"return": {
+			"type": "bool",
+			"hint": {
+				"en_US": "Matched",
+				"zh_CN": "是否命中"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Starts With",
+				"zh_CN": "Starts With"
+			}
 		}
 	}, {
 		"name": "split_value",
 		"example": "split_value(col1, str_splitter, index)",
 		"hint": {
-			"en_US": "Splitthevalueofthe1stparameterwiththe2ndparameter,andreturnthevalueofsplitarraythatindexedwiththe3rdparameter.split_value(\"/test/device001/message\",\"/\",0)ASa,thereturnedvalueoffunctionisempty;split_value(\"/test/device001/message\",\"/\",3)ASa,thereturnedvalueoffunctionismessage;",
-			"zh_CN": "用第二个参数分割第一个参数的值，并返回用第三个参数索引的分割数组的值。split_value（“/test/device001/message”，“/”，0）ASa，函数的返回值为空；split_value（“/test/device001/message”，“/”，3）ASa，函数的返回值为message。"
+			"en_US": "split_value(col1, str_splitter, index)",
+			"zh_CN": "用第二个参数分割第一个参数的值，并返回用第三个参数索引的分割数组的值。split_value（“/test/device001/message”，“/”，0） AS a，函数的返回值为空；split_value（“/test/device001/message”，“/”，3） AS a，函数的返回值为message。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			},
+			{
+				"name": "str_splitter",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "str_splitter",
+					"zh_CN": "分隔符"
+				},
+				"label": {
+					"en_US": "str_splitter",
+					"zh_CN": "分隔符"
+				}
+			},
+			{
+				"name": "index",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "index",
+					"zh_CN": "位次值"
+				},
+				"label": {
+					"en_US": "index",
+					"zh_CN": "位次值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Split String",
+				"zh_CN": "字符串分割值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Split String",
+				"zh_CN": "字符串分割"
+			}
 		}
 	}, {
 		"name": "trim",
@@ -365,6 +2155,37 @@
 		"hint": {
 			"en_US": "Removes all leading and trailing whitespace (tabs and spaces) from the provided String.",
 			"zh_CN": "从提供的字符串中删除所有前导和尾随空格（制表符和空格）。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Trim",
+				"zh_CN": "去空格值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Trim",
+				"zh_CN": "去空格"
+			}
 		}
 	}, {
 		"name": "upper",
@@ -372,6 +2193,37 @@
 		"hint": {
 			"en_US": "Returns the uppercase version of the given String.",
 			"zh_CN": "返回给定 String 的大写版本。"
+		},
+		"args": [
+			{
+				"name": "string",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "A String",
+					"zh_CN": "String 值"
+				},
+				"label": {
+					"en_US": "String",
+					"zh_CN": "String 值"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Upper Case",
+				"zh_CN": "大写值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Upper Case",
+				"zh_CN": "大写"
+			}
 		}
 	}, {
 		"name": "cast",
@@ -379,6 +2231,52 @@
 		"hint": {
 			"en_US": "Converts a value from one data type to another. The supported types includes: bigint, float, string, boolean and datetime(not supported now).",
 			"zh_CN": "将值从一种数据类型转换为另一种数据类型。 支持的类型包括：bigint，float，string，boolean 和 datetime（现在不支持）。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "any",
+				"hint": {
+					"en_US": "The field to cast.",
+					"zh_CN": "需要转换的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			},
+			{
+				"name": "datatype",
+				"optional": false,
+				"control": "select",
+				"type": "string",
+				"values": ["bigint", "float", "string", "boolean", "datetime"],
+				"hint": {
+					"en_US": "The data type to cast to",
+					"zh_CN": "转换类型"
+				},
+				"label": {
+					"en_US": "Cast Type",
+					"zh_CN": "转换类型"
+				}
+			}
+		],
+		"return": {
+			"type": "any",
+			"hint": {
+				"en_US": "Casted Value",
+				"zh_CN": "类型转换值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Cast",
+				"zh_CN": "类型转换"
+			}
 		}
 	}, {
 		"name": "chr",
@@ -386,20 +2284,142 @@
 		"hint": {
 			"en_US": "Returns the ASCII character that corresponds to the given Int argument",
 			"zh_CN": "返回与给定 Int 参数对应的 ASCII 字符"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "int",
+				"hint": {
+					"en_US": "An int field or value",
+					"zh_CN": "Int 类型字段"
+				},
+				"label": {
+					"en_US": "Int Expression",
+					"zh_CN": "Int 表达式"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "ASCII Character",
+				"zh_CN": "ASCII 字符"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "ASCII Character",
+				"zh_CN": "ASCII 字符"
+			}
 		}
 	}, {
 		"name": "encode",
 		"example": "encode(col1, \"base64\")",
 		"hint": {
-			"en_US": "Use the encode function to encode the payload, which potentially might be non-JSON data, into its string representation based on the encoding scheme. Currently, only \"base64\" econding type is supported.",
+			"en_US": "Use the encode function to encode the payload, which potentially might be non-JSON data, into its string representation based on the encoding scheme. Currently, only \"base64\" encoding type is supported.",
 			"zh_CN": "使用 encode 函数根据编码方案将负载（可能是非 JSON 数据）编码为其字符串表示形式。目前，只支持\"base64\"econding 类型。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "The field to encode.",
+					"zh_CN": "需要编码的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			},
+			{
+				"name": "encodingType",
+				"optional": false,
+				"control": "select",
+				"type": "string",
+				"values": ["base64"],
+				"hint": {
+					"en_US": "The type of encoding",
+					"zh_CN": "编码方案"
+				},
+				"label": {
+					"en_US": "Encoding Type",
+					"zh_CN": "编码方案"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "Encoded Value",
+				"zh_CN": "编码值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Encode",
+				"zh_CN": "Encode"
+			}
 		}
 	}, {
 		"name": "trunc",
 		"example": "trunc(dec, int)",
 		"hint": {
-			"en_US": "iTruncates the first argument to the number of Decimal places specified by the second argument. If the second argument is less than zero, it is set to zero. If the second argument is greater than 34, it is set to 34. Trailing zeroes are stripped from the result.",
+			"en_US": "Truncates the first argument to the number of Decimal places specified by the second argument. If the second argument is less than zero, it is set to zero. If the second argument is greater than 34, it is set to 34. Trailing zeroes are stripped from the result.",
 			"zh_CN": "将第一个参数截断为第二个参数指定的小数位数。 如果第二个参数小于零，则将其设置为零。 如果第二个参数大于34，则将其设置为34。从结果中去除尾随零。"
+		},
+		"args": [
+			{
+				"name": "decimal",
+				"optional": false,
+				"control": "field",
+				"type": "number",
+				"hint": {
+					"en_US": "The decimal number to be truncated.",
+					"zh_CN": "需要截断的数字"
+				},
+				"label": {
+					"en_US": "Decimal",
+					"zh_CN": "浮点数"
+				}
+			},
+			{
+				"name": "length",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Number of decimal places",
+					"zh_CN": "小数位数"
+				},
+				"label": {
+					"en_US": "Decimal Places",
+					"zh_CN": "小数位数"
+				}
+			}
+		],
+		"return": {
+			"type": "number",
+			"hint": {
+				"en_US": "Truncated Value",
+				"zh_CN": "截断值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Truncate",
+				"zh_CN": "截断"
+			}
 		}
 	}, {
 		"name": "md5",
@@ -407,34 +2427,189 @@
 		"hint": {
 			"en_US": "Hashed value of the argument",
 			"zh_CN": "参数的哈希值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "The field to encode.",
+					"zh_CN": "需要编码的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "MD5 Value",
+				"zh_CN": "MD5 值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "MD5",
+				"zh_CN": "MD5"
+			}
 		}
 	}, {
 		"name": "sha1",
 		"example": "sha1(col1)",
 		"hint": {
-			"en_US": "Hashed value of the argument",
-			"zh_CN": "参数的哈希值"
+			"en_US": "Hashed by SHA1",
+			"zh_CN": "参数的 SHA1 哈希值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "The field to encode.",
+					"zh_CN": "需要编码的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "SHA1 Value",
+				"zh_CN": "SHA1 值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "SHA1",
+				"zh_CN": "SHA1"
+			}
 		}
 	}, {
 		"name": "sha256",
 		"example": "sha256(col1)",
 		"hint": {
-			"en_US": "Hashed value of the argument",
-			"zh_CN": "参数的哈希值"
+			"en_US": "SHA256 hashed value of the argument",
+			"zh_CN": "参数的 SHA256 哈希值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "The field to encode.",
+					"zh_CN": "需要编码的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "SHA256 Value",
+				"zh_CN": "SHA256 值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "SHA256",
+				"zh_CN": "SHA256"
+			}
 		}
 	}, {
 		"name": "sha384",
 		"example": "sha384(col1)",
 		"hint": {
-			"en_US": "Hashed value of the argument",
-			"zh_CN": "参数的哈希值"
+			"en_US": "SHA384 Hashed value of the argument",
+			"zh_CN": "参数的 SHA384 哈希值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "The field to encode.",
+					"zh_CN": "需要编码的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "SHA384 Value",
+				"zh_CN": "SHA384 值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "SHA384",
+				"zh_CN": "SHA384"
+			}
 		}
 	}, {
 		"name": "sha512",
 		"example": "sha512(col1)",
 		"hint": {
-			"en_US": "Hashed value of the argument",
-			"zh_CN": "参数的哈希值"
+			"en_US": "SHA512 hashed value of the argument",
+			"zh_CN": "参数的 SHA512 哈希值"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "The field to encode.",
+					"zh_CN": "需要编码的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "SHA512 Value",
+				"zh_CN": "SHA512 值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "SHA 512 hashed value of the argument",
+				"zh_CN": "参数的 SHA512 哈希值"
+			}
 		}
 	}, {
 		"name": "json_path_exists",
@@ -442,6 +2617,51 @@
 		"hint": {
 			"en_US": "Checks whether JSON path returns any item for the specified JSON value. Return bool value.",
 			"zh_CN": "检查 JSON 路径是否返回指定JSON 值的任何项目。 返回布尔值。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "string",
+				"hint": {
+					"en_US": "The field to apply JSON Path, could be a struct or a string.",
+					"zh_CN": "应用 JSON Path 的字段名，可为 struct 或者 string 类型。"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			},
+			{
+				"name": "expr",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "JSON path expression",
+					"zh_CN": "JSON 路径表达式"
+				},
+				"label": {
+					"en_US": "JSON Path",
+					"zh_CN": "JSON 路径"
+				}
+			}
+		],
+		"return": {
+			"type": "bool",
+			"hint": {
+				"en_US": "Existed",
+				"zh_CN": "是否存在"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "JSON Path Exist",
+				"zh_CN": "JSON Path 检查"
+			}
 		}
 	}, {
 		"name": "json_path_query",
@@ -449,6 +2669,51 @@
 		"hint": {
 			"en_US": "Gets all items returned by JSON path for the specified JSON value.",
 			"zh_CN": "获取 JSON 路径返回的指定 JSON值的所有项目。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "any",
+				"hint": {
+					"en_US": "The field to apply JSON Path, could be a struct or a string.",
+					"zh_CN": "应用 JSON Path 的字段名，可为 struct 或者 string 类型。"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			},
+			{
+				"name": "expr",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "JSON path expression",
+					"zh_CN": "JSON 路径表达式"
+				},
+				"label": {
+					"en_US": "JSON Path",
+					"zh_CN": "JSON 路径"
+				}
+			}
+		],
+		"return": {
+			"type": "array",
+			"hint": {
+				"en_US": "JSON Path Queried Value",
+				"zh_CN": "JSON Path 查询返回值"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "JSON Path Query",
+				"zh_CN": "JSON Path 查询"
+			}
 		}
 	}, {
 		"name": "json_path_query_first",
@@ -456,6 +2721,51 @@
 		"hint": {
 			"en_US": "Gets the first item returned by JSON path for the specified JSON value.",
 			"zh_CN": "获取 JSON 路径返回的指定 JSON值的第一项。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "any",
+				"hint": {
+					"en_US": "The field to apply JSON Path, could be a struct or a string.",
+					"zh_CN": "应用 JSON Path 的字段名，可为 struct 或者 string 类型。"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			},
+			{
+				"name": "expr",
+				"optional": false,
+				"control": "text",
+				"type": "string",
+				"hint": {
+					"en_US": "JSON path expression",
+					"zh_CN": "JSON 路径表达式"
+				},
+				"label": {
+					"en_US": "JSON Path",
+					"zh_CN": "JSON 路径"
+				}
+			}
+		],
+		"return": {
+			"type": "any",
+			"hint": {
+				"en_US": "JSON Path Queried First Value",
+				"zh_CN": "JSON Path 查询返回值的第一项"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "JSON Path Query First",
+				"zh_CN": "JSON Path 查询第一项"
+			}
 		}
 	}, {
 		"name": "isNull",
@@ -463,6 +2773,37 @@
 		"hint": {
 			"en_US": "Returns true if the argument is the Null value.",
 			"zh_CN": "如果参数为空值，则返回 true。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "any",
+				"hint": {
+					"en_US": "The field to check.",
+					"zh_CN": "需要检查的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "bool",
+			"hint": {
+				"en_US": "Is null",
+				"zh_CN": "是否为空"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "IsNull",
+				"zh_CN": "空值测试"
+			}
 		}
 	}, {
 		"name": "newuuid",
@@ -470,6 +2811,22 @@
 		"hint": {
 			"en_US": "Returns a random 16-byte UUID.",
 			"zh_CN": "返回一个随机的16字节 UUID。"
+		},
+		"args": [],
+		"return": {
+			"type": "string",
+			"hint": {
+				"en_US": "UUID",
+				"zh_CN": "UUID"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "UUID",
+				"zh_CN": "UUID"
+			}
 		}
 	}, {
 		"name": "tstamp",
@@ -477,6 +2834,22 @@
 		"hint": {
 			"en_US": "Returns the current timestamp in milliseconds from 00:00:00 Coordinated Universal Time (UTC), Thursday, 1 January 1970",
 			"zh_CN": "返回当前时间戳，以1970年1月1日星期四00:00:00协调世界时（UTC）为单位。"
+		},
+		"args": [],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "UTC value of current timestamp",
+				"zh_CN": "UTC 时间戳"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Current Timestamp",
+				"zh_CN": "时间戳"
+			}
 		}
 	}, {
 		"name": "mqtt",
@@ -484,6 +2857,37 @@
 		"hint": {
 			"en_US": "Returns the MQTT meta-data of specified key. The current supported keys - topic: return the topic of message. If there are multiple stream source, then specify the source name in parameter. Such as mqtt(src1.topic) - 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)",
 			"zh_CN": "返回指定键的 MQTT 元数据。 当前支持的键包括 -topic：返回消息的主题。 如果有多个流源，则在参数中指定源名称。 如 mqtt(src1.topic) - messageid：返回消息的消息ID。 如果有多个流源，则在参数中指定源名称。 如 mqtt(src2.messageid)"
+		},
+		"args": [
+			{
+				"name": "key",
+				"optional": false,
+				"control": "text",
+				"type": "any",
+				"hint": {
+					"en_US": "Returns the MQTT meta-data of specified key.",
+					"zh_CN": "返回指定键的 MQTT 元数据"
+				},
+				"label": {
+					"en_US": "Returns the MQTT meta-data of specified key.",
+					"zh_CN": "返回指定键的 MQTT 元数据"
+				}
+			}
+		],
+		"return": {
+			"type": "any",
+			"hint": {
+				"en_US": "MQTT Meta",
+				"zh_CN": "MQTT 元数据"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "MQTT Meta",
+				"zh_CN": "MQTT 元数据"
+			}
 		}
 	}, {
 		"name": "meta",
@@ -491,6 +2895,37 @@
 		"hint": {
 			"en_US": "Returns the meta-data of specified key. The key could be: - a standalone key if there is only one source in the from clause, such as meta(device) - A qualified key to specify the stream, such as meta(src1.device) - 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.",
 			"zh_CN": "返回指定键的元数据。 键可能是：-如果 from 子句中只有一个来源，则为独立键，例如meta(device) -用于指定流的合格键，例如 meta(src1.device) -用于多级元数据的带有箭头的键，例如 meta(src1.reading->device->name)。这里假定读取是地图结构元数据。"
+		},
+		"args": [
+			{
+				"name": "key",
+				"optional": false,
+				"control": "text",
+				"type": "any",
+				"hint": {
+					"en_US": "Returns the meta-data of specified key.",
+					"zh_CN": "返回指定键的元数据"
+				},
+				"label": {
+					"en_US": "Returns the meta-data of specified key.",
+					"zh_CN": "返回指定键的元数据"
+				}
+			}
+		],
+		"return": {
+			"type": "any",
+			"hint": {
+				"en_US": "MQTT Meta",
+				"zh_CN": "MQTT 元数据"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Metadata",
+				"zh_CN": "元数据"
+			}
 		}
 	}, {
 		"name": "cardinality",
@@ -498,6 +2933,37 @@
 		"hint": {
 			"en_US": "The number of members in the group. The null value is 0.",
 			"zh_CN": "组中成员的数量。空值为0。"
+		},
+		"args": [
+			{
+				"name": "field",
+				"optional": false,
+				"control": "field",
+				"type": "array",
+				"hint": {
+					"en_US": "The field to check.",
+					"zh_CN": "需要检查的字段名"
+				},
+				"label": {
+					"en_US": "Field",
+					"zh_CN": "字段"
+				}
+			}
+		],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Cardinality",
+				"zh_CN": "数组长度"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Cardinality",
+				"zh_CN": "数组长度"
+			}
 		}
 	}, {
 		"name": "window_start",
@@ -505,6 +2971,22 @@
 		"hint": {
 			"en_US": "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.",
 			"zh_CN": "返回窗口的开始时间戳，格式为 int64。若运行时没有时间窗口，则返回默认值0。窗口的时间与规则所用的时间系统相同。若规则采用处理时间，则窗口的时间也为处理时间；若规则采用事件事件，则窗口的时间也为事件时间。"
+		},
+		"args": [],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Window Start Timestamp",
+				"zh_CN": "窗口开始时间戳"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Window Start Time",
+				"zh_CN": "窗口开始时间"
+			}
 		}
 	}, {
 		"name": "window_end",
@@ -512,6 +2994,22 @@
 		"hint": {
 			"en_US": "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.",
 			"zh_CN": "返回窗口的结束时间戳，格式为 int64。若运行时没有时间窗口，则返回默认值0。窗口的时间与规则所用的时间系统相同。若规则采用处理时间，则窗口的时间也为处理时间；若规则采用事件事件，则窗口的时间也为事件时间。"
+		},
+		"args": [],
+		"return": {
+			"type": "int",
+			"hint": {
+				"en_US": "Window Start Timestamp",
+				"zh_CN": "窗口开始时间戳"
+			}
+		},
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Window End Time",
+				"zh_CN": "窗口结束时间"
+			}
 		}
 	}]
 }
diff --git a/etc/functions/windows.json b/etc/functions/windows.json
index be2e8ba6..ec45f35c 100644
--- a/etc/functions/windows.json
+++ b/etc/functions/windows.json
@@ -22,6 +22,45 @@
 		"hint": {
 			"en_US": "Tumbling window functions are used to segment a data stream into distinct time segments and perform a function against them, such as the example below. The key differentiators of a Tumbling window are that they repeat, do not overlap, and an event cannot belong to more than one tumbling window.",
 			"zh_CN": "滚动窗口函数用于将数据流分割成不同的时间段，并对其执行函数，例如下面的示例。滚动窗口的关键区别在于它们重复不重叠，并且一个事件不能属于多个翻滚窗口。"
+		},
+		"args": [
+			{
+				"name": "timeunit",
+				"optional": false,
+				"control": "select",
+				"type": "string",
+				"values": ["ms", "ss", "mi", "hh", "dd"],
+				"hint": {
+					"en_US": "The time unit to calculate window time",
+					"zh_CN": "窗口时间单位"
+				},
+				"label": {
+					"en_US": "Timeunit",
+					"zh_CN": "时间单位"
+				}
+			},
+			{
+				"name": "length",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Window length",
+					"zh_CN": "窗口长度"
+				},
+				"label": {
+					"en_US": "Window Length",
+					"zh_CN": "窗口长度"
+				}
+			}
+		],
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Tumbling Window",
+				"zh_CN": "滚动窗口"
+			}
 		}
 	}, {
 		"name": "HOPPINGWINDOW",
@@ -29,6 +68,59 @@
 		"hint": {
 			"en_US": "Hopping window functions hop forward in time by a fixed period. It may be easy to think of them as Tumbling windows that can overlap, so events can belong to more than one Hopping window result set. To make a Hopping window the same as a Tumbling window, specify the hop size to be the same as the window size.",
 			"zh_CN": "跳跃窗口功能会在时间上向前跳一段固定的时间。 将它们视为可能重叠的翻转窗口可能很容易，因此事件可以属于多个跳跃窗口结果集。 要使跳跃窗口与翻转窗口相同，请将跳跃大小指定为与窗口大小相同。"
+		},
+		"args": [
+			{
+				"name": "timeunit",
+				"optional": false,
+				"control": "select",
+				"type": "string",
+				"values": ["ms", "ss", "mi", "hh", "dd"],
+				"hint": {
+					"en_US": "The time unit to calculate window time",
+					"zh_CN": "窗口时间单位"
+				},
+				"label": {
+					"en_US": "Timeunit",
+					"zh_CN": "时间单位"
+				}
+			},
+			{
+				"name": "length",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Window length",
+					"zh_CN": "窗口长度"
+				},
+				"label": {
+					"en_US": "Window Length",
+					"zh_CN": "窗口长度"
+				}
+			},
+			{
+				"name": "hopsize",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Hop size",
+					"zh_CN": "窗口跳跃长度"
+				},
+				"label": {
+					"en_US": "Hop Size",
+					"zh_CN": "窗口跳跃长度"
+				}
+			}
+		],
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Hopping Window",
+				"zh_CN": "跳跃窗口"
+			}
 		}
 	}, {
 		"name": "SLIDINGWINDOW",
@@ -36,6 +128,45 @@
 		"hint": {
 			"en_US": "Sliding window functions, unlike Tumbling or Hopping windows, produce an output ONLY when an event occurs. Every window will have at least one event and the window continuously moves forward by an € (epsilon). Like hopping windows, events can belong to more than one sliding window.",
 			"zh_CN": "滑动窗口功能与翻转或跳动窗口不同，仅在事件发生时会产生输出。 每个窗口至少会有一个事件，并且该窗口连续向前移动€（ε）。 就像跳跃窗口一样，事件可以属于多个滑动窗口。"
+		},
+		"args": [
+			{
+				"name": "timeunit",
+				"optional": false,
+				"control": "select",
+				"type": "string",
+				"values": ["ms", "ss", "mi", "hh", "dd"],
+				"hint": {
+					"en_US": "The time unit to calculate window time",
+					"zh_CN": "窗口时间单位"
+				},
+				"label": {
+					"en_US": "Timeunit",
+					"zh_CN": "时间单位"
+				}
+			},
+			{
+				"name": "length",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Window length",
+					"zh_CN": "窗口长度"
+				},
+				"label": {
+					"en_US": "Window Length",
+					"zh_CN": "窗口长度"
+				}
+			}
+		],
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Sliding Window",
+				"zh_CN": "滑动窗口"
+			}
 		}
 	}, {
 		"name": "SESSIONWINDOW",
@@ -43,6 +174,59 @@
 		"hint": {
 			"en_US": "Session window functions group events that arrive at similar times, filtering out periods of time where there is no data. It has two main parameters: timeout and maximum duration.",
 			"zh_CN": "会话窗口功能对在相似时间到达的事件进行分组，以过滤掉没有数据的时间段。 它有两个主要参数：超时和最大持续时间。"
+		},
+		"args": [
+			{
+				"name": "timeunit",
+				"optional": false,
+				"control": "select",
+				"type": "string",
+				"values": ["ms", "ss", "mi", "hh", "dd"],
+				"hint": {
+					"en_US": "The time unit to calculate window time",
+					"zh_CN": "窗口时间单位"
+				},
+				"label": {
+					"en_US": "Timeunit",
+					"zh_CN": "时间单位"
+				}
+			},
+			{
+				"name": "length",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Window length",
+					"zh_CN": "最大窗口长度"
+				},
+				"label": {
+					"en_US": "Window Length",
+					"zh_CN": "最大窗口长度"
+				}
+			},
+			{
+				"name": "timeout",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Timeout",
+					"zh_CN": "超时时长"
+				},
+				"label": {
+					"en_US": "Timeout",
+					"zh_CN": "超时时长"
+				}
+			}
+		],
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Session Window",
+				"zh_CN": "会话窗口"
+			}
 		}
 	}, {
 		"name": "COUNTWINDOW",
@@ -50,6 +234,44 @@
 		"hint": {
 			"en_US": "Tumbling count window is similar to general tumbling window, events in a tumbling window can not repeat, do not overlap, and an event cannot belong to more than one tumbling window. Below is a count window with 5 events length.",
 			"zh_CN": "滚动计数窗口与一般的滚动窗口类似，在滚动窗口中的事件不重复、不重叠，一个事件不会属于多个滚动窗口。以下是一个长度为 5 的滚动计数窗口。"
+		},
+		"args": [
+			{
+				"name": "length",
+				"optional": false,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Window length",
+					"zh_CN": "窗口长度"
+				},
+				"label": {
+					"en_US": "Window Length",
+					"zh_CN": "窗口长度"
+				}
+			},
+			{
+				"name": "timeout",
+				"optional": true,
+				"control": "text",
+				"type": "int",
+				"hint": {
+					"en_US": "Trigger Length",
+					"zh_CN": "触发长度"
+				},
+				"label": {
+					"en_US": "Trigger Length",
+					"zh_CN": "触发长度"
+				}
+			}
+		],
+		"node": {
+			"category": "function",
+			"icon": "iconPath",
+			"label": {
+				"en_US": "Count Window",
+				"zh_CN": "计数窗口"
+			}
 		}
 	}]
 }
diff --git a/etc/mqtt_source.json b/etc/mqtt_source.json
index 7898f1bc..151dff9d 100644
--- a/etc/mqtt_source.json
+++ b/etc/mqtt_source.json
@@ -212,5 +212,22 @@
 				"zh_CN": "KubeEdge 模型文件"
 			}
 		}]
+	},
+	"outputs": [
+		{
+			"label": {
+				"en_US": "Output",
+				"zh_CN": "输出"
+			},
+			"value": "signal"
+		}
+	],
+	"node": {
+		"category": "source",
+		"icon": "iconPath",
+		"label": {
+			"en_US": "MQTT",
+			"zh_CN": "MQTT"
+		}
 	}
 }
diff --git a/etc/multilingual/en_US.ini b/etc/multilingual/en_US.ini
index 1e8568ef..69879623 100644
--- a/etc/multilingual/en_US.ini
+++ b/etc/multilingual/en_US.ini
@@ -1,6 +1,11 @@
 [sink]
 not_found_plugin=This plugin was not found:
+not_found_confkey=This confkey was not found:
+confkey_already_exist=This confkey already exists:
+json_marshal_fail=Failed to format json data:
 type_conversion_fail=Type conversion failed:
+not_found_file=Can't find this file:
+write_data_fail=Failed to write data to file:
 [source]
 not_found_plugin=This plugin was not found:
 not_found_confkey=This confkey was not found:
diff --git a/etc/multilingual/zh_CN.ini b/etc/multilingual/zh_CN.ini
index f340b67f..14337dc4 100644
--- a/etc/multilingual/zh_CN.ini
+++ b/etc/multilingual/zh_CN.ini
@@ -1,6 +1,11 @@
 [sink]
 not_found_plugin=没有找到这个插件：
+not_found_confkey=没有找到这个配置项：
+confkey_already_exist=这个配置项已经存在：
+json_marshal_fail=格式化 json 错误：
 type_conversion_fail=类型转换错误：
+not_found_file=找不到这个文件：
+write_data_fail=数据写入文件失败：
 [source]
 not_found_plugin=没有找到这个插件：
 not_found_confkey=没有找到这个配置项：
diff --git a/etc/ops/filter.json b/etc/ops/filter.json
index dc0706ee..02b4cf70 100644
--- a/etc/ops/filter.json
+++ b/etc/ops/filter.json
@@ -1,4 +1,5 @@
 {
+  "name": "filter",
   "about": {
     "trial": false,
     "author": {
@@ -16,24 +17,26 @@
       "zh_CN": "用于过滤数据流的操作"
     }
   },
-  "properties": [{
-    "name": "expr",
-    "default": "",
-    "optional": false,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "filter condition expression",
-      "zh_CN": "过滤条件语句"
-    },
-    "label": {
-      "en_US": "Condition",
-      "zh_CN": "条件"
+  "properties": [
+    {
+      "name": "expr",
+      "default": "12",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "filter condition expression",
+        "zh_CN": "过滤条件语句"
+      },
+      "label": {
+        "en_US": "Condition",
+        "zh_CN": "条件"
+      }
     }
-  }],
+  ],
   "node": {
     "display": true,
-    "category": "op",
+    "category": "operator",
     "input": {
       "type": "any",
       "rowType": "any",
diff --git a/etc/ops/function.json b/etc/ops/function.json
index e8dce08b..a692397f 100644
--- a/etc/ops/function.json
+++ b/etc/ops/function.json
@@ -1,4 +1,5 @@
 {
+  "name": "function",
   "about": {
     "trial": false,
     "author": {
@@ -33,7 +34,7 @@
   }],
   "node": {
     "display": false,
-    "category": "op",
+    "category": "operator",
     "input": {
       "type": "any",
       "rowType": "any",
diff --git a/etc/ops/groupby.json b/etc/ops/groupby.json
index d2e648d2..aca97499 100644
--- a/etc/ops/groupby.json
+++ b/etc/ops/groupby.json
@@ -1,4 +1,5 @@
 {
+  "name": "groupby",
   "about": {
     "trial": false,
     "author": {
@@ -18,7 +19,7 @@
   },
   "properties": [{
     "name": "dimensions",
-    "default": "[]",
+    "default": "",
     "optional": false,
     "control": "text",
     "type": "list_string",
@@ -33,7 +34,7 @@
   }],
   "node": {
     "display": true,
-    "category": "op",
+    "category": "operator",
     "input": {
       "type": "collection",
       "rowType": "single",
diff --git a/etc/ops/join.json b/etc/ops/join.json
index b6322642..472870b9 100644
--- a/etc/ops/join.json
+++ b/etc/ops/join.json
@@ -1,4 +1,5 @@
 {
+  "name": "join",
   "about": {
     "trial": false,
     "author": {
@@ -100,7 +101,7 @@
   }],
   "node": {
     "display": true,
-    "category": "op",
+    "category": "operator",
     "input": {
       "type": "collection",
       "rowType": "single",
diff --git a/etc/ops/orderby.json b/etc/ops/orderby.json
index 9f8d57f1..52c918c7 100644
--- a/etc/ops/orderby.json
+++ b/etc/ops/orderby.json
@@ -1,4 +1,5 @@
 {
+  "name": "orderby",
   "about": {
     "trial": false,
     "author": {
@@ -64,7 +65,7 @@
   }],
   "node": {
     "display": true,
-    "category": "op",
+    "category": "operator",
     "input": {
       "type": "any",
       "rowType": "any",
@@ -76,8 +77,8 @@
     },
     "icon": "iconPath",
     "label": {
-      "en": "Filter",
-      "zh": "过滤"
+      "en": "Sort",
+      "zh": "排序"
     }
   }
 }
\ No newline at end of file
diff --git a/etc/ops/pick.json b/etc/ops/pick.json
index 1ddd694f..e174219e 100644
--- a/etc/ops/pick.json
+++ b/etc/ops/pick.json
@@ -1,4 +1,5 @@
 {
+  "name": "pick",
   "about": {
     "trial": false,
     "author": {
@@ -20,7 +21,7 @@
     "name": "fields",
     "default": "",
     "optional": false,
-    "control": "text",
+    "control": "list",
     "type": "list_string",
     "hint": {
       "en_US": "select fields",
@@ -33,7 +34,7 @@
   }],
   "node": {
     "display": true,
-    "category": "op",
+    "category": "operator",
     "input": {
       "type": "any",
       "rowType": "any",
diff --git a/etc/ops/window.json b/etc/ops/window.json
index 1e6c246c..351adf57 100644
--- a/etc/ops/window.json
+++ b/etc/ops/window.json
@@ -1,4 +1,5 @@
 {
+  "name": "window",
   "about": {
     "trial": false,
     "author": {
@@ -21,7 +22,7 @@
       "name": "type",
       "default": "",
       "optional": false,
-      "control": "text",
+      "control": "select",
       "type": "string",
       "values": [
         "tumblingwindow",
@@ -90,7 +91,7 @@
   }],
   "node": {
     "display": false,
-    "category": "op",
+    "category": "operator",
     "input": {
       "type": "row",
       "rowType": "any",
diff --git a/etc/sinks/edgex.json b/etc/sinks/edgex.json
index 8c88fac7..ac712c5d 100644
--- a/etc/sinks/edgex.json
+++ b/etc/sinks/edgex.json
@@ -418,11 +418,11 @@
     }
   ],
   "node": {
-    "category": "Sink",
+    "category": "sink",
     "icon": "iconPath",
     "label": {
-      "en": "EDGEX output",
-      "zh": "EDGEX 输出"
+      "en": "EDGEX",
+      "zh": "EDGEX"
     }
   }
 }
diff --git a/etc/sinks/log.json b/etc/sinks/log.json
index 9a2eba17..c1c701c2 100644
--- a/etc/sinks/log.json
+++ b/etc/sinks/log.json
@@ -18,11 +18,11 @@
   },
   "properties": [],
   "node": {
-    "category": "Sink",
+    "category": "sink",
     "icon": "iconPath",
     "label": {
-      "en": "Log output",
-      "zh": "Log 输出"
+      "en": "Log",
+      "zh": "Log"
     }
   }
 }
\ No newline at end of file
diff --git a/etc/sinks/memory.json b/etc/sinks/memory.json
index d7aa6b60..90191083 100644
--- a/etc/sinks/memory.json
+++ b/etc/sinks/memory.json
@@ -34,10 +34,10 @@
     }
   ],
   "node": {
-    "category": "Sink",
+    "category": "sink",
     "icon": "iconPath",
     "label": {
-      "en": "Memory output",
+      "en": "Memory",
       "zh": "内存输出"
     }
   }
diff --git a/etc/sinks/mqtt.json b/etc/sinks/mqtt.json
index c875234a..6bca64f3 100644
--- a/etc/sinks/mqtt.json
+++ b/etc/sinks/mqtt.json
@@ -218,11 +218,11 @@
     }
   ],
   "node": {
-    "category": "Sink",
+    "category": "sink",
     "icon": "iconPath",
     "label": {
-      "en": "MQTT output",
-      "zh": "MQTT 输出"
+      "en": "MQTT",
+      "zh": "MQTT"
     }
   }
 }
diff --git a/etc/sinks/neuron.json b/etc/sinks/neuron.json
index 9dc8ef1c..9b8fb6d1 100644
--- a/etc/sinks/neuron.json
+++ b/etc/sinks/neuron.json
@@ -79,11 +79,11 @@
     }
   ],
   "node": {
-    "category": "Sink",
+    "category": "sink",
     "icon": "iconPath",
     "label": {
-      "en": "Neuron output",
-      "zh": "Neuron 输出"
+      "en": "Neuron",
+      "zh": "Neuron"
     }
   }
 }
\ No newline at end of file
diff --git a/etc/sinks/nop.json b/etc/sinks/nop.json
index f0c9d4f2..2359380b 100644
--- a/etc/sinks/nop.json
+++ b/etc/sinks/nop.json
@@ -34,11 +34,11 @@
     }
   ],
   "node": {
-    "category": "Sink",
+    "category": "sink",
     "icon": "iconPath",
     "label": {
-      "en": "Nop output",
-      "zh": "Nop 输出"
+      "en": "Nop",
+      "zh": "Nop"
     }
   }
 }
\ No newline at end of file
diff --git a/etc/sinks/rest.json b/etc/sinks/rest.json
index 04b860bd..0a197862 100644
--- a/etc/sinks/rest.json
+++ b/etc/sinks/rest.json
@@ -188,11 +188,11 @@
     }
   ],
   "node": {
-    "category": "Sink",
+    "category": "sink",
     "icon": "iconPath",
     "label": {
-      "en": "Rest output",
-      "zh": "rest 输出"
+      "en": "Rest",
+      "zh": "Rest"
     }
   }
 }
diff --git a/etc/sources/edgex.json b/etc/sources/edgex.json
index 6633ee29..cdb62c3a 100644
--- a/etc/sources/edgex.json
+++ b/etc/sources/edgex.json
@@ -350,7 +350,7 @@
 		}
 	],
 	"node": {
-		"category": "Protocol/Protocol1",
+		"category": "source",
 		"icon": "iconPath",
 		"label": {
 			"en_US": "Edgex",
diff --git a/etc/sources/file.json b/etc/sources/file.json
index d9589e7e..aa85df86 100644
--- a/etc/sources/file.json
+++ b/etc/sources/file.json
@@ -73,7 +73,7 @@
     }
   ],
   "node": {
-    "category": "Protocol/Protocol1",
+    "category": "source",
     "icon": "iconPath",
     "label": {
       "en_US": "File",
diff --git a/etc/sources/httppull.json b/etc/sources/httppull.json
index 449de1ed..82ea1d49 100644
--- a/etc/sources/httppull.json
+++ b/etc/sources/httppull.json
@@ -216,7 +216,7 @@
 		}
 	],
 	"node": {
-		"category": "Protocol/Protocol1",
+		"category": "source",
 		"icon": "iconPath",
 		"label": {
 			"en_US": "HTTP PULL",
diff --git a/etc/sources/httppush.json b/etc/sources/httppush.json
index 522b4671..7d1fdd97 100644
--- a/etc/sources/httppush.json
+++ b/etc/sources/httppush.json
@@ -43,7 +43,7 @@
 		"value": "signal"
 	}],
 	"node": {
-		"category": "Protocol/Protocol1",
+		"category": "source",
 		"icon": "iconPath",
 		"label": {
 			"en_US": "HTTP PUSH",
diff --git a/etc/sources/memory.json b/etc/sources/memory.json
index ae55c719..16811aaa 100644
--- a/etc/sources/memory.json
+++ b/etc/sources/memory.json
@@ -29,7 +29,7 @@
     }
   ],
   "node": {
-    "category": "Protocol/Protocol1",
+    "category": "source",
     "icon": "iconPath",
     "label": {
       "en_US": "Memory",
diff --git a/etc/sources/neuron.json b/etc/sources/neuron.json
index b0308146..6051dcb1 100644
--- a/etc/sources/neuron.json
+++ b/etc/sources/neuron.json
@@ -28,7 +28,7 @@
     }
   ],
   "node": {
-    "category": "Protocol/Protocol1",
+    "category": "source",
     "icon": "iconPath",
     "label": {
       "en_US": "Neuron",
diff --git a/etc/sources/redis.json b/etc/sources/redis.json
index 116052e8..b8181abe 100644
--- a/etc/sources/redis.json
+++ b/etc/sources/redis.json
@@ -86,7 +86,7 @@
     ]
   },
   "node": {
-    "category": ["Protocol","Protocol1"],
+    "category": "source",
     "icon": "iconPath",
     "label": {
       "en_US": "Redis",
diff --git a/extensions/sources/sql/sql.go b/extensions/sources/sql/sql.go
index 55cdf843..c4dad27e 100644
--- a/extensions/sources/sql/sql.go
+++ b/extensions/sources/sql/sql.go
@@ -63,26 +63,18 @@ func (m *sqlsource) Configure(_ string, props map[string]interface{}) error {
 
 	m.Query = generator
 	m.conf = cfg
-	return nil
-}
-
-func (m *sqlsource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
-	logger := ctx.GetLogger()
-	logger.Debugf("Opening sql stream %v", m.conf)
 
 	db, err := dburl.Open(m.conf.Url)
 	if err != nil {
-		logger.Errorf("connection to %s Open with error %v", m.conf.Url, err)
-		logger.Errorf("support build tags are %v", driver2.KnownBuildTags())
-		errCh <- err
-		return
+		return fmt.Errorf("connection to %s Open with error %v, support build tags are %v", m.conf.Url, err, driver2.KnownBuildTags())
 	}
 	m.db = db
 
-	defer func() {
-		_ = m.db.Close()
-	}()
+	return nil
+}
 
+func (m *sqlsource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
+	logger := ctx.GetLogger()
 	t := time.NewTicker(time.Duration(m.conf.Interval) * time.Millisecond)
 	defer t.Stop()
 	for {
@@ -142,6 +134,9 @@ func (m *sqlsource) Rewind(offset interface{}) error {
 func (m *sqlsource) Close(ctx api.StreamContext) error {
 	logger := ctx.GetLogger()
 	logger.Debugf("Closing sql stream to %v", m.conf)
+	if m.db != nil {
+		_ = m.db.Close()
+	}
 
 	return nil
 }
diff --git a/extensions/sources/zmq/zmq.go b/extensions/sources/zmq/zmq.go
index 9b3773f3..91b5720c 100644
--- a/extensions/sources/zmq/zmq.go
+++ b/extensions/sources/zmq/zmq.go
@@ -35,20 +35,22 @@ func (s *zmqSource) Configure(topic string, props map[string]interface{}) error
 		return fmt.Errorf("zmq source is missing property server")
 	}
 	s.srv = srv.(string)
-	return nil
-}
 
-func (s *zmqSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
-	logger := ctx.GetLogger()
 	var err error
 	s.subscriber, err = zmq.NewSocket(zmq.SUB)
 	if err != nil {
-		errCh <- fmt.Errorf("zmq source fails to create socket: %v", err)
+		return fmt.Errorf("zmq source fails to create socket: %v", err)
 	}
 	err = s.subscriber.Connect(s.srv)
 	if err != nil {
-		errCh <- fmt.Errorf("zmq source fails to connect to %s: %v", s.srv, err)
+		return fmt.Errorf("zmq source fails to connect to %s: %v", s.srv, err)
 	}
+
+	return nil
+}
+
+func (s *zmqSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
+	logger := ctx.GetLogger()
 	s.subscriber.SetSubscribe(s.topic)
 	logger.Debugf("zmq source subscribe to topic %s", s.topic)
 	exeCtx, cancel := ctx.WithCancel()
diff --git a/internal/binder/meta/bind.go b/internal/binder/meta/bind.go
index 974398e3..3c298904 100644
--- a/internal/binder/meta/bind.go
+++ b/internal/binder/meta/bind.go
@@ -15,7 +15,6 @@
 package meta
 
 import (
-	"github.com/lf-edge/ekuiper/internal/binder/function"
 	"github.com/lf-edge/ekuiper/internal/binder/io"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/meta"
@@ -35,11 +34,6 @@ func Bind() {
 	}); nil != err {
 		conf.Log.Errorf("readSinkMetaDir:%v", err)
 	}
-	if err := meta.ReadFuncMetaDir(func(name string) bool {
-		return function.HasFunctionSet(name)
-	}); nil != err {
-		conf.Log.Errorf("readFuncMetaDir:%v", err)
-	}
 	if err := meta.ReadUiMsgDir(); nil != err {
 		conf.Log.Errorf("readUiMsgDir:%v", err)
 	}
diff --git a/internal/conf/yaml_config_ops.go b/internal/conf/yaml_config_ops.go
index d262c738..51a92baa 100644
--- a/internal/conf/yaml_config_ops.go
+++ b/internal/conf/yaml_config_ops.go
@@ -97,6 +97,7 @@ func (c *ConfigKeys) CopyConfContent() map[string]map[string]interface{} {
 		cf[key] = aux
 	}
 
+	//note: config keys in data directory will overwrite those in etc directory with same name
 	for key, kvs := range c.dataCfg {
 		aux := make(map[string]interface{})
 		for k, v := range kvs {
@@ -230,10 +231,6 @@ func (c *ConfigKeys) AddConfKey(confKey string, reqField map[string]interface{})
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	if _, ok := c.etcCfg[confKey]; ok {
-		return fmt.Errorf("duplicate key %s, already exist in etc folder", confKey)
-	}
-
 	c.dataCfg[confKey] = reqField
 
 	return nil
@@ -279,6 +276,32 @@ func (c *SourceConfigKeysOps) SaveCfgToFile() error {
 	return nil
 }
 
+// SinkConfigKeysOps implement ConfOperator interface, load the configs from data/sinks/xx.yaml
+type SinkConfigKeysOps struct {
+	*ConfigKeys
+}
+
+func (c *SinkConfigKeysOps) IsSource() bool {
+	return false
+}
+
+func (c *SinkConfigKeysOps) SaveCfgToFile() error {
+	pluginName := c.pluginName
+	confDir, err := GetDataLoc()
+	if nil != err {
+		return err
+	}
+
+	dir := path.Join(confDir, "sinks")
+	filePath := path.Join(dir, pluginName+".yaml")
+	cfg := c.CopyUpdatableConfContent()
+	err = filex.WriteYamlMarshal(filePath, cfg)
+	if nil != err {
+		return err
+	}
+	return nil
+}
+
 // ConnectionConfigKeysOps implement ConfOperator interface, load the configs from et/connections/connection.yaml
 type ConnectionConfigKeysOps struct {
 	*ConfigKeys
@@ -361,6 +384,42 @@ func NewConfigOperatorFromSourceYaml(pluginName string) (ConfigOperator, error)
 	return c, nil
 }
 
+// NewConfigOperatorForSink construct function
+func NewConfigOperatorForSink(pluginName string) ConfigOperator {
+	c := &SinkConfigKeysOps{
+		&ConfigKeys{
+			lock:       sync.RWMutex{},
+			pluginName: pluginName,
+			etcCfg:     map[string]map[string]interface{}{},
+			dataCfg:    map[string]map[string]interface{}{},
+		},
+	}
+	return c
+}
+
+// NewConfigOperatorFromSinkYaml construct function, Load the configs from etc/sources/xx.yaml
+func NewConfigOperatorFromSinkYaml(pluginName string) (ConfigOperator, error) {
+	c := &SinkConfigKeysOps{
+		&ConfigKeys{
+			lock:       sync.RWMutex{},
+			pluginName: pluginName,
+			etcCfg:     map[string]map[string]interface{}{},
+			dataCfg:    map[string]map[string]interface{}{},
+		},
+	}
+
+	dataDir, err := GetDataLoc()
+	if nil != err {
+		return nil, err
+	}
+	dir := path.Join(dataDir, "sinks")
+
+	filePath := path.Join(dir, pluginName+`.yaml`)
+	_ = filex.ReadYamlUnmarshal(filePath, &c.dataCfg)
+
+	return c, nil
+}
+
 // NewConfigOperatorForConnection construct function
 func NewConfigOperatorForConnection(pluginName string) ConfigOperator {
 	c := &ConnectionConfigKeysOps{
@@ -432,5 +491,6 @@ func NewConfigOperatorFromConnectionYaml(pluginName string) (ConfigOperator, err
 			return nil, fmt.Errorf("file content is not right: %v", plgCnfs)
 		}
 	}
+
 	return c, nil
 }
diff --git a/internal/meta/func_meta.go b/internal/meta/func_meta.go
index 52adc617..f80c41f3 100644
--- a/internal/meta/func_meta.go
+++ b/internal/meta/func_meta.go
@@ -15,147 +15,63 @@
 package meta
 
 import (
-	"fmt"
+	"bytes"
 	"os"
 	"path"
 	"strings"
 
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 )
 
-type (
-	fileFunc struct {
-		Name       string        `json:"name"`
-		Example    string        `json:"example"`
-		Hint       *fileLanguage `json:"hint"`
-		Aggregate  bool          `json:"aggregate"`
-		ArgsFields []*fileField  `json:"args"`
-		Node       *fileNode     `json:"node"`
-		Outputs    []interface{} `json:"outputs"`
-	}
-	fileFuncs struct {
-		About   *fileAbout  `json:"about"`
-		Name    string      `json:"name"`
-		Version string      `json:"version"`
-		FiFuncs []*fileFunc `json:"functions"`
-	}
-	uiFunc struct {
-		Name       string        `json:"name"`
-		Example    string        `json:"example"`
-		Hint       *language     `json:"hint"`
-		Aggregate  bool          `json:"aggregate"`
-		ArgsFields []*fileField  `json:"args"`
-		Node       *node         `json:"node"`
-		Outputs    []interface{} `json:"outputs"`
-	}
-	uiFuncs struct {
-		About   *about    `json:"about"`
-		Name    string    `json:"name"`
-		Version string    `json:"version"`
-		UiFuncs []*uiFunc `json:"functions"`
-	}
-)
-
-func newUiFuncs(fi *fileFuncs) *uiFuncs {
-	if nil == fi {
-		return nil
-	}
-	uis := new(uiFuncs)
-	uis.About = newAbout(fi.About)
-	uis.Name = fi.Name
-	for _, v := range fi.FiFuncs {
-		ui := new(uiFunc)
-		ui.Name = v.Name
-		ui.Example = v.Example
-		ui.Hint = newLanguage(v.Hint)
-		ui.Aggregate = v.Aggregate
-		ui.ArgsFields = v.ArgsFields
-		ui.Node = newNode(v.Node)
-		ui.Outputs = make([]interface{}, len(v.Outputs))
-		for k, field := range v.Outputs {
-			ui.Outputs[k] = field
-		}
-		uis.UiFuncs = append(uis.UiFuncs, ui)
-	}
-	return uis
-}
+func readFuncMetaDir() []fileContent {
+	var filesByte []fileContent
 
-var gFuncmetadata = make(map[string]*uiFuncs)
-
-func ReadFuncMetaDir(checker InstallChecker) error {
 	confDir, err := conf.GetConfLoc()
 	if nil != err {
-		return err
+		return nil
 	}
 
 	dir := path.Join(confDir, "functions")
 	files, err := os.ReadDir(dir)
 	if nil != err {
-		return err
+		return nil
 	}
 	for _, file := range files {
 		fname := file.Name()
 		if !strings.HasSuffix(fname, ".json") {
 			continue
 		}
-
-		if err := ReadFuncMetaFile(path.Join(dir, fname), checker(strings.TrimSuffix(fname, ".json"))); nil != err {
-			return err
-		}
+		filesByte = append(filesByte, readFuncMetaFile(path.Join(dir, fname)))
 	}
 
 	confDir, err = conf.GetDataLoc()
 	if nil != err {
-		return err
+		return nil
 	}
 
 	dir = path.Join(confDir, "functions")
 	files, err = os.ReadDir(dir)
 	if nil != err {
-		return err
+		return nil
 	}
 	for _, file := range files {
 		fname := file.Name()
 		if !strings.HasSuffix(fname, ".json") {
 			continue
 		}
-
-		if err := ReadFuncMetaFile(path.Join(dir, fname), checker(strings.TrimSuffix(fname, ".json"))); nil != err {
-			return err
-		}
-	}
-	return nil
-}
-
-func UninstallFunc(name string) {
-	if ui, ok := gFuncmetadata[name+".json"]; ok {
-		if nil != ui.About {
-			ui.About.Installed = false
-		}
+		filesByte = append(filesByte, readFuncMetaFile(path.Join(dir, fname)))
 	}
+	return filesByte
 }
 
-func ReadFuncMetaFile(filePath string, installed bool) error {
+func readFuncMetaFile(filePath string) fileContent {
 	fiName := path.Base(filePath)
-	fis := new(fileFuncs)
-	err := filex.ReadJsonUnmarshal(filePath, fis)
-	if nil != err {
-		return fmt.Errorf("filePath:%s err:%v", filePath, err)
-	}
-	if nil == fis.About {
-		return fmt.Errorf("not found about of %s", filePath)
-	} else {
-		fis.About.Installed = installed
-	}
-	gFuncmetadata[fiName] = newUiFuncs(fis)
+	sliByte, _ := os.ReadFile(filePath)
 	conf.Log.Infof("funcMeta file : %s", fiName)
-	return nil
+	return sliByte
 }
 
-func GetFunctions() (ret []*uiFuncs) {
-	for _, v := range gFuncmetadata {
-		ret = append(ret, v)
-	}
-	return ret
+func GetFunctions() bytes.Buffer {
+	files := readFuncMetaDir()
+	return ConstructJsonArray(files)
 }
diff --git a/internal/meta/msgUtil.go b/internal/meta/msgUtil.go
index b3d6dff0..34e7fc2d 100644
--- a/internal/meta/msgUtil.go
+++ b/internal/meta/msgUtil.go
@@ -15,6 +15,7 @@
 package meta
 
 import (
+	"bytes"
 	"os"
 	"path"
 
@@ -59,3 +60,23 @@ func ReadUiMsgDir() error {
 	}
 	return nil
 }
+
+func ConstructJsonArray(jsonByteItems []fileContent) bytes.Buffer {
+	var buf bytes.Buffer
+	var length = len(jsonByteItems)
+	if length == 0 {
+		buf.Write([]byte("[]"))
+		return buf
+	}
+
+	buf.Write([]byte("["))
+	buf.Write(jsonByteItems[0])
+
+	for i := 1; i < length; i++ {
+		buf.Write([]byte(","))
+		buf.Write(jsonByteItems[i])
+	}
+
+	buf.Write([]byte("]"))
+	return buf
+}
diff --git a/internal/meta/operator_meta.go b/internal/meta/operator_meta.go
new file mode 100644
index 00000000..54201538
--- /dev/null
+++ b/internal/meta/operator_meta.go
@@ -0,0 +1,64 @@
+// 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.
+
+package meta
+
+import (
+	"bytes"
+	"os"
+	"path"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/internal/conf"
+)
+
+type fileContent []byte
+
+func readOpsMetaDir() ([]fileContent, error) {
+	var filesByte []fileContent
+
+	confDir, err := conf.GetConfLoc()
+	if nil != err {
+		return nil, err
+	}
+
+	dir := path.Join(confDir, "ops")
+	files, err := os.ReadDir(dir)
+	if nil != err {
+		return nil, err
+	}
+	for _, file := range files {
+		fname := file.Name()
+		if !strings.HasSuffix(fname, ".json") {
+			continue
+		}
+
+		filesByte = append(filesByte, readOpsMetaFile(path.Join(dir, fname)))
+
+	}
+
+	return filesByte, nil
+}
+
+func readOpsMetaFile(filePath string) fileContent {
+	fiName := path.Base(filePath)
+	sliByte, _ := os.ReadFile(filePath)
+	conf.Log.Infof("operatorMeta file : %s", fiName)
+	return sliByte
+}
+
+func GetOperators() bytes.Buffer {
+	files, _ := readOpsMetaDir()
+	return ConstructJsonArray(files)
+}
diff --git a/internal/meta/sinkMeta.go b/internal/meta/sinkMeta.go
index 0a33c49b..38700b04 100644
--- a/internal/meta/sinkMeta.go
+++ b/internal/meta/sinkMeta.go
@@ -59,16 +59,16 @@ type (
 		HelpUrl     *fileLanguage `json:"helpUrl"`
 		Description *fileLanguage `json:"description"`
 	}
-	fileNode struct {
-		Category string        `json:"category"`
-		Icon     string        `json:"iconPath"`
-		Label    *fileLanguage `json:"label"`
-	}
+	//fileNode struct {
+	//	Category string        `json:"category"`
+	//	Icon     string        `json:"iconPath"`
+	//	Label    *fileLanguage `json:"label"`
+	//}
 	fileSink struct {
 		About  *fileAbout   `json:"about"`
 		Libs   []string     `json:"libs"`
 		Fields []*fileField `json:"properties"`
-		Node   *fileNode    `json:"node"`
+		Node   interface{}  `json:"node"`
 	}
 	language struct {
 		English string `json:"en"`
@@ -99,10 +99,10 @@ type (
 		Label    *language `json:"label"`
 	}
 	uiSink struct {
-		About  *about   `json:"about"`
-		Libs   []string `json:"libs"`
-		Fields []field  `json:"properties"`
-		Node   *node    `json:"node"`
+		About  *about      `json:"about"`
+		Libs   []string    `json:"libs"`
+		Fields []field     `json:"properties"`
+		Node   interface{} `json:"node"`
 	}
 )
 
@@ -158,16 +158,17 @@ func newAbout(fi *fileAbout) *about {
 	ui.Description = newLanguage(fi.Description)
 	return ui
 }
-func newNode(fi *fileNode) *node {
-	if nil == fi {
-		return nil
-	}
-	ui := new(node)
-	ui.Category = fi.Category
-	ui.Icon = fi.Icon
-	ui.Label = newLanguage(fi.Label)
-	return ui
-}
+
+//func newNode(fi *fileNode) *node {
+//	if nil == fi {
+//		return nil
+//	}
+//	ui := new(node)
+//	ui.Category = fi.Category
+//	ui.Icon = fi.Icon
+//	ui.Label = newLanguage(fi.Label)
+//	return ui
+//}
 func newUiSink(fi *fileSink) (*uiSink, error) {
 	if nil == fi {
 		return nil, nil
@@ -175,8 +176,8 @@ func newUiSink(fi *fileSink) (*uiSink, error) {
 	var err error
 	ui := new(uiSink)
 	ui.Libs = fi.Libs
+	ui.Node = fi.Node
 	ui.About = newAbout(fi.About)
-	ui.Node = newNode(fi.Node)
 	ui.Fields, err = newField(fi.Fields)
 	return ui, err
 }
@@ -253,6 +254,7 @@ func ReadSinkMetaFile(filePath string, installed bool) error {
 	if nil != err {
 		return err
 	}
+	loadConfigOperatorForSink(strings.TrimSuffix(finame, `.json`))
 	conf.Log.Infof("Loading metadata file for sink: %s", finame)
 	return nil
 }
diff --git a/internal/meta/sourceMeta.go b/internal/meta/sourceMeta.go
index 4bf4de7c..a8c2ced7 100644
--- a/internal/meta/sourceMeta.go
+++ b/internal/meta/sourceMeta.go
@@ -30,15 +30,13 @@ type (
 		About    *fileAbout              `json:"about"`
 		Libs     []string                `json:"libs"`
 		ConfKeys map[string][]*fileField `json:"properties"`
-		Node     *fileNode               `json:"node"`
-		Outputs  []interface{}           `json:"outputs"`
+		Node     interface{}             `json:"node"`
 	}
 	uiSource struct {
 		About    *about             `json:"about"`
 		Libs     []string           `json:"libs"`
 		ConfKeys map[string][]field `json:"properties"`
-		Node     *node              `json:"node"`
-		Outputs  []interface{}      `json:"outputs"`
+		Node     interface{}        `json:"node"`
 	}
 )
 
@@ -50,11 +48,7 @@ func newUiSource(fi *fileSource) (*uiSource, error) {
 	ui := new(uiSource)
 	ui.Libs = fi.Libs
 	ui.About = newAbout(fi.About)
-	ui.Node = newNode(fi.Node)
-	ui.Outputs = make([]interface{}, len(fi.Outputs))
-	for k, field := range fi.Outputs {
-		ui.Outputs[k] = field
-	}
+	ui.Node = fi.Node
 	ui.ConfKeys = make(map[string][]field)
 	for k, fields := range fi.ConfKeys {
 		if ui.ConfKeys[k], err = newField(fields); nil != err {
diff --git a/internal/meta/yamlConfigMeta.go b/internal/meta/yamlConfigMeta.go
index 3f53e021..d8779eb6 100644
--- a/internal/meta/yamlConfigMeta.go
+++ b/internal/meta/yamlConfigMeta.go
@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"strings"
 	"sync"
 )
 
@@ -35,11 +36,15 @@ var ConfigManager = configManager{
 }
 
 const SourceCfgOperatorKeyTemplate = "sources.%s"
+const SourceCfgOperatorKeyPrefix = "sources."
+const SinkCfgOperatorKeyTemplate = "sinks.%s"
+const SinkCfgOperatorKeyPrefix = "sinks."
 const ConnectionCfgOperatorKeyTemplate = "connections.%s"
+const ConnectionCfgOperatorKeyPrefix = "connections."
 
 // loadConfigOperatorForSource
-// Try to load ConfigOperator for plugin xxx from /etc/sources/xxx.yaml
-// If plugin xxx not exist in /etc/sources/xxx.yaml, no error response
+// Try to load ConfigOperator for plugin xxx from /etc/sources/xxx.yaml  /data/sources/xxx.yaml
+// If plugin xxx not exist, no error response
 func loadConfigOperatorForSource(pluginName string) {
 	yamlKey := fmt.Sprintf(SourceCfgOperatorKeyTemplate, pluginName)
 
@@ -51,8 +56,22 @@ func loadConfigOperatorForSource(pluginName string) {
 	}
 }
 
+// loadConfigOperatorForSink
+// Try to load ConfigOperator for plugin xxx from /data/sinks/xxx.yaml
+// If plugin xxx not exist, no error response
+func loadConfigOperatorForSink(pluginName string) {
+	yamlKey := fmt.Sprintf(SinkCfgOperatorKeyTemplate, pluginName)
+
+	if cfg, _ := conf.NewConfigOperatorFromSinkYaml(pluginName); cfg != nil {
+		ConfigManager.lock.Lock()
+		ConfigManager.cfgOperators[yamlKey] = cfg
+		ConfigManager.lock.Unlock()
+		conf.Log.Infof("Loading yaml file for sink: %s", pluginName)
+	}
+}
+
 // loadConfigOperatorForConnection
-// Try to load ConfigOperator for plugin from /etc/connections/connection.yaml
+// Try to load ConfigOperator for plugin from /etc/connections/connection.yaml /data/connections/connection.yaml
 // If plugin not exist in /etc/connections/connection.yaml, no error response
 func loadConfigOperatorForConnection(pluginName string) {
 	yamlKey := fmt.Sprintf(ConnectionCfgOperatorKeyTemplate, pluginName)
@@ -65,22 +84,22 @@ func loadConfigOperatorForConnection(pluginName string) {
 	}
 }
 
-func GetYamlConf(configOperatorKey, language string) (b []byte, err error) {
-
-	ConfigManager.lock.RLock()
-	defer ConfigManager.lock.RUnlock()
+func delConfKey(configOperatorKey, confKey, language string) error {
+	ConfigManager.lock.Lock()
+	defer ConfigManager.lock.Unlock()
 
 	cfgOps, ok := ConfigManager.cfgOperators[configOperatorKey]
 	if !ok {
-		return nil, fmt.Errorf(`%s%s`, getMsg(language, source, "not_found_plugin"), configOperatorKey)
+		return fmt.Errorf(`%s%s`, getMsg(language, source, "not_found_plugin"), configOperatorKey)
 	}
 
-	cf := cfgOps.CopyConfContent()
-	if b, err = json.Marshal(cf); nil != err {
-		return nil, fmt.Errorf(`%s%v`, getMsg(language, source, "json_marshal_fail"), cf)
-	} else {
-		return b, err
+	cfgOps.DeleteConfKey(confKey)
+
+	err := cfgOps.SaveCfgToFile()
+	if err != nil {
+		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "write_data_fail"), configOperatorKey, err)
 	}
+	return nil
 }
 
 func DelSourceConfKey(plgName, confKey, language string) error {
@@ -88,27 +107,32 @@ func DelSourceConfKey(plgName, confKey, language string) error {
 	return delConfKey(configOperatorKey, confKey, language)
 }
 
+func DelSinkConfKey(plgName, confKey, language string) error {
+	configOperatorKey := fmt.Sprintf(SinkCfgOperatorKeyTemplate, plgName)
+	return delConfKey(configOperatorKey, confKey, language)
+}
+
 func DelConnectionConfKey(plgName, confKey, language string) error {
 	configOperatorKey := fmt.Sprintf(ConnectionCfgOperatorKeyTemplate, plgName)
 	return delConfKey(configOperatorKey, confKey, language)
 }
 
-func delConfKey(configOperatorKey, confKey, language string) error {
-	ConfigManager.lock.Lock()
-	defer ConfigManager.lock.Unlock()
+func GetYamlConf(configOperatorKey, language string) (b []byte, err error) {
+
+	ConfigManager.lock.RLock()
+	defer ConfigManager.lock.RUnlock()
 
 	cfgOps, ok := ConfigManager.cfgOperators[configOperatorKey]
 	if !ok {
-		return fmt.Errorf(`%s%s`, getMsg(language, source, "not_found_plugin"), configOperatorKey)
+		return nil, fmt.Errorf(`%s%s`, getMsg(language, source, "not_found_plugin"), configOperatorKey)
 	}
 
-	cfgOps.DeleteConfKey(confKey)
-
-	err := cfgOps.SaveCfgToFile()
-	if err != nil {
-		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "write_data_fail"), configOperatorKey, err)
+	cf := cfgOps.CopyConfContent()
+	if b, err = json.Marshal(cf); nil != err {
+		return nil, fmt.Errorf(`%s%v`, getMsg(language, source, "json_marshal_fail"), cf)
+	} else {
+		return b, err
 	}
-	return nil
 }
 
 func AddSourceConfKey(plgName, confKey, language string, content []byte) error {
@@ -143,6 +167,38 @@ func AddSourceConfKey(plgName, confKey, language string, content []byte) error {
 	return nil
 }
 
+func AddSinkConfKey(plgName, confKey, language string, content []byte) error {
+	ConfigManager.lock.Lock()
+	defer ConfigManager.lock.Unlock()
+
+	configOperatorKey := fmt.Sprintf(SinkCfgOperatorKeyTemplate, plgName)
+
+	reqField := make(map[string]interface{})
+	err := json.Unmarshal(content, &reqField)
+	if nil != err {
+		return fmt.Errorf(`%s%s.%v`, getMsg(language, sink, "type_conversion_fail"), plgName, err)
+	}
+
+	var cfgOps conf.ConfigOperator
+	var found bool
+
+	cfgOps, found = ConfigManager.cfgOperators[configOperatorKey]
+	if !found {
+		cfgOps = conf.NewConfigOperatorForSink(plgName)
+		ConfigManager.cfgOperators[configOperatorKey] = cfgOps
+	}
+
+	if err := cfgOps.AddConfKey(confKey, reqField); err != nil {
+		return err
+	}
+
+	err = cfgOps.SaveCfgToFile()
+	if err != nil {
+		return fmt.Errorf(`%s%s.%v`, getMsg(language, sink, "write_data_fail"), configOperatorKey, err)
+	}
+	return nil
+}
+
 func AddConnectionConfKey(plgName, confKey, language string, content []byte) error {
 	ConfigManager.lock.Lock()
 	defer ConfigManager.lock.Unlock()
@@ -174,3 +230,50 @@ func AddConnectionConfKey(plgName, confKey, language string, content []byte) err
 	}
 	return nil
 }
+
+func GetResources(language string) (b []byte, err error) {
+	ConfigManager.lock.RLock()
+	defer ConfigManager.lock.RUnlock()
+	var srcResources []map[string]string
+	var sinkResources []map[string]string
+
+	for key, ops := range ConfigManager.cfgOperators {
+		if strings.HasPrefix(key, ConnectionCfgOperatorKeyPrefix) {
+			continue
+		}
+		if strings.HasPrefix(key, SourceCfgOperatorKeyPrefix) {
+			plugin := strings.TrimPrefix(key, SourceCfgOperatorKeyPrefix)
+			resourceIds := ops.GetUpdatableConfKeys()
+			if len(resourceIds) > 0 {
+				item := map[string]string{}
+				for _, v := range resourceIds {
+					item[v] = plugin
+				}
+				srcResources = append(srcResources, item)
+			}
+			continue
+		}
+		if strings.HasPrefix(key, SinkCfgOperatorKeyPrefix) {
+			plugin := strings.TrimPrefix(key, SinkCfgOperatorKeyPrefix)
+			resourceIds := ops.GetUpdatableConfKeys()
+			if len(resourceIds) > 0 {
+				item := map[string]string{}
+				for _, v := range resourceIds {
+					item[v] = plugin
+				}
+				sinkResources = append(sinkResources, item)
+			}
+			continue
+		}
+	}
+
+	result := map[string]interface{}{}
+	result["sources"] = srcResources
+	result["sinks"] = sinkResources
+
+	if b, err = json.Marshal(result); nil != err {
+		return nil, fmt.Errorf(`%s%v`, getMsg(language, source, "json_marshal_fail"), result)
+	} else {
+		return b, err
+	}
+}
diff --git a/internal/plugin/native/manager.go b/internal/plugin/native/manager.go
index 3eb4139e..7ccc7e20 100644
--- a/internal/plugin/native/manager.go
+++ b/internal/plugin/native/manager.go
@@ -289,10 +289,6 @@ func (rr *Manager) Register(t plugin2.PluginType, j plugin2.Plugin) error {
 		if err := meta.ReadSourceMetaFile(path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], name+`.json`), true); nil != err {
 			conf.Log.Errorf("readSourceFile:%v", err)
 		}
-	case plugin2.FUNCTION:
-		if err := meta.ReadFuncMetaFile(path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], name+`.json`), true); nil != err {
-			conf.Log.Errorf("readFuncFile:%v", err)
-		}
 	}
 	return nil
 }
@@ -362,7 +358,6 @@ func (rr *Manager) Delete(t plugin2.PluginType, name string, stop bool) error {
 		} else if !ok {
 			rr.removeSymbols([]string{name})
 		}
-		meta.UninstallFunc(name)
 	}
 
 	for _, p := range paths {
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index 01c66721..4eb15d4a 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.go
@@ -142,11 +142,6 @@ func (m *Manager) doRegister(name string, pi *PluginInfo, isInit bool) error {
 				conf.Log.Errorf("read sink json file:%v", err)
 			}
 		}
-		for _, s := range pi.Functions {
-			if err := meta.ReadFuncMetaFile(path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.FUNCTION], s+`.json`), true); nil != err {
-				conf.Log.Errorf("read function json file:%v", err)
-			}
-		}
 	}
 
 	conf.Log.Infof("Installed portable plugin %s successfully", name)
@@ -368,7 +363,6 @@ func (m *Manager) Delete(name string) error {
 	for _, s := range pinfo.Functions {
 		p := path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.FUNCTION], s+".json")
 		os.Remove(p)
-		meta.UninstallFunc(s)
 	}
 	_ = os.RemoveAll(path.Join(m.pluginDir, name))
 	return nil
diff --git a/internal/server/meta_init.go b/internal/server/meta_init.go
index ae400480..54fddfba 100644
--- a/internal/server/meta_init.go
+++ b/internal/server/meta_init.go
@@ -18,7 +18,9 @@
 package server
 
 import (
+	"encoding/json"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/topo/node"
 	"io"
 	"net/http"
 	"strings"
@@ -42,18 +44,24 @@ func (m metaComp) register() {
 
 func (m metaComp) rest(r *mux.Router) {
 	r.HandleFunc("/metadata/functions", functionsMetaHandler).Methods(http.MethodGet)
+	r.HandleFunc("/metadata/operators", operatorsMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sinks", sinksMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sinks/{name}", newSinkMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sources", sourcesMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sources/{name}", sourceMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sources/yaml/{name}", sourceConfHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/sources/{name}/confKeys/{confKey}", sourceConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
+	r.HandleFunc("/metadata/sinks/yaml/{name}", sinkConfHandler).Methods(http.MethodGet)
+	r.HandleFunc("/metadata/sinks/{name}/confKeys/{confKey}", sinkConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 
 	r.HandleFunc("/metadata/connections", connectionsMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/connections/{name}", connectionMetaHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/connections/yaml/{name}", connectionConfHandler).Methods(http.MethodGet)
 	r.HandleFunc("/metadata/connections/{name}/confKeys/{confKey}", connectionConfKeyHandler).Methods(http.MethodDelete, http.MethodPut)
 
+	r.HandleFunc("/metadata/resources", resourcesHandler).Methods(http.MethodGet)
+	r.HandleFunc("/metadata/sources/connection/{name}", sourceConnectionHandler).Methods(http.MethodPost)
+	r.HandleFunc("/metadata/sinks/connection/{name}", sinkConnectionHandler).Methods(http.MethodPost)
 	for _, endpoint := range metaEndpoints {
 		endpoint(r)
 	}
@@ -85,8 +93,16 @@ func newSinkMetaHandler(w http.ResponseWriter, r *http.Request) {
 // list functions
 func functionsMetaHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
-	sinks := meta.GetFunctions()
-	jsonResponse(sinks, w, logger)
+	funcs := meta.GetFunctions()
+	jsonByteResponse(funcs, w, logger)
+	return
+}
+
+// list operators
+func operatorsMetaHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	ops := meta.GetOperators()
+	jsonByteResponse(ops, w, logger)
 	return
 }
 
@@ -172,7 +188,23 @@ func connectionConfHandler(w http.ResponseWriter, r *http.Request) {
 		handleError(w, err, "", logger)
 		return
 	} else {
-		w.Write(ret)
+		_, _ = w.Write(ret)
+	}
+}
+
+// Get sink yaml
+func sinkConfHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+	pluginName := vars["name"]
+	language := getLanguage(r)
+	configOperatorKey := fmt.Sprintf(meta.SinkCfgOperatorKeyTemplate, pluginName)
+	ret, err := meta.GetYamlConf(configOperatorKey, language)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	} else {
+		_, _ = w.Write(ret)
 	}
 }
 
@@ -202,6 +234,31 @@ func sourceConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// Add  del confkey
+func sinkConfKeyHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	var err error
+	vars := mux.Vars(r)
+	pluginName := vars["name"]
+	confKey := vars["confKey"]
+	language := getLanguage(r)
+	switch r.Method {
+	case http.MethodDelete:
+		err = meta.DelSinkConfKey(pluginName, confKey, language)
+	case http.MethodPut:
+		v, err1 := io.ReadAll(r.Body)
+		if err1 != nil {
+			handleError(w, err, "Invalid body", logger)
+			return
+		}
+		err = meta.AddSinkConfKey(pluginName, confKey, language, v)
+	}
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+}
+
 // Add  del confkey
 func connectionConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 
@@ -228,6 +285,19 @@ func connectionConfKeyHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// get updatable resources
+func resourcesHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	language := getLanguage(r)
+	ret, err := meta.GetResources(language)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	} else {
+		_, _ = w.Write(ret)
+	}
+}
+
 func getLanguage(r *http.Request) string {
 	language := r.Header.Get("Content-Language")
 	if 0 == len(language) {
@@ -237,3 +307,47 @@ func getLanguage(r *http.Request) string {
 	}
 	return language
 }
+
+func sinkConnectionHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+
+	sinkNm := vars["name"]
+	config := map[string]interface{}{}
+	v, _ := io.ReadAll(r.Body)
+	err := json.Unmarshal(v, &config)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+
+	err = node.SinkOpen(sinkNm, config)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}
+
+func sourceConnectionHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	vars := mux.Vars(r)
+
+	sourceNm := vars["name"]
+	config := map[string]interface{}{}
+	v, _ := io.ReadAll(r.Body)
+	err := json.Unmarshal(v, &config)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+
+	err = node.SourceOpen(sourceNm, config)
+	if err != nil {
+		handleError(w, err, "", logger)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}
diff --git a/internal/server/prome_init.go b/internal/server/prome_init.go
index 4da0dd4c..fbda06c0 100644
--- a/internal/server/prome_init.go
+++ b/internal/server/prome_init.go
@@ -20,6 +20,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"github.com/gorilla/mux"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"net/http"
@@ -27,38 +28,58 @@ import (
 )
 
 func init() {
-	servers["prometheus"] = promeComp{}
+	p := promeComp{}
+	servers["prometheus"] = p
+	components["prometheus"] = p
 }
 
 type promeComp struct {
 	s *http.Server
 }
 
+func (p promeComp) register() {
+	// Do nothing
+}
+
+func (p promeComp) rest(r *mux.Router) {
+	portPrometheus := conf.Config.Basic.PrometheusPort
+	portRest := conf.Config.Basic.RestPort
+	if portPrometheus == portRest {
+		r.Handle("/metrics", promhttp.Handler())
+		msg := fmt.Sprintf("Register prometheus metrics to http://localhost:%d/metrics", portPrometheus)
+		logger.Infof(msg)
+		fmt.Println(msg)
+	}
+}
+
 func (p promeComp) serve() {
-	//Start prometheus service
 	if conf.Config.Basic.Prometheus {
+		//Start prometheus service
 		portPrometheus := conf.Config.Basic.PrometheusPort
 		if portPrometheus <= 0 {
 			logger.Fatal("Miss configuration prometheusPort")
 		}
-		mux := http.NewServeMux()
-		mux.Handle("/metrics", promhttp.Handler())
-		srvPrometheus := &http.Server{
-			Addr:         fmt.Sprintf("0.0.0.0:%d", portPrometheus),
-			WriteTimeout: time.Second * 15,
-			ReadTimeout:  time.Second * 15,
-			IdleTimeout:  time.Second * 60,
-			Handler:      mux,
-		}
-		go func() {
-			if err := srvPrometheus.ListenAndServe(); err != nil && err != http.ErrServerClosed {
-				logger.Fatal("Listen prometheus error: ", err)
+		portRest := conf.Config.Basic.RestPort
+		if portPrometheus != portRest {
+			mux := http.NewServeMux()
+			mux.Handle("/metrics", promhttp.Handler())
+			srvPrometheus := &http.Server{
+				Addr:         fmt.Sprintf("0.0.0.0:%d", portPrometheus),
+				WriteTimeout: time.Second * 15,
+				ReadTimeout:  time.Second * 15,
+				IdleTimeout:  time.Second * 60,
+				Handler:      mux,
 			}
-		}()
-		p.s = srvPrometheus
-		msg := fmt.Sprintf("Serving prometheus metrics on port http://localhost:%d/metrics", portPrometheus)
-		logger.Infof(msg)
-		fmt.Println(msg)
+			go func() {
+				if err := srvPrometheus.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+					logger.Fatal("Listen prometheus error: ", err)
+				}
+			}()
+			p.s = srvPrometheus
+			msg := fmt.Sprintf("Serving prometheus metrics on port http://localhost:%d/metrics", portPrometheus)
+			logger.Infof(msg)
+			fmt.Println(msg)
+		}
 	}
 }
 
diff --git a/internal/server/rest.go b/internal/server/rest.go
index e58523ae..807b6547 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -98,6 +98,18 @@ func jsonResponse(i interface{}, w http.ResponseWriter, logger api.Logger) {
 	}
 }
 
+func jsonByteResponse(buffer bytes.Buffer, w http.ResponseWriter, logger api.Logger) {
+	w.Header().Add(ContentType, ContentTypeJSON)
+
+	w.Header().Add("Content-Length", strconv.Itoa(buffer.Len()))
+
+	_, err := w.Write(buffer.Bytes())
+	// Problems encoding
+	if err != nil {
+		handleError(w, err, "", logger)
+	}
+}
+
 func createRestServer(ip string, port int, needToken bool) *http.Server {
 	dataDir, err := conf.GetDataLoc()
 	if err != nil {
diff --git a/internal/server/rpc_plugin.go b/internal/server/rpc_plugin.go
index 19f66355..4ca51bfc 100644
--- a/internal/server/rpc_plugin.go
+++ b/internal/server/rpc_plugin.go
@@ -13,8 +13,6 @@
 // limitations under the License.
 
 //go:build (rpc || !core) && (plugin || portable || !core)
-// +build rpc !core
-// +build plugin portable !core
 
 package server
 
@@ -90,6 +88,16 @@ func (t *Server) DescPlugin(arg *model.PluginDesc, reply *string) error {
 	return nil
 }
 
+func (t *Server) ShowPlugins(arg int, reply *string) error {
+	pt := plugin.PluginType(arg)
+	l, err := t.doShow(pt)
+	if err != nil {
+		return fmt.Errorf("Show plugin error: %s", err)
+	}
+	*reply = l
+	return nil
+}
+
 func getPluginByJson(arg *model.PluginDesc, pt plugin.PluginType) (plugin.Plugin, error) {
 	p := plugin.NewPluginByType(pt)
 	if arg.Json != "" {
diff --git a/internal/server/rpc_plugin_both.go b/internal/server/rpc_plugin_both.go
index 8b354e9f..faa5be4a 100644
--- a/internal/server/rpc_plugin_both.go
+++ b/internal/server/rpc_plugin_both.go
@@ -13,13 +13,14 @@
 // limitations under the License.
 
 //go:build !core || (rpc && portable && plugin)
-// +build !core rpc,portable,plugin
 
 package server
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/plugin"
+	"strings"
 )
 
 func (t *Server) doRegister(pt plugin.PluginType, p plugin.Plugin) error {
@@ -53,3 +54,23 @@ func (t *Server) doDesc(pt plugin.PluginType, name string) (interface{}, error)
 	}
 	return result, nil
 }
+
+func (t *Server) doShow(pt plugin.PluginType) (string, error) {
+	var result string
+	if pt == plugin.PORTABLE {
+		l := portableManager.List()
+		jb, err := json.Marshal(l)
+		if err != nil {
+			return "", err
+		}
+		return string(jb), nil
+	} else {
+		l := nativeManager.List(pt)
+		if len(l) == 0 {
+			result = "No plugin is found."
+		} else {
+			result = strings.Join(l, "\n")
+		}
+		return result, nil
+	}
+}
diff --git a/internal/server/rpc_plugin_hasnative.go b/internal/server/rpc_plugin_hasnative.go
new file mode 100644
index 00000000..af979569
--- /dev/null
+++ b/internal/server/rpc_plugin_hasnative.go
@@ -0,0 +1,68 @@
+// 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.
+
+//go:build !core || (rpc && plugin)
+
+package server
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/internal/plugin"
+	"strings"
+)
+
+func (t *Server) RegisterPlugin(arg *model.PluginDesc, reply *string) error {
+	p, err := getPluginByJson(arg, plugin.FUNCTION)
+	if err != nil {
+		return fmt.Errorf("Register plugin functions error: %s", err)
+	}
+	if len(p.GetSymbols()) == 0 {
+		return fmt.Errorf("Register plugin functions error: Missing function list.")
+	}
+	err = nativeManager.RegisterFuncs(p.GetName(), p.GetSymbols())
+	if err != nil {
+		return fmt.Errorf("Create plugin error: %s", err)
+	} else {
+		*reply = fmt.Sprintf("Plugin %s is created.", p.GetName())
+	}
+	return nil
+}
+
+func (t *Server) ShowUdfs(_ int, reply *string) error {
+	l := nativeManager.ListSymbols()
+	if len(l) == 0 {
+		l = append(l, "No udf is found.")
+	}
+	*reply = strings.Join(l, "\n")
+	return nil
+}
+
+func (t *Server) DescUdf(arg string, reply *string) error {
+	m, ok := nativeManager.GetPluginBySymbol(plugin.FUNCTION, arg)
+	if !ok {
+		return fmt.Errorf("Describe udf error: not found")
+	} else {
+		j := map[string]string{
+			"name":   arg,
+			"plugin": m,
+		}
+		r, err := marshalDesc(j)
+		if err != nil {
+			return fmt.Errorf("Describe udf error: %v", err)
+		}
+		*reply = r
+	}
+	return nil
+}
diff --git a/internal/server/rpc_plugin_native.go b/internal/server/rpc_plugin_native.go
index 6de4cabf..3d16fe21 100644
--- a/internal/server/rpc_plugin_native.go
+++ b/internal/server/rpc_plugin_native.go
@@ -19,7 +19,6 @@ package server
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/pkg/model"
 	"github.com/lf-edge/ekuiper/internal/plugin"
 	"strings"
 )
@@ -52,56 +51,17 @@ func (t *Server) doDesc(pt plugin.PluginType, name string) (interface{}, error)
 	}
 }
 
-func (t *Server) RegisterPlugin(arg *model.PluginDesc, reply *string) error {
-	p, err := getPluginByJson(arg, plugin.FUNCTION)
-	if err != nil {
-		return fmt.Errorf("Register plugin functions error: %s", err)
-	}
-	if len(p.GetSymbols()) == 0 {
-		return fmt.Errorf("Register plugin functions error: Missing function list.")
-	}
-	err = nativeManager.RegisterFuncs(p.GetName(), p.GetSymbols())
-	if err != nil {
-		return fmt.Errorf("Create plugin error: %s", err)
-	} else {
-		*reply = fmt.Sprintf("Plugin %s is created.", p.GetName())
-	}
-	return nil
-}
-
-func (t *Server) ShowPlugins(arg int, reply *string) error {
-	pt := plugin.PluginType(arg)
-	l := nativeManager.List(pt)
-	if len(l) == 0 {
-		l = append(l, "No plugin is found.")
-	}
-	*reply = strings.Join(l, "\n")
-	return nil
-}
-
-func (t *Server) ShowUdfs(_ int, reply *string) error {
-	l := nativeManager.ListSymbols()
-	if len(l) == 0 {
-		l = append(l, "No udf is found.")
-	}
-	*reply = strings.Join(l, "\n")
-	return nil
-}
-
-func (t *Server) DescUdf(arg string, reply *string) error {
-	m, ok := nativeManager.GetPluginBySymbol(plugin.FUNCTION, arg)
-	if !ok {
-		return fmt.Errorf("Describe udf error: not found")
+func (t *Server) doShow(pt plugin.PluginType) (string, error) {
+	var result string
+	if pt == plugin.PORTABLE {
+		return "", fmt.Errorf("portable plugin support is disabled")
 	} else {
-		j := map[string]string{
-			"name":   arg,
-			"plugin": m,
-		}
-		r, err := marshalDesc(j)
-		if err != nil {
-			return fmt.Errorf("Describe udf error: %v", err)
+		l := nativeManager.List(pt)
+		if len(l) == 0 {
+			result = "No plugin is found."
+		} else {
+			result = strings.Join(l, "\n")
 		}
-		*reply = r
+		return result, nil
 	}
-	return nil
 }
diff --git a/internal/server/rpc_plugin_portable.go b/internal/server/rpc_plugin_portable.go
index f5280cf5..f92a4ad8 100644
--- a/internal/server/rpc_plugin_portable.go
+++ b/internal/server/rpc_plugin_portable.go
@@ -18,6 +18,7 @@
 package server
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/plugin"
 )
@@ -49,3 +50,16 @@ func (t *Server) doDesc(pt plugin.PluginType, name string) (interface{}, error)
 		return nil, fmt.Errorf("native plugin support is disabled")
 	}
 }
+
+func (t *Server) doShow(pt plugin.PluginType) (string, error) {
+	if pt == plugin.PORTABLE {
+		l := portableManager.List()
+		jb, err := json.Marshal(l)
+		if err != nil {
+			return "", err
+		}
+		return string(jb), nil
+	} else {
+		return "", fmt.Errorf("native plugin support is disabled")
+	}
+}
diff --git a/internal/topo/connection/clients/client_registy.go b/internal/topo/connection/clients/client_registy.go
index f3f547c1..3a776180 100644
--- a/internal/topo/connection/clients/client_registy.go
+++ b/internal/topo/connection/clients/client_registy.go
@@ -26,21 +26,21 @@ type clientRegistry struct {
 	Lock                sync.Mutex
 	supportedClientType []string
 	clientFactory       map[string]ClientFactoryFunc
-	ShareClientStore    map[string]ClientWrapper
+	shareClientStore    map[string]ClientWrapper
 }
 
-var ClientRegistry = clientRegistry{
+var gClientRegistry = clientRegistry{
 	clientFactory:       make(map[string]ClientFactoryFunc),
 	Lock:                sync.Mutex{},
 	supportedClientType: make([]string, 0),
-	ShareClientStore:    make(map[string]ClientWrapper),
+	shareClientStore:    make(map[string]ClientWrapper),
 }
 
 func RegisterClientFactory(clientType string, creatorFunc ClientFactoryFunc) {
-	ClientRegistry.Lock.Lock()
-	ClientRegistry.clientFactory[clientType] = creatorFunc
-	ClientRegistry.supportedClientType = append(ClientRegistry.supportedClientType, clientType)
-	ClientRegistry.Lock.Unlock()
+	gClientRegistry.Lock.Lock()
+	gClientRegistry.clientFactory[clientType] = creatorFunc
+	gClientRegistry.supportedClientType = append(gClientRegistry.supportedClientType, clientType)
+	gClientRegistry.Lock.Unlock()
 }
 
 func getConnectionSelector(props map[string]interface{}) (ConnectionSelector string, err error) {
@@ -57,24 +57,24 @@ func getConnectionSelector(props map[string]interface{}) (ConnectionSelector str
 }
 
 func GetClient(connectionType string, props map[string]interface{}) (api.MessageClient, error) {
-	ClientRegistry.Lock.Lock()
-	defer ClientRegistry.Lock.Unlock()
+	gClientRegistry.Lock.Lock()
+	defer gClientRegistry.Lock.Unlock()
 
 	connectSelector, err := getConnectionSelector(props)
 	if err != nil {
 		return nil, err
 	}
 	if connectSelector != "" {
-		if cliWpr, found := ClientRegistry.ShareClientStore[connectSelector]; found {
+		if cliWpr, found := gClientRegistry.shareClientStore[connectSelector]; found {
 			cliWpr.AddRef()
 			return cliWpr, nil
 		}
 	}
 
-	clientCreator, ok := ClientRegistry.clientFactory[connectionType]
+	clientCreator, ok := gClientRegistry.clientFactory[connectionType]
 	if !ok {
-		conf.Log.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, ClientRegistry.supportedClientType)
-		return nil, fmt.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, ClientRegistry.supportedClientType)
+		conf.Log.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, gClientRegistry.supportedClientType)
+		return nil, fmt.Errorf("can not find clientCreator for connection type : %s. only support %s", connectionType, gClientRegistry.supportedClientType)
 	}
 
 	if connectSelector != "" {
@@ -95,7 +95,7 @@ func GetClient(connectionType string, props map[string]interface{}) (api.Message
 		}
 		cliWpr.SetConnectionSelector(connectSelector)
 		conf.Log.Infof("Init client wrapper for client type %s and connection selector %s", connectionType, connectSelector)
-		ClientRegistry.ShareClientStore[connectSelector] = cliWpr
+		gClientRegistry.shareClientStore[connectSelector] = cliWpr
 		return cliWpr, nil
 	} else {
 		cliWpr, err := clientCreator(props)
@@ -107,3 +107,18 @@ func GetClient(connectionType string, props map[string]interface{}) (api.Message
 		return cliWpr, nil
 	}
 }
+
+func ReleaseClient(ctx api.StreamContext, cli api.MessageClient) {
+	log := ctx.GetLogger()
+
+	wrapper := cli.(ClientWrapper)
+	sel := wrapper.GetConnectionSelector()
+	ok := wrapper.Release(ctx)
+
+	if sel != "" && ok {
+		log.Infof("remove mqtt client wrapper for connection selector %s", sel)
+		gClientRegistry.Lock.Lock()
+		delete(gClientRegistry.shareClientStore, sel)
+		gClientRegistry.Lock.Unlock()
+	}
+}
diff --git a/internal/topo/connection/clients/edgex/edgex_wrapper.go b/internal/topo/connection/clients/edgex/edgex_wrapper.go
index 530fc221..48deba6b 100644
--- a/internal/topo/connection/clients/edgex/edgex_wrapper.go
+++ b/internal/topo/connection/clients/edgex/edgex_wrapper.go
@@ -228,12 +228,14 @@ func (mc *edgexClientWrapper) SetConnectionSelector(conSelector string) {
 	mc.conSelector = conSelector
 }
 
-func (mc *edgexClientWrapper) Release(c api.StreamContext) {
+func (mc *edgexClientWrapper) GetConnectionSelector() string {
+	return mc.conSelector
+}
+
+func (mc *edgexClientWrapper) Release(c api.StreamContext) bool {
 	mc.unsubscribe(c)
 
-	clients.ClientRegistry.Lock.Lock()
-	mc.DeRef(c)
-	clients.ClientRegistry.Lock.Unlock()
+	return mc.deRef(c)
 }
 
 func (mc *edgexClientWrapper) AddRef() {
@@ -244,7 +246,7 @@ func (mc *edgexClientWrapper) AddRef() {
 	conf.Log.Infof("edgex client wrapper add refence for connection selector %s total refcount %d", mc.conSelector, mc.refCnt)
 }
 
-func (mc *edgexClientWrapper) DeRef(c api.StreamContext) {
+func (mc *edgexClientWrapper) deRef(c api.StreamContext) bool {
 	log := c.GetLogger()
 	mc.refLock.Lock()
 	defer mc.refLock.Unlock()
@@ -255,14 +257,13 @@ func (mc *edgexClientWrapper) DeRef(c api.StreamContext) {
 	}
 	if mc.refCnt == 0 {
 		log.Infof("mqtt client wrapper reference count 0")
-		if mc.conSelector != "" {
-			conf.Log.Infof("remove mqtt client wrapper for connection selector %s", mc.conSelector)
-			delete(clients.ClientRegistry.ShareClientStore, mc.conSelector)
-		}
 		// clean the go routine that waiting on the messages
 		for _, sub := range mc.topicSubscriptions {
 			sub.stop <- struct{}{}
 		}
 		_ = mc.cli.Disconnect()
+		return true
+	} else {
+		return false
 	}
 }
diff --git a/internal/topo/connection/clients/mqtt/mqtt_wrapper.go b/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
index 499b6c60..7a9aa5ef 100644
--- a/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
+++ b/internal/topo/connection/clients/mqtt/mqtt_wrapper.go
@@ -260,18 +260,19 @@ func (mc *mqttClientWrapper) unsubscribe(c api.StreamContext) {
 	delete(mc.subscribers, subId)
 }
 
-func (mc *mqttClientWrapper) Release(c api.StreamContext) {
+func (mc *mqttClientWrapper) Release(c api.StreamContext) bool {
 	mc.unsubscribe(c)
-
-	clients.ClientRegistry.Lock.Lock()
-	mc.DeRef(c)
-	clients.ClientRegistry.Lock.Unlock()
+	return mc.deRef(c)
 }
 
 func (mc *mqttClientWrapper) SetConnectionSelector(conSelector string) {
 	mc.conSelector = conSelector
 }
 
+func (mc *mqttClientWrapper) GetConnectionSelector() string {
+	return mc.conSelector
+}
+
 func (mc *mqttClientWrapper) AddRef() {
 	mc.refLock.Lock()
 	defer mc.refLock.Unlock()
@@ -279,18 +280,17 @@ func (mc *mqttClientWrapper) AddRef() {
 	conf.Log.Infof("mqtt client wrapper add refence for connection selector %s total refcount %d", mc.conSelector, mc.refCnt)
 }
 
-func (mc *mqttClientWrapper) DeRef(c api.StreamContext) {
+func (mc *mqttClientWrapper) deRef(c api.StreamContext) bool {
 	log := c.GetLogger()
 	mc.refLock.Lock()
 	defer mc.refLock.Unlock()
 
 	mc.refCnt = mc.refCnt - 1
+	log.Infof("mqtt client wrapper reference count %d", mc.refCnt)
 	if mc.refCnt == 0 {
-		log.Infof("mqtt client wrapper reference count 0")
-		if mc.conSelector != "" {
-			conf.Log.Infof("remove mqtt client wrapper for connection selector %s", mc.conSelector)
-			delete(clients.ClientRegistry.ShareClientStore, mc.conSelector)
-		}
 		_ = mc.cli.Disconnect()
+		return true
+	} else {
+		return false
 	}
 }
diff --git a/internal/topo/connection/clients/type.go b/internal/topo/connection/clients/type.go
index aee98fbd..4c6ec009 100644
--- a/internal/topo/connection/clients/type.go
+++ b/internal/topo/connection/clients/type.go
@@ -32,8 +32,9 @@ type ClientFactoryFunc func(props map[string]interface{}) (ClientWrapper, error)
 
 type ClientWrapper interface {
 	Subscribe(c api.StreamContext, subChan []api.TopicChannel, messageErrors chan error, params map[string]interface{}) error
-	Release(c api.StreamContext)
+	Release(c api.StreamContext) bool
 	Publish(c api.StreamContext, topic string, message []byte, params map[string]interface{}) error
 	SetConnectionSelector(conSelector string)
+	GetConnectionSelector() string
 	AddRef()
 }
diff --git a/internal/topo/context/default.go b/internal/topo/context/default.go
index 6b52a92a..6f7fe270 100644
--- a/internal/topo/context/default.go
+++ b/internal/topo/context/default.go
@@ -19,7 +19,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/internal/topo/transform"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
@@ -251,7 +250,3 @@ func (c *DefaultContext) SaveState(checkpointId int64) error {
 	c.snapshot = nil
 	return nil
 }
-
-func (c *DefaultContext) GetClient(clientType string, config map[string]interface{}) (api.MessageClient, error) {
-	return clients.GetClient(clientType, config)
-}
diff --git a/internal/topo/context/func_context.go b/internal/topo/context/func_context.go
index ff1b050c..02a493e3 100644
--- a/internal/topo/context/func_context.go
+++ b/internal/topo/context/func_context.go
@@ -16,7 +16,6 @@ package context
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/pkg/api"
 )
 
@@ -56,10 +55,6 @@ func (c *DefaultFuncContext) GetFuncId() int {
 	return c.funcId
 }
 
-func (c *DefaultFuncContext) GetClient(clientType string, config map[string]interface{}) (api.MessageClient, error) {
-	return clients.GetClient(clientType, config)
-}
-
 func (c *DefaultFuncContext) convertKey(key string) string {
 	return fmt.Sprintf("$$func%d_%s", c.funcId, key)
 }
diff --git a/internal/topo/node/conf/sink.go b/internal/topo/node/conf/sink.go
new file mode 100644
index 00000000..5d1a4a69
--- /dev/null
+++ b/internal/topo/node/conf/sink.go
@@ -0,0 +1,55 @@
+// 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 conf
+
+import (
+	"github.com/lf-edge/ekuiper/internal/conf"
+)
+
+const ResourceID = "resourceId"
+
+func GetSinkConf(sinkType string, action map[string]interface{}) map[string]interface{} {
+	resourceId, ok := action[ResourceID].(string)
+	if !ok {
+		return action
+	}
+	delete(action, ResourceID)
+
+	yamlOps, err := conf.NewConfigOperatorFromSinkYaml(sinkType)
+	if err != nil {
+		conf.Log.Warnf("fail to parse yaml for sink %s. Return error %v", sinkType, err)
+		return action
+	}
+	props := make(map[string]interface{})
+	cfg := yamlOps.CopyConfContent()
+	if len(cfg) == 0 {
+		conf.Log.Warnf("fail to parse yaml for sink %s. Return an empty configuration", sinkType)
+		return action
+	} else {
+		def, ok := cfg[resourceId]
+		if !ok {
+			conf.Log.Warnf("resource id %s is not found", resourceId)
+			return action
+		} else {
+			props = def
+			for k, v := range action {
+				props[k] = v
+			}
+		}
+	}
+
+	conf.Log.Debugf("get conf for %s with resource id %s: %v", sinkType, resourceId, printable(props))
+	return props
+}
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index b33c22d3..4397a824 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -16,10 +16,14 @@ package node
 
 import (
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/binder/io"
+	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/checkpoint"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 )
 
 type OperatorNode interface {
@@ -166,3 +170,42 @@ func (o *defaultSinkNode) preprocess(data interface{}) (interface{}, bool) {
 	}
 	return data, false
 }
+
+func SinkOpen(sinkType string, config map[string]interface{}) error {
+
+	sink, err := getSink(sinkType, config)
+	if err != nil {
+		return err
+	}
+
+	contextLogger := conf.Log.WithField("rule", "TestSinkOpen"+"_"+sinkType)
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+
+	defer func() {
+		_ = sink.Close(ctx)
+	}()
+
+	return sink.Open(ctx)
+}
+
+func SourceOpen(sourceType string, config map[string]interface{}) error {
+
+	options := &ast.Options{}
+	//_ = nodeConf.GetSourceConf(sourceType, options)
+
+	ns, err := io.Source(sourceType)
+	if err != nil {
+		return err
+	}
+	err = ns.Configure(options.DATASOURCE, config)
+	if err != nil {
+		return err
+	}
+
+	contextLogger := conf.Log.WithField("rule", "TestSourceOpen"+"_"+sourceType)
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	defer ns.Close(ctx)
+
+	return nil
+
+}
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index 26803e07..95def7e6 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -20,6 +20,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/topo/node/cache"
+	nodeConf "github.com/lf-edge/ekuiper/internal/topo/node/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/node/metric"
 	"github.com/lf-edge/ekuiper/internal/topo/transform"
 	"github.com/lf-edge/ekuiper/internal/xsql"
@@ -388,7 +389,8 @@ func getSink(name string, action map[string]interface{}) (api.Sink, error) {
 	)
 	s, err = io.Sink(name)
 	if s != nil {
-		err = s.Configure(action)
+		newAction := nodeConf.GetSinkConf(name, action)
+		err = s.Configure(newAction)
 		if err != nil {
 			return nil, err
 		}
diff --git a/internal/topo/planner/planner_graph.go b/internal/topo/planner/planner_graph.go
index c3f52d5e..4b6688f6 100644
--- a/internal/topo/planner/planner_graph.go
+++ b/internal/topo/planner/planner_graph.go
@@ -17,6 +17,7 @@ package planner
 import (
 	"errors"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/binder/function"
 	"github.com/lf-edge/ekuiper/internal/topo"
 	"github.com/lf-edge/ekuiper/internal/topo/graph"
 	"github.com/lf-edge/ekuiper/internal/topo/node"
@@ -460,7 +461,7 @@ func parseFunc(props map[string]interface{}) (*operator.FuncOp, error) {
 	} else {
 		name = f.Name
 	}
-	return &operator.FuncOp{CallExpr: c, Name: name}, nil
+	return &operator.FuncOp{CallExpr: c, Name: name, IsAgg: function.IsAggFunc(name)}, nil
 }
 
 func parseFilter(props map[string]interface{}) (*operator.FilterOp, error) {
diff --git a/internal/topo/sink/edgex_sink.go b/internal/topo/sink/edgex_sink.go
index 3af2be27..a9814922 100644
--- a/internal/topo/sink/edgex_sink.go
+++ b/internal/topo/sink/edgex_sink.go
@@ -25,6 +25,7 @@ import (
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos/requests"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -93,7 +94,7 @@ func (ems *EdgexMsgBusSink) Configure(ps map[string]interface{}) error {
 func (ems *EdgexMsgBusSink) Open(ctx api.StreamContext) error {
 	log := ctx.GetLogger()
 
-	cli, err := ctx.GetClient("edgex", ems.config)
+	cli, err := clients.GetClient("edgex", ems.config)
 	if err != nil {
 		log.Errorf("found error when get edgex client, error %s", err.Error())
 		return err
@@ -522,7 +523,7 @@ func (ems *EdgexMsgBusSink) Close(ctx api.StreamContext) error {
 	logger := ctx.GetLogger()
 	logger.Infof("Closing edgex sink")
 	if ems.cli != nil {
-		ems.cli.Release(ctx)
+		clients.ReleaseClient(ctx, ems.cli)
 	}
 	return nil
 }
diff --git a/internal/topo/sink/mqtt_sink.go b/internal/topo/sink/mqtt_sink.go
index b195995d..34d640a0 100644
--- a/internal/topo/sink/mqtt_sink.go
+++ b/internal/topo/sink/mqtt_sink.go
@@ -16,6 +16,7 @@ package sink
 
 import (
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -72,8 +73,7 @@ func (ms *MQTTSink) Configure(ps map[string]interface{}) error {
 
 func (ms *MQTTSink) Open(ctx api.StreamContext) error {
 	log := ctx.GetLogger()
-
-	cli, err := ctx.GetClient("mqtt", ms.config)
+	cli, err := clients.GetClient("mqtt", ms.config)
 	if err != nil {
 		log.Errorf("found error when get mqtt client config %v, error %s", ms.config, err.Error())
 		return err
@@ -112,7 +112,7 @@ func (ms *MQTTSink) Close(ctx api.StreamContext) error {
 	logger := ctx.GetLogger()
 	logger.Infof("Closing mqtt sink")
 	if ms.cli != nil {
-		ms.cli.Release(ctx)
+		clients.ReleaseClient(ctx, ms.cli)
 	}
 	return nil
 }
diff --git a/internal/topo/source/edgex_source.go b/internal/topo/source/edgex_source.go
index 8255d042..8ae596cb 100644
--- a/internal/topo/source/edgex_source.go
+++ b/internal/topo/source/edgex_source.go
@@ -25,6 +25,7 @@ import (
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos/requests"
 	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
 	"github.com/fxamacker/cbor/v2"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"strconv"
@@ -69,20 +70,18 @@ func (es *EdgexSource) Configure(_ string, props map[string]interface{}) error {
 	es.topic = c.Topic
 	es.config = props
 
+	cli, err := clients.GetClient("edgex", es.config)
+	if err != nil {
+		return err
+	}
+	es.cli = cli.(api.MessageClient)
+
 	return nil
 }
 
 func (es *EdgexSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
 	log := ctx.GetLogger()
 
-	cli, err := ctx.GetClient("edgex", es.config)
-	if err != nil {
-		errCh <- err
-		log.Errorf("found error when get edgex client, error %s", err.Error())
-		return
-	}
-	es.cli = cli.(api.MessageClient)
-
 	messages := make(chan interface{}, es.buflen)
 	topics := []api.TopicChannel{{Topic: es.topic, Messages: messages}}
 	subErrs := make(chan error, len(topics))
@@ -309,7 +308,7 @@ func (es *EdgexSource) Close(ctx api.StreamContext) error {
 	log := ctx.GetLogger()
 	log.Infof("EdgeX Source instance %d Done.", ctx.GetInstanceId())
 	if es.cli != nil {
-		es.cli.Release(ctx)
+		clients.ReleaseClient(ctx, es.cli)
 	}
 	return nil
 }
diff --git a/internal/topo/source/httppull_source.go b/internal/topo/source/httppull_source.go
index 8611bd9d..a131b3a3 100644
--- a/internal/topo/source/httppull_source.go
+++ b/internal/topo/source/httppull_source.go
@@ -18,6 +18,7 @@ import (
 	"crypto/md5"
 	"encoding/hex"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/cert"
 	"io"
 	"net/http"
 	"net/url"
@@ -25,7 +26,6 @@ import (
 	"time"
 
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/cert"
 	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
@@ -167,17 +167,10 @@ func (hps *HTTPPullSource) Configure(device string, props map[string]interface{}
 		}
 	}
 
-	conf.Log.Debugf("Initialized with configurations %#v.", hps)
-	return nil
-}
-
-func (hps *HTTPPullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
-	_, e := url.Parse(hps.url)
-	if e != nil {
-		errCh <- e
-		return
+	_, err := url.Parse(hps.url)
+	if err != nil {
+		return err
 	}
-	log := ctx.GetLogger()
 
 	tlsOpts := cert.TlsConfigurationOptions{
 		SkipCertVerify: hps.insecureSkipVerify,
@@ -185,11 +178,10 @@ func (hps *HTTPPullSource) Open(ctx api.StreamContext, consumer chan<- api.Sourc
 		KeyFile:        hps.privateKeyPath,
 		CaFile:         hps.rootCaPath,
 	}
-	log.Infof("Connect http source with TLS configs. %v", tlsOpts)
+
 	tlscfg, err := cert.GenerateTLSForClient(tlsOpts)
 	if err != nil {
-		errCh <- err
-		return
+		return err
 	}
 
 	tr := &http.Transport{
@@ -199,6 +191,12 @@ func (hps *HTTPPullSource) Open(ctx api.StreamContext, consumer chan<- api.Sourc
 	hps.client = &http.Client{
 		Transport: tr,
 		Timeout:   time.Duration(hps.timeout) * time.Millisecond}
+
+	conf.Log.Debugf("Initialized with configurations %#v.", hps)
+	return nil
+}
+
+func (hps *HTTPPullSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
 	hps.initTimerPull(ctx, consumer, errCh)
 }
 
diff --git a/internal/topo/source/mqtt_source.go b/internal/topo/source/mqtt_source.go
index 9fe287ef..af7420db 100644
--- a/internal/topo/source/mqtt_source.go
+++ b/internal/topo/source/mqtt_source.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/connection/clients"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
@@ -75,20 +76,18 @@ func (ms *MQTTSource) Configure(topic string, props map[string]interface{}) erro
 			return err
 		}
 	}
-	return nil
-}
-
-func (ms *MQTTSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
-	log := ctx.GetLogger()
 
-	cli, err := ctx.GetClient("mqtt", ms.config)
+	cli, err := clients.GetClient("mqtt", ms.config)
 	if err != nil {
-		errCh <- err
-		log.Errorf("found error when get mqtt client config %v, error %s", ms.config, err.Error())
-		return
+		return err
 	}
 	ms.cli = cli
-	err = subscribe(ms, ctx, consumer)
+
+	return nil
+}
+
+func (ms *MQTTSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
+	err := subscribe(ms, ctx, consumer)
 	if err != nil {
 		errCh <- err
 	}
@@ -167,7 +166,7 @@ func getTuple(ctx api.StreamContext, ms *MQTTSource, env interface{}) api.Source
 func (ms *MQTTSource) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Mqtt Source instance %d Done", ctx.GetInstanceId())
 	if ms.cli != nil {
-		ms.cli.Release(ctx)
+		clients.ReleaseClient(ctx, ms.cli)
 	}
 	return nil
 }
diff --git a/pkg/api/stream.go b/pkg/api/stream.go
index 5f973647..0a6c99d1 100644
--- a/pkg/api/stream.go
+++ b/pkg/api/stream.go
@@ -173,8 +173,6 @@ type StreamContext interface {
 	PutState(key string, value interface{}) error
 	GetState(key string) (interface{}, error)
 	DeleteState(key string) error
-	// GetClient Connection related methods
-	GetClient(clientType string, config map[string]interface{}) (MessageClient, error)
 	// ParseTemplate parse the template string with the given data
 	ParseTemplate(template string, data interface{}) (string, error)
 	// ParseJsonPath parse the jsonPath string with the given data
@@ -220,7 +218,6 @@ type Qos int
 
 type MessageClient interface {
 	Subscribe(c StreamContext, subChan []TopicChannel, messageErrors chan error, params map[string]interface{}) error
-	Release(c StreamContext)
 	Publish(c StreamContext, topic string, message []byte, params map[string]interface{}) error
 }
 


Test patch:
diff --git a/internal/meta/msgUtil_test.go b/internal/meta/msgUtil_test.go
new file mode 100644
index 00000000..d77b5d05
--- /dev/null
+++ b/internal/meta/msgUtil_test.go
@@ -0,0 +1,76 @@
+// 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 meta
+
+import (
+	"bytes"
+	"reflect"
+	"testing"
+)
+
+func TestConstructJsonArray(t *testing.T) {
+	type args struct {
+		jsonByteItems []fileContent
+	}
+
+	var buf1 bytes.Buffer
+	buf1.Write([]byte("[]"))
+
+	var buf2 bytes.Buffer
+	buf2.Write([]byte(`[{"key": "key1"}]`))
+
+	var buf3 bytes.Buffer
+	buf3.Write([]byte(`[{"key1": "value1"},{"key2": "value2"}]`))
+
+	tests := []struct {
+		name string
+		args args
+		want bytes.Buffer
+	}{
+		{
+			name: "no value",
+			args: args{
+				jsonByteItems: nil,
+			},
+			want: buf1,
+		},
+		{
+			name: "one value",
+			args: args{
+				jsonByteItems: []fileContent{
+					[]byte(`{"key": "key1"}`),
+				},
+			},
+			want: buf2,
+		},
+		{
+			name: "two value",
+			args: args{
+				jsonByteItems: []fileContent{
+					[]byte(`{"key1": "value1"}`),
+					[]byte(`{"key2": "value2"}`),
+				},
+			},
+			want: buf3,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ConstructJsonArray(tt.args.jsonByteItems); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ConstructJsonArray() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/internal/meta/sinkMeta_test.go b/internal/meta/sinkMeta_test.go
index 4b24b183..888150a7 100644
--- a/internal/meta/sinkMeta_test.go
+++ b/internal/meta/sinkMeta_test.go
@@ -15,37 +15,28 @@
 package meta
 
 import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"path"
 	"testing"
 )
 
 func TestHintWhenModifySink(t *testing.T) {
-	taosMeta := &uiSink{
-		Fields: []field{
-			{
-				Name:    "ip",
-				Default: "911.911.911.911",
-			},
-		},
-	}
-	logMeta := &uiSink{
-		Fields: []field{
-			{
-				Name:    "ip",
-				Default: "911.911.911.911",
-			},
-		},
+	confDir, err := conf.GetConfLoc()
+	if nil != err {
+		return
 	}
 
-	gSinkmetadata = make(map[string]*uiSink)
-	gSinkmetadata["taos.json"] = taosMeta
-	gSinkmetadata["log.json"] = logMeta
+	if err = ReadSinkMetaFile(path.Join(confDir, "sinks", "mqtt.json"), true); nil != err {
+		t.Error(err)
+		return
+	}
 
-	oldSink, err := GetSinkMeta("taos", "en_US")
-	if err != nil {
-		t.Errorf("%v", err)
-	} else {
-		if "911.911.911.911" != oldSink.Fields[0].Default {
-			t.Errorf("fail")
-		}
+	showMeta, err := GetSinkMeta("mqtt", "zh_CN")
+	if nil != err {
+		t.Error(err)
 	}
+
+	fmt.Printf("%+v", showMeta)
+
 }