Repository: lf-edge/ekuiper
Pull request #895 (2021-07-15 06:45:17)

sinks add redis plugin
sinks add redis plugin

ngjaying: DCO fails with invalid email. You'll need to setup the email
```
$ git config --global user.email name@example.com
```
ngjaying: @feng-crazy We would like to add this to 1.3.0 release. Are you able to sign off before next Wednesday? Thanks.
ngjaying: @feng-crazy Thank you again for contribute this plugin. I squash your commits into one and format the user/email to pass the DCO. I force push the updates to the master branch of your fork. Sorry for any inconvenience.

Patch:
diff --git a/.github/workflows/build_packages.yaml b/.github/workflows/build_packages.yaml
index cf66060b..16e55502 100644
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@@ -91,11 +91,12 @@ jobs:
             fail-fast: false
             matrix:
                 plugin:
-                - sinks/file
+                  - sinks/file
                 - sinks/image
                 - sinks/influx
                 - sinks/tdengine
                 - sinks/zmq
+                  - sinks/redis
                 - sources/random
                 - sources/zmq
                 - functions/accumulateWordCount
diff --git a/Makefile b/Makefile
index 93503469..d27b576b 100644
--- a/Makefile
+++ b/Makefile
@@ -138,6 +138,7 @@ PLUGINS := sinks/file \
 	sinks/influx \
 	sinks/zmq \
 	sinks/image \
+	sinks/redis \
 	sources/random \
 	sources/zmq \
 	functions/accumulateWordCount \
diff --git a/etc/sinks/redis.json b/etc/sinks/redis.json
new file mode 100755
index 00000000..148fa154
--- /dev/null
+++ b/etc/sinks/redis.json
@@ -0,0 +1,133 @@
+{
+	"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/emqx/kuiper/blob/master/docs/en_US/plugins/sinks/reids.md",
+			"zh_CN": "https://github.com/emqx/kuiper/blob/master/docs/zh_CN/plugins/sinks/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/v8"
+	],
+	"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": "超时时间"
+			}
+		}
+	]
+}
diff --git a/extensions/go.mod b/extensions/go.mod
index 1a9e58f4..68450269 100644
--- a/extensions/go.mod
+++ b/extensions/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646
 	github.com/pebbe/zmq4 v1.2.7
 	github.com/taosdata/driver-go v0.0.0-20210525062356-2bd1b495d5f3
+	github.com/go-redis/redis/v8 v8.4.11
 )
 
 replace github.com/lf-edge/ekuiper => ../
diff --git a/extensions/go.sum b/extensions/go.sum
index 05fb1720..a0944f23 100644
--- a/extensions/go.sum
+++ b/extensions/go.sum
@@ -20,6 +20,8 @@ github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+Ce
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash/v2 v2.1.0/go.mod h1:dgIUBU3pDso/gPgZ1osOZ0iQf77oPR28Tjxl5dIMyVM=
+github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
+github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
@@ -27,6 +29,8 @@ github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwc
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/disintegration/imaging v1.6.2/go.mod h1:44/5580QXChDfwIclfc/PCwrr44amcmDAg8hxG0Ewe4=
 github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.0/go.mod h1:pfXURRetgIto0GR0sCjDrfa71hqJ1wxmQWi/mOzWfWU=
@@ -43,6 +47,8 @@ github.com/faiface/pixel v0.8.0/go.mod h1:CEUU/s9E82Kqp01Boj1O67KnBskqiLghANqvUJ
 github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 h1:Ghm4eQYC0nEPnSJdVkTrXpu9KtoVCSo1hg7mtI7G9KU=
 github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239/go.mod h1:Gdwt2ce0yfBxPvZrHkprdPPTTS3N5rwmLE8T22KBXlw=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
