Repository: lf-edge/ekuiper
Pull request #994 (2021-09-03 06:35:59)

Merge Dev 1.4.0 to master


Patch:
diff --git a/.gitignore b/.gitignore
index f791795d..df66e032 100644
--- a/.gitignore
+++ b/.gitignore
@@ -34,3 +34,5 @@ corss_build_for_rpm.tar
 
 *.swp
 *.history
+
+*/**/*.db
\ No newline at end of file
diff --git a/deploy/chart/kuiper/values.yaml b/deploy/chart/kuiper/values.yaml
index ee4cb994..e5670ff7 100644
--- a/deploy/chart/kuiper/values.yaml
+++ b/deploy/chart/kuiper/values.yaml
@@ -96,6 +96,19 @@ kuiperConfig:
       cacheThreshold: 10
       cacheTriggerCount: 15
       disableCache: true
+    store:
+      type: sqlite
+      redis:
+        host: localhost
+        port: 6379
+        password: kuiper
+        #Timeout in ms
+        timeout: 1000
+      sqlite:
+        #Sqlite absolute database path, if left empty default directory of the application will be used
+        path: ""
+        #Sqlite file name, if left empty name of db will be sqliteKV.db and tskv.db respectively
+        name: ""
   "mqtt_source.yaml":
     #Global MQTT configurations
     default:
diff --git a/docs/en_US/operation/configuration_file.md b/docs/en_US/operation/configuration_file.md
index 0f9eb8eb..b3002e6e 100644
--- a/docs/en_US/operation/configuration_file.md
+++ b/docs/en_US/operation/configuration_file.md
@@ -138,3 +138,39 @@ The content of the page should be similar as below.
   disableCache: false
 ```
 
+## Store configurations
+
+There is possibility to configure storage of state for application. Default storage layer is sqlite database. There is option to set redis as storage.
+In order to use redis as store type property must be changed into redis value.
+
+### Sqlite
+    
+It has properties
+* path - if left empty database will be created in the data directory of the kuiper base key
+* name - name of database file - if left empty it will be `sqliteKV.db`
+ 
+### Redis
+
+It has properties
+* host     - host of redis
+* port     - port of redis
+* password - password used for auth in redis, if left empty auth won't be used
+* timeout  - timeout fo connection
+
+### Config
+```yaml
+    store:
+      #Type of store that will be used for keeping state of the application
+      type: sqlite
+      redis:
+        host: localhost
+        port: 6379
+        password: kuiper
+        #Timeout in ms
+        timeout: 1000
+      sqlite:
+        #Sqlite absolute database path, if left empty default directory of the application will be used
+        path:
+        #Sqlite file name, if left empty name of db will be sqliteKV.db
+        name:
+```
\ No newline at end of file
diff --git a/docs/zh_CN/operation/configuration_file.md b/docs/zh_CN/operation/configuration_file.md
index 6c2e4560..7c732e4a 100755
--- a/docs/zh_CN/operation/configuration_file.md
+++ b/docs/zh_CN/operation/configuration_file.md
@@ -138,3 +138,39 @@ http://host:port/kuiper-plugins/0.9.1/alpine/functions
   disableCache: false
 ```
 
+## 存储配置
+
+可通过配置修改创建的流和规则等状态的存储方式。默认情况下，程序状态存储在 sqlite 数据库中。把存储类型改成 redis，可使用 redis 作为存储方式。
+
+### Sqlite
+
+可配置如下属性：
+* path - 配置 sqlite 存储路径。若为空则存储在默认的 data 目录中。
+* name - 数据库文件名。若为空，则设置为默认名字`sqliteKV.db`。
+
+### Redis
+
+可配置如下属性：
+* host     - redis 服务器地址。
+* port     - redis 服务器端口。
+* password - redis 服务器密码。若 redis 未配置认证系统，则可不设置密码。
+* timeout  - 连接超时时间。
+
+### 配置示例
+
+```yaml
+    store:
+      #Type of store that will be used for keeping state of the application
+      type: sqlite
+      redis:
+        host: localhost
+        port: 6379
+        password: kuiper
+        #Timeout in ms
+        timeout: 1000
+      sqlite:
+        #Sqlite absolute database path, if left empty default directory of the application will be used
+        path:
+        #Sqlite file name, if left empty name of db will be sqliteKV.db
+        name:
+```
\ No newline at end of file
diff --git a/etc/kuiper.yaml b/etc/kuiper.yaml
index 535c158e..336359a4 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -52,3 +52,18 @@ sink:
 
   # Control to disable cache or not. If it's set to true, then the cache will be disabled, otherwise, it will be enabled.
   disableCache: true
+
+store:
+  #Type of store that will be used for keeping state of the application
+  type: sqlite
+  redis:
+    host: localhost
+    port: 6379
+    password: kuiper
+    #Timeout in ms
+    timeout: 1000
+  sqlite:
+    #Sqlite absolute database path, if left empty default directory of the application will be used
+    path:
+    #Sqlite file name, if left empty name of db will be sqliteKV.db
+    name:
\ No newline at end of file
diff --git a/go.mod b/go.mod
index 61084cdb..d00c5d80 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.1
 	github.com/PaesslerAG/gval v1.0.0
 	github.com/PaesslerAG/jsonpath v0.1.1
+	github.com/alicebob/miniredis/v2 v2.15.1
 	github.com/benbjohnson/clock v1.0.0
 	github.com/eclipse/paho.mqtt.golang v1.3.5
 	github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.0
@@ -12,6 +13,7 @@ require (
 	github.com/gdexlab/go-render v1.0.1
 	github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3
 	github.com/golang/protobuf v1.5.0
+	github.com/gomodule/redigo v2.0.0+incompatible
 	github.com/google/uuid v1.2.0
 	github.com/gorilla/handlers v1.4.2
 	github.com/gorilla/mux v1.7.3
diff --git a/go.sum b/go.sum
index 658ee156..71b0f1bd 100644
--- a/go.sum
+++ b/go.sum
@@ -15,6 +15,10 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a h1:HbKu58rmZpUGpz5+4FfNmIU+FmZg2P3Xaj2v2bfNWmk=
+github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
+github.com/alicebob/miniredis/v2 v2.15.1 h1:Fw+ixAJPmKhCLBqDwHlTDqxUxp0xjEwXczEpt1B6r7k=
+github.com/alicebob/miniredis/v2 v2.15.1/go.mod h1:gquAfGbzn92jvtrSC69+6zZnwSODVXVpYDRaGhWaL6I=
 github.com/benbjohnson/clock v1.0.0 h1:78Jk/r6m4wCi6sndMpty7A//t4dw/RW5fV4ZgDVfX1w=
 github.com/benbjohnson/clock v1.0.0/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
@@ -24,6 +28,9 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash/v2 v2.1.0 h1:yTUvW7Vhb89inJ+8irsUqiWjh8iT6sQPZiQzI6ReGkA=
 github.com/cespare/xxhash/v2 v2.1.0/go.mod h1:dgIUBU3pDso/gPgZ1osOZ0iQf77oPR28Tjxl5dIMyVM=
+github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
+github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
+github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 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/cpuguy83/go-md2man v1.0.10 h1:BSKMNlYxDvnunlTymqtgONjNnaRV1sTpcovwwjF22jk=
@@ -80,6 +87,8 @@ github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QD
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/protobuf v1.5.0 h1:LUVKkCeviFUMKqHa4tXIIij/lbhnMbP7Fn5wKdKkRh4=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/gomodule/redigo v2.0.0+incompatible h1:K/R+8tc58AaqLkqG2Ol3Qk+DR/TlNuhuh457pBFPtt0=
+github.com/gomodule/redigo v2.0.0+incompatible/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -222,6 +231,8 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 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=
+github.com/yuin/gopher-lua v0.0.0-20200816102855-ee81675732da h1:NimzV1aGyq29m5ukMK0AMWEhFaL/lrEOaephfuoiARg=
+github.com/yuin/gopher-lua v0.0.0-20200816102855-ee81675732da/go.mod h1:E1AXubJBdNmFERAOucpDIxNzeGfLzg0mYh+UfMWdChA=
 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=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -261,6 +272,7 @@ golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5h
 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-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/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-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index cba79ed8..ca191cc8 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -73,7 +73,20 @@ type KuiperConf struct {
 	Sink struct {
 		CacheThreshold    int  `yaml:"cacheThreshold"`
 		CacheTriggerCount int  `yaml:"cacheTriggerCount"`
-		DisableCache      bool `yaml:"disableCache""`
+		DisableCache      bool `yaml:"disableCache"`
+	}
+	Store struct {
+		Type  string `yaml:"type"`
+		Redis struct {
+			Host     string `yaml:"host"`
+			Port     int    `yaml:"port"`
+			Password string `yaml:"password"`
+			Timeout  int    `yaml:"timeout"`
+		}
+		Sqlite struct {
+			Path string `yaml:"path"`
+			Name string `yaml:"name"`
+		}
 	}
 }
 
diff --git a/internal/pkg/db/config.go b/internal/pkg/db/config.go
new file mode 100644
index 00000000..20cde7e6
--- /dev/null
+++ b/internal/pkg/db/config.go
@@ -0,0 +1,26 @@
+// Copyright 2021 INTECH Process Automation 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 db
+
+import (
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql/sqlite"
+)
+
+type Config struct {
+	Type   string
+	Redis  redis.Config
+	Sqlite sqlite.Config
+}
diff --git a/internal/pkg/db/database.go b/internal/pkg/db/database.go
new file mode 100644
index 00000000..44782e50
--- /dev/null
+++ b/internal/pkg/db/database.go
@@ -0,0 +1,49 @@
+// 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 db
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql/sqlite"
+)
+
+type Database interface {
+	Connect() error
+	Disconnect() error
+}
+
+func CreateDatabase(conf Config) (Database, error) {
+	var db Database
+	var err error
+	databaseType := conf.Type
+	switch databaseType {
+	case "redis":
+		r := redis.NewRedisFromConf(conf.Redis)
+		db = &r
+	case "sqlite":
+		err, db = sqlite.NewSqliteDatabase(conf.Sqlite)
+		if err != nil {
+			return nil, err
+		}
+	default:
+		return nil, fmt.Errorf("unrecognized database type - %s", databaseType)
+	}
+	err = db.Connect()
+	if err != nil {
+		return nil, err
+	}
+	return db, nil
+}
diff --git a/internal/pkg/db/redis/config.go b/internal/pkg/db/redis/config.go
new file mode 100644
index 00000000..f99c1cb2
--- /dev/null
+++ b/internal/pkg/db/redis/config.go
@@ -0,0 +1,22 @@
+// Copyright 2021 INTECH Process Automation 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
+
+type Config struct {
+	Host     string
+	Port     int
+	Password string
+	Timeout  int
+}
diff --git a/internal/pkg/db/redis/redis.go b/internal/pkg/db/redis/redis.go
new file mode 100644
index 00000000..63aa004d
--- /dev/null
+++ b/internal/pkg/db/redis/redis.go
@@ -0,0 +1,121 @@
+// Copyright 2021 INTECH Process Automation 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 (
+	"fmt"
+	"github.com/gomodule/redigo/redis"
+	"sync"
+	"time"
+)
+
+type Instance struct {
+	ConnectionString string
+	pool             *redis.Pool
+	mu               *sync.Mutex
+	config           Config
+}
+
+func NewRedisFromConf(conf Config) Instance {
+	host := conf.Host
+	port := conf.Port
+	return Instance{
+		ConnectionString: connectionString(host, port),
+		pool:             nil,
+		mu:               &sync.Mutex{},
+		config:           conf,
+	}
+}
+
+func NewRedis(host string, port int) Instance {
+	return Instance{
+		ConnectionString: connectionString(host, port),
+		pool:             nil,
+		mu:               &sync.Mutex{},
+	}
+}
+
+func (r *Instance) Connect() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.ConnectionString == "" {
+		return fmt.Errorf("connection string for redis not initalized")
+	}
+	err, pool := r.connectRedis()
+	if err != nil {
+		return err
+	}
+	conn := pool.Get()
+	defer conn.Close()
+	reply, err := conn.Do("PING")
+	if err != nil {
+		return err
+	}
+	response, err := redis.String(reply, err)
+	if err != nil {
+		return err
+	}
+	if response != "PONG" {
+		return fmt.Errorf("failed to connect to redis")
+	}
+	r.pool = pool
+	return nil
+}
+
+func (r *Instance) connectRedis() (error, *redis.Pool) {
+	opts := []redis.DialOption{
+		redis.DialConnectTimeout(time.Duration(r.config.Timeout) * time.Millisecond),
+	}
+	if r.config.Password != "" {
+		opts = append(opts, redis.DialPassword(r.config.Password))
+	}
+	dialFunction := func() (redis.Conn, error) {
+		conn, err := redis.Dial("tcp", r.ConnectionString, opts...)
+		if err == nil {
+			_, err = conn.Do("PING")
+			if err == nil {
+				return conn, nil
+			}
+		}
+		return nil, fmt.Errorf("could not dial redis: %s", err)
+	}
+	pool := &redis.Pool{
+		IdleTimeout: 0,
+		MaxIdle:     10,
+		Dial:        dialFunction,
+	}
+	return nil, pool
+}
+
+func connectionString(host string, port int) string {
+	return fmt.Sprintf("%s:%d", host, port)
+}
+
+func (r *Instance) Disconnect() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r == nil {
+		return nil
+	}
+	err := r.pool.Close()
+	r.pool = nil
+	return err
+}
+
+func (r *Instance) Apply(f func(conn redis.Conn) error) error {
+	connection := r.pool.Get()
+	defer connection.Close()
+	return f(connection)
+}
diff --git a/internal/pkg/sqlkv/database.go b/internal/pkg/db/sql/database.go
similarity index 87%
rename from internal/pkg/sqlkv/database.go
rename to internal/pkg/db/sql/database.go
index c68c7629..490c334e 100644
--- a/internal/pkg/sqlkv/database.go
+++ b/internal/pkg/db/sql/database.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021 INTECH Process Automation Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,12 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package sqlkv
+package sql
 
 import "database/sql"
 
 type Database interface {
-	Connect() error
-	Disconnect() error
 	Apply(f func(db *sql.DB) error) error
 }
