Repository: lf-edge/ekuiper
Pull request #1423 (2022-09-29 09:10:33)

Merge 1.6.2 bugfixes


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/etc/sources/memory.yaml b/etc/sources/memory.yaml
index 3a248136..e69de29b 100644
--- a/etc/sources/memory.yaml
+++ b/etc/sources/memory.yaml
@@ -1,2 +0,0 @@
-default:
-  index: ["id"]
\ No newline at end of file
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/binder/function/binder.go b/internal/binder/function/binder.go
index 27b5cdee..8ef545ea 100644
--- a/internal/binder/function/binder.go
+++ b/internal/binder/function/binder.go
@@ -17,6 +17,7 @@ package function
 import (
 	"github.com/lf-edge/ekuiper/internal/binder"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 )
 
@@ -25,15 +26,6 @@ var ( // init once and read only
 	funcFactoriesNames []string
 )
 
-type FuncType int
-
-const (
-	FuncTypeUnknown FuncType = iota - 1
-	FuncTypeScalar
-	FuncTypeAgg
-	FuncTypeCols
-)
-
 func init() {
 	f := binder.FactoryEntry{
 		Name:    "built-in",
@@ -92,14 +84,14 @@ func ConvName(name string) (string, bool) {
 }
 
 type multiAggFunc interface {
-	GetFuncType(name string) FuncType
+	GetFuncType(name string) ast.FuncType
 }
 
 func IsAggFunc(funcName string) bool {
 	f, _ := Function(funcName)
 	if f != nil {
 		if mf, ok := f.(multiAggFunc); ok {
-			return mf.GetFuncType(funcName) == FuncTypeAgg
+			return mf.GetFuncType(funcName) == ast.FuncTypeAgg
 		} else {
 			return f.IsAggregate()
 		}
@@ -107,16 +99,16 @@ func IsAggFunc(funcName string) bool {
 	return false
 }
 
-func GetFuncType(funcName string) FuncType {
+func GetFuncType(funcName string) ast.FuncType {
 	f, _ := Function(funcName)
 	if f != nil {
 		if mf, ok := f.(multiAggFunc); ok {
 			return mf.GetFuncType(funcName)
 		}
 		if f.IsAggregate() {
-			return FuncTypeAgg
+			return ast.FuncTypeAgg
 		}
-		return FuncTypeScalar
+		return ast.FuncTypeScalar
 	}
-	return FuncTypeUnknown
+	return ast.FuncTypeUnknown
 }
diff --git a/internal/binder/function/funcs_agg.go b/internal/binder/function/funcs_agg.go
index 9ee1eab7..38172a06 100644
--- a/internal/binder/function/funcs_agg.go
+++ b/internal/binder/function/funcs_agg.go
@@ -22,7 +22,7 @@ import (
 
 func registerAggFunc() {
 	builtins["avg"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			c := getCount(arg0)
@@ -52,7 +52,7 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["count"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			return getCount(arg0), true
@@ -60,7 +60,7 @@ func registerAggFunc() {
 		val: ValidateOneArg,
 	}
 	builtins["max"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			if len(arg0) > 0 {
@@ -101,7 +101,7 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["min"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			if len(arg0) > 0 {
@@ -142,7 +142,7 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sum"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
 			if len(arg0) > 0 {
@@ -171,14 +171,14 @@ func registerAggFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["collect"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return args[0], true
 		},
 		val: ValidateOneArg,
 	}
 	builtins["deduplicate"] = builtinFunc{
-		fType: FuncTypeAgg,
+		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].([]interface{})
 			v2, ok2 := args[1].([]interface{})
diff --git a/internal/binder/function/funcs_cols.go b/internal/binder/function/funcs_cols.go
index 055d9edc..4826c821 100644
--- a/internal/binder/function/funcs_cols.go
+++ b/internal/binder/function/funcs_cols.go
@@ -42,7 +42,7 @@ func wrapColFunc(colFunc ColFunc) funcExe {
 
 func registerColsFunc() {
 	builtins["changed_cols"] = builtinFunc{
-		fType: FuncTypeCols,
+		fType: ast.FuncTypeCols,
 		exec:  wrapColFunc(changedFunc),
 		val: func(_ api.FunctionContext, args []ast.Expr) error {
 			if len(args) <= 2 {
diff --git a/internal/binder/function/funcs_math.go b/internal/binder/function/funcs_math.go
index 0fe831c9..15693ab5 100644
--- a/internal/binder/function/funcs_math.go
+++ b/internal/binder/function/funcs_math.go
@@ -24,7 +24,7 @@ import (
 
 func registerMathFunc() {
 	builtins["abs"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[0].(int); ok {
 				t := float64(v)
@@ -39,7 +39,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["acos"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Acos(v), true
@@ -50,7 +50,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["asin"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Asin(v), true
@@ -61,7 +61,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["atan"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Atan(v), true
@@ -72,7 +72,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["atan2"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v1, e := toF64(args[0]); e == nil {
 				if v2, e1 := toF64(args[1]); e1 == nil {
@@ -87,7 +87,7 @@ func registerMathFunc() {
 		val: ValidateTwoNumberArg,
 	}
 	builtins["bitand"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			v2, ok2 := args[1].(int)
@@ -100,7 +100,7 @@ func registerMathFunc() {
 		val: ValidateTwoIntArg,
 	}
 	builtins["bitor"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			v2, ok2 := args[1].(int)
@@ -113,7 +113,7 @@ func registerMathFunc() {
 		val: ValidateTwoIntArg,
 	}
 	builtins["bitxor"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			v2, ok2 := args[1].(int)
@@ -126,7 +126,7 @@ func registerMathFunc() {
 		val: ValidateTwoIntArg,
 	}
 	builtins["bitnot"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			v1, ok1 := args[0].(int)
 			if ok1 {
@@ -146,7 +146,7 @@ func registerMathFunc() {
 		},
 	}
 	builtins["ceil"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Ceil(v), true
@@ -157,7 +157,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["cos"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Cos(v), true
@@ -168,7 +168,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["cosh"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Cosh(v), true
@@ -179,7 +179,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["exp"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Exp(v), true
@@ -190,7 +190,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["ln"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Log2(v), true
@@ -201,7 +201,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["log"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Log10(v), true
@@ -212,7 +212,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["mod"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				if v1, e1 := toF64(args[1]); e == nil {
@@ -227,7 +227,7 @@ func registerMathFunc() {
 		val: ValidateTwoNumberArg,
 	}
 	builtins["power"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v1, e := toF64(args[0]); e == nil {
 				if v2, e2 := toF64(args[1]); e2 == nil {
@@ -242,14 +242,14 @@ func registerMathFunc() {
 		val: ValidateTwoNumberArg,
 	}
 	builtins["rand"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return rand.Float64(), true
 		},
 		val: ValidateOneArg,
 	}
 	builtins["round"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Round(v), true
@@ -260,7 +260,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sign"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				if v > 0 {
@@ -277,7 +277,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sin"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Sin(v), true
@@ -288,7 +288,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sinh"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Sinh(v), true
@@ -299,7 +299,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["sqrt"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Sqrt(v), true
@@ -310,7 +310,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["tan"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Tan(v), true
@@ -321,7 +321,7 @@ func registerMathFunc() {
 		val: ValidateOneNumberArg,
 	}
 	builtins["tanh"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, e := toF64(args[0]); e == nil {
 				return math.Tanh(v), true
diff --git a/internal/binder/function/funcs_misc.go b/internal/binder/function/funcs_misc.go
index f9011f76..74fcf178 100644
--- a/internal/binder/function/funcs_misc.go
+++ b/internal/binder/function/funcs_misc.go
@@ -35,7 +35,7 @@ import (
 
 func registerMiscFunc() {
 	builtins["cast"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[1].(string); ok {
 				switch v {
@@ -150,7 +150,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["chr"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[0].(int); ok {
 				return rune(v), true
@@ -178,7 +178,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["encode"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[1].(string); ok {
 				if strings.EqualFold(v, "base64") {
@@ -215,7 +215,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["trunc"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			var v0 float64
 			if v1, ok := args[0].(int); ok {
@@ -247,7 +247,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["md5"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -263,7 +263,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha1"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -279,7 +279,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha256"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -295,7 +295,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha384"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -311,7 +311,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["sha512"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -327,7 +327,7 @@ func registerMiscFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["isnull"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return true, true
@@ -344,7 +344,7 @@ func registerMiscFunc() {
 		val: ValidateOneArg,
 	}
 	builtins["newuuid"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if newUUID, err := uuid.NewUUID(); err != nil {
 				return err, false
@@ -355,14 +355,14 @@ func registerMiscFunc() {
 		val: ValidateNoArg,
 	}
 	builtins["tstamp"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return conf.GetNowInMilli(), true
 		},
 		val: ValidateNoArg,
 	}
 	builtins["mqtt"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if v, ok := args[0].(string); ok {
 				return v, true
@@ -386,7 +386,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["meta"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			return args[0], true
 		},
@@ -412,7 +412,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["cardinality"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			val := reflect.ValueOf(args[0])
 			if val.Kind() == reflect.Slice {
@@ -423,7 +423,7 @@ func registerMiscFunc() {
 		val: ValidateOneArg,
 	}
 	builtins["json_path_query"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result, err := jsonCall(ctx, args)
 			if err != nil {
@@ -434,7 +434,7 @@ func registerMiscFunc() {
 		val: ValidateJsonFunc,
 	}
 	builtins["json_path_query_first"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result, err := jsonCall(ctx, args)
 			if err != nil {
@@ -449,7 +449,7 @@ func registerMiscFunc() {
 		val: ValidateJsonFunc,
 	}
 	builtins["json_path_exists"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result, err := jsonCall(ctx, args)
 			if err != nil {
@@ -470,17 +470,17 @@ func registerMiscFunc() {
 		val: ValidateJsonFunc,
 	}
 	builtins["window_start"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec:  nil, // directly return in the valuer
 		val:   ValidateNoArg,
 	}
 	builtins["window_end"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec:  nil, // directly return in the valuer
 		val:   ValidateNoArg,
 	}
 	builtins["changed_col"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			ignoreNull, ok := args[0].(bool)
 			if !ok {
@@ -513,7 +513,7 @@ func registerMiscFunc() {
 		},
 	}
 	builtins["had_changed"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if len(args) <= 1 {
 				return fmt.Errorf("expect more than one arg but got %d", len(args)), false
@@ -555,7 +555,7 @@ func registerMiscFunc() {
 	}
 
 	builtins["lag"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if len(args) != 1 && len(args) != 2 && len(args) != 3 {
 				return fmt.Errorf("expect one two or three args but got %d", len(args)), false
@@ -625,7 +625,7 @@ func registerMiscFunc() {
 	}
 
 	builtins["object_construct"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			result := make(map[string]interface{})
 			for i := 0; i < len(args); i += 2 {
diff --git a/internal/binder/function/funcs_str.go b/internal/binder/function/funcs_str.go
index b5d83764..9b74dcb6 100644
--- a/internal/binder/function/funcs_str.go
+++ b/internal/binder/function/funcs_str.go
@@ -28,7 +28,7 @@ import (
 
 func registerStrFunc() {
 	builtins["concat"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			var b bytes.Buffer
 			for _, arg := range args {
@@ -49,7 +49,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["endswith"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return false, true
@@ -60,7 +60,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["indexof"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return -1, true
@@ -71,7 +71,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["length"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := cast.ToStringAlways(args[0])
 			return utf8.RuneCountInString(arg0), true
@@ -79,7 +79,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["lower"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -90,7 +90,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["lpad"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -105,7 +105,7 @@ func registerStrFunc() {
 		val: ValidateOneStrOneInt,
 	}
 	builtins["ltrim"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -116,7 +116,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["numbytes"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := cast.ToStringAlways(args[0])
 			return len(arg0), true
@@ -124,7 +124,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["format_time"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -155,7 +155,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["regexp_matches"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return false, true
@@ -170,7 +170,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["regexp_replace"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil || args[2] == nil {
 				return nil, true
@@ -195,7 +195,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["regexp_substr"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return nil, true
@@ -210,7 +210,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["rpad"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -225,7 +225,7 @@ func registerStrFunc() {
 		val: ValidateOneStrOneInt,
 	}
 	builtins["rtrim"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -236,7 +236,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["substring"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -306,7 +306,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["startswith"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return false, true
@@ -317,7 +317,7 @@ func registerStrFunc() {
 		val: ValidateTwoStrArg,
 	}
 	builtins["split_value"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil || args[1] == nil {
 				return nil, true
@@ -354,7 +354,7 @@ func registerStrFunc() {
 		},
 	}
 	builtins["trim"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
@@ -365,7 +365,7 @@ func registerStrFunc() {
 		val: ValidateOneStrArg,
 	}
 	builtins["upper"] = builtinFunc{
-		fType: FuncTypeScalar,
+		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			if args[0] == nil {
 				return nil, true
diff --git a/internal/binder/function/function.go b/internal/binder/function/function.go
index 71a66474..7499eea2 100644
--- a/internal/binder/function/function.go
+++ b/internal/binder/function/function.go
@@ -25,7 +25,7 @@ type funcExe func(ctx api.FunctionContext, args []interface{}) (interface{}, boo
 type funcVal func(ctx api.FunctionContext, args []ast.Expr) error
 
 type builtinFunc struct {
-	fType FuncType
+	fType ast.FuncType
 	exec  funcExe
 	val   funcVal
 }
@@ -87,10 +87,10 @@ func (f *funcExecutor) IsAggregate() bool {
 	return false
 }
 
-func (f *funcExecutor) GetFuncType(name string) FuncType {
+func (f *funcExecutor) GetFuncType(name string) ast.FuncType {
 	fs, ok := builtins[name]
 	if !ok {
-		return FuncTypeUnknown
+		return ast.FuncTypeUnknown
 	}
 	return fs.fType
 }
diff --git a/internal/plugin/native/manager.go b/internal/plugin/native/manager.go
index 5bc723a2..76049662 100644
--- a/internal/plugin/native/manager.go
+++ b/internal/plugin/native/manager.go
@@ -607,8 +607,8 @@ func (rr *Manager) HasFunctionSet(name string) bool {
 }
 
 func (rr *Manager) ConvName(name string) (string, bool) {
-	_, err := rr.Function(name)
-	if err == nil {
+	_, ok := rr.GetPluginBySymbol(plugin2.FUNCTION, name)
+	if ok {
 		return name, true
 	}
 	return name, false
diff --git a/internal/plugin/portable/manager.go b/internal/plugin/portable/manager.go
index 88aaa1ba..c695a047 100644
--- a/internal/plugin/portable/manager.go
+++ b/internal/plugin/portable/manager.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.
@@ -345,6 +345,11 @@ func (m *Manager) Delete(name string) error {
 	if !ok {
 		return fmt.Errorf("portable plugin %s is not found", name)
 	}
+	pm := runtime.GetPluginInsManager()
+	err := pm.Kill(name)
+	if err != nil {
+		conf.Log.Errorf("fail to kill portable plugin %s process, please try to kill it manually", name)
+	}
 	// unregister the plugin
 	m.reg.Delete(name)
 	// delete files and uninstall metas
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager.go b/internal/plugin/portable/runtime/plugin_ins_manager.go
index 191da6cb..3c210cf3 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager.go
@@ -178,16 +178,22 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 		return nil, fmt.Errorf("invalid conf: %v", pconf)
 	}
 	var cmd *exec.Cmd
-	switch pluginMeta.Language {
-	case "go":
-		conf.Log.Printf("starting go plugin executable %s", pluginMeta.Executable)
-		cmd = exec.Command(pluginMeta.Executable, string(jsonArg))
+	err = infra.SafeRun(func() error {
+		switch pluginMeta.Language {
+		case "go":
+			conf.Log.Printf("starting go plugin executable %s", pluginMeta.Executable)
+			cmd = exec.Command(pluginMeta.Executable, string(jsonArg))
 
-	case "python":
-		conf.Log.Printf("starting python plugin executable %s with script %s\n", conf.Config.Portable.PythonBin, pluginMeta.Executable)
-		cmd = exec.Command(conf.Config.Portable.PythonBin, pluginMeta.Executable, string(jsonArg))
-	default:
-		return nil, fmt.Errorf("unsupported language: %s", pluginMeta.Language)
+		case "python":
+			conf.Log.Printf("starting python plugin executable %s with script %s\n", conf.Config.Portable.PythonBin, pluginMeta.Executable)
+			cmd = exec.Command(conf.Config.Portable.PythonBin, pluginMeta.Executable, string(jsonArg))
+		default:
+			return fmt.Errorf("unsupported language: %s", pluginMeta.Language)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("fail to start plugin %s: %v", pluginMeta.Name, err)
 	}
 	cmd.Stdout = conf.Log.Out
 	cmd.Stderr = conf.Log.Out
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/memory/lookupsource.go b/internal/topo/memory/lookupsource.go
index c5bf4f05..02521a6e 100644
--- a/internal/topo/memory/lookupsource.go
+++ b/internal/topo/memory/lookupsource.go
@@ -15,9 +15,9 @@
 package memory
 
 import (
+	"fmt"
 	"github.com/lf-edge/ekuiper/internal/topo/memory/store"
 	"github.com/lf-edge/ekuiper/pkg/api"
-	"github.com/lf-edge/ekuiper/pkg/cast"
 	"regexp"
 	"strings"
 )
@@ -27,14 +27,14 @@ import (
 type lookupsource struct {
 	topic      string
 	topicRegex *regexp.Regexp
-	keys       []string
 	table      *store.Table
+	key        string
 }
 
 func (s *lookupsource) Open(ctx api.StreamContext) error {
-	ctx.GetLogger().Infof("lookup source %s is opened with keys %v", s.topic, s.keys)
+	ctx.GetLogger().Infof("lookup source %s is opened with key %v", s.topic, s.key)
 	var err error
-	s.table, err = store.Reg(s.topic, s.topicRegex, s.keys)
+	s.table, err = store.Reg(s.topic, s.topicRegex, s.key)
 	return err
 }
 
@@ -47,11 +47,14 @@ func (s *lookupsource) Configure(datasource string, props map[string]interface{}
 		}
 		s.topicRegex = r
 	}
-	if c, ok := props["index"]; ok {
-		if bl, err := cast.ToStringSlice(c, cast.CONVERT_SAMEKIND); err != nil {
-			s.keys = bl
+	if k, ok := props["key"]; ok {
+		if kk, ok := k.(string); ok {
+			s.key = kk
 		}
 	}
+	if s.key == "" {
+		return fmt.Errorf("key is required for lookup source")
+	}
 	return nil
 }
 
@@ -62,5 +65,5 @@ func (s *lookupsource) Lookup(ctx api.StreamContext, _ []string, keys []string,
 
 func (s *lookupsource) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("lookup source %s is closing", s.topic)
-	return store.Unreg(s.topic)
+	return store.Unreg(s.topic, s.key)
 }
diff --git a/internal/topo/memory/pubsub/manager.go b/internal/topo/memory/pubsub/manager.go
index 8bc2232b..7467726f 100644
--- a/internal/topo/memory/pubsub/manager.go
+++ b/internal/topo/memory/pubsub/manager.go
@@ -110,6 +110,18 @@ func RemovePub(topic string) {
 }
 
 func Produce(ctx api.StreamContext, topic string, data map[string]interface{}) {
+	doProduce(ctx, topic, api.NewDefaultSourceTuple(data, map[string]interface{}{"topic": topic}))
+}
+
+func ProduceUpdatable(ctx api.StreamContext, topic string, data map[string]interface{}, rowkind string, keyval interface{}) {
+	doProduce(ctx, topic, &UpdatableTuple{
+		DefaultSourceTuple: api.NewDefaultSourceTuple(data, map[string]interface{}{"topic": topic}),
+		Rowkind:            rowkind,
+		Keyval:             keyval,
+	})
+}
+
+func doProduce(ctx api.StreamContext, topic string, data api.SourceTuple) {
 	c, exists := pubTopics[topic]
 	if !exists {
 		return
@@ -120,7 +132,7 @@ func Produce(ctx api.StreamContext, topic string, data map[string]interface{}) {
 	// broadcast to all consumers
 	for name, out := range c.consumers {
 		select {
-		case out <- api.NewDefaultSourceTuple(data, map[string]interface{}{"topic": topic}):
+		case out <- data:
 			logger.Debugf("memory source broadcast from topic %s to %s done", topic, name)
 		case <-ctx.Done():
 			// rule stop so stop waiting
@@ -128,7 +140,6 @@ func Produce(ctx api.StreamContext, topic string, data map[string]interface{}) {
 			logger.Errorf("memory source topic %s drop message to %s", topic, name)
 		}
 	}
-
 }
 
 func ProduceError(ctx api.StreamContext, topic string, err error) {
diff --git a/internal/topo/memory/pubsub/tuple.go b/internal/topo/memory/pubsub/tuple.go
new file mode 100644
index 00000000..9384d8de
--- /dev/null
+++ b/internal/topo/memory/pubsub/tuple.go
@@ -0,0 +1,23 @@
+// 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 pubsub
+
+import "github.com/lf-edge/ekuiper/pkg/api"
+
+type UpdatableTuple struct {
+	*api.DefaultSourceTuple
+	Rowkind string
+	Keyval  interface{}
+}
diff --git a/internal/topo/memory/sink.go b/internal/topo/memory/sink.go
index da3356d6..c97ec421 100644
--- a/internal/topo/memory/sink.go
+++ b/internal/topo/memory/sink.go
@@ -19,12 +19,23 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/ast"
+	"github.com/lf-edge/ekuiper/pkg/cast"
 	"strings"
 )
 
+type config struct {
+	Topic        string `json:"topic"`
+	DataTemplate string `json:"dataTemplate"`
+	RowkindField string `json:"rowkindField"`
+	KeyField     string `json:"keyField"`
+}
+
 type sink struct {
 	topic        string
 	hasTransform bool
+	keyField     string
+	rowkindField string
 }
 
 func (s *sink) Open(ctx api.StreamContext) error {
@@ -34,19 +45,23 @@ func (s *sink) Open(ctx api.StreamContext) error {
 }
 
 func (s *sink) Configure(props map[string]interface{}) error {
-	if t, ok := props[pubsub.IdProperty]; ok {
-		if id, casted := t.(string); casted {
-			if strings.ContainsAny(id, "#+") {
-				return fmt.Errorf("invalid memory topic %s: wildcard found", id)
-			}
-			s.topic = id
-		} else {
-			return fmt.Errorf("can't cast value %s to string", t)
-		}
+	cfg := &config{}
+	err := cast.MapToStruct(props, cfg)
+	if err != nil {
+		return err
+	}
+	if strings.ContainsAny(cfg.Topic, "#+") {
+		return fmt.Errorf("invalid memory topic %s: wildcard found", cfg.Topic)
 	}
-	if _, ok := props["dataTemplate"]; ok {
+	s.topic = cfg.Topic
+	if cfg.DataTemplate != "" {
 		s.hasTransform = true
 	}
+	s.rowkindField = cfg.RowkindField
+	s.keyField = cfg.KeyField
+	if s.rowkindField != "" && s.keyField == "" {
+		return fmt.Errorf("keyField is required when rowkindField is set")
+	}
 	return nil
 }
 
@@ -68,14 +83,19 @@ func (s *sink) Collect(ctx api.StreamContext, data interface{}) error {
 		}
 		data = m
 	}
-
 	switch d := data.(type) {
 	case []map[string]interface{}:
 		for _, el := range d {
-			pubsub.Produce(ctx, topic, el)
+			err := s.publish(ctx, topic, el)
+			if err != nil {
+				return fmt.Errorf("fail to publish data %v for error %v", d, err)
+			}
 		}
 	case map[string]interface{}:
-		pubsub.Produce(ctx, topic, d)
+		err := s.publish(ctx, topic, d)
+		if err != nil {
+			return fmt.Errorf("fail to publish data %v for error %v", d, err)
+		}
 	default:
 		return fmt.Errorf("unrecognized format of %s", data)
 	}
@@ -87,3 +107,29 @@ func (s *sink) Close(ctx api.StreamContext) error {
 	pubsub.RemovePub(s.topic)
 	return nil
 }
+
+func (s *sink) publish(ctx api.StreamContext, topic string, el map[string]interface{}) error {
+	if s.rowkindField != "" {
+		c, ok := el[s.rowkindField]
+		var rowkind string
+		if !ok {
+			rowkind = ast.RowkindUpsert
+		} else {
+			rowkind, ok = c.(string)
+			if !ok {
+				return fmt.Errorf("rowkind field %s is not a string in data %v", s.rowkindField, el)
+			}
+			if rowkind != ast.RowkindInsert && rowkind != ast.RowkindUpdate && rowkind != ast.RowkindDelete && rowkind != ast.RowkindUpsert {
+				return fmt.Errorf("invalid rowkind %s", rowkind)
+			}
+		}
+		key, ok := el[s.keyField]
+		if !ok {
+			return fmt.Errorf("key field %s not found in data %v", s.keyField, el)
+		}
+		pubsub.ProduceUpdatable(ctx, topic, el, rowkind, key)
+	} else {
+		pubsub.Produce(ctx, topic, el)
+	}
+	return nil
+}
diff --git a/internal/topo/memory/store/db.go b/internal/topo/memory/store/db.go
index 821c8719..03d570c9 100644
--- a/internal/topo/memory/store/db.go
+++ b/internal/topo/memory/store/db.go
@@ -17,6 +17,7 @@ package store
 import (
 	"context"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"sync"
 )
@@ -39,21 +40,22 @@ func (tc *tableCount) Decrease() int {
 	defer tc.Unlock()
 	tc.count--
 	if tc.count < 0 {
-		fmt.Errorf("Table count is less than 0: %d", tc.count)
+		conf.Log.Errorf("Table count is less than 0: %d", tc.count)
 	}
 	return tc.count
 }
 
 type database struct {
 	sync.RWMutex
-	tables map[string]*tableCount // topic: table
+	tables map[string]*tableCount // topic_key: table
 }
 
 // getTable return the table of the topic.
-func (db *database) getTable(topic string) (*Table, bool) {
+func (db *database) getTable(topic string, key string) (*Table, bool) {
 	db.RLock()
 	defer db.RUnlock()
-	tc, ok := db.tables[topic]
+	tableId := fmt.Sprintf("%s_%s", topic, key)
+	tc, ok := db.tables[tableId]
 	if ok {
 		return tc.t, true
 	} else {
@@ -65,19 +67,20 @@ func (db *database) getTable(topic string) (*Table, bool) {
 // If the table already exists, return the existing table;
 // otherwise, create a new table and return it.
 // The second argument is to indicate if the table is newly created
-func (db *database) addTable(topic string, keys []string) (*Table, bool) {
+func (db *database) addTable(topic string, key string) (*Table, bool) {
 	db.Lock()
 	defer db.Unlock()
-	tc, ok := db.tables[topic]
+	tableId := fmt.Sprintf("%s_%s", topic, key)
+	tc, ok := db.tables[tableId]
 	if ok {
 		tc.Increase()
 	} else {
-		t := createTable(keys)
+		t := createTable(topic, key)
 		tc = &tableCount{
 			count: 1,
 			t:     t,
 		}
-		db.tables[topic] = tc
+		db.tables[tableId] = tc
 	}
 	return tc.t, !ok
 }
@@ -85,79 +88,81 @@ func (db *database) addTable(topic string, keys []string) (*Table, bool) {
 // dropTable drop the table of the topic/values
 // stops to accumulate job
 // deletes the cache data
-func (db *database) dropTable(topic string) error {
+func (db *database) dropTable(topic string, key string) error {
+	tableId := fmt.Sprintf("%s_%s", topic, key)
 	db.Lock()
 	defer db.Unlock()
-	if tc, ok := db.tables[topic]; ok {
+	if tc, ok := db.tables[tableId]; ok {
 		if tc.Decrease() == 0 {
 			if tc.t != nil && tc.t.cancel != nil {
 				tc.t.cancel()
 			}
-			delete(db.tables, topic)
+			delete(db.tables, tableId)
 		}
 		return nil
 	}
-	return fmt.Errorf("Table %s not found", topic)
+	return fmt.Errorf("Table %s not found", tableId)
 }
 
 // Table has one writer and multiple reader
 type Table struct {
 	sync.RWMutex
-	// datamap is the overall data
-	datamap  []api.SourceTuple
-	hasIndex bool
-	// indexes is the indexed data
-	indexes map[string]map[interface{}][]api.SourceTuple
+	topic string
+	key   string
+	// datamap is the overall data indexed by primary key
+	datamap map[interface{}]api.SourceTuple
 	cancel  context.CancelFunc
 }
 
-func createTable(keys []string) *Table {
-	t := &Table{}
-	if len(keys) > 0 {
-		t.indexes = make(map[string]map[interface{}][]api.SourceTuple, len(keys))
-		for _, k := range keys {
-			t.indexes[k] = make(map[interface{}][]api.SourceTuple)
-		}
-		t.hasIndex = true
-	}
+func createTable(topic string, key string) *Table {
+	t := &Table{topic: topic, key: key, datamap: make(map[interface{}]api.SourceTuple)}
 	return t
 }
 
 func (t *Table) add(value api.SourceTuple) {
 	t.Lock()
 	defer t.Unlock()
-	t.datamap = append(t.datamap, value)
-	for k, v := range t.indexes {
-		if val, ok := value.Message()[k]; ok {
-			if _, ok := v[val]; !ok {
-				v[val] = make([]api.SourceTuple, 0)
-			}
-			v[val] = append(v[val], value)
-		}
+	keyval, ok := value.Message()[t.key]
+	if !ok {
+		conf.Log.Errorf("add to table %s omitted, value not found for key %s", t.topic, t.key)
 	}
+	t.datamap[keyval] = value
+}
+
+func (t *Table) delete(key interface{}) {
+	t.Lock()
+	defer t.Unlock()
+	delete(t.datamap, key)
 }
 
 func (t *Table) Read(keys []string, values []interface{}) ([]api.SourceTuple, error) {
 	t.RLock()
 	defer t.RUnlock()
-	data := t.datamap
-	excludeKey := -1
-	if t.hasIndex {
-		// Find the first indexed key
+	// Find the primary key
+	var matched api.SourceTuple
+	for i, k := range keys {
+		if k == t.key {
+			matched = t.datamap[values[i]]
+		}
+	}
+	if matched != nil {
+		match := true
 		for i, k := range keys {
-			if d, ok := t.indexes[k]; ok {
-				data = d[values[i]]
-				excludeKey = i
+			if val, ok := matched.Message()[k]; !ok || val != values[i] {
+				match = false
+				break
 			}
 		}
+		if match {
+			return []api.SourceTuple{matched}, nil
+		} else {
+			return nil, nil
+		}
 	}
 	var result []api.SourceTuple
-	for _, v := range data {
+	for _, v := range t.datamap {
 		match := true
 		for i, k := range keys {
-			if i == excludeKey {
-				continue
-			}
 			if val, ok := v.Message()[k]; !ok || val != values[i] {
 				match = false
 				break
diff --git a/internal/topo/memory/store/store.go b/internal/topo/memory/store/store.go
index aee41187..6317db0b 100644
--- a/internal/topo/memory/store/store.go
+++ b/internal/topo/memory/store/store.go
@@ -19,13 +19,14 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/pkg/ast"
 	"regexp"
 )
 
 // Reg registers a topic to save it to memory store
 // Create a new go routine to listen to the topic and save the data to memory
-func Reg(topic string, topicRegex *regexp.Regexp, keys []string) (*Table, error) {
-	t, isNew := db.addTable(topic, keys)
+func Reg(topic string, topicRegex *regexp.Regexp, key string) (*Table, error) {
+	t, isNew := db.addTable(topic, key)
 	if isNew {
 		go runTable(topic, topicRegex, t)
 	}
@@ -46,7 +47,17 @@ func runTable(topic string, topicRegex *regexp.Regexp, t *Table) {
 			if !opened { // exit go routine is not sync with drop table
 				return
 			}
-			t.add(v)
+			switch vv := v.(type) {
+			case *pubsub.UpdatableTuple:
+				switch vv.Rowkind {
+				case ast.RowkindInsert, ast.RowkindUpdate, ast.RowkindUpsert:
+					t.add(vv.DefaultSourceTuple)
+				case ast.RowkindDelete:
+					t.delete(vv.Keyval)
+				}
+			default:
+				t.add(v)
+			}
 			conf.Log.Debugf("receive data %v for %s", v, topic)
 		case <-ctx.Done():
 			return
@@ -55,7 +66,7 @@ func runTable(topic string, topicRegex *regexp.Regexp, t *Table) {
 }
 
 // Unreg unregisters a topic to remove it from memory store
-func Unreg(topic string) error {
+func Unreg(topic string, key string) error {
 	// Must be an atomic operation
-	return db.dropTable(topic)
+	return db.dropTable(topic, key)
 }
diff --git a/internal/topo/node/conf/source.go b/internal/topo/node/conf/source.go
index 689288e6..7d59a9be 100644
--- a/internal/topo/node/conf/source.go
+++ b/internal/topo/node/conf/source.go
@@ -54,6 +54,7 @@ func GetSourceConf(sourceType string, options *ast.Options) map[string]interface
 		f = "json"
 	}
 	props["format"] = strings.ToLower(f)
+	props["key"] = strings.ToLower(options.KEY)
 	conf.Log.Debugf("get conf for %s with conf key %s: %v", sourceType, confkey, printable(props))
 	return props
 }
diff --git a/internal/topo/node/dynamic_channel_buffer.go b/internal/topo/node/dynamic_channel_buffer.go
index ec2a668a..dc98b45f 100644
--- a/internal/topo/node/dynamic_channel_buffer.go
+++ b/internal/topo/node/dynamic_channel_buffer.go
@@ -29,7 +29,7 @@ type DynamicChannelBuffer struct {
 
 func NewDynamicChannelBuffer() *DynamicChannelBuffer {
 	buffer := &DynamicChannelBuffer{
-		In:     make(chan api.SourceTuple, 10),
+		In:     make(chan api.SourceTuple, 1024),
 		Out:    make(chan api.SourceTuple),
 		buffer: make([]api.SourceTuple, 0),
 		limit:  102400,
@@ -58,8 +58,10 @@ func (b *DynamicChannelBuffer) run() {
 		} else if l > 0 {
 			select {
 			case b.Out <- b.buffer[0]:
+				//fmt.Println("out loud")
 				b.buffer = b.buffer[1:]
 			case value := <-b.In:
+				//fmt.Printf("in loud with length %d\n", len(b.In))
 				b.buffer = append(b.buffer, value)
 			case <-b.done:
 				return
@@ -67,6 +69,7 @@ func (b *DynamicChannelBuffer) run() {
 		} else {
 			select {
 			case value := <-b.In:
+				//fmt.Printf("in quiet with length %d \n", len(b.In))
 				b.buffer = append(b.buffer, value)
 			case <-b.done:
 				return
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{}
 }
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index f4d0babe..2d4b7ae2 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -823,7 +823,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 	p.inFunc = name
 	defer func() { p.inFunc = "" }()
 	ft := function.GetFuncType(name)
-	if ft == function.FuncTypeCols && p.clause != "select" {
+	if ft == ast.FuncTypeCols && p.clause != "select" {
 		return nil, fmt.Errorf("function %s can only be used inside the select clause", n)
 	}
 	var args []ast.Expr
@@ -832,7 +832,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 			if valErr := validateFuncs(name, nil); valErr != nil {
 				return nil, valErr
 			}
-			c := &ast.Call{Name: name, Args: args, FuncId: p.fn}
+			c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
 			p.fn += 1
 			return c, nil
 		} else {
@@ -842,7 +842,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 		if exp, err := p.ParseExpr(); err != nil {
 			return nil, err
 		} else {
-			if ft == function.FuncTypeCols {
+			if ft == ast.FuncTypeCols {
 				field := &ast.ColFuncField{Expr: exp, Name: nameExpr(exp)}
 				args = append(args, field)
 			} else {
@@ -867,7 +867,7 @@ func (p *Parser) parseCall(n string) (ast.Expr, error) {
 		if name == "deduplicate" {
 			args = append([]ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}}, args...)
 		}
-		c := &ast.Call{Name: name, Args: args, FuncId: p.fn}
+		c := &ast.Call{Name: name, Args: args, FuncId: p.fn, FuncType: ft}
 		p.fn += 1
 		return c, nil
 	} else {
@@ -1461,6 +1461,9 @@ func (p *Parser) parseStreamOptions() (*ast.Options, error) {
 	} else {
 		return nil, fmt.Errorf("found %q, expect stream options.", lit)
 	}
+	if opts.KIND == ast.StreamKindLookup && opts.TYPE == "memory" && opts.KEY == "" {
+		return nil, fmt.Errorf("Option \"key\" is required for memory lookup table.")
+	}
 	return opts, nil
 }
 
diff --git a/internal/xsql/valuer.go b/internal/xsql/valuer.go
index 8ec00626..0eaf90a7 100644
--- a/internal/xsql/valuer.go
+++ b/internal/xsql/valuer.go
@@ -281,11 +281,11 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 			if valuer, ok := v.Valuer.(CallValuer); ok {
 				var (
 					args []interface{}
-					ft   = function.GetFuncType(expr.Name)
+					ft   = expr.FuncType
 				)
 				if len(expr.Args) > 0 {
 					switch ft {
-					case function.FuncTypeAgg:
+					case ast.FuncTypeAgg:
 						args = make([]interface{}, len(expr.Args))
 						for i, arg := range expr.Args {
 							if aggreValuer, ok := valuer.(AggregateCallValuer); ok {
@@ -297,7 +297,7 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 								}
 							}
 						}
-					case function.FuncTypeScalar:
+					case ast.FuncTypeScalar:
 						args = make([]interface{}, len(expr.Args))
 						for i, arg := range expr.Args {
 							args[i] = v.Eval(arg)
@@ -305,7 +305,7 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 								return args[i]
 							}
 						}
-					case function.FuncTypeCols:
+					case ast.FuncTypeCols:
 						var keys []string
 						for _, arg := range expr.Args { // In the parser, the col func arguments must be ColField
 							cf, ok := arg.(*ast.ColFuncField)
diff --git a/pkg/ast/expr.go b/pkg/ast/expr.go
index 3ca9d981..689b13db 100644
--- a/pkg/ast/expr.go
+++ b/pkg/ast/expr.go
@@ -122,10 +122,20 @@ func (sl *StringLiteral) expr()    {}
 func (sl *StringLiteral) literal() {}
 func (sl *StringLiteral) node()    {}
 
+type FuncType int
+
+const (
+	FuncTypeUnknown FuncType = iota - 1
+	FuncTypeScalar
+	FuncTypeAgg
+	FuncTypeCols
+)
+
 type Call struct {
-	Name   string
-	FuncId int
-	Args   []Expr
+	Name     string
+	FuncId   int
+	FuncType FuncType
+	Args     []Expr
 }
 
 func (c *Call) expr()    {}
diff --git a/pkg/ast/statement.go b/pkg/ast/statement.go
index 3d4c0ed1..394f140f 100644
--- a/pkg/ast/statement.go
+++ b/pkg/ast/statement.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.
@@ -189,3 +189,10 @@ type SortField struct {
 type SortFields []SortField
 
 func (d SortFields) node() {}
+
+const (
+	RowkindInsert = "insert"
+	RowkindUpdate = "update"
+	RowkindUpsert = "upsert"
+	RowkindDelete = "delete"
+)


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/processor/stream_test.go b/internal/processor/stream_test.go
index 059e9e68..08584b3f 100644
--- a/internal/processor/stream_test.go
+++ b/internal/processor/stream_test.go
@@ -205,8 +205,8 @@ func TestTableProcessor(t *testing.T) {
 func TestTableList(t *testing.T) {
 	p := NewStreamProcessor()
 	p.ExecStmt(`CREATE TABLE tt1 () WITH (DATASOURCE="users", FORMAT="JSON", KIND="scan")`)
-	p.ExecStmt(`CREATE TABLE tt2 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KIND="lookup")`)
-	p.ExecStmt(`CREATE TABLE tt3 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KIND="lookup")`)
+	p.ExecStmt(`CREATE TABLE tt2 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`)
+	p.ExecStmt(`CREATE TABLE tt3 () WITH (DATASOURCE="users", TYPE="memory", FORMAT="JSON", KEY="id", KIND="lookup")`)
 	p.ExecStmt(`CREATE TABLE tt4 () WITH (DATASOURCE="users", FORMAT="JSON")`)
 	defer func() {
 		p.ExecStmt(`DROP TABLE tt1`)
diff --git a/internal/topo/lookup/table_test.go b/internal/topo/lookup/table_test.go
index 7422ef15..99af264f 100644
--- a/internal/topo/lookup/table_test.go
+++ b/internal/topo/lookup/table_test.go
@@ -27,6 +27,7 @@ func TestTable(t *testing.T) {
 		DATASOURCE: "test",
 		TYPE:       "memory",
 		KIND:       "lookup",
+		KEY:        "id",
 	})
 	if err != nil {
 		t.Error(err)
@@ -36,6 +37,7 @@ func TestTable(t *testing.T) {
 		DATASOURCE: "test2",
 		TYPE:       "memory",
 		KIND:       "lookup",
+		KEY:        "id",
 	})
 	if err != nil {
 		t.Error(err)
diff --git a/internal/topo/memory/lookupsource_test.go b/internal/topo/memory/lookupsource_test.go
index 74a38b98..e28a9afd 100644
--- a/internal/topo/memory/lookupsource_test.go
+++ b/internal/topo/memory/lookupsource_test.go
@@ -25,11 +25,11 @@ import (
 	"time"
 )
 
-func TestNoIndexLookup(t *testing.T) {
+func TestUpdateLookup(t *testing.T) {
 	contextLogger := conf.Log.WithField("rule", "test")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	ls := GetLookupSource()
-	err := ls.Configure("test", map[string]interface{}{"option": "value"})
+	err := ls.Configure("test", map[string]interface{}{"key": "ff"})
 	if err != nil {
 		t.Error(err)
 		return
@@ -42,8 +42,12 @@ func TestNoIndexLookup(t *testing.T) {
 	// wait for the source to be ready
 	time.Sleep(100 * time.Millisecond)
 	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value2"})
-	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"})
+	pubsub.ProduceUpdatable(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value2"}, "delete", "value1")
+	pubsub.ProduceUpdatable(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"}, "insert", "value2")
 	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value4"})
+	pubsub.ProduceUpdatable(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"}, "delete", "value2")
+	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value1", "gg": "value2"})
+	pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value2", "gg": "value2"})
 	// wait for table accumulation
 	time.Sleep(100 * time.Millisecond)
 	canctx, cancel := gocontext.WithCancel(gocontext.Background())
@@ -60,7 +64,6 @@ func TestNoIndexLookup(t *testing.T) {
 	}()
 	expected := []api.SourceTuple{
 		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value2"}, map[string]interface{}{"topic": "test"}),
-		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value4"}, map[string]interface{}{"topic": "test"}),
 	}
 	result, err := ls.Lookup(ctx, []string{}, []string{"ff"}, []interface{}{"value1"})
 	if !reflect.DeepEqual(result, expected) {
@@ -73,11 +76,11 @@ func TestNoIndexLookup(t *testing.T) {
 	}
 }
 
-func TestSingleIndexLookup(t *testing.T) {
+func TestLookup(t *testing.T) {
 	contextLogger := conf.Log.WithField("rule", "test2")
 	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
 	ls := GetLookupSource()
-	err := ls.Configure("test2", map[string]interface{}{"index": []string{"ff"}})
+	err := ls.Configure("test2", map[string]interface{}{"key": "gg"})
 	if err != nil {
 		t.Error(err)
 		return
@@ -90,7 +93,7 @@ func TestSingleIndexLookup(t *testing.T) {
 	// wait for the source to be ready
 	time.Sleep(100 * time.Millisecond)
 	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value1", "gg": "value2"})
-	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value2", "gg": "value2"})
+	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value2", "gg": "value3"})
 	pubsub.Produce(ctx, "test2", map[string]interface{}{"ff": "value1", "gg": "value4"})
 	// wait for table accumulation
 	time.Sleep(100 * time.Millisecond)
@@ -102,7 +105,7 @@ func TestSingleIndexLookup(t *testing.T) {
 			case <-canctx.Done():
 				return
 			case <-time.After(10 * time.Millisecond):
-				pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value4", "gg": "value2"})
+				pubsub.Produce(ctx, "test", map[string]interface{}{"ff": "value4", "gg": "value5"})
 			}
 		}
 	}()
@@ -111,6 +114,13 @@ func TestSingleIndexLookup(t *testing.T) {
 		api.NewDefaultSourceTuple(map[string]interface{}{"ff": "value1", "gg": "value4"}, map[string]interface{}{"topic": "test2"}),
 	}
 	result, err := ls.Lookup(ctx, []string{}, []string{"ff"}, []interface{}{"value1"})
+	if len(result) != 2 {
+		t.Errorf("expect %v but got %v", expected, result)
+	} else {
+		if result[0].Message()["gg"] != "value2" {
+			result[0], result[1] = result[1], result[0]
+		}
+	}
 	if !reflect.DeepEqual(result, expected) {
 		t.Errorf("expect %v but got %v", expected, result)
 	}
diff --git a/internal/topo/memory/sink_test.go b/internal/topo/memory/sink_test.go
new file mode 100644
index 00000000..bbabef50
--- /dev/null
+++ b/internal/topo/memory/sink_test.go
@@ -0,0 +1,84 @@
+// 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 memory
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/topo/context"
+	"github.com/lf-edge/ekuiper/internal/topo/memory/pubsub"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"reflect"
+	"testing"
+)
+
+func TestUpdate(t *testing.T) {
+	contextLogger := conf.Log.WithField("rule", "test2")
+	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+	ms := GetSink()
+	err := ms.Configure(map[string]interface{}{"topic": "testupdate", "rowkindField": "verb", "keyField": "id"})
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	err = ms.Open(ctx)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	var data = []map[string]interface{}{
+		{"id": "1", "verb": "insert", "name": "test1"},
+		{"id": "2", "verb": "insert", "name": "test2"},
+		{"id": "1", "verb": "update", "name": "test1"},
+		{"id": "2", "verb": "delete", "name": "test2"},
+	}
+	c := pubsub.CreateSub("testupdate", nil, "testSource", 100)
+	go func() {
+		for _, d := range data {
+			ms.Collect(ctx, d)
+		}
+	}()
+	var actual []api.SourceTuple
+	for i := 0; i < 4; i++ {
+		d := <-c
+		fmt.Println(d)
+		actual = append(actual, d)
+	}
+	expects := []api.SourceTuple{
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "1", "verb": "insert", "name": "test1"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "insert",
+			Keyval:             "1",
+		},
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "2", "verb": "insert", "name": "test2"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "insert",
+			Keyval:             "2",
+		},
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "1", "verb": "update", "name": "test1"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "update",
+			Keyval:             "1",
+		},
+		&pubsub.UpdatableTuple{
+			DefaultSourceTuple: api.NewDefaultSourceTuple(map[string]interface{}{"id": "2", "verb": "delete", "name": "test2"}, map[string]interface{}{"topic": "testupdate"}),
+			Rowkind:            "delete",
+			Keyval:             "2",
+		},
+	}
+	if !reflect.DeepEqual(actual, expects) {
+		t.Errorf("expect %v but got %v", expects, actual)
+	}
+}
diff --git a/internal/topo/memory/store/db_test.go b/internal/topo/memory/store/db_test.go
index bd00e4c3..0dd826d6 100644
--- a/internal/topo/memory/store/db_test.go
+++ b/internal/topo/memory/store/db_test.go
@@ -21,94 +21,111 @@ import (
 )
 
 func TestTable(t *testing.T) {
-	tb := createTable([]string{"a"})
+	tb := createTable("topicT", "a")
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "0"}, nil))
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 2, "b": "0"}, nil))
-	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 3, "b": "0"}, nil))
+	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 3, "b": "4"}, nil))
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil))
 	v, _ := tb.Read([]string{"a"}, []interface{}{1})
 	exp := []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "0"}, nil),
 		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
 	}
 	if !reflect.DeepEqual(v, exp) {
-		t.Errorf("read 1 expect %v, but got %v", exp, v)
+		t.Errorf("read a 1 expect %v, but got %v", exp, v)
 		return
 	}
-	v, _ = tb.Read([]string{"a"}, []interface{}{3})
+	v, _ = tb.Read([]string{"b"}, []interface{}{"0"})
 	exp = []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 3, "b": "0"}, nil),
+		api.NewDefaultSourceTuple(map[string]interface{}{"a": 2, "b": "0"}, nil),
 	}
 	if !reflect.DeepEqual(v, exp) {
-		t.Errorf("read 3 expect %v, but got %v", exp, v)
+		t.Errorf("read b 0 expect %v, but got %v", exp, v)
 		return
 	}
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 5, "b": "0"}, nil))
+	tb.delete(3)
 	tb.add(api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil))
-	v, _ = tb.Read([]string{"a"}, []interface{}{1})
+	v, _ = tb.Read([]string{"b"}, []interface{}{"0"})
 	exp = []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "0"}, nil),
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
+		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),
-		api.NewDefaultSourceTuple(map[string]interface{}{"a": 1, "b": "1"}, nil),
 	}
 	if !reflect.DeepEqual(v, exp) {
 		t.Errorf("read a,b expect %v, but got %v", exp, v)
 		return
 	}
+	v, _ = tb.Read([]string{"a"}, []interface{}{3})
+	if v != nil {
+		t.Errorf("read a 3 expect nil, but got %v", v)
+		return
+	}
+	tb.delete(1)
+	v, _ = tb.Read([]string{"a"}, []interface{}{1})
+	if v != nil {
+		t.Errorf("read a 1 expect nil, but got %v", v)
+	}
 }
 
 func TestDb(t *testing.T) {
 	db = &database{
 		tables: make(map[string]*tableCount),
 	}
-	db.addTable("t1", []string{"a"})
-	db.addTable("t1", []string{"a", "b"})
-	db.addTable("t2", []string{"a"})
-	db.addTable("t1", []string{"a"})
-	_, ok := db.getTable("t1")
+	db.addTable("t1", "a")
+	db.addTable("t1", "b")
+	db.addTable("t2", "a")
+	db.addTable("t1", "a")
+	_, ok := db.getTable("t1", "a")
 	if !ok {
 		t.Errorf("table t1 a should exist")
 		return
 	}
-	_, ok = db.getTable("t1")
+	_, ok = db.getTable("t1", "b")
 	if !ok {
 		t.Errorf("table t1 b should exist")
 		return
 	}
-	_, ok = db.getTable("t3")
+	_, ok = db.getTable("t3", "a")
 	if ok {
 		t.Errorf("table t1 c should not exist")
 		return
 	}
-	tc := db.tables["t1"]
-	if tc.count != 3 {
+	tc := db.tables["t1_a"]
+	if tc.count != 2 {
 		t.Errorf("table t1 a should have 2 instances but got %d", tc.count)
 		return
 	}
-	tc = db.tables["t2"]
+	tc = db.tables["t2_a"]
 	if tc.count != 1 {
-		t.Errorf("table t1 a should have 1 instances")
+		t.Errorf("table t2 a should have 1 instances")
 		return
 	}
-	db.dropTable("t1")
-	db.dropTable("t2")
-	_, ok = db.getTable("t2")
+	db.dropTable("t1", "a")
+	db.dropTable("t2", "a")
+	_, ok = db.getTable("t2", "a")
 	if ok {
-		t.Errorf("table ta a should not exist")
+		t.Errorf("table t2 a should not exist")
 		return
 	}
-	tc = db.tables["t1"]
-	if tc.count != 2 {
-		t.Errorf("table t1 a should have 2 instances but got %d", tc.count)
+	tc = db.tables["t1_a"]
+	if tc.count != 1 {
+		t.Errorf("table t1 a should have 1 instances but got %d", tc.count)
 		return
 	}
 }
diff --git a/internal/topo/memory/store/store_test.go b/internal/topo/memory/store/store_test.go
index a63a6ec9..647966c3 100644
--- a/internal/topo/memory/store/store_test.go
+++ b/internal/topo/memory/store/store_test.go
@@ -23,18 +23,18 @@ func TestReg(t *testing.T) {
 	db = &database{
 		tables: make(map[string]*tableCount),
 	}
-	reg1, err := Reg("test", nil, []string{"a"})
+	reg1, err := Reg("test", nil, "a")
 	if err != nil {
 		t.Errorf("register test error: %v", err)
 		return
 	}
-	_, err2 := Reg("test", nil, []string{"a", "b"})
+	_, err2 := Reg("test", nil, "a")
 	if err2 != nil {
 		t.Errorf("register test error: %v", err2)
 		return
 	}
 	exp := map[string]*tableCount{
-		"test": {
+		"test_a": {
 			count: 2,
 			t:     reg1,
 		},
@@ -43,9 +43,9 @@ func TestReg(t *testing.T) {
 		t.Errorf("register expect %v, but got %v", exp, db.tables)
 		return
 	}
-	Unreg("test")
+	Unreg("test", "a")
 	exp = map[string]*tableCount{
-		"test": {
+		"test_a": {
 			count: 1,
 			t:     reg1,
 		},
diff --git a/internal/topo/node/dynamic_channel_buffer_test.go b/internal/topo/node/dynamic_channel_buffer_test.go
new file mode 100644
index 00000000..3de4ee7f
--- /dev/null
+++ b/internal/topo/node/dynamic_channel_buffer_test.go
@@ -0,0 +1,49 @@
+// 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 node
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/api"
+	"testing"
+	"time"
+)
+
+func TestBuffer(t *testing.T) {
+	b := NewDynamicChannelBuffer()
+	b.SetLimit(100)
+	stopSign := make(chan struct{})
+	go func(done chan struct{}) {
+		for i := 0; i < 100; i++ {
+			select {
+			case b.In <- api.NewDefaultSourceTuple(map[string]interface{}{"a": 5}, nil):
+				fmt.Printf("feed in %d\n", i)
+			default:
+				t.Errorf("message %d dropped, should not drop message", i)
+			}
+		}
+		close(done)
+	}(stopSign)
+	for i := 0; i < 50; i++ {
+		_ = <-b.Out
+		fmt.Printf("eaten %d\n", i)
+		time.Sleep(10 * time.Millisecond)
+	}
+	<-stopSign
+	l := b.GetLength()
+	if l != 50 {
+		t.Errorf("Expect buffer length 50, but got %d", l)
+	}
+}
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index f50afabf..7b5f813f 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.go
@@ -31,6 +31,7 @@ func TestGetConf_Apply(t *testing.T) {
 		"incremental":        false,
 		"body":               "{}",
 		"bodyType":           "json",
+		"key":                "",
 		"format":             "json",
 		"insecureSkipVerify": true,
 		"headers": map[string]interface{}{
@@ -56,6 +57,7 @@ func TestGetConfAndConvert_Apply(t *testing.T) {
 		"incremental":        true,
 		"body":               "{}",
 		"bodyType":           "json",
+		"key":                "",
 		"format":             "json",
 		"insecureSkipVerify": true,
 		"headers": map[string]interface{}{
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index cb557895..e2de868f 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -406,7 +406,7 @@ func Test_createLogicalPlan(t *testing.T) {
 							condition: &ast.BinaryExpr{
 								LHS: &ast.Call{Name: "count", FuncId: 0, Args: []ast.Expr{&ast.Wildcard{
 									Token: ast.ASTERISK,
-								}}},
+								}}, FuncType: ast.FuncTypeAgg},
 								OP:  ast.GT,
 								RHS: &ast.IntegerLiteral{Val: 2},
 							},
@@ -1430,7 +1430,7 @@ func Test_createLogicalPlanSchemaless(t *testing.T) {
 							condition: &ast.BinaryExpr{
 								LHS: &ast.Call{Name: "count", FuncId: 0, Args: []ast.Expr{&ast.Wildcard{
 									Token: ast.ASTERISK,
-								}}},
+								}}, FuncType: ast.FuncTypeAgg},
 								OP:  ast.GT,
 								RHS: &ast.IntegerLiteral{Val: 2},
 							},
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)
+			}
+		}
+	}
+}
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index 5c03bf92..6dd5bcd7 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -431,8 +431,9 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: "",
 						Name:  "count",
 						Expr: &ast.Call{
-							Name: "count",
-							Args: []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}},
+							Name:     "count",
+							Args:     []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}},
+							FuncType: ast.FuncTypeAgg,
 						},
 					},
 				},
@@ -483,8 +484,9 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: "",
 						Name:  "deduplicate",
 						Expr: &ast.Call{
-							Name: "deduplicate",
-							Args: []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}, &ast.FieldRef{Name: "temperature", StreamName: ast.DefaultStream}, &ast.BooleanLiteral{Val: false}},
+							Name:     "deduplicate",
+							Args:     []ast.Expr{&ast.Wildcard{Token: ast.ASTERISK}, &ast.FieldRef{Name: "temperature", StreamName: ast.DefaultStream}, &ast.BooleanLiteral{Val: false}},
+							FuncType: ast.FuncTypeAgg,
 						},
 					},
 				},
@@ -1160,7 +1162,7 @@ func TestParser_ParseStatement(t *testing.T) {
 				Sources:    []ast.Source{&ast.Table{Name: "topic/sensor1"}},
 				Condition:  &ast.BinaryExpr{LHS: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}, OP: ast.EQ, RHS: &ast.StringLiteral{Val: "dname"}},
 				Dimensions: ast.Dimensions{ast.Dimension{Expr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
-				Having:     &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
+				Having:     &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}, FuncType: ast.FuncTypeAgg}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
 			},
 		},
 
@@ -1173,7 +1175,7 @@ func TestParser_ParseStatement(t *testing.T) {
 				},
 				Sources:   []ast.Source{&ast.Table{Name: "topic/sensor1"}},
 				Condition: &ast.BinaryExpr{LHS: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}, OP: ast.EQ, RHS: &ast.StringLiteral{Val: "dname"}},
-				Having:    &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
+				Having:    &ast.BinaryExpr{LHS: &ast.Call{Name: "count", Args: []ast.Expr{&ast.FieldRef{StreamName: ast.DefaultStream, Name: "name"}}, FuncType: ast.FuncTypeAgg}, OP: ast.GT, RHS: &ast.IntegerLiteral{Val: 3}},
 			},
 		},
 
@@ -1657,6 +1659,7 @@ func TestParser_ParseStatement(t *testing.T) {
 								Args: []ast.Expr{
 									&ast.Wildcard{Token: ast.ASTERISK},
 								},
+								FuncType: ast.FuncTypeAgg,
 							},
 							RHS: &ast.IntegerLiteral{Val: 10},
 						},
@@ -1721,6 +1724,7 @@ func TestParser_ParseStatement(t *testing.T) {
 									Name:       "c",
 								}},
 							},
+							FuncType: ast.FuncTypeCols,
 						},
 					},
 				},
@@ -1756,6 +1760,7 @@ func TestParser_ParseStatement(t *testing.T) {
 									Name:       "c",
 								}},
 							},
+							FuncType: ast.FuncTypeCols,
 						},
 					},
 				},
diff --git a/internal/xsql/parser_tree_test.go b/internal/xsql/parser_tree_test.go
index 937f2e28..7567974e 100644
--- a/internal/xsql/parser_tree_test.go
+++ b/internal/xsql/parser_tree_test.go
@@ -112,7 +112,7 @@ func TestParser_ParseTree(t *testing.T) {
 					name STRING,
 					size BIGINT,
 					id BIGINT
-				) WITH (DATASOURCE="devices", KIND="LOOKUP", TYPE="sql");`,
+				) WITH (DATASOURCE="devices", KIND="LOOKUP", TYPE="sql", KEY="id");`,
 			stmt: &ast.StreamStmt{
 				Name: ast.StreamName("table1"),
 				StreamFields: []ast.StreamField{
@@ -125,10 +125,19 @@ func TestParser_ParseTree(t *testing.T) {
 					STRICT_VALIDATION: true,
 					KIND:              ast.StreamKindLookup,
 					TYPE:              "sql",
+					KEY:               "id",
 				},
 				StreamType: ast.TypeTable,
 			},
 		},
+		{
+			s: `CREATE TABLE table1 (
+					name STRING,
+					size BIGINT,
+					id BIGINT
+				) WITH (DATASOURCE="devices", KIND="LOOKUP", TYPE="memory");`,
+			err: `Option "key" is required for memory lookup table.`,
+		},
 		{
 			s:    `SHOW STREAMS`,
 			stmt: &ast.ShowStreamsStatement{},