Repository: lf-edge/ekuiper
Pull request #1149 (2022-01-07 09:08:17)

refactor(portable): cache isAgg result to avoid frequent communication


Patch:
diff --git a/docs/en_US/rules/overview.md b/docs/en_US/rules/overview.md
index d9860c40..21133831 100644
--- a/docs/en_US/rules/overview.md
+++ b/docs/en_US/rules/overview.md
@@ -24,12 +24,12 @@ The following 3 parameters are required for creating a rule.
 
 ## Parameters
 
-| Parameter name | Optional | Description                                                  |
-| ------------- | -------- | ------------------------------------------------------------ |
-| id | false   | The id of the rule |
-| sql        | false   | The sql query to run for the rule |
-| actions           | false    | An array of sink actions        |
-| options           | true    | A map of options        |
+| Parameter name | Optional | Description                       |
+|----------------|----------|-----------------------------------|
+| id             | false    | The id of the rule                |
+| sql            | false    | The sql query to run for the rule |
+| actions        | false    | An array of sink actions          |
+| options        | true     | A map of options                  |
 
 ### id
 
@@ -43,16 +43,16 @@ The sql query to run for the rule.
 
 The current options includes:
 
-| Option name | Type & Default Value | Description                                                  |
-| ------------- | -------- | ------------------------------------------------------------ |
-| isEventTime | boolean: false   | Whether to use event time or processing time as the timestamp for an event. If event time is used, the timestamp will be extracted from the payload. The timestamp filed must be specified by the [stream](../sqls/streams.md) definition. |
-| lateTolerance        | int64:0   | When working with event-time windowing, it can happen that elements arrive late. LateTolerance can specify by how much time(unit is millisecond) elements can be late before they are dropped. By default, the value is 0 which means late elements are dropped.  |
-| concurrency | int: 1   | A rule is processed by several phases of plans according to the sql statement. This option will specify how many instances will be run for each plan. If the value is bigger than 1, the order of the messages may not be retained. |
-| bufferLength | int: 1024   | Specify how many messages can be buffered in memory for each plan. If the buffered messages exceed the limit, the plan will block message receiving until the buffered messages have been sent out so that the buffered size is less than the limit. A bigger value will accommodate more throughput but will also take up more memory footprint.  |
-| sendMetaToSink | bool:false   | Specify whether the meta data of an event will be sent to the sink. If true, the sink can get te meta data information.  |
-| sendError  | bool: true | Whether to send the error to sink. If true, any runtime error will be sent through the whole rule into sinks. Otherwise, the error will only be printed out in the log. |
-| qos | int:0   | Specify the qos of the stream. The options are 0: At most once; 1: At least once and 2: Exactly once. If qos is bigger than 0, the checkpoint mechanism will be activated to save states periodically so that the rule can be resumed from errors.  |
-| checkpointInterval | int:300000   | Specify the time interval in milliseconds to trigger a checkpoint. This is only effective when qos is bigger than 0.  |
+| Option name        | Type & Default Value | Description                                                                                                                                                                                                                                                                                                                                       |
+|--------------------|----------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| isEventTime        | boolean: false       | Whether to use event time or processing time as the timestamp for an event. If event time is used, the timestamp will be extracted from the payload. The timestamp filed must be specified by the [stream](../sqls/streams.md) definition.                                                                                                        |
+| lateTolerance      | int64:0              | When working with event-time windowing, it can happen that elements arrive late. LateTolerance can specify by how much time(unit is millisecond) elements can be late before they are dropped. By default, the value is 0 which means late elements are dropped.                                                                                  |
+| concurrency        | int: 1               | A rule is processed by several phases of plans according to the sql statement. This option will specify how many instances will be run for each plan. If the value is bigger than 1, the order of the messages may not be retained.                                                                                                               |
+| bufferLength       | int: 1024            | Specify how many messages can be buffered in memory for each plan. If the buffered messages exceed the limit, the plan will block message receiving until the buffered messages have been sent out so that the buffered size is less than the limit. A bigger value will accommodate more throughput but will also take up more memory footprint. |
+| sendMetaToSink     | bool:false           | Specify whether the meta data of an event will be sent to the sink. If true, the sink can get te meta data information.                                                                                                                                                                                                                           |
+| sendError          | bool: true           | Whether to send the error to sink. If true, any runtime error will be sent through the whole rule into sinks. Otherwise, the error will only be printed out in the log.                                                                                                                                                                           |
+| qos                | int:0                | Specify the qos of the stream. The options are 0: At most once; 1: At least once and 2: Exactly once. If qos is bigger than 0, the checkpoint mechanism will be activated to save states periodically so that the rule can be resumed from errors.                                                                                                |
+| checkpointInterval | int:300000           | Specify the time interval in milliseconds to trigger a checkpoint. This is only effective when qos is bigger than 0.                                                                                                                                                                                                                              |
 
 For detail about `qos` and `checkpointInterval`, please check [state and fault tolerance](./state_and_fault_tolerance.md).
 
@@ -79,18 +79,18 @@ Currently, below kinds of sinks/actions are supported:
 
 Each action can define its own properties. There are several common properties:
 