diff --git a/internal/pkg/db/sql/sqlite/config.go b/internal/pkg/db/sql/sqlite/config.go
new file mode 100644
index 00000000..bc2db1f2
--- /dev/null
+++ b/internal/pkg/db/sql/sqlite/config.go
@@ -0,0 +1,20 @@
+// Copyright 2021 INTECH Process Automation 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 sqlite
+
+type Config struct {
+	Path string
+	Name string
+}
diff --git a/internal/pkg/sqlkv/sqliteDatabase.go b/internal/pkg/db/sql/sqlite/database.go
similarity index 74%
rename from internal/pkg/sqlkv/sqliteDatabase.go
rename to internal/pkg/db/sql/sqlite/database.go
index 6be1e0fc..e807846d 100644
--- a/internal/pkg/sqlkv/sqliteDatabase.go
+++ b/internal/pkg/db/sql/sqlite/database.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021 INTECH Process Automation Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package sqlkv
+package sqlite
 
 import (
 	"database/sql"
@@ -23,25 +23,30 @@ import (
 	"sync"
 )
 
-type SqliteDatabase struct {
+type Database struct {
 	db   *sql.DB
 	Path string
 	mu   sync.Mutex
 }
 
-func newSqliteDatabase(dir string) (error, *SqliteDatabase) {
+func NewSqliteDatabase(conf Config) (error, *Database) {
+	dir := conf.Path
+	name := "sqliteKV.db"
+	if conf.Name != "" {
+		name = conf.Name
+	}
 	if _, err := os.Stat(dir); os.IsNotExist(err) {
 		os.MkdirAll(dir, os.ModePerm)
 	}
-	dbPath := path.Join(dir, "sqliteKV.db")
-	return nil, &SqliteDatabase{
+	dbPath := path.Join(dir, name)
+	return nil, &Database{
 		db:   nil,
 		Path: dbPath,
 		mu:   sync.Mutex{},
 	}
 }
 
-func (d *SqliteDatabase) Connect() error {
+func (d *Database) Connect() error {
 	db, err := sql.Open("sqlite3", connectionString(d.Path))
 	if err != nil {
 		return err
@@ -57,12 +62,12 @@ func connectionString(dpath string) string {
 	return fmt.Sprintf("file:%s?cache=shared", dpath)
 }
 
-func (d *SqliteDatabase) Disconnect() error {
+func (d *Database) Disconnect() error {
 	err := d.db.Close()
 	return err
 }
 
-func (d *SqliteDatabase) Apply(f func(db *sql.DB) error) error {
+func (d *Database) Apply(f func(db *sql.DB) error) error {
 	d.mu.Lock()
 	err := f(d.db)
 	d.mu.Unlock()
diff --git a/internal/pkg/sqlkv/kvStore.go b/internal/pkg/sqlkv/kvStore.go
deleted file mode 100644
index ad458371..00000000
--- a/internal/pkg/sqlkv/kvStore.go
+++ /dev/null
@@ -1,70 +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 sqlkv
-
-import (
-	"github.com/lf-edge/ekuiper/pkg/kv"
-	"sync"
-)
-
-type kvstores struct {
-	stores map[string]kv.KeyValue
-	mu     sync.Mutex
-}
-
-var stores = kvstores{
-	stores: make(map[string]kv.KeyValue),
-	mu:     sync.Mutex{},
-}
-
-var database Database
-
-func Setup(dataDir string) error {
-	err, db := newSqliteDatabase(dataDir)
-	if err != nil {
-		return err
-	}
-	err = db.Connect()
-	if err != nil {
-		return err
-	}
-	database = db
-	return nil
-}
-
-func Close() {
-	if database != nil {
-		database.Disconnect()
-		database = nil
-	}
-}
-
-func (s *kvstores) get(table string) (kv.KeyValue, error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if store, contains := s.stores[table]; contains {
-		return store, nil
-	}
-	err, store := CreateSqlKvStore(database, table)
-	if err != nil {
-		return nil, err
-	}
-	s.stores[table] = store
-	return store, nil
-}
-
-func GetKVStore(table string) (kv.KeyValue, error) {
-	return stores.get(table)
-}
diff --git a/internal/pkg/store/builder.go b/internal/pkg/store/builder.go
new file mode 100644
index 00000000..c917c885
--- /dev/null
+++ b/internal/pkg/store/builder.go
@@ -0,0 +1,42 @@
+// 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 store
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/db"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql"
+	rb "github.com/lf-edge/ekuiper/internal/pkg/store/redis"
+	sb "github.com/lf-edge/ekuiper/internal/pkg/store/sql"
+	"github.com/lf-edge/ekuiper/pkg/kv"
+)
+
+type Builder interface {
+	CreateStore(table string) (kv.KeyValue, error)
+}
+
+func CreateStoreBuilder(database db.Database) (Builder, error) {
+	switch database.(type) {
+	case *redis.Instance:
+		d := *database.(*redis.Instance)
+		return rb.NewStoreBuilder(d), nil
+	case sql.Database:
+		d := database.(sql.Database)
+		return sb.NewStoreBuilder(d), nil
+	default:
+		return nil, fmt.Errorf("unrecognized database type")
+	}
+}
diff --git a/internal/pkg/store/encoding/encoding.go b/internal/pkg/store/encoding/encoding.go
new file mode 100644
index 00000000..41e8b06a
--- /dev/null
+++ b/internal/pkg/store/encoding/encoding.go
@@ -0,0 +1,30 @@
+// 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 encoding
+
+import (
+	"bytes"
+	"encoding/gob"
+)
+
+func Encode(value interface{}) (error, []byte) {
+	var buff bytes.Buffer
+	gob.Register(value)
+	enc := gob.NewEncoder(&buff)
+	if err := enc.Encode(value); err != nil {
+		return err, nil
+	}
+	return nil, buff.Bytes()
+}
diff --git a/internal/pkg/store/redis/redisKv.go b/internal/pkg/store/redis/redisKv.go
new file mode 100644
index 00000000..061059db
--- /dev/null
+++ b/internal/pkg/store/redis/redisKv.go
@@ -0,0 +1,152 @@
+// 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 redis
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"github.com/gomodule/redigo/redis"
+	dbRedis "github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	kvEncoding "github.com/lf-edge/ekuiper/internal/pkg/store/encoding"
+	"strings"
+)
+
+const KvPrefix = "KV:STORE"
+
+type redisKvStore struct {
+	database  dbRedis.Instance
+	table     string
+	keyPrefix string
+}
+
+func CreateRedisKvStore(redis dbRedis.Instance, table string) (*redisKvStore, error) {
+	store := &redisKvStore{
+		database:  redis,
+		table:     table,
+		keyPrefix: fmt.Sprintf("%s:%s", KvPrefix, table),
+	}
+	return store, nil
+}
+
+func (kv redisKvStore) Setnx(key string, value interface{}) error {
+	return kv.database.Apply(func(conn redis.Conn) error {
+		err, b := kvEncoding.Encode(value)
+		if nil != err {
+			return err
+		}
+		tKey := kv.tableKey(key)
+		reply, err := conn.Do("SETNX", tKey, b)
+		if err != nil {
+			return err
+		}
+		code, err := redis.Int(reply, err)
+		if code == 0 {
+			return fmt.Errorf("item %s already exists under %s key because of %s", key, tKey, err)
+		}
+		return nil
+	})
+}
+
+func (kv redisKvStore) Set(key string, value interface{}) error {
+	err, b := kvEncoding.Encode(value)
+	if nil != err {
+		return err
+	}
+	err = kv.database.Apply(func(conn redis.Conn) error {
+		tKey := kv.tableKey(key)
+		reply, err := conn.Do("SET", tKey, b)
+		code, err := redis.String(reply, err)
+		if err != nil {
+			return err
+		}
+		if code != "OK" {
+			return fmt.Errorf("item %s (under key %s) not set because of %s", key, tKey, err)
+		}
+		return nil
+	})
+	return err
+}
+
+func (kv redisKvStore) Get(key string, value interface{}) (bool, error) {
+	result := false
+	err := kv.database.Apply(func(conn redis.Conn) error {
+		tKey := kv.tableKey(key)
+		reply, err := conn.Do("GET", tKey)
+		if err != nil {
+			return err
+		}
+		buff, err := redis.Bytes(reply, err)
+		if err != nil {
+			result = false
+			return nil
+		}
+		dec := gob.NewDecoder(bytes.NewBuffer(buff))
+		if err := dec.Decode(value); err != nil {
+			return err
+		}
+		result = true
+		return nil
+	})
+	return result, err
+}
+
+func (kv redisKvStore) Delete(key string) error {
+	return kv.database.Apply(func(conn redis.Conn) error {
+		tKey := kv.tableKey(key)
+		_, err := conn.Do("DEL", tKey)
+		return err
+	})
+}
+
+func (kv redisKvStore) Keys() ([]string, error) {
+	keys := make([]string, 0)
+	err := kv.database.Apply(func(conn redis.Conn) error {
+		pattern := fmt.Sprintf("%s:*", kv.keyPrefix)
+		reply, err := conn.Do("KEYS", pattern)
+		keys, err = redis.Strings(reply, err)
+		return err
+	})
+	result := make([]string, 0)
+	for _, k := range keys {
+		result = append(result, kv.trimPrefix(k))
+	}
+	return result, err
+}
+
+func (kv redisKvStore) Clean() error {
+	keys, err := kv.Keys()
+	if err != nil {
+		return err
+	}
+	keysToRemove := make([]interface{}, len(keys))
+	for i, v := range keysToRemove {
+		keysToRemove[i] = v
+	}
+	err = kv.database.Apply(func(conn redis.Conn) error {
+		_, err := conn.Do("DEL", keysToRemove...)
+		return err
+	})
+	return err
+}
+
+func (kv redisKvStore) tableKey(key string) string {
+	return fmt.Sprintf("%s:%s:%s", KvPrefix, kv.table, key)
+}
+
+func (kv redisKvStore) trimPrefix(fullKey string) string {
+	prefixToTrim := fmt.Sprintf("%s:%s:", KvPrefix, kv.table)
+	return strings.TrimPrefix(fullKey, prefixToTrim)
+}
diff --git a/internal/pkg/store/redis/redisStoreBuilder.go b/internal/pkg/store/redis/redisStoreBuilder.go
new file mode 100644
index 00000000..cd48ef96
--- /dev/null
+++ b/internal/pkg/store/redis/redisStoreBuilder.go
@@ -0,0 +1,34 @@
+// 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 redis
+
+import (
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	"github.com/lf-edge/ekuiper/pkg/kv"
+)
+
+type StoreBuilder struct {
+	database redis.Instance
+}
+
+func NewStoreBuilder(redis redis.Instance) StoreBuilder {
+	return StoreBuilder{
+		database: redis,
+	}
+}
+
+func (b StoreBuilder) CreateStore(table string) (kv.KeyValue, error) {
+	return CreateRedisKvStore(b.database, table)
+}
diff --git a/internal/pkg/store/setup.go b/internal/pkg/store/setup.go
new file mode 100644
index 00000000..7ce6a055
--- /dev/null
+++ b/internal/pkg/store/setup.go
@@ -0,0 +1,71 @@
+// 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 store
+
+import (
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/db"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql/sqlite"
+)
+
+func SetupDefault() error {
+	dir, err := conf.GetDataLoc()
+	if err != nil {
+		return err
+	}
+
+	c := db.Config{
+		Type:  "sqlite",
+		Redis: redis.Config{},
+		Sqlite: sqlite.Config{
+			Path: dir,
+			Name: "",
+		},
+	}
+
+	return Setup(c)
+}
+
+func SetupWithKuiperConfig(conf *conf.KuiperConf) error {
+	c := db.Config{
+		Type: conf.Store.Type,
+		Redis: redis.Config{
+			Host:     conf.Store.Redis.Host,
+			Port:     conf.Store.Redis.Port,
+			Password: conf.Store.Redis.Password,
+			Timeout:  conf.Store.Redis.Timeout,
+		},
+		Sqlite: sqlite.Config{
+			Path: conf.Store.Sqlite.Path,
+			Name: conf.Store.Sqlite.Name,
+		},
+	}
+	return Setup(c)
+}
+
+func Setup(config db.Config) error {
+	database, err := db.CreateDatabase(config)
+	if err != nil {
+		return err
+	}
+
+	err = InitGlobalStores(database)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
diff --git a/internal/pkg/sqlkv/sqlKv.go b/internal/pkg/store/sql/sqlKv.go
similarity index 90%
rename from internal/pkg/sqlkv/sqlKv.go
rename to internal/pkg/store/sql/sqlKv.go
index 1674dddd..36deea20 100644
--- a/internal/pkg/sqlkv/sqlKv.go
+++ b/internal/pkg/store/sql/sqlKv.go
@@ -12,23 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package sqlkv
+package sql
 
 import (
 	"bytes"
 	"database/sql"
 	"encoding/gob"
 	"fmt"
+	dbSql "github.com/lf-edge/ekuiper/internal/pkg/db/sql"
+	kvEncoding "github.com/lf-edge/ekuiper/internal/pkg/store/encoding"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"strings"
 )
 
 type sqlKvStore struct {
-	database Database
+	database dbSql.Database
 	table    string
 }
 
-func CreateSqlKvStore(database Database, table string) (error, *sqlKvStore) {
+func createSqlKvStore(database dbSql.Database, table string) (*sqlKvStore, error) {
 	store := &sqlKvStore{
 		database: database,
 		table:    table,
@@ -39,24 +41,14 @@ func CreateSqlKvStore(database Database, table string) (error, *sqlKvStore) {
 		return err
 	})
 	if err != nil {
-		return err, nil
-	}
-	return nil, store
-}
-
-func encode(value interface{}) ([]byte, error) {
-	var buf bytes.Buffer
-	gob.Register(value)
-	enc := gob.NewEncoder(&buf)
-	if err := enc.Encode(value); err != nil {
 		return nil, err
 	}
-	return buf.Bytes(), nil
+	return store, nil
 }
 
 func (kv *sqlKvStore) Setnx(key string, value interface{}) error {
 	return kv.database.Apply(func(db *sql.DB) error {
-		b, err := encode(value)
+		err, b := kvEncoding.Encode(value)
 		if nil != err {
 			return err
 		}
@@ -74,7 +66,7 @@ func (kv *sqlKvStore) Setnx(key string, value interface{}) error {
 }
 
 func (kv *sqlKvStore) Set(key string, value interface{}) error {
-	b, err := encode(value)
+	err, b := kvEncoding.Encode(value)
 	if nil != err {
 		return err
 	}
diff --git a/internal/pkg/store/sql/sqlStoreBuilder.go b/internal/pkg/store/sql/sqlStoreBuilder.go
new file mode 100644
index 00000000..0ad4d884
--- /dev/null
+++ b/internal/pkg/store/sql/sqlStoreBuilder.go
@@ -0,0 +1,34 @@
+// 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 sql
+
+import (
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql"
+	"github.com/lf-edge/ekuiper/pkg/kv"
+)
+
+type StoreBuilder struct {
+	database sql.Database
+}
+
+func NewStoreBuilder(d sql.Database) StoreBuilder {
+	return StoreBuilder{
+		database: d,
+	}
+}
+
+func (b StoreBuilder) CreateStore(table string) (kv.KeyValue, error) {
+	return createSqlKvStore(b.database, table)
+}
diff --git a/internal/pkg/store/stores.go b/internal/pkg/store/stores.go
new file mode 100644
index 00000000..015c2095
--- /dev/null
+++ b/internal/pkg/store/stores.go
@@ -0,0 +1,102 @@
+// 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 store
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/db"
+	"github.com/lf-edge/ekuiper/internal/pkg/ts"
+	kv2 "github.com/lf-edge/ekuiper/pkg/kv"
+	"sync"
+)
+
+type stores struct {
+	kv        map[string]kv2.KeyValue
+	ts        map[string]kv2.Tskv
+	mu        sync.Mutex
+	kvBuilder Builder
+	tsBuilder ts.Builder
+}
+
+func newStores(db db.Database) (error, *stores) {
+	var err error
+	var kvBuilder Builder
+	var tsBuilder ts.Builder
+	kvBuilder, err = CreateStoreBuilder(db)
+	if err != nil {
+		return err, nil
+	}
+	err, tsBuilder = ts.CreateTsBuilder(db)
+	if err != nil {
+		return err, nil
+	}
+	return nil, &stores{
+		kv:        make(map[string]kv2.KeyValue),
+		ts:        make(map[string]kv2.Tskv),
+		mu:        sync.Mutex{},
+		kvBuilder: kvBuilder,
+		tsBuilder: tsBuilder,
+	}
+}
+
+func (s *stores) GetKV(table string) (error, kv2.KeyValue) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if ks, contains := s.kv[table]; contains {
+		return nil, ks
+	}
+	ks, err := s.kvBuilder.CreateStore(table)
+	if err != nil {
+		return err, nil
+	}
+	s.kv[table] = ks
+	return nil, ks
+}
+
+func (s *stores) GetTS(table string) (error, kv2.Tskv) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if tts, contains := s.ts[table]; contains {
+		return nil, tts
+	}
+	err, tts := s.tsBuilder.CreateTs(table)
+	if err != nil {
+		return err, nil
+	}
+	s.ts[table] = tts
+	return nil, tts
+}
+
+var globalStores *stores = nil
+
+func InitGlobalStores(db db.Database) error {
+	var err error
+	err, globalStores = newStores(db)
+	return err
+}
+
+func GetKV(table string) (error, kv2.KeyValue) {
+	if globalStores == nil {
+		return fmt.Errorf("global stores are not initialized"), nil
+	}
+	return globalStores.GetKV(table)
+}
+
+func GetTS(table string) (error, kv2.Tskv) {
+	if globalStores == nil {
+		return fmt.Errorf("global stores are not initialized"), nil
+	}
+	return globalStores.GetTS(table)
+}
diff --git a/internal/pkg/ts/builder.go b/internal/pkg/ts/builder.go
new file mode 100644
index 00000000..d5533341
--- /dev/null
+++ b/internal/pkg/ts/builder.go
@@ -0,0 +1,42 @@
+// 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 ts
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/db"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql"
+	rb "github.com/lf-edge/ekuiper/internal/pkg/ts/redis"
+	sb "github.com/lf-edge/ekuiper/internal/pkg/ts/sql"
+	"github.com/lf-edge/ekuiper/pkg/kv"
+)
+
+type Builder interface {
+	CreateTs(table string) (error, kv.Tskv)
+}
+
+func CreateTsBuilder(database db.Database) (error, Builder) {
+	switch database.(type) {
+	case *redis.Instance:
+		d := *database.(*redis.Instance)
+		return nil, rb.NewTsBuilder(d)
+	case sql.Database:
+		d := database.(sql.Database)
+		return nil, sb.NewTsBuilder(d)
+	default:
+		return fmt.Errorf("unrecognized database type"), nil
+	}
+}
diff --git a/internal/pkg/ts/redis/redisTs.go b/internal/pkg/ts/redis/redisTs.go
new file mode 100644
index 00000000..46954dcc
--- /dev/null
+++ b/internal/pkg/ts/redis/redisTs.go
@@ -0,0 +1,190 @@
+// 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 redis
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"github.com/gomodule/redigo/redis"
+	dbRedis "github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	kvEncoding "github.com/lf-edge/ekuiper/internal/pkg/store/encoding"
+	"strconv"
+)
+
+const (
+	TsPrefix             = "KV:TS"
+	AddToSortedSet       = "ZADD"
+	ReversedRangeByScore = "ZREVRANGEBYSCORE"
+	RemoveRangeByScore   = "ZREMRANGEBYSCORE"
+	Delete               = "DEL"
+	ReversedRange        = "ZREVRANGE"
+)
+
+type ts struct {
+	redis dbRedis.Instance
+	table string
+	last  int64
+	key   string
+}
+
+func init() {
+	gob.Register(make(map[string]interface{}))
+}
+
+func createRedisTs(redis dbRedis.Instance, table string) (error, *ts) {
+	key := fmt.Sprintf("%s:%s", TsPrefix, table)
+	err, lastTs := getLast(redis, key)
+	if err != nil {
+		return err, nil
+	}
+	s := &ts{
+		redis: redis,
+		table: table,
+		last:  lastTs,
+		key:   key,
+	}
+	return nil, s
+}
+
+func (t *ts) Set(key int64, value interface{}) (bool, error) {
+	if key <= t.last {
+		return false, nil
+	}
+	err, b := kvEncoding.Encode(value)
+	if err != nil {
+		return false, err
+	}
+	err = t.redis.Apply(func(conn redis.Conn) error {
+		reply, err := conn.Do(AddToSortedSet, t.key, key, b)
+		if err != nil {
+			return err
+		}
+		length, err := redis.Int(reply, err)
+		if err != nil {
+			return err
+		}
+		if length == 0 {
+			return fmt.Errorf("list at %s key should be non empty", t.key)
+		}
+		t.last = key
+		return nil
+	})
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+func (t ts) Get(key int64, value interface{}) (bool, error) {
+	err := t.redis.Apply(func(conn redis.Conn) error {
+		reply, err := conn.Do(ReversedRangeByScore, t.key, key, key)
+		if err != nil {
+			return err
+		}
+		var tmp [][]byte
+		tmp, err = redis.ByteSlices(reply, err)
+		if err != nil {
+			return err
+		}
+		if len(tmp) == 0 {
+			return fmt.Errorf("record under %s key and %d score not found", t.key, key)
+		}
+		dec := gob.NewDecoder(bytes.NewBuffer(tmp[0]))
+		err = dec.Decode(value)
+		return err
+	})
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+func (t ts) Last(value interface{}) (int64, error) {
+	var last int64 = 0
+	err := t.redis.Apply(func(conn redis.Conn) error {
+		reply, err := conn.Do(ReversedRange, t.key, 0, 0, "WITHSCORES")
+		if err != nil {
+			return err
+		}
+		var tmp [][]byte
+		tmp, err = redis.ByteSlices(reply, err)
+		if err != nil {
+			return err
+		}
+		if len(tmp) > 0 {
+			dec := gob.NewDecoder(bytes.NewBuffer(tmp[0]))
+			if err = dec.Decode(value); err != nil {
+				return err
+			}
+			last, err = strconv.ParseInt(string(tmp[1]), 10, 64)
+		}
+		return err
+	})
+	if err != nil {
+		return 0, err
+	}
+	return last, nil
+}
+
+func (t ts) Delete(key int64) error {
+	return t.redis.Apply(func(conn redis.Conn) error {
+		_, err := conn.Do(RemoveRangeByScore, t.key, key, key)
+		return err
+	})
+}
+
+func (t ts) DeleteBefore(key int64) error {
+	return t.redis.Apply(func(conn redis.Conn) error {
+		bound := fmt.Sprintf("(%d", key)
+		_, err := conn.Do(RemoveRangeByScore, t.key, "-INF", bound)
+		return err
+	})
+}
+
+func (t ts) Close() error {
+	return nil
+}
+
+func (t ts) Drop() error {
+	return t.redis.Apply(func(conn redis.Conn) error {
+		_, err := conn.Do(Delete, t.key)
+		return err
+	})
+}
+
+func getLast(db dbRedis.Instance, key string) (error, int64) {
+	var lastTs int64
+	err := db.Apply(func(conn redis.Conn) error {
+		reply, err := conn.Do(ReversedRange, key, 0, 0, "WITHSCORES")
+		if err != nil {
+			return err
+		}
+		var tmp [][]byte
+		tmp, err = redis.ByteSlices(reply, err)
+		if err != nil {
+			return err
+		}
+		if len(tmp) == 0 {
+			return nil
+		}
+		lastTs, err = strconv.ParseInt(string(tmp[1]), 10, 64)
+		return err
+	})
+	if err != nil {
+		return err, 0
+	}
+	return nil, lastTs
+}
diff --git a/tools/migration/main.go b/internal/pkg/ts/redis/redisTsBuilder.go
similarity index 64%
rename from tools/migration/main.go
rename to internal/pkg/ts/redis/redisTsBuilder.go
index 37074ab8..7e1ef976 100644
--- a/tools/migration/main.go
+++ b/internal/pkg/ts/redis/redisTsBuilder.go
@@ -12,22 +12,23 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package main
+package redis
 
 import (
-	"github.com/lf-edge/ekuiper/tools/migration/util"
-	"log"
-	"os"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	st "github.com/lf-edge/ekuiper/pkg/kv"
 )
 
-func main() {
-	if 2 != len(os.Args) {
-		log.Fatal("Please enter the correct path. For example: ./migration kuiper/bin/data")
-	}
+type TsBuilder struct {
+	redis redis.Instance
+}
 
-	if err := util.DataMigration(os.Args[1]); nil != err {
-		log.Fatal(err)
-	} else {
-		log.Println("The data migration was successful.")
+func NewTsBuilder(d redis.Instance) TsBuilder {
+	return TsBuilder{
+		redis: d,
 	}
 }
+
+func (b TsBuilder) CreateTs(table string) (error, st.Tskv) {
+	return createRedisTs(b.redis, table)
+}
diff --git a/internal/pkg/ts/sql/sqlTs.go b/internal/pkg/ts/sql/sqlTs.go
new file mode 100644
index 00000000..57c84ca6
--- /dev/null
+++ b/internal/pkg/ts/sql/sqlTs.go
@@ -0,0 +1,153 @@
+// Copyright 2021 INTECH Process Automation 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 sql
+
+import (
+	"bytes"
+	"database/sql"
+	"encoding/gob"
+	"fmt"
+	dbSql "github.com/lf-edge/ekuiper/internal/pkg/db/sql"
+	kvEncoding "github.com/lf-edge/ekuiper/internal/pkg/store/encoding"
+)
+
+type ts struct {
+	database dbSql.Database
+	table    string
+	last     int64
+}
+
+func init() {
+	gob.Register(make(map[string]interface{}))
+}
+
+func createSqlTs(database dbSql.Database, table string) (error, *ts) {
+	store := &ts{
+		database: database,
+		table:    table,
+		last:     getLast(database, table),
+	}
+	err := store.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS '%s'('key' INTEGER PRIMARY KEY, 'val' BLOB);", table)
+		_, err := db.Exec(query)
+		return err
+	})
+	if err != nil {
+		return err, nil
+	}
+	return nil, store
+}
+
+func (t *ts) Set(key int64, value interface{}) (bool, error) {
+	if key <= t.last {
+		return false, nil
+	}
+	err, b := kvEncoding.Encode(value)
+	if err != nil {
+		return false, err
+	}
+	err = t.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("INSERT INTO %s(key,val) values(?,?);", t.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			return err
+		}
+		defer stmt.Close()
+		_, err = stmt.Exec(key, b)
+		if err != nil {
+			return err
+		}
+		t.last = key
+		return nil
+	})
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+func (t ts) Get(key int64, value interface{}) (bool, error) {
+	result := false
+	err := t.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("SELECT val FROM %s WHERE key=%d;", t.table, key)
+		row := db.QueryRow(query)
+		var tmp []byte
+		switch err := row.Scan(&tmp); err {
+		case sql.ErrNoRows:
+			return nil
+		case nil:
+		default:
+			return err
+		}
+
+		dec := gob.NewDecoder(bytes.NewBuffer(tmp))
+		if err := dec.Decode(value); err != nil {
+			return err
+		}
+		result = true
+		return nil
+	})
+	if err != nil {
+		return false, err
+	}
+	return result, nil
+}
+
+func (t ts) Last(value interface{}) (int64, error) {
+	_, err := t.Get(t.last, value)
+	if err != nil {
+		return 0, err
+	}
+	return t.last, nil
+}
+
+func (t ts) Delete(key int64) error {
+	return t.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("DELETE FROM %s WHERE key=%d;", t.table, key)
+		_, err := db.Exec(query)
+		return err
+	})
+}
+
+func (t ts) DeleteBefore(key int64) error {
+	return t.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("DELETE FROM %s WHERE key<%d;", t.table, key)
+		_, err := db.Exec(query)
+		return err
+	})
+}
+
+func (t ts) Close() error {
+	return nil
+}
+
+func (t ts) Drop() error {
+	return t.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("Drop table %s;", t.table)
+		_, err := db.Exec(query)
+		return err
+	})
+}
+
+func getLast(d dbSql.Database, table string) int64 {
+	var last int64 = 0
+	_ = d.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("SELECT key FROM %s Order by key DESC Limit 1;", table)
+		row := db.QueryRow(query)
+		err := row.Scan(&last)
+		return err
+	})
+	return last
+}
diff --git a/internal/pkg/ts/sql/sqlTsBuilder.go b/internal/pkg/ts/sql/sqlTsBuilder.go
new file mode 100644
index 00000000..6dff2207
--- /dev/null
+++ b/internal/pkg/ts/sql/sqlTsBuilder.go
@@ -0,0 +1,34 @@
+// 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 sql
+
+import (
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql"
+	ts2 "github.com/lf-edge/ekuiper/pkg/kv"
+)
+
+type TsBuilder struct {
+	database sql.Database
+}
+
+func NewTsBuilder(d sql.Database) TsBuilder {
+	return TsBuilder{
+		database: d,
+	}
+}
+
+func (b TsBuilder) CreateTs(table string) (error, ts2.Tskv) {
+	return createSqlTs(b.database, table)
+}
diff --git a/internal/pkg/tskv/sqlite.go b/internal/pkg/tskv/sqlite.go
deleted file mode 100644
index 23532982..00000000
--- a/internal/pkg/tskv/sqlite.go
+++ /dev/null
@@ -1,169 +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 tskv
-
-import (
-	"bytes"
-	"database/sql"
-	"encoding/gob"
-	"fmt"
-	"github.com/lf-edge/ekuiper/internal/conf"
-	_ "github.com/mattn/go-sqlite3"
-	"path"
-	"sync"
-)
-
-// All TSKV instances share ONE database with different tables
-var (
-	db   *sql.DB
-	once sync.Once
-)
-
-// SqliteTskv All TSKV instances share the same database but with different tables
-// Each table must have ONLY ONE instance
-type SqliteTskv struct {
-	table string
-	// only append key bigger than the latest key inside; ONLY check in the instance itself
-	last int64
-}
-
-func init() {
-	gob.Register(make(map[string]interface{}))
-}
-
-func NewSqlite(table string) (*SqliteTskv, error) {
-	var outerError error
-	once.Do(func() {
-		d, err := conf.GetDataLoc()
-		if err != nil {
-			outerError = err
-			return
-		}
-		db, outerError = sql.Open("sqlite3", path.Join(d, "tskv.db"))
-	})
-	if outerError != nil {
-		return nil, outerError
-	}
-	if db == nil {
-		return nil, fmt.Errorf("cannot initiate sqlite db, please restart")
-	}
-	sqlStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS '%s'('key' INTEGER PRIMARY KEY, 'val' BLOB);", table)
-	_, outerError = db.Exec(sqlStr)
-	if outerError != nil {
-		return nil, fmt.Errorf("cannot create table: %v", outerError)
-	}
-	return &SqliteTskv{
-		table: table,
-		last:  last(table),
-	}, nil
-}
-
-func (m *SqliteTskv) Set(key int64, value interface{}) (bool, error) {
-	if key > m.last {
-		b, err := m.encode(value)
-		if err != nil {
-			return false, err
-		}
-		sqlStr := fmt.Sprintf("INSERT INTO %s(key,val) values(?,?);", m.table)
-		stmt, err := db.Prepare(sqlStr)
-		if err != nil {
-			return false, err
-		}
-		defer stmt.Close()
-		_, err = stmt.Exec(key, b)
-		if err != nil {
-			return false, err
-		} else {
-			m.last = key
-			return true, nil
-		}
-	} else {
-		return false, nil
-	}
-}
-
-func (m *SqliteTskv) Get(key int64, value interface{}) (bool, error) {
-	sqlStr := fmt.Sprintf("SELECT val FROM %s WHERE key=%d;", m.table, key)
-	row := db.QueryRow(sqlStr)
-	var tmp []byte
-	switch err := row.Scan(&tmp); err {
-	case sql.ErrNoRows:
-		return false, nil
-	case nil:
-		// do nothing, continue processing
-	default:
-		return false, err
-	}
-
-	dec := gob.NewDecoder(bytes.NewBuffer(tmp))
-	if err := dec.Decode(value); err != nil {
-		return false, err
-	}
-	return true, nil
-}
-
-func (m *SqliteTskv) Last(value interface{}) (int64, error) {
-	_, err := m.Get(m.last, value)
-	if err != nil {
-		return 0, err
-	}
-	return m.last, nil
-}
-
-func (m *SqliteTskv) Delete(k int64) error {
-	sqlStr := fmt.Sprintf("DELETE FROM %s WHERE key=%d;", m.table, k)
-	_, err := db.Exec(sqlStr)
-	return err
-}
-
-func (m *SqliteTskv) DeleteBefore(k int64) error {
-	sqlStr := fmt.Sprintf("DELETE FROM %s WHERE key<%d;", m.table, k)
-	_, err := db.Exec(sqlStr)
-	return err
-}
-
-func (m *SqliteTskv) Close() error {
-	return nil
-}
-
-func (m *SqliteTskv) Drop() error {
-	sqlStr := fmt.Sprintf("Drop table %s;", m.table)
-	_, err := db.Exec(sqlStr)
-	return err
-}
-
-func (m *SqliteTskv) encode(value interface{}) ([]byte, error) {
-	var buf bytes.Buffer
-	gob.Register(value)
-	enc := gob.NewEncoder(&buf)
-	if err := enc.Encode(value); err != nil {
-		return nil, err
-	}
-	return buf.Bytes(), nil
-}
-
-func last(table string) int64 {
-	sqlStr := fmt.Sprintf("SELECT key FROM %s Order by key DESC Limit 1;", table)
-	row := db.QueryRow(sqlStr)
-	var tmp int64
-	switch err := row.Scan(&tmp); err {
-	case sql.ErrNoRows:
-		return 0
-	case nil:
-		return tmp
-	default:
-		return 0
-	}
-}
diff --git a/internal/plugin/manager.go b/internal/plugin/manager.go
index af27fd01..e71d5ed9 100644
--- a/internal/plugin/manager.go
+++ b/internal/plugin/manager.go
@@ -22,7 +22,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/lf-edge/ekuiper/pkg/kv"
@@ -295,7 +295,7 @@ func NewPluginManager() (*Manager, error) {
 			outerErr = fmt.Errorf("cannot find etc folder: %s", err)
 			return
 		}
