Repository: lf-edge/ekuiper
Pull request #1428 (2022-10-09 04:01:02)

doc(lookup): doc for lookup table


Patch:
diff --git a/.github/workflows/deploy_docs.yaml b/.github/workflows/deploy_docs.yaml
index 15c277b0..655c94d1 100644
--- a/.github/workflows/deploy_docs.yaml
+++ b/.github/workflows/deploy_docs.yaml
@@ -95,8 +95,7 @@ jobs:
         tccli cdn PurgePathCache --Paths '["https://ekuiper.org/docs/"]' --FlushType delete
 
     - name: update search index
-      if: env.VERSION == 'latest'
-      uses: Swilder-M/docsearch-scraper-simple@v3
+      uses: Swilder-M/docsearch-scraper-simple@v4
       env:
         APPLICATION_ID: ${{ secrets.ALGOLIA_APPLICATION_ID }}
         API_KEY: ${{ secrets.ALGOLIA_API_KEY }}
diff --git a/docs/directory.json b/docs/directory.json
index 77be8c35..1c308328 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -207,6 +207,10 @@
 								{
 									"title": "文件源",
 									"path": "rules/sources/builtin/file"
+								},
+								{
+									"title": "Redis 源",
+									"path": "rules/sources/builtin/redis"
 								}
 							]
 						},
@@ -270,6 +274,10 @@
 								{
 									"title": "Nop 动作",
 									"path": "rules/sinks/builtin/nop"
+								},
+								{
+									"title": "Redis 动作",
+									"path": "rules/sinks/builtin/redis"
 								}
 							]
 						},
@@ -300,10 +308,6 @@
 									"title": "TDengine 动作",
 									"path": "rules/sinks/plugin/tdengine"
 								},
-								{
-									"title": "Redis 动作",
-									"path": "rules/sinks/plugin/redis"
-								},
 								{
 									"title": "图像动作",
 									"path": "rules/sinks/plugin/image"
@@ -769,6 +773,10 @@
 								{
 									"title": "File source",
 									"path": "rules/sources/builtin/file"
+								},
+								{
+									"title": "Redis source",
+									"path": "rules/sources/builtin/redis"
 								}
 							]
 						},
@@ -832,6 +840,10 @@
 								{
 									"title": "Nop action",
 									"path": "rules/sinks/builtin/nop"
+								},
+								{
+									"title": "Redis sink",
+									"path": "rules/sinks/builtin/redis"
 								}
 							]
 						},
@@ -862,10 +874,6 @@
 									"title": "TDengine sink",
 									"path": "rules/sinks/plugin/tdengine"
 								},