+github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fxamacker/cbor/v2 v2.2.0/go.mod h1:TA1xS00nchWmaBnEIxPSE5oHLuJBAVvqrtAnWBwBCVo=
 github.com/gdexlab/go-render v1.0.1/go.mod h1:wRi5nW2qfjiGj4mPukH4UV0IknS1cHD4VgFTmJX5JzM=
 github.com/go-gl/gl v0.0.0-20190320180904-bf2b1f2f34d7/go.mod h1:482civXOzJJCPzJ4ZOX/pwvXBWSnzD4OKMdH4ClKGbk=
@@ -57,6 +63,8 @@ github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTM
 github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
 github.com/go-playground/validator/v10 v10.6.1/go.mod h1:xm76BBt941f7yWdGnI2DVPFFg1UK3YY04qifoXU3lOk=
 github.com/go-redis/redis/v7 v7.3.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
+github.com/go-redis/redis/v8 v8.4.11 h1:t2lToev01VTrqYQcv+QFbxtGgcf64K+VUMgf9Ap6A/E=
+github.com/go-redis/redis/v8 v8.4.11/go.mod h1:d5yY/TlkQyYBSBHnXUmnf1OrHbyQere5JV4dLKwvXmo=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3/go.mod h1:nPpo7qLxd6XL3hWJG/O60sR8ZKfMCiIoNap5GvD12KU=
@@ -78,6 +86,8 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
@@ -145,9 +155,18 @@ github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRW
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=
 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/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
+github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
+github.com/onsi/ginkgo v1.14.2 h1:8mVmC9kjFFmA8H4pKMUhcblgifdkOIXPvbhN1T36q1M=
+github.com/onsi/ginkgo v1.14.2/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
+github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
+github.com/onsi/gomega v1.10.4 h1:NiTx7EEvBzu9sFOD1zORteLSt3o8gnlvZZwSE9TnY9U=
+github.com/onsi/gomega v1.10.4/go.mod h1:g/HbgYopi++010VEqkFgJHKC09uJiW9UkXvMUuKHUCQ=
 github.com/pebbe/zmq4 v1.2.7 h1:6EaX83hdFSRUEhgzSW1E/SPoTS3JeYZgYkBvwdcrA9A=
 github.com/pebbe/zmq4 v1.2.7/go.mod h1:nqnPueOapVhE2wItZ0uOErngczsJdLOGkebMxaO8r48=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -181,6 +200,7 @@ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXf
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/taosdata/driver-go v0.0.0-20210525062356-2bd1b495d5f3 h1:zGuQLIFConaigelrFtyvPh0+7FJuqJFFPN+35yKJCsA=
@@ -193,6 +213,8 @@ github.com/urfave/cli v1.22.0/go.mod h1:b3D7uWrF2GilkNgYpgcg6J+JMUw7ehmNkE8sZdli
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+go.opentelemetry.io/otel v0.16.0 h1:uIWEbdeb4vpKPGITLsRVUS44L5oDbDUCZxn8lkxhmgw=
+go.opentelemetry.io/otel v0.16.0/go.mod h1:e4GKElweB8W2gWUqbghw0B8t5MCTccc9212eNHnOHwA=
 gocv.io/x/gocv v0.21.0/go.mod h1:Rar2PS6DV+T4FL+PM535EImD/h13hGVaHhnCu1xarBs=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -221,7 +243,10 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
+golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb h1:eBmm0M9fYhWpKZLjQUUKka/LtIxf46G4fxeEz5KJr9U=
+golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -236,11 +261,18 @@ golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd h1:xhmwyvizuTgC2qz7ZlMluP20uW+C3Rm0FD/WLDX8884=
+golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
@@ -253,6 +285,7 @@ golang.org/x/tools v0.0.0-20200522201501-cb1345f3a375/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20200717024301-6ddee64345a6/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -284,12 +317,14 @@ gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/ini.v1 v1.62.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
+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.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+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 h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/extensions/sinks/redis/redis.go b/extensions/sinks/redis/redis.go
new file mode 100755
index 00000000..fb88a869
--- /dev/null
+++ b/extensions/sinks/redis/redis.go
@@ -0,0 +1,222 @@
+// 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 (
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+
+	"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()
+
+	if v, ok := data.([]byte); ok {
+		if r.field != "" {
+			if !r.sendSingle {
+				var out []map[string]interface{}
+				if err := json.Unmarshal(v, &out); err != nil {
+					logger.Debug("Failed to unmarshal data with error: ", err, " data:", string(v))
+					return err
+				}
+
+				for _, m := range out {
+					key := r.field
+					field, ok := m[key].(string)
+					if ok {
+						key = field
+					}
+
+					if r.dataType == "list" {
+						err := r.cli.LPush(ctx, key, v).Err()
+						if err != nil {
+							logger.Error(err)
+							return err
+						}
+						logger.Debugf("send redis list success, key:%s data: %s", key, string(v))
+					} else {
+						err := r.cli.Set(ctx, key, v, r.expiration*time.Second).Err()
+						if err != nil {
+							logger.Error(err)
+							return err
+						}
+						logger.Debugf("send redis string success, key:%s data: %s", key, string(v))
+					}
+				}
+			} else {
+				var out map[string]interface{}
+				if err := json.Unmarshal(v, &out); err != nil {
+					logger.Debug("Failed to unmarshal data with error: ", err, " data:", string(v))
+					return err
+				}
+				key := r.field
+				field, ok := out[key].(string)
+				if ok {
+					key = field
+				}
+
+				if r.dataType == "list" {
+					err := r.cli.LPush(ctx, key, v).Err()
+					if err != nil {
+						logger.Error(err)
+						return err
+					}
+					logger.Debugf("send redis list success, key:%s data: %s", key, string(v))
+				} else {
+					err := r.cli.Set(ctx, key, v, r.expiration*time.Second).Err()
+					if err != nil {
+						logger.Error(err)
+						return err
+					}
+					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(ctx, r.key, v).Err()
+				if err != nil {
+					logger.Error(err)
+					return err
+				}
+				logger.Debugf("send redis list success, key:%s data: %s", r.key, string(v))
+			} else {
+				err := r.cli.Set(ctx, r.key, v, r.expiration*time.Second).Err()
+				if err != nil {
+					logger.Error(err)
+					return err
+				}
+				logger.Debugf("send redis string success, key:%s data: %s", r.key, string(v))
+			}
+		}
+
+		logger.Debug("insert success", string(v))
+	} else {
+		logger.Debug("insert failed data is not []byte data:", 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/internal/topo/operator/join_operator.go b/internal/topo/operator/join_operator.go
index 6301d605..a9e58d12 100644
--- a/internal/topo/operator/join_operator.go
+++ b/internal/topo/operator/join_operator.go
@@ -73,6 +73,10 @@ func (jp *JoinOp) getStreamNames(join *ast.Join) ([]string, error) {
 	ast.WalkFunc(join, func(node ast.Node) bool {
 		if f, ok := node.(*ast.FieldRef); ok {
 			for _, v := range f.RefSources() {
+				// Exclude default stream as it is a virtual stream name.
+				if v == ast.DefaultStream {
+					continue
+				}
 				if _, ok := keys[v]; !ok {
 					srcs = append(srcs, string(v))
 					keys[v] = true


Test patch:
diff --git a/internal/topo/operator/join_test.go b/internal/topo/operator/join_test.go
index f69554e3..eca039e1 100644
--- a/internal/topo/operator/join_test.go
+++ b/internal/topo/operator/join_test.go
@@ -35,7 +35,7 @@ func TestLeftJoinPlan_Apply(t *testing.T) {
 		result interface{}
 	}{
 		{ //0
-			sql: "SELECT id1 FROM src1 left join src2 on src1.id1 = src2.id2",
+			sql: "SELECT id1 FROM src1 left join src2 on id1 = id2",
 			data: xsql.WindowTuplesSet{
 				Content: []xsql.WindowTuples{
 					{