-		db, err := sqlkv.GetKVStore("pluginFuncs")
+		err, db := store.GetKV("pluginFuncs")
 		if err != nil {
 			outerErr = fmt.Errorf("error when opening db: %v.", err)
 		}
diff --git a/internal/processor/rule.go b/internal/processor/rule.go
index f6ead577..e6a5e479 100644
--- a/internal/processor/rule.go
+++ b/internal/processor/rule.go
@@ -19,8 +19,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
-	"github.com/lf-edge/ekuiper/internal/pkg/tskv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/topo"
 	"github.com/lf-edge/ekuiper/internal/topo/node"
 	"github.com/lf-edge/ekuiper/internal/topo/planner"
@@ -36,7 +35,7 @@ type RuleProcessor struct {
 }
 
 func NewRuleProcessor() *RuleProcessor {
-	db, err := sqlkv.GetKVStore("rule")
+	err, db := store.GetKV("rule")
 	if err != nil {
 		panic(fmt.Sprintf("Can not initalize store for the rule processor at path 'rule': %v", err))
 	}
@@ -234,7 +233,7 @@ func (p *RuleProcessor) ExecDrop(name string) (string, error) {
 }
 
 func cleanCheckpoint(name string) error {
-	db, err := tskv.NewSqlite(name)
+	err, db := store.GetTS(name)
 	if err != nil {
 		return err
 	}
@@ -242,7 +241,7 @@ func cleanCheckpoint(name string) error {
 }
 
 func cleanSinkCache(rule *api.Rule) error {
-	store, err := sqlkv.GetKVStore("sink")
+	err, store := store.GetKV("sink")
 	if err != nil {
 		return err
 	}
diff --git a/internal/processor/stream.go b/internal/processor/stream.go
index e907422b..e16ad141 100644
--- a/internal/processor/stream.go
+++ b/internal/processor/stream.go
@@ -19,7 +19,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -36,7 +36,7 @@ type StreamProcessor struct {
 }
 
 func NewStreamProcessor() *StreamProcessor {
-	db, err := sqlkv.GetKVStore("stream")
+	err, db := store.GetKV("stream")
 	if err != nil {
 		panic(fmt.Sprintf("Can not initalize store for the stream processor at path 'stream': %v", err))
 	}
diff --git a/internal/server/server.go b/internal/server/server.go
index d437b458..a13ea3e0 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -16,7 +16,7 @@ package server
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/plugin"
 	"github.com/lf-edge/ekuiper/internal/processor"
 	"github.com/lf-edge/ekuiper/internal/service"
@@ -34,7 +34,6 @@ import (
 )
 
 var (
-	dataDir         string
 	logger          = conf.Log
 	startTimeStamp  int64
 	version         = ""
@@ -50,15 +49,7 @@ func StartUp(Version, LoadFileType string) {
 	startTimeStamp = time.Now().Unix()
 	conf.InitConf()
 
-	dr, err := conf.GetDataLoc()
-	if err != nil {
-		panic(err)
-	} else {
-		logger.Infof("db location is %s", dr)
-		dataDir = dr
-	}
-
-	err = sqlkv.Setup(dataDir)
+	err := store.SetupWithKuiperConfig(conf.Config)
 	if err != nil {
 		panic(err)
 	}
@@ -185,6 +176,5 @@ func StartUp(Version, LoadFileType string) {
 		logger.Info("prometheus server successfully shutdown.")
 	}
 
-	sqlkv.Close()
 	os.Exit(0)
 }
diff --git a/internal/service/manager.go b/internal/service/manager.go
index a9f18696..2d8f898f 100644
--- a/internal/service/manager.go
+++ b/internal/service/manager.go
@@ -20,7 +20,7 @@ import (
 	kconf "github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/filex"
 	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/kv"
 	"io/ioutil"
@@ -60,11 +60,11 @@ func GetServiceManager() (*Manager, error) {
 		if err != nil {
 			return nil, fmt.Errorf("cannot find etc/services folder: %s", err)
 		}
-		sdb, err := sqlkv.GetKVStore("services")
+		err, sdb := store.GetKV("services")
 		if err != nil {
 			return nil, fmt.Errorf("cannot open service db: %s", err)
 		}
-		fdb, err := sqlkv.GetKVStore("serviceFuncs")
+		err, fdb := store.GetKV("serviceFuncs")
 		if err != nil {
 			return nil, fmt.Errorf("cannot open function db: %s", err)
 		}
diff --git a/internal/topo/node/sink_cache.go b/internal/topo/node/sink_cache.go
index 078a92a8..79827363 100644
--- a/internal/topo/node/sink_cache.go
+++ b/internal/topo/node/sink_cache.go
@@ -18,7 +18,7 @@ import (
 	"encoding/gob"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/topo/checkpoint"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/kv"
@@ -101,7 +101,7 @@ func (c *Cache) initStore(ctx api.StreamContext) {
 		Tail: 0,
 	}
 	var err error
-	c.store, err = sqlkv.GetKVStore(path.Join("sink", ctx.GetRuleId()))
+	err, c.store = store.GetKV(path.Join("sink", ctx.GetRuleId()))
 	if err != nil {
 		c.drainError(err)
 	}
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index 9a110b57..e12308e0 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -18,7 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	store2 "github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/topo"
 	"github.com/lf-edge/ekuiper/internal/topo/node"
 	"github.com/lf-edge/ekuiper/internal/topo/operator"
@@ -49,7 +49,7 @@ func PlanWithSourcesAndSinks(rule *api.Rule, sources []*node.SourceNode, sinks [
 	if rule.Options.SendMetaToSink && (len(streamsFromStmt) > 1 || stmt.Dimensions != nil) {
 		return nil, fmt.Errorf("Invalid option sendMetaToSink, it can not be applied to window")
 	}
-	store, err := sqlkv.GetKVStore("stream")
+	err, store := store2.GetKV("stream")
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/topo/state/kv_store.go b/internal/topo/state/kv_store.go
index c47d521c..4e8ec1ef 100644
--- a/internal/topo/state/kv_store.go
+++ b/internal/topo/state/kv_store.go
@@ -18,9 +18,10 @@ import (
 	"encoding/gob"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/tskv"
+	ts "github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/topo/checkpoint"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	ts2 "github.com/lf-edge/ekuiper/pkg/kv"
 	"sync"
 )
 
@@ -35,7 +36,7 @@ func init() {
 //  { "checkpoint1", "checkpoint2" ... "checkpointn" : The complete or incomplete snapshot
 //
 type KVStore struct {
-	db          tskv.Tskv
+	db          ts2.Tskv
 	mapStore    *sync.Map //The current root store of a rule
 	checkpoints []int64
 	max         int
@@ -48,7 +49,7 @@ type KVStore struct {
 //"$checkpointId":A map with key of checkpoint id and value of snapshot(gob serialized)
 //Assume each operator only has one instance
 func getKVStore(ruleId string) (*KVStore, error) {
-	db, err := tskv.NewSqlite(ruleId)
+	err, db := ts.GetTS(ruleId)
 	if err != nil {
 		return nil, err
 	}
@@ -66,8 +67,10 @@ func (s *KVStore) restore() error {
 	if err != nil {
 		return err
 	}
-	s.checkpoints = []int64{k}
-	s.mapStore.Store(k, cast.MapToSyncMap(m))
+	if k > 0 {
+		s.checkpoints = []int64{k}
+		s.mapStore.Store(k, cast.MapToSyncMap(m))
+	}
 	return nil
 }
 
diff --git a/internal/pkg/tskv/tskv.go b/pkg/kv/tskv.go
similarity index 97%
rename from internal/pkg/tskv/tskv.go
rename to pkg/kv/tskv.go
index e0a234d3..d709dd89 100644
--- a/internal/pkg/tskv/tskv.go
+++ b/pkg/kv/tskv.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package tskv
+package kv
 
 type Tskv interface {
 	Set(k int64, v interface{}) (inserted bool, err error)
@@ -21,4 +21,5 @@ type Tskv interface {
 	Delete(k int64) error
 	DeleteBefore(int64) error
 	Close() error
+	Drop() error
 }
diff --git a/tools/migration/README-CN.md b/tools/migration/README-CN.md
deleted file mode 100644
index 23d3ca77..00000000
--- a/tools/migration/README-CN.md
+++ /dev/null
@@ -1,20 +0,0 @@
-## 1 程序说明及应用：
-
-### 1.1 程序说明：
-
-​    本程序用于历史数据迁移，将版本号小于1.0.2的 eKuiper 数据迁移到1.0.2版本中。程序运行时将 `dada` 文件夹及其子目录下所有名为 `stores.data` 的文件数据迁移到 data 目录下的 `sqliteKV.db` 数据库中。若原数据存储路径为 `data/rule/store.data`,则迁移后数据位于`data/sqliteKV.db `路径下名为`rule`的表中。
-
-### 1.2 编译程序：
-
-执行 `go build -o tools/migration/migration tools/migration/main.go` 命令即可生成 migration 程序。
-
-### 1.3 操作示例
-
-用户需要提供 eKuiper 的 data 文件夹的路径
-
-```shell
-./migration $(ekuiper/data)
-```
-
-## 
-
diff --git a/tools/migration/README.md b/tools/migration/README.md
deleted file mode 100644
index 8cbae6fd..00000000
--- a/tools/migration/README.md
+++ /dev/null
@@ -1,21 +0,0 @@
-## 1 Program description and application:
-
-### 1.1 Program description:
-
-This program is used for historical data migration, which migrates eKuiper data whose version number is less than 1.0.2 to 1.0.2 version. When the program is running, all file data named `stores.data` in the `dada` folder and its subdirectories will be migrated to the `sqliteKV.db` database under the data directory. If the original data storage path is `data/rule/store.data`, the migrated data will be located in the table named `rule` under the path `data/sqliteKV.db`.
-
-### 1.2 Compile the program:
-
-Execute the `go build -o tools/migration/migration tools/migration/main.go` command to generate the migration program.
-
-### 1.3 Operation example
-
-The user needs to provide the path of eKuiper's data folder
-
-```shell
-./migration $(ekuiper/data)
-```
-
-## 
-
-
diff --git a/tools/migration/go.mod b/tools/migration/go.mod
deleted file mode 100644
index c5354861..00000000
--- a/tools/migration/go.mod
+++ /dev/null
@@ -1,10 +0,0 @@
-module github.com/lf-edge/ekuiper/tools/migration
-
-go 1.16
-
-require (
-	github.com/lf-edge/ekuiper v0.0.0
-	github.com/patrickmn/go-cache v2.1.0+incompatible
-)
-
-replace github.com/lf-edge/ekuiper => ../../
diff --git a/tools/migration/go.sum b/tools/migration/go.sum
deleted file mode 100644
index e28c0e5e..00000000
--- a/tools/migration/go.sum
+++ /dev/null
@@ -1,274 +0,0 @@
-cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
-github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
-github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=
-github.com/Masterminds/sprig/v3 v3.2.1/go.mod h1:UoaO7Yp8KlPnJIYWTFkMaqPUYKTfGFPhxNuwnnxkKlk=
-github.com/PaesslerAG/gval v1.0.0/go.mod h1:y/nm5yEyTeX6av0OfKJNp9rBNj2XrGhAf5+v24IBN1I=
-github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
-github.com/PaesslerAG/jsonpath v0.1.1/go.mod h1:lVboNxFGal/VwW6d9JzIy56bUsYAP6tH/x80vjnCseY=
-github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
-github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
-github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
-github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
-github.com/benbjohnson/clock v1.0.0 h1:78Jk/r6m4wCi6sndMpty7A//t4dw/RW5fV4ZgDVfX1w=
-github.com/benbjohnson/clock v1.0.0/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM=
-github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
-github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
-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/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/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
-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/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=
-github.com/edgexfoundry/go-mod-messaging/v2 v2.0.1/go.mod h1:bLKWB9yeOHLZoQtHLZlGwz8MjsMJIvHDFce7CcUb4fE=
-github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
-github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
-github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
-github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
-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/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-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
-github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
-github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
-github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
-github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
-github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
-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-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
-github.com/go-yaml/yaml v2.1.0+incompatible h1:RYi2hDdss1u4YE7GwixGzWwVo47T8UQwnTLB6vQiq+o=
-github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=
-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=
-github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
-github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
-github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
-github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
-github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
-github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
-github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
-github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
-github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
-github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
-github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
-github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
-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.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=
-github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gordonklaus/ineffassign v0.0.0-20200309095847-7953dde2c7bf/go.mod h1:cuNKsD1zp2v6XfE/orVX2QE1LC+i254ceGcVeDT3pTU=
-github.com/gorilla/handlers v1.4.2/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/OAirnOIQ=
-github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
-github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-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/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
-github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
-github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
-github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869 h1:IPJ3dvxmJ4uczJe5YQdrYB16oTJlGSC/OyZDqUk9xX4=
-github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869/go.mod h1:cJ6Cj7dQo+O6GJNiMx+Pa94qKj+TG8ONdKHgMNIyyag=
-github.com/jhump/protoreflect v1.8.2/go.mod h1:7GcYQDdMU/O/BBrl/cX6PNHpXh6cenjd8pneu5yW7Tg=
-github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
-github.com/jonboulle/clockwork v0.2.2/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
-github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
-github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
-github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
-github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
-github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1 h1:JL2rWnBX8jnbHHlLcLde3BBWs+jzqZvOmF+M3sXoNOE=
-github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1/go.mod h1:nNLjpEi4xVFB7358xLPpPscdvXP+pbhiHgSmjIur8z0=
-github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
-github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
-github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
-github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
-github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
-github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
-github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
-github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc h1:RKf14vYWi2ttpEmkA4aQ3j4u9dStX2t4M8UM6qqNsG8=
-github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc/go.mod h1:kopuH9ugFRkIXf3YoqHKyrJ9YfUFsckUU9S7B+XP+is=
-github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible h1:Y6sqxHMyB1D2YSzWkLibYKgg+SwmyFU9dF2hn6MdTj4=
-github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible/go.mod h1:ZQnN8lSECaebrkQytbHj4xNgtg8CR7RYXnPok8e0EHA=
-github.com/lestrrat-go/strftime v1.0.3 h1:qqOPU7y+TM8Y803I8fG9c/DyKG3xH/xkng6keC1015Q=
-github.com/lestrrat-go/strftime v1.0.3/go.mod h1:E1nN3pCbtMSu1yjSVeyuRFVm/U0xoR76fd03sz+Qz4g=
-github.com/mattn/go-sqlite3 v1.14.5 h1:1IdxlwTNazvbKJQSxoJ5/9ECbEeaTTyeU7sEAZ5KKTQ=
-github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI=
-github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
-github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFWgYaq1OVrnRcwhnw=
-github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
-github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
-github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
-github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
-github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
-github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/msgpack-rpc/msgpack-rpc-go v0.0.0-20131026060856-c76397e1782b/go.mod h1:YUWGR1lA7NRnzRjOmYpSgNVip20OCp6NMBwiVFPPhZ8=
-github.com/msgpack/msgpack-go v0.0.0-20130625150338-8224460e6fa3/go.mod h1:jDCQZQaHCHpBYqM4WoGyujFc55bazGAEwK27iK4PQTI=
-github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/nishanths/predeclared v0.0.0-20200524104333-86fad755b4d3/go.mod h1:nt3d53pc1VYcphSCIaYAJtnPYnr3Zyn8fMq2wvPGPso=
-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/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
-github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
-github.com/pebbe/zmq4 v1.2.7/go.mod h1:nqnPueOapVhE2wItZ0uOErngczsJdLOGkebMxaO8r48=
-github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
-github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
-github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
-github.com/prometheus/client_golang v1.2.1/go.mod h1:XMU6Z2MjaRKVu/dC1qupJI9SiNkDYzz3xecMgSW/F+U=
-github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
-github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
-github.com/prometheus/common v0.7.0/go.mod h1:DjGbpBbp5NYNiECxcL/VnbXCCaQpKd3tt26CguLLsqA=
-github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
-github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
-github.com/prometheus/procfs v0.0.5/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ=
-github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
-github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
-github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
-github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
-github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
-github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
-github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
-github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
-github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
-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.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
-github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/tebeka/strftime v0.1.5 h1:1NQKN1NiQgkqd/2moD6ySP/5CoZQsKa1d3ZhJ44Jpmg=
-github.com/tebeka/strftime v0.1.5/go.mod h1:29/OidkoWHdEKZqzyDLUyC+LmgDgdHo4WAFCDT7D/Ig=
-github.com/ugorji/go v1.2.5/go.mod h1:gat2tIT8KJG8TVI8yv77nEO/KYT6dV7JE1gfUa8Xuls=
-github.com/ugorji/go/codec v1.2.5/go.mod h1:QPxoTbPKSEAlAHPYt02++xp/en9B/wUdwFCz+hj5caA=
-github.com/urfave/cli v1.22.0/go.mod h1:b3D7uWrF2GilkNgYpgcg6J+JMUw7ehmNkE8sZdliGLc=
-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=
-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=
-golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20200414173820-0848c9571904/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
-golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
-golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
-golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
-golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.3.0/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-20181114220301-adae6a3d119a/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=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20190620200207-3b0461eec859/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-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-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
-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=
-golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-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-20181116152217-5ac8a444bdc5/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-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-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-20200323222414-85ca7c5b95cd/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/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=
-golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
-golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
-golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
-golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20191130070609-6e064ea0cf2d/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20200522201501-cb1345f3a375/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
-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/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=
-google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
-google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
-google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
-google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
-google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
-google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
-google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
-google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
-google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
-google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
-google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
-google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
-google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
-google.golang.org/protobuf v1.25.1-0.20200805231151-a709e31e5d12/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
-google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
-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 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-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/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.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=
-honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
-honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
-honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
diff --git a/tools/migration/util/migration.go b/tools/migration/util/migration.go
deleted file mode 100644
index 620dab67..00000000
--- a/tools/migration/util/migration.go
+++ /dev/null
@@ -1,123 +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 util
-
-import (
-	"fmt"
-	"github.com/lf-edge/ekuiper/pkg/kv"
-	"github.com/patrickmn/go-cache"
-	"io/ioutil"
-	"os"
-	"path"
-)
-
-func migration(dir string) error {
-	fpath := path.Join(dir, "stores.data")
-	c := cache.New(cache.NoExpiration, 0)
-	if err := gocacheOpen(c, fpath); nil != err {
-		return err
-	}
-	defer gocacheClose(c, fpath)
-
-	keys, err := gocacheKeys(c)
-	if nil != err {
-		return err
-	}
-
-	err, store := kv.GetKVStore(dir)
-	if err != nil {
-		return err
-	}
-
-	for _, k := range keys {
-		if value, ok := c.Get(k); !ok {
-			return fmt.Errorf("not found %s from %s", k, fpath)
-		} else {
-			if err := store.Setnx(k, value); nil != err {
-				return err
-			}
-			if err := gocacheDel(c, k); nil != err {
-				return err
-			}
-		}
-	}
-	return os.Remove(fpath)
-}
-
-func DataMigration(dir string) error {
-	var dirs []string
-	dirs = append(dirs, dir)
-	for i := 0; i < len(dirs); i++ {
-		files, err := ioutil.ReadDir(dirs[i])
-		if nil != err {
-			return err
-		}
-		for _, file := range files {
-			fname := file.Name()
-			if file.IsDir() {
-				dirs = append(dirs, path.Join(dirs[i], fname))
-			} else if "stores.data" == fname {
-				return migration(dirs[i])
-			}
-		}
-	}
-	kv.CloseAll()
-	return nil
-}
-
-func gocacheClose(c *cache.Cache, path string) error {
-	if e := c.SaveFile(path); e != nil {
-		return e
-	}
-	c.Flush()
-	return nil
-}
-
-func gocacheOpen(c *cache.Cache, path string) error {
-	if _, err := os.Stat(path); os.IsNotExist(err) {
-		return nil
-	}
-	if e := c.LoadFile(path); e != nil {
-		return e
-	}
-	return nil
-}
-
-func gocacheKeys(c *cache.Cache) (keys []string, err error) {
-	if c == nil {
-		return nil, fmt.Errorf("cache has not been initialized yet.")
-	}
-	its := c.Items()
-	keys = make([]string, 0, len(its))
-	for k := range its {
-		keys = append(keys, k)
-	}
-	return keys, nil
-}
-
-func gocacheSet(c *cache.Cache, path, key string, value interface{}) error {
-	if c == nil {
-		return fmt.Errorf("cache has not been initialized yet.")
-	}
-	return c.Add(key, value, cache.NoExpiration)
-}
-
-func gocacheDel(c *cache.Cache, key string) error {
-	if c == nil {
-		return fmt.Errorf("cache has not been initialized yet.")
-	}
-	c.Delete(key)
-	return nil
-}


Test patch:
diff --git a/internal/pkg/sqlkv/sqlKv_test.go b/internal/pkg/sqlkv/sqlKv_test.go
deleted file mode 100644
index 538dac59..00000000
--- a/internal/pkg/sqlkv/sqlKv_test.go
+++ /dev/null
@@ -1,107 +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 sqlkv
-
-import (
-	"os"
-	"path"
-	"path/filepath"
-	"reflect"
-	"testing"
-)
-
-func TestSqlKVStore_Funcs(t *testing.T) {
-	abs, _ := filepath.Abs("test")
-	if f, _ := os.Stat(abs); f != nil {
-		os.Remove(abs)
-	}
-	_, db := newSqliteDatabase(abs)
-	db.Connect()
-	database = db
-
-	ks, _ := GetKVStore("test")
-	if err := ks.Setnx("foo", "bar"); nil != err {
-		t.Error(err)
-	}
-
-	var v string
-	if ok, _ := ks.Get("foo", &v); ok {
-		if !reflect.DeepEqual("bar", v) {
-			t.Error("expect:bar", "get:", v)
-		}
-	} else {
-		t.Errorf("Should not find the foo key.")
-	}
-
-	if err := ks.Setnx("foo1", "bar1"); nil != err {
-		t.Error(err)
-	}
-
-	if err := ks.Set("foo1", "bar2"); nil != err {
-		t.Error(err)
-	}
-
-	var v1 string
-	if ok, _ := ks.Get("foo1", &v1); ok {
-		if !reflect.DeepEqual("bar2", v1) {
-			t.Error("expect:bar2", "get:", v1)
-		}
-	} else {
-		t.Errorf("Should not find the foo1 key.")
-	}
-
-	if keys, e1 := ks.Keys(); e1 != nil {
-		t.Errorf("Failed to get value: %s.", e1)
-	} else {
-		if !reflect.DeepEqual(2, len(keys)) {
-			t.Error("expect:2", "get:", len(keys))
-		}
-	}
-
-	var v2 string
-	if ok, _ := ks.Get("foo", &v2); ok {
-		if !reflect.DeepEqual("bar", v2) {
-			t.Error("expect:bar", "get:", v)
-		}
-	} else {
-		t.Errorf("Should not find the foo key.")
-	}
-
-	if err := ks.Delete("foo1"); nil != err {
-		t.Error(err)
-	}
-
-	if keys, e1 := ks.Keys(); e1 != nil {
-		t.Errorf("Failed to get value: %s.", e1)
-	} else {
-		reflect.DeepEqual(1, len(keys))
-	}
-
-	if err := ks.Clean(); nil != err {
-		t.Error(err)
-	}
-
-	if keys, e1 := ks.Keys(); e1 != nil {
-		t.Errorf("Failed to get value: %s.", e1)
-	} else {
-		reflect.DeepEqual(0, len(keys))
-	}
-
-	database.Disconnect()
-	dir, _ := filepath.Split(abs)
-	abs = path.Join(dir, "sqliteKV.db")
-	os.Remove(abs)
-
-}
diff --git a/internal/pkg/store/redisKv_test.go b/internal/pkg/store/redisKv_test.go
new file mode 100644
index 00000000..63118fe6
--- /dev/null
+++ b/internal/pkg/store/redisKv_test.go
@@ -0,0 +1,83 @@
+// 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 store
+
+import (
+	"github.com/alicebob/miniredis/v2"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	rb "github.com/lf-edge/ekuiper/internal/pkg/store/redis"
+	"github.com/lf-edge/ekuiper/internal/pkg/store/test/common"
+	"github.com/lf-edge/ekuiper/pkg/kv"
+	"strconv"
+	"testing"
+)
+
+func TestRedisKvSetnx(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+	common.TestKvSetnx(ks, t)
+}
+
+func TestRedisKvSet(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+	common.TestKvSet(ks, t)
+}
+
+func TestRedisKvGet(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+	common.TestKvGet(ks, t)
+}
+
+func TestRedisKvKeys(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+
+	length := 10
+	common.TestKvKeys(length, ks, t)
+}
+
+func setupRedisKv() (kv.KeyValue, *redis.Instance, *miniredis.Miniredis) {
+	minRedis, err := miniredis.Run()
+	if err != nil {
+		panic(err)
+	}
+	redisDB := redis.NewRedis("localhost", stringToInt(minRedis.Port()))
+	err = redisDB.Connect()
+	if err != nil {
+		panic(err)
+	}
+	builder := rb.NewStoreBuilder(redisDB)
+	var ks kv.KeyValue
+	ks, err = builder.CreateStore("test")
+	if err != nil {
+		panic(err)
+	}
+	return ks, &redisDB, minRedis
+}
+
+func cleanRedisKv(instance *redis.Instance, minRedis *miniredis.Miniredis) {
+	instance.Disconnect()
+	minRedis.Close()
+}
+
+func stringToInt(svalue string) int {
+	ivalue, err := strconv.Atoi(svalue)
+	if err != nil {
+		panic(err)
+	}
+	return ivalue
+}
diff --git a/internal/pkg/store/sqlKv_test.go b/internal/pkg/store/sqlKv_test.go
new file mode 100644
index 00000000..ded2c64a
--- /dev/null
+++ b/internal/pkg/store/sqlKv_test.go
@@ -0,0 +1,100 @@
+// 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 store
+
+import (
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql/sqlite"
+	sb "github.com/lf-edge/ekuiper/internal/pkg/store/sql"
+	"github.com/lf-edge/ekuiper/internal/pkg/store/test/common"
+	"github.com/lf-edge/ekuiper/pkg/kv"
+	"os"
+	"path"
+	"path/filepath"
+	"testing"
+)
+
+const DbName = "sqliteKV.db"
+const Table = "test"
+
+func TestSqlKvSetnx(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+	common.TestKvSetnx(ks, t)
+}
+
+func TestSqlKvSet(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+	common.TestKvSet(ks, t)
+}
+
+func TestSqlKvGet(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+	common.TestKvGet(ks, t)
+}
+
+func TestSqlKvKeys(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+
+	length := 10
+	common.TestKvKeys(length, ks, t)
+}
+
+func deleteIfExists(abs string) error {
+	absPath := path.Join(abs, DbName)
+	if f, _ := os.Stat(absPath); f != nil {
+		return os.Remove(absPath)
+	}
+	return nil
+}
+
+func setupSqlKv() (kv.KeyValue, *sqlite.Database, string) {
+	absPath, err := filepath.Abs("test")
+	if err != nil {
+		panic(err)
+	}
+	err = deleteIfExists(absPath)
+	if err != nil {
+		panic(err)
+	}
+	config := sqlite.Config{
+		Path: absPath,
+		Name: DbName,
+	}
+	_, db := sqlite.NewSqliteDatabase(config)
+	err = db.Connect()
+	if err != nil {
+		panic(err)
+	}
+
+	builder := sb.NewStoreBuilder(db)
+	var store kv.KeyValue
+	store, err = builder.CreateStore(Table)
+	if err != nil {
+		panic(err)
+	}
+	return store, db, absPath
+}
+
+func cleanSqlKv(db *sqlite.Database, abs string) {
+	if err := db.Disconnect(); err != nil {
+		panic(err)
+	}
+	if err := deleteIfExists(abs); err != nil {
+		panic(err)
+	}
+}
diff --git a/internal/pkg/store/test/common/test.go b/internal/pkg/store/test/common/test.go
new file mode 100644
index 00000000..e22c94da
--- /dev/null
+++ b/internal/pkg/store/test/common/test.go
@@ -0,0 +1,84 @@
+// 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 common
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/kv"
+	"reflect"
+	"testing"
+)
+
+func TestKvSetnx(ks kv.KeyValue, t *testing.T) {
+
+	if err := ks.Setnx("foo", "bar"); nil != err {
+		t.Error(err)
+	}
+
+	if err := ks.Setnx("foo", "bar1"); nil == err {
+		t.Errorf("Can't overwrite an existing intem")
+	}
+}
+
+func TestKvSet(ks kv.KeyValue, t *testing.T) {
+
+	if err := ks.Set("foo", "bar"); nil != err {
+		t.Error(err)
+	}
+
+	if err := ks.Set("foo", "bar1"); nil != err {
+		t.Errorf("Set should overwrite an existing record")
+	}
+}
+
+func TestKvGet(ks kv.KeyValue, t *testing.T) {
+
+	if err := ks.Setnx("foo", "bar"); nil != err {
+		t.Error(err)
+	}
+
+	var v string
+	if ok, _ := ks.Get("foo", &v); ok {
+		if !reflect.DeepEqual("bar", v) {
+			t.Error("expect:bar", "get:", v)
+		}
+	} else {
+		t.Errorf("Should find the foo key")
+	}
+}
+
+func TestKvKeys(length int, ks kv.KeyValue, t *testing.T) {
+
+	expected := make([]string, 0)
+	for i := 0; i < length; i++ {
+		key := fmt.Sprintf("key-%d", i)
+		value := fmt.Sprintf("value-%d", i)
+		if err := ks.Setnx(key, value); err != nil {
+			t.Errorf("It should be set")
+		}
+		expected = append(expected, key)
+	}
+
+	var keys []string
+	var err error
+	if keys, err = ks.Keys(); err != nil {
+		t.Errorf("Failed to get value: %s.", err)
+	} else if !reflect.DeepEqual(length, len(keys)) {
+		t.Errorf("expect: %d, got: %d", length, len(keys))
+	}
+	if !reflect.DeepEqual(keys, expected) {
+		t.Errorf("Keys do not match expected %s != %s", keys, expected)
+	}
+}
diff --git a/internal/pkg/ts/redis/redisTs_test.go b/internal/pkg/ts/redis/redisTs_test.go
new file mode 100644
index 00000000..a088dc9d
--- /dev/null
+++ b/internal/pkg/ts/redis/redisTs_test.go
@@ -0,0 +1,92 @@
+// 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 redis
+
+import (
+	"github.com/alicebob/miniredis/v2"
+	"github.com/lf-edge/ekuiper/internal/pkg/db/redis"
+	"github.com/lf-edge/ekuiper/internal/pkg/ts/test/common"
+	ts2 "github.com/lf-edge/ekuiper/pkg/kv"
+	"strconv"
+	"testing"
+)
+
+func TestRedisTsSet(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+
+	common.TestTsSet(ks, t)
+}
+
+func TestRedisTsLast(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+
+	common.TestTsLast(ks, t)
+}
+
+func TestRedisTsGet(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+
+	common.TestTsGet(ks, t)
+}
+
+func TestRedisTsDelete(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+
+	common.TestTsDelete(ks, t)
+}
+
+func TestRedisTsDeleteBefore(t *testing.T) {
+	ks, db, minRedis := setupRedisKv()
+	defer cleanRedisKv(db, minRedis)
+
+	common.TestTsDeleteBefore(ks, t)
+}
+
+func setupRedisKv() (ts2.Tskv, *redis.Instance, *miniredis.Miniredis) {
+	minRedis, err := miniredis.Run()
+	if err != nil {
+		panic(err)
+	}
+	redisDB := redis.NewRedis("localhost", stringToInt(minRedis.Port()))
+	err = redisDB.Connect()
+	if err != nil {
+		panic(err)
+	}
+
+	builder := NewTsBuilder(redisDB)
+	var ks ts2.Tskv
+	err, ks = builder.CreateTs("test")
+	if err != nil {
+		panic(err)
+	}
+	return ks, &redisDB, minRedis
+}
+
+func cleanRedisKv(instance *redis.Instance, minRedis *miniredis.Miniredis) {
+	instance.Disconnect()
+	minRedis.Close()
+}
+
+func stringToInt(svalue string) int {
+	ivalue, err := strconv.Atoi(svalue)
+	if err != nil {
+		panic(err)
+	}
+	return ivalue
+}
diff --git a/internal/pkg/ts/sql/sqlTs_test.go b/internal/pkg/ts/sql/sqlTs_test.go
new file mode 100644
index 00000000..df3a4afe
--- /dev/null
+++ b/internal/pkg/ts/sql/sqlTs_test.go
@@ -0,0 +1,113 @@
+// 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 sql
+
+import (
+	"github.com/lf-edge/ekuiper/internal/pkg/db/sql/sqlite"
+	"github.com/lf-edge/ekuiper/internal/pkg/ts/test/common"
+	ts2 "github.com/lf-edge/ekuiper/pkg/kv"
+	"os"
+	"path"
+	"path/filepath"
+	"testing"
+)
+
+const (
+	DbName = "sqliteTS.db"
+	Table  = "test"
+)
+
+func TestSqlTsSet(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+
+	common.TestTsSet(ks, t)
+}
+
+func TestSqlTsLast(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+
+	common.TestTsLast(ks, t)
+}
+
+func TestSqlTsGet(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+
+	common.TestTsGet(ks, t)
+}
+
+func TestSqlTsDelete(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+
+	common.TestTsDelete(ks, t)
+}
+
+func TestSqlTsDeleteBefore(t *testing.T) {
+	ks, db, abs := setupSqlKv()
+	defer cleanSqlKv(db, abs)
+
+	common.TestTsDeleteBefore(ks, t)
+}
+
+func deleteIfExists(abs string) error {
+	absPath := path.Join(abs, DbName)
+	if f, _ := os.Stat(absPath); f != nil {
+		return os.Remove(absPath)
+	}
+	return nil
+}
+
+func setupSqlKv() (ts2.Tskv, *sqlite.Database, string) {
+	absPath, err := filepath.Abs("test")
+	if err != nil {
+		panic(err)
+	}
+	err = deleteIfExists(absPath)
+	if err != nil {
+		panic(err)
+	}
+	config := sqlite.Config{
+		Path: absPath,
+		Name: DbName,
+	}
+	_, db := sqlite.NewSqliteDatabase(config)
+	err = db.Connect()
+	if err != nil {
+		panic(err)
+	}
+
+	builder := NewTsBuilder(db)
+	if err != nil {
+		panic(err)
+	}
+	var store ts2.Tskv
+	err, store = builder.CreateTs(Table)
+	if err != nil {
+		panic(err)
+	}
+	return store, db, absPath
+}
+
+func cleanSqlKv(db *sqlite.Database, abs string) {
+	if err := db.Disconnect(); err != nil {
+		panic(err)
+	}
+	if err := deleteIfExists(abs); err != nil {
+		panic(err)
+	}
+}
diff --git a/internal/pkg/tskv/sqlite_test.go b/internal/pkg/ts/test/common/test.go
similarity index 53%
rename from internal/pkg/tskv/sqlite_test.go
rename to internal/pkg/ts/test/common/test.go
index f1288786..69b642e7 100644
--- a/internal/pkg/tskv/sqlite_test.go
+++ b/internal/pkg/ts/test/common/test.go
@@ -12,49 +12,31 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package tskv
+package common
 
 import (
+	"github.com/lf-edge/ekuiper/pkg/kv"
 	"reflect"
 	"testing"
 )
 
-func TestSqlite_Funcs(t *testing.T) {
-	ks, e := NewSqlite("test")
-	if e != nil {
-		t.Errorf("Failed to create tskv %s.", e)
-		return
-	}
-
-	if ok, err := ks.Set(1000, "bar1"); nil != err {
-		t.Error(err)
-	} else if !ok {
-		t.Error("should allow key 1000")
-	}
-
-	if ok, err := ks.Set(1500, "bar15"); nil != err {
-		t.Error(err)
-	} else if !ok {
-		t.Error("should allow key 1500")
-	}
+var (
+	Keys   = []int64{1000, 1500, 2000, 3000}
+	Values = []string{"bar1", "bar15", "bar2", "bar3"}
+)
 
-	if ok, err := ks.Set(2000, "bar2"); nil != err {
-		t.Error(err)
-	} else if !ok {
-		t.Error("should allow key 2000")
-	}
-
-	if ok, err := ks.Set(3000, "bar3"); nil != err {
-		t.Error(err)
-	} else if !ok {
-		t.Error("should allow key 3000")
-	}
+func TestTsSet(ks kv.Tskv, t *testing.T) {
+	load(ks, t)
 
 	if ok, err := ks.Set(2500, "bar25"); nil != err {
 		t.Error(err)
 	} else if ok {
-		t.Error("should deny key 2500")
+		t.Errorf("should deny key 2500 while last one is 3000")
 	}
+}
+
+func TestTsLast(ks kv.Tskv, t *testing.T) {
+	load(ks, t)
 
 	var v string
 	if k, err := ks.Last(&v); err != nil {
@@ -62,22 +44,36 @@ func TestSqlite_Funcs(t *testing.T) {
 	} else if k != 3000 || v != "bar3" {
 		t.Errorf("Last expect 3000/bar3 but got %d/%s", k, v)
 	}
+}
+
+func TestTsGet(ks kv.Tskv, t *testing.T) {
+	load(ks, t)
 
-	if ok, _ := ks.Get(2000, &v); ok {
-		if !reflect.DeepEqual("bar2", v) {
-			t.Error("expect:bar", "get:", v)
+	var value string
+	if ok, _ := ks.Get(2000, &value); ok {
+		if !reflect.DeepEqual("bar2", value) {
+			t.Error("expect:bar", "get:", value)
 		}
 	} else {
 		t.Errorf("Should find key 2000.")
 	}
+}
+
+func TestTsDelete(ks kv.Tskv, t *testing.T) {
+	load(ks, t)
 
 	if err := ks.Delete(1500); nil != err {
 		t.Error(err)
 	}
 
-	if ok, _ := ks.Get(1500, &v); ok {
+	var value string
+	if ok, _ := ks.Get(1500, &value); ok {
 		t.Errorf("Should not find deleted key 1500.")
 	}
+}
+
+func TestTsDeleteBefore(ks kv.Tskv, t *testing.T) {
+	load(ks, t)
 
 	if ok, err := ks.Set(3500, "bar35"); nil != err {
 		t.Error(err)
@@ -89,31 +85,39 @@ func TestSqlite_Funcs(t *testing.T) {
 		t.Error(err)
 	}
 
-	if ok, _ := ks.Get(1000, &v); ok {
+	var value string
+	if ok, _ := ks.Get(1000, &value); ok {
 		t.Errorf("Should not find deleted key 1000.")
 	}
-
-	if ok, _ := ks.Get(2000, &v); ok {
+	if ok, _ := ks.Get(2000, &value); ok {
 		t.Errorf("Should not find deleted key 2000.")
 	}
 
-	if ok, _ := ks.Get(3000, &v); ok {
-		if !reflect.DeepEqual("bar3", v) {
-			t.Error("expect:bar3", "get:", v)
+	if ok, _ := ks.Get(3000, &value); ok {
+		if !reflect.DeepEqual("bar3", value) {
+			t.Error("expect:bar3", "get:", value)
 		}
 	} else {
 		t.Errorf("Should find key 3000.")
 	}
 
-	if ok, _ := ks.Get(3500, &v); ok {
-		if !reflect.DeepEqual("bar35", v) {
-			t.Error("expect:bar35", "get:", v)
+	if ok, _ := ks.Get(3500, &value); ok {
+		if !reflect.DeepEqual("bar35", value) {
+			t.Error("expect:bar35", "get:", value)
 		}
 	} else {
 		t.Errorf("Should find key 3500.")
 	}
+}
 
-	if err := ks.Drop(); err != nil {
-		t.Error(err)
+func load(ks kv.Tskv, t *testing.T) {
+	for i := 0; i < len(Keys); i++ {
+		k := Keys[i]
+		v := Values[i]
+		if ok, err := ks.Set(k, v); nil != err {
+			t.Error(err)
+		} else if !ok {
+			t.Errorf("should allow key %d", k)
+		}
 	}
 }
diff --git a/internal/testx/testUtil.go b/internal/testx/testUtil.go
index f5229c7c..dd335d1c 100644
--- a/internal/testx/testUtil.go
+++ b/internal/testx/testUtil.go
@@ -16,7 +16,7 @@ package testx
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 )
 
 // errstring returns the string representation of an error.
@@ -27,12 +27,10 @@ func Errstring(err error) string {
 	return ""
 }
 
-func InitEnv() string {
+func InitEnv() {
 	conf.InitConf()
-	dbDir, err := conf.GetDataLoc()
+	err := store.SetupDefault()
 	if err != nil {
 		conf.Log.Fatal(err)
 	}
-	sqlkv.Setup(dbDir)
-	return dbDir
 }
diff --git a/internal/topo/context/default_test.go b/internal/topo/context/default_test.go
index c68d8401..a4321807 100644
--- a/internal/topo/context/default_test.go
+++ b/internal/topo/context/default_test.go
@@ -16,6 +16,7 @@ package context
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"log"
@@ -26,6 +27,10 @@ import (
 )
 
 func TestState(t *testing.T) {
+	err := store.SetupDefault()
+	if err != nil {
+		t.Error(err)
+	}
 	var (
 		i      = 0
 		ruleId = "testStateRule"
diff --git a/internal/topo/planner/analyzer_test.go b/internal/topo/planner/analyzer_test.go
index 398404dc..2606e53e 100644
--- a/internal/topo/planner/analyzer_test.go
+++ b/internal/topo/planner/analyzer_test.go
@@ -17,7 +17,7 @@ package planner
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
@@ -123,7 +123,7 @@ var tests = []struct {
 }
 
 func Test_validation(t *testing.T) {
-	store, err := sqlkv.GetKVStore("stream")
+	err, store := store.GetKV("stream")
 	if err != nil {
 		t.Error(err)
 		return
@@ -185,7 +185,7 @@ func Test_validation(t *testing.T) {
 }
 
 func Test_validationSchemaless(t *testing.T) {
-	store, err := sqlkv.GetKVStore("stream")
+	err, store := store.GetKV("stream")
 	if err != nil {
 		t.Error(err)
 		return
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index fe9fb3bd..a1cec4c1 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -18,7 +18,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/gdexlab/go-render/render"
-	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
@@ -33,7 +33,7 @@ func init() {
 }
 
 func Test_createLogicalPlan(t *testing.T) {
-	store, err := sqlkv.GetKVStore("stream")
+	err, store := store.GetKV("stream")
 	if err != nil {
 		t.Error(err)
 		return
@@ -1136,7 +1136,7 @@ func Test_createLogicalPlan(t *testing.T) {
 }
 
 func Test_createLogicalPlanSchemaless(t *testing.T) {
-	store, err := sqlkv.GetKVStore("stream")
+	err, store := store.GetKV("stream")
 	if err != nil {
 		t.Error(err)
 		return
diff --git a/internal/topo/state/kv_store_test.go b/internal/topo/state/kv_store_test.go
index 94aac1ae..521e8b22 100644
--- a/internal/topo/state/kv_store_test.go
+++ b/internal/topo/state/kv_store_test.go
@@ -17,6 +17,7 @@ package state
 import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"log"
 	"os"
@@ -153,6 +154,10 @@ func TestLifecycle(t *testing.T) {
 	)
 	func() {
 		cleanStateData()
+		err := store.SetupDefault()
+		if err != nil {
+			t.Error(err)
+		}
 		store, err := getKVStore(ruleId)
 		if err != nil {
 			t.Errorf("Get store for rule %s error: %s", ruleId, err)
diff --git a/tools/migration/util/migration_test.go b/tools/migration/util/migration_test.go
deleted file mode 100644
index aabf17d2..00000000
--- a/tools/migration/util/migration_test.go
+++ /dev/null
@@ -1,81 +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 util
-
-import (
-	"github.com/lf-edge/ekuiper/pkg/kv"
-	"github.com/patrickmn/go-cache"
-	"os"
-	"path"
-	"path/filepath"
-	"testing"
-)
-
-func TestDataMigration(t *testing.T) {
-	kvs := make(map[string]string)
-	kvs["mqtt"] = `{"sql":"create stream mqtt(age BIGINT) WITH (DATASOURCE = \"dev/+/msg\", FORMAT = \"json\");"}`
-	kvs["log"] = `{"id":"log","sql":"SELECT  * FROM mqtt","actions":[{"log":{}}]}`
-
-	dir, _ := filepath.Abs("testMigration")
-	dirSqlite, _ := filepath.Split(dir)
-	fpath := path.Join(dir, "stores.data")
-	if f, _ := os.Stat(fpath); f != nil {
-		os.Remove(fpath)
-	}
-	if _, err := os.Stat(dir); os.IsNotExist(err) {
-		os.MkdirAll(dir, os.ModePerm)
-	}
-	defer os.RemoveAll(dir)
-	defer os.Remove(path.Join(dirSqlite, "sqliteKV.db"))
-
-	c := cache.New(cache.NoExpiration, 0)
-	if err := gocacheOpen(c, fpath); nil != err {
-		t.Error(err)
-		return
-	}
-
-	for k, v := range kvs {
-		if err := gocacheSet(c, fpath, k, v); nil != err {
-			t.Error(err)
-			return
-		}
-	}
-	if err := gocacheClose(c, fpath); nil != err {
-		t.Error(err)
-	}
-
-	if err := DataMigration(dir); nil != err {
-		t.Error(err)
-		return
-	}
-
-	store := kv.GetDefaultKVStore(dir)
-	if err := store.Open(); nil != err {
-		t.Error(err)
-		return
-	}
-	defer store.Close()
-	for k, v := range kvs {
-		var dbVal string
-		if ok, _ := store.Get(k, &dbVal); !ok {
-			t.Error("not found key ", k)
-			return
-		} else if v != dbVal {
-			t.Error("gocache save:", v, "sqlite save:", dbVal)
-			return
-		}
-	}
-
-}