-								{
-									"title": "Redis sink",
-									"path": "rules/sinks/plugin/redis"
-								},
 								{
 									"title": "Image sink",
 									"path": "rules/sinks/plugin/image"
diff --git a/docs/en_US/concepts/sources/table.md b/docs/en_US/concepts/sources/table.md
index 01c6ea3e..f1392579 100644
--- a/docs/en_US/concepts/sources/table.md
+++ b/docs/en_US/concepts/sources/table.md
@@ -1,16 +1,27 @@
 # Table
 
-In eKuiper, a table is a snapshot of the source data. In contrast to the common static tables that represent batch data, eKuiper tables can change over time.
+A table is a snapshot of the source data. We support two kinds of table: scan table and lookup table.
 
-The source for table can be either bounded or unbounded. For bounded source table, the content of the table is static. For unbounded table, the content of the table is dynamic.
+- Scan table: Consume the stream data as changelog and update the table continuously. In contrast to the common static tables that represent batch data, scan tables can change over time. All stream sources like MQTT, neuron source can also be a scan table source. Scan table was supported since v1.2.0.
+- Lookup table: an external table whose content is usually never read entirely but queried for individual values when necessary. We support to bind physical table as table and generate lookup command(e.g. a SQL on db) on demand. Notice that, not all source type can be a lookup table source, only sources like SQL source which has an external storage can be a lookup source. Lookup table was supported since v1.7.0.
 
-## Table Updates
+## Scan table
 
-Currently, the table update in eKuiper is append-only. Users can specify the properties to limit the table size to avoid too much memory consumption.
+The source for table can be either bounded or unbounded. For bounded source table, the content of the table is static. For unbounded table, the content of the table is dynamic. The content of the table are stored in memory.
 
-## Table Usages
+Currently, the scan table update in eKuiper is append-only. Users can specify the properties to limit the table size to avoid too much memory consumption.
 
-Table cannot be used standalone in a rule. It is usually used to join with streams. It can be used to enrich stream data or as a switch for calculation.
+Scan table cannot be used standalone in a rule. It is usually used to join with streams. It can be used to enrich stream data or as a switch for calculation.
+
+## Lookup Table
+
+Lookup table do not store the table content in memory but refer to the external table. Apparently, only a few of sources is suitable as lookup table which requires the source itself is queryable. The supported sources are:
+
+- Memory source: if a memory source is used as table type, we need to accumulate the data as a table in memory. It can serve as a intermediate to convert any stream into a lookup table.
+- Redis source: Support to query by redis key.
+- SQL source: This is the most typical lookup source. We can use SQL directly to query. 
+
+Unlike scan tables, lookup table will run separately from rules. Thus, all rules that refer to a lookup table can actually query the same table content.
 
 ## More Readings
 
diff --git a/docs/en_US/extension/native/develop/sink.md b/docs/en_US/extension/native/develop/sink.md
index 29b49b71..47bad032 100644
--- a/docs/en_US/extension/native/develop/sink.md
+++ b/docs/en_US/extension/native/develop/sink.md
@@ -53,6 +53,12 @@ func MySink() api.Sink {
 
 The [Memory Sink](https://github.com/lf-edge/ekuiper/blob/master/extensions/sinks/memory/memory.go) is a good example.
 
+#### Updatable Sink
+
+If your sink is updatable, you'll need to deal with the `rowkindField` property. Some sink may also need a `keyField` property to specify which field is the primary key to update.
+
+So in the _Configure_ method, parse the `rowkindField` to know which field in the data is the update action. Then in the _Collect_ method, retrieve the rowkind by the `rowkindField` and perform the proper action. The rowkind value could be `insert`, `update`, `upsert` and `delete`. For example, in SQL sink, each rowkind value will generate different SQL statement to execute.
+
 #### Parse dynamic properties
 
 For customized sink plugins, users may still want to support [dynamic properties](../../../rules/overview.md#dynamic-properties) like the built-in ones.
diff --git a/docs/en_US/extension/native/develop/source.md b/docs/en_US/extension/native/develop/source.md
index 64e96c32..a11fd372 100644
--- a/docs/en_US/extension/native/develop/source.md
+++ b/docs/en_US/extension/native/develop/source.md
@@ -4,6 +4,8 @@ Sources feed data into eKuiper from other systems. eKuiper has built-in source s
 
 ## Developing
 
+There are two kinds of sources. One is the normal source also named scan source, the other is the lookup source. A normal source can be used as a stream or scan table; A lookup source can be used as a lookup table. Users can develop one kind or both in a source plugin.
+
 ### Develop a source
 
 To develop a source for eKuiper is to implement [api.Source](https://github.com/lf-edge/ekuiper/blob/master/pkg/api/stream.go) interface and export it as a golang plugin.
@@ -40,6 +42,49 @@ function MySource() api.Source{
 
 The [Random Source](https://github.com/lf-edge/ekuiper/blob/master/extensions/sources/random/random.go) is a good example.
 
+### Develop a lookup source
+
+To develop a lookup source for eKuiper is to implement [api.LookupSource](https://github.com/lf-edge/ekuiper/blob/master/pkg/api/stream.go) interface and export it.
+
+Before starting the development, you must [setup the environment for golang plugin](../overview.md#setup-the-plugin-developing-environment).
+
+To develop a lookup source, the _Configure_ method must be implemented. This method will be called once the source is initialized. In this method, you can retrieve the _DATASOURCE_ property of the stream (which is topic for mqtt and other messaging system) from the first parameter. Then in the second parameter, a map that contains the configuration in your _yaml_ file is passed. See [configuration](#deal-with-configuration) for more detail. Typically, there will be information such as host, port, user and password of the external system. You can use this map to initialize this source.
+
+```go
+//Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties read from the yaml 
+Configure(datasource string, props map[string]interface{}) error
+```
+
+The next task is to implement _open_ method. The method will be called once the source is created. It is responsible for initialization like establish the connection.
+
+```go
+// Open creates the connection to the external data source
+Open(ctx StreamContext) error
+```
+
+The main task for a Source is to implement _Lookup_ method. The method will be run for each join operation. The parameters are gotten at runtime about the fields, keys and values to be retrieved from the external system. Each lookup source have a different lookup mechanism. For example, the SQL lookup source will build a SQL query from these parameters to retrieve the lookup data.
+
+```go
+// Lookup receive lookup values to construct the query and return query results
+Lookup(ctx StreamContext, fields []string, keys []string, values []interface{}) ([]SourceTuple, error)
+```  
+
+The last method to implement is _Close_ which literally close the connection. It is called when the stream is about to terminate. You could also do any clean up work in this function.
+
+```go
+Close(ctx StreamContext) error
+```
+
+As the source itself is a plugin, it must be in the main package. The exported name must end with `Lookup` so that it can be refereed as the source named `MySource`. For source extension, states are usually needed, so it is recommended to export a constructor function.
+
+```go
+function MySourceLookup() api.LookupSource{
+    return &mySource{}
+}
+```
+
+The [SQL Lookup Source](https://github.com/lf-edge/ekuiper/blob/master/extensions/sources/sql/sqlLookup.go) is a good example.
+
 ### Rewindable source
 If the [rule checkpoint](../../../rules/state_and_fault_tolerance.md#source-consideration) is enabled, the source requires to be rewindable. That means the source need to implement both `api.Source` and `api.Rewindable` interface. 
 
diff --git a/docs/en_US/rules/sinks/builtin/memory.md b/docs/en_US/rules/sinks/builtin/memory.md
index 0c8848c9..bc128068 100644
--- a/docs/en_US/rules/sinks/builtin/memory.md
+++ b/docs/en_US/rules/sinks/builtin/memory.md
@@ -1,5 +1,7 @@
 # Memory action
 
+<span style="background:green;color:white">updatable</span>
+
 The action is used to flush the result into an in-memory topic so that it can be consumed by the [memory source](../../sources/builtin/memory.md). The topic is like pubsub topic such as mqtt, so that there could be multiple memory sinks which publish to the same topic and multiple memory sources which subscribe to the same topic. The typical usage for memory action is to form [rule pipelines](../../rule_pipeline.md).
 
 | Property name | Optional | Description                                    |
@@ -30,4 +32,25 @@ Below is another sample for dynamic topic action:
 
 ::: v-pre
 The data transfer between the memory action and the memory source is in internal format and is not coded or decoded for efficiency. Therefore, the format-related configuration items of the memory action are ignored, except for the data template. The memory action can support data templates to vary the result format, but the result of the data template must be in the object form of a JSON string, e.g. `"{\"key\":\"{{.key}}\"}"`. JSON strings in the form of arrays or non-JSON strings are not supported.
-:::
\ No newline at end of file
+:::
+
+## Updatable Sink
+
+The memory sink support [updatable](../overview.md#updatable-sink). It is used to update the lookup table which subscribes to the same topic as the sink. A typical usage is to create a rule that use the updatable sink to accumulate the memory table. In below example, the data from stream alertStream will update the memory topic `alertVal`. The action verb is specified by the `action` field in the ingested data.
+
+```json
+{
+  "id": "ruleUpdateAlert",
+  "sql":"SELECT * FROM alertStream",
+  "actions":[
+    {
+      "memory": {
+        "keyField": "id",
+        "rowkindField": "action",
+        "topic": "alertVal",
+        "sendSingle": true
+      }
+    }
+  ]
+}
+```
\ No newline at end of file
diff --git a/docs/en_US/rules/sinks/plugin/redis.md b/docs/en_US/rules/sinks/builtin/redis.md
similarity index 88%
rename from docs/en_US/rules/sinks/plugin/redis.md
rename to docs/en_US/rules/sinks/builtin/redis.md
index 20d6ddad..1a3d653c 100644
--- a/docs/en_US/rules/sinks/plugin/redis.md
+++ b/docs/en_US/rules/sinks/builtin/redis.md
@@ -50,9 +50,25 @@ Below is a sample for selecting temperature great than 50 degree, and some profi
   ]
 }
 ```
-### /tmp/redis.txt
+
+### Updatable sample
+
+By specifying the `rowkindField` property, the sink can update according the action specified in that field.
+
 ```json
 {
-   "file":"http://localhost:8080/redis.zip"
- }
+  "id": "ruleUpdateAlert",
+  "sql":"SELECT * FROM alertStream",
+  "actions":[
+    {
+      "redis": {
+        "addr": "127.0.0.1:6379",
+        "dataType": "string",
+        "field": "id",
+        "rowkindField": "action",
+        "sendSingle": true
+      }
+    }
+  ]
+}
 ```
diff --git a/docs/en_US/rules/sinks/overview.md b/docs/en_US/rules/sinks/overview.md
index c26645a1..217efb23 100644
--- a/docs/en_US/rules/sinks/overview.md
+++ b/docs/en_US/rules/sinks/overview.md
@@ -13,6 +13,7 @@ Users can directly use the built-in sinks in the standard eKuiper instance. The
 - [Memory sink](./builtin/memory.md): sink to eKuiper memory topic to form rule pipelines.
 - [Log sink](./builtin/log.md): sink to log, usually for debug only.
 - [Nop sink](./builtin/nop.md): sink to nowhere. It is used for performance testing now.
+- [Redis sink](./builtin/redis.md): sink to redis.
 
 ## Predefined Sink Plugins
 
@@ -27,16 +28,43 @@ The list of predefined sink plugins:
 - [InfluxDB sink](./plugin/influx.md): sink to influx db `v1.x`. 
 - [InfluxDBV2 sink](./plugin/influx2.md): sink to influx db `v2.x`.
 - [Tdengine sink](./plugin/tdengine.md): sink to tdengine.
-- [Redis sink](./plugin/redis.md): sink to redis.
 - [Image sink](./plugin/image.md): sink to an image file. Only used to handle binary result.
 
-### Caching
+## Updatable Sink
+
+By default, sinks append data to the external system. Some external system such as SQL DB is updatable which allows to update or delete data. Similar to lookup source, only a few sinks are "updatable" naturally. The sink must support insert, update and delete. The shipped updatable sinks include:
+
+- Memory sink
+- Redis sink
+- SQL sink
+
+To activate the update feature, the sink must set the `rowkindField` property to specify which field in the data represents to action to take. In the below example, `rowkindField` is set to `action`. 
+
+```json
+{"redis": {
+  "addr": "127.0.0.1:6379",
+  "dataType": "string",
+  "field": "id",
+  "rowkindField": "action",
+  "sendSingle": true
+}}
+```
+
+The data ingested must have a field to indicate the update action. In the below example, the `action` field is the action to perform. The actions could be `insert`, `update`, `upsert` and `delete`. The action implementation varies between sinks. Some sinks may perform the same action for insert, upsert and update.
+
+```json
+{"action":"update", "id":5, "name":"abc"}
+```
+
+This message will update the data of id 5 to the new name.
+
+## Caching
 
 Sinks are used to send processing results to external systems. There are situations where the external system is not available, especially in edge-to-cloud scenarios. For example, in a weak network scenario, the edge-to-cloud network connection may be disconnected and reconnected from time to time. Therefore, sinks provide caching capabilities to temporarily store data in case of recoverable errors and automatically resend the cached data after the error is recovered. Sink's cache can be divided into two levels of storage, namely memory and disk. The user can configure the number of memory cache entries and when the limit is exceeded, the new cache will be stored offline to disk. The cache will be stored in both memory and disk so that the cache capacity becomes larger; it will also continuously detect the failure state and resend without restarting the rule.
 
 The storage location of the offline cache is determined by the storage configuration in `etc/kuiper.yaml`, which defaults to sqlite. If the disk storage is sqlite, all caches will be saved to the `data/cache.db` file. Each sink will have a unique sqlite table to hold the cache. The number of caches is added to the buffer length section of the sink's metrics.
 
-#### Flow
+### Flow
 
 Each sink can configure its own caching mechanism. The caching process is the same for each sink. If caching is enabled, all sink's events go through two phases: first, saving all content to the cache; then deleting the cache after receiving an ack.
 
@@ -44,7 +72,7 @@ Each sink can configure its own caching mechanism. The caching process is the sa
 - Cache mechanism: The cache will first be kept in memory. If the memory threshold is exceeded, the later cache will be saved to disk. Once the disk cache exceeds the disk storage threshold, the cache will start to rotate, i.e. the earliest cache in memory will be discarded and the earliest cache on disk will be loaded instead.
 - Resend policy: Currently the caching mechanism can only run in the default synchronous mode, where if a message is being sent, it will wait for the result of the send to continue sending the next cached data. Otherwise, when new data arrives, the first data in the cache is sent to detect network conditions. If the send is successful, all caches in memory and on disk are sent in a sequential chain. Chained sends can define a send interval to prevent message storms.
 
-#### Configuration
+### Configuration
 
 There are two levels of configuration for the Sink cache. A global configuration in `etc/kuiper.yaml` that defines the default behavior of all rules. There is also a rule sink level definition to override the default behavior.
 
diff --git a/docs/en_US/rules/sinks/plugin/sql.md b/docs/en_US/rules/sinks/plugin/sql.md
index 8c308c79..7d73c6b3 100644
--- a/docs/en_US/rules/sinks/plugin/sql.md
+++ b/docs/en_US/rules/sinks/plugin/sql.md
@@ -94,3 +94,25 @@ The following configuration will write telemetry field's values into database
 }
 ```
 
+### Update Sample
+
+By specifying the `rowkindField` and `keyField`, the sink can generate insert, update or delete statement against the primary key.
+
+```json
+{
+  "id": "ruleUpdateAlert",
+  "sql":"SELECT * FROM alertStream",
+  "actions":[
+    {
+      "sql": {
+        "url": "sqlite://test.db",
+        "keyField": "id",
+        "rowkindField": "action",
+        "table": "alertTable",
+        "sendSingle": true
+      }
+    }
+  ]
+}
+```
+
diff --git a/docs/en_US/rules/sources/builtin/edgex.md b/docs/en_US/rules/sources/builtin/edgex.md
index 4b4f4561..c1ad4769 100644
--- a/docs/en_US/rules/sources/builtin/edgex.md
+++ b/docs/en_US/rules/sources/builtin/edgex.md
@@ -1,5 +1,8 @@
 ## EdgeX Source
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper provides built-in support for EdgeX source stream, which can subscribe the message from [EdgeX message bus](https://github.com/edgexfoundry/go-mod-messaging) and feed into the eKuiper streaming process pipeline.  
 
 ### Stream definition for EdgeX
diff --git a/docs/en_US/rules/sources/builtin/file.md b/docs/en_US/rules/sources/builtin/file.md
index be645138..f1e762bd 100644
--- a/docs/en_US/rules/sources/builtin/file.md
+++ b/docs/en_US/rules/sources/builtin/file.md
@@ -1,5 +1,7 @@
 ## File source
 
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper provides built-in support for reading file content into the eKuiper processing pipeline. The file source is usually used as a [table](../../../sqls/tables.md) and it is the default type for create table statement.
 
 ```sql
diff --git a/docs/en_US/rules/sources/builtin/http_pull.md b/docs/en_US/rules/sources/builtin/http_pull.md
index f0b3d738..89cb2b4e 100644
--- a/docs/en_US/rules/sources/builtin/http_pull.md
+++ b/docs/en_US/rules/sources/builtin/http_pull.md
@@ -1,5 +1,8 @@
 # HTTP pull source 
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper provides built-in support for pulling HTTP source stream, which can pull the message from HTTP server broker and feed into the eKuiper processing pipeline.  The configuration file of HTTP pull source is at `etc/sources/httppull.yaml`. Below is the file format.
 
 ```yaml
diff --git a/docs/en_US/rules/sources/builtin/http_push.md b/docs/en_US/rules/sources/builtin/http_push.md
index df88c0c1..bfb9fde3 100644
--- a/docs/en_US/rules/sources/builtin/http_push.md
+++ b/docs/en_US/rules/sources/builtin/http_push.md
@@ -1,5 +1,8 @@
 # HTTP push source 
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper provides built-in HTTP source stream, which serves as an HTTP server and can receive the message from HTTP client. There will be a single global HTTP server for all HTTP push sources. Each source can have its own endpoint so that multiple endpoints are supported.
 
 ## Configurations
diff --git a/docs/en_US/rules/sources/builtin/memory.md b/docs/en_US/rules/sources/builtin/memory.md
index bbec1203..5d843943 100644
--- a/docs/en_US/rules/sources/builtin/memory.md
+++ b/docs/en_US/rules/sources/builtin/memory.md
@@ -1,11 +1,15 @@
 # Memory Source
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+<span style="background:green;color:white">lookup table source</span>
+
 Memory source is provided to consume events produced by the [memory sink](../../sinks/builtin/memory.md) through topics. 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). The data transfer between the memory action and the memory source is in internal format and is not coded or decoded for efficiency. Therefore, the `format` attribute of the memory source is ignored.
 
 There is no configuration properties. The topic is specified by the stream data source property like below examples:
 
 ```text
-CREATE TABLE table1 (
+CREATE STREAM stream1 (
     name STRING,
     size BIGINT,
     id BIGINT
@@ -21,4 +25,16 @@ Similar to mqtt topic, memory source also supports topic wildcards. Currently, t
 
 Examples:
 1. `home/device1/+/sensor1`
-2. `home/device1/#`
\ No newline at end of file
+2. `home/device1/#`
+
+## Lookup Table
+
+The memory source supports lookup table. Below is an example to create a lookup table against memory topic `topicName`. Notice that, `KEY` property is required as a lookup table which will be served as a primary key for the virtual table and accelerate the query.
+
+```text
+CREATE TABLE alertTable() WITH (DATASOURCE="topicName", TYPE="memory", KIND="lookup", KEY="id")
+```
+
+After creating a memory lookup table, it will start to accumulate the data from the memory topic indexed by the key field. It will keep running independently of rules. Each topic and key pair will have a single in-memory copy of the virtual table. All rules that refer to the same table or the memory tables with the same topic/key pair will share the same copy of data.
+
+The memory lookup table can be used like a pipeline between multiple rules which is similar to the [rule pipeline](../../rule_pipeline.md) concept. It can store the history of any stream type in memory so that other streams can work with. By working together with [updatable memory sink](../../sinks/builtin/memory.md#updatable-sink), the table content can be updated.
\ No newline at end of file
diff --git a/docs/en_US/rules/sources/builtin/mqtt.md b/docs/en_US/rules/sources/builtin/mqtt.md
index 67c3accc..63f17080 100644
--- a/docs/en_US/rules/sources/builtin/mqtt.md
+++ b/docs/en_US/rules/sources/builtin/mqtt.md
@@ -1,5 +1,8 @@
 # MQTT source 
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper provides built-in support for MQTT source stream, which can subscribe the message from MQTT broker and feed into the eKuiper processing pipeline.  The configuration file of MQTT source is at `$ekuiper/etc/mqtt_source.yaml`. Below is the file format.
 
 ```yaml
diff --git a/docs/en_US/rules/sources/builtin/neuron.md b/docs/en_US/rules/sources/builtin/neuron.md
index 74b55179..41f0c2ae 100644
--- a/docs/en_US/rules/sources/builtin/neuron.md
+++ b/docs/en_US/rules/sources/builtin/neuron.md
@@ -1,5 +1,8 @@
 # Neuron Source
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 Neuron source is provided to consume events from the local neuron instance. Notice that, the source is bound to the local neuron only which must be able to communicate through nanomsg ipc protocol without network. In the eKuiper side, all neuron source and sink instances share the same connection, thus the events consumed are also the same. The events are formatted in json by default. Notice that, the dial to neuron is async which will run in the background and always redial when the previous attempt fails, which means that the rule using neuron sink will not see an error even when neuron is down. While debugging, we need to check the rule status for the message in counts.
 
 The event content and format are fixed as below: 
diff --git a/docs/en_US/rules/sources/builtin/redis.md b/docs/en_US/rules/sources/builtin/redis.md
new file mode 100644
index 00000000..18308a7d
--- /dev/null
+++ b/docs/en_US/rules/sources/builtin/redis.md
@@ -0,0 +1,25 @@
+## Redis source
+
+<span style="background:green;color:white">lookup table source</span>
+
+eKuiper provides built-in support for looking up data in redis. Notice that, the redis source can only be used as a lookup table now. Stream and scan table is not supported.
+
+```text
+create table table1 () WITH (DATASOURCE="0", FORMAT="json", TYPE="redis", KIND="lookup");
+```
+
+You can use [cli](../../../operation/cli/tables.md) or [rest api](../../../operation/restapi/tables.md) to manage the tables.
+
+The configure file for the redis source is in */etc/sources/redis.yaml* in which the path to the file can be specified.
+
+```yaml
+default:
+  # the redis host address
+  addr: "127.0.0.1:6379"
+  # currently supports string and list only
+  datatype: "string"
+#  username: ""
+#  password: ""
+```
+
+With this yaml file, the table will refer to the database 0 in redis instance of address 127.0.0.1:6379. The value type is `string`.
\ No newline at end of file
diff --git a/docs/en_US/rules/sources/overview.md b/docs/en_US/rules/sources/overview.md
index b213bf86..ae199c12 100644
--- a/docs/en_US/rules/sources/overview.md
+++ b/docs/en_US/rules/sources/overview.md
@@ -12,6 +12,7 @@ Users can directly use the built-in sources in the standard eKuiper instance. Th
 - [Http pull source](./builtin/http_pull.md): source to pull data from http servers.
 - [Memory source](./builtin/memory.md): source to read from eKuiper memory topic to form rule pipelines.
 - [File source](./builtin/file.md): source to read from file, usually used as tables.
+- [Redis source](./builtin/redis.md): source to lookup from redis as a lookup table.
 
 ## Predefined Source Plugins
 
@@ -23,6 +24,7 @@ The list of predefined source plugins:
 
 - [Zero MQ source](./plugin/zmq.md): read data from zero mq.
 - [Random source](./plugin/random.md): a source to generate random data for testing.
+- [SQL source](./plugin/sql.md): a source to periodically fetch data from SQL DB.
 
 ## Use of sources
 
diff --git a/docs/en_US/rules/sources/plugin/random.md b/docs/en_US/rules/sources/plugin/random.md
index b136f3a5..edd37f3f 100644
--- a/docs/en_US/rules/sources/plugin/random.md
+++ b/docs/en_US/rules/sources/plugin/random.md
@@ -1,6 +1,9 @@
 # Random Source
 
-he source will generate random inputs with a specified pattern.
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
+The source will generate random inputs with a specified pattern.
 
 ## Compile & deploy plugin
 
diff --git a/docs/en_US/rules/sources/plugin/sql.md b/docs/en_US/rules/sources/plugin/sql.md
index 6baefa70..f0140a6a 100644
--- a/docs/en_US/rules/sources/plugin/sql.md
+++ b/docs/en_US/rules/sources/plugin/sql.md
@@ -1,5 +1,9 @@
 # Sql Source
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+<span style="background:green;color:white">lookup table source</span>
+
 The source will query the database periodically to get data stream.
 
 ## Compile & deploy plugin
@@ -127,3 +131,30 @@ demo (
 ```
 
 The configuration keys "template_config" will be used.
+
+## Lookup Table
+
+The SQL source supports to be a lookup table. We can use create table statement to create a SQL lookup table. It will bind to the physical SQL DB and query on demand.
+
+```text
+CREATE TABLE alertTable() WITH (DATASOURCE="tableName", CONF_KEY="sqlite_config", TYPE="sql", KIND="lookup")
+```
+
+### Lookup cache
+
+Query external DB is supposed to be slower than in memory calculation. If the throughput is high, the lookup cache can be used to improve the performance.
+
+If lookup cache is not enabled, so all the requests are sent to external database. When lookup cache is enabled, each lookup table instance will hold a cache. When querying, we will first look up the cache before sending to the external database.
+
+The cache configuration lies in the `sql.yaml`.
+
+```yaml
+  lookup:
+    cache: true
+    cacheTtl: 600
+    cacheMissingKey: true
+```
+
+- cache: bool value to indicate whether to enable cache.
+- cacheTtl: the time to live of the cache in seconds.
+- cacheMissingKey: whether to cache nil value for a key.
\ No newline at end of file
diff --git a/docs/en_US/rules/sources/plugin/zmq.md b/docs/en_US/rules/sources/plugin/zmq.md
index 556e1af1..4d4695ef 100644
--- a/docs/en_US/rules/sources/plugin/zmq.md
+++ b/docs/en_US/rules/sources/plugin/zmq.md
@@ -1,5 +1,8 @@
 # Zmq Source
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 The source will subscribe to a Zero Mq topic to import the messages into eKuiper
 
 ## Compile & deploy plugin
diff --git a/docs/en_US/sqls/tables.md b/docs/en_US/sqls/tables.md
index f41d32e0..6abbb2fd 100644
--- a/docs/en_US/sqls/tables.md
+++ b/docs/en_US/sqls/tables.md
@@ -2,7 +2,10 @@
 
 eKuiper streams is unbounded and immutable, any new data are appended in the current stream for processing.  **Table** is provided to represent the current state of the stream. It can be considered as a snapshot of the stream. Users can use table to retain a batch of data for processing.
 
-Table is not allowed to use alone in eKuiper. It is only recommended to join with streams. When joining with stream, table will be updated continuously when new event coming. However, only events arriving on the stream side trigger downstream updates and produce join output.
+There are two kinds of table:
+
+- Scan table: accumulates the data in memory. It is suitable for smaller dataset and the table content do NOT need to share between rules.
+- Lookup table: refer to external table content. It is suitable for bigger dataset and share table content across rules.
 
 ## Syntax
 
@@ -19,11 +22,34 @@ Table supports the same [data types](./streams.md#data-types) as stream.
 
 Table also supports all [the properties of the stream](./streams.md#language-definitions). Thus, all the source type are also supported in table. Many sources are not batched which have one event at any given time point, which means the table will always have only one event. An additional property `RETAIN_SIZE` to specify the size of the table snapshot so that the table can hold an arbitrary amount of history data.
 
+### Lookup Table Syntax
+
+The syntax is the same as creating a normal scan table, just need to specify kind property to be `lookup`. Below is an example to create a lookup data, which binds to redis database 0.
+
+```sql
+CREATE TABLE alertTable() WITH (DATASOURCE="0", TYPE="redis", KIND="lookup")
+```
+
+Currently, only `memory`, 'redis' and 'sql' source can be lookup table.
+
+### Table properties
+
+| Property name | Optional | Description                                                                                                                                                          |
+|---------------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| DATASOURCE    | false    | The value is determined by source type. The topic names list if it's a MQTT data source. Please refer to related document for other sources.                         |
+| FORMAT        | true     | The data format, currently the value can be "JSON", "PROTOBUF" and "BINARY". The default is "JSON". Check [Binary Stream](#binary-stream) for more detail.           |
+| SCHEMAID      | true     | The schema to be used when decoding the events. Currently, only use when format is PROTOBUF.                                                                         |
+| KEY           | true     | The primary key of the table. For example, for SQL source key specifies the primary key in the SQL table. It is not obeyed by all source types.                      |
+| TYPE          | true     | The source type. Each source type may support one kind or both kind of tables. Please refer to related documents.                                                    |
+| CONF_KEY      | true     | If additional configuration items are requied to be configured, then specify the config key here. See [MQTT stream](../rules/sources/builtin/mqtt.md) for more info. |
+| KIND          | true     | The table kind, could be `scan` or `lookup`. If not specified, the default value is `scan`.                                                                          |
+
+
 ## Usage scenarios
 
-Typically, table will be joined with stream with or without a window. When joining with stream, table data won't affect the downstream updata, it is treated like a static referenced data although it may be updated internally.
+Typically, table will be joined with stream with or without a window. When joining with stream, table data won't affect the downstream data, it is treated like a static referenced data, although it may be updated internally.
 
-### Lookup table
+### Enrich data
 
 A typical usage for table is as a lookup table. Sample SQL will be like:
 
diff --git a/docs/zh_CN/concepts/sources/table.md b/docs/zh_CN/concepts/sources/table.md
index 6fa4e5c4..767b800d 100644
--- a/docs/zh_CN/concepts/sources/table.md
+++ b/docs/zh_CN/concepts/sources/table.md
@@ -1,16 +1,27 @@
 # 表
 
-在 eKuiper 中，表是源数据在当前时间的快照。相比于普通的数据库表，eKuiper 中的表会随着时间而变化。
+表是源数据在当前时间的快照。我们支持两种类型的表：扫描表（Scan Table）和查询表（Lookup Table）。
+
+- 扫描表。消费流数据作为变化日志，并持续更新表。与常见的代表批处理数据的静态表相比，扫描表可以随时间变化。所有的流数据源如 MQTT、 Neuron 源等都可以是扫描表源。扫描表从 v1.2.0 版本开始支持。
+- 查询表：一个外部表，其内容通常不会被完全读取，仅在必要时进行查询。我们支持将实体表绑定为查询表，并根据需要生成查询命令（例如，数据库上的 SQL）。请注意，不是所有的源类型都可以成为查询表源，只有像 SQL 源这样的有外部存储的源可以成为查询源。我们从 v1.7.0 版本开始支持查询表。
+
+## 扫描表
 
 表的数据源既可以是无界的也可以是有界的。对于有界数据源来说，其表的内容为静态的。若表的数据源为无界数据流，则其内容会动态变化。
 
-## 表内容更新
+当前，扫描表的内容更新仅支持追加。用户创建表时，可以指定参数限制表的大小，防止占用过多内存。
+
+扫描表不能在规则中单独使用，必须与流搭配，通常用于与流进行连接。扫描表可用于补全流数据或作为计算的开关。
+
+## 查询表
 
-当前，表的内容更新仅支持追加。用户创建表时，可以指定参数限制表的大小，防止占用过多内存。
+查询表不在内存中存储表的内容，而是引用外部表。显然，只有少数源适合作为查询表，这需要源本身是可查询的。支持的源包括：
 
-## 表的使用
+- 内存源：如果内存源被用作表类型，我们需要在内存中把数据积累成表。它可以作为一个中间环节，将任何流转换为查询表。
+- Redis：支持按Redis键查询。
+- SQL源：这是最典型的查询源。我们可以直接使用SQL来查询。
 
-表不能在规则中单独使用，必须与流搭配，通常用于与流进行连接。表可用于补全流数据或作为计算的开关。
+与扫描表不同，查询表将与规则分开运行。因此，所有使用查询表的规则实际上可以查询同一个表的内容。
 
 ## 更多信息
 
diff --git a/docs/zh_CN/extension/native/develop/sink.md b/docs/zh_CN/extension/native/develop/sink.md
index 2be4ba53..62a5398d 100644
--- a/docs/zh_CN/extension/native/develop/sink.md
+++ b/docs/zh_CN/extension/native/develop/sink.md
@@ -56,6 +56,12 @@ func MySink() api.Sink {
 
 [Memory Sink](https://github.com/lf-edge/ekuiper/blob/master/extensions/sinks/memory/memory.go) 是一个很好的示例。
 
+#### 可更新的 Sink
+
+如果你的 Sink 是可更新的，你将需要处理 `rowkindField` 属性。有些 sink 可能还需要一个 `keyField' 属性来指定哪个字段是要更新的主键。
+
+因此，在_Configure_方法中，需要解析 `rowkindField` 以知道数据中的哪个字段表示更新的动作。然后在_Collect_方法中，通过该字段获取动作类型，并执行适当的操作。rowkind 的值可以是 `insert`、`update`、`upsert` 和 `delete`。例如，在 SQL sink 中，每种 rowkind 值将产生不同的SQL语句来执行。
+
 #### 解析动态属性
 
 在自定义的 sink 插件中，用户可能仍然想要像内置的 sink 一样支持[动态属性](../../../rules/overview.md#动态属性)。 我们在 context 对象中提供了 `ParseTemplate` 方法使得开发者可以方便地解析动态属性并应用于插件中。开发组应当根据业务逻辑，设计那些属性支持动态值。然后在代码编写时，使用此方法解析用户传入的属性值。
diff --git a/docs/zh_CN/extension/native/develop/source.md b/docs/zh_CN/extension/native/develop/source.md
index e3dac306..c932b62d 100644
--- a/docs/zh_CN/extension/native/develop/source.md
+++ b/docs/zh_CN/extension/native/develop/source.md
@@ -4,7 +4,9 @@
 
 ## 开发
 
-### 开发一个源
+有两种类型的源。一种是普通源，即扫描源（Scan Source），另一种是查询源（Lookup Source）。一个正常的源可以作为一个流或扫描表使用；一个查询源可以作为一个查询表使用。用户可以在一个源插件中开发一种或两种源。
+
+### 开发普通源
 
 为 eKuiper 开发源的 是实现 [api.Source](https://github.com/lf-edge/ekuiper/blob/master/pkg/api/stream.go) 接口并将其导出为 golang 插件。
 
@@ -40,6 +42,49 @@ function MySource() api.Source{
 
 [Random Source](https://github.com/lf-edge/ekuiper/blob/master/extensions/sources/random/random.go)  是一个很好的示例。
 
+### 开发查询源
+
+为 eKuiper 开发一个查询源就是实现 [api.LookupSource](https://github.com/lf-edge/ekuiper/blob/master/pkg/api/stream.go) 接口并将其导出。
+
+在开始开发之前，您必须为 [golang 插件设置环境](../overview.md#插件开发环境设置)。
+
+要开发一个查询源，必须实现 _Configure_ 方法。 初始化源后，将调用此方法。 在此方法中，您可以从第一个参数检索流的 _DATASOURCE_ 属性（这是 mqtt 和其他消息传递系统的主题）。 然后在第二个参数中，传递包含 _yaml_ 文件中的配置的映射。 有关更多详细信息，请参见 [配置](#处理配置)。 通常，将有外部系统的信息，例如主机、端口、用户和密码。 您可以使用此映射来初始化此源。
+
+```go
+//在初始化过程中调用。用数据源（例如mqtt的topic）和从yaml中读取的属性来配置这个源 
+Configure(datasource string, props map[string]interface{}) error
+```
+
+下一个任务是实现_open_方法。一旦源被创建，该方法将被调用。它负责初始化，比如建立连接。
+
+```go
+// Open 创建与外部数据源的连接
+Open(ctx StreamContext) error
+```
+
+查询源的主要任务是实现 _Lookup_ 方法。该方法将在每个连接操作中运行。参数是在运行时获得的，包括要从外部系统中检索的字段、键和值等信息。每个查询源都有不同的查询机制。例如，SQL查询源将从这些参数中组装一个 SQL 查询来检索查询数据。
+
+```go
+// Lookup 接收查询值以构建查询并返回查询结果
+Lookup(ctx StreamContext, fields []string, keys []string, values []interface{}) ([]SourceTuple, error)
+```  
+
+最后要实现的方法是 _Close_，它实际上用来关闭连接。 当流即将终止时调用它。 您也可以在此功能中执行任何清理工作。
+
+```go
+Close(ctx StreamContext) error
+```
+
+由于源本身是一个插件，因此它必须位于主程序包中。Export 的名称必须以 `Lookup` 结尾，这样它就可以被称为 `MySource` 的查询源。对于源扩展，通常需要状态，所以建议导出一个构造函数。
+
+```go
+function MySourceLookup() api.LookupSource{
+    return &mySource{}。
+}
+```
+
+[SQL Lookup Source](https://github.com/lf-edge/ekuiper/blob/master/extensions/sources/sql/sqlLookup.go) 是一个很好的示例。。
+
 ### 处理配置
 
 eKuiper 配置的格式为 yaml，它提供了一个集中位置  _/etc_  来保存所有配置。 在其中，为源配置提供了一个子文件夹  _sources_，同时也适用于扩展源。
diff --git a/docs/zh_CN/rules/sinks/builtin/memory.md b/docs/zh_CN/rules/sinks/builtin/memory.md
index 47dd7ce4..33bd8858 100644
--- a/docs/zh_CN/rules/sinks/builtin/memory.md
+++ b/docs/zh_CN/rules/sinks/builtin/memory.md
@@ -1,5 +1,7 @@
 # 内存动作
 
+<span style="background:green;color:white">updatable</span>
+
 该动作用于将结果刷新到内存中的主题中，以便 [内存源](../../sources/builtin/memory.md) 可以使用它。 该主题类似于 pubsub 主题，例如 mqtt，因此可能有多个内存目标发布到同一主题，也可能有多个内存源订阅同一主题。 内存动作的典型用途是形成[规则管道](../../rule_pipeline.md)。
 
 | 属性名称  | 是否可选 | 描述                                  |
@@ -30,4 +32,25 @@
 
 ::: v-pre
 内存动作和内存源之间的数据传输采用内部格式，不经过编解码以提高效率。因此，内存动作的格式相关配置项，除了数据模板之外都会被忽略。内存动作可支持数据模板对结果格式进行变化，但是数据模板的结果必须为 JSON 字符串的 object 形式，例如 `"{\"key\":\"{{.key}}\"}"`。数组形式的 JSON 字符串或者非 JSON 字符串都不支持。
-:::
\ No newline at end of file
+:::
+
+## 更新
+
+内存动作支持[更新](../overview.md#更新)。可用于更新订阅了与 sink 相同的主题的查询表。一个典型的用法是创建一个规则，使用可更新的 sink 来累积更新内存表。在下面的例子中，来自流alertStream的数据将更新内存主题`alertVal`。更新动作是由流入的数据中的 `action` 字段指定的。
+
+```json
+{
+  "id": "ruleUpdateAlert",
+  "sql":"SELECT * FROM alertStream",
+  "actions":[
+    {
+      "memory": {
+        "keyField": "id",
+        "rowkindField": "action",
+        "topic": "alertVal",
+        "sendSingle": true
+      }
+    }
+  ]
+}
+```
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sinks/plugin/redis.md b/docs/zh_CN/rules/sinks/builtin/redis.md
similarity index 90%
rename from docs/zh_CN/rules/sinks/plugin/redis.md
rename to docs/zh_CN/rules/sinks/builtin/redis.md
index 2e2713c6..1e91b491 100644
--- a/docs/zh_CN/rules/sinks/plugin/redis.md
+++ b/docs/zh_CN/rules/sinks/builtin/redis.md
@@ -56,4 +56,26 @@ redis 源代码在 extensions 目录中，但是需要在 eKuiper 根目录编
 {
   "file":"http://localhost:8080/redis.zip"
 }
+```
+
+### 更新示例
+
+通过指定 `rowkindField` 属性，sink 可以根据该字段中指定的动作进行更新。
+
+```json
+{
+  "id": "ruleUpdateAlert",
+  "sql":"SELECT * FROM alertStream",
+  "actions":[
+    {
+      "redis": {
+        "addr": "127.0.0.1:6379",
+        "dataType": "string",
+        "field": "id",
+        "rowkindField": "action",
+        "sendSingle": true
+      }
+    }
+  ]
+}
 ```
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sinks/overview.md b/docs/zh_CN/rules/sinks/overview.md
index b1d81104..efe10aa7 100644
--- a/docs/zh_CN/rules/sinks/overview.md
+++ b/docs/zh_CN/rules/sinks/overview.md
@@ -13,6 +13,7 @@
 - [Memory sink](./builtin/memory.md)：输出到 eKuiper 内存主题以形成规则管道。
 - [Log sink](./builtin/log.md)：写入日志，通常只用于调试。
 - [Nop sink](./builtin/nop.md)：不输出，用于性能测试。
+- [Redis sink](./builtin/redis.md): 写入 Redis 。
 
 ## 预定义的动作插件
 
@@ -27,16 +28,43 @@
 - [InfluxDB sink](./plugin/influx.md)： 写入 Influx DB `v1.x`。
 - [InfluxDBV2 sink](./plugin/influx2.md)： 写入 Influx DB `v2.x`。
 - [Tdengine sink](./plugin/tdengine.md)： 写入 Tdengine 。
-- [Redis sink](./plugin/redis.md)： 写入 redis 。
 - [Image sink](./plugin/image.md)： 写入一个图像文件。仅用于处理二进制结果。
 
-### 缓存
+## 更新
+
+默认情况下，Sink 将数据附加到外部系统中。一些外部系统，如 SQL DB 本身是可更新的，允许更新或删除数据。与查找源类似，只有少数 Sink 是天然 "可更新 "的。可更新的 Sink 必须支持插入、更新和删除。产品自带的 Sink 种，可更新的包括: 
+
+- Memory Sink
+- Redis Sink
+- SQL Sink
+
+为了激活更新功能，Sink 必须设置 `rowkindField` 属性，以指定数据中的哪个字段代表要采取的动作。在下面的例子中，`rowkindField` 被设置为 `action`。
+
+```json
+{"redis": {
+  "addr": "127.0.0.1:6379",
+  "dataType": "string",
+  "field": "id",
+  "rowkindField": "action",
+  "sendSingle": true
+}}
+```
+
+流入的数据必须有一个字段来表示更新的动作。在下面的例子中，`action` 字段是要执行的动作。动作可以是插入、更新、 upsert 和删除。不同的 sink 的动作实现是不同的。有些 sink 可能对插入、upsert 和更新执行相同的动作。
+
+```json
+{"action":"update", "id":5, "name":"abc"}
+```
+
+这条信息将把id 为 5的数据更新为新的名字。
+
+## 缓存
 
 动作用于将处理结果发送到外部系统中，存在外部系统不可用的情况，特别是在从边到云的场景中。例如，在弱网情况下，边到云的网络连接可能会不时断开和重连。因此，动作提供了缓存功能，用于在发送错误的情况下暂存数据，并在错误恢复之后自动重发缓存数据。动作的缓存可分为内存和磁盘的两级存储。用户可配置内存缓存条数，超过上限后，新的缓存将离线存储到磁盘中。缓存将同时保存在内存和磁盘中，这样缓存的容量就变得更大了；它还将持续检测故障恢复状态，并在不重新启动规则的情况下重新发送。
 
 离线缓存的保存位置根据 `etc/kuiper.yaml` 里的 store 配置决定，默认为 sqlite 。如果磁盘存储是sqlite，所有的缓存将被保存到`data/cache.db`文件。每个 sink 将有一个唯一的 sqlite 表来保存缓存。缓存的计数添加到 sink 的 指标中的 buffer length 部分。
 
-#### 流程
+### 流程
 
 每个sink都可以配置自己的缓存机制。每个 sink 的缓存流程是相同的。如果启用了缓存，所有 sink 的事件都会经过两个阶段：首先是将所有内容保存到缓存中；然后在收到ack后删除缓存。
 
@@ -44,7 +72,7 @@
 - 缓存机制：缓存将首先被保存在内存中。如果超过了内存的阈值，后面的缓存将被保存到磁盘中。一旦磁盘缓存超过磁盘存储阈值，缓存将开始rotate，即内存中最早的缓存将被丢弃，并加载磁盘中最早的缓存来代替。
 - 重发策略：目前缓存机制仅可运行在默认的同步模式中，如果有一条消息正在发送中，则会等待发送的结果以继续发送下个缓存数据。否则，当有新的数据到来时，发送缓存中的第一个数据以检测网络状况。如果发送成功，将按顺序链式发送所有内存和磁盘中的所有缓存。链式发送可定义一个发送间隔，防止形成消息风暴。
 
-#### 配置
+### 配置
 
 Sink 缓存的配置有两个层次。`etc/kuiper.yaml` 中的全局配置，定义所有规则的默认行为。还有一个规则 sink 层的定义，用来覆盖默认行为。
 
diff --git a/docs/zh_CN/rules/sinks/plugin/sql.md b/docs/zh_CN/rules/sinks/plugin/sql.md
index dacfe6f6..2a5ee1d2 100644
--- a/docs/zh_CN/rules/sinks/plugin/sql.md
+++ b/docs/zh_CN/rules/sinks/plugin/sql.md
@@ -90,5 +90,24 @@
 }
 ```
 
+### 更新示例
 
+通过指定 `rowkindField` 和 `keyField` 属性，sink 可以生成针对主键的插入、更新或删除语句。
 
+```json
+{
+  "id": "ruleUpdateAlert",
+  "sql":"SELECT * FROM alertStream",
+  "actions":[
+    {
+      "sql": {
+        "url": "sqlite://test.db",
+        "keyField": "id",
+        "rowkindField": "action",
+        "table": "alertTable",
+        "sendSingle": true
+      }
+    }
+  ]
+}
+```
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sources/builtin/edgex.md b/docs/zh_CN/rules/sources/builtin/edgex.md
index d59e88a0..6eacd9df 100644
--- a/docs/zh_CN/rules/sources/builtin/edgex.md
+++ b/docs/zh_CN/rules/sources/builtin/edgex.md
@@ -1,7 +1,8 @@
-
-
 # EdgeX 源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper 提供了内置的 EdgeX 源支持，它可以被用来订阅来自于[EdgeX 消息总线](https://github.com/edgexfoundry/go-mod-messaging)的数据，并且将数据放入 eKuiper 数据处理流水线中。
 
 ## EdgeX 流定义
@@ -55,7 +56,7 @@ EdgeX 中所有的 `INT8` , `INT16`, `INT32`,  `INT64` , `UINT8` , `UINT16` ,  `
 
 EdgeX 中所有的 `FLOAT32`, `FLOAT64`  数组类型会被转换为 `Float` 数组。 
 
-# 全局配置
+## 全局配置
 
 EdgeX 源配置文件为 `$ekuiper/etc/sources/edgex.yaml`，以下配置文件内容。
 
diff --git a/docs/zh_CN/rules/sources/builtin/file.md b/docs/zh_CN/rules/sources/builtin/file.md
index d6dccc9f..7209178e 100644
--- a/docs/zh_CN/rules/sources/builtin/file.md
+++ b/docs/zh_CN/rules/sources/builtin/file.md
@@ -1,5 +1,7 @@
 ## 文件源
 
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper 提供了内置支持，可将文件内容读入 eKuiper 处理管道。 文件源通常用作 [表格](../../../sqls/tables.md)， 并且采用 create table 语句的默认类型。
 
 ```sql
@@ -10,7 +12,7 @@ CREATE TABLE table1 (
 ) WITH (DATASOURCE="lookup.json", FORMAT="json", TYPE="file");
 ```
 
-您可以使用 [cli](../../../operation/cli/tables.md) 或 [rest api](../../../operation/restapi/tables.md) 来管理表
+您可以使用 [cli](../../../operation/cli/tables.md) 或 [rest api](../../../operation/restapi/tables.md) 来管理表。
 
 文件源的配置文件是 */etc/sources/file.yaml* ，可以在其中指定文件的路径。
 
diff --git a/docs/zh_CN/rules/sources/builtin/http_pull.md b/docs/zh_CN/rules/sources/builtin/http_pull.md
index 6caa7457..cc1ad82d 100644
--- a/docs/zh_CN/rules/sources/builtin/http_pull.md
+++ b/docs/zh_CN/rules/sources/builtin/http_pull.md
@@ -1,5 +1,8 @@
 # HTTP 提取源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper 为提取 HTTP 源流提供了内置支持，该支持可从 HTTP 服务器代理提取消息并输入 eKuiper 处理管道。 HTTP提取源的配置文件位于 `etc/sources/httppull.yaml`中。 以下是文件格式。
 
 ```yaml
diff --git a/docs/zh_CN/rules/sources/builtin/http_push.md b/docs/zh_CN/rules/sources/builtin/http_push.md
index 62a3bd95..6722b2e5 100644
--- a/docs/zh_CN/rules/sources/builtin/http_push.md
+++ b/docs/zh_CN/rules/sources/builtin/http_push.md
@@ -1,5 +1,8 @@
 # HTTP push 源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper 提供了内置的 HTTP 源，它作为一个 HTTP 服务器，可以接收来自 HTTP 客户端的消息。所有的 HTTP 推送源共用单一的全局 HTTP 数据服务器。每个源可以有自己的 URL，这样就可以支持多个端点。
 
 ## 配置
diff --git a/docs/zh_CN/rules/sources/builtin/memory.md b/docs/zh_CN/rules/sources/builtin/memory.md
index 82038562..cf11dbd2 100644
--- a/docs/zh_CN/rules/sources/builtin/memory.md
+++ b/docs/zh_CN/rules/sources/builtin/memory.md
@@ -1,11 +1,15 @@
 # 内存源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+<span style="background:green;color:white">lookup table source</span>
+
 内存源通过主题消费由 [内存目标](../../sinks/builtin/memory.md) 生成的事件。该主题类似于 pubsub 主题，例如 mqtt，因此可能有多个内存目标发布到同一主题，也可能有多个内存源订阅同一主题。 内存动作的典型用途是形成[规则管道](../../rule_pipeline.md)。内存动作和内存源之间的数据传输采用内部格式，不经过编解码以提高效率。因此，内存源的`format`属性会被忽略。
 
 主题没有配置属性，由流数据源属性指定，如以下示例所示：
 
 ```text
-CREATE TABLE table1 (
+CREATE STREAM table1 (
     name STRING,
     size BIGINT,
     id BIGINT
@@ -21,4 +25,16 @@ CREATE TABLE table1 (
 
 示例：
 1. `home/device1/+/sensor1`
-2. `home/device1/#`
\ No newline at end of file
+2. `home/device1/#`
+
+## 查询表
+
+内存源支持查询表。下面是一个针对内存主题 "topicName" 创建查询表的例子。注意，作为查询表使用时，`KEY` 属性是必须的，它将作为虚拟表的主键来加速查询。
+
+```text
+CREATE TABLE alertTable() WITH (DATASOURCE="topicName", TYPE="memory", KIND="lookup", KEY="id")
+```
+
+在创建一个内存查询表后，它将开始积累由键字段索引的内存主题的数据。它将一直独立于规则运行。每个主题和键对将有一个虚拟表的内存拷贝。所有引用同一表或具有相同主题/键对的内存表的规则将共享同一数据副本。
+
+内存查询表可以像多个规则之间的管道一样使用，这与[规则管道](../../rule_pipeline.md)的概念类似。它可以在内存中存储任何流类型的历史，以便其他流可以与之合作。通过与[可更新的内存动作](../../sinks/builtin/memory.md#更新)一起工作，表格内容可以被更新。
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sources/builtin/mqtt.md b/docs/zh_CN/rules/sources/builtin/mqtt.md
index 141122a4..9621ca1f 100644
--- a/docs/zh_CN/rules/sources/builtin/mqtt.md
+++ b/docs/zh_CN/rules/sources/builtin/mqtt.md
@@ -1,5 +1,8 @@
 # MQTT源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 eKuiper 为 MQTT 源流提供了内置支持，流可以订阅来自 MQTT 代理的消息并输入eKuiper 处理管道。 MQTT 源的配置文件位于 `$ekuiper/etc/mqtt_source.yaml`。 以下是文件格式。
 
 ```yaml
diff --git a/docs/zh_CN/rules/sources/builtin/neuron.md b/docs/zh_CN/rules/sources/builtin/neuron.md
index 6ce10be3..33217d7d 100644
--- a/docs/zh_CN/rules/sources/builtin/neuron.md
+++ b/docs/zh_CN/rules/sources/builtin/neuron.md
@@ -1,5 +1,8 @@
 # Neuron 源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 Neuron 源订阅本地 neuron 实例的消息。需要注意的是，该源仅可用于本地的 neuron，因为与 neuron 的通信基于 nanomsg ipc 协议，无法通过网络进行。在 eKuiper 端，所有 neuron 源和动作共享同一个 neuron 连接。需要注意的是，拨号到 Neuron 是异步的，它将在后台运行，不断重拨直到连接成功。这意味着使用 Neuron sink 的规则即使在 Neuron 停机时也不会看到错误。用户调试时，可查看规则状态，注意消息流入数量是否正常。
 
 Neuron 发过来的消息为固定的 json 格式，如下所示： 
diff --git a/docs/zh_CN/rules/sources/builtin/redis.md b/docs/zh_CN/rules/sources/builtin/redis.md
new file mode 100644
index 00000000..f2026c81
--- /dev/null
+++ b/docs/zh_CN/rules/sources/builtin/redis.md
@@ -0,0 +1,25 @@
+## Redis 源
+
+<span style="background:green;color:white">lookup table source</span>
+
+eKuiper 提供了对 redis 中数据查询的内置支持。请注意，现在 redis 源只能作为一个查询表使用。不支持流和扫描表。
+
+```text
+create table table1 () WITH (DATASOURCE="0", FORMAT="json", TYPE="redis", KIND="lookup");
+```
+
+您可以使用 [cli](../../../operation/cli/tables.md) 或 [rest api](../../../operation/restapi/tables.md) 来管理表。
+
+Redis 源的配置文件是 */etc/sources/redis.yaml* ，可以在其中指定 redis 的连接信息等属性。
+
+```yaml
+default:
+  # the redis host address
+  addr: "127.0.0.1:6379"
+  # currently supports string and list only
+  datatype: "string"
+#  username: ""
+#  password: ""
+```
+
+在这个 yaml 文件的配置中，表将引用的 redis 实例地址是127.0.0.1:6379。值的类型是 "string"。
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sources/overview.md b/docs/zh_CN/rules/sources/overview.md
index 237726f9..94d1476b 100644
--- a/docs/zh_CN/rules/sources/overview.md
+++ b/docs/zh_CN/rules/sources/overview.md
@@ -12,6 +12,7 @@
 - [Http pull source](./builtin/http_pull.md)：从http服务器中拉取数据。
 - [Memory source](./builtin/memory.md)：从 eKuiper 内存主题读取数据以形成规则管道。
 - [File source](./builtin/file.md)：从文件中读取数据，通常用作表格。
+- [Redis source](./builtin/redis.md): 从 Redis 中查询数据，用作查询表。
 
 ## 预定义的源插件
 
@@ -23,6 +24,7 @@
 
 - [Zero MQ source](./plugin/zmq.md)：从Zero MQ读取数据。
 - [Random source](./plugin/random.md): 一个生成随机数据的源，用于测试。
+- [SQL source](./plugin/sql.md): 定期从关系数据库中拉取数据。
 
 ## 源的使用
 
diff --git a/docs/zh_CN/rules/sources/plugin/random.md b/docs/zh_CN/rules/sources/plugin/random.md
index c036ec73..a4883e32 100644
--- a/docs/zh_CN/rules/sources/plugin/random.md
+++ b/docs/zh_CN/rules/sources/plugin/random.md
@@ -1,5 +1,8 @@
 # 随机源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 随机源将生成具有指定样式的随机输入。
 
 ## 编译和部署插件
diff --git a/docs/zh_CN/rules/sources/plugin/sql.md b/docs/zh_CN/rules/sources/plugin/sql.md
index be5d0180..72fe3e1d 100644
--- a/docs/zh_CN/rules/sources/plugin/sql.md
+++ b/docs/zh_CN/rules/sources/plugin/sql.md
@@ -1,5 +1,9 @@
 # Sql 源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+<span style="background:green;color:white">lookup table source</span>
+
 源将定期查询数据库以获取数据流。
 
 ## 编译和部署插件
@@ -125,3 +129,28 @@ demo (
 ```
 
 将使用配置键 `template_config`
+
+## 查询表
+
+SQL 源支持成为一个查询表。我们可以使用创建表语句来创建一个 SQL 查询表。它将与实体关系数据库绑定并按需查询。
+
+```text
+CREATE TABLE alertTable() WITH (DATASOURCE="tableName", CONF_KEY="sqlite_config", TYPE="sql", KIND="lookup")
+```
+
+### 查询缓存
+
+查询外部数据库比在内存中计算要慢。如果吞吐量很高，可以使用查找缓存来提高性能。如果不启用查找缓存，那么所有的请求都被发送到外部数据库。当启用查找缓存时，每个查找表实例将持有一个缓存。当查询时，我们将首先查询缓存，然后再发送到外部数据库。
+
+缓存的配置在`sql.yaml`中。
+
+```yaml
+  lookup:
+    cache: true
+    cacheTtl: 600
+    cacheMissingKey: true
+```
+
+- cache: bool值，表示是否启用缓存。
+- cacheTtl: 缓存的生存时间，单位是秒。
+- cacheMissingKey：是否对空值进行缓存。
\ No newline at end of file
diff --git a/docs/zh_CN/rules/sources/plugin/zmq.md b/docs/zh_CN/rules/sources/plugin/zmq.md
index 340b47e8..8a5b3b79 100644
--- a/docs/zh_CN/rules/sources/plugin/zmq.md
+++ b/docs/zh_CN/rules/sources/plugin/zmq.md
@@ -1,5 +1,8 @@
 # Zmq 源
 
+<span style="background:green;color:white;">stream source</span>
+<span style="background:green;color:white">scan table source</span>
+
 源将订阅 Zero Mq 主题以将消息导入 eKuiper。
 
 ## 编译和部署插件
diff --git a/docs/zh_CN/sqls/tables.md b/docs/zh_CN/sqls/tables.md
index 29d3adf5..eb186d2b 100644
--- a/docs/zh_CN/sqls/tables.md
+++ b/docs/zh_CN/sqls/tables.md
@@ -2,7 +2,10 @@
 
 eKuiper 流是无界且不可变的，任何新数据都会附加到当前流中进行处理。 **Table** 用于表示流的当前状态。它可以被认为是流的快照。用户可以使用 table 来保留一批数据进行处理。
 
-在 eKuiper 中不允许单独使用表格。仅建议与流进行 join 操作。join 流时，表格将在新事件到来时不断更新。但是，只有到达流端的事件才会触发下游更新并产生连接输出。
+有两种类型的表。
+
+- 扫描表（Scan Table）：在内存中积累数据。它适用于较小的数据集，表的内容不需要在规则之间共享。
+- 查询表（Lookup Table）：绑定外部表并按需查询。它适用于更大的数据集，并且在规则之间共享表的内容。
 
 ## 语法定义
 
@@ -17,11 +20,34 @@ CREATE TABLE
 表支持与流相同的 [数据类型](./streams.md#数据类型)。
 表还支持所有[流的属性](./streams.md#语言定义)。因此，表中也支持所有源类型。许多源不是批处理的，它们在任何给定时间点都有一个事件，这意味着表将始终只有一个事件。一个附加属性 `RETAIN_SIZE` 来指定表快照的大小，以便表可以保存任意数量的历史数据。
 
+### 查询表的语法
+
+语法与创建普通扫描表相同，只需指定 `KIND` 的属性为 `lookup`。下面是一个创建查询表的例子，它会与 redis 的数据库 0 绑定。
+
+```sql
+CREATE TABLE alertTable() WITH (DATASOURCE="0", TYPE="redis", KIND="lookup")
+```
+
+目前，只有 `memory`、`redis` 和 `sql` 源可以作为查找表。
+
+### 表的属性
+
+| 属性名称       | 可选   | 描述                                                                                                                                                                      |
+|------------|------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| DATASOURCE | 否    | 取决于不同的源类型；如果是 MQTT 源，则为 MQTT 数据源主题名；其它源请参考相关的文档。                                                                                                                        |
+| FORMAT     | 是    | 传入的数据类型，支持 "JSON", "PROTOBUF" 和 "BINARY"，默认为 "JSON" 。关于 "BINARY" 类型的更多信息，请参阅 [Binary Stream](#二进制流)。该属性是否生效取决于源的类型，某些源自身解析的时固定私有格式的数据，则该配置不起作用。可支持该属性的源包括 MQTT 和 ZMQ 等。 |
+| SCHEMAID   | 是    | 解码时使用的模式，目前仅在格式为 PROTOBUF 的情况下使用。                                                                                                                                       |
+| KEY        | true | 表的主键。例如，对于SQL源，用于指定SQL表中的主键。并非所有的源类型都支持该属性。                                                                                                                             |
+| TYPE       | true | 源类型。每个源类型可以支持一种或两种表。请参考相关文件。                                                                                                                                            |
+| CONF_KEY   | 是    | 如果需要配置其他配置项，请在此处指定 config 键。 有关更多信息，请参见 [MQTT stream](../rules/sources/builtin/mqtt.md) 。                                                                               |
+| KIND       | true | 表的种类，可以是 `scan` 或 `lookup`。如果没有指定，默认值是`scan`。                                                                                                                           |
+
+
 ## 使用场景
 
 通常，表格将与带有或不带有窗口的流连接。与流连接时，表数据不会影响下游更新数据，它被视为静态引用数据，尽管它可能会在内部更新。
 
-### 查询表
+### 数据补全
 
 表的典型用法是作为查找表。示例 SQL 将类似于：
 ```sql
diff --git a/etc/sinks/influx2.json b/etc/sinks/influx2.json
deleted file mode 100644
index 7bf3bb51..00000000
--- a/etc/sinks/influx2.json
+++ /dev/null
@@ -1,135 +0,0 @@
-{
-  "about": {
-    "trial": true,
-    "author": {
-      "name": "elpsyr",
-      "email": "hellccqcq@gmail.com",
-      "company": "",
-      "website": "https://github.com/elpsyr"
-    },
-    "helpUrl": {
-      "en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/rules/sinks/plugin/influx2.md",
-      "zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sinks/plugin/influx2.md"
-    },
-    "description": {
-      "en_US": "This a sink plugin for InfluxDB2, it can be used for saving the analysis data into InfluxDB V2.X .",
-      "zh_CN": "本插件为 InfluxDB2 的持久化插件，可以用于将分析数据存入 InfluxDB V2.X 中"
-    }
-  },
-  "libs": [
-    "github.com/influxdata/influxdb-client-go/v2"
-  ],
-  "properties": [{
-    "name": "addr",
-    "default": "http://192.168.100.245:8086",
-    "optional": false,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The addr of the InfluxDB",
-      "zh_CN": "InfluxDB的地址"
-    },
-    "label": {
-      "en_US": "Addr",
-      "zh_CN": "地址"
-    }
-  }, {
-    "name": "measurement",
-    "default": "test",
-    "optional": true,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The measurement of the InfluxDB",
-      "zh_CN": "InfluxDB的测量"
-    },
-    "label": {
-      "en_US": "Measurement",
-      "zh_CN": "测量"
-    }
-  }, {
-    "name": "token",
-    "default": "",
-    "optional": false,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The InfluxDB api token",
-      "zh_CN": "InfluxDB 验证 api token"
-    },
-    "label": {
-      "en_US": "API Token",
-      "zh_CN": "API Token"
-    }
-  }, {
-    "name": "org",
-    "default": "",
-    "optional": false,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The InfluxDB org",
-      "zh_CN": "InfluxDB org"
-    },
-    "label": {
-      "en_US": "org",
-      "zh_CN": "组织"
-    }
-  }, {
-    "name": "bucket",
-    "default": "",
-    "optional": false,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The InfluxDB bucket",
-      "zh_CN": "InfluxDB bucket"
-    },
-    "label": {
-      "en_US": "bucket",
-      "zh_CN": "桶"
-    }
-  }, {
-    "name": "tagKey",
-    "default": "tagKey",
-    "optional": true,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The tag key of the InfluxDB",
-      "zh_CN": "InfluxDB 的标签键"
-    },
-    "label": {
-      "en_US": "Tag",
-      "zh_CN": "标签键"
-    }
-  }, {
-    "name": "tagValue",
-    "default": "tagValue",
-    "optional": true,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The tag value of the InfluxDB",
-      "zh_CN": "InfluxDB 的标签值"
-    },
-    "label": {
-      "en_US": "Tag value",
-      "zh_CN": "标签值"
-    }
-  }, {
-    "name": "fields",
-    "default": "humidity,temperature,pressure",
-    "optional": true,
-    "control": "text",
-    "type": "string",
-    "hint": {
-      "en_US": "The column of the InfluxDB",
-      "zh_CN": "InfluxDB 的列名"
-    },
-    "label": {
-      "en_US": "Column",
-      "zh_CN": "列名"
-    }
-  }]
-}
diff --git a/extensions/sources/sql/sql.yaml b/extensions/sources/sql/sql.yaml
index 78f6ed17..8e1a73db 100644
--- a/extensions/sources/sql/sql.yaml
+++ b/extensions/sources/sql/sql.yaml
@@ -13,6 +13,9 @@ default:
     cacheTtl: 600
     cacheMissingKey: true
 
+sqlite_config:
+  url: sqlite:/tmp/test.db
+
 sqlserver_config:
   url: sqlserver://username:password@140.210.204.147/testdb
   internalSqlQueryCfg:
diff --git a/internal/binder/function/function.go b/internal/binder/function/function.go
index 7499eea2..e6fac3f9 100644
--- a/internal/binder/function/function.go
+++ b/internal/binder/function/function.go
@@ -46,6 +46,19 @@ func init() {
 //	"count":   "",
 //}
 
+var analyticFuncs = map[string]struct{}{
+	"lag":         {},
+	"changed_col": {},
+	"had_changed": {},
+}
+
+const AnalyticPrefix = "$$a"
+
+func IsAnalyticFunc(name string) bool {
+	_, ok := analyticFuncs[name]
+	return ok
+}
+
 type funcExecutor struct{}
 
 func (f *funcExecutor) ValidateWithName(args []ast.Expr, name string) error {
diff --git a/internal/meta/yaml_config_ops.go b/internal/conf/yaml_config_ops.go
similarity index 62%
rename from internal/meta/yaml_config_ops.go
rename to internal/conf/yaml_config_ops.go
index 9688db74..83ebda07 100644
--- a/internal/meta/yaml_config_ops.go
+++ b/internal/conf/yaml_config_ops.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.
@@ -12,12 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package meta
+package conf
 
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"path"
@@ -30,7 +29,11 @@ type ConfKeysOperator interface {
 	GetPluginName() string
 	GetConfContentByte() ([]byte, error)
 	CopyConfContent() map[string]map[string]interface{}
+	CopyReadOnlyConfContent() map[string]map[string]interface{}
+	CopyUpdatableConfContent() map[string]map[string]interface{}
 	GetConfKeys() (keys []string)
+	GetReadOnlyConfKeys() (keys []string)
+	GetUpdatableConfKeys() (keys []string)
 	DeleteConfKey(confKey string)
 	DeleteConfKeyField(confKey string, reqField map[string]interface{}) error
 	AddConfKey(confKey string, reqField map[string]interface{}) error
@@ -45,12 +48,13 @@ type ConfigOperator interface {
 }
 
 // ConfigKeys implement ConfKeysOperator interface, load the configs from etc/sources/xx.yaml and et/connections/connection.yaml
-// Hold the connection configs for each connection type in cf field
+// Hold the connection configs for each connection type in etcCfg field
 // Provide method to query/add/update/delete the configs
 type ConfigKeys struct {
 	lock       sync.RWMutex
 	pluginName string                            // source type, can be mqtt/edgex/httppull
-	cf         map[string]map[string]interface{} // configs defined in yaml
+	etcCfg     map[string]map[string]interface{} // configs defined in etc/sources/yaml
+	dataCfg    map[string]map[string]interface{} // configs defined in etc/sources/
 }
 
 func (c *ConfigKeys) GetPluginName() string {
@@ -61,7 +65,15 @@ func (c *ConfigKeys) GetConfContentByte() ([]byte, error) {
 	cf := make(map[string]map[string]interface{})
 	c.lock.RLock()
 	defer c.lock.RUnlock()
-	for key, kvs := range c.cf {
+	for key, kvs := range c.etcCfg {
+		aux := make(map[string]interface{})
+		for k, v := range kvs {
+			aux[k] = v
+		}
+		cf[key] = aux
+	}
+
+	for key, kvs := range c.dataCfg {
 		aux := make(map[string]interface{})
 		for k, v := range kvs {
 			aux[k] = v
@@ -77,7 +89,47 @@ func (c *ConfigKeys) CopyConfContent() map[string]map[string]interface{} {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
-	for key, kvs := range c.cf {
+	for key, kvs := range c.etcCfg {
+		aux := make(map[string]interface{})
+		for k, v := range kvs {
+			aux[k] = v
+		}
+		cf[key] = aux
+	}
+
+	for key, kvs := range c.dataCfg {
+		aux := make(map[string]interface{})
+		for k, v := range kvs {
+			aux[k] = v
+		}
+		cf[key] = aux
+	}
+
+	return cf
+}
+
+func (c *ConfigKeys) CopyReadOnlyConfContent() map[string]map[string]interface{} {
+	cf := make(map[string]map[string]interface{})
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	for key, kvs := range c.etcCfg {
+		aux := make(map[string]interface{})
+		for k, v := range kvs {
+			aux[k] = v
+		}
+		cf[key] = aux
+	}
+
+	return cf
+}
+
+func (c *ConfigKeys) CopyUpdatableConfContent() map[string]map[string]interface{} {
+	cf := make(map[string]map[string]interface{})
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	for key, kvs := range c.dataCfg {
 		aux := make(map[string]interface{})
 		for k, v := range kvs {
 			aux[k] = v
@@ -89,10 +141,30 @@ func (c *ConfigKeys) CopyConfContent() map[string]map[string]interface{} {
 }
 
 func (c *ConfigKeys) GetConfKeys() (keys []string) {
+	ro := c.GetReadOnlyConfKeys()
+	keys = append(keys, ro...)
+
+	up := c.GetUpdatableConfKeys()
+	keys = append(keys, up...)
+
+	return keys
+}
+
+func (c *ConfigKeys) GetReadOnlyConfKeys() (keys []string) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	for k := range c.etcCfg {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+func (c *ConfigKeys) GetUpdatableConfKeys() (keys []string) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
-	for k := range c.cf {
+	for k := range c.dataCfg {
 		keys = append(keys, k)
 	}
 	return keys
@@ -102,7 +174,7 @@ func (c *ConfigKeys) DeleteConfKey(confKey string) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	delete(c.cf, confKey)
+	delete(c.dataCfg, confKey)
 }
 
 func recursionDelMap(cf, fields map[string]interface{}) error {
@@ -117,7 +189,6 @@ func recursionDelMap(cf, fields map[string]interface{}) error {
 				delete(cf, k)
 				continue
 			}
-
 			var auxCf map[string]interface{}
 			if err := cast.MapToStruct(cf[k], &auxCf); nil != err {
 				return fmt.Errorf(`%s%s.%s`, "type_conversion_fail", k, delKey)
@@ -147,7 +218,7 @@ func (c *ConfigKeys) DeleteConfKeyField(confKey string, reqField map[string]inte
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	err := recursionDelMap(c.cf[confKey], reqField)
+	err := recursionDelMap(c.dataCfg[confKey], reqField)
 	if nil != err {
 		return err
 	}
@@ -159,7 +230,11 @@ func (c *ConfigKeys) AddConfKey(confKey string, reqField map[string]interface{})
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	c.cf[confKey] = reqField
+	if _, ok := c.etcCfg[confKey]; ok {
+		return fmt.Errorf("duplicate key %s, already exist in etc folder", confKey)
+	}
+
+	c.dataCfg[confKey] = reqField
 
 	return nil
 }
@@ -168,12 +243,12 @@ func (c *ConfigKeys) AddConfKeyField(confKey string, reqField map[string]interfa
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	if nil == c.cf[confKey] {
+	if nil == c.dataCfg[confKey] {
 		return fmt.Errorf(`%s`, "not_found_confkey")
 	}
 
 	for k, v := range reqField {
-		c.cf[confKey][k] = v
+		c.dataCfg[confKey][k] = v
 	}
 	return nil
 }
@@ -189,18 +264,14 @@ func (c *SourceConfigKeysOps) IsSource() bool {
 
 func (c *SourceConfigKeysOps) SaveCfgToFile() error {
 	pluginName := c.pluginName
-	confDir, err := conf.GetConfLoc()
+	confDir, err := GetDataLoc()
 	if nil != err {
 		return err
 	}
 
 	dir := path.Join(confDir, "sources")
-	if "mqtt" == pluginName {
-		pluginName = "mqtt_source"
-		dir = confDir
-	}
 	filePath := path.Join(dir, pluginName+".yaml")
-	cfg := c.CopyConfContent()
+	cfg := c.CopyUpdatableConfContent()
 	err = filex.WriteYamlMarshal(filePath, cfg)
 	if nil != err {
 		return err
@@ -219,12 +290,12 @@ func (p *ConnectionConfigKeysOps) IsSource() bool {
 
 func (p *ConnectionConfigKeysOps) SaveCfgToFile() error {
 	pluginName := p.pluginName
-	confDir, err := conf.GetConfLoc()
+	confDir, err := GetDataLoc()
 	if nil != err {
 		return err
 	}
 
-	cfg := p.CopyConfContent()
+	cfg := p.CopyUpdatableConfContent()
 
 	yamlPath := path.Join(confDir, "connections/connection.yaml")
 
@@ -239,21 +310,34 @@ func (p *ConnectionConfigKeysOps) SaveCfgToFile() error {
 	return filex.WriteYamlMarshal(yamlPath, yamlData)
 }
 
-// NewConfigOperatorFromSourceYaml construct function, Load the configs from etc/sources/xx.yaml
-func NewConfigOperatorFromSourceYaml(pluginName string) (ConfigOperator, error) {
-	confDir, err := conf.GetConfLoc()
-	if nil != err {
-		return nil, err
+// NewConfigOperatorForSource construct function
+func NewConfigOperatorForSource(pluginName string) ConfigOperator {
+	c := &SourceConfigKeysOps{
+		&ConfigKeys{
+			lock:       sync.RWMutex{},
+			pluginName: pluginName,
+			etcCfg:     map[string]map[string]interface{}{},
+			dataCfg:    map[string]map[string]interface{}{},
+		},
 	}
+	return c
+}
 
+// NewConfigOperatorFromSourceYaml construct function, Load the configs from etc/sources/xx.yaml
+func NewConfigOperatorFromSourceYaml(pluginName string) (ConfigOperator, error) {
 	c := &SourceConfigKeysOps{
 		&ConfigKeys{
 			lock:       sync.RWMutex{},
 			pluginName: pluginName,
-			cf:         map[string]map[string]interface{}{},
+			etcCfg:     map[string]map[string]interface{}{},
+			dataCfg:    map[string]map[string]interface{}{},
 		},
 	}
 
+	confDir, err := GetConfLoc()
+	if nil != err {
+		return nil, err
+	}
 	dir := path.Join(confDir, "sources")
 	fileName := pluginName
 	if "mqtt" == pluginName {
@@ -261,42 +345,60 @@ func NewConfigOperatorFromSourceYaml(pluginName string) (ConfigOperator, error)
 		dir = confDir
 	}
 	filePath := path.Join(dir, fileName+`.yaml`)
-	err = filex.ReadYamlUnmarshal(filePath, &c.cf)
+	_ = filex.ReadYamlUnmarshal(filePath, &c.etcCfg)
+
+	dataDir, err := GetDataLoc()
 	if nil != err {
 		return nil, err
 	}
+	dir = path.Join(dataDir, "sources")
+	fileName = pluginName
+
+	filePath = path.Join(dir, fileName+`.yaml`)
+	_ = filex.ReadYamlUnmarshal(filePath, &c.dataCfg)
 
 	return c, nil
 }
 
+// NewConfigOperatorForConnection construct function
+func NewConfigOperatorForConnection(pluginName string) ConfigOperator {
+	c := &ConnectionConfigKeysOps{
+		&ConfigKeys{
+			lock:       sync.RWMutex{},
+			pluginName: pluginName,
+			etcCfg:     map[string]map[string]interface{}{},
+			dataCfg:    map[string]map[string]interface{}{},
+		},
+	}
+	return c
+}
+
 // NewConfigOperatorFromConnectionYaml construct function, Load the configs from et/connections/connection.yaml
 func NewConfigOperatorFromConnectionYaml(pluginName string) (ConfigOperator, error) {
+	c := &ConnectionConfigKeysOps{
+		&ConfigKeys{
+			lock:       sync.RWMutex{},
+			pluginName: pluginName,
+			etcCfg:     map[string]map[string]interface{}{},
+			dataCfg:    map[string]map[string]interface{}{},
+		},
+	}
 
-	confDir, err := conf.GetConfLoc()
+	confDir, err := GetConfLoc()
 	if nil != err {
 		return nil, err
 	}
-
 	yamlPath := path.Join(confDir, "connections/connection.yaml")
 	yamlData := make(map[string]interface{})
 	err = filex.ReadYamlUnmarshal(yamlPath, &yamlData)
 	if nil != err {
 		return nil, err
 	}
-
-	c := &ConnectionConfigKeysOps{
-		&ConfigKeys{
-			lock:       sync.RWMutex{},
-			pluginName: pluginName,
-			cf:         map[string]map[string]interface{}{},
-		},
-	}
-
 	if plgCnfs, ok := yamlData[pluginName]; ok {
 		if cf, ok1 := plgCnfs.(map[string]interface{}); ok1 {
 			for confKey, confVal := range cf {
 				if conf, ok := confVal.(map[string]interface{}); ok {
-					c.cf[confKey] = conf
+					c.etcCfg[confKey] = conf
 				} else {
 					return nil, fmt.Errorf("file content is not right: %s.%v", confKey, confVal)
 				}
@@ -308,5 +410,26 @@ func NewConfigOperatorFromConnectionYaml(pluginName string) (ConfigOperator, err
 		return nil, fmt.Errorf("not find the target connection type: %s", c.pluginName)
 	}
 
+	confDir, err = GetDataLoc()
+	if nil != err {
+		return nil, err
+	}
+	yamlPath = path.Join(confDir, "connections/connection.yaml")
+	yamlData = make(map[string]interface{})
+	_ = filex.ReadYamlUnmarshal(yamlPath, &yamlData)
+
+	if plgCnfs, ok := yamlData[pluginName]; ok {
+		if cf, ok1 := plgCnfs.(map[string]interface{}); ok1 {
+			for confKey, confVal := range cf {
+				if conf, ok := confVal.(map[string]interface{}); ok {
+					c.dataCfg[confKey] = conf
+				} else {
+					return nil, fmt.Errorf("file content is not right: %s.%v", confKey, confVal)
+				}
+			}
+		} else {
+			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 b3ccc03f..52adc617 100644
--- a/internal/meta/func_meta.go
+++ b/internal/meta/func_meta.go
@@ -104,6 +104,27 @@ func ReadFuncMetaDir(checker InstallChecker) error {
 			return err
 		}
 	}
+
+	confDir, err = conf.GetDataLoc()
+	if nil != err {
+		return err
+	}
+
+	dir = path.Join(confDir, "functions")
+	files, err = os.ReadDir(dir)
+	if nil != err {
+		return err
+	}
+	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
 }
 
diff --git a/internal/meta/sinkMeta.go b/internal/meta/sinkMeta.go
index b5b14be0..0a33c49b 100644
--- a/internal/meta/sinkMeta.go
+++ b/internal/meta/sinkMeta.go
@@ -205,6 +205,28 @@ func ReadSinkMetaDir(checker InstallChecker) error {
 			return err
 		}
 	}
+
+	confDir, err = conf.GetDataLoc()
+	if nil != err {
+		return err
+	}
+
+	dir = path.Join(confDir, "sinks")
+	files, err = os.ReadDir(dir)
+	if nil != err {
+		return err
+	}
+	for _, file := range files {
+		fname := file.Name()
+		if !strings.HasSuffix(fname, ".json") {
+			continue
+		}
+
+		filePath := path.Join(dir, fname)
+		if err := ReadSinkMetaFile(filePath, checker(strings.TrimSuffix(fname, ".json"))); nil != err {
+			return err
+		}
+	}
 	return nil
 }
 
diff --git a/internal/meta/sourceMeta.go b/internal/meta/sourceMeta.go
index aada28bc..4bf4de7c 100644
--- a/internal/meta/sourceMeta.go
+++ b/internal/meta/sourceMeta.go
@@ -106,6 +106,7 @@ func ReadSourceMetaFile(filePath string, installed bool) error {
 }
 
 func ReadSourceMetaDir(checker InstallChecker) error {
+	//load etc/sources meta data
 	confDir, err := conf.GetConfLoc()
 	if nil != err {
 		return err
@@ -132,6 +133,30 @@ func ReadSourceMetaDir(checker InstallChecker) error {
 			conf.Log.Infof("Loading metadata file for source : %s", fileName)
 		}
 	}
+
+	//load data/sources meta data
+	confDir, err = conf.GetDataLoc()
+	if nil != err {
+		return err
+	}
+
+	dir = path.Join(confDir, "sources")
+	dirEntries, err = os.ReadDir(dir)
+	if nil != err {
+		return err
+	}
+
+	for _, entry := range dirEntries {
+		fileName := entry.Name()
+		if strings.HasSuffix(fileName, ".json") {
+			filePath := path.Join(dir, fileName)
+			if err = ReadSourceMetaFile(filePath, checker(strings.TrimSuffix(fileName, ".json"))); nil != err {
+				return err
+			}
+			conf.Log.Infof("Loading metadata file for source : %s", fileName)
+		}
+	}
+
 	return nil
 }
 
diff --git a/internal/meta/yamlConfigMeta.go b/internal/meta/yamlConfigMeta.go
index aa0cd4e1..3f53e021 100644
--- a/internal/meta/yamlConfigMeta.go
+++ b/internal/meta/yamlConfigMeta.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.
@@ -23,7 +23,7 @@ import (
 
 type configManager struct {
 	lock         sync.RWMutex
-	cfgOperators map[string]ConfigOperator
+	cfgOperators map[string]conf.ConfigOperator
 }
 
 //ConfigManager Hold the ConfigOperator for yaml configs defined in etc/sources/xxx.yaml and etc/connections/connection.yaml
@@ -31,7 +31,7 @@ type configManager struct {
 // for configs in etc/connections/connection.yaml, the map key is connections.xxx format, xxx will be mqtt/edgex
 var ConfigManager = configManager{
 	lock:         sync.RWMutex{},
-	cfgOperators: make(map[string]ConfigOperator),
+	cfgOperators: make(map[string]conf.ConfigOperator),
 }
 
 const SourceCfgOperatorKeyTemplate = "sources.%s"
@@ -43,7 +43,7 @@ const ConnectionCfgOperatorKeyTemplate = "connections.%s"
 func loadConfigOperatorForSource(pluginName string) {
 	yamlKey := fmt.Sprintf(SourceCfgOperatorKeyTemplate, pluginName)
 
-	if cfg, _ := NewConfigOperatorFromSourceYaml(pluginName); cfg != nil {
+	if cfg, _ := conf.NewConfigOperatorFromSourceYaml(pluginName); cfg != nil {
 		ConfigManager.lock.Lock()
 		ConfigManager.cfgOperators[yamlKey] = cfg
 		ConfigManager.lock.Unlock()
@@ -57,7 +57,7 @@ func loadConfigOperatorForSource(pluginName string) {
 func loadConfigOperatorForConnection(pluginName string) {
 	yamlKey := fmt.Sprintf(ConnectionCfgOperatorKeyTemplate, pluginName)
 
-	if cfg, _ := NewConfigOperatorFromConnectionYaml(pluginName); cfg != nil {
+	if cfg, _ := conf.NewConfigOperatorFromConnectionYaml(pluginName); cfg != nil {
 		ConfigManager.lock.Lock()
 		ConfigManager.cfgOperators[yamlKey] = cfg
 		ConfigManager.lock.Unlock()
@@ -123,18 +123,12 @@ func AddSourceConfKey(plgName, confKey, language string, content []byte) error {
 		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "type_conversion_fail"), plgName, err)
 	}
 
-	var cfgOps ConfigOperator
+	var cfgOps conf.ConfigOperator
 	var found bool
 
 	cfgOps, found = ConfigManager.cfgOperators[configOperatorKey]
 	if !found {
-		cfgOps = &SourceConfigKeysOps{
-			ConfigKeys: &ConfigKeys{
-				lock:       sync.RWMutex{},
-				pluginName: plgName,
-				cf:         map[string]map[string]interface{}{},
-			},
-		}
+		cfgOps = conf.NewConfigOperatorForSource(plgName)
 		ConfigManager.cfgOperators[configOperatorKey] = cfgOps
 	}
 
@@ -161,18 +155,12 @@ func AddConnectionConfKey(plgName, confKey, language string, content []byte) err
 		return fmt.Errorf(`%s%s.%v`, getMsg(language, source, "type_conversion_fail"), plgName, err)
 	}
 
-	var cfgOps ConfigOperator
+	var cfgOps conf.ConfigOperator
 	var found bool
 
 	cfgOps, found = ConfigManager.cfgOperators[configOperatorKey]
 	if !found {
-		cfgOps = &ConnectionConfigKeysOps{
-			ConfigKeys: &ConfigKeys{
-				lock:       sync.RWMutex{},
-				pluginName: plgName,
-				cf:         map[string]map[string]interface{}{},
-			},
-		}
+		cfgOps = conf.NewConfigOperatorForConnection(plgName)
 		ConfigManager.cfgOperators[configOperatorKey] = cfgOps
 	}
 
diff --git a/internal/plugin/native/manager.go b/internal/plugin/native/manager.go
index 76049662..3eb4139e 100644
--- a/internal/plugin/native/manager.go
+++ b/internal/plugin/native/manager.go
@@ -57,8 +57,8 @@ type Manager struct {
 	// loaded symbols in current runtime
 	runtime map[string]*plugin.Plugin
 	// dirs
-	pluginDir string
-	etcDir    string
+	pluginDir     string
+	pluginConfDir string
 	// the access to db
 	db kv.KeyValue
 }
@@ -69,15 +69,15 @@ func InitManager() (*Manager, error) {
 	if err != nil {
 		return nil, fmt.Errorf("cannot find plugins folder: %s", err)
 	}
-	etcDir, err := conf.GetConfLoc()
+	dataDir, err := conf.GetDataLoc()
 	if err != nil {
-		return nil, fmt.Errorf("cannot find etc folder: %s", err)
+		return nil, fmt.Errorf("cannot find data folder: %s", err)
 	}
 	err, db := store.GetKV("pluginFuncs")
 	if err != nil {
 		return nil, fmt.Errorf("error when opening db: %v", err)
 	}
-	registry := &Manager{symbols: make(map[string]string), db: db, pluginDir: pluginDir, etcDir: etcDir, runtime: make(map[string]*plugin.Plugin)}
+	registry := &Manager{symbols: make(map[string]string), db: db, pluginDir: pluginDir, pluginConfDir: dataDir, runtime: make(map[string]*plugin.Plugin)}
 	manager = registry
 	plugins := make([]map[string]string, 3)
 	for i := range plugin2.PluginTypes {
@@ -282,15 +282,15 @@ func (rr *Manager) Register(t plugin2.PluginType, j plugin2.Plugin) error {
 
 	switch t {
 	case plugin2.SINK:
-		if err := meta.ReadSinkMetaFile(path.Join(rr.etcDir, plugin2.PluginTypes[t], name+`.json`), true); nil != err {
+		if err := meta.ReadSinkMetaFile(path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], name+`.json`), true); nil != err {
 			conf.Log.Errorf("readSinkFile:%v", err)
 		}
 	case plugin2.SOURCE:
-		if err := meta.ReadSourceMetaFile(path.Join(rr.etcDir, plugin2.PluginTypes[t], name+`.json`), true); nil != err {
+		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.etcDir, plugin2.PluginTypes[t], name+`.json`), true); nil != err {
+		if err := meta.ReadFuncMetaFile(path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], name+`.json`), true); nil != err {
 			conf.Log.Errorf("readFuncFile:%v", err)
 		}
 	}
@@ -331,7 +331,7 @@ func (rr *Manager) Delete(t plugin2.PluginType, name string, stop bool) error {
 		soPath,
 	}
 	// Find etc folder
-	etcPath := path.Join(rr.etcDir, plugin2.PluginTypes[t], name)
+	etcPath := path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], name)
 	if fi, err := os.Stat(etcPath); err == nil {
 		if fi.Mode().IsDir() {
 			paths = append(paths, etcPath)
@@ -339,16 +339,16 @@ func (rr *Manager) Delete(t plugin2.PluginType, name string, stop bool) error {
 	}
 	switch t {
 	case plugin2.SOURCE:
-		paths = append(paths, path.Join(rr.etcDir, plugin2.PluginTypes[t], name+".yaml"))
-		srcJsonPath := path.Join(rr.etcDir, plugin2.PluginTypes[plugin2.SOURCE], name+".json")
+		paths = append(paths, path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], name+".yaml"))
+		srcJsonPath := path.Join(rr.pluginConfDir, plugin2.PluginTypes[plugin2.SOURCE], name+".json")
 		_ = os.RemoveAll(srcJsonPath)
 		meta.UninstallSource(name)
 	case plugin2.SINK:
-		sinkJsonPaths := path.Join(rr.etcDir, plugin2.PluginTypes[plugin2.SINK], name+".json")
+		sinkJsonPaths := path.Join(rr.pluginConfDir, plugin2.PluginTypes[plugin2.SINK], name+".json")
 		_ = os.RemoveAll(sinkJsonPaths)
 		meta.UninstallSink(name)
 	case plugin2.FUNCTION:
-		funcJsonPath := path.Join(rr.etcDir, plugin2.PluginTypes[plugin2.FUNCTION], name+".json")
+		funcJsonPath := path.Join(rr.pluginConfDir, plugin2.PluginTypes[plugin2.FUNCTION], name+".json")
 		_ = os.RemoveAll(funcJsonPath)
 		old := make([]string, 0)
 		if ok, err := rr.db.Get(name, &old); err != nil {
@@ -439,7 +439,7 @@ func (rr *Manager) install(t plugin2.PluginType, name, src string, shellParas []
 	expFiles := 1
 	if t == plugin2.SOURCE {
 		yamlFile = name + ".yaml"
-		yamlPath = path.Join(rr.etcDir, plugin2.PluginTypes[t], yamlFile)
+		yamlPath = path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], yamlFile)
 		expFiles = 2
 	}
 	var revokeFiles []string
@@ -460,7 +460,7 @@ func (rr *Manager) install(t plugin2.PluginType, name, src string, shellParas []
 			revokeFiles = append(revokeFiles, yamlPath)
 			filenames = append(filenames, yamlPath)
 		} else if fileName == name+".json" {
-			jsonPath := path.Join(rr.etcDir, plugin2.PluginTypes[t], fileName)
+			jsonPath := path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], fileName)
 			if err := filex.UnzipTo(file, jsonPath); nil != err {
 				conf.Log.Errorf("Failed to decompress the metadata %s file", fileName)
 			} else {
@@ -476,7 +476,7 @@ func (rr *Manager) install(t plugin2.PluginType, name, src string, shellParas []
 			revokeFiles = append(revokeFiles, soPath)
 			soName, version = parseName(fileName)
 		} else if strings.HasPrefix(fileName, "etc/") {
-			err = filex.UnzipTo(file, path.Join(rr.etcDir, plugin2.PluginTypes[t], strings.Replace(fileName, "etc", name, 1)))
+			err = filex.UnzipTo(file, path.Join(rr.pluginConfDir, plugin2.PluginTypes[t], strings.Replace(fileName, "etc", name, 1)))
 			if err != nil {
 				return version, err
 			}
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index c695a047..01c66721 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.go
@@ -38,9 +38,9 @@ import (
 var manager *Manager
 
 type Manager struct {
-	pluginDir string
-	etcDir    string
-	reg       *registry // can be replaced with kv
+	pluginDir     string
+	pluginConfDir string
+	reg           *registry // can be replaced with kv
 }
 
 // InitManager must only be called once
@@ -49,9 +49,9 @@ func InitManager() (*Manager, error) {
 	if err != nil {
 		return nil, fmt.Errorf("cannot find plugins folder: %s", err)
 	}
-	etcDir, err := conf.GetConfLoc()
+	dataDir, err := conf.GetDataLoc()
 	if err != nil {
-		return nil, fmt.Errorf("cannot find etc folder: %s", err)
+		return nil, fmt.Errorf("cannot find data folder: %s", err)
 	}
 	registry := &registry{
 		RWMutex:   sync.RWMutex{},
@@ -63,9 +63,9 @@ func InitManager() (*Manager, error) {
 	// Read plugin info from file system
 	pluginDir = filepath.Join(pluginDir, "portable")
 	m := &Manager{
-		pluginDir: pluginDir,
-		etcDir:    etcDir,
-		reg:       registry,
+		pluginDir:     pluginDir,
+		pluginConfDir: dataDir,
+		reg:           registry,
 	}
 	err = m.syncRegistry()
 	if err != nil {
@@ -133,17 +133,17 @@ func (m *Manager) doRegister(name string, pi *PluginInfo, isInit bool) error {
 
 	if !isInit {
 		for _, s := range pi.Sources {
-			if err := meta.ReadSourceMetaFile(path.Join(m.etcDir, plugin.PluginTypes[plugin.SOURCE], s+`.json`), true); nil != err {
+			if err := meta.ReadSourceMetaFile(path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.SOURCE], s+`.json`), true); nil != err {
 				conf.Log.Errorf("read source json file:%v", err)
 			}
 		}
 		for _, s := range pi.Sinks {
-			if err := meta.ReadSinkMetaFile(path.Join(m.etcDir, plugin.PluginTypes[plugin.SINK], s+`.json`), true); nil != err {
+			if err := meta.ReadSinkMetaFile(path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.SINK], s+`.json`), true); nil != err {
 				conf.Log.Errorf("read sink json file:%v", err)
 			}
 		}
 		for _, s := range pi.Functions {
-			if err := meta.ReadFuncMetaFile(path.Join(m.etcDir, plugin.PluginTypes[plugin.FUNCTION], s+`.json`), true); nil != err {
+			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)
 			}
 		}
@@ -270,7 +270,7 @@ func (m *Manager) install(name, src string, shellParas []string) (resultErr erro
 	for _, file := range r.File {
 		fileName := file.Name
 		if strings.HasPrefix(fileName, "sources/") || strings.HasPrefix(fileName, "sinks/") || strings.HasPrefix(fileName, "functions/") {
-			target = path.Join(m.etcDir, fileName)
+			target = path.Join(m.pluginConfDir, fileName)
 		} else {
 			target = path.Join(pluginTarget, fileName)
 			if fileName == "install.sh" {
@@ -354,19 +354,19 @@ func (m *Manager) Delete(name string) error {
 	m.reg.Delete(name)
 	// delete files and uninstall metas
 	for _, s := range pinfo.Sources {
-		p := path.Join(m.etcDir, plugin.PluginTypes[plugin.SOURCE], s+".yaml")
+		p := path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.SOURCE], s+".yaml")
 		os.Remove(p)
-		p = path.Join(m.etcDir, plugin.PluginTypes[plugin.SOURCE], s+".json")
+		p = path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.SOURCE], s+".json")
 		os.Remove(p)
 		meta.UninstallSource(s)
 	}
 	for _, s := range pinfo.Sinks {
-		p := path.Join(m.etcDir, plugin.PluginTypes[plugin.SINK], s+".json")
+		p := path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.SINK], s+".json")
 		os.Remove(p)
 		meta.UninstallSink(s)
 	}
 	for _, s := range pinfo.Functions {
-		p := path.Join(m.etcDir, plugin.PluginTypes[plugin.FUNCTION], s+".json")
+		p := path.Join(m.pluginConfDir, plugin.PluginTypes[plugin.FUNCTION], s+".json")
 		os.Remove(p)
 		meta.UninstallFunc(s)
 	}
diff --git a/internal/server/rest.go b/internal/server/rest.go
index d9221add..391a88b9 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -17,6 +17,7 @@ package server
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
 	"io"
 	"net/http"
 	"os"
@@ -28,7 +29,6 @@ import (
 
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
-	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/server/middleware"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
@@ -97,16 +97,11 @@ func jsonResponse(i interface{}, w http.ResponseWriter, logger api.Logger) {
 }
 
 func createRestServer(ip string, port int, needToken bool) *http.Server {
-	// Create upload path for upload api
-	etcDir, err := conf.GetConfLoc()
-	if err != nil {
-		panic(err)
-	}
-	uploadDir = filepath.Join(etcDir, "uploads")
-	err = os.MkdirAll(uploadDir, os.ModePerm)
+	dataDir, err := conf.GetDataLoc()
 	if err != nil {
 		panic(err)
 	}
+	uploadDir = filepath.Join(dataDir, "uploads")
 
 	r := mux.NewRouter()
 	r.HandleFunc("/", rootHandler).Methods(http.MethodGet, http.MethodPost)
diff --git a/internal/server/server.go b/internal/server/server.go
index 78f42263..724ddfff 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -27,6 +27,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"sort"
 	"syscall"
 	"time"
@@ -40,10 +41,41 @@ var (
 	streamProcessor *processor.StreamProcessor
 )
 
+func createPaths() {
+	dataDir, err := conf.GetDataLoc()
+	if err != nil {
+		panic(err)
+	}
+	dirs := []string{"uploads", "sources", "sinks", "functions", "services", "services/schemas", "connections"}
+
+	for _, v := range dirs {
+		// Create dir if not exist
+		realDir := filepath.Join(dataDir, v)
+		if _, err := os.Stat(realDir); os.IsNotExist(err) {
+			if err := os.MkdirAll(realDir, os.ModePerm); err != nil {
+				panic(err)
+			}
+		}
+	}
+
+	files := []string{"connections/connection.yaml"}
+	for _, v := range files {
+		// Create dir if not exist
+		realFile := filepath.Join(dataDir, v)
+		if _, err := os.Stat(realFile); os.IsNotExist(err) {
+			if _, err := os.Create(realFile); err != nil {
+				panic(err)
+			}
+		}
+	}
+
+}
+
 func StartUp(Version, LoadFileType string) {
 	version = Version
 	conf.LoadFileType = LoadFileType
 	startTimeStamp = time.Now().Unix()
+	createPaths()
 	conf.InitConf()
 	factory.InitClientsFactory()
 
diff --git a/internal/service/manager.go b/internal/service/manager.go
index 1bbfb6dc..b41150a4 100644
--- a/internal/service/manager.go
+++ b/internal/service/manager.go
@@ -52,7 +52,7 @@ func InitManager() (*Manager, error) {
 	mutex.Lock()
 	defer mutex.Unlock()
 	if singleton == nil {
-		dir := "etc/services"
+		dir := "data/services"
 		if kconf.IsTesting {
 			dir = "service/test"
 		}
diff --git a/internal/service/schema.go b/internal/service/schema.go
index 94d4befc..c01ab209 100644
--- a/internal/service/schema.go
+++ b/internal/service/schema.go
@@ -69,7 +69,7 @@ var ( //Do not call these directly, use the get methods
 
 func ProtoParser() *protoparse.Parser {
 	once.Do(func() {
-		dir := "etc/services/schemas/"
+		dir := "data/services/schemas/"
 		if kconf.IsTesting {
 			dir = "service/test/schemas/"
 		}
diff --git a/internal/topo/node/conf/source.go b/internal/topo/node/conf/source.go
index 7d59a9be..a3836d49 100644
--- a/internal/topo/node/conf/source.go
+++ b/internal/topo/node/conf/source.go
@@ -22,29 +22,24 @@ import (
 
 func GetSourceConf(sourceType string, options *ast.Options) map[string]interface{} {
 	confkey := options.CONF_KEY
-	confPath := "sources/" + sourceType + ".yaml"
-	if sourceType == "mqtt" {
-		confPath = "mqtt_source.yaml"
+
+	yamlOps, err := conf.NewConfigOperatorFromSourceYaml(sourceType)
+	if err != nil {
+		conf.Log.Warnf("fail to parse yaml for source %s. Return error %v", sourceType, err)
 	}
 	props := make(map[string]interface{})
-	cfg := make(map[string]interface{})
-	err := conf.LoadConfigByName(confPath, &cfg)
-	if err != nil {
+	cfg := yamlOps.CopyConfContent()
+	if len(cfg) == 0 {
 		conf.Log.Warnf("fail to parse yaml for source %s. Return an empty configuration", sourceType)
 	} else {
 		def, ok := cfg["default"]
 		if !ok {
 			conf.Log.Warnf("default conf %s is not found", confkey)
 		} else {
-			if def1, ok1 := def.(map[string]interface{}); ok1 {
-				props = def1
-			}
+			props = def
 			if c, ok := cfg[strings.ToLower(confkey)]; ok {
-				if c1, ok := c.(map[string]interface{}); ok {
-					c2 := c1
-					for k, v := range c2 {
-						props[k] = v
-					}
+				for k, v := range c {
+					props[k] = v
 				}
 			}
 		}
diff --git a/internal/topo/operator/analyticfuncs_operator.go b/internal/topo/operator/analyticfuncs_operator.go
new file mode 100644
index 00000000..54e114f2
--- /dev/null
+++ b/internal/topo/operator/analyticfuncs_operator.go
@@ -0,0 +1,61 @@
+// 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 operator
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/xsql"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+)
+
+type AnalyticFuncsOp struct {
+	Funcs []*ast.Call
+}
+
+func (p *AnalyticFuncsOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{} {
+	ctx.GetLogger().Debugf("AnalyticFuncsOp receive: %s", data)
+	switch input := data.(type) {
+	case error:
+		return input
+	case xsql.TupleRow:
+		ve := &xsql.ValuerEval{Valuer: xsql.MultiValuer(input, fv)}
+		for _, f := range p.Funcs {
+			result := ve.Eval(f)
+			if e, ok := result.(error); ok {
+				return e
+			}
+			input.Set(f.CachedField, result)
+		}
+	case xsql.SingleCollection:
+		err := input.RangeSet(func(_ int, row xsql.Row) (bool, error) {
+			ve := &xsql.ValuerEval{Valuer: xsql.MultiValuer(row, &xsql.WindowRangeValuer{WindowRange: input.GetWindowRange()}, fv, &xsql.WildcardValuer{Data: row})}
+			for _, f := range p.Funcs {
+				result := ve.Eval(f)
+				if e, ok := result.(error); ok {
+					return false, e
+				}
+				row.Set(f.CachedField, result)
+			}
+			return true, nil
+		})
+		if err != nil {
+			return err
+		}
+	default:
+		return fmt.Errorf("run analytic funcs op error: invalid input %[1]T(%[1]v)", input)
+	}
+	return data
+}
diff --git a/internal/topo/planner/analyticFuncsPlan.go b/internal/topo/planner/analyticFuncsPlan.go
new file mode 100644
index 00000000..dd402f42
--- /dev/null
+++ b/internal/topo/planner/analyticFuncsPlan.go
@@ -0,0 +1,40 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package planner
+
+import "github.com/lf-edge/ekuiper/pkg/ast"
+
+type AnalyticFuncsPlan struct {
+	baseLogicalPlan
+	funcs []*ast.Call
+}
+
+func (p AnalyticFuncsPlan) Init() *AnalyticFuncsPlan {
+	p.baseLogicalPlan.self = &p
+	return &p
+}
+
+// PushDownPredicate this op must run before any filters
+func (p *AnalyticFuncsPlan) PushDownPredicate(condition ast.Expr) (ast.Expr, LogicalPlan) {
+	return condition, p
+}
+
+func (p *AnalyticFuncsPlan) PruneColumns(fields []ast.Expr) error {
+	for _, f := range p.funcs {
+		ff := getFields(f)
+		fields = append(fields, ff...)
+	}
+	return p.baseLogicalPlan.PruneColumns(fields)
+}
diff --git a/internal/topo/planner/analyzer.go b/internal/topo/planner/analyzer.go
index 1d10c0ff..a1856d95 100644
--- a/internal/topo/planner/analyzer.go
+++ b/internal/topo/planner/analyzer.go
@@ -25,14 +25,14 @@ import (
 
 // Analyze the select statement by decorating the info from stream statement.
 // Typically, set the correct stream name for fieldRefs
-func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt, error) {
+func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt, []*ast.Call, error) {
 	streamsFromStmt := xsql.GetStreams(s)
 	streamStmts := make([]*ast.StreamStmt, len(streamsFromStmt))
 	isSchemaless := false
 	for i, s := range streamsFromStmt {
 		streamStmt, err := xsql.GetDataSource(store, s)
 		if err != nil {
-			return nil, fmt.Errorf("fail to get stream %s, please check if stream is created", s)
+			return nil, nil, fmt.Errorf("fail to get stream %s, please check if stream is created", s)
 		}
 		streamStmts[i] = streamStmt
 		// TODO fine grain control of schemaless
@@ -55,8 +55,9 @@ func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt,
 		}
 	}
 	var (
-		walkErr     error
-		aliasFields []*ast.Field
+		walkErr       error
+		aliasFields   []*ast.Field
+		analyticFuncs []*ast.Call
 	)
 	// Scan columns fields: bind all field refs, collect alias
 	for i, f := range s.Fields {
@@ -68,7 +69,7 @@ func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt,
 			return true
 		})
 		if walkErr != nil {
-			return nil, walkErr
+			return nil, nil, walkErr
 		}
 		if f.AName != "" {
 			aliasFields = append(aliasFields, &s.Fields[i])
@@ -88,7 +89,7 @@ func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt,
 			walkErr = fieldsMap.save(f.AName, ast.AliasStream, ar)
 		}
 	}
-	// bind field ref for alias AND set StreamName for all field ref
+	// Bind field ref for alias AND set StreamName for all field ref
 	ast.WalkFunc(s, func(n ast.Node) bool {
 		switch f := n.(type) {
 		case ast.Fields: // do not bind selection fields, should have done above
@@ -113,10 +114,31 @@ func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*ast.StreamStmt,
 		return true
 	})
 	if walkErr != nil {
-		return nil, walkErr
+		return nil, nil, walkErr
 	}
 	walkErr = validate(s)
-	return streamStmts, walkErr
+	// Collect all analytic function calls so that we can let them run firstly
+	ast.WalkFunc(s, func(n ast.Node) bool {
+		switch f := n.(type) {
+		case *ast.Call:
+			if function.IsAnalyticFunc(f.Name) {
+				f.CachedField = fmt.Sprintf("%s_%s_%d", function.AnalyticPrefix, f.Name, f.FuncId)
+				f.Cached = true
+				analyticFuncs = append(analyticFuncs, &ast.Call{
+					Name:        f.Name,
+					FuncId:      f.FuncId,
+					FuncType:    f.FuncType,
+					Args:        f.Args,
+					CachedField: f.CachedField,
+				})
+			}
+		}
+		return true
+	})
+	if walkErr != nil {
+		return nil, nil, walkErr
+	}
+	return streamStmts, analyticFuncs, walkErr
 }
 
 func validate(s *ast.SelectStatement) (err error) {
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index 00b1a048..0f01ed1b 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -155,6 +155,8 @@ func buildOps(lp LogicalPlan, tp *topo.Topo, options *api.RuleOption, sources []
 			inputs = []api.Emitter{srcNode}
 			op = srcNode
 		}
+	case *AnalyticFuncsPlan:
+		op = Transform(&operator.AnalyticFuncsOp{Funcs: t.funcs}, fmt.Sprintf("%d_analytic", newIndex), options)
 	case *WindowPlan:
 		if t.condition != nil {
 			wfilterOp := Transform(&operator.FilterOp{Condition: t.condition}, fmt.Sprintf("%d_windowFilter", newIndex), options)
@@ -222,7 +224,7 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *api.RuleOption, store kv.
 		ds                  ast.Dimensions
 	)
 
-	streamStmts, err := decorateStmt(stmt, store)
+	streamStmts, analyticFuncs, err := decorateStmt(stmt, store)
 	if err != nil {
 		return nil, err
 	}
@@ -248,6 +250,13 @@ func createLogicalPlan(stmt *ast.SelectStatement, opt *api.RuleOption, store kv.
 			}
 		}
 	}
+	if len(analyticFuncs) > 0 {
+		p = AnalyticFuncsPlan{
+			funcs: analyticFuncs,
+		}.Init()
+		p.SetChildren(children)
+		children = []LogicalPlan{p}
+	}
 	if dimensions != nil {
 		w = dimensions.GetWindow()
 		if w != nil {
diff --git a/internal/xsql/valuer.go b/internal/xsql/valuer.go
index 0eaf90a7..e301ae7d 100644
--- a/internal/xsql/valuer.go
+++ b/internal/xsql/valuer.go
@@ -270,6 +270,15 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 		}
 		return &BracketEvalResult{Start: ii, End: ii}
 	case *ast.Call:
+		// The analytic function are calculated prior to all ops, so just get the cached field value
+		if expr.Cached {
+			val, ok := v.Valuer.Value(expr.CachedField, "")
+			if ok {
+				return val
+			} else {
+				return fmt.Errorf("call %s error: %v", expr.Name, val)
+			}
+		}
 		if _, ok := implicitValueFuncs[expr.Name]; ok {
 			if vv, ok := v.Valuer.(FuncValuer); ok {
 				val, ok := vv.FuncValue(expr.Name)
diff --git a/pkg/ast/expr.go b/pkg/ast/expr.go
index 689b13db..7fbf7417 100644
--- a/pkg/ast/expr.go
+++ b/pkg/ast/expr.go
@@ -136,6 +136,11 @@ type Call struct {
 	FuncId   int
 	FuncType FuncType
 	Args     []Expr
+	// This is used for analytic functions.
+	// In planner, all analytic functions are planned to calculate in analytic_op which produce a new field.
+	// This cachedField cached the new field name and when evaluating, just return the field access evaluated value.
+	CachedField string
+	Cached      bool
 }
 
 func (c *Call) expr()    {}


Test patch:
diff --git a/internal/meta/yaml_config_ops_test.go b/internal/conf/yaml_config_ops_test.go
similarity index 98%
rename from internal/meta/yaml_config_ops_test.go
rename to internal/conf/yaml_config_ops_test.go
index b3031cf6..a5907ea3 100644
--- a/internal/meta/yaml_config_ops_test.go
+++ b/internal/conf/yaml_config_ops_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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package meta
+package conf
 
 import (
 	"encoding/json"
diff --git a/internal/meta/yamlConfigMeta_test.go b/internal/meta/yamlConfigMeta_test.go
index 80fbeec8..338fa1ac 100644
--- a/internal/meta/yamlConfigMeta_test.go
+++ b/internal/meta/yamlConfigMeta_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.
@@ -15,37 +15,47 @@
 package meta
 
 import (
-	"fmt"
-	"sync"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"os"
+	"path/filepath"
 	"testing"
 )
 
-type MockSourcesConfigOps struct {
-	*ConfigKeys
-}
+func createPaths() {
+	dataDir, err := conf.GetDataLoc()
+	if err != nil {
+		panic(err)
+	}
+	dirs := []string{"sources", "sinks", "functions", "services", "services/schemas", "connections"}
 
-func (m MockSourcesConfigOps) IsSource() bool {
-	return true
-}
+	for _, v := range dirs {
+		// Create dir if not exist
+		realDir := filepath.Join(dataDir, v)
+		if _, err := os.Stat(realDir); os.IsNotExist(err) {
+			if err := os.MkdirAll(realDir, os.ModePerm); err != nil {
+				panic(err)
+			}
+		}
+	}
+
+	files := []string{"connections/connection.yaml"}
+	for _, v := range files {
+		// Create dir if not exist
+		realFile := filepath.Join(dataDir, v)
+		if _, err := os.Stat(realFile); os.IsNotExist(err) {
+			if _, err := os.Create(realFile); err != nil {
+				panic(err)
+			}
+		}
+	}
 
-func (m MockSourcesConfigOps) SaveCfgToFile() error {
-	return nil
 }
 
 func TestYamlConfigMeta_Ops(t *testing.T) {
-	plgName := "mocksource"
+	createPaths()
 
-	yamlKey := fmt.Sprintf(SourceCfgOperatorKeyTemplate, plgName)
+	plgName := "mocksource"
 	addData := `{"url":"127.0.0.1","method":"post","headers":{"Accept":"json"}}`
-
-	ConfigManager.cfgOperators[yamlKey] = MockSourcesConfigOps{
-		&ConfigKeys{
-			sync.RWMutex{},
-			plgName,
-			make(map[string]map[string]interface{}),
-		},
-	}
-
 	// init new ConfigOperator, success
 	err := AddSourceConfKey(plgName, "new", "en_US", []byte(addData))
 	if err != nil {
diff --git a/internal/plugin/native/manager_test.go b/internal/plugin/native/manager_test.go
index fa2a61e9..150151dd 100644
--- a/internal/plugin/native/manager_test.go
+++ b/internal/plugin/native/manager_test.go
@@ -141,7 +141,7 @@ func TestManager_Register(t *testing.T) {
 		if !reflect.DeepEqual(tt.err, err) {
 			t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.err, err)
 		} else if tt.err == nil {
-			err := checkFile(manager.pluginDir, manager.etcDir, tt.t, tt.n, tt.v, tt.lowerSo)
+			err := checkFile(manager.pluginDir, manager.pluginConfDir, tt.t, tt.n, tt.v, tt.lowerSo)
 			if err != nil {
 				t.Errorf("%d: error : %s\n\n", i, err)
 			}
diff --git a/internal/plugin/portable/manager_test.go b/internal/plugin/portable/manager_test.go
index 18b1e636..35dcdb54 100644
--- a/internal/plugin/portable/manager_test.go
+++ b/internal/plugin/portable/manager_test.go
@@ -83,7 +83,7 @@ func TestManager_Install(t *testing.T) {
 		if !reflect.DeepEqual(tt.err, err) {
 			t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.err, err)
 		} else if tt.err == nil {
-			err := checkFileForMirror(manager.pluginDir, manager.etcDir, true)
+			err := checkFileForMirror(manager.pluginDir, manager.pluginConfDir, true)
 			if err != nil {
 				t.Errorf("%d: error : %s\n\n", i, err)
 			}
@@ -164,7 +164,7 @@ func TestDelete(t *testing.T) {
 	if err != nil {
 		t.Errorf("delete plugin error: %v", err)
 	}
-	err = checkFileForMirror(manager.pluginDir, manager.etcDir, false)
+	err = checkFileForMirror(manager.pluginDir, manager.pluginConfDir, false)
 	if err != nil {
 		t.Errorf("error : %s\n\n", err)
 	}
diff --git a/internal/topo/operator/analyticfuncs_operator_test.go b/internal/topo/operator/analyticfuncs_operator_test.go
new file mode 100644
index 00000000..536db349
--- /dev/null
+++ b/internal/topo/operator/analyticfuncs_operator_test.go
@@ -0,0 +1,197 @@
+// 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 operator
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/state"
+	"github.com/lf-edge/ekuiper/internal/xsql"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"reflect"
+	"strconv"
+	"testing"
+)
+
+func TestAnalyticFuncs(t *testing.T) {
+	var tests = []struct {
+		funcs  []*ast.Call
+		data   []interface{}
+		result [][]map[string]interface{}
+	}{
+		{ // 0 Lag test
+			funcs: []*ast.Call{
+				{
+					Name: "lag",
+					Args: []ast.Expr{
+						&ast.FieldRef{Name: "a"},
+					},
+					FuncId:      0,
+					CachedField: "$$a_lag_0",
+				},
+				{
+					Name: "lag",
+					Args: []ast.Expr{
+						&ast.FieldRef{Name: "b"},
+					},
+					FuncId:      1,
+					CachedField: "$$a_lag_1",
+				},
+			},
+			data: []interface{}{
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"b": "b1",
+						"c": "c1",
+					},
+				},
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"b": "b2",
+						"c": "c1",
+					},
+				},
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"c": "c1",
+					},
+				},
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"b": "b2",
+						"c": "c2",
+					},
+				},
+			},
+			result: [][]map[string]interface{}{{{
+				"$$a_lag_0": nil,
+				"$$a_lag_1": nil,
+				"a":         "a1", "b": "b1", "c": "c1",
+			}}, {{
+				"$$a_lag_0": "a1", "$$a_lag_1": "b1", "a": "a1", "b": "b2", "c": "c1",
+			}}, {{
+				"$$a_lag_0": "a1", "$$a_lag_1": "b2", "a": "a1", "c": "c1",
+			}}, {{
+				"$$a_lag_0": "a1", "$$a_lag_1": interface{}(nil), "a": "a1", "b": "b2", "c": "c2",
+			}}},
+		},
+		{ // 1 changed test
+			funcs: []*ast.Call{
+				{
+					Name: "changed_col",
+					Args: []ast.Expr{
+						&ast.BooleanLiteral{Val: false},
+						&ast.FieldRef{Name: "a"},
+					},
+					FuncId:      0,
+					CachedField: "$$a_changed_col_0",
+				},
+				{
+					Name: "lag",
+					Args: []ast.Expr{
+						&ast.FieldRef{Name: "b"},
+					},
+					FuncId:      1,
+					CachedField: "$$a_lag_1",
+				},
+				{
+					Name: "had_changed",
+					Args: []ast.Expr{
+						&ast.BooleanLiteral{Val: true},
+						&ast.FieldRef{Name: "c"},
+					},
+					FuncId:      0,
+					CachedField: "$$a_had_changed_0",
+				},
+			},
+			data: []interface{}{
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"b": "b1",
+					},
+				},
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"c": "c1",
+					},
+				},
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"c": "c1",
+					},
+				},
+				&xsql.Tuple{
+					Emitter: "test",
+					Message: xsql.Message{
+						"a": "a1",
+						"b": "b2",
+						"c": "c2",
+					},
+				},
+			},
+			result: [][]map[string]interface{}{
+				{{
+					"$$a_changed_col_0": "a1", "$$a_had_changed_0": false, "$$a_lag_1": nil, "a": "a1", "b": "b1",
+				}}, {{
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": "b1", "a": "a1", "c": "c1",
+				}}, {{
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": false, "$$a_lag_1": nil, "a": "a1", "c": "c1",
+				}}, {{
+					"$$a_changed_col_0": nil, "$$a_had_changed_0": true, "$$a_lag_1": nil, "a": "a1", "b": "b2", "c": "c2",
+				}},
+			},
+		},
+	}
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	contextLogger := conf.Log.WithField("rule", "TestChangedFuncs_Apply1")
+
+	for i, tt := range tests {
+		tempStore, _ := state.CreateStore("mockRule"+strconv.Itoa(i), api.AtMostOnce)
+		ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger).WithMeta("mockRule"+strconv.Itoa(i), "project", tempStore)
+
+		pp := &AnalyticFuncsOp{Funcs: tt.funcs}
+		fv, afv := xsql.NewFunctionValuersForOp(ctx)
+		r := make([][]map[string]interface{}, 0, len(tt.data))
+		for _, d := range tt.data {
+			opResult := pp.Apply(ctx, d, fv, afv)
+			result, err := parseResult(opResult, false)
+			if err != nil {
+				t.Errorf("parse result error： %s", err)
+				continue
+			}
+			r = append(r, result)
+		}
+
+		if !reflect.DeepEqual(tt.result, r) {
+			t.Errorf("%d.\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.result, r)
+		}
+	}
+}
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index e2de868f..96cb6128 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -1101,6 +1101,86 @@ func Test_createLogicalPlan(t *testing.T) {
 				isAggregate: false,
 				sendMeta:    false,
 			}.Init(),
+		}, { // 13 analytic function plan
+			sql: `SELECT lag(name), id1 FROM src1 WHERE lag(temp) > temp`,
+			p: ProjectPlan{
+				baseLogicalPlan: baseLogicalPlan{
+					children: []LogicalPlan{
+						FilterPlan{
+							baseLogicalPlan: baseLogicalPlan{
+								children: []LogicalPlan{
+									AnalyticFuncsPlan{
+										baseLogicalPlan: baseLogicalPlan{
+											children: []LogicalPlan{
+												DataSourcePlan{
+													name: "src1",
+													streamFields: []interface{}{
+														&ast.StreamField{
+															Name:      "id1",
+															FieldType: &ast.BasicType{Type: ast.BIGINT},
+														},
+														&ast.StreamField{
+															Name:      "name",
+															FieldType: &ast.BasicType{Type: ast.STRINGS},
+														},
+														&ast.StreamField{
+															Name:      "temp",
+															FieldType: &ast.BasicType{Type: ast.BIGINT},
+														},
+													},
+													streamStmt: streams["src1"],
+													metaFields: []string{},
+												}.Init(),
+											},
+										},
+										funcs: []*ast.Call{
+											{
+												Name: "lag", FuncId: 0, CachedField: "$$a_lag_0", FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}},
+											}, {
+												Name:        "lag",
+												FuncId:      1,
+												CachedField: "$$a_lag_1",
+												Args: []ast.Expr{&ast.FieldRef{
+													Name:       "temp",
+													StreamName: "src1",
+												}},
+											},
+										},
+									}.Init(),
+								},
+							},
+							condition: &ast.BinaryExpr{
+								LHS: &ast.Call{
+									Name:   "lag",
+									FuncId: 1,
+									Args: []ast.Expr{&ast.FieldRef{
+										Name:       "temp",
+										StreamName: "src1",
+									}},
+									CachedField: "$$a_lag_1",
+									Cached:      true,
+								},
+								OP: ast.GT,
+								RHS: &ast.FieldRef{
+									Name:       "temp",
+									StreamName: "src1",
+								},
+							},
+						}.Init(),
+					},
+				},
+				fields: []ast.Field{
+					{
+						Expr: &ast.Call{Name: "lag", FuncId: 0, FuncType: ast.FuncTypeScalar, Args: []ast.Expr{&ast.FieldRef{Name: "name", StreamName: "src1"}}, CachedField: "$$a_lag_0", Cached: true},
+						Name: "lag",
+					}, {
+						Expr: &ast.FieldRef{Name: "id1", StreamName: "src1"},
+						Name: "id1",
+					},
+				},
+				isAggregate: false,
+				sendMeta:    false,
+			}.Init(),
 		},
 	}
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
diff --git a/internal/topo/topotest/rule_test.go b/internal/topo/topotest/rule_test.go
index a2f1163e..4c479a3d 100644
--- a/internal/topo/topotest/rule_test.go
+++ b/internal/topo/topotest/rule_test.go
@@ -573,7 +573,7 @@ func TestSingleSQL(t *testing.T) {
 		},
 		{
 			Name: `TestLagAlias`,
-			Sql:  "SELECT lag(size) as lastSize, size, lastSize/size as changeRate FROM demo",
+			Sql:  "SELECT lag(size) as lastSize, size, lastSize/size as changeRate FROM demo WHERE size > 2",
 			R: [][]map[string]interface{}{
 				{{
 					"size": float64(3),
@@ -583,31 +583,16 @@ func TestSingleSQL(t *testing.T) {
 					"size":       float64(6),
 					"changeRate": float64(0),
 				}},
-				{{
-					"lastSize":   float64(6),
-					"size":       float64(2),
-					"changeRate": float64(3),
-				}},
 				{{
 					"lastSize":   float64(2),
 					"size":       float64(4),
 					"changeRate": float64(0),
 				}},
-				{{
-					"lastSize":   float64(4),
-					"size":       float64(1),
-					"changeRate": float64(4),
-				}},
 			},
 			M: map[string]interface{}{
-				"op_2_project_0_exceptions_total":   int64(0),
-				"op_2_project_0_process_latency_us": int64(0),
-				"op_2_project_0_records_in_total":   int64(5),
-				"op_2_project_0_records_out_total":  int64(5),
-
 				"sink_mockSink_0_exceptions_total":  int64(0),
-				"sink_mockSink_0_records_in_total":  int64(5),
-				"sink_mockSink_0_records_out_total": int64(5),
+				"sink_mockSink_0_records_in_total":  int64(3),
+				"sink_mockSink_0_records_out_total": int64(3),
 
 				"source_demo_0_exceptions_total":  int64(0),
 				"source_demo_0_records_in_total":  int64(5),
diff --git a/test/lookup_table_redis.jmx b/test/lookup_table_redis.jmx
new file mode 100644
index 00000000..b13b1a8e
--- /dev/null
+++ b/test/lookup_table_redis.jmx
@@ -0,0 +1,918 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<jmeterTestPlan version="1.2" properties="5.0" jmeter="5.3">
+  <hashTree>
+    <TestPlan guiclass="TestPlanGui" testclass="TestPlan" testname="Test Plan" enabled="true">
+      <stringProp name="TestPlan.comments"></stringProp>
+      <boolProp name="TestPlan.functional_mode">false</boolProp>
+      <boolProp name="TestPlan.tearDown_on_shutdown">true</boolProp>
+      <boolProp name="TestPlan.serialize_threadgroups">false</boolProp>
+      <elementProp name="TestPlan.user_defined_variables" elementType="Arguments" guiclass="ArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+        <collectionProp name="Arguments.arguments"/>
+      </elementProp>
+      <stringProp name="TestPlan.user_define_classpath"></stringProp>
+    </TestPlan>
+    <hashTree>
+      <Arguments guiclass="ArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+        <collectionProp name="Arguments.arguments">
+          <elementProp name="srv" elementType="Argument">
+            <stringProp name="Argument.name">srv</stringProp>
+            <stringProp name="Argument.value">127.0.0.1</stringProp>
+            <stringProp name="Argument.metadata">=</stringProp>
+          </elementProp>
+          <elementProp name="rest_port" elementType="Argument">
+            <stringProp name="Argument.name">rest_port</stringProp>
+            <stringProp name="Argument.value">9081</stringProp>
+            <stringProp name="Argument.metadata">=</stringProp>
+          </elementProp>
+          <elementProp name="mqtt_srv" elementType="Argument">
+            <stringProp name="Argument.name">mqtt_srv</stringProp>
+            <stringProp name="Argument.value">127.0.0.1</stringProp>
+            <stringProp name="Argument.metadata">=</stringProp>
+          </elementProp>
+        </collectionProp>
+      </Arguments>
+      <hashTree/>
+      <ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="Rules" enabled="true">
+        <stringProp name="ThreadGroup.on_sample_error">continue</stringProp>
+        <elementProp name="ThreadGroup.main_controller" elementType="LoopController" guiclass="LoopControlPanel" testclass="LoopController" testname="Loop Controller" enabled="true">
+          <boolProp name="LoopController.continue_forever">false</boolProp>
+          <stringProp name="LoopController.loops">1</stringProp>
+        </elementProp>
+        <stringProp name="ThreadGroup.num_threads">1</stringProp>
+        <stringProp name="ThreadGroup.ramp_time">1</stringProp>
+        <boolProp name="ThreadGroup.scheduler">false</boolProp>
+        <stringProp name="ThreadGroup.duration"></stringProp>
+        <stringProp name="ThreadGroup.delay"></stringProp>
+        <boolProp name="ThreadGroup.same_user_on_next_iteration">true</boolProp>
+      </ThreadGroup>
+      <hashTree>
+        <ResultCollector guiclass="ViewResultsFullVisualizer" testclass="ResultCollector" testname="View Results Tree" enabled="true">
+          <boolProp name="ResultCollector.error_logging">false</boolProp>
+          <objProp>
+            <name>saveConfig</name>
+            <value class="SampleSaveConfiguration">
+              <time>true</time>
+              <latency>true</latency>
+              <timestamp>true</timestamp>
+              <success>true</success>
+              <label>true</label>
+              <code>true</code>
+              <message>true</message>
+              <threadName>true</threadName>
+              <dataType>true</dataType>
+              <encoding>false</encoding>
+              <assertions>true</assertions>
+              <subresults>true</subresults>
+              <responseData>false</responseData>
+              <samplerData>false</samplerData>
+              <xml>false</xml>
+              <fieldNames>true</fieldNames>
+              <responseHeaders>false</responseHeaders>
+              <requestHeaders>false</requestHeaders>
+              <responseDataOnError>false</responseDataOnError>
+              <saveAssertionResultsFailureMessage>true</saveAssertionResultsFailureMessage>
+              <assertionsResultsToSave>0</assertionsResultsToSave>
+              <bytes>true</bytes>
+              <sentBytes>true</sentBytes>
+              <threadCounts>true</threadCounts>
+              <idleTime>true</idleTime>
+              <connectTime>true</connectTime>
+            </value>
+          </objProp>
+          <stringProp name="filename"></stringProp>
+        </ResultCollector>
+        <hashTree/>
+        <TransactionController guiclass="TransactionControllerGui" testclass="TransactionController" testname="API" enabled="true">
+          <boolProp name="TransactionController.includeTimers">false</boolProp>
+          <boolProp name="TransactionController.parent">false</boolProp>
+        </TransactionController>
+        <hashTree>
+          <net.xmeter.samplers.ConnectSampler guiclass="net.xmeter.gui.ConnectSamplerUI" testclass="net.xmeter.samplers.ConnectSampler" testname="MQTT Connect" enabled="true">
+            <stringProp name="mqtt.server">${mqtt_srv}</stringProp>
+            <stringProp name="mqtt.port">1883</stringProp>
+            <stringProp name="mqtt.version">3.1</stringProp>
+            <stringProp name="mqtt.conn_timeout">10</stringProp>
+            <boolProp name="mqtt.private_protocol">false</boolProp>
+            <stringProp name="mqtt.listener_timeout">10</stringProp>
+            <stringProp name="mqtt.protocol">TCP</stringProp>
+            <boolProp name="mqtt.dual_ssl_authentication">false</boolProp>
+            <stringProp name="mqtt.keystore_file_path"></stringProp>
+            <stringProp name="mqtt.keystore_password"></stringProp>
+            <stringProp name="mqtt.clientcert_file_path"></stringProp>
+            <stringProp name="mqtt.clientcert_password"></stringProp>
+            <stringProp name="mqtt.user_name"></stringProp>
+            <stringProp name="mqtt.password"></stringProp>
+            <stringProp name="mqtt.client_id_prefix">conn_</stringProp>
+            <boolProp name="mqtt.client_id_suffix">true</boolProp>
+            <stringProp name="mqtt.conn_keep_alive">300</stringProp>
+            <stringProp name="mqtt.conn_attampt_max">0</stringProp>
+            <stringProp name="mqtt.reconn_attampt_max">0</stringProp>
+            <boolProp name="mqtt.conn_clean_session">true</boolProp>
+          </net.xmeter.samplers.ConnectSampler>
+          <hashTree/>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateAlertSettingStream" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&quot;sql&quot;:&quot;CREATE STREAM alertStream() WITH (DATASOURCE=\&quot;scene/alert\&quot;, FORMAT=\&quot;json\&quot;, TYPE=\&quot;mqtt\&quot;)&quot;}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+            <stringProp name="TestPlan.comments">Create stream for alert setting actions</stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="2128378328">Stream alertStream is created.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateRuleUpdateAlert" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&#xd;
+  &quot;id&quot;: &quot;ruleUpdateAlert&quot;,&#xd;
+    &quot;sql&quot;:&quot;SELECT * FROM alertStream&quot;,&#xd;
+    &quot;actions&quot;:[&#xd;
+      {&#xd;
+        &quot;redis&quot;: {&#xd;
+          &quot;addr&quot;: &quot;127.0.0.1:6379&quot;,&#xd;
+          &quot;dataType&quot;: &quot;string&quot;,&#xd;
+          &quot;field&quot;: &quot;id&quot;,&#xd;
+          &quot;rowkindField&quot;: &quot;action&quot;,&#xd;
+          &quot;sendSingle&quot;: true&#xd;
+        }&#xd;
+      }&#xd;
+    ]&#xd;
+}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-1451809660">Rule ruleUpdateAlert was created</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateDataStream" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&quot;sql&quot;:&quot;CREATE STREAM demoStream() WITH (DATASOURCE=\&quot;scene/data\&quot;, FORMAT=\&quot;json\&quot;, TYPE=\&quot;mqtt\&quot;)&quot;}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+            <stringProp name="TestPlan.comments">Create stream for alert setting actions</stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-894755937">Stream demoStream is created.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateRedisTable" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&quot;sql&quot;:&quot;CREATE TABLE alertTable() WITH (DATASOURCE=\&quot;0\&quot;, TYPE=\&quot;redis\&quot;, KIND=\&quot;lookup\&quot;)&quot;}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/tables</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-1341540896">Table alertTable is created.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateRuleAlert" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&#xd;
+  &quot;id&quot;: &quot;ruleAlert&quot;,&#xd;
+  &quot;sql&quot;:&quot;SELECT device, value FROM demoStream INNER JOIN alertTable ON demoStream.deviceKind = alertTable.id WHERE demoStream.value &gt; alertTable.alarm&quot;,&#xd;
+  &quot;actions&quot;:[&#xd;
+    {&#xd;
+      &quot;mqtt&quot;: {&#xd;
+        &quot;server&quot;: &quot;tcp://${mqtt_srv}:1883&quot;,&#xd;
+        &quot;topic&quot;: &quot;rule/alert&quot;,&#xd;
+        &quot;sendSingle&quot;: true&#xd;
+      }&#xd;
+    }&#xd;
+  ]&#xd;
+}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-2079532467">Rule ruleAlert was created</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_GetRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_demoStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">0</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="FeedTable" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">alert_init_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">action,id,alarm</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/alert</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;id&quot;: ${id}, &quot;action&quot;: ${action}, &quot;alarm&quot; : ${alarm}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="GetTableRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleUpdateAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_alertStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">4</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_memory_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">4</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+              <stringProp name="ConstantTimer.delay">5000</stringProp>
+            </ConstantTimer>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="SendData" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">collect_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">device,deviceKind,value</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/data</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;deviceKind&quot;: ${deviceKind}, &quot;device&quot;: ${device}, &quot;value&quot; : ${value}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_GetRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_demoStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">4</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_mqtt_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">2</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="UpdateTable" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">alert_update_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">action,id,alarm</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/alert</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;id&quot;: ${id}, &quot;action&quot;: ${action}, &quot;alarm&quot; : ${alarm}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="GetTableRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleUpdateAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_alertStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">8</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_memory_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">8</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+              <stringProp name="ConstantTimer.delay">5000</stringProp>
+            </ConstantTimer>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="SendData" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">collect_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">device,deviceKind,value</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/data</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;deviceKind&quot;: ${deviceKind}, &quot;device&quot;: ${device}, &quot;value&quot; : ${value}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_GetRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_demoStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">8</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_mqtt_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">3</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_DropRule" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="659914816">Rule ruleAlert is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_Drop_Stream" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams/demoStream</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="1148079559">Stream demoStream is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_Drop_Table" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/tables/alertTable</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="701294600">Table alertTable is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_DropRule" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleUpdateAlert</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="1287637623">Rule ruleUpdateAlert is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_Drop_Stream" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams/alertStream</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-123753472">Stream alertStream is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+            <stringProp name="ConstantTimer.delay">500</stringProp>
+          </ConstantTimer>
+          <hashTree/>
+        </hashTree>
+      </hashTree>
+      <ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="Result" enabled="true">
+        <stringProp name="ThreadGroup.on_sample_error">continue</stringProp>
+        <elementProp name="ThreadGroup.main_controller" elementType="LoopController" guiclass="LoopControlPanel" testclass="LoopController" testname="Loop Controller" enabled="true">
+          <boolProp name="LoopController.continue_forever">false</boolProp>
+          <stringProp name="LoopController.loops">3</stringProp>
+        </elementProp>
+        <stringProp name="ThreadGroup.num_threads">1</stringProp>
+        <stringProp name="ThreadGroup.ramp_time">1</stringProp>
+        <boolProp name="ThreadGroup.scheduler">false</boolProp>
+        <stringProp name="ThreadGroup.duration"></stringProp>
+        <stringProp name="ThreadGroup.delay"></stringProp>
+        <boolProp name="ThreadGroup.same_user_on_next_iteration">true</boolProp>
+      </ThreadGroup>
+      <hashTree>
+        <ResultCollector guiclass="ViewResultsFullVisualizer" testclass="ResultCollector" testname="View Results Tree" enabled="true">
+          <boolProp name="ResultCollector.error_logging">false</boolProp>
+          <objProp>
+            <name>saveConfig</name>
+            <value class="SampleSaveConfiguration">
+              <time>true</time>
+              <latency>true</latency>
+              <timestamp>true</timestamp>
+              <success>true</success>
+              <label>true</label>
+              <code>true</code>
+              <message>true</message>
+              <threadName>true</threadName>
+              <dataType>true</dataType>
+              <encoding>false</encoding>
+              <assertions>true</assertions>
+              <subresults>true</subresults>
+              <responseData>false</responseData>
+              <samplerData>false</samplerData>
+              <xml>false</xml>
+              <fieldNames>true</fieldNames>
+              <responseHeaders>false</responseHeaders>
+              <requestHeaders>false</requestHeaders>
+              <responseDataOnError>false</responseDataOnError>
+              <saveAssertionResultsFailureMessage>true</saveAssertionResultsFailureMessage>
+              <assertionsResultsToSave>0</assertionsResultsToSave>
+              <bytes>true</bytes>
+              <sentBytes>true</sentBytes>
+              <threadCounts>true</threadCounts>
+              <idleTime>true</idleTime>
+              <connectTime>true</connectTime>
+            </value>
+          </objProp>
+          <stringProp name="filename"></stringProp>
+        </ResultCollector>
+        <hashTree/>
+        <OnceOnlyController guiclass="OnceOnlyControllerGui" testclass="OnceOnlyController" testname="Once Only Controller" enabled="true"/>
+        <hashTree>
+          <net.xmeter.samplers.ConnectSampler guiclass="net.xmeter.gui.ConnectSamplerUI" testclass="net.xmeter.samplers.ConnectSampler" testname="MQTT Connect" enabled="true">
+            <stringProp name="mqtt.server">${mqtt_srv}</stringProp>
+            <stringProp name="mqtt.port">1883</stringProp>
+            <stringProp name="mqtt.version">3.1</stringProp>
+            <stringProp name="mqtt.conn_timeout">10</stringProp>
+            <boolProp name="mqtt.private_protocol">false</boolProp>
+            <stringProp name="mqtt.listener_timeout">10</stringProp>
+            <stringProp name="mqtt.protocol">TCP</stringProp>
+            <boolProp name="mqtt.dual_ssl_authentication">false</boolProp>
+            <stringProp name="mqtt.keystore_file_path"></stringProp>
+            <stringProp name="mqtt.keystore_password"></stringProp>
+            <stringProp name="mqtt.clientcert_file_path"></stringProp>
+            <stringProp name="mqtt.clientcert_password"></stringProp>
+            <stringProp name="mqtt.user_name"></stringProp>
+            <stringProp name="mqtt.password"></stringProp>
+            <stringProp name="mqtt.client_id_prefix">conn_</stringProp>
+            <boolProp name="mqtt.client_id_suffix">true</boolProp>
+            <stringProp name="mqtt.conn_keep_alive">300</stringProp>
+            <stringProp name="mqtt.conn_attampt_max">0</stringProp>
+            <stringProp name="mqtt.reconn_attampt_max">0</stringProp>
+            <boolProp name="mqtt.conn_clean_session">true</boolProp>
+          </net.xmeter.samplers.ConnectSampler>
+          <hashTree/>
+        </hashTree>
+        <net.xmeter.samplers.SubSampler guiclass="net.xmeter.gui.SubSamplerUI" testclass="net.xmeter.samplers.SubSampler" testname="AnalysisResult" enabled="true">
+          <stringProp name="mqtt.topic_name">rule/alert</stringProp>
+          <stringProp name="mqtt.qos_level">0</stringProp>
+          <boolProp name="mqtt.add_timestamp">false</boolProp>
+          <boolProp name="mqtt.debug_response">true</boolProp>
+          <stringProp name="mqtt.sample_condition">number of received messages</stringProp>
+          <stringProp name="mqtt.sample_condition_value">1</stringProp>
+        </net.xmeter.samplers.SubSampler>
+        <hashTree>
+          <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="value Assertion" enabled="true">
+            <stringProp name="JSON_PATH">$.value</stringProp>
+            <stringProp name="EXPECTED_VALUE">${value}</stringProp>
+            <boolProp name="JSONVALIDATION">true</boolProp>
+            <boolProp name="EXPECT_NULL">false</boolProp>
+            <boolProp name="INVERT">false</boolProp>
+            <boolProp name="ISREGEX">false</boolProp>
+          </JSONPathAssertion>
+          <hashTree/>
+          <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="device Assertion" enabled="true">
+            <stringProp name="JSON_PATH">$.device</stringProp>
+            <stringProp name="EXPECTED_VALUE">${device}</stringProp>
+            <boolProp name="JSONVALIDATION">true</boolProp>
+            <boolProp name="EXPECT_NULL">false</boolProp>
+            <boolProp name="INVERT">false</boolProp>
+            <boolProp name="ISREGEX">false</boolProp>
+          </JSONPathAssertion>
+          <hashTree/>
+        </hashTree>
+        <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+          <stringProp name="delimiter">,</stringProp>
+          <stringProp name="fileEncoding"></stringProp>
+          <stringProp name="filename">lookup_table_result_data.txt</stringProp>
+          <boolProp name="ignoreFirstLine">false</boolProp>
+          <boolProp name="quotedData">false</boolProp>
+          <boolProp name="recycle">true</boolProp>
+          <stringProp name="shareMode">shareMode.thread</stringProp>
+          <boolProp name="stopThread">false</boolProp>
+          <stringProp name="variableNames">device,value</stringProp>
+        </CSVDataSet>
+        <hashTree/>
+      </hashTree>
+    </hashTree>
+  </hashTree>
+</jmeterTestPlan>
diff --git a/test/lookup_table_sql.jmx b/test/lookup_table_sql.jmx
new file mode 100644
index 00000000..bfdc3b6a
--- /dev/null
+++ b/test/lookup_table_sql.jmx
@@ -0,0 +1,1070 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<jmeterTestPlan version="1.2" properties="5.0" jmeter="5.3">
+  <hashTree>
+    <TestPlan guiclass="TestPlanGui" testclass="TestPlan" testname="Test Plan" enabled="true">
+      <stringProp name="TestPlan.comments"></stringProp>
+      <boolProp name="TestPlan.functional_mode">false</boolProp>
+      <boolProp name="TestPlan.tearDown_on_shutdown">true</boolProp>
+      <boolProp name="TestPlan.serialize_threadgroups">false</boolProp>
+      <elementProp name="TestPlan.user_defined_variables" elementType="Arguments" guiclass="ArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+        <collectionProp name="Arguments.arguments"/>
+      </elementProp>
+      <stringProp name="TestPlan.user_define_classpath"></stringProp>
+    </TestPlan>
+    <hashTree>
+      <Arguments guiclass="ArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+        <collectionProp name="Arguments.arguments">
+          <elementProp name="srv" elementType="Argument">
+            <stringProp name="Argument.name">srv</stringProp>
+            <stringProp name="Argument.value">127.0.0.1</stringProp>
+            <stringProp name="Argument.metadata">=</stringProp>
+          </elementProp>
+          <elementProp name="rest_port" elementType="Argument">
+            <stringProp name="Argument.name">rest_port</stringProp>
+            <stringProp name="Argument.value">9081</stringProp>
+            <stringProp name="Argument.metadata">=</stringProp>
+          </elementProp>
+          <elementProp name="mqtt_srv" elementType="Argument">
+            <stringProp name="Argument.name">mqtt_srv</stringProp>
+            <stringProp name="Argument.value">127.0.0.1</stringProp>
+            <stringProp name="Argument.metadata">=</stringProp>
+          </elementProp>
+        </collectionProp>
+      </Arguments>
+      <hashTree/>
+      <ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="Rules" enabled="true">
+        <stringProp name="ThreadGroup.on_sample_error">continue</stringProp>
+        <elementProp name="ThreadGroup.main_controller" elementType="LoopController" guiclass="LoopControlPanel" testclass="LoopController" testname="Loop Controller" enabled="true">
+          <boolProp name="LoopController.continue_forever">false</boolProp>
+          <stringProp name="LoopController.loops">1</stringProp>
+        </elementProp>
+        <stringProp name="ThreadGroup.num_threads">1</stringProp>
+        <stringProp name="ThreadGroup.ramp_time">1</stringProp>
+        <boolProp name="ThreadGroup.scheduler">false</boolProp>
+        <stringProp name="ThreadGroup.duration"></stringProp>
+        <stringProp name="ThreadGroup.delay"></stringProp>
+        <boolProp name="ThreadGroup.same_user_on_next_iteration">true</boolProp>
+      </ThreadGroup>
+      <hashTree>
+        <ResultCollector guiclass="ViewResultsFullVisualizer" testclass="ResultCollector" testname="View Results Tree" enabled="true">
+          <boolProp name="ResultCollector.error_logging">false</boolProp>
+          <objProp>
+            <name>saveConfig</name>
+            <value class="SampleSaveConfiguration">
+              <time>true</time>
+              <latency>true</latency>
+              <timestamp>true</timestamp>
+              <success>true</success>
+              <label>true</label>
+              <code>true</code>
+              <message>true</message>
+              <threadName>true</threadName>
+              <dataType>true</dataType>
+              <encoding>false</encoding>
+              <assertions>true</assertions>
+              <subresults>true</subresults>
+              <responseData>false</responseData>
+              <samplerData>false</samplerData>
+              <xml>false</xml>
+              <fieldNames>true</fieldNames>
+              <responseHeaders>false</responseHeaders>
+              <requestHeaders>false</requestHeaders>
+              <responseDataOnError>false</responseDataOnError>
+              <saveAssertionResultsFailureMessage>true</saveAssertionResultsFailureMessage>
+              <assertionsResultsToSave>0</assertionsResultsToSave>
+              <bytes>true</bytes>
+              <sentBytes>true</sentBytes>
+              <threadCounts>true</threadCounts>
+              <idleTime>true</idleTime>
+              <connectTime>true</connectTime>
+            </value>
+          </objProp>
+          <stringProp name="filename"></stringProp>
+        </ResultCollector>
+        <hashTree/>
+        <TransactionController guiclass="TransactionControllerGui" testclass="TransactionController" testname="API" enabled="true">
+          <boolProp name="TransactionController.includeTimers">false</boolProp>
+          <boolProp name="TransactionController.parent">false</boolProp>
+        </TransactionController>
+        <hashTree>
+          <net.xmeter.samplers.ConnectSampler guiclass="net.xmeter.gui.ConnectSamplerUI" testclass="net.xmeter.samplers.ConnectSampler" testname="MQTT Connect" enabled="true">
+            <stringProp name="mqtt.server">${mqtt_srv}</stringProp>
+            <stringProp name="mqtt.port">1883</stringProp>
+            <stringProp name="mqtt.version">3.1</stringProp>
+            <stringProp name="mqtt.conn_timeout">10</stringProp>
+            <boolProp name="mqtt.private_protocol">false</boolProp>
+            <stringProp name="mqtt.listener_timeout">10</stringProp>
+            <stringProp name="mqtt.protocol">TCP</stringProp>
+            <boolProp name="mqtt.dual_ssl_authentication">false</boolProp>
+            <stringProp name="mqtt.keystore_file_path"></stringProp>
+            <stringProp name="mqtt.keystore_password"></stringProp>
+            <stringProp name="mqtt.clientcert_file_path"></stringProp>
+            <stringProp name="mqtt.clientcert_password"></stringProp>
+            <stringProp name="mqtt.user_name"></stringProp>
+            <stringProp name="mqtt.password"></stringProp>
+            <stringProp name="mqtt.client_id_prefix">conn_</stringProp>
+            <boolProp name="mqtt.client_id_suffix">true</boolProp>
+            <stringProp name="mqtt.conn_keep_alive">300</stringProp>
+            <stringProp name="mqtt.conn_attampt_max">0</stringProp>
+            <stringProp name="mqtt.reconn_attampt_max">0</stringProp>
+            <boolProp name="mqtt.conn_clean_session">true</boolProp>
+          </net.xmeter.samplers.ConnectSampler>
+          <hashTree/>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="AddSourcePlugin" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&#xd;
+  &quot;name&quot;:&quot;sql&quot;,&#xd;
+  &quot;file&quot;:&quot;http://127.0.0.1:9090/plugins/sqlSrc.zip&quot;&#xd;
+}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/plugins/sources</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="49587">201</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_code</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+            <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+              <stringProp name="ConstantTimer.delay">500</stringProp>
+            </ConstantTimer>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="DescSourcePlugin" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/plugins/sources/sql</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="114126">sql</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="DescSinkPlugin" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/plugins/sinks/sql</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="114126">sql</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <SystemSampler guiclass="SystemSamplerGui" testclass="SystemSampler" testname="CreateTable" enabled="true">
+            <boolProp name="SystemSampler.checkReturnCode">false</boolProp>
+            <stringProp name="SystemSampler.expectedReturnCode">0</stringProp>
+            <stringProp name="SystemSampler.command">test/plugins/sql/create_table</stringProp>
+            <elementProp name="SystemSampler.arguments" elementType="Arguments" guiclass="ArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <elementProp name="SystemSampler.environment" elementType="Arguments" guiclass="ArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="SystemSampler.directory">${__property(fvt,,)}</stringProp>
+          </SystemSampler>
+          <hashTree>
+            <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+              <stringProp name="ConstantTimer.delay">500</stringProp>
+            </ConstantTimer>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateAlertSettingStream" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&quot;sql&quot;:&quot;CREATE STREAM alertStream() WITH (DATASOURCE=\&quot;scene/alert\&quot;, FORMAT=\&quot;json\&quot;, TYPE=\&quot;mqtt\&quot;)&quot;}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+            <stringProp name="TestPlan.comments">Create stream for alert setting actions</stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="2128378328">Stream alertStream is created.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateRuleUpdateAlert" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&#xd;
+  &quot;id&quot;: &quot;ruleUpdateAlert&quot;,&#xd;
+  &quot;sql&quot;:&quot;SELECT * FROM alertStream&quot;,&#xd;
+  &quot;actions&quot;:[&#xd;
+    {&#xd;
+      &quot;sql&quot;: {&#xd;
+        &quot;url&quot;: &quot;sqlite:/tmp/test.db&quot;,&#xd;
+        &quot;keyField&quot;: &quot;id&quot;,&#xd;
+        &quot;rowkindField&quot;: &quot;action&quot;,&#xd;
+        &quot;table&quot;: &quot;alertTable&quot;,&#xd;
+        &quot;sendSingle&quot;: true&#xd;
+      }&#xd;
+    }&#xd;
+  ]&#xd;
+}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-1451809660">Rule ruleUpdateAlert was created</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateDataStream" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&quot;sql&quot;:&quot;CREATE STREAM demoStream() WITH (DATASOURCE=\&quot;scene/data\&quot;, FORMAT=\&quot;json\&quot;, TYPE=\&quot;mqtt\&quot;)&quot;}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+            <stringProp name="TestPlan.comments">Create stream for alert setting actions</stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-894755937">Stream demoStream is created.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateSqlTable" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&quot;sql&quot;:&quot;CREATE TABLE alertTable() WITH (DATASOURCE=\&quot;alertTable\&quot;, CONF_KEY=\&quot;sqlite_config\&quot;, TYPE=\&quot;sql\&quot;, KIND=\&quot;lookup\&quot;)&quot;}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/tables</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-1341540896">Table alertTable is created.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="CreateRuleAlert" enabled="true">
+            <boolProp name="HTTPSampler.postBodyRaw">true</boolProp>
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments">
+              <collectionProp name="Arguments.arguments">
+                <elementProp name="" elementType="HTTPArgument">
+                  <boolProp name="HTTPArgument.always_encode">false</boolProp>
+                  <stringProp name="Argument.value">{&#xd;
+  &quot;id&quot;: &quot;ruleAlert&quot;,&#xd;
+  &quot;sql&quot;:&quot;SELECT demoStream.device, demoStream.value FROM demoStream INNER JOIN alertTable ON demoStream.deviceKind = alertTable.id WHERE demoStream.value &gt; alertTable.alarm&quot;,&#xd;
+  &quot;actions&quot;:[&#xd;
+    {&#xd;
+      &quot;mqtt&quot;: {&#xd;
+        &quot;server&quot;: &quot;tcp://${mqtt_srv}:1883&quot;,&#xd;
+        &quot;topic&quot;: &quot;rule/alert&quot;,&#xd;
+        &quot;sendSingle&quot;: true&#xd;
+      }&#xd;
+    }&#xd;
+  ]&#xd;
+}</stringProp>
+                  <stringProp name="Argument.metadata">=</stringProp>
+                </elementProp>
+              </collectionProp>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules</stringProp>
+            <stringProp name="HTTPSampler.method">POST</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-2079532467">Rule ruleAlert was created</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_GetRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_demoStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">0</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="FeedTable" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">alert_init_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">action,id,alarm</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/alert</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;id&quot;: ${id}, &quot;action&quot;: ${action}, &quot;alarm&quot; : ${alarm}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="GetTableRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleUpdateAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_alertStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">4</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_memory_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">4</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+              <stringProp name="ConstantTimer.delay">5000</stringProp>
+            </ConstantTimer>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="SendData" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">collect_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">device,deviceKind,value</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/data</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;deviceKind&quot;: ${deviceKind}, &quot;device&quot;: ${device}, &quot;value&quot; : ${value}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_GetRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_demoStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">4</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_mqtt_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">2</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="UpdateTable" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">alert_update_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">action,id,alarm</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/alert</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;id&quot;: ${id}, &quot;action&quot;: ${action}, &quot;alarm&quot; : ${alarm}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="GetTableRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleUpdateAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_alertStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">8</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_memory_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">8</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+              <stringProp name="ConstantTimer.delay">5000</stringProp>
+            </ConstantTimer>
+            <hashTree/>
+          </hashTree>
+          <LoopController guiclass="LoopControlPanel" testclass="LoopController" testname="SendData" enabled="true">
+            <boolProp name="LoopController.continue_forever">true</boolProp>
+            <stringProp name="LoopController.loops">4</stringProp>
+          </LoopController>
+          <hashTree>
+            <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+              <stringProp name="delimiter">,</stringProp>
+              <stringProp name="fileEncoding"></stringProp>
+              <stringProp name="filename">collect_data.txt</stringProp>
+              <boolProp name="ignoreFirstLine">false</boolProp>
+              <boolProp name="quotedData">false</boolProp>
+              <boolProp name="recycle">true</boolProp>
+              <stringProp name="shareMode">shareMode.thread</stringProp>
+              <boolProp name="stopThread">false</boolProp>
+              <stringProp name="variableNames">device,deviceKind,value</stringProp>
+            </CSVDataSet>
+            <hashTree/>
+            <net.xmeter.samplers.PubSampler guiclass="net.xmeter.gui.PubSamplerUI" testclass="net.xmeter.samplers.PubSampler" testname="MQTT Pub Sampler" enabled="true">
+              <stringProp name="mqtt.topic_name">scene/data</stringProp>
+              <stringProp name="mqtt.qos_level">0</stringProp>
+              <boolProp name="mqtt.add_timestamp">false</boolProp>
+              <stringProp name="mqtt.message_type">String</stringProp>
+              <stringProp name="mqtt.message_type_fixed_length">1024</stringProp>
+              <stringProp name="mqtt.message_to_sent">{&quot;deviceKind&quot;: ${deviceKind}, &quot;device&quot;: ${device}, &quot;value&quot; : ${value}}</stringProp>
+              <boolProp name="mqtt.retained_message">false</boolProp>
+            </net.xmeter.samplers.PubSampler>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_GetRuleStatus" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert/status</stringProp>
+            <stringProp name="HTTPSampler.method">GET</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
+              <stringProp name="JSON_PATH">$.source_demoStream_0_records_in_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">8</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+            <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="false">
+              <stringProp name="JSON_PATH">$.sink_mqtt_0_0_records_out_total</stringProp>
+              <stringProp name="EXPECTED_VALUE">3</stringProp>
+              <boolProp name="JSONVALIDATION">true</boolProp>
+              <boolProp name="EXPECT_NULL">false</boolProp>
+              <boolProp name="INVERT">false</boolProp>
+              <boolProp name="ISREGEX">false</boolProp>
+            </JSONPathAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_DropRule" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleAlert</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="659914816">Rule ruleAlert is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_Drop_Stream" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams/demoStream</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="1148079559">Stream demoStream is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_Drop_Table" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/tables/alertTable</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="701294600">Table alertTable is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_DropRule" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/rules/ruleUpdateAlert</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="1287637623">Rule ruleUpdateAlert is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_Drop_Stream" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/streams/alertStream</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-123753472">Stream alertStream is dropped.</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">false</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="DropSourcePlugin" enabled="true">
+            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
+              <collectionProp name="Arguments.arguments"/>
+            </elementProp>
+            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
+            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
+            <stringProp name="HTTPSampler.protocol"></stringProp>
+            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
+            <stringProp name="HTTPSampler.path">/plugins/sources/sql</stringProp>
+            <stringProp name="HTTPSampler.method">DELETE</stringProp>
+            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
+            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
+            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
+            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
+            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
+            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
+            <stringProp name="HTTPSampler.response_timeout"></stringProp>
+          </HTTPSamplerProxy>
+          <hashTree>
+            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
+              <collectionProp name="Asserion.test_strings">
+                <stringProp name="-57041190">sources plugin sql is deleted</stringProp>
+              </collectionProp>
+              <stringProp name="Assertion.custom_message"></stringProp>
+              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
+              <boolProp name="Assertion.assume_success">true</boolProp>
+              <intProp name="Assertion.test_type">16</intProp>
+            </ResponseAssertion>
+            <hashTree/>
+          </hashTree>
+          <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
+            <stringProp name="ConstantTimer.delay">500</stringProp>
+          </ConstantTimer>
+          <hashTree/>
+        </hashTree>
+      </hashTree>
+      <ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="Result" enabled="true">
+        <stringProp name="ThreadGroup.on_sample_error">continue</stringProp>
+        <elementProp name="ThreadGroup.main_controller" elementType="LoopController" guiclass="LoopControlPanel" testclass="LoopController" testname="Loop Controller" enabled="true">
+          <boolProp name="LoopController.continue_forever">false</boolProp>
+          <stringProp name="LoopController.loops">2</stringProp>
+        </elementProp>
+        <stringProp name="ThreadGroup.num_threads">1</stringProp>
+        <stringProp name="ThreadGroup.ramp_time">1</stringProp>
+        <boolProp name="ThreadGroup.scheduler">false</boolProp>
+        <stringProp name="ThreadGroup.duration"></stringProp>
+        <stringProp name="ThreadGroup.delay"></stringProp>
+        <boolProp name="ThreadGroup.same_user_on_next_iteration">true</boolProp>
+      </ThreadGroup>
+      <hashTree>
+        <ResultCollector guiclass="ViewResultsFullVisualizer" testclass="ResultCollector" testname="View Results Tree" enabled="true">
+          <boolProp name="ResultCollector.error_logging">false</boolProp>
+          <objProp>
+            <name>saveConfig</name>
+            <value class="SampleSaveConfiguration">
+              <time>true</time>
+              <latency>true</latency>
+              <timestamp>true</timestamp>
+              <success>true</success>
+              <label>true</label>
+              <code>true</code>
+              <message>true</message>
+              <threadName>true</threadName>
+              <dataType>true</dataType>
+              <encoding>false</encoding>
+              <assertions>true</assertions>
+              <subresults>true</subresults>
+              <responseData>false</responseData>
+              <samplerData>false</samplerData>
+              <xml>false</xml>
+              <fieldNames>true</fieldNames>
+              <responseHeaders>false</responseHeaders>
+              <requestHeaders>false</requestHeaders>
+              <responseDataOnError>false</responseDataOnError>
+              <saveAssertionResultsFailureMessage>true</saveAssertionResultsFailureMessage>
+              <assertionsResultsToSave>0</assertionsResultsToSave>
+              <bytes>true</bytes>
+              <sentBytes>true</sentBytes>
+              <threadCounts>true</threadCounts>
+              <idleTime>true</idleTime>
+              <connectTime>true</connectTime>
+            </value>
+          </objProp>
+          <stringProp name="filename"></stringProp>
+        </ResultCollector>
+        <hashTree/>
+        <OnceOnlyController guiclass="OnceOnlyControllerGui" testclass="OnceOnlyController" testname="Once Only Controller" enabled="true"/>
+        <hashTree>
+          <net.xmeter.samplers.ConnectSampler guiclass="net.xmeter.gui.ConnectSamplerUI" testclass="net.xmeter.samplers.ConnectSampler" testname="MQTT Connect" enabled="true">
+            <stringProp name="mqtt.server">${mqtt_srv}</stringProp>
+            <stringProp name="mqtt.port">1883</stringProp>
+            <stringProp name="mqtt.version">3.1</stringProp>
+            <stringProp name="mqtt.conn_timeout">10</stringProp>
+            <boolProp name="mqtt.private_protocol">false</boolProp>
+            <stringProp name="mqtt.listener_timeout">10</stringProp>
+            <stringProp name="mqtt.protocol">TCP</stringProp>
+            <boolProp name="mqtt.dual_ssl_authentication">false</boolProp>
+            <stringProp name="mqtt.keystore_file_path"></stringProp>
+            <stringProp name="mqtt.keystore_password"></stringProp>
+            <stringProp name="mqtt.clientcert_file_path"></stringProp>
+            <stringProp name="mqtt.clientcert_password"></stringProp>
+            <stringProp name="mqtt.user_name"></stringProp>
+            <stringProp name="mqtt.password"></stringProp>
+            <stringProp name="mqtt.client_id_prefix">conn_</stringProp>
+            <boolProp name="mqtt.client_id_suffix">true</boolProp>
+            <stringProp name="mqtt.conn_keep_alive">300</stringProp>
+            <stringProp name="mqtt.conn_attampt_max">0</stringProp>
+            <stringProp name="mqtt.reconn_attampt_max">0</stringProp>
+            <boolProp name="mqtt.conn_clean_session">true</boolProp>
+          </net.xmeter.samplers.ConnectSampler>
+          <hashTree/>
+        </hashTree>
+        <net.xmeter.samplers.SubSampler guiclass="net.xmeter.gui.SubSamplerUI" testclass="net.xmeter.samplers.SubSampler" testname="AnalysisResult" enabled="true">
+          <stringProp name="mqtt.topic_name">rule/alert</stringProp>
+          <stringProp name="mqtt.qos_level">0</stringProp>
+          <boolProp name="mqtt.add_timestamp">false</boolProp>
+          <boolProp name="mqtt.debug_response">true</boolProp>
+          <stringProp name="mqtt.sample_condition">number of received messages</stringProp>
+          <stringProp name="mqtt.sample_condition_value">1</stringProp>
+        </net.xmeter.samplers.SubSampler>
+        <hashTree>
+          <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="value Assertion" enabled="true">
+            <stringProp name="JSON_PATH">$.value</stringProp>
+            <stringProp name="EXPECTED_VALUE">${value}</stringProp>
+            <boolProp name="JSONVALIDATION">true</boolProp>
+            <boolProp name="EXPECT_NULL">false</boolProp>
+            <boolProp name="INVERT">false</boolProp>
+            <boolProp name="ISREGEX">false</boolProp>
+          </JSONPathAssertion>
+          <hashTree/>
+          <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="device Assertion" enabled="true">
+            <stringProp name="JSON_PATH">$.device</stringProp>
+            <stringProp name="EXPECTED_VALUE">${device}</stringProp>
+            <boolProp name="JSONVALIDATION">true</boolProp>
+            <boolProp name="EXPECT_NULL">false</boolProp>
+            <boolProp name="INVERT">false</boolProp>
+            <boolProp name="ISREGEX">false</boolProp>
+          </JSONPathAssertion>
+          <hashTree/>
+        </hashTree>
+        <CSVDataSet guiclass="TestBeanGUI" testclass="CSVDataSet" testname="CSV Data Set Config" enabled="true">
+          <stringProp name="delimiter">,</stringProp>
+          <stringProp name="fileEncoding"></stringProp>
+          <stringProp name="filename">lookup_table_result_data.txt</stringProp>
+          <boolProp name="ignoreFirstLine">false</boolProp>
+          <boolProp name="quotedData">false</boolProp>
+          <boolProp name="recycle">true</boolProp>
+          <stringProp name="shareMode">shareMode.thread</stringProp>
+          <boolProp name="stopThread">false</boolProp>
+          <stringProp name="variableNames">device,value</stringProp>
+        </CSVDataSet>
+        <hashTree/>
+      </hashTree>
+    </hashTree>
+  </hashTree>
+</jmeterTestPlan>
diff --git a/test/plugin_end_2_end.jmx b/test/plugin_end_2_end.jmx
index 20be10e3..eb67875b 100644
--- a/test/plugin_end_2_end.jmx
+++ b/test/plugin_end_2_end.jmx
@@ -1,4 +1,20 @@
 <?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
 <jmeterTestPlan version="1.2" properties="5.0" jmeter="5.3">
   <hashTree>
     <TestPlan guiclass="TestPlanGui" testclass="TestPlan" testname="Test Plan" enabled="true">
@@ -962,36 +978,6 @@
             </ResponseAssertion>
             <hashTree/>
           </hashTree>
-          <HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API_DropPlugin" enabled="true">
-            <elementProp name="HTTPsampler.Arguments" elementType="Arguments" guiclass="HTTPArgumentsPanel" testclass="Arguments" testname="User Defined Variables" enabled="true">
-              <collectionProp name="Arguments.arguments"/>
-            </elementProp>
-            <stringProp name="HTTPSampler.domain">${srv}</stringProp>
-            <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
-            <stringProp name="HTTPSampler.protocol"></stringProp>
-            <stringProp name="HTTPSampler.contentEncoding"></stringProp>
-            <stringProp name="HTTPSampler.path">/plugins/sinks/sql</stringProp>
-            <stringProp name="HTTPSampler.method">DELETE</stringProp>
-            <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
-            <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
-            <boolProp name="HTTPSampler.use_keepalive">true</boolProp>
-            <boolProp name="HTTPSampler.DO_MULTIPART_POST">false</boolProp>
-            <stringProp name="HTTPSampler.embedded_url_re"></stringProp>
-            <stringProp name="HTTPSampler.connect_timeout"></stringProp>
-            <stringProp name="HTTPSampler.response_timeout"></stringProp>
-          </HTTPSamplerProxy>
-          <hashTree>
-            <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
-              <collectionProp name="Asserion.test_strings">
-                <stringProp name="367681890">sinks plugin sql is deleted</stringProp>
-              </collectionProp>
-              <stringProp name="Assertion.custom_message"></stringProp>
-              <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
-              <boolProp name="Assertion.assume_success">true</boolProp>
-              <intProp name="Assertion.test_type">16</intProp>
-            </ResponseAssertion>
-            <hashTree/>
-          </hashTree>
           <ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="Constant Timer" enabled="true">
             <stringProp name="ConstantTimer.delay">500</stringProp>
           </ConstantTimer>
diff --git a/test/plugins/sql/create_table.go b/test/plugins/sql/create_table.go
new file mode 100644
index 00000000..902b9ab2
--- /dev/null
+++ b/test/plugins/sql/create_table.go
@@ -0,0 +1,32 @@
+// Copyright 2022 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"database/sql"
+	_ "github.com/mattn/go-sqlite3"
+)
+
+func main() {
+	db, err := sql.Open("sqlite3", "file:/tmp/test.db")
+	if err != nil {
+		panic(err)
+	}
+	defer db.Close()
+	_, err = db.Exec("CREATE TABLE IF NOT EXISTS alertTable (id BIGINT PRIMARY KEY, action TEXT NOT NULL, alarm BIGINT)")
+	if err != nil {
+		panic(err)
+	}
+}
diff --git a/test/prepare_plugins.sh b/test/prepare_plugins.sh
index c8b1aba2..34ec7d77 100755
--- a/test/prepare_plugins.sh
+++ b/test/prepare_plugins.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 #
-# Copyright 2021 EMQ Technologies Co., Ltd.
+# Copyright 2021-2022 EMQ Technologies Co., Ltd.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -23,6 +23,9 @@ chmod +x test/plugins/pub/zmq_pub
 go build -o test/plugins/service/http_server test/plugins/service/server.go
 chmod +x test/plugins/service/http_server
 
+go build -o test/plugins/sql/create_table test/plugins/sql/create_table.go
+chmod +x test/plugins/sql/create_table
+
 cd test
 
 rm -rf zmq.* Zmq.so
@@ -36,7 +39,7 @@ else
 fi
 
 mv ../plugins/sources/Zmq.so .
-cp plugins/zmq.yaml .
+cp ../extensions/sources/zmq/zmq.yaml .
 zip zmq.zip Zmq.so zmq.yaml
 rm -rf zmq.yaml Zmq.so
 
@@ -70,7 +73,7 @@ rm -rf Tdengine.so
 # build sql plugins
 FILE=../plugins/sinks/Sql.so
 if [ -f "$FILE" ]; then
-    echo "$FILE exists, not requried to build plugin."
+    echo "$FILE exists, not required to build plugin."
 else
     echo "$FILE does not exist, will build the plugin."
     go build -trimpath -modfile ../extensions.mod --buildmode=plugin -o ../plugins/sinks/Sql.so ../extensions/sinks/sql/*.go
@@ -80,12 +83,26 @@ mv ../plugins/sinks/Sql.so .
 zip sql.zip Sql.so
 rm -rf Sql.so
 
+FILE=../plugins/sources/Sql.so
+if [ -f "$FILE" ]; then
+    echo "$FILE exists, not required to build plugin."
+else
+    echo "$FILE does not exist, will build the plugin."
+    go build -trimpath -modfile ../extensions.mod --buildmode=plugin -o ../plugins/sources/Sql.so ../extensions/sources/sql/*.go
+fi
+
+mv ../plugins/sources/Sql.so .
+cp ../extensions/sources/sql/sql.yaml .
+zip sqlSrc.zip Sql.so sql.yaml
+rm -rf Sql.so
+
 rm -rf plugins/service/web/plugins/
 mkdir -p plugins/service/web/plugins/
 mv zmq.zip plugins/service/web/plugins/
 mv image.zip plugins/service/web/plugins/
 mv tdengine.zip plugins/service/web/plugins/
 mv sql.zip plugins/service/web/plugins/
+mv sqlSrc.zip plugins/service/web/plugins/
 
 # prepare portable plugins
 cd ..
diff --git a/test/run_jmeter.sh b/test/run_jmeter.sh
index 863993bb..30e242c2 100755
--- a/test/run_jmeter.sh
+++ b/test/run_jmeter.sh
@@ -154,4 +154,10 @@ echo -e "---------------------------------------------\n"
 echo -e "---------------------------------------------\n"
 
 /opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/lookup_table_memory.jmx -Dfvt="$fvt_dir" -l jmeter_logs/lookup_table_memory.jtl -j jmeter_logs/lookup_table_memory.log
+echo -e "---------------------------------------------\n"
+
+/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/lookup_table_redis.jmx -Dfvt="$fvt_dir" -l jmeter_logs/lookup_table_redis.jtl -j jmeter_logs/lookup_table_redis.log
+echo -e "---------------------------------------------\n"
+
+/opt/jmeter/bin/jmeter.sh -Jjmeter.save.saveservice.output_format=xml -n -t test/lookup_table_sql.jmx -Dfvt="$fvt_dir" -l jmeter_logs/lookup_table_sql.jtl -j jmeter_logs/lookup_table_sql.log
 echo -e "---------------------------------------------\n"
\ No newline at end of file