Repository: lf-edge/ekuiper
Pull request #1422 (2022-09-29 09:06:19)

feat(sink): updatable SQL sink
1. Support updatable for SQL sink
2. Add unit test
3. Fix a bug for writing []map, generated SQL miss table


Patch:
diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index bbd58bb5..e3b98fb6 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -181,7 +181,6 @@ jobs:
           - sinks/influx
           - sinks/tdengine
           - sinks/zmq
-          - sinks/redis
           - sinks/sql
           - sources/random
           - sources/zmq
diff --git a/Makefile b/Makefile
index 64da927f..07c2c2a3 100644
--- a/Makefile
+++ b/Makefile
@@ -85,7 +85,6 @@ PLUGINS := sinks/file \
 	sinks/influx2 \
 	sinks/zmq \
 	sinks/image \
-	sinks/redis \
 	sinks/sql   \
 	sources/random \
 	sources/zmq \
diff --git a/extensions/go.mod b/extensions/go.mod
index e68b6493..5ec97697 100644
--- a/extensions/go.mod
+++ b/extensions/go.mod
@@ -13,7 +13,6 @@ require (
 	github.com/btnguyen2k/gocosmos v0.1.6
 	github.com/couchbase/go_n1ql v0.0.0-20220303011133-0ed4bf93e31d
 	github.com/denisenkom/go-mssqldb v0.12.2
-	github.com/go-redis/redis/v7 v7.3.0
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/godror/godror v0.34.0
 	github.com/googleapis/go-sql-spanner v0.0.0-20220321120010-12780e57be1c
diff --git a/extensions/go.sum b/extensions/go.sum
index 922716a5..cfae42f0 100644
--- a/extensions/go.sum
+++ b/extensions/go.sum
@@ -277,7 +277,6 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoD
 github.com/form3tech-oss/jwt-go v3.2.5+incompatible h1:/l4kBbb4/vGSsdtB5nUe8L7B9mImVMaBPw9L/0TBHU8=
 github.com/form3tech-oss/jwt-go v3.2.5+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
-github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/gabriel-vasile/mimetype v1.4.0 h1:Cn9dkdYsMIu56tGho+fqzh7XmvY2YyGU0FnbhiOsEro=
 github.com/gabriel-vasile/mimetype v1.4.0/go.mod h1:fA8fi6KUiG7MgQQ+mEWotXoEOvmxRtOJlERCzSmRvr8=
 github.com/getkin/kin-openapi v0.61.0/go.mod h1:7Yn5whZr5kJi6t+kShccXS8ae1APpYTW6yheSwk8Yi4=
@@ -307,8 +306,6 @@ github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34
 github.com/go-openapi/strfmt v0.19.5 h1:0utjKrw+BAh8s57XE9Xz8DUBsVvPmRUB6styvl9wWIM=
 github.com/go-openapi/strfmt v0.19.5/go.mod h1:eftuHTlB/dI8Uq8JJOyRlieZf+WkkxUuk0dgdHXr2Qk=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
-github.com/go-redis/redis/v7 v7.3.0 h1:3oHqd0W7f/VLKBxeYTEpqdMUsmMectngjM9OtoRoIgg=
-github.com/go-redis/redis/v7 v7.3.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=
 github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
@@ -435,8 +432,6 @@ github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/C
 github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
-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=
@@ -624,11 +619,6 @@ github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S/RLdc7JQKbRpFeM1dOBd8T9ki5s+AY8=
 github.com/nishanths/predeclared v0.0.0-20200524104333-86fad755b4d3/go.mod h1:nt3d53pc1VYcphSCIaYAJtnPYnr3Zyn8fMq2wvPGPso=
 github.com/oklog/ulid/v2 v2.0.2 h1:r4fFzBm+bv0wNKNh5eXTwU7i85y5x+uwkxCUTNVQqLc=
-github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
-github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
-github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.0.2 h1:9yCKha/T5XdGtO0q9Q9a6T5NUCsTn/DrBg0D7ufOcFM=
@@ -868,7 +858,6 @@ golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -880,7 +869,6 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191112182307-2180aed22343/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -943,7 +931,6 @@ golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20220513210516-0976fa681c29 h1:w8s32wxx3sY+OjLlv9qltkLU5yvJzxjjgiHWLjdIcw4=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -960,7 +947,6 @@ golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191112214154-59a1497f0cea/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191115151921-52ab43148777/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1278,12 +1264,9 @@ google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscL
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 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=
-gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/inconshreveable/log15.v2 v2.0.0-20180818164646-67afb5ed74ec/go.mod h1:aPpfJ7XW+gOuirDoZ8gHhLh3kZ1B08FtV2bbmy7Jv3s=
 gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw=
 gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo=
@@ -1295,12 +1278,8 @@ gopkg.in/jcmturner/gokrb5.v6 v6.1.1 h1:n0KFjpbuM5pFMN38/Ay+Br3l91netGSVqHPHEXeWU
 gopkg.in/jcmturner/gokrb5.v6 v6.1.1/go.mod h1:NFjHNLrHQiruory+EmqDXCGv6CrjkeYeA+bR9mIfNFk=
 gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU=
 gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8=
-gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
-gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
-gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
@@ -1308,7 +1287,6 @@ gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA=
 gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/extensions/sinks/redis/redis.go b/extensions/sinks/redis/redis.go
deleted file mode 100755
index ab850622..00000000
--- a/extensions/sinks/redis/redis.go
+++ /dev/null
@@ -1,211 +0,0 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package main
-
-import (
-	"errors"
-	"fmt"
-	"github.com/lf-edge/ekuiper/pkg/errorx"
-	"time"
-
-	"github.com/go-redis/redis/v7"
-
-	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/cast"
-)
-
-type RedisSink struct {
-	// host:port address.
-	addr     string
-	username string
-	// Optional password. Must match the password specified in the
-	password string
-	// Database to be selected after connecting to the server.
-	db int
-
-	// key of field
-	field string
-
-	// key define
-	key string
-
-	dataType string
-
-	expiration time.Duration
-
-	sendSingle bool
-
-	cli *redis.Client
-}
-
-func (r *RedisSink) Configure(props map[string]interface{}) error {
-	if i, ok := props["addr"]; ok {
-		if i, ok := i.(string); ok {
-			r.addr = i
-		}
-	} else {
-		return errors.New("redis addr is null")
-	}
-
-	if i, ok := props["password"]; ok {
-		if i, ok := i.(string); ok {
-			r.password = i
-		}
-	}
-
-	r.db = 0
-	if i, ok := props["db"]; ok {
-		if t, err := cast.ToInt(i, cast.STRICT); err == nil {
-			r.db = t
-		}
-	}
-
-	if i, ok := props["key"]; ok {
-		if i, ok := i.(string); ok {
-			r.key = i
-		}
-	} else {
-		return errors.New("not config data key for redis")
-	}
-
-	if i, ok := props["field"]; ok {
-		if i, ok := i.(string); ok {
-			r.field = i
-		}
-	}
-
-	r.sendSingle = true
-	if i, ok := props["sendSingle"]; ok {
-		if i, ok := i.(bool); ok {
-			r.sendSingle = i
-		}
-	}
-
-	r.dataType = "string"
-	if i, ok := props["dataType"]; ok {
-		if i, ok := i.(string); ok {
-			r.dataType = i
-		}
-	}
-
-	r.expiration = -1
-	if i, ok := props["expiration"]; ok {
-		if t, err := cast.ToInt(i, cast.STRICT); err == nil {
-			r.expiration = time.Duration(t)
-		}
-	}
-
-	return nil
-}
-
-func (r *RedisSink) Open(ctx api.StreamContext) (err error) {
-	logger := ctx.GetLogger()
-	logger.Debug("Opening redis sink")
-
-	r.cli = redis.NewClient(&redis.Options{
-		Addr:     r.addr,
-		Username: r.username,
-		Password: r.password,
-		DB:       r.db, // use default DB
-	})
-
-	return nil
-}
-
-func (r *RedisSink) Collect(ctx api.StreamContext, data interface{}) error {
-	logger := ctx.GetLogger()
-	v, _, err := ctx.TransformOutput(data)
-	if err != nil {
-		logger.Error(err)
-		return err
-	}
-	if r.field != "" {
-		switch out := data.(type) {
-		case []map[string]interface{}:
-			for _, m := range out {
-				key := r.field
-				field, ok := m[key].(string)
-				if ok {
-					key = field
-				}
-
-				if r.dataType == "list" {
-					err := r.cli.LPush(key, v).Err()
-					if err != nil {
-						logger.Error(err)
-						return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-					}
-					logger.Debugf("send redis list success, key:%s data: %s", key, string(v))
-				} else {
-					err := r.cli.Set(key, v, r.expiration*time.Second).Err()
-					if err != nil {
-						logger.Error(err)
-						return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-					}
-					logger.Debugf("send redis string success, key:%s data: %s", key, string(v))
-				}
-			}
-		case map[string]interface{}:
-			key := r.field
-			field, ok := out[key].(string)
-			if ok {
-				key = field
-			}
-
-			if r.dataType == "list" {
-				err := r.cli.LPush(key, v).Err()
-				if err != nil {
-					logger.Error(err)
-					return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-				}
-				logger.Debugf("send redis list success, key:%s data: %s", key, string(v))
-			} else {
-				err := r.cli.Set(key, v, r.expiration*time.Second).Err()
-				if err != nil {
-					logger.Error(err)
-					return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-				}
-				logger.Debugf("send redis string success, key:%s data: %s", key, string(v))
-			}
-		}
-	} else if r.key != "" {
-		if r.dataType == "list" {
-			err := r.cli.LPush(r.key, v).Err()
-			if err != nil {
-				logger.Error(err)
-				return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-			}
-			logger.Debugf("send redis list success, key:%s data: %s", r.key, string(v))
-		} else {
-			err := r.cli.Set(r.key, v, r.expiration*time.Second).Err()
-			if err != nil {
-				logger.Error(err)
-				return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-			}
-			logger.Debugf("send redis string success, key:%s data: %s", r.key, string(v))
-		}
-	}
-	logger.Debug("insert success %v", data)
-	return nil
-}
-
-func (r *RedisSink) Close(ctx api.StreamContext) error {
-	err := r.cli.Close()
-	return err
-}
-
-func Redis() api.Sink {
-	return &RedisSink{}
-}
diff --git a/extensions/sinks/redis/redis.json b/extensions/sinks/redis/redis.json
deleted file mode 100644
index f115d146..00000000
--- a/extensions/sinks/redis/redis.json
+++ /dev/null
@@ -1,141 +0,0 @@
-{
-	"about": {
-		"trial": false,
-		"author": {
-			"name": "DengFeng HE",
-			"email": "894220128@qq.com",
-			"company": "personal",
-			"website": "https://github.com/feng-crazy"
-		},
-		"helpUrl": {
-			"en_US": "https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/rules/sinks/plugin/reids.md",
-			"zh_CN": "https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/rules/sinks/plugin/reids.md"
-		},
-		"description": {
-			"en_US": "This a sink for Redis, it can be used for saving the analysis data into Redis.",
-			"zh_CN": "为 Redis 的持久化插件，可以用于将分析数据存入 Redis 中"
-		}
-	},
-	"libs": [
-		"github.com/go-redis/redis/v7"
-	],
-	"properties": [
-		{
-			"name": "addr",
-			"default": "10.122.48.17:6379",
-			"optional": false,
-			"control": "text",
-			"type": "string",
-			"hint": {
-				"en_US": "The addr of the Redis,example: 10.122.48.17:6379",
-				"zh_CN": "Redis的地址, 例如: 10.122.48.17:6379"
-			},
-			"label": {
-				"en_US": "Addr",
-				"zh_CN": "地址"
-			}
-		},
-		{
-			"name": "password",
-			"default": "",
-			"optional": true,
-			"control": "text",
-			"type": "string",
-			"hint": {
-				"en_US": "The Redis login password",
-				"zh_CN": "Redis 登陆密码"
-			},
-			"label": {
-				"en_US": "Password",
-				"zh_CN": "密码"
-			}
-		},
-		{
-			"name": "db",
-			"default": 0,
-			"optional": false,
-			"control": "text",
-			"type": "int",
-			"hint": {
-				"en_US": "The database of the Redis,example: 0",
-				"zh_CN": "Redis 的数据库,例如0"
-			},
-			"label": {
-				"en_US": "DataBase name",
-				"zh_CN": "数据库名"
-			}
-		},
-		{
-			"name": "key",
-			"default": "key",
-			"optional": false,
-			"control": "text",
-			"type": "string",
-			"hint": {
-				"en_US": "Select one of the Key, Key and field of Redis data and give priority to field",
-				"zh_CN": "Redis 数据的Key， key与field选择其中一个,优先field"
-			},
-			"label": {
-				"en_US": "Key",
-				"zh_CN": "Key"
-			}
-		},
-		{
-			"name": "field",
-			"default": "deviceName",
-			"optional": true,
-			"control": "text",
-			"type": "string",
-			"hint": {
-				"en_US": "This field must exist and be of type string. Otherwise, use the field character as the key. Note: Do not use a data template to configure this value",
-				"zh_CN": "json数据某一个属性，配置它作为redis数据的key值, 例如 deviceName,该字段必须存在且为string类型，否则以field字符作为key,注意:配置该值不要使用数据模板"
-			},
-			"label": {
-				"en_US": "field",
-				"zh_CN": "field"
-			}
-		},
-		{
-			"name": "dataType",
-			"default": "string",
-			"optional": false,
-			"control": "select",
-			"type": "string",
-			"values": [
-				"string",
-				"list"
-			],
-			"hint": {
-				"en_US": "The default Redis data type is string. Note that the original key must be deleted after the Redis data type is changed. Otherwise, the modification is invalid。",
-				"zh_CN": "Redis 数据的类型, 默认是 string, 注意修改类型之后，需在redis中删除原有key，否则修改无效。"
-			},
-			"label": {
-				"en_US": "data type",
-				"zh_CN": "数据类型"
-			}
-		},
-		{
-			"name": "expiration",
-			"default": -1,
-			"optional": false,
-			"control": "text",
-			"type": "int",
-			"hint": {
-				"en_US": "Timeout duration of Redis data. This parameter is valid only for string data in seconds. The default value is -1 ",
-				"zh_CN": "Redis数据的超时时间，仅在string类型数据有效,单位是秒,默认是永久保存-1 "
-			},
-			"label": {
-				"en_US": "expiration",
-				"zh_CN": "超时时间"
-			}
-		}
-	],
-	"node": {
-		"category": "Sink",
-		"icon": "iconPath",
-		"label": {
-			"en": "Redis",
-			"zh": "Redis"
-		}
-	}
-}
diff --git a/extensions/sinks/sql/sql.go b/extensions/sinks/sql/sql.go
index 3d50885c..b734c8fa 100644
--- a/extensions/sinks/sql/sql.go
+++ b/extensions/sinks/sql/sql.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/extensions/sqldatabase/driver"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/xo/dburl"
@@ -33,11 +34,22 @@ type sqlConfig struct {
 	Fields         []string `json:"fields"`
 	DataTemplate   string   `json:"dataTemplate"`
 	TableDataField string   `json:"tableDataField"`
+	RowkindField   string   `json:"rowkindField"`
+	KeyField       string   `json:"keyField"`
 }
 
-func (t *sqlConfig) buildSql(ctx api.StreamContext, mapData map[string]interface{}) ([]string, string, error) {
+func (t *sqlConfig) buildInsertSql(ctx api.StreamContext, mapData map[string]interface{}) ([]string, string, error) {
+	keys, vals, err := t.getKeyValues(ctx, mapData)
+	if err != nil {
+		return keys, "", err
+	}
+	sqlStr := "(" + strings.Join(vals, ",") + ")"
+	return keys, sqlStr, nil
+}
+
+func (t *sqlConfig) getKeyValues(ctx api.StreamContext, mapData map[string]interface{}) ([]string, []string, error) {
 	if 0 == len(mapData) {
-		return nil, "", fmt.Errorf("data is empty.")
+		return nil, nil, fmt.Errorf("data is empty.")
 	}
 	logger := ctx.GetLogger()
 	var keys, vals []string
@@ -66,9 +78,7 @@ func (t *sqlConfig) buildSql(ctx api.StreamContext, mapData map[string]interface
 			}
 		}
 	}
-
-	sqlStr := "(" + strings.Join(vals, ",") + ")"
-	return keys, sqlStr, nil
+	return keys, vals, nil
 }
 
 type sqlSink struct {
@@ -89,6 +99,9 @@ func (m *sqlSink) Configure(props map[string]interface{}) error {
 	if cfg.Table == "" {
 		return fmt.Errorf("property Table is required")
 	}
+	if cfg.RowkindField != "" && cfg.KeyField == "" {
+		return fmt.Errorf("keyField is required when rowkindField is set")
+	}
 	m.conf = cfg
 	return nil
 }
@@ -108,11 +121,16 @@ func (m *sqlSink) Open(ctx api.StreamContext) (err error) {
 
 func (m *sqlSink) writeToDB(ctx api.StreamContext, sqlStr *string) error {
 	ctx.GetLogger().Debugf(*sqlStr)
-	rows, err := m.db.Query(*sqlStr)
+	r, err := m.db.Exec(*sqlStr)
 	if err != nil {
 		return fmt.Errorf("%s: %s", errorx.IOErr, err.Error())
 	}
-	return rows.Close()
+	d, err := r.RowsAffected()
+	if err != nil {
+		ctx.GetLogger().Errorf("get rows affected error: %s", err.Error())
+	}
+	ctx.GetLogger().Debugf("Rows affected: %d", d)
+	return nil
 }
 
 func (m *sqlSink) Collect(ctx api.StreamContext, item interface{}) error {
@@ -130,10 +148,12 @@ func (m *sqlSink) Collect(ctx api.StreamContext, item interface{}) error {
 		item = tm
 	}
 
-	var table string
-	var err error
-	v, ok := item.(map[string]interface{})
-	if ok {
+	var (
+		table string
+		err   error
+	)
+	switch v := item.(type) {
+	case map[string]interface{}:
 		table, err = ctx.ParseTemplate(m.conf.Table, v)
 		if err != nil {
 			ctx.GetLogger().Errorf("parse template for table %s error: %v", m.conf.Table, err)
@@ -142,60 +162,102 @@ func (m *sqlSink) Collect(ctx api.StreamContext, item interface{}) error {
 		if m.conf.TableDataField != "" {
 			item = v[m.conf.TableDataField]
 		}
+	case []map[string]interface{}:
+		if len(v) == 0 {
+			ctx.GetLogger().Warnf("empty data array")
+			return nil
+		}
+		table, err = ctx.ParseTemplate(m.conf.Table, v[0])
+		if err != nil {
+			ctx.GetLogger().Errorf("parse template for table %s error: %v", m.conf.Table, err)
+			return err
+		}
 	}
 
 	var keys []string = nil
 	var values []string = nil
 	var vars string
 
-	switch v := item.(type) {
-	case []map[string]interface{}:
-		for _, mapData := range v {
-			keys, vars, err = m.conf.buildSql(ctx, mapData)
+	if m.conf.RowkindField == "" {
+		switch v := item.(type) {
+		case []map[string]interface{}:
+			for _, mapData := range v {
+				keys, vars, err = m.conf.buildInsertSql(ctx, mapData)
+				if err != nil {
+					return err
+				}
+				values = append(values, vars)
+			}
+			if keys != nil {
+				sqlStr := fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
+				return m.writeToDB(ctx, &sqlStr)
+			}
+			return nil
+		case map[string]interface{}:
+			keys, vars, err = m.conf.buildInsertSql(ctx, v)
 			if err != nil {
 				return err
 			}
 			values = append(values, vars)
-		}
-		if keys != nil {
-			sqlStr := fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
-			return m.writeToDB(ctx, &sqlStr)
-		}
-		return nil
-	case map[string]interface{}:
-		keys, vars, err = m.conf.buildSql(ctx, v)
-		if err != nil {
-			return err
-		}
-		values = append(values, vars)
-		if keys != nil {
-			sqlStr := fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
-			return m.writeToDB(ctx, &sqlStr)
-		}
-		return nil
-	case []interface{}:
-		for _, data := range v {
-			mapData, ok := data.(map[string]interface{})
-			if !ok {
-				ctx.GetLogger().Errorf("unsupported type: %T", data)
-				return fmt.Errorf("unsupported type: %T", data)
+			if keys != nil {
+				sqlStr := fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
+				return m.writeToDB(ctx, &sqlStr)
 			}
+			return nil
+		case []interface{}:
+			for _, data := range v {
+				mapData, ok := data.(map[string]interface{})
+				if !ok {
+					ctx.GetLogger().Errorf("unsupported type: %T", data)
+					return fmt.Errorf("unsupported type: %T", data)
+				}
 
-			keys, vars, err = m.conf.buildSql(ctx, mapData)
+				keys, vars, err = m.conf.buildInsertSql(ctx, mapData)
+				if err != nil {
+					ctx.GetLogger().Errorf("sql sink build sql error %v for data", err, mapData)
+					return err
+				}
+				values = append(values, vars)
+			}
+
+			if keys != nil {
+				sqlStr := fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
+				return m.writeToDB(ctx, &sqlStr)
+			}
+			return nil
+		default: // never happen
+			return fmt.Errorf("unsupported type: %T", item)
+		}
+	} else {
+		switch d := item.(type) {
+		case []map[string]interface{}:
+			for _, el := range d {
+				err := m.save(ctx, table, el)
+				if err != nil {
+					ctx.GetLogger().Error(err)
+				}
+			}
+		case map[string]interface{}:
+			err := m.save(ctx, table, d)
 			if err != nil {
-				ctx.GetLogger().Errorf("sql sink build sql error %v for data", err, mapData)
 				return err
 			}
-			values = append(values, vars)
-		}
-
-		if keys != nil {
-			sqlStr := fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
-			return m.writeToDB(ctx, &sqlStr)
+		case []interface{}:
+			for _, vv := range d {
+				el, ok := vv.(map[string]interface{})
+				if !ok {
+					ctx.GetLogger().Errorf("unsupported type: %T", vv)
+					return fmt.Errorf("unsupported type: %T", vv)
+				}
+				err := m.save(ctx, table, el)
+				if err != nil {
+					ctx.GetLogger().Error(err)
+				}
+			}
+		default:
+			return fmt.Errorf("unrecognized format of %s", item)
 		}
 		return nil
-	default: // never happen
-		return fmt.Errorf("unsupported type: %T", item)
 	}
 }
 
@@ -206,6 +268,67 @@ func (m *sqlSink) Close(_ api.StreamContext) error {
 	return nil
 }
 
+// save save updatable data only to db
+func (m *sqlSink) save(ctx api.StreamContext, table string, data map[string]interface{}) error {
+	rowkind := ast.RowkindInsert
+	c, ok := data[m.conf.RowkindField]
+	if ok {
+		rowkind, ok = c.(string)
+		if !ok {
+			return fmt.Errorf("rowkind field %s is not a string in data %v", m.conf.RowkindField, data)
+		}
+		if rowkind != ast.RowkindInsert && rowkind != ast.RowkindUpdate && rowkind != ast.RowkindDelete {
+			return fmt.Errorf("invalid rowkind %s", rowkind)
+		}
+	}
+	var sqlStr string
+	switch rowkind {
+	case ast.RowkindInsert:
+		keys, vars, err := m.conf.buildInsertSql(ctx, data)
+		if err != nil {
+			return err
+		}
+		values := []string{vars}
+		if keys != nil {
+			sqlStr = fmt.Sprintf("INSERT INTO %s (%s) values ", table, strings.Join(keys, ",")) + strings.Join(values, ",") + ";"
+		}
+	case ast.RowkindUpdate:
+		keyval, ok := data[m.conf.KeyField]
+		if !ok {
+			return fmt.Errorf("field %s does not exist in data %v", m.conf.KeyField, data)
+		}
+		keys, vals, err := m.conf.getKeyValues(ctx, data)
+		if err != nil {
+			return err
+		}
+		sqlStr = fmt.Sprintf("UPDATE %s SET ", table)
+		for i, key := range keys {
+			if i != 0 {
+				sqlStr += ","
+			}
+			sqlStr += fmt.Sprintf("%s=%s", key, vals[i])
+		}
+		if _, ok := keyval.(string); ok {
+			sqlStr += fmt.Sprintf(" WHERE %s = \"%s\";", m.conf.KeyField, keyval)
+		} else {
+			sqlStr += fmt.Sprintf(" WHERE %s = %v;", m.conf.KeyField, keyval)
+		}
+	case ast.RowkindDelete:
+		keyval, ok := data[m.conf.KeyField]
+		if !ok {
+			return fmt.Errorf("field %s does not exist in data %v", m.conf.KeyField, data)
+		}
+		if _, ok := keyval.(string); ok {
+			sqlStr = fmt.Sprintf("DELETE FROM %s WHERE %s = \"%s\";", table, m.conf.KeyField, keyval)
+		} else {
+			sqlStr = fmt.Sprintf("DELETE FROM %s WHERE %s = %v;", table, m.conf.KeyField, keyval)
+		}
+	default:
+		return fmt.Errorf("invalid rowkind %s", rowkind)
+	}
+	return m.writeToDB(ctx, &sqlStr)
+}
+
 func Sql() api.Sink {
 	return &sqlSink{}
 }
diff --git a/internal/server/meta_plugin_init.go b/internal/server/meta_plugin_init.go
index d5fec378..8e0f8858 100644
--- a/internal/server/meta_plugin_init.go
+++ b/internal/server/meta_plugin_init.go
@@ -75,7 +75,7 @@ func prebuildPluginsHandler(w http.ResponseWriter, _ *http.Request, t plugin.Plu
 }
 
 var NativeSourcePlugin = []string{"random", "zmq", "sql"}
-var NativeSinkPlugin = []string{"file", "image", "influx", "redis", "tdengine", "zmq", "sql"}
+var NativeSinkPlugin = []string{"file", "image", "influx", "tdengine", "zmq", "sql"}
 var NativeFunctionPlugin = []string{"accumulateWordCount", "countPlusOne", "echo", "geohash", "image", "labelImage"}
 
 func fetchPluginList(t plugin.PluginType, hosts, os, arch string) (err error, result map[string]string) {
diff --git a/internal/topo/redis/sink.go b/internal/topo/redis/sink.go
index 01ce832f..38f895f3 100644
--- a/internal/topo/redis/sink.go
+++ b/internal/topo/redis/sink.go
@@ -17,9 +17,10 @@
 package redis
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 	"time"
 
 	"github.com/go-redis/redis/v7"
@@ -28,87 +29,42 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/cast"
 )
 
-type RedisSink struct {
+type config struct {
 	// host:port address.
-	addr     string
-	username string
+	Addr     string `json:"addr,omitempty"`
+	Username string `json:"username,omitempty"`
 	// Optional password. Must match the password specified in the
-	password string
+	Password string `json:"password,omitempty"`
 	// Database to be selected after connecting to the server.
-	db int
-
+	Db int `json:"db,omitempty"`
 	// key of field
-	field string
-
+	Field string `json:"field,omitempty"`
 	// key define
-	key string
-
-	dataType string
-
-	expiration time.Duration
-
-	sendSingle bool
+	Key          string        `json:"key,omitempty"`
+	DataType     string        `json:"dataType,omitempty"`
+	Expiration   time.Duration `json:"expiration,omitempty"`
+	RowkindField string        `json:"rowkindField"`
+	DataTemplate string        `json:"dataTemplate"`
+}
 
+type RedisSink struct {
+	c   *config
 	cli *redis.Client
 }
 
 func (r *RedisSink) Configure(props map[string]interface{}) error {
-	if i, ok := props["addr"]; ok {
-		if i, ok := i.(string); ok {
-			r.addr = i
-		}
-	} else {
-		return errors.New("redis addr is null")
-	}
-
-	if i, ok := props["password"]; ok {
-		if i, ok := i.(string); ok {
-			r.password = i
-		}
-	}
-
-	r.db = 0
-	if i, ok := props["db"]; ok {
-		if t, err := cast.ToInt(i, cast.STRICT); err == nil {
-			r.db = t
-		}
-	}
-
-	if i, ok := props["key"]; ok {
-		if i, ok := i.(string); ok {
-			r.key = i
-		}
-	} else {
-		return errors.New("not config data key for redis")
+	c := &config{DataType: "string", Expiration: -1}
+	err := cast.MapToStruct(props, c)
+	if err != nil {
+		return err
 	}
-
-	if i, ok := props["field"]; ok {
-		if i, ok := i.(string); ok {
-			r.field = i
-		}
+	if c.Key == "" && c.Field == "" {
+		return errors.New("redis sink must have key or field")
 	}
-
-	r.sendSingle = true
-	if i, ok := props["sendSingle"]; ok {
-		if i, ok := i.(bool); ok {
-			r.sendSingle = i
-		}
+	if c.DataType != "string" && c.DataType != "list" {
+		return errors.New("redis sink only support string or list data type")
 	}
-
-	r.dataType = "string"
-	if i, ok := props["dataType"]; ok {
-		if i, ok := i.(string); ok {
-			r.dataType = i
-		}
-	}
-
-	r.expiration = -1
-	if i, ok := props["expiration"]; ok {
-		if t, err := cast.ToInt(i, cast.STRICT); err == nil {
-			r.expiration = time.Duration(t)
-		}
-	}
-
+	r.c = c
 	return nil
 }
 
@@ -117,10 +73,10 @@ func (r *RedisSink) Open(ctx api.StreamContext) (err error) {
 	logger.Debug("Opening redis sink")
 
 	r.cli = redis.NewClient(&redis.Options{
-		Addr:     r.addr,
-		Username: r.username,
-		Password: r.password,
-		DB:       r.db, // use default DB
+		Addr:     r.c.Addr,
+		Username: r.c.Username,
+		Password: r.c.Password,
+		DB:       r.c.Db, // use default DB
 	})
 
 	return nil
@@ -128,86 +84,118 @@ func (r *RedisSink) Open(ctx api.StreamContext) (err error) {
 
 func (r *RedisSink) Collect(ctx api.StreamContext, data interface{}) error {
 	logger := ctx.GetLogger()
-	v, _, err := ctx.TransformOutput(data)
-	if err != nil {
-		logger.Error(err)
-		return err
+	var val string
+	if r.c.DataTemplate != "" { // The result is a string
+		v, _, err := ctx.TransformOutput(data)
+		if err != nil {
+			logger.Error(err)
+			return err
+		}
+		m := make(map[string]interface{})
+		err = json.Unmarshal(v, &m)
+		if err != nil {
+			return fmt.Errorf("fail to decode data %s after applying dataTemplate for error %v", string(v), err)
+		}
+		data = m
+		val = string(v)
 	}
-	if r.field != "" {
-		switch out := data.(type) {
-		case []map[string]interface{}:
-			for _, m := range out {
-				key := r.field
-				k, err := cast.ToString(m[key], cast.CONVERT_ALL)
-				if err != nil {
-					return fmt.Errorf("key must be string or convertible to string, but got %v", m[key])
-				}
+	switch d := data.(type) {
+	case []map[string]interface{}:
+		for _, el := range d {
+			err := r.save(ctx, el, val)
+			if err != nil {
+				return err
+			}
+		}
+	case map[string]interface{}:
+		err := r.save(ctx, d, val)
+		if err != nil {
+			return err
+		}
+	default:
+		return fmt.Errorf("unrecognized format of %s", data)
+	}
+	logger.Debug("insert success %v", data)
+	return nil
+}
 
-				if r.dataType == "list" {
-					err := r.cli.LPush(k, v).Err()
-					if err != nil {
-						logger.Error(err)
-						return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-					}
-					logger.Debugf("send redis list success, key:%s data: %s", k, string(v))
-				} else {
-					err := r.cli.Set(k, v, r.expiration*time.Second).Err()
-					if err != nil {
-						logger.Error(err)
-						return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-					}
-					logger.Debugf("send redis string success, key:%s data: %s", k, string(v))
-				}
+func (r *RedisSink) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Closing redis sink")
+	err := r.cli.Close()
+	return err
+}
+
+func (r *RedisSink) save(ctx api.StreamContext, data map[string]interface{}, val string) error {
+	logger := ctx.GetLogger()
+	if val == "" {
+		jsonBytes, err := json.Marshal(data)
+		if err != nil {
+			return err
+		}
+		val = string(jsonBytes)
+	}
+	key := r.c.Key
+	var err error
+	if r.c.Field != "" {
+		keyval, ok := data[r.c.Field]
+		if !ok {
+			return fmt.Errorf("field %s does not exist in data %v", r.c.Field, data)
+		}
+		key, err = cast.ToString(keyval, cast.CONVERT_ALL)
+		if err != nil {
+			return fmt.Errorf("key must be string or convertible to string, but got %v", keyval)
+		}
+	}
+	rowkind := ast.RowkindUpsert
+	if r.c.RowkindField != "" {
+		c, ok := data[r.c.RowkindField]
+		if ok {
+			rowkind, ok = c.(string)
+			if !ok {
+				return fmt.Errorf("rowkind field %s is not a string in data %v", r.c.RowkindField, data)
+			}
+			if rowkind != ast.RowkindInsert && rowkind != ast.RowkindUpdate && rowkind != ast.RowkindDelete && rowkind != ast.RowkindUpsert {
+				return fmt.Errorf("invalid rowkind %s", rowkind)
 			}
-		case map[string]interface{}:
-			key := r.field
-			k, err := cast.ToString(out[key], cast.CONVERT_ALL)
+		}
+	}
+	switch rowkind {
+	case ast.RowkindInsert, ast.RowkindUpdate, ast.RowkindUpsert:
+		if r.c.DataType == "list" {
+			err = r.cli.LPush(key, val).Err()
 			if err != nil {
-				return fmt.Errorf("key must be string or convertible to string, but got %v", out[key])
+				return fmt.Errorf("lpush %s:%s error, %v", key, val, err)
 			}
-
-			if r.dataType == "list" {
-				err := r.cli.LPush(k, v).Err()
-				if err != nil {
-					logger.Error(err)
-					return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-				}
-				logger.Debugf("send redis list success, key:%s data: %s", k, string(v))
-			} else {
-				err := r.cli.Set(k, v, r.expiration*time.Second).Err()
-				if err != nil {
-					logger.Error(err)
-					return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
-				}
-				logger.Debugf("send redis string success, key:%s data: %s", k, string(v))
+			logger.Debugf("push redis list success, key:%s data: %v", key, val)
+		} else {
+			err = r.cli.Set(key, val, r.c.Expiration*time.Second).Err()
+			if err != nil {
+				return fmt.Errorf("set %s:%s error, %v", key, val, err)
 			}
+			logger.Debugf("set redis string success, key:%s data: %s", key, val)
 		}
-	} else if r.key != "" {
-		if r.dataType == "list" {
-			err := r.cli.LPush(r.key, v).Err()
+	case ast.RowkindDelete:
+		if r.c.DataType == "list" {
+			err = r.cli.LPop(key).Err()
 			if err != nil {
-				logger.Error(err)
-				return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+				return fmt.Errorf("lpop %s error, %v", key, err)
 			}
-			logger.Debugf("send redis list success, key:%s data: %s", r.key, string(v))
+			logger.Debugf("pop redis list success, key:%s data: %v", key, val)
 		} else {
-			err := r.cli.Set(r.key, v, r.expiration*time.Second).Err()
+			err = r.cli.Del(key).Err()
 			if err != nil {
 				logger.Error(err)
-				return fmt.Errorf("%s:%s", errorx.IOErr, err.Error())
+				return err
 			}
-			logger.Debugf("send redis string success, key:%s data: %s", r.key, string(v))
+			logger.Debugf("delete redis string success, key:%s data: %s", key, val)
 		}
+	default:
+		// never happen
+		logger.Errorf("unexpected rowkind %s", rowkind)
 	}
-	logger.Debug("insert success %v", data)
 	return nil
 }
 
-func (r *RedisSink) Close(ctx api.StreamContext) error {
-	err := r.cli.Close()
-	return err
-}
-
 func GetSink() api.Sink {
 	return &RedisSink{}
 }


Test patch:
diff --git a/extensions/sinks/sql/sql_test.go b/extensions/sinks/sql/sql_test.go
new file mode 100644
index 00000000..adc8a59a
--- /dev/null
+++ b/extensions/sinks/sql/sql_test.go
@@ -0,0 +1,310 @@
+// 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"
+	"database/sql/driver"
+	"fmt"
+	econf "github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"os"
+	"reflect"
+	"testing"
+)
+
+func TestSingle(t *testing.T) {
+	db, err := sql.Open("sqlite3", "file:test.db")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	s := &sqlSink{}
+	defer func() {
+		db.Close()
+		s.Close(ctx)
+		err := os.Remove("test.db")
+		if err != nil {
+			fmt.Println(err)
+		}
+	}()
+	_, err = db.Exec("CREATE TABLE IF NOT EXISTS single (id BIGINT PRIMARY KEY, name TEXT NOT NULL, address varchar(20), mobile varchar(20))")
+	if err != nil {
+		panic(err)
+	}
+	err = s.Configure(map[string]interface{}{
+		"url":   "sqlite://test.db",
+		"table": "single",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	err = s.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var data = []map[string]interface{}{
+		{"id": 1, "name": "John", "address": "343", "mobile": "334433"},
+		{"id": 2, "name": "Susan", "address": "34", "mobile": "334433"},
+		{"id": 3, "name": "Susan", "address": "34", "mobile": "334433"},
+	}
+	for _, d := range data {
+		err = s.Collect(ctx, d)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+	}
+	s.Close(ctx)
+	rows, err := db.Query("SELECT * FROM single")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	act, _ := rowsToMap(rows)
+	exp := []map[string]interface{}{
+		{"id": int64(1), "name": "John", "address": "343", "mobile": "334433"},
+		{"id": int64(2), "name": "Susan", "address": "34", "mobile": "334433"},
+		{"id": int64(3), "name": "Susan", "address": "34", "mobile": "334433"},
+	}
+	if !reflect.DeepEqual(act, exp) {
+		t.Errorf("Expect %v but got %v", exp, act)
+	}
+}
+
+func TestBatch(t *testing.T) {
+	db, err := sql.Open("sqlite3", "file:test.db")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	s := &sqlSink{}
+	defer func() {
+		db.Close()
+		s.Close(ctx)
+		err := os.Remove("test.db")
+		if err != nil {
+			fmt.Println(err)
+		}
+	}()
+	_, err = db.Exec("CREATE TABLE IF NOT EXISTS batch (id BIGINT PRIMARY KEY, name TEXT NOT NULL)")
+	if err != nil {
+		panic(err)
+	}
+	err = s.Configure(map[string]interface{}{
+		"url":    "sqlite://test.db",
+		"table":  "batch",
+		"fields": []string{"id", "name"},
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	err = s.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var data = []map[string]interface{}{
+		{"id": 1, "name": "John", "address": "343", "mobile": "334433"},
+		{"id": 2, "name": "Susan", "address": "34", "mobile": "334433"},
+		{"id": 3, "name": "Susan", "address": "34", "mobile": "334433"},
+	}
+	err = s.Collect(ctx, data)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	s.Close(ctx)
+	rows, err := db.Query("SELECT * FROM batch")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	act, _ := rowsToMap(rows)
+	exp := []map[string]interface{}{
+		{"id": int64(1), "name": "John"},
+		{"id": int64(2), "name": "Susan"},
+		{"id": int64(3), "name": "Susan"},
+	}
+	if !reflect.DeepEqual(act, exp) {
+		t.Errorf("Expect %v but got %v", exp, act)
+	}
+}
+
+func TestUpdate(t *testing.T) {
+	db, err := sql.Open("sqlite3", "file:test.db")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	s := &sqlSink{}
+	defer func() {
+		db.Close()
+		s.Close(ctx)
+		err := os.Remove("test.db")
+		if err != nil {
+			fmt.Println(err)
+		}
+	}()
+	_, err = db.Exec("CREATE TABLE IF NOT EXISTS updateTable (id BIGINT PRIMARY KEY, name TEXT NOT NULL)")
+	if err != nil {
+		panic(err)
+	}
+	err = s.Configure(map[string]interface{}{
+		"url":          "sqlite://test.db",
+		"table":        "updateTable",
+		"rowkindField": "action",
+		"keyField":     "id",
+		"fields":       []string{"id", "name"},
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = s.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var test = []struct {
+		d []map[string]interface{}
+		b bool
+		r []map[string]interface{}
+	}{
+		{
+			d: []map[string]interface{}{
+				{"id": 1, "name": "John", "address": "343", "mobile": "334433"},
+				{"action": "insert", "id": 2, "name": "Susan", "address": "34", "mobile": "334433"},
+				{"action": "update", "id": 2, "name": "Diana"},
+			},
+			b: true,
+			r: []map[string]interface{}{
+				{"id": int64(1), "name": "John"},
+				{"id": int64(2), "name": "Diana"},
+			},
+		}, {
+			d: []map[string]interface{}{
+				{"id": 4, "name": "Charles", "address": "343", "mobile": "334433"},
+				{"action": "delete", "id": 2},
+				{"action": "update", "id": 1, "name": "Lizz"},
+			},
+			b: false,
+			r: []map[string]interface{}{
+				{"id": int64(1), "name": "Lizz"},
+				{"id": int64(4), "name": "Charles"},
+			},
+		}, {
+			d: []map[string]interface{}{
+				{"action": "upsert", "id": 4, "name": "Charles", "address": "343", "mobile": "334433"},
+				{"action": "update", "id": 3, "name": "Lizz"},
+				{"action": "update", "id": 1, "name": "Philips"},
+			},
+			b: true,
+			r: []map[string]interface{}{
+				{"id": int64(1), "name": "Philips"},
+				{"id": int64(4), "name": "Charles"},
+			},
+		},
+	}
+	for i, tt := range test {
+		if tt.b {
+			err = s.Collect(ctx, tt.d)
+			if err != nil {
+				fmt.Println(err)
+			}
+		} else {
+			for _, d := range tt.d {
+				err = s.Collect(ctx, d)
+				if err != nil {
+					fmt.Println(err)
+				}
+			}
+		}
+		rows, err := db.Query("SELECT * FROM updateTable")
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		act, _ := rowsToMap(rows)
+		if !reflect.DeepEqual(act, tt.r) {
+			t.Errorf("Case %d Expect %v but got %v", i, tt.r, act)
+		}
+	}
+}
+
+func rowsToMap(rows *sql.Rows) ([]map[string]interface{}, error) {
+	cols, _ := rows.Columns()
+
+	types, err := rows.ColumnTypes()
+	if err != nil {
+		return nil, err
+	}
+	var result []map[string]interface{}
+	for rows.Next() {
+		data := make(map[string]interface{})
+		columns := make([]interface{}, len(cols))
+		prepareValues(columns, types, cols)
+
+		err := rows.Scan(columns...)
+		if err != nil {
+			return nil, err
+		}
+		scanIntoMap(data, columns, cols)
+		result = append(result, data)
+	}
+	return result, nil
+}
+
+func scanIntoMap(mapValue map[string]interface{}, values []interface{}, columns []string) {
+	for idx, column := range columns {
+		if reflectValue := reflect.Indirect(reflect.Indirect(reflect.ValueOf(values[idx]))); reflectValue.IsValid() {
+			mapValue[column] = reflectValue.Interface()
+			if valuer, ok := mapValue[column].(driver.Valuer); ok {
+				mapValue[column], _ = valuer.Value()
+			} else if b, ok := mapValue[column].(sql.RawBytes); ok {
+				mapValue[column] = string(b)
+			}
+		} else {
+			mapValue[column] = nil
+		}
+	}
+}
+
+func prepareValues(values []interface{}, columnTypes []*sql.ColumnType, columns []string) {
+	if len(columnTypes) > 0 {
+		for idx, columnType := range columnTypes {
+			if columnType.ScanType() != nil {
+				values[idx] = reflect.New(reflect.PtrTo(columnType.ScanType())).Interface()
+			} else {
+				values[idx] = new(interface{})
+			}
+		}
+	} else {
+		for idx := range columns {
+			values[idx] = new(interface{})
+		}
+	}
+}
diff --git a/internal/topo/memory/store/db_test.go b/internal/topo/memory/store/db_test.go
index f5a4d4cb..0dd826d6 100644
--- a/internal/topo/memory/store/db_test.go
+++ b/internal/topo/memory/store/db_test.go
@@ -50,10 +50,19 @@ func TestTable(t *testing.T) {
 		api.NewDefaultSourceTuple(map[string]interface{}{"a": 2, "b": "0"}, nil),
 		api.NewDefaultSourceTuple(map[string]interface{}{"a": 5, "b": "0"}, nil),
 	}
-	if !reflect.DeepEqual(v, exp) {
+	if len(v) != 2 {
 		t.Errorf("read 1 again expect %v, but got %v", exp, v)
 		return
+	} else {
+		if v[0].Message()["a"] != 2 {
+			v[0], v[1] = v[1], v[0]
+		}
+		if !reflect.DeepEqual(v, exp) {
+			t.Errorf("read 1 again expect %v, but got %v", exp, v)
+			return
+		}
 	}
+
 	v, _ = tb.Read([]string{"a", "b"}, []interface{}{1, "1"})
 	exp = []api.SourceTuple{
 		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
diff --git a/internal/topo/redis/lookup_test.go b/internal/topo/redis/lookup_test.go
index c95b270d..6c638326 100644
--- a/internal/topo/redis/lookup_test.go
+++ b/internal/topo/redis/lookup_test.go
@@ -25,7 +25,10 @@ import (
 	"testing"
 )
 
-var addr string
+var (
+	addr string
+	mr   *miniredis.Miniredis
+)
 
 func init() {
 	s, err := miniredis.Run()
@@ -41,6 +44,7 @@ func init() {
 	s.Lpush("group1", `{"id":2,"name":"Susan"}`)
 	s.Lpush("group2", `{"id":3,"name":"Nancy"}`)
 	s.Lpush("group3", `{"id":4,"name":"Tom"}`)
+	mr = s
 }
 
 // TestSingle test lookup value of a single map
diff --git a/internal/topo/redis/sink_test.go b/internal/topo/redis/sink_test.go
new file mode 100644
index 00000000..4b886053
--- /dev/null
+++ b/internal/topo/redis/sink_test.go
@@ -0,0 +1,274 @@
+// 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 redis
+
+import (
+	econf "github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/pkg/cast"
+	"reflect"
+	"testing"
+)
+
+func TestSink(t *testing.T) {
+	s := &RedisSink{}
+	err := s.Configure(map[string]interface{}{
+		"addr": addr,
+		"key":  "test",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	err = s.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var tests = []struct {
+		c map[string]interface{}
+		d interface{}
+		k string
+		v interface{}
+	}{
+		{
+			c: map[string]interface{}{"key": "1"},
+			d: map[string]interface{}{"id": 1, "name": "John", "address": 34, "mobile": "334433"},
+			k: "1",
+			v: `{"address":34,"id":1,"mobile":"334433","name":"John"}`,
+		},
+		{
+			c: map[string]interface{}{"field": "id"},
+			d: map[string]interface{}{"id": 2, "name": "Susan", "address": 34, "mobile": "334433"},
+			k: "2",
+			v: `{"address":34,"id":2,"mobile":"334433","name":"Susan"}`,
+		},
+		{
+			c: map[string]interface{}{"field": "name", "datatype": "list"},
+			d: map[string]interface{}{"id": 3, "name": "Susan"},
+			k: "Susan",
+			v: `{"id":3,"name":"Susan"}`,
+		},
+		{
+			c: map[string]interface{}{"field": "id", "datatype": "list"},
+			d: []map[string]interface{}{
+				{"id": 4, "name": "Susan"},
+				{"id": 4, "name": "Bob"},
+				{"id": 4, "name": "John"},
+			},
+			k: "4",
+			v: `{"id":4,"name":"John"}`,
+		},
+		{
+			c: map[string]interface{}{"field": "id", "datatype": "string"},
+			d: []map[string]interface{}{
+				{"id": 25, "name": "Susan"},
+				{"id": 25, "name": "Bob"},
+				{"id": 25, "name": "John"},
+			},
+			k: "25",
+			v: `{"id":25,"name":"John"}`,
+		},
+	}
+	for i, tt := range tests {
+		cast.MapToStruct(tt.c, s.c)
+		err = s.Collect(ctx, tt.d)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		var (
+			r   string
+			err error
+		)
+		switch tt.c["datatype"] {
+		case "list":
+			r, err = mr.Lpop(tt.k)
+		default:
+			r, err = mr.Get(tt.k)
+		}
+		if err != nil {
+			t.Errorf("case %d err %v", i, err)
+			return
+		}
+		if !reflect.DeepEqual(r, tt.v) {
+			t.Errorf("case %d expect %v, but got %v", i, tt.v, r)
+		}
+	}
+}
+
+func TestUpdateString(t *testing.T) {
+	s := &RedisSink{}
+	err := s.Configure(map[string]interface{}{
+		"addr":         addr,
+		"field":        "id",
+		"rowkindField": "action",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	err = s.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var tests = []struct {
+		d interface{}
+		k string
+		v interface{}
+	}{
+		{
+			d: map[string]interface{}{ // add without action
+				"id": "testUpdate1", "name": "Susan",
+			},
+			k: "testUpdate1",
+			v: `{"id":"testUpdate1","name":"Susan"}`,
+		},
+		{
+			d: map[string]interface{}{ // update with action
+				"action": "update", "id": "testUpdate1", "name": "John",
+			},
+			k: "testUpdate1",
+			v: `{"action":"update","id":"testUpdate1","name":"John"}`,
+		},
+		{
+			d: map[string]interface{}{ // delete
+				"action": "delete", "id": "testUpdate1",
+			},
+			k: "testUpdate1",
+			v: ``,
+		},
+		{
+			d: []map[string]interface{}{ // multiple actions
+				{"action": "delete", "id": "testUpdate1"},
+				{"action": "insert", "id": "testUpdate1", "name": "Susan"},
+			},
+			k: "testUpdate1",
+			v: `{"action":"insert","id":"testUpdate1","name":"Susan"}`,
+		},
+	}
+	for i, tt := range tests {
+		err = s.Collect(ctx, tt.d)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		r, err := mr.Get(tt.k)
+		if tt.v == "" {
+			if err == nil || err.Error() != "ERR no such key" {
+				t.Errorf("case %d err %v", i, err)
+				return
+			}
+		} else {
+			if err != nil {
+				t.Errorf("case %d err %v", i, err)
+				return
+			}
+			if !reflect.DeepEqual(r, tt.v) {
+				t.Errorf("case %d expect %v, but got %v", i, tt.v, r)
+			}
+		}
+	}
+}
+
+func TestUpdateList(t *testing.T) {
+	s := &RedisSink{}
+	err := s.Configure(map[string]interface{}{
+		"addr":         addr,
+		"field":        "id",
+		"datatype":     "list",
+		"rowkindField": "action",
+	})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	contextLogger := econf.Log.WithField("rule", "test")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	err = s.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var tests = []struct {
+		d interface{}
+		k string
+		v []string
+	}{
+		{
+			d: map[string]interface{}{ // add without action
+				"id": "testUpdateList", "name": "Susan",
+			},
+			k: "testUpdateList",
+			v: []string{`{"id":"testUpdateList","name":"Susan"}`},
+		},
+		{
+			d: map[string]interface{}{ // update with action
+				"action": "update", "id": "testUpdateList", "name": "John",
+			},
+			k: "testUpdateList",
+			v: []string{`{"action":"update","id":"testUpdateList","name":"John"}`, `{"id":"testUpdateList","name":"Susan"}`},
+		},
+		{
+			d: map[string]interface{}{ // delete
+				"action": "delete", "id": "testUpdateList",
+			},
+			k: "testUpdateList",
+			v: []string{`{"id":"testUpdateList","name":"Susan"}`},
+		},
+		{
+			d: []map[string]interface{}{ // multiple actions
+				{"action": "delete", "id": "testUpdateList"},
+				{"action": "insert", "id": "testUpdateList", "name": "Susan"},
+			},
+			k: "testUpdateList",
+			v: []string{`{"action":"insert","id":"testUpdateList","name":"Susan"}`},
+		},
+		{
+			d: map[string]interface{}{ // delete
+				"action": "delete", "id": "testUpdateList",
+			},
+			k: "testUpdateList",
+			v: nil,
+		},
+	}
+	for i, tt := range tests {
+		err = s.Collect(ctx, tt.d)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		r, err := mr.List(tt.k)
+		if tt.v == nil {
+			if err == nil || err.Error() != "ERR no such key" {
+				t.Errorf("case %d err %v", i, err)
+				return
+			}
+		} else {
+			if err != nil {
+				t.Errorf("case %d err %v", i, err)
+				return
+			}
+			if !reflect.DeepEqual(r, tt.v) {
+				t.Errorf("case %d expect %v, but got %v", i, tt.v, r)
+			}
+		}
+	}
+}