Tdengine sTable support & dynamic table (#1239)

* feat(plugin): tdengine sTable support

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

* doc(plugin): tdengine doc

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

docs/en_US/rules/sinks/plugin/tdengine.md

@@ -1,41 +1,31 @@
 ## Compile the plugins
 
-### plugins/go.mod
-
-```go
-module plugins
-
-go 1.13
-
-replace github.com/lf-edge/ekuiper => /$eKuiper
-
-require (
-    github.com/lf-edge/ekuiper v0.0.0-00010101000000-000000000000 // indirect
-    github.com/taosdata/driver-go v0.0.0-20200723061832-5be6460b0c20
-)
-```
+In eKuiper source code root path, run the below command.
 
 ```shell
-go mod edit -replace github.com/lf-edge/ekuiper=/$eKuiper
-go build -trimpath -modfile extensions.mod --buildmode=plugin -o /$ekuiper/plugins/sinks/Tdengine@v1.0.0.so /$ekuiper/extensions/sinks/tdengine/tdengine.go
+go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sinks/Tdengine@v1.0.0.so extensions/sinks/tdengine/tdengine.go
 ```
 ### Install plugin
+
 Since the operation of the tdengine plug-in depends on the tdengine client, for the convenience of users, the tdengine client will be downloaded when the plug-in is installed. However, the tdengine client version corresponds to the server version one-to-one, which is not compatible with each other, so the user must inform the tdengine server version used.
-## Rule Actions Description
 
-As the tdengine database requires a time stamp field in the table, the user must inform the time stamp field name of the data table (required tsFieldName). The user can choose whether to provide time stamp data. If not (provideTs=false), the content of the time stamp field is automatically generated by the tdengine database.
+## Rule Actions Description
 
-| Name        | Type     | Optional                                          | Description                                 |
-| ----------- | -------- | ------------------------------------------------- | ------------------------------------------- |
-| ip          | string   | false                                             | Database ip                                 |
-| port        | int      | false                                             | Database port                               |
-| user        | string   | false                                             | Username                                    |
-| password    | string   | false                                             | Password                                    |
-| database    | string   | false                                             | Database name                               |
-| table       | string   | false                                             | Table Name                                  |
-| fields      | []string | true（Replace with data key when not filling in） | Table field collection                      |
-| provideTs   | Bool     | false                                             | Whether the user provides a timestamp field |
-| tsFieldName | String   | false                                             | Timestamp field name                        |
+As the tdengine database requires a timestamp field in the table, the user must inform the timestamp field name of the data table (required tsFieldName). The user can choose whether to provide timestamp data. If not (provideTs=false), the content of the timestamp field is automatically generated by the tdengine database.
+
+| Name        | Type     | Optional | Description                                                                                                                                                      |
+|-------------|----------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| host        | string   | true     | Database host, it must be a host name aka. [FQDN](https://www.taosdata.com/blog/2020/09/11/1824.html). An IP address is invalid. The default value is localhost. |
+| port        | int      | false    | Database port                                                                                                                                                    |
+| user        | string   | true     | Username, default to `root`.                                                                                                                                     |
+| password    | string   | true     | Password, default to `taosdata`.                                                                                                                                 |
+| database    | string   | false    | Database name.                                                                                                                                                   |
+| table       | string   | false    | Table Name, could be a [dynamic property](../../overview.md#dynamic-properties).                                                                                 |
+| fields      | []string | true     | The fields to be inserted to. The result map and the database should both have these fields. If not specified, all fields in the result map will be inserted.    |
+| provideTs   | Bool     | true     | Whether the user provides a timestamp field, default to false.                                                                                                   |
+| tsFieldName | String   | true     | Timestamp field name                                                                                                                                             |
+| sTable      | String   | true     | The super table to be use, could be a [dynamic property](../../overview.md#dynamic-properties).                                                                  |
+| tagFields   | []String | true     | The result fields to be used as the tag values in order. If sTable is specified, this is required.                                                               |
 
 ## Operation example
 
@@ -57,9 +47,36 @@ curl --location --request POST 'http://127.0.0.1:9081/streams' --header 'Content
 curl --location --request POST 'http://127.0.0.1:9081/rules' --header 'Content-Type:application/json' --data '{"id":"demoRule","sql":"SELECT * FROM demoStream;","actions":[{"tdengine":{"provideTs":true,"tsFieldName":"time","port":0,"ip":"127.0.0.1","user":"root","password":"taosdata","database":"dbName","table":"tableName","fields":["time","age"]}}]}'
 ```
 
-### Send data
+Write into fixed table:
+
+```json
+{
+  "tdengine": {
+    "host":        "hostname",
+    "port":        6030,
+    "user":        "root",
+    "password":    "taosdata",
+    "database":    "db",
+    "table":       "tableName",
+    "tsfieldname": "ts"
+  }
+}
+```
 
-```bash
-mosquitto_pub -h broker.emqx.io -m '{"time":"2020-01-11 18:18:18", "age" : 18}' -t device/device_001/message
+Write into dynamic table:
+
+```json lines
+{
+  "tdengine": {
+    "host":        "hostname",
+    "port":        6030,
+    "database":    "dab",
+    "table":       "{{.table}}", // dynamic value, get from the table field of the result
+    "tsfieldname": "ts",
+    "fields":      []string{"f1", "f2"}, // Write f1, f2 fields in result into f1, f2 columns in the db
+    "sTable":      "myStable", // super table name, also allow dynamic
+    "tagFields":   []string{"f3","f4"} // Write f3, f4 fields' values in the result as tags in order
+  }
+}
 ```
 

docs/zh_CN/rules/sinks/plugin/tdengine.md

@@ -1,41 +1,31 @@
 ## 编译插件
 
-### plugins/go.mod
-
-```go
-module plugins
-
-go 1.13
-
-replace github.com/lf-edge/ekuiper => /$eKuiper
-
-require (
-    github.com/lf-edge/ekuiper v0.0.0-00010101000000-000000000000 // indirect
-    github.com/taosdata/driver-go v0.0.0-20200723061832-5be6460b0c20
-)
-```
+在 eKuiper 项目主目录运行如下命令：
 
 ```shell
-go mod edit -replace github.com/lf-edge/ekuiper=/$eKuiper
-go build -trimpath -modfile extensions.mod --buildmode=plugin -o /$ekuiper/plugins/sinks/Tdengine@v1.0.0.so /$ekuiper/extensions/sinks/tdengine/tdengine.go
+go build -trimpath -modfile extensions.mod --buildmode=plugin -o plugins/sinks/Tdengine@v1.0.0.so extensions/sinks/tdengine/tdengine.go
 ```
 ### 安装插件
+
 由于 tdengine 插件的运行依赖于 tdengine 客户端，为了便于用户使用，安装插件时将下载 tdengine 客户端。但是 tdengine 客户端版本与其服务器版本一一对应，互不兼容，所以用户必须告知所用 tdengine 服务器版本。
+
 ## 规则 Actions 说明
 
 由于 tdengine 数据库要求表中必须有时间戳字段，所以用户必须告知数据表的时间戳字段名称（必填tsFieldName）。用户可以选择是否提供时间戳数据，若不提供（provideTs=false），时间戳字段的内容由 tdengine 数据库自动生成。
 
-| 名称        | 类型     | 是否必填                      | 释义                   |
-| ----------- | -------- | ----------------------------- | ---------------------- |
-| ip          | string   | 必填                          | 数据库ip               |
-| port        | int      | 必填                          | 数据库端口             |
-| user        | string   | 必填                          | 用户名                 |
-| password    | string   | 必填                          | 密码                   |
-| database    | string   | 必填                          | 数据库名               |
-| table       | string   | 必填                          | 表名                   |
-| fields      | []string | 选填（不填时用数据的key替代） | 表字段集合             |
-| provideTs   | Bool     | 必填                          | 用户是否提供时间戳字段 |
-| tsFieldName | String   | 必填                          | 时间戳字段名称         |
+| 名称          | 类型       | 是否必填 | 释义                                                                                                    |
+|-------------|----------|------|-------------------------------------------------------------------------------------------------------|
+| host        | string   | 否    | 数据库域名，其值必须为域名，即 [FQDN](https://www.taosdata.com/blog/2020/09/11/1824.html)，不能为 IP 地址。其默认值为 localhost。 |
+| port        | int      | 是    | 数据库端口                                                                                                 |
+| user        | string   | 否    | 用户名，默认值为 `root` 。                                                                                     |
+| password    | string   | 否    | 密码，默认值为 `taosdata` 。                                                                                  |
+| database    | string   | 是    | 数据库名                                                                                                  |
+| table       | string   | 是    | 表名，可设置[动态属性](../../overview.md#动态属性)。                                                                 |
+| fields      | []string | 否    | 将要插入的表字段集合。sink 收到的数据和数据库表中均有该字段。若为设置，则所有结果字段写入数据库。                                                   |
+| provideTs   | Bool     | 否    | 用户是否提供时间戳字段，默认为否。                                                                                     |
+| tsFieldName | String   | 是    | 时间戳字段名称                                                                                               |
+| sTable      | String   | 否    | 使用的超级表，可设置[动态属性](../../overview.md#动态属性)。                                                             |
+| tagFields   | []String | 否    | 结果中作为标签的字段。若设置 sTable 属性，则该属性必填。                                                                      |
 
 ## 操作示例
 
@@ -57,9 +47,37 @@ curl --location --request POST 'http://127.0.0.1:9081/streams' --header 'Content
 curl --location --request POST 'http://127.0.0.1:9081/rules' --header 'Content-Type:application/json' --data '{"id":"demoRule","sql":"SELECT * FROM demoStream;","actions":[{"tdengine":{"provideTs":true,"tsFieldName":"time","port":0,"ip":"127.0.0.1","user":"root","password":"taosdata","database":"dbName","table":"tableName","fields":["time","age"]}}]}'
 ```
 
-### 发送数据
+写入固定表格的例子：
+
+```json
+{
+  "tdengine": {
+    "host":        "hostname",
+    "port":        6030,
+    "user":        "root",
+    "password":    "taosdata",
+    "database":    "db",
+    "table":       "tableName",
+    "tsfieldname": "ts"
+  }
+}
+```
 
-```bash
-mosquitto_pub -h broker.emqx.io -m '{"time":"2020-01-11 18:18:18", "age" : 18}' -t device/device_001/message
+写入动态表的例子：
+
+```json lines
+{
+  "tdengine": {
+    "host":        "hostname",
+    "port":        6030,
+    "database":    "dab",
+    "table":       "{{.table}}", // 动态值，从结果中的 table 字段获取
+    "tsfieldname": "ts",
+    "fields":      []string{"f1", "f2"}, // 结果中的 f1, f2 字段写入数据库中的 f1, f2 列
+    "sTable":      "myStable", // 超级表名，也可以动态
+    "tagFields":   []string{"f3","f4"} // 结果中的 f3, f4 字段的值按顺序作为标签值写入
+  }
+}
 ```
 
+

extensions/go.mod

@@ -13,11 +13,6 @@ require (
 	github.com/taosdata/driver-go/v2 v2.0.0
 )
 
-require (
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/mattn/go-pointer v0.0.0-20190911064623-a0a44394634f // indirect
-	github.com/mitchellh/mapstructure v1.4.1 // indirect
-	gopkg.in/yaml.v2 v2.4.0 // indirect
-)
+require gopkg.in/yaml.v2 v2.4.0 // indirect
 
 replace github.com/lf-edge/ekuiper => ../

extensions/go.sum

@@ -2,15 +2,20 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
+github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI=
 github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
 github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
+github.com/Masterminds/semver/v3 v3.1.1 h1:hLg3sBzpNErnxhQtUy/mmLR2I9foDujNK030IGemrRc=
 github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=
 github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=
+github.com/Masterminds/sprig/v3 v3.2.1 h1:n6EPaDyLSvCEa3frruQvAiHuNp2dhBlMSmkEr+HuzGc=
 github.com/Masterminds/sprig/v3 v3.2.1/go.mod h1:UoaO7Yp8KlPnJIYWTFkMaqPUYKTfGFPhxNuwnnxkKlk=
 github.com/Masterminds/sprig/v3 v3.2.1/go.mod h1:UoaO7Yp8KlPnJIYWTFkMaqPUYKTfGFPhxNuwnnxkKlk=
 github.com/Microsoft/go-winio v0.4.11/go.mod h1:VhR8bwka0BXejwEJY73c50VrPtXAaKcyvVC4A4RozmA=
+github.com/PaesslerAG/gval v1.0.0 h1:GEKnRwkWDdf9dOmKcNrar9EA1bz1z9DqPIO1+iLzhd8=
 github.com/PaesslerAG/gval v1.0.0/go.mod h1:y/nm5yEyTeX6av0OfKJNp9rBNj2XrGhAf5+v24IBN1I=
 github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
+github.com/PaesslerAG/jsonpath v0.1.1 h1:c1/AToHQMVsduPAa4Vh6xp2U0evy4t8SWp8imEsylIk=
 github.com/PaesslerAG/jsonpath v0.1.1/go.mod h1:lVboNxFGal/VwW6d9JzIy56bUsYAP6tH/x80vjnCseY=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
@@ -21,7 +26,9 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
+github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a h1:HbKu58rmZpUGpz5+4FfNmIU+FmZg2P3Xaj2v2bfNWmk=
 github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
+github.com/alicebob/miniredis/v2 v2.15.1 h1:Fw+ixAJPmKhCLBqDwHlTDqxUxp0xjEwXczEpt1B6r7k=
 github.com/alicebob/miniredis/v2 v2.15.1/go.mod h1:gquAfGbzn92jvtrSC69+6zZnwSODVXVpYDRaGhWaL6I=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
@@ -32,6 +39,7 @@ github.com/aws/aws-sdk-go v1.38.68/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2z
 github.com/aws/aws-sdk-go-v2 v1.7.0/go.mod h1:tb9wi5s61kTDA5qCkcDbt3KRVV74GGslQkl/DRdX/P4=
 github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.5.0/go.mod h1:acH3+MQoiMzozT/ivU+DbRg7Ooo2298RdRaWcOv+4vM=
 github.com/aws/smithy-go v1.5.0/go.mod h1:SObp3lf9smib00L/v3U2eAKG8FyQ7iLrJnQiAmR5n+E=
+github.com/benbjohnson/clock v1.0.0 h1:78Jk/r6m4wCi6sndMpty7A//t4dw/RW5fV4ZgDVfX1w=
 github.com/benbjohnson/clock v1.0.0/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
@@ -75,6 +83,7 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7
 github.com/faiface/glhf v0.0.0-20181018222622-82a6317ac380/go.mod h1:zqnPFFIuYFFxl7uH2gYByJwIVKG7fRqlqQCbzAnHs9g=
 github.com/faiface/mainthread v0.0.0-20171120011319-8b78f0a41ae3/go.mod h1:VEPNJUlxl5KdWjDvz6Q1l+rJlxF2i6xqDeGuGAxa87M=
 github.com/faiface/pixel v0.8.0/go.mod h1:CEUU/s9E82Kqp01Boj1O67KnBskqiLghANqvUJGgDAM=
+github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 h1:Ghm4eQYC0nEPnSJdVkTrXpu9KtoVCSo1hg7mtI7G9KU=
 github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239/go.mod h1:Gdwt2ce0yfBxPvZrHkprdPPTTS3N5rwmLE8T22KBXlw=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
@@ -127,6 +136,7 @@ github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/gomodule/redigo v2.0.0+incompatible h1:K/R+8tc58AaqLkqG2Ol3Qk+DR/TlNuhuh457pBFPtt0=
 github.com/gomodule/redigo v2.0.0+incompatible/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -141,6 +151,7 @@ github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gordonklaus/ineffassign v0.0.0-20200309095847-7953dde2c7bf/go.mod h1:cuNKsD1zp2v6XfE/orVX2QE1LC+i254ceGcVeDT3pTU=
@@ -174,16 +185,20 @@ github.com/hashicorp/serf v0.9.5/go.mod h1:UWDWwZeL5cuWDJdl0C6wrvrUwEqtQ4ZKBKKEN
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/huandu/xstrings v1.3.1/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
+github.com/huandu/xstrings v1.3.2 h1:L18LIDzqlW6xN2rEkpdV8+oL/IXWJ1APd+vsdYy4Wdw=
 github.com/huandu/xstrings v1.3.2/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
 github.com/hudl/fargo v1.3.0/go.mod h1:y3CKSmjA+wD2gak7sUSXTAoopbhU08POFhmITJgmKTg=
 github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
+github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/influxdata/influxdb1-client v0.0.0-20200827194710-b269163b24ab h1:HqW4xhhynfjrtEiiSGcQUd6vrK23iMam1FO8rI7mwig=
 github.com/influxdata/influxdb1-client v0.0.0-20200827194710-b269163b24ab/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
+github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869 h1:IPJ3dvxmJ4uczJe5YQdrYB16oTJlGSC/OyZDqUk9xX4=
 github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869/go.mod h1:cJ6Cj7dQo+O6GJNiMx+Pa94qKj+TG8ONdKHgMNIyyag=
 github.com/jhump/protoreflect v1.8.2/go.mod h1:7GcYQDdMU/O/BBrl/cX6PNHpXh6cenjd8pneu5yW7Tg=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
+github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
 github.com/jonboulle/clockwork v0.2.2/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
@@ -192,6 +207,7 @@ github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
+github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1 h1:JL2rWnBX8jnbHHlLcLde3BBWs+jzqZvOmF+M3sXoNOE=
 github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1/go.mod h1:nNLjpEi4xVFB7358xLPpPscdvXP+pbhiHgSmjIur8z0=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
@@ -202,13 +218,18 @@ github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFB
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
 github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
+github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc h1:RKf14vYWi2ttpEmkA4aQ3j4u9dStX2t4M8UM6qqNsG8=
 github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc/go.mod h1:kopuH9ugFRkIXf3YoqHKyrJ9YfUFsckUU9S7B+XP+is=
+github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible h1:Y6sqxHMyB1D2YSzWkLibYKgg+SwmyFU9dF2hn6MdTj4=
 github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible/go.mod h1:ZQnN8lSECaebrkQytbHj4xNgtg8CR7RYXnPok8e0EHA=
+github.com/lestrrat-go/strftime v1.0.3 h1:qqOPU7y+TM8Y803I8fG9c/DyKG3xH/xkng6keC1015Q=
 github.com/lestrrat-go/strftime v1.0.3/go.mod h1:E1nN3pCbtMSu1yjSVeyuRFVm/U0xoR76fd03sz+Qz4g=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
 github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
@@ -220,6 +241,7 @@ github.com/mattn/go-isatty v0.0.11/go.mod h1:PhnuNfih5lzO57/f3n+odYbM4JtupLOxQOA
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/mattn/go-pointer v0.0.0-20190911064623-a0a44394634f h1:QTRRO+ozoYgT3CQRIzNVYJRU3DB8HRnkZv6mr4ISmMA=
 github.com/mattn/go-pointer v0.0.0-20190911064623-a0a44394634f/go.mod h1:2zXcozF6qYGgmsG+SeTZz3oAbFLdD3OWqnUbNvJZAlc=
+github.com/mattn/go-sqlite3 v1.14.5 h1:1IdxlwTNazvbKJQSxoJ5/9ECbEeaTTyeU7sEAZ5KKTQ=
 github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI=
 github.com/mattn/go-tflite v1.0.1 h1:bTfbF7HIF0n3vQsl2JdMUhsFT/KkQuQlCy0UlnF9D4M=
 github.com/mattn/go-tflite v1.0.1/go.mod h1:LME9BQINAkZIOGDVDJJcCa2v0NMuV2AKaf1U47NVS4w=
@@ -229,6 +251,7 @@ github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKju
 github.com/minio/highwayhash v1.0.1/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/mitchellh/cli v1.1.0/go.mod h1:xcISNoH86gajksDmfB23e/pu+B+GeFRMYmoHXxx3xhI=
 github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFWgYaq1OVrnRcwhnw=
+github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw=
 github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
@@ -237,6 +260,7 @@ github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh
 github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag=
 github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
+github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ=
 github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/mmcloughlin/geohash v0.10.0 h1:9w1HchfDfdeLc+jFEf/04D27KP7E2QmpDu52wPbJWRE=
 github.com/mmcloughlin/geohash v0.10.0/go.mod h1:oNZxQo5yWJh0eMQEP/8hwQuVx9Z9tjwFUqcTB1SmG0c=
@@ -276,6 +300,7 @@ github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -301,18 +326,22 @@ github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqn
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
+github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=
 github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
+github.com/shopspring/decimal v1.2.0 h1:abSATXmQEYyShuxI4/vyW3tV1MrKAJzCZ/0zLUXYbsQ=
 github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
+github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
 github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/sony/gobreaker v0.4.1/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=
+github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/streadway/amqp v0.0.0-20190404075320-75d898a42a94/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/amqp v1.0.0/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
@@ -328,6 +357,7 @@ github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5Cc
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/taosdata/driver-go/v2 v2.0.0 h1:ncE+Cz9LfYuHRF4ppzclawodIPf7zpcVaXfvc1VeItk=
 github.com/taosdata/driver-go/v2 v2.0.0/go.mod h1:W7pu74rSvDmGjJPO6fzp+GCtwOelrMgXEhPD0aQJ1xw=
+github.com/tebeka/strftime v0.1.5 h1:1NQKN1NiQgkqd/2moD6ySP/5CoZQsKa1d3ZhJ44Jpmg=
 github.com/tebeka/strftime v0.1.5/go.mod h1:29/OidkoWHdEKZqzyDLUyC+LmgDgdHo4WAFCDT7D/Ig=
 github.com/ugorji/go v1.2.5/go.mod h1:gat2tIT8KJG8TVI8yv77nEO/KYT6dV7JE1gfUa8Xuls=
 github.com/ugorji/go/codec v1.2.5/go.mod h1:QPxoTbPKSEAlAHPYt02++xp/en9B/wUdwFCz+hj5caA=
@@ -337,6 +367,7 @@ github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+github.com/yuin/gopher-lua v0.0.0-20200816102855-ee81675732da h1:NimzV1aGyq29m5ukMK0AMWEhFaL/lrEOaephfuoiARg=
 github.com/yuin/gopher-lua v0.0.0-20200816102855-ee81675732da/go.mod h1:E1AXubJBdNmFERAOucpDIxNzeGfLzg0mYh+UfMWdChA=
 go.etcd.io/etcd/api/v3 v3.5.0/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=
 go.etcd.io/etcd/client/pkg/v3 v3.5.0/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3YSwc9/Ac1g=
@@ -358,6 +389,7 @@ golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200414173820-0848c9571904/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
+golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97 h1:/UOmuWzQfxxo9UtlXMwuQU8CMgg1eZXqTRwkSQJWKOI=
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/image v0.0.0-20190321063152-3fc05d484e9f/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
@@ -499,6 +531,7 @@ gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLks
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=

extensions/sinks/tdengine/tdengine.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,17 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//go:build plugins
-// +build plugins
-
 package main
 
 import (
 	"database/sql"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
-	"github.com/lf-edge/ekuiper/pkg/errorx"
 	_ "github.com/taosdata/driver-go/v2/taosSql"
 	"reflect"
 	"strings"
@@ -32,103 +29,142 @@ type (
 	taosConfig struct {
 		ProvideTs   bool     `json:"provideTs"`
 		Port        int      `json:"port"`
-		Ip          string   `json:"ip"`
+		Ip          string   `json:"ip"` // To be deprecated
+		Host        string   `json:"host"`
 		User        string   `json:"user"`
 		Password    string   `json:"password"`
 		Database    string   `json:"database"`
 		Table       string   `json:"table"`
 		TsFieldName string   `json:"tsFieldName"`
 		Fields      []string `json:"fields"`
+		STable      string   `json:"sTable"`
+		TagFields   []string `json:"tagFields"`
 	}
 	taosSink struct {
 		conf *taosConfig
+		url  string
 		db   *sql.DB
 	}
 )
 
-func (this *taosConfig) delTsField() {
+func (t *taosConfig) delTsField() {
 	var auxFields []string
-	for _, v := range this.Fields {
-		if v != this.TsFieldName {
+	for _, v := range t.Fields {
+		if v != t.TsFieldName {
 			auxFields = append(auxFields, v)
 		}
 	}
-	this.Fields = auxFields
+	t.Fields = auxFields
 }
 
-func (this *taosConfig) buildSql(ctx api.StreamContext, mapData map[string]interface{}) (string, error) {
+func (t *taosConfig) buildSql(ctx api.StreamContext, mapData map[string]interface{}) (string, error) {
 	if 0 == len(mapData) {
 		return "", fmt.Errorf("data is empty.")
 	}
-	if 0 == len(this.TsFieldName) {
-		return "", fmt.Errorf("tsFieldName is empty.")
-	}
-
 	logger := ctx.GetLogger()
-	var keys, vals []string
+	var (
+		table, sTable    string
+		keys, vals, tags []string
+		err              error
+	)
+	table, err = ctx.ParseTemplate(t.Table, mapData)
+	if err != nil {
+		logger.Errorf("parse template for table %s error: %v", t.Table, err)
+		return "", err
+	}
+	sTable, err = ctx.ParseTemplate(t.STable, mapData)
+	if err != nil {
+		logger.Errorf("parse template for sTable %s error: %v", t.STable, err)
+		return "", err
+	}
 
-	if this.ProvideTs {
-		if v, ok := mapData[this.TsFieldName]; !ok {
-			return "", fmt.Errorf("Timestamp field not found : %s.", this.TsFieldName)
+	if t.ProvideTs {
+		if v, ok := mapData[t.TsFieldName]; !ok {
+			return "", fmt.Errorf("Timestamp field not found : %s.", t.TsFieldName)
 		} else {
-			keys = append(keys, this.TsFieldName)
+			keys = append(keys, t.TsFieldName)
 			vals = append(vals, fmt.Sprintf(`"%v"`, v))
-			delete(mapData, this.TsFieldName)
-			this.delTsField()
 		}
 	} else {
 		vals = append(vals, "now")
-		keys = append(keys, this.TsFieldName)
+		keys = append(keys, t.TsFieldName)
 	}
 
-	for _, k := range this.Fields {
-		if v, ok := mapData[k]; ok {
+	if len(t.Fields) != 0 {
+		for _, k := range t.Fields {
+			if k == t.TsFieldName {
+				continue
+			}
+			if v, ok := mapData[k]; ok {
+				keys = append(keys, k)
+				if reflect.String == reflect.TypeOf(v).Kind() {
+					vals = append(vals, fmt.Sprintf(`"%v"`, v))
+				} else {
+					vals = append(vals, fmt.Sprintf(`%v`, v))
+				}
+			} else {
+				logger.Warnln("not found field:", k)
+			}
+		}
+	} else {
+		for k, v := range mapData {
+			if k == t.TsFieldName {
+				continue
+			}
 			keys = append(keys, k)
 			if reflect.String == reflect.TypeOf(v).Kind() {
 				vals = append(vals, fmt.Sprintf(`"%v"`, v))
 			} else {
 				vals = append(vals, fmt.Sprintf(`%v`, v))
 			}
-		} else {
-			logger.Warnln("not found field:", k)
 		}
 	}
 
-	if 0 != len(this.Fields) {
-		if len(this.Fields) < len(mapData) {
-			logger.Warnln("some of values will be ignored.")
+	if len(t.TagFields) > 0 {
+		for _, v := range t.TagFields {
+			switch mapData[v].(type) {
+			case string:
+				tags = append(tags, fmt.Sprintf(`"%s"`, mapData[v]))
+			default:
+				tags = append(tags, fmt.Sprintf(`%v`, mapData[v]))
+			}
 		}
-		return fmt.Sprintf(`INSERT INTO %s (%s)VALUES(%s);`, this.Table, strings.Join(keys, `,`), strings.Join(vals, `,`)), nil
 	}
 
-	for k, v := range mapData {
-		keys = append(keys, k)
-		if reflect.String == reflect.TypeOf(v).Kind() {
-			vals = append(vals, fmt.Sprintf(`"%v"`, v))
-		} else {
-			vals = append(vals, fmt.Sprintf(`%v`, v))
-		}
+	sqlStr := fmt.Sprintf("INSERT INTO %s (%s)", table, strings.Join(keys, ","))
+	if sTable != "" {
+		sqlStr += " using " + sTable
 	}
-	if 0 != len(keys) {
-		return fmt.Sprintf(`INSERT INTO %s (%s)VALUES(%s);`, this.Table, strings.Join(keys, `,`), strings.Join(vals, `,`)), nil
+	if len(tags) != 0 {
+		sqlStr += " tags (" + strings.Join(tags, ",") + ")"
 	}
-	return "", nil
+	sqlStr += " values (" + strings.Join(vals, ",") + ");"
+	return sqlStr, nil
 }
 
 func (m *taosSink) Configure(props map[string]interface{}) error {
-	cfg := &taosConfig{}
+	cfg := &taosConfig{
+		User:     "root",
+		Password: "taosdata",
+	}
 	err := cast.MapToStruct(props, cfg)
 	if err != nil {
 		return fmt.Errorf("read properties %v fail with error: %v", props, err)
 	}
-	if cfg.Ip == "" {
-		cfg.Ip = "127.0.0.1"
+	if cfg.Ip != "" {
+		conf.Log.Warnf("Deprecated: Tdengine sink ip property is deprecated, use host instead.")
+		if cfg.Host == "" {
+			cfg.Host = cfg.Ip
+		}
+	}
+	if cfg.Host == "" {
+		cfg.Host = "localhost"
 	}
 	if cfg.User == "" {
-		cfg.User = "root"
+		return fmt.Errorf("propert user is required.")
 	}
 	if cfg.Password == "" {
-		cfg.Password = "taosdata"
+		return fmt.Errorf("propert password is required.")
 	}
 	if cfg.Database == "" {
 		return fmt.Errorf("property database is required")
@@ -139,38 +175,51 @@ func (m *taosSink) Configure(props map[string]interface{}) error {
 	if cfg.TsFieldName == "" {
 		return fmt.Errorf("property TsFieldName is required")
 	}
+	if cfg.STable != "" && len(cfg.TagFields) == 0 {
+		return fmt.Errorf("property tagFields is required when sTable is set")
+	}
+	m.url = fmt.Sprintf(`%s:%s@tcp(%s:%d)/%s`, cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.Database)
 	m.conf = cfg
 	return nil
 }
 
 func (m *taosSink) Open(ctx api.StreamContext) (err error) {
-	logger := ctx.GetLogger()
-	logger.Debug("Opening tdengine sink")
-	url := fmt.Sprintf(`%s:%s@tcp(%s:%d)/%s`, m.conf.User, m.conf.Password, m.conf.Ip, m.conf.Port, m.conf.Database)
-	m.db, err = sql.Open("taosSql", url)
+	ctx.GetLogger().Debug("Opening tdengine sink")
+	m.db, err = sql.Open("taosSql", m.url)
 	return err
 }
 
 func (m *taosSink) Collect(ctx api.StreamContext, item interface{}) error {
-	logger := ctx.GetLogger()
-	logger.Debugf("tdengine sink receive %s", item)
+	ctx.GetLogger().Debugf("tdengine sink receive %s", item)
+
+	switch v := item.(type) {
+	case []map[string]interface{}:
+		var err error
+		for _, mapData := range v {
+			e := m.writeToDB(ctx, mapData)
+			if e != nil {
+				err = e
+			}
+		}
+		return err
+	case map[string]interface{}:
+		return m.writeToDB(ctx, v)
+	default: // never happen
+		return fmt.Errorf("unsupported type: %T", item)
+	}
+}
 
-	sliData, ok := item.([]map[string]interface{})
-	if !ok {
-		return fmt.Errorf("tdengine sink receive non map slice data: %#v", item)
+func (m *taosSink) writeToDB(ctx api.StreamContext, mapData map[string]interface{}) error {
+	sql, err := m.conf.buildSql(ctx, mapData)
+	if nil != err {
+		return err
 	}
-	for _, mapData := range sliData {
-		sql, err := m.conf.buildSql(ctx, mapData)
-		if nil != err {
-			return err
-		}
-		logger.Debugf(sql)
-		rows, err := m.db.Query(sql)
-		if err != nil {
-			return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-		}
-		rows.Close()
+	ctx.GetLogger().Debugf(sql)
+	rows, err := m.db.Query(sql)
+	if err != nil {
+		return err
 	}
+	rows.Close()
 	return nil
 }
 

extensions/sinks/tdengine/tdengine_test.go

@@ -0,0 +1,237 @@
+// 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 (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/testx"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/state"
+	"reflect"
+	"testing"
+)
+
+func TestConfig(t *testing.T) {
+	var tests = []struct {
+		conf     map[string]interface{}
+		expected *taosConfig
+		error    string
+	}{
+		{ //0
+			conf: map[string]interface{}{
+				"host":        "e0d9d8089bef",
+				"port":        6030,
+				"user":        "root",
+				"password":    "taosdata",
+				"database":    "db",
+				"table":       "t",
+				"tsfieldname": "ts",
+			},
+			expected: &taosConfig{
+				ProvideTs:   false,
+				Host:        "e0d9d8089bef",
+				Port:        6030,
+				User:        "root",
+				Password:    "taosdata",
+				Database:    "db",
+				Table:       "t",
+				TsFieldName: "ts",
+				Fields:      nil,
+			},
+		},
+		{ //1
+			conf: map[string]interface{}{
+				"ip":          "e0d9d8089bef",
+				"port":        6030,
+				"user":        "root1",
+				"password":    "taosdata1",
+				"database":    "db",
+				"table":       "t",
+				"provideTs":   true,
+				"tsfieldname": "ts",
+			},
+			expected: &taosConfig{
+				ProvideTs:   true,
+				Ip:          "e0d9d8089bef",
+				Host:        "e0d9d8089bef",
+				Port:        6030,
+				User:        "root1",
+				Password:    "taosdata1",
+				Database:    "db",
+				Table:       "t",
+				TsFieldName: "ts",
+				Fields:      nil,
+			},
+		},
+		{ //2
+			conf: map[string]interface{}{
+				"port":        6030,
+				"database":    "dab",
+				"table":       "tt",
+				"tsfieldname": "tst",
+				"fields":      []string{"f1", "f2"},
+				"sTable":      "s",
+				"tagFields":   []string{"a", "b"},
+			},
+			expected: &taosConfig{
+				ProvideTs:   false,
+				Ip:          "",
+				Host:        "localhost",
+				Port:        6030,
+				User:        "root",
+				Password:    "taosdata",
+				Database:    "dab",
+				Table:       "tt",
+				TsFieldName: "tst",
+				Fields:      []string{"f1", "f2"},
+				STable:      "s",
+				TagFields:   []string{"a", "b"},
+			},
+		},
+		{ //3
+			conf: map[string]interface{}{
+				"port":     6030,
+				"database": "dab",
+				"table":    "t",
+				"fields":   []string{"f1", "f2"},
+			},
+			error: "property TsFieldName is required",
+		},
+		{ //4
+			conf: map[string]interface{}{
+				"port":        6030,
+				"database":    "dab",
+				"table":       "tt",
+				"tsfieldname": "tst",
+				"fields":      []string{"f1", "f2"},
+				"sTable":      "s",
+			},
+			error: "property tagFields is required when sTable is set",
+		},
+	}
+
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	for i, test := range tests {
+		tdsink := &taosSink{}
+		err := tdsink.Configure(test.conf)
+		if !reflect.DeepEqual(test.error, testx.Errstring(err)) {
+			t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, test.error, err)
+		} else if test.error == "" && !reflect.DeepEqual(test.expected, tdsink.conf) {
+			t.Errorf("%d\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, test.expected, tdsink.conf)
+		}
+	}
+}
+
+func TestBuildSql(t *testing.T) {
+	var tests = []struct {
+		conf     *taosConfig
+		data     map[string]interface{}
+		expected string
+		error    string
+	}{
+		{
+			conf: &taosConfig{
+				ProvideTs:   false,
+				Host:        "e0d9d8089bef",
+				Port:        6030,
+				User:        "root",
+				Password:    "taosdata",
+				Database:    "db",
+				Table:       "t",
+				TsFieldName: "ts",
+				Fields:      nil,
+			},
+			data: map[string]interface{}{
+				"f1": "v1",
+			},
+			expected: `INSERT INTO t (ts,f1) values (now,"v1");`,
+		},
+		{
+			conf: &taosConfig{
+				ProvideTs:   true,
+				Ip:          "e0d9d8089bef",
+				Host:        "e0d9d8089bef",
+				Port:        6030,
+				User:        "root1",
+				Password:    "taosdata1",
+				Database:    "db",
+				Table:       "t",
+				TsFieldName: "ts",
+				Fields:      nil,
+			},
+			data: map[string]interface{}{
+				"ts": 12345678,
+				"f2": 65,
+			},
+			expected: `INSERT INTO t (ts,f2) values ("12345678",65);`,
+		},
+		{
+			conf: &taosConfig{
+				ProvideTs:   true,
+				Ip:          "e0d9d8089bef",
+				Host:        "e0d9d8089bef",
+				Port:        6030,
+				User:        "root1",
+				Password:    "taosdata1",
+				Database:    "db",
+				Table:       "t",
+				TsFieldName: "ts",
+				Fields:      nil,
+			},
+			data: map[string]interface{}{
+				"ts": 12345678,
+				"f2": 65,
+			},
+			expected: `INSERT INTO t (ts,f2) values ("12345678",65);`,
+		},
+		{
+			conf: &taosConfig{
+				ProvideTs:   false,
+				Ip:          "",
+				Host:        "localhost",
+				Port:        6030,
+				User:        "root",
+				Password:    "taosdata",
+				Database:    "dab",
+				Table:       "{{.table}}",
+				TsFieldName: "tst",
+				Fields:      []string{"f1", "f2"},
+				STable:      "s",
+				TagFields:   []string{"a", "b"},
+			},
+			data: map[string]interface{}{
+				"table": "t1",
+				"ts":    12345678,
+				"f2":    65,
+				"f1":    12.3,
+				"a":     "a1",
+				"b":     2,
+			},
+			expected: `INSERT INTO t1 (tst,f1,f2) using s tags ("a1",2) values (now,12.3,65);`,
+		},
+	}
+	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
+	contextLogger := conf.Log.WithField("rule", "mockRule0")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger).WithMeta("testTD", "op1", &state.MemoryStore{})
+	for i, test := range tests {
+		sql, err := test.conf.buildSql(ctx, test.data)
+		if !reflect.DeepEqual(test.error, testx.Errstring(err)) {
+			t.Errorf("%d: error mismatch:\n  exp=%s\n  got=%s\n\n", i, test.error, err)
+		} else if test.error == "" && !reflect.DeepEqual(test.expected, sql) {
+			t.Errorf("%d\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, test.expected, sql)
+		}
+	}
+}