-| property name | Type & Default Value | Description                                                  |
-| ------------- | -------- | ------------------------------------------------------------ |
-| concurrency | int: 1   | Specify how many instances of the sink will be run. If the value is bigger than 1, the order of the messages may not be retained. |
-| bufferLength | int: 1024   | Specify how many messages can be buffered in memory. If the buffered messages exceed the limit, the sink will block message receiving until the buffered messages have been sent out so that the buffered size is less than the limit. |
-| runAsync        | bool:false   | Whether the sink will run asynchronously for better performance. If it is true, the sink result order is not promised.  |
-| retryInterval   | int:1000   | Specify how many milliseconds will the sink retry to send data out if the previous send failed. If the specified value <= 0, then it will not retry. |
-| retryCount | int:3 | Specify how many will the sink retry to send data out if the previous send failed. If the specified value <= 0, then it will not retry. |
-| cacheLength     | int:1024   | Specify how many messages can be cached. The cached messages will be resent to external system until the data sent out successfully. The cached message will be sent in order except in runAsync or concurrent mode. The cached message will be saved to disk in fixed intervals.  |
-| cacheSaveInterval  | int:1000   | Specify the interval to save cached message to the disk. Notice that, if the rule is closed in plan, all the cached messages will be saved at close. A larger value can reduce the saving overhead but may lose more cache messages when the system is interrupted in error.  |
-| omitIfEmpty | bool: false | If the configuration item is set to true, when SELECT result is empty, then the result will not feed to sink operator. |
-| sendSingle        | true     | The output messages are received as an array. This is indicate whether to send the results one by one. If false, the output message will be ``{"result":"${the string of received message}"}``. For example, ``{"result":"[{\"count\":30},"\"count\":20}]"}``. Otherwise, the result message will be sent one by one with the actual field name. For the same example as above, it will send ``{"count":30}``, then send ``{"count":20}`` to the RESTful endpoint.Default to false. |
-| dataTemplate      | true     | The [golang template](https://golang.org/pkg/html/template) format string to specify the output data format. The input of the template is the sink message which is always an array of map. If no data template is specified, the raw input will be the data. |
+| property name     | Type & Default Value | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
+|-------------------|----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| concurrency       | int: 1               | Specify how many instances of the sink will be run. If the value is bigger than 1, the order of the messages may not be retained.                                                                                                                                                                                                                                                                                                                                                   |
+| bufferLength      | int: 1024            | Specify how many messages can be buffered in memory. If the buffered messages exceed the limit, the sink will block message receiving until the buffered messages have been sent out so that the buffered size is less than the limit.                                                                                                                                                                                                                                              |
+| runAsync          | bool:false           | Whether the sink will run asynchronously for better performance. If it is true, the sink result order is not promised.                                                                                                                                                                                                                                                                                                                                                              |
+| retryInterval     | int:1000             | Specify how many milliseconds will the sink retry to send data out if the previous send failed. If the specified value <= 0, then it will not retry.                                                                                                                                                                                                                                                                                                                                |
+| retryCount        | int:3                | Specify how many will the sink retry to send data out if the previous send failed. If the specified value <= 0, then it will not retry.                                                                                                                                                                                                                                                                                                                                             |
+| cacheLength       | int:1024             | Specify how many messages can be cached. The cached messages will be resent to external system until the data sent out successfully. The cached message will be sent in order except in runAsync or concurrent mode. The cached message will be saved to disk in fixed intervals.                                                                                                                                                                                                   |
+| cacheSaveInterval | int:1000             | Specify the interval to save cached message to the disk. Notice that, if the rule is closed in plan, all the cached messages will be saved at close. A larger value can reduce the saving overhead but may lose more cache messages when the system is interrupted in error.                                                                                                                                                                                                        |
+| omitIfEmpty       | bool: false          | If the configuration item is set to true, when SELECT result is empty, then the result will not feed to sink operator.                                                                                                                                                                                                                                                                                                                                                              |
+| sendSingle        | true                 | The output messages are received as an array. This is indicate whether to send the results one by one. If false, the output message will be ``{"result":"${the string of received message}"}``. For example, ``{"result":"[{\"count\":30},"\"count\":20}]"}``. Otherwise, the result message will be sent one by one with the actual field name. For the same example as above, it will send ``{"count":30}``, then send ``{"count":20}`` to the RESTful endpoint.Default to false. |
+| dataTemplate      | true                 | The [golang template](https://golang.org/pkg/html/template) format string to specify the output data format. The input of the template is the sink message which is always an array of map. If no data template is specified, the raw input will be the data.                                                                                                                                                                                                                       |
 
 ### Data Template
 
@@ -169,7 +169,7 @@ eKuiper extends several functions that can be used in data template.
 
 ### Dynamic properties
 
-In the sink, it is common to fetch a property value from the result data to achieve dynamic output. For example, to write data into a dynamic topic of mqtt. The dynamic properties will all follow the json path syntax. In below example, the sink topic is gotten from the selected topic by using jsonpath syntax.
+In the sink, it is common to fetch a property value from the result data to achieve dynamic output. For example, to write data into a dynamic topic of mqtt. The dynamic properties will be parsed as a [data template](#data-template). In below example, the sink topic is gotten from the selected topic using data template.
 
 ```json
 {
@@ -178,10 +178,10 @@ In the sink, it is common to fetch a property value from the result data to achi
   "actions": [{
     "mqtt": {
       "sendSingle": true,
-      "topic": "$.topic"
+      "topic": "prefix/{{.topic}}"
     }
   }]
 }
 ```
 
-In the above example, `sendSingle` property is used, so the sink data is a map by default. If not using `sendSingle`, you can get the topic by index with jsonapth `$[0].topic`.
\ No newline at end of file
+In the above example, `sendSingle` property is used, so the sink data is a map by default. If not using `sendSingle`, you can get the topic by index with data template `{{index . 0 \"topic\"}}`.
\ No newline at end of file
diff --git a/docs/en_US/rules/sinks/edgex.md b/docs/en_US/rules/sinks/edgex.md
index bddc7ecd..0882c53b 100644
--- a/docs/en_US/rules/sinks/edgex.md
+++ b/docs/en_US/rules/sinks/edgex.md
@@ -6,22 +6,22 @@ The action is used for publishing output message into EdgeX message bus.
 
 **Also, you need to expose the port number to host server before running the eKuiper server if you want to have the service available to other hosts.**
 
-| Property name | Optional | Description                                                  |
-| ------------- | -------- | ------------------------------------------------------------ |
-| type          | true     | The message bus type, three types of message buses are supported, ``zero``, ``mqtt`` and ``redis``, and ``redis`` is the default value. |
-| protocol      | true     | The protocol. If it's not specified, then use default value ``redis``. |
-| host          | true     | The host of message bus. If not specified, then use default value ``localhost``. |
-| port          | true     | The port of message bus. If not specified, then use default value ``6379``. |
-| connectionSelector | true     | reuse the connection to EdgeX message bus. [more info](../sources/edgex.md#connectionselector)
-| topic         | true     | The topic to be published. The topic is static across all messages. To use dynamic topic, leave this empty and specify the topicPrefix property. Only one of the topic and topicPrefix properties can be specified. If both are not specified, then use default topic value ``application``. |
-| topicPrefix         | true     | The prefix of a dynamic topic to be published. The topic will become a concatenation of `$topicPrefix/$profileName/$deviceName/$sourceName`. |
-| contentType   | true     | The content type of message to be published. If not specified, then use the default value ``application/json``. |
-| messageType   | true     | The EdgeX message model type. To publish the message as an event like EdgeX application service, use `event`. Otherwise, to publish the message as an event request like EdgeX device service or core data service, use `request`. If not specified, then use the default value ``event``. |
-| metadata      | true     | The property is a field name that allows user to specify a field name of SQL  select clause,  the field name should use ``meta(*) AS xxx``  to select all of EdgeX metadata from message. |
-| profileName    | true     | Allows user to specify the profile name in the event structure that are sent from eKuiper. The profileName in the meta take precedence if specified. |
-| deviceName    | true     | Allows user to specify the device name in the event structure that are sent from eKuiper. The deviceName in the meta take precedence if specified.  |
-| sourceName    | true     | Allows user to specify the source name in the event structure that are sent from eKuiper. The sourceName in the meta take precedence if specified. |
-| optional      | true     | If ``mqtt`` message bus type is specified, then some optional values can be specified. Please refer to below for supported optional supported configurations. |
+| Property name      | Optional | Description                                                                                                                                                                                                                                                                                  |
+|--------------------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| type               | true     | The message bus type, three types of message buses are supported, ``zero``, ``mqtt`` and ``redis``, and ``redis`` is the default value.                                                                                                                                                      |
+| protocol           | true     | The protocol. If it's not specified, then use default value ``redis``.                                                                                                                                                                                                                       |
+| host               | true     | The host of message bus. If not specified, then use default value ``localhost``.                                                                                                                                                                                                             |
+| port               | true     | The port of message bus. If not specified, then use default value ``6379``.                                                                                                                                                                                                                  |
+| connectionSelector | true     | reuse the connection to EdgeX message bus. [more info](../sources/edgex.md#connectionselector)                                                                                                                                                                                               |
+| topic              | true     | The topic to be published. The topic is static across all messages. To use dynamic topic, leave this empty and specify the topicPrefix property. Only one of the topic and topicPrefix properties can be specified. If both are not specified, then use default topic value ``application``. |
+| topicPrefix        | true     | The prefix of a dynamic topic to be published. The topic will become a concatenation of `$topicPrefix/$profileName/$deviceName/$sourceName`.                                                                                                                                                 |
+| contentType        | true     | The content type of message to be published. If not specified, then use the default value ``application/json``.                                                                                                                                                                              |
+| messageType        | true     | The EdgeX message model type. To publish the message as an event like EdgeX application service, use `event`. Otherwise, to publish the message as an event request like EdgeX device service or core data service, use `request`. If not specified, then use the default value ``event``.   |
+| metadata           | true     | The property is a field name that allows user to specify a field name of SQL  select clause,  the field name should use ``meta(*) AS xxx``  to select all of EdgeX metadata from message.                                                                                                    |
+| profileName        | true     | Allows user to specify the profile name in the event structure that are sent from eKuiper. The profileName in the meta take precedence if specified.                                                                                                                                         |
+| deviceName         | true     | Allows user to specify the device name in the event structure that are sent from eKuiper. The deviceName in the meta take precedence if specified.                                                                                                                                           |
+| sourceName         | true     | Allows user to specify the source name in the event structure that are sent from eKuiper. The sourceName in the meta take precedence if specified.                                                                                                                                           |
+| optional           | true     | If ``mqtt`` message bus type is specified, then some optional values can be specified. Please refer to below for supported optional supported configurations.                                                                                                                                |
 
 Below optional configurations are supported, please check MQTT specification for the detailed information.
 
diff --git a/docs/en_US/rules/sinks/memory.md b/docs/en_US/rules/sinks/memory.md
index 805834ce..af3dd074 100644
--- a/docs/en_US/rules/sinks/memory.md
+++ b/docs/en_US/rules/sinks/memory.md
@@ -2,9 +2,9 @@
 
 The action is used to flush the result into an in-memory topic so that it can be consumed by the [memory source](../sources/memory.md). The topic is like pubsub topic such as mqtt, so that there could be multiple memory sinks which publish to the same topic and multiple memory sources which subscribe to the same topic. The typical usage for memory action is to form [rule pipelines](../rule_pipeline.md).
 
-| Property name      | Optional | Description                                                  |
-| ------------------ | -------- | ------------------------------------------------------------ |
-| topic              | false    | The in-memory topic, such as `analysis/result`                    |
+| Property name | Optional | Description                                    |
+|---------------|----------|------------------------------------------------|
+| topic         | false    | The in-memory topic, such as `analysis/result` |
 
 Below is a sample memory action configuration:
 
@@ -21,7 +21,7 @@ Below is another sample for dynamic topic action:
 ```json
 {
   "memory": {
-    "topic": "$.topic"
+    "topic": "{{.topic}}"
   }
 }
 ```
\ No newline at end of file
diff --git a/docs/en_US/rules/sinks/mqtt.md b/docs/en_US/rules/sinks/mqtt.md
index 7527ef09..f649798b 100644
--- a/docs/en_US/rules/sinks/mqtt.md
+++ b/docs/en_US/rules/sinks/mqtt.md
@@ -2,21 +2,21 @@
 
 The action is used for publish output message into an MQTT server. 
 
-| Property name      | Optional | Description                                                  |
-| ------------------ | -------- | ------------------------------------------------------------ |
-| server             | false    | The broker address of the MQTT server, such as `tcp://127.0.0.1:1883` |
-| topic              | false    | The MQTT topic, such as `analysis/result`                    |
-| clientId           | true     | The client id for MQTT connection. If not specified, an uuid will be used |
-| protocolVersion    | true     | MQTT protocol version. 3.1 (also refer as MQTT 3) or 3.1.1 (also refer as MQTT 4).  If not specified, the default value is 3.1. |
-| qos                | true     | The QoS for message delivery. Only int type value 0 or 1 or 2. |
-| username           | true     | The username for the connection.                             |
-| password           | true     | The password for the connection.                             |
+| Property name      | Optional | Description                                                                                                                                                                                                                                                                                                                                               |
+|--------------------|----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| server             | false    | The broker address of the MQTT server, such as `tcp://127.0.0.1:1883`                                                                                                                                                                                                                                                                                     |
+| topic              | false    | The MQTT topic, such as `analysis/result`                                                                                                                                                                                                                                                                                                                 |
+| clientId           | true     | The client id for MQTT connection. If not specified, an uuid will be used                                                                                                                                                                                                                                                                                 |
+| protocolVersion    | true     | MQTT protocol version. 3.1 (also refer as MQTT 3) or 3.1.1 (also refer as MQTT 4).  If not specified, the default value is 3.1.                                                                                                                                                                                                                           |
+| qos                | true     | The QoS for message delivery. Only int type value 0 or 1 or 2.                                                                                                                                                                                                                                                                                            |
+| username           | true     | The username for the connection.                                                                                                                                                                                                                                                                                                                          |
+| password           | true     | The password for the connection.                                                                                                                                                                                                                                                                                                                          |
 | certificationPath  | true     | The certification path. It can be an absolute path, or a relative path. If it is an relative path, then the base path is where you excuting the `kuiperd` command. For example, if you run `bin/kuiperd` from `/var/kuiper`, then the base path is `/var/kuiper`; If you run `./kuiperd` from `/var/kuiper/bin`, then the base path is `/var/kuiper/bin`. |
-| privateKeyPath     | true     | The private key path. It can be either absolute path, or relative path, which is similar to use of certificationPath. |
-| rootCaPath     | true     | The location of root ca path. It can be an absolute path, or a relative path, which is similar to use of certificationPath. |
-| insecureSkipVerify | true     | If InsecureSkipVerify is `true`, TLS accepts any certificate presented by the server and any host name in that certificate.  In this mode, TLS is susceptible to man-in-the-middle attacks. The default value is `false`. The configuration item can only be used with TLS connections. |
-| retained           | true     | If retained is `true`,The broker stores the last retained message and the corresponding QoS for that topic.The default value is `false`.
-| connectionSelector | true     | reuse the connection to mqtt broker. [more info](../sources/mqtt.md#connectionselector)
+| privateKeyPath     | true     | The private key path. It can be either absolute path, or relative path, which is similar to use of certificationPath.                                                                                                                                                                                                                                     |
+| rootCaPath         | true     | The location of root ca path. It can be an absolute path, or a relative path, which is similar to use of certificationPath.                                                                                                                                                                                                                               |
+| insecureSkipVerify | true     | If InsecureSkipVerify is `true`, TLS accepts any certificate presented by the server and any host name in that certificate.  In this mode, TLS is susceptible to man-in-the-middle attacks. The default value is `false`. The configuration item can only be used with TLS connections.                                                                   |
+| retained           | true     | If retained is `true`,The broker stores the last retained message and the corresponding QoS for that topic.The default value is `false`.                                                                                                                                                                                                                  |
+| connectionSelector | true     | reuse the connection to mqtt broker. [more info](../sources/mqtt.md#connectionselector)                                                                                                                                                                                                                                                                   | 
 
 Below is sample configuration for connecting to Azure IoT Hub by using SAS authentication.
 ```json
@@ -53,13 +53,13 @@ Below is another sample configuration for connecting to AWS IoT by using certifi
 
 ## Dynamic Topic
 
-If the result data contains the topic name, we can use it as the property of the mqtt action to achieve dynamic topic support. Assume the selected data has a field named `mytopic`, we can use jsonpath syntax to set it as the property value for `topic` as below:
+If the result data contains the topic name, we can use it as the property of the mqtt action to achieve dynamic topic support. Assume the selected data has a field named `mytopic`, we can use data template syntax to set it as the property value for `topic` as below:
 
 ```json
     {
       "mqtt": {
         "server": "ssl://xyz-ats.iot.us-east-1.amazonaws.com:8883",
-        "topic": "$.mytopic",
+        "topic": "{{.mytopic}}",
         "qos": 1,
         "clientId": "demo_001",
         "certificationPath": "keys/d3807d9fa5-certificate.pem",
diff --git a/docs/en_US/rules/sinks/nop.md b/docs/en_US/rules/sinks/nop.md
index ea40a44f..0c1e19be 100644
--- a/docs/en_US/rules/sinks/nop.md
+++ b/docs/en_US/rules/sinks/nop.md
@@ -2,8 +2,8 @@
 
 The action is an Nop sink, the result sent to this sink will be ignored. If specify the `log` property to `true`, then the result will be saved into log file, the log file is at `$eKuiper_install/log/stream.log` by default.
 
-| Property name      | Optional | Description                                                  |
-| ------------------ | -------- | ------------------------------------------------------------ |
-| log             | true | true/false - print the sink result to log or not. By default is `false`, that will not print the result to log file. |
+| Property name | Optional | Description                                                                                                          |
+|---------------|----------|----------------------------------------------------------------------------------------------------------------------|
+| log           | true     | true/false - print the sink result to log or not. By default is `false`, that will not print the result to log file. |
 
 
diff --git a/docs/en_US/rules/sinks/rest.md b/docs/en_US/rules/sinks/rest.md
index a5c6e25c..ed1ae398 100644
--- a/docs/en_US/rules/sinks/rest.md
+++ b/docs/en_US/rules/sinks/rest.md
@@ -2,18 +2,18 @@
 
 The action is used for publish output message into a RESTful API.
 
-| Property name     | Optional | Description                                                  |
-| ----------------- | -------- | ------------------------------------------------------------ |
-| method            | true    | The HTTP method for the RESTful API. It is a case insensitive string whose value is among "get", "post", "put", "patch", "delete" and "head". The default value is "get". |
-| url             | false    | The RESTful API endpoint, such as ``https://www.example.com/api/dummy``                  |
-| bodyType          | true     | The type of the body. Currently, these types are supported: "none", "json", "text", "html", "xml", "javascript" and "form". For "get" and "head", no body is required so the default value is "none". For other http methods, the default value is "json" For "html", "xml" and "javascript", the dataTemplate must be carefully set up to make sure the format is correct. |
-| timeout   | true     | The timeout (milliseconds) for a HTTP request, defaults to 5000 ms |
-| headers            | true     | The additional headers to be set for the HTTP request. |
-| debugResp | true | Control if print the response information into the console. If set it to `true`, then print response; If set to `false`, then skip print log. The default is `false`. |
-| certificationPath  | true     | The certification path. It can be an absolute path, or a relative path. If it is an relative path, then the base path is where you excuting the `kuiperd` command. For example, if you run `bin/kuiperd` from `/var/kuiper`, then the base path is `/var/kuiper`; If you run `./kuiperd` from `/var/kuiper/bin`, then the base path is `/var/kuiper/bin`. |
-| privateKeyPath     | true     | The private key path. It can be either absolute path, or relative path, which is similar to use of certificationPath. |
-| rootCaPath     | true     | The location of root ca path. It can be an absolute path, or a relative path, which is similar to use of certificationPath. |
-| insecureSkipVerify | true | Control if to skip the certification verification. If it is set to `true`, then skip certification verification; Otherwise, verify the certification. The default value is `true`. |
+| Property name      | Optional | Description                                                                                                                                                                                                                                                                                                                                                                 |
+|--------------------|----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| method             | true     | The HTTP method for the RESTful API. It is a case insensitive string whose value is among "get", "post", "put", "patch", "delete" and "head". The default value is "get".                                                                                                                                                                                                   |
+| url                | false    | The RESTful API endpoint, such as ``https://www.example.com/api/dummy``                                                                                                                                                                                                                                                                                                     |
+| bodyType           | true     | The type of the body. Currently, these types are supported: "none", "json", "text", "html", "xml", "javascript" and "form". For "get" and "head", no body is required so the default value is "none". For other http methods, the default value is "json" For "html", "xml" and "javascript", the dataTemplate must be carefully set up to make sure the format is correct. |
+| timeout            | true     | The timeout (milliseconds) for a HTTP request, defaults to 5000 ms                                                                                                                                                                                                                                                                                                          |
+| headers            | true     | The additional headers to be set for the HTTP request.                                                                                                                                                                                                                                                                                                                      |
+| debugResp          | true     | Control if print the response information into the console. If set it to `true`, then print response; If set to `false`, then skip print log. The default is `false`.                                                                                                                                                                                                       |
+| certificationPath  | true     | The certification path. It can be an absolute path, or a relative path. If it is an relative path, then the base path is where you excuting the `kuiperd` command. For example, if you run `bin/kuiperd` from `/var/kuiper`, then the base path is `/var/kuiper`; If you run `./kuiperd` from `/var/kuiper/bin`, then the base path is `/var/kuiper/bin`.                   |
+| privateKeyPath     | true     | The private key path. It can be either absolute path, or relative path, which is similar to use of certificationPath.                                                                                                                                                                                                                                                       |
+| rootCaPath         | true     | The location of root ca path. It can be an absolute path, or a relative path, which is similar to use of certificationPath.                                                                                                                                                                                                                                                 |
+| insecureSkipVerify | true     | Control if to skip the certification verification. If it is set to `true`, then skip certification verification; Otherwise, verify the certification. The default value is `true`.                                                                                                                                                                                          |
 
 ::: v-pre
 REST service usually requires a specific data format. That can be imposed by the common sink property `dataTemplate`. Please check the [data template](../overview.md#data-template). Below is a sample configuration for connecting to Edgex Foundry core command. The dataTemplate `{{.key}}` means it will print out the value of key, that is result[key]. So the template here is to select only field ``key`` in the result and change the field name to ``newKey``. `sendSingle` is another common property. Set to true means that if the result is an array, each element will be sent individually.
@@ -71,7 +71,7 @@ There are many scenarios that we need to sink to dynamic url and configurations
 }
 ```
 
-Then in the action, we set the `method` and `url` to be the value of the result by using jsonpath syntax as below:
+Then in the action, we set the `method` and `url` to be the value of the result by using data template syntax as below:
 
 ```json
 {"id": "rest2",
@@ -83,9 +83,9 @@ Then in the action, we set the `method` and `url` to be the value of the result
         "dataTemplate": "insert into mqtt.kuiper values (now, {{.temperature}}, {{.humidity}})", 
         "debugResp": true,
         "headers": {"Authorization": "Basic cm9vdDp0YW9zZGF0YQ=="},
-        "method": "$.method",
+        "method": "{{.method}}",
         "sendSingle": true,
-        "url": "$.url"
+        "url": "{{.url}}"
       }
     }
   ]
diff --git a/docs/zh_CN/rules/overview.md b/docs/zh_CN/rules/overview.md
index 1289713c..97d8c3b4 100644
--- a/docs/zh_CN/rules/overview.md
+++ b/docs/zh_CN/rules/overview.md
@@ -24,12 +24,12 @@
 
 ## 参数
 
-| 参数名 | 是否可选 | 说明                |
-| ------------- | -------- | ------------------------------------------------------------ |
-| id | 否  | 规则 id |
-| sql        | 否  | 为规则运行的 sql 查询 |
-| actions           | 否   | Sink 动作数组 |
-| options           | 是       | 选项图     |
+| 参数名     | 是否可选 | 说明            |
+|---------|------|---------------|
+| id      | 否    | 规则 id         |
+| sql     | 否    | 为规则运行的 sql 查询 |
+| actions | 否    | Sink 动作数组     |
+| options | 是    | 选项图           |
 
 ### id
 
@@ -43,16 +43,16 @@
 
 当前的选项包括：
 
-| 选项名             | 类型和默认值 | 说明                                                         |
-| ------------------ | ------------ | ------------------------------------------------------------ |
-| isEventTime        | bool:false   | 使用事件时间还是将时间用作事件的时间戳。 如果使用事件时间，则将从有效负载中提取时间戳。 必须通过 [stream](../sqls/streams.md) 定义指定时间戳记。 |
-| lateTolerance      | int64:0      | 在使用事件时间窗口时，可能会出现元素延迟到达的情况。 LateTolerance 可以指定在删除元素之前可以延迟多少时间（单位为 ms）。 默认情况下，该值为0，表示后期元素将被删除。 |
-| concurrency        | int: 1       | 一条规则运行时会根据 sql 语句分解成多个 plan 运行。该参数设置每个 plan 运行的线程数。该参数值大于1时，消息处理顺序可能无法保证。 |
-| bufferLength       | int: 1024    | 指定每个 plan 可缓存消息数。若缓存消息数超过此限制，plan 将阻塞消息接收，直到缓存消息被消费使得缓存消息数目小于限制为止。此选项值越大，则消息吞吐能力越强，但是内存占用也会越多。 |
-| sendMetaToSink     | bool:false   | 指定是否将事件的元数据发送到目标。 如果为 true，则目标可以获取元数据信息。 |
-| sendError  | bool: true | 指定是否将运行时错误发送到目标。如果为 true，则错误会在整个流中传递直到目标。否则，错误会被忽略，仅打印到日志中。 |
-| qos                | int:0        | 指定流的 qos。 值为0对应最多一次； 1对应至少一次，2对应恰好一次。 如果 qos 大于0，将激活检查点机制以定期保存状态，以便可以从错误中恢复规则。 |
-| checkpointInterval | int:300000   | 指定触发检查点的时间间隔（单位为 ms）。 仅当 qos 大于0时才有效。 |
+| 选项名                | 类型和默认值     | 说明                                                                                             |
+|--------------------|------------|------------------------------------------------------------------------------------------------|
+| isEventTime        | bool:false | 使用事件时间还是将时间用作事件的时间戳。 如果使用事件时间，则将从有效负载中提取时间戳。 必须通过 [stream](../sqls/streams.md) 定义指定时间戳记。       |
+| lateTolerance      | int64:0    | 在使用事件时间窗口时，可能会出现元素延迟到达的情况。 LateTolerance 可以指定在删除元素之前可以延迟多少时间（单位为 ms）。 默认情况下，该值为0，表示后期元素将被删除。   |
+| concurrency        | int: 1     | 一条规则运行时会根据 sql 语句分解成多个 plan 运行。该参数设置每个 plan 运行的线程数。该参数值大于1时，消息处理顺序可能无法保证。                      |
+| bufferLength       | int: 1024  | 指定每个 plan 可缓存消息数。若缓存消息数超过此限制，plan 将阻塞消息接收，直到缓存消息被消费使得缓存消息数目小于限制为止。此选项值越大，则消息吞吐能力越强，但是内存占用也会越多。 |
+| sendMetaToSink     | bool:false | 指定是否将事件的元数据发送到目标。 如果为 true，则目标可以获取元数据信息。                                                       |
+| sendError          | bool: true | 指定是否将运行时错误发送到目标。如果为 true，则错误会在整个流中传递直到目标。否则，错误会被忽略，仅打印到日志中。                                    |
+| qos                | int:0      | 指定流的 qos。 值为0对应最多一次； 1对应至少一次，2对应恰好一次。 如果 qos 大于0，将激活检查点机制以定期保存状态，以便可以从错误中恢复规则。                 |
+| checkpointInterval | int:300000 | 指定触发检查点的时间间隔（单位为 ms）。 仅当 qos 大于0时才有效。                                                          |
 
 有关 `qos` 和 `checkpointInterval` 的详细信息，请查看[状态和容错](./state_and_fault_tolerance.md)。
 
@@ -79,18 +79,18 @@
 
 每个动作可以定义自己的属性。当前有以下的公共属性:
 
-| 属性名 | 类型和默认值 | 描述                                                  |
-| ------------- | -------- | ------------------------------------------------------------ |
-| concurrency | int: 1   | 设置运行的线程数。该参数值大于1时，消息发出的顺序可能无法保证。 |
-| bufferLength | int: 1024   | 设置可缓存消息数目。若缓存消息数超过此限制，sink将阻塞消息接收，直到缓存消息被消费使得缓存消息数目小于限制为止。|
-| runAsync        | bool:false   | 设置是否异步运行输出操作以提升性能。请注意，异步运行的情况下，输出结果顺序不能保证。  |
-| retryInterval   | int:1000   | 设置信息发送失败后重试等待时间，单位为毫秒。如果该值的设置 <= 0，那么不会尝试重新发送。 |
-| retryCount | int:3 | 设置信息发送失败后重试次数，如果该值的设置 <= 0，那么不会尝试重新发送。 |
-| cacheLength     | int:1024   | 设置最大消息缓存数量。缓存的消息会一直保留直到消息发送成功。缓存消息将按顺序发送，除非运行在异步或者并发模式下。缓存消息会定期存储到磁盘中。  |
-| cacheSaveInterval  | int:1000   | 设置缓存存储间隔时间。需要注意的是，当规则关闭时，缓存会自动存储。该值越大，则缓存保存开销越小，但系统意外退出时缓存丢失的风险变大。 |
-| omitIfEmpty | bool: false | 如果配置项设置为 true，则当 SELECT 结果为空时，该结果将不提供给目标运算符。 |
-| sendSingle        | true     | 输出消息以数组形式接收，该属性意味着是否将结果一一发送。 如果为false，则输出消息将为`{"result":"${the string of received message}"}`。 例如，`{"result":"[{\"count\":30},"\"count\":20}]"}`。否则，结果消息将与实际字段名称一一对应发送。 对于与上述相同的示例，它将发送 `{"count":30}`，然后发送`{"count":20}`到 RESTful 端点。默认为 false。 |
-| dataTemplate      | true     | [golang 模板](https://golang.org/pkg/html/template)格式字符串，用于指定输出数据格式。 模板的输入是目标消息，该消息始终是映射数组。 如果未指定数据模板，则将数据作为原始输入。 |
+| 属性名               | 类型和默认值      | 描述                                                                                                                                                                                                                                               |
+|-------------------|-------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| concurrency       | int: 1      | 设置运行的线程数。该参数值大于1时，消息发出的顺序可能无法保证。                                                                                                                                                                                                                 |
+| bufferLength      | int: 1024   | 设置可缓存消息数目。若缓存消息数超过此限制，sink将阻塞消息接收，直到缓存消息被消费使得缓存消息数目小于限制为止。                                                                                                                                                                                       |
+| runAsync          | bool:false  | 设置是否异步运行输出操作以提升性能。请注意，异步运行的情况下，输出结果顺序不能保证。                                                                                                                                                                                                       |
+| retryInterval     | int:1000    | 设置信息发送失败后重试等待时间，单位为毫秒。如果该值的设置 <= 0，那么不会尝试重新发送。                                                                                                                                                                                                   |
+| retryCount        | int:3       | 设置信息发送失败后重试次数，如果该值的设置 <= 0，那么不会尝试重新发送。                                                                                                                                                                                                           |
+| cacheLength       | int:1024    | 设置最大消息缓存数量。缓存的消息会一直保留直到消息发送成功。缓存消息将按顺序发送，除非运行在异步或者并发模式下。缓存消息会定期存储到磁盘中。                                                                                                                                                                           |
+| cacheSaveInterval | int:1000    | 设置缓存存储间隔时间。需要注意的是，当规则关闭时，缓存会自动存储。该值越大，则缓存保存开销越小，但系统意外退出时缓存丢失的风险变大。                                                                                                                                                                               |
+| omitIfEmpty       | bool: false | 如果配置项设置为 true，则当 SELECT 结果为空时，该结果将不提供给目标运算符。                                                                                                                                                                                                     |
+| sendSingle        | true        | 输出消息以数组形式接收，该属性意味着是否将结果一一发送。 如果为false，则输出消息将为`{"result":"${the string of received message}"}`。 例如，`{"result":"[{\"count\":30},"\"count\":20}]"}`。否则，结果消息将与实际字段名称一一对应发送。 对于与上述相同的示例，它将发送 `{"count":30}`，然后发送`{"count":20}`到 RESTful 端点。默认为 false。 |
+| dataTemplate      | true        | [golang 模板](https://golang.org/pkg/html/template)格式字符串，用于指定输出数据格式。 模板的输入是目标消息，该消息始终是映射数组。 如果未指定数据模板，则将数据作为原始输入。                                                                                                                                  |
 
 ### 数据模板
 
@@ -171,7 +171,7 @@ eKuiper 扩展了几个可以在模版中使用的函数。
 
 ### 动态属性
 
-有些情况下，用户需要按照数据把结果发送到不同的目标中。例如，根据收到的数据，把计算结果发到不同的 mqtt 主题中。使用基于 jsonpath 格式的动态属性，可以实现这样的功能。在以下的例子中，目标的 topic 属性是一个 jsonpath 格式的字符串从而在运行时会将消息发送到动态的主题中。 
+有些情况下，用户需要按照数据把结果发送到不同的目标中。例如，根据收到的数据，把计算结果发到不同的 mqtt 主题中。使用基于[数据模板](#数据模板)格式的动态属性，可以实现这样的功能。在以下的例子中，目标的 topic 属性是一个数据模板格式的字符串从而在运行时会将消息发送到动态的主题中。 
 
 ```json
 {
@@ -180,10 +180,10 @@ eKuiper 扩展了几个可以在模版中使用的函数。
   "actions": [{
     "mqtt": {
       "sendSingle": true,
-      "topic": "$.topic"
+      "topic": "prefix/{{.topic}}"
     }
   }]
 }
 ```
 
-需要注意的是，上例中的 `sendSingle` 属性已设置。在默认情况下，目标接收到的是数组，使用的 jsonpath 需要采用 `$[0].topic`。
\ No newline at end of file
+需要注意的是，上例中的 `sendSingle` 属性已设置。在默认情况下，目标接收到的是数组，使用的 jsonpath 需要采用 `{{index . 0 \"topic\"}}`。
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sinks/edgex.md b/docs/zh_CN/rules/sinks/edgex.md
index 3261839e..3dac6a51 100644
--- a/docs/zh_CN/rules/sinks/edgex.md
+++ b/docs/zh_CN/rules/sinks/edgex.md
@@ -6,22 +6,22 @@
 
 **另外，如果你需要在别的主机上对你的端口可以进行访问，你需要在开始运行 eKuiper 服务之前，把端口号映射到主机上。**
 
-| 名称        | 可选 | Description                                                  |
-| ----------- | -------- | ------------------------------------------------------------ |
-| type          | 是    | 消息总线类型，目前支持三种类型的消息总线， `redis`, `zero` 或者 `mqtt`，其中 `redis` 为缺省类型。 |
-| protocol    | 是     | 协议，如未指定，使用缺省值 `tcp` 。  |
-| host        | 是    | 消息总线主机地址，使用缺省值 `*` 。                    |
-| port        | 是    | 消息总线端口号。 如未指定，使用缺省值 `5563` 。              |
-| connectionSelector | 是     | 重用到 EdgeX 消息总线的连接，详细信息，[请参考](../sources/edgex.md#connectionselector)
-| topic       | 是    | 发布的主题名称。该主题为固定值。若不同的消息需要动态指定主题，则将该属性置空，并设置 topicPrefix 属性。这两个属性只能设置一个。若两者都未设置，则使用缺省主题 `application` 。          |
-| topicPrefix | 是     | 发布的主题的前缀。发送的主题将采用动态拼接，格式为`$topicPrefix/$profileName/$deviceName/$sourceName` 。|
-| contentType | 是    | 发布消息的内容类型，如未指定，使用缺省值 `application/json` 。|
-| messageType   | 是   | EdgeX 消息模型类型。若要将消息发送为类似 apllication service 的 event 类型，则应设置为 `event`。否则，若要将消息发送为类似 device service 或者 core data service 的 event request 类型，则应设置为 `request`。如未指定，使用缺省值 ``event`` 。|
-| metadata    | 是    | 该属性为一个字段名称，该字段是 SQL SELECT 子句的一个字段名称，这个字段应该类似于 `meta(*) AS xxx` ，用于选出消息中所有的 EdgeX 元数据 。 |
-| profileName  | 是    | 允许用户指定 Profile 名称，该名称将作为从 eKuiper 中发送出来的 Event 结构体的 profile 名称。若在 metadata 中设置了 profileName 将会优先采用。|
-| deviceName  | 是    | 允许用户指定设备名称，该名称将作为从 eKuiper 中发送出来的 Event 结构体的设备名称。若在 metadata 中设置了 deviceName 将会优先采用。 |
-| sourceName    | 是   | 允许用户指定源名称，该名称将作为从 eKuiper 中发送出来的 Event 结构体的源名称。若在 metadata 中设置了 sourceName 将会优先采用。 |
-| optional      | 是    | 如果指定了 `mqtt` 消息总线，那么还可以指定一下可选的值。请参考以下可选的支持的配置类型。 |
+| 名称                 | 可选  | Description                                                                                                                                                                      |
+|--------------------|-----|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| type               | 是   | 消息总线类型，目前支持三种类型的消息总线， `redis`, `zero` 或者 `mqtt`，其中 `redis` 为缺省类型。                                                                                                                |
+| protocol           | 是   | 协议，如未指定，使用缺省值 `tcp` 。                                                                                                                                                            |
+| host               | 是   | 消息总线主机地址，使用缺省值 `*` 。                                                                                                                                                             |
+| port               | 是   | 消息总线端口号。 如未指定，使用缺省值 `5563` 。                                                                                                                                                     |
+| connectionSelector | 是   | 重用到 EdgeX 消息总线的连接，详细信息，[请参考](../sources/edgex.md#connectionselector)                                                                                                             |
+| topic              | 是   | 发布的主题名称。该主题为固定值。若不同的消息需要动态指定主题，则将该属性置空，并设置 topicPrefix 属性。这两个属性只能设置一个。若两者都未设置，则使用缺省主题 `application` 。                                                                            |
+| topicPrefix        | 是   | 发布的主题的前缀。发送的主题将采用动态拼接，格式为`$topicPrefix/$profileName/$deviceName/$sourceName` 。                                                                                                   |
+| contentType        | 是   | 发布消息的内容类型，如未指定，使用缺省值 `application/json` 。                                                                                                                                        |
+| messageType        | 是   | EdgeX 消息模型类型。若要将消息发送为类似 apllication service 的 event 类型，则应设置为 `event`。否则，若要将消息发送为类似 device service 或者 core data service 的 event request 类型，则应设置为 `request`。如未指定，使用缺省值 ``event`` 。 |
+| metadata           | 是   | 该属性为一个字段名称，该字段是 SQL SELECT 子句的一个字段名称，这个字段应该类似于 `meta(*) AS xxx` ，用于选出消息中所有的 EdgeX 元数据 。                                                                                          |
+| profileName        | 是   | 允许用户指定 Profile 名称，该名称将作为从 eKuiper 中发送出来的 Event 结构体的 profile 名称。若在 metadata 中设置了 profileName 将会优先采用。                                                                              |
+| deviceName         | 是   | 允许用户指定设备名称，该名称将作为从 eKuiper 中发送出来的 Event 结构体的设备名称。若在 metadata 中设置了 deviceName 将会优先采用。                                                                                             |
+| sourceName         | 是   | 允许用户指定源名称，该名称将作为从 eKuiper 中发送出来的 Event 结构体的源名称。若在 metadata 中设置了 sourceName 将会优先采用。                                                                                               |
+| optional           | 是   | 如果指定了 `mqtt` 消息总线，那么还可以指定一下可选的值。请参考以下可选的支持的配置类型。                                                                                                                                 |
 
 以下为支持的可选的配置列表，您可以参考 MQTT 协议规范来获取更详尽的信息。
 
diff --git a/docs/zh_CN/rules/sinks/memory.md b/docs/zh_CN/rules/sinks/memory.md
index ff1968f6..1f5fe204 100644
--- a/docs/zh_CN/rules/sinks/memory.md
+++ b/docs/zh_CN/rules/sinks/memory.md
@@ -21,7 +21,7 @@
 ```json
 {
   "memory": {
-    "topic": "$.topic"
+    "topic": "{{.topic}}"
   }
 }
 ```
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sinks/mqtt.md b/docs/zh_CN/rules/sinks/mqtt.md
index 8b182fb4..2d4d4a94 100755
--- a/docs/zh_CN/rules/sinks/mqtt.md
+++ b/docs/zh_CN/rules/sinks/mqtt.md
@@ -2,21 +2,22 @@
 
 该操作用于将输出消息发布到 MQTT 服务器中。
 
-| 属性名称 | 是否可选 | 说明                                          |
-| ------------- | -------- | ---------------------------------------------------- |
-| server        | 否    | MQTT  服务器地址，例如 `tcp://127.0.0.1:1883` |
-| topic          | 否    | MQTT 主题，例如 `analysis/result` , 也可设置为动态属性，例如 `$.col`, 将会把结果中的 col 列的值作为主题                   |
-| clientId      | 是     | MQTT 连接的客户端 ID。 如果未指定，将使用一个 uuid |
-| protocolVersion   | 是    | MQTT 协议版本。3.1 (也被称为 MQTT 3) 或者 3.1.1 (也被称为 MQTT 4)。 如果未指定，缺省值为 3.1。 |
-| qos               | 是    | 消息转发的服务质量                               |
-| username          | 是    | 连接用户名                            |
-| password          | 是    | 连接密码                             |
-| certificationPath | 是    | 证书路径。可以为绝对路径，也可以为相对路径。如果指定的是相对路径，那么父目录为执行 `kuiperd` 命令的路径。比如，如果你在 `/var/kuiper` 中运行 `bin/kuiperd` ，那么父目录为 `/var/kuiper`; 如果运行从 `/var/kuiper/bin` 中运行`./kuiperd`，那么父目录为 `/var/kuiper/bin`。 |
-| privateKeyPath    | 是    | 私钥路径。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。 |
-| rootCaPath    | 是    | 根证书路径，用以验证服务器证书。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。 |
-| insecureSkipVerify | 是     | 如果 InsecureSkipVerify 设置为 `true`, TLS接受服务器提供的任何证书以及该证书中的任何主机名。 在这种模式下，TLS容易受到中间人攻击。默认值为`false`。配置项只能用于TLS连接。|
-| retained           | 是     | 如果 retained 设置为 `true`,Broker会存储每个Topic的最后一条保留消息及其Qos。默认值是 `false`   
-| connectionSelector | 是     | 重用到 MQTT Broker 的连接，详细信息，[请参考](../sources/mqtt.md#connectionselector)
+| 属性名称               | 是否可选 | 说明                                                                                                                                                                                        |
+|--------------------|------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| server             | 否    | MQTT  服务器地址，例如 `tcp://127.0.0.1:1883`                                                                                                                                                     |
+| topic              | 否    | MQTT 主题，例如 `analysis/result` , 也可设置为动态属性，例如 `$.col`, 将会把结果中的 col 列的值作为主题                                                                                                                  |
+| clientId           | 是    | MQTT 连接的客户端 ID。 如果未指定，将使用一个 uuid                                                                                                                                                          |
+| protocolVersion    | 是    | MQTT 协议版本。3.1 (也被称为 MQTT 3) 或者 3.1.1 (也被称为 MQTT 4)。 如果未指定，缺省值为 3.1。                                                                                                                       |
+| qos                | 是    | 消息转发的服务质量                                                                                                                                                                                 |
+| username           | 是    | 连接用户名                                                                                                                                                                                     |
+| password           | 是    | 连接密码                                                                                                                                                                                      |
+| certificationPath  | 是    | 证书路径。可以为绝对路径，也可以为相对路径。如果指定的是相对路径，那么父目录为执行 `kuiperd` 命令的路径。比如，如果你在 `/var/kuiper` 中运行 `bin/kuiperd` ，那么父目录为 `/var/kuiper`; 如果运行从 `/var/kuiper/bin` 中运行`./kuiperd`，那么父目录为 `/var/kuiper/bin`。 |
+| privateKeyPath     | 是    | 私钥路径。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。                                                                                                                                    |
+| rootCaPath         | 是    | 根证书路径，用以验证服务器证书。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。                                                                                                                         |
+| insecureSkipVerify | 是    | 如果 InsecureSkipVerify 设置为 `true`, TLS接受服务器提供的任何证书以及该证书中的任何主机名。 在这种模式下，TLS容易受到中间人攻击。默认值为`false`。配置项只能用于TLS连接。                                                                              |
+| retained           | 是    | 如果 retained 设置为 `true`,Broker会存储每个Topic的最后一条保留消息及其Qos。默认值是 `false`                                                                                                                        |
+| connectionSelector | 是    | 重用到 MQTT Broker 的连接，详细信息，[请参考](../sources/mqtt.md#connectionselector)                                                                                                                     |
+
 以下为使用 SAS 连接到 Azure IoT Hub 的样例。
 ```json
     {
@@ -51,13 +52,13 @@
 
 ## 动态主题
 
-若结果数据中包含主题内容，可以将其作为主题属性，从而实现动态主题的需求。假设 SQL 选出的数据包含 `mytopic`, 则可以使用 jsonpath 语法将其设置为 `topic` 属性的值，如下所示：
+若结果数据中包含主题内容，可以将其作为主题属性，从而实现动态主题的需求。假设 SQL 选出的数据包含 `mytopic`, 则可以使用数据模板的语法将其设置为 `topic` 属性的值，如下所示：
 
 ```json
     {
       "mqtt": {
         "server": "ssl://xyz-ats.iot.us-east-1.amazonaws.com:8883",
-        "topic": "$.mytopic",
+        "topic": "{{.mytopic}}",
         "qos": 1,
         "clientId": "demo_001",
         "certificationPath": "keys/d3807d9fa5-certificate.pem",
diff --git a/docs/zh_CN/rules/sinks/nop.md b/docs/zh_CN/rules/sinks/nop.md
index 7fadb02f..ebada56b 100644
--- a/docs/zh_CN/rules/sinks/nop.md
+++ b/docs/zh_CN/rules/sinks/nop.md
@@ -2,8 +2,8 @@
 
 该 action 是一个空操作目标，所有发送到此的结果将被忽略。如果指定 `log` 属性为 `true`，那么结果将会保存到日志文件，日志文件缺省保存在  `$eKuiper_install/log/stream.log`。
 
-| 属性名称  | 是否可选 | 说明                                                  |
-| ------------------ | -------- | ------------------------------------------------------------ |
-| log             | true | true/false - 是否将结果打印到日志。缺省为 `false`，这种情况下将不会打印到日志文件。 |
+| 属性名称 | 是否可选 | 说明                                                   |
+|------|------|------------------------------------------------------|
+| log  | true | true/false - 是否将结果打印到日志。缺省为 `false`，这种情况下将不会打印到日志文件。 |
 
 
diff --git a/docs/zh_CN/rules/sinks/rest.md b/docs/zh_CN/rules/sinks/rest.md
index d654fc25..d0f6467e 100644
--- a/docs/zh_CN/rules/sinks/rest.md
+++ b/docs/zh_CN/rules/sinks/rest.md
@@ -2,18 +2,18 @@
 
 该动作用于将输出消息发布到 RESTful API 中。
 
-| 属性名称   | 是否可选 | 说明                                                  |
-| ----------------- | -------- | ------------------------------------------------------------ |
-| method            | 是    | RESTful API 的 HTTP 方法。 这是一个不区分大小写的字符串，其值范围为"get"，"post"，"put"，"patch"，"delete" 和 "head"。 默认值为 "get"，支持动态获取。 |
-| url             | 否    | RESTful API 终端地址，例如 `https://www.example.com/api/dummy`，支持动态获取。 |
-| bodyType          | 是    | 消息体的类型。 当前，支持以下类型："none", "json", "text", "html", "xml", "javascript"  和 "form"。 对于 "get" 和 "head"，不需要正文，因此默认值为 "none"。 对于其他 http 方法，默认值为 "json"。对于 "html"，"xml" 和 "javascript"，必须仔细设置 dataTemplate 以确保格式正确。支持动态获取。 |
-| timeout   | 是    | HTTP 请求超时的时间（毫秒），默认为5000毫秒 |
-| headers            | 是    | 要为 HTTP 请求设置的其它 HTTP 头。支持动态获取。 |
-| debugResp | 是 | 控制是否将响应信息打印到控制台中。 如果将其设置为 `true`，则打印响应。 如果设置为`false`，则跳过打印日志。 默认值为 `false`。 |
-| certificationPath | 是    | 证书路径。可以为绝对路径，也可以为相对路径。如果指定的是相对路径，那么父目录为执行 `kuiperd` 命令的路径。比如，如果你在 `/var/kuiper` 中运行 `bin/kuiperd` ，那么父目录为 `/var/kuiper`; 如果运行从 `/var/kuiper/bin` 中运行`./kuiperd`，那么父目录为 `/var/kuiper/bin`。 |
-| privateKeyPath    | 是    | 私钥路径。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。 |
-| rootCaPath    | 是    | 根证书路径，用以验证服务器证书。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。 |
-| insecureSkipVerify | 是 | 控制是否跳过证书认证。如果被设置为 `true`，那么跳过证书认证；否则进行证书验证。缺省为 `true`。 |
+| 属性名称               | 是否可选 | 说明                                                                                                                                                                                                                  |
+|--------------------|------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| method             | 是    | RESTful API 的 HTTP 方法。 这是一个不区分大小写的字符串，其值范围为"get"，"post"，"put"，"patch"，"delete" 和 "head"。 默认值为 "get"，支持动态获取。                                                                                                         |
+| url                | 否    | RESTful API 终端地址，例如 `https://www.example.com/api/dummy`，支持动态获取。                                                                                                                                                     |
+| bodyType           | 是    | 消息体的类型。 当前，支持以下类型："none", "json", "text", "html", "xml", "javascript"  和 "form"。 对于 "get" 和 "head"，不需要正文，因此默认值为 "none"。 对于其他 http 方法，默认值为 "json"。对于 "html"，"xml" 和 "javascript"，必须仔细设置 dataTemplate 以确保格式正确。支持动态获取。 |
+| timeout            | 是    | HTTP 请求超时的时间（毫秒），默认为5000毫秒                                                                                                                                                                                          |
+| headers            | 是    | 要为 HTTP 请求设置的其它 HTTP 头。支持动态获取。                                                                                                                                                                                      |
+| debugResp          | 是    | 控制是否将响应信息打印到控制台中。 如果将其设置为 `true`，则打印响应。 如果设置为`false`，则跳过打印日志。 默认值为 `false`。                                                                                                                                         |
+| certificationPath  | 是    | 证书路径。可以为绝对路径，也可以为相对路径。如果指定的是相对路径，那么父目录为执行 `kuiperd` 命令的路径。比如，如果你在 `/var/kuiper` 中运行 `bin/kuiperd` ，那么父目录为 `/var/kuiper`; 如果运行从 `/var/kuiper/bin` 中运行`./kuiperd`，那么父目录为 `/var/kuiper/bin`。                           |
+| privateKeyPath     | 是    | 私钥路径。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。                                                                                                                                                              |
+| rootCaPath         | 是    | 根证书路径，用以验证服务器证书。可以为绝对路径，也可以为相对路径，相对路径的用法与 `certificationPath` 类似。                                                                                                                                                   |
+| insecureSkipVerify | 是    | 控制是否跳过证书认证。如果被设置为 `true`，那么跳过证书认证；否则进行证书验证。缺省为 `true`。                                                                                                                                                              |
 
 ::: v-pre
 REST 服务通常需要特定的数据格式。 这可以由公共目标属性 `dataTemplate` 强制使用。 请参考[数据模板](../overview.md#数据模板)。 以下是用于连接到 Edgex Foundry core 命令的示例配置。dataTemplate`{{.key}}` 表示将打印出键值，即 result [key]。 因此，这里的模板是在结果中仅选择字段 `key` ，并将字段名称更改为 `newKey`。 `sendSingle` 是另一个常见属性。 设置为 true 表示如果结果是数组，则每个元素将单独发送。
@@ -59,7 +59,7 @@ Text mode
 
 ## 设置动态输出参数
 
-很多情况下，我们需要根据结果数据，决定写入的目的地址和参数。在 REST sink 里，`method`, `url`, `bodyType` 和 `headers` 支持动态参数。动态参数可通过 jsonpath 语法配置。接下来，让我们使用动态参数改写上例。假设我们收到了数据中包含了 http 方法和 url 后缀等元数据。我们可以通过改写 SQL 语句，在输出结果中得到这两个值。规则输出的单条数据类似：
+很多情况下，我们需要根据结果数据，决定写入的目的地址和参数。在 REST sink 里，`method`, `url`, `bodyType` 和 `headers` 支持动态参数。动态参数可通过数据模板语法配置。接下来，让我们使用动态参数改写上例。假设我们收到了数据中包含了 http 方法和 url 后缀等元数据。我们可以通过改写 SQL 语句，在输出结果中得到这两个值。规则输出的单条数据类似：
 
 ```json
 {
@@ -70,7 +70,7 @@ Text mode
 }
 ```
 
-在规则 action 中，可以通过 jsonpath 语法取得结果数据作为属性变量。如下例子中，`method` 和 `url` 为动态变量。
+在规则 action 中，可以通过数据模板语法取得结果数据作为属性变量。如下例子中，`method` 和 `url` 为动态变量。
 
 ```json
 {"id": "rest2",
@@ -82,9 +82,9 @@ Text mode
         "dataTemplate": "insert into mqtt.kuiper values (now, {{.temperature}}, {{.humidity}})", 
         "debugResp": true,
         "headers": {"Authorization": "Basic cm9vdDp0YW9zZGF0YQ=="},
-        "method": "$.method",
+        "method": "{{.method}}",
         "sendSingle": true,
-        "url": "$.url"
+        "url": "{{.url}}"
       }
     }
   ]
diff --git a/internal/binder/function/funcs_misc.go b/internal/binder/function/funcs_misc.go
index efd8b845..e7d6d907 100644
--- a/internal/binder/function/funcs_misc.go
+++ b/internal/binder/function/funcs_misc.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -255,7 +255,8 @@ func otherCall(name string, args []interface{}) (interface{}, bool) {
 }
 
 func jsonCall(ctx api.StreamContext, name string, args []interface{}) (interface{}, bool) {
-	result, err := ctx.ParseDynamicProp(args[1].(string), args[0])
+	// TO BE REMOVED prefix { to avoid conflict with regular string
+	result, err := ctx.ParseDynamicProp("{"+args[1].(string), args[0])
 	if err != nil {
 		if name == "json_path_exists" {
 			return false, true
diff --git a/internal/plugin/portable/runtime/function.go b/internal/plugin/portable/runtime/function.go
index fca5f0b7..56c1404f 100644
--- a/internal/plugin/portable/runtime/function.go
+++ b/internal/plugin/portable/runtime/function.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -29,6 +29,7 @@ type PortableFunc struct {
 	symbolName string
 	reg        *PluginMeta
 	dataCh     DataReqChannel
+	isAgg      int // 0 - not calculate yet, 1 - no, 2 - yes
 }
 
 func NewPortableFunc(symbolName string, reg *PluginMeta) (*PortableFunc, error) {
@@ -115,7 +116,9 @@ func (f *PortableFunc) Exec(args []interface{}, ctx api.FunctionContext) (interf
 }
 
 func (f *PortableFunc) IsAggregate() bool {
-	// TODO error handling
+	if f.isAgg > 0 {
+		return f.isAgg > 1
+	}
 	jsonArg, err := encode("IsAggregate", nil)
 	if err != nil {
 		conf.Log.Error(err)
@@ -138,6 +141,11 @@ func (f *PortableFunc) IsAggregate() bool {
 			conf.Log.Errorf("IsAggregate result is not bool, got %s", string(res))
 			return false
 		} else {
+			if r {
+				f.isAgg = 2
+			} else {
+				f.isAgg = 1
+			}
 			return r
 		}
 	} else {
diff --git a/internal/topo/context/default.go b/internal/topo/context/default.go
index db3da6e1..b55cd160 100644
--- a/internal/topo/context/default.go
+++ b/internal/topo/context/default.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -15,15 +15,19 @@
 package context
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/connection"
+	"github.com/lf-edge/ekuiper/internal/topo/transform"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/sirupsen/logrus"
+	"regexp"
 	"strings"
 	"sync"
+	"text/template"
 	"time"
 )
 
@@ -40,7 +44,8 @@ type DefaultContext struct {
 	state    *sync.Map
 	snapshot map[string]interface{}
 	// cache
-	jsonEvalReg sync.Map
+	tpReg sync.Map
+	jpReg sync.Map
 }
 
 func Background() *DefaultContext {
@@ -110,24 +115,45 @@ func (c *DefaultContext) SetError(err error) {
 }
 
 func (c *DefaultContext) ParseDynamicProp(prop string, data interface{}) (interface{}, error) {
-	// If not a json path, just return itself
-	if !strings.HasPrefix(prop, "$") {
-		return prop, nil
-	}
-	var (
-		je  conf.JsonPathEval
-		err error
-	)
-	if raw, ok := c.jsonEvalReg.Load(prop); ok {
-		je = raw.(conf.JsonPathEval)
-	} else {
-		je, err = conf.GetJsonPathEval(prop)
+	re := regexp.MustCompile(`{{(.*?)}}`)
+	if re.Match([]byte(prop)) {
+		var (
+			tp  *template.Template
+			err error
+		)
+		if raw, ok := c.tpReg.Load(prop); ok {
+			tp = raw.(*template.Template)
+		} else {
+			tp, err = transform.GenTp(prop)
+			if err != nil {
+				return fmt.Sprintf("%v", data), err
+			}
+			c.tpReg.Store(prop, tp)
+		}
+		var output bytes.Buffer
+		err = tp.Execute(&output, data)
 		if err != nil {
-			return nil, err
+			return fmt.Sprintf("%v", data), err
 		}
-		c.jsonEvalReg.Store(prop, je)
+		return output.String(), nil
+	} else if strings.HasPrefix(prop, "{$") { //TO BE REMOVED: will be extracted as a new function in the next release
+		var (
+			je  conf.JsonPathEval
+			err error
+		)
+		if raw, ok := c.jpReg.Load(prop); ok {
+			je = raw.(conf.JsonPathEval)
+		} else {
+			je, err = conf.GetJsonPathEval(prop[1:])
+			if err != nil {
+				return nil, err
+			}
+			c.jpReg.Store(prop, je)
+		}
+		return je.Eval(data)
+	} else {
+		return prop, nil
 	}
-	return je.Eval(data)
 }
 
 func (c *DefaultContext) WithMeta(ruleId string, opId string, store api.Store) api.StreamContext {
@@ -136,13 +162,14 @@ func (c *DefaultContext) WithMeta(ruleId string, opId string, store api.Store) a
 		c.GetLogger().Warnf("Initialize context store error for %s: %s", opId, err)
 	}
 	return &DefaultContext{
-		ruleId:      ruleId,
-		opId:        opId,
-		instanceId:  0,
-		ctx:         c.ctx,
-		store:       store,
-		state:       s,
-		jsonEvalReg: sync.Map{},
+		ruleId:     ruleId,
+		opId:       opId,
+		instanceId: 0,
+		ctx:        c.ctx,
+		store:      store,
+		state:      s,
+		tpReg:      sync.Map{},
+		jpReg:      sync.Map{},
 	}
 }
 
diff --git a/internal/topo/sink/rest_sink.go b/internal/topo/sink/rest_sink.go
index 02017b48..96770ad7 100644
--- a/internal/topo/sink/rest_sink.go
+++ b/internal/topo/sink/rest_sink.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -15,6 +15,7 @@
 package sink
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/pkg/cert"
 	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
@@ -280,7 +281,7 @@ func (ms *RestSink) Send(ctx api.StreamContext, v interface{}, logger api.Logger
 	if err != nil {
 		return nil, err
 	}
-	url, ok := temp.(string)
+	u, ok := temp.(string)
 	if !ok {
 		return nil, fmt.Errorf("the value %v of dynamic prop %s for url is not a string", ms.url, temp)
 	}
@@ -292,12 +293,16 @@ func (ms *RestSink) Send(ctx api.StreamContext, v interface{}, logger api.Logger
 		if err != nil {
 			return nil, err
 		}
-		headers, ok = temp.(map[string]string)
+		tstr, ok := temp.(string)
 		if !ok {
-			return nil, fmt.Errorf("the value %v of dynamic prop %s for headers is not a map[string]string", ms.headersTemplate, temp)
+			return nil, fmt.Errorf("the value %v of dynamic prop %s for headersTemplate is not a string", ms.headersTemplate, temp)
+		}
+		err = json.Unmarshal([]byte(tstr), &headers)
+		if err != nil {
+			return nil, fmt.Errorf("rest sink headers template decode error: %v", err)
 		}
 	}
-	return httpx.Send(logger, ms.client, bodyType, method, url, headers, ms.sendSingle, v)
+	return httpx.Send(logger, ms.client, bodyType, method, u, headers, ms.sendSingle, v)
 }
 
 func (ms *RestSink) Close(ctx api.StreamContext) error {
diff --git a/internal/topo/transform/template.go b/internal/topo/transform/template.go
index cc4ac68b..b34f6f68 100644
--- a/internal/topo/transform/template.go
+++ b/internal/topo/transform/template.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -47,3 +47,7 @@ func GenTransform(dt string) (TransFunc, error) {
 		}
 	}, nil
 }
+
+func GenTp(dt string) (*template.Template, error) {
+	return template.New("sink").Funcs(ct.FuncMap).Parse(dt)
+}


Test patch:
diff --git a/internal/topo/context/default_test.go b/internal/topo/context/default_test.go
index ea7c9dcd..e699c6e5 100644
--- a/internal/topo/context/default_test.go
+++ b/internal/topo/context/default_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// 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.
@@ -116,7 +116,7 @@ func TestDynamicProp(t *testing.T) {
 		r []interface{} // parsed results
 	}{
 		{
-			j: "$.a",
+			j: "{$.a",
 			v: []interface{}{
 				map[string]interface{}{
 					"a": 123,
@@ -137,7 +137,7 @@ func TestDynamicProp(t *testing.T) {
 				nil,
 			},
 		}, {
-			j: "$[0].a",
+			j: "{$[0].a",
 			v: []interface{}{
 				[]map[string]interface{}{{
 					"a": 123,
@@ -161,7 +161,7 @@ func TestDynamicProp(t *testing.T) {
 				"single",
 			},
 		}, {
-			j: "a",
+			j: "$a",
 			v: []interface{}{
 				map[string]interface{}{
 					"a": 123,
@@ -177,9 +177,30 @@ func TestDynamicProp(t *testing.T) {
 				},
 			},
 			r: []interface{}{
-				"a",
-				"a",
-				"a",
+				"$a",
+				"$a",
+				"$a",
+			},
+		}, {
+			j: "devices/{{.a}}",
+			v: []interface{}{
+				map[string]interface{}{
+					"a": 123,
+					"b": "dafds",
+				},
+				map[string]interface{}{
+					"a": "single",
+					"c": 20.2,
+				},
+				map[string]interface{}{
+					"b": "b",
+					"c": "c",
+				},
+			},
+			r: []interface{}{
+				"devices/123",
+				"devices/single",
+				"devices/<no value>",
 			},
 		},
 	}
diff --git a/test/rule_pipeline.jmx b/test/rule_pipeline.jmx
index e7ebd5ed..b97b1df9 100644
--- a/test/rule_pipeline.jmx
+++ b/test/rule_pipeline.jmx
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright 2021 EMQ Technologies Co., Ltd.
+  ~ 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.
@@ -344,7 +344,7 @@
     &quot;mqtt&quot;: {&#xd;
       &quot;server&quot;: &quot;tcp://${mqtt_srv}:1883&quot;,&#xd;
       &quot;sendSingle&quot;: true,&#xd;
-      &quot;topic&quot;: &quot;$.t&quot;&#xd;
+      &quot;topic&quot;: &quot;{{.t}}&quot;&#xd;
     }&#xd;
   }],&#xd;
   &quot;options&quot;: {&#xd;