Repository: lf-edge/ekuiper
Pull request #921 (2021-08-04 12:44:07)

fix,refactor(sqlite): Fix (and refactoring) related to the too many files open issue
Fix (and refactoring) related to the too many files open issue - since sqlite does not work well with concurrent writes - kv store was rewritten to use single connection for access to the database,
* introduced database structure which will take away sql database specifics (sqlite in this case)
* fix in topo.go - there was possible scenario that during Open execution of underlying goroutine Cancel was called and store property was set to nil which was leading to null pointer exception

ngjaying: @rwadowski Awesome!
You need to signoff. Please click into the detail of DCO check and follow the instruction to pass the check.
rwadowski: @ngjaying  what are fvt tests ? How can I run them locally ?

Patch:
diff --git a/internal/pkg/sqlkv/database.go b/internal/pkg/sqlkv/database.go
new file mode 100644
index 00000000..c68c7629
--- /dev/null
+++ b/internal/pkg/sqlkv/database.go
@@ -0,0 +1,23 @@
+// 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 "database/sql"
+
+type Database interface {
+	Connect() error
+	Disconnect() error
+	Apply(f func(db *sql.DB) error) error
+}
diff --git a/internal/pkg/sqlkv/kvStore.go b/internal/pkg/sqlkv/kvStore.go
new file mode 100644
index 00000000..ad458371
--- /dev/null
+++ b/internal/pkg/sqlkv/kvStore.go
@@ -0,0 +1,70 @@
+// 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/sqlkv/sqlKv.go b/internal/pkg/sqlkv/sqlKv.go
new file mode 100644
index 00000000..1674dddd
--- /dev/null
+++ b/internal/pkg/sqlkv/sqlKv.go
@@ -0,0 +1,159 @@
+// 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 (
+	"bytes"
+	"database/sql"
+	"encoding/gob"
+	"fmt"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"strings"
+)
+
+type sqlKvStore struct {
+	database Database
+	table    string
+}
+
+func CreateSqlKvStore(database Database, table string) (error, *sqlKvStore) {
+	store := &sqlKvStore{
+		database: database,
+		table:    table,
+	}
+	err := store.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS '%s'('key' VARCHAR(255) PRIMARY KEY, 'val' BLOB);", table)
+		_, err := db.Exec(query)
+		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
+}
+
+func (kv *sqlKvStore) Setnx(key string, value interface{}) error {
+	return kv.database.Apply(func(db *sql.DB) error {
+		b, err := encode(value)
+		if nil != err {
+			return err
+		}
+		query := fmt.Sprintf("INSERT INTO '%s'(key,val) values(?,?);", kv.table)
+		stmt, err := db.Prepare(query)
+		_, err = stmt.Exec(key, b)
+		stmt.Close()
+		if err != nil {
+			if strings.Contains(err.Error(), "UNIQUE constraint failed") {
+				return fmt.Errorf(`Item %s already exists`, key)
+			}
+		}
+		return err
+	})
+}
+
+func (kv *sqlKvStore) Set(key string, value interface{}) error {
+	b, err := encode(value)
+	if nil != err {
+		return err
+	}
+	err = kv.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("REPLACE INTO '%s'(key,val) values(?,?);", kv.table)
+		stmt, err := db.Prepare(query)
+		if err != nil {
+			return err
+		}
+		_, err = stmt.Exec(key, b)
+		stmt.Close()
+		return err
+	})
+	return err
+}
+
+func (kv *sqlKvStore) Get(key string, value interface{}) (bool, error) {
+	result := false
+	err := kv.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("SELECT val FROM '%s' WHERE key='%s';", kv.table, key)
+		row := db.QueryRow(query)
+		var tmp []byte
+		err := row.Scan(&tmp)
+		if err != nil {
+			result = false
+			return nil
+		}
+		dec := gob.NewDecoder(bytes.NewBuffer(tmp))
+		if err := dec.Decode(value); err != nil {
+			return err
+		}
+		result = true
+		return nil
+	})
+	return result, err
+}
+
+func (kv *sqlKvStore) Delete(key string) error {
+	return kv.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("SELECT key FROM '%s' WHERE key='%s';", kv.table, key)
+		row := db.QueryRow(query)
+		var tmp []byte
+		err := row.Scan(&tmp)
+		if nil != err || 0 == len(tmp) {
+			return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("%s is not found", key))
+		}
+		query = fmt.Sprintf("DELETE FROM '%s' WHERE key='%s';", kv.table, key)
+		_, err = db.Exec(query)
+		return err
+	})
+}
+
+func (kv *sqlKvStore) Keys() ([]string, error) {
+	keys := make([]string, 0)
+	err := kv.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("SELECT key FROM '%s'", kv.table)
+		row, err := db.Query(query)
+		if nil != err {
+			return err
+		}
+		defer row.Close()
+		for row.Next() {
+			var val string
+			err = row.Scan(&val)
+			if nil != err {
+				return err
+			} else {
+				keys = append(keys, val)
+			}
+		}
+		return nil
+	})
+	return keys, err
+}
+
+func (kv *sqlKvStore) Clean() error {
+	return kv.database.Apply(func(db *sql.DB) error {
+		query := fmt.Sprintf("DELETE FROM '%s'", kv.table)
+		_, err := db.Exec(query)
+		return err
+	})
+}
diff --git a/internal/pkg/sqlkv/sqliteDatabase.go b/internal/pkg/sqlkv/sqliteDatabase.go
new file mode 100644
index 00000000..6be1e0fc
--- /dev/null
+++ b/internal/pkg/sqlkv/sqliteDatabase.go
@@ -0,0 +1,70 @@
+// 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 (
+	"database/sql"
+	"fmt"
+	_ "github.com/mattn/go-sqlite3"
+	"os"
+	"path"
+	"sync"
+)
+
+type SqliteDatabase struct {
+	db   *sql.DB
+	Path string
+	mu   sync.Mutex
+}
+
+func newSqliteDatabase(dir string) (error, *SqliteDatabase) {
+	if _, err := os.Stat(dir); os.IsNotExist(err) {
+		os.MkdirAll(dir, os.ModePerm)
+	}
+	dbPath := path.Join(dir, "sqliteKV.db")
+	return nil, &SqliteDatabase{
+		db:   nil,
+		Path: dbPath,
+		mu:   sync.Mutex{},
+	}
+}
+
+func (d *SqliteDatabase) Connect() error {
+	db, err := sql.Open("sqlite3", connectionString(d.Path))
+	if err != nil {
+		return err
+	}
+	db.SetMaxIdleConns(1)
+	db.SetMaxOpenConns(1)
+	db.SetConnMaxLifetime(-1)
+	d.db = db
+	return nil
+}
+
+func connectionString(dpath string) string {
+	return fmt.Sprintf("file:%s?cache=shared", dpath)
+}
+
+func (d *SqliteDatabase) Disconnect() error {
+	err := d.db.Close()
+	return err
+}
+
+func (d *SqliteDatabase) Apply(f func(db *sql.DB) error) error {
+	d.mu.Lock()
+	err := f(d.db)
+	d.mu.Unlock()
+	return err
+}
diff --git a/internal/plugin/manager.go b/internal/plugin/manager.go
index 4c1b0805..af27fd01 100644
--- a/internal/plugin/manager.go
+++ b/internal/plugin/manager.go
@@ -22,6 +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/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/lf-edge/ekuiper/pkg/kv"
@@ -294,17 +295,10 @@ func NewPluginManager() (*Manager, error) {
 			outerErr = fmt.Errorf("cannot find etc folder: %s", err)
 			return
 		}
-		dbDir, err := conf.GetDataLoc()
-		if err != nil {
-			outerErr = fmt.Errorf("cannot find db folder: %s", err)
-			return
-		}
-		db := kv.GetDefaultKVStore(path.Join(dbDir, "pluginFuncs"))
-		err = db.Open()
+		db, err := sqlkv.GetKVStore("pluginFuncs")
 		if err != nil {
 			outerErr = fmt.Errorf("error when opening db: %v.", err)
 		}
-		defer db.Close()
 		plugins := make([]map[string]string, 3)
 		for i := 0; i < 3; i++ {
 			names, err := findAll(PluginType(i), dir)
@@ -400,15 +394,10 @@ func (m *Manager) Register(t PluginType, j Plugin) error {
 	var err error
 	if t == FUNCTION {
 		if len(j.GetSymbols()) > 0 {
-			err = m.db.Open()
-			if err != nil {
-				return err
-			}
 			err = m.db.Set(name, j.GetSymbols())
 			if err != nil {
 				return err
 			}
-			m.db.Close()
 			err = m.registry.StoreSymbols(name, j.GetSymbols())
 		} else {
 			err = m.registry.StoreSymbols(name, []string{name})
@@ -430,16 +419,13 @@ func (m *Manager) Register(t PluginType, j Plugin) error {
 	//unzip and copy to destination
 	unzipFiles, version, err := m.install(t, name, zipPath, shellParas)
 	if err == nil && len(j.GetSymbols()) > 0 {
-		if err = m.db.Open(); err == nil {
-			err = m.db.Set(name, j.GetSymbols())
-		}
+		err = m.db.Set(name, j.GetSymbols())
 	}
 	if err != nil { //Revert for any errors
 		if t == SOURCE && len(unzipFiles) == 1 { //source that only copy so file
 			os.RemoveAll(unzipFiles[0])
 		}
 		if len(j.GetSymbols()) > 0 {
-			m.db.Close()
 			m.registry.RemoveSymbols(j.GetSymbols())
 		} else {
 			m.registry.RemoveSymbols([]string{name})
@@ -470,11 +456,6 @@ func (m *Manager) RegisterFuncs(name string, functions []string) error {
 	if len(functions) == 0 {
 		return fmt.Errorf("property 'functions' must not be empty")
 	}
-	err := m.db.Open()
-	if err != nil {
-		return err
-	}
-	defer m.db.Close()
 	old := make([]string, 0)
 	if ok, err := m.db.Get(name, &old); err != nil {
 		return err
@@ -483,7 +464,7 @@ func (m *Manager) RegisterFuncs(name string, functions []string) error {
 	} else if !ok {
 		m.registry.RemoveSymbols([]string{name})
 	}
-	err = m.db.Set(name, functions)
+	err := m.db.Set(name, functions)
 	if err != nil {
 		return err
 	}
@@ -518,10 +499,6 @@ func (m *Manager) Delete(t PluginType, name string, stop bool) error {
 		m.uninstalSink(name)
 	case FUNCTION:
 		old := make([]string, 0)
-		err = m.db.Open()
-		if err != nil {
-			return err
-		}
 		if ok, err := m.db.Get(name, &old); err != nil {
 			return err
 		} else if ok {
@@ -533,7 +510,6 @@ func (m *Manager) Delete(t PluginType, name string, stop bool) error {
 		} else if !ok {
 			m.registry.RemoveSymbols([]string{name})
 		}
-		m.db.Close()
 		m.uninstalFunc(name)
 	}
 
@@ -573,12 +549,9 @@ func (m *Manager) Get(t PluginType, name string) (map[string]interface{}, bool)
 			"version": v,
 		}
 		if t == FUNCTION {
-			if err := m.db.Open(); err == nil {
-				l := make([]string, 0)
-				if ok, _ := m.db.Get(name, &l); ok {
-					r["functions"] = l
-				}
-				m.db.Close()
+			l := make([]string, 0)
+			if ok, _ := m.db.Get(name, &l); ok {
+				r["functions"] = l
 			}
 			// ignore the error
 		}
diff --git a/internal/processor/rule.go b/internal/processor/rule.go
index cd2f9294..f6ead577 100644
--- a/internal/processor/rule.go
+++ b/internal/processor/rule.go
@@ -19,6 +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/topo"
 	"github.com/lf-edge/ekuiper/internal/topo/node"
@@ -28,18 +29,19 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/lf-edge/ekuiper/pkg/kv"
-	"path"
 )
 
 type RuleProcessor struct {
-	db        kv.KeyValue
-	rootDbDir string
+	db kv.KeyValue
 }
 
-func NewRuleProcessor(d string) *RuleProcessor {
+func NewRuleProcessor() *RuleProcessor {
+	db, err := sqlkv.GetKVStore("rule")
+	if err != nil {
+		panic(fmt.Sprintf("Can not initalize store for the rule processor at path 'rule': %v", err))
+	}
 	processor := &RuleProcessor{
-		db:        kv.GetDefaultKVStore(path.Join(d, "rule")),
-		rootDbDir: d,
+		db: db,
 	}
 	return processor
 }
@@ -50,12 +52,6 @@ func (p *RuleProcessor) ExecCreate(name, ruleJson string) (*api.Rule, error) {
 		return nil, err
 	}
 
-	err = p.db.Open()
-	if err != nil {
-		return nil, err
-	}
-	defer p.db.Close()
-
 	err = p.db.Setnx(rule.Id, ruleJson)
 	if err != nil {
 		return nil, err
@@ -71,12 +67,6 @@ func (p *RuleProcessor) ExecUpdate(name, ruleJson string) (*api.Rule, error) {
 		return nil, err
 	}
 
-	err = p.db.Open()
-	if err != nil {
-		return nil, err
-	}
-	defer p.db.Close()
-
 	err = p.db.Set(rule.Id, ruleJson)
 	if err != nil {
 		return nil, err
@@ -99,12 +89,6 @@ func (p *RuleProcessor) ExecReplaceRuleState(name string, triggered bool) (err e
 		return fmt.Errorf("Marshal rule %s error : %s.", name, err)
 	}
 
-	err = p.db.Open()
-	if err != nil {
-		return err
-	}
-	defer p.db.Close()
-
 	err = p.db.Set(name, string(ruleJson))
 	if err != nil {
 		return err
@@ -115,11 +99,6 @@ func (p *RuleProcessor) ExecReplaceRuleState(name string, triggered bool) (err e
 }
 
 func (p *RuleProcessor) GetRuleByName(name string) (*api.Rule, error) {
-	err := p.db.Open()
-	if err != nil {
-		return nil, err
-	}
-	defer p.db.Close()
 	var s1 string
 	f, _ := p.db.Get(name, &s1)
 	if !f {
@@ -194,7 +173,7 @@ func (p *RuleProcessor) getRuleByJson(name, ruleJson string) (*api.Rule, error)
 }
 
 func (p *RuleProcessor) ExecQuery(ruleid, sql string) (*topo.Topo, error) {
-	if tp, err := planner.PlanWithSourcesAndSinks(p.getDefaultRule(ruleid, sql), p.rootDbDir, nil, []*node.SinkNode{node.NewSinkNode("sink_memory_log", "logToMemory", nil)}); err != nil {
+	if tp, err := planner.PlanWithSourcesAndSinks(p.getDefaultRule(ruleid, sql), nil, []*node.SinkNode{node.NewSinkNode("sink_memory_log", "logToMemory", nil)}); err != nil {
 		return nil, err
 	} else {
 		go func() {
@@ -214,11 +193,6 @@ func (p *RuleProcessor) ExecQuery(ruleid, sql string) (*topo.Topo, error) {
 }
 
 func (p *RuleProcessor) ExecDesc(name string) (string, error) {
-	err := p.db.Open()
-	if err != nil {
-		return "", err
-	}
-	defer p.db.Close()
 	var s1 string
 	f, _ := p.db.Get(name, &s1)
 	if !f {
@@ -233,20 +207,10 @@ func (p *RuleProcessor) ExecDesc(name string) (string, error) {
 }
 
 func (p *RuleProcessor) GetAllRules() ([]string, error) {
-	err := p.db.Open()
-	if err != nil {
-		return nil, err
-	}
-	defer p.db.Close()
 	return p.db.Keys()
 }
 
 func (p *RuleProcessor) ExecDrop(name string) (string, error) {
-	err := p.db.Open()
-	if err != nil {
-		return "", err
-	}
-	defer p.db.Close()
 	result := fmt.Sprintf("Rule %s is dropped.", name)
 	var ruleJson string
 	if ok, _ := p.db.Get(name, &ruleJson); ok {
@@ -261,7 +225,7 @@ func (p *RuleProcessor) ExecDrop(name string) (string, error) {
 			result = fmt.Sprintf("%s. Clean checkpoint cache faile: %s.", result, err)
 		}
 	}
-	err = p.db.Delete(name)
+	err := p.db.Delete(name)
 	if err != nil {
 		return "", err
 	} else {
@@ -278,16 +242,10 @@ func cleanCheckpoint(name string) error {
 }
 
 func cleanSinkCache(rule *api.Rule) error {
-	dbDir, err := conf.GetDataLoc()
-	if err != nil {
-		return err
-	}
-	store := kv.GetDefaultKVStore(path.Join(dbDir, "sink"))
-	err = store.Open()
+	store, err := sqlkv.GetKVStore("sink")
 	if err != nil {
 		return err
 	}
-	defer store.Close()
 	for d, m := range rule.Actions {
 		con := 1
 		for name, action := range m {
diff --git a/internal/processor/stream.go b/internal/processor/stream.go
index 5b29dca8..e907422b 100644
--- a/internal/processor/stream.go
+++ b/internal/processor/stream.go
@@ -19,6 +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/xsql"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
@@ -34,10 +35,13 @@ type StreamProcessor struct {
 	db kv.KeyValue
 }
 
-//@params d : the directory of the DB to save the stream info
-func NewStreamProcessor(d string) *StreamProcessor {
+func NewStreamProcessor() *StreamProcessor {
+	db, err := sqlkv.GetKVStore("stream")
+	if err != nil {
+		panic(fmt.Sprintf("Can not initalize store for the stream processor at path 'stream': %v", err))
+	}
 	processor := &StreamProcessor{
-		db: kv.GetDefaultKVStore(d),
+		db: db,
 	}
 	return processor
 }
@@ -96,11 +100,6 @@ func (p *StreamProcessor) ExecStmt(statement string) (result []string, err error
 }
 
 func (p *StreamProcessor) execSave(stmt *ast.StreamStmt, statement string, replace bool) error {
-	err := p.db.Open()
-	if err != nil {
-		return fmt.Errorf("error when opening db: %v.", err)
-	}
-	defer p.db.Close()
 	s, err := json.Marshal(xsql.StreamInfo{
 		StreamType: stmt.StreamType,
 		Statement:  statement,
@@ -160,11 +159,6 @@ func (p *StreamProcessor) execShow(st ast.StreamType) ([]string, error) {
 
 func (p *StreamProcessor) ShowStream(st ast.StreamType) ([]string, error) {
 	stt := ast.StreamTypeMap[st]
-	err := p.db.Open()
-	if err != nil {
-		return nil, fmt.Errorf("Show %ss fails, error when opening db: %v.", stt, err)
-	}
-	defer p.db.Close()
 	keys, err := p.db.Keys()
 	if err != nil {
 		return nil, fmt.Errorf("Show %ss fails, error when loading data from db: %v.", stt, err)
@@ -281,17 +275,11 @@ func (p *StreamProcessor) execDrop(stmt ast.NameNode, st ast.StreamType) (string
 }
 
 func (p *StreamProcessor) DropStream(name string, st ast.StreamType) (string, error) {
-	defer p.db.Close()
 	_, err := p.getStream(name, st)
 	if err != nil {
 		return "", err
 	}
 
-	err = p.db.Open()
-	if err != nil {
-		return "", fmt.Errorf("error when opening db: %v", err)
-	}
-	defer p.db.Close()
 	err = p.db.Delete(name)
 	if err != nil {
 		return "", err
diff --git a/internal/server/ruleManager.go b/internal/server/ruleManager.go
index e3ec7003..02ad8858 100644
--- a/internal/server/ruleManager.go
+++ b/internal/server/ruleManager.go
@@ -89,7 +89,7 @@ func createRuleState(rule *api.Rule) (*RuleState, error) {
 		Name: rule.Id,
 	}
 	registry.Store(rule.Id, rs)
-	if tp, err := planner.Plan(rule, dataDir); err != nil {
+	if tp, err := planner.Plan(rule); err != nil {
 		return rs, err
 	} else {
 		rs.Topology = tp
diff --git a/internal/server/server.go b/internal/server/server.go
index 1d325bd9..d437b458 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -16,6 +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/plugin"
 	"github.com/lf-edge/ekuiper/internal/processor"
 	"github.com/lf-edge/ekuiper/internal/service"
@@ -28,7 +29,6 @@ import (
 	"net/rpc"
 	"os"
 	"os/signal"
-	"path"
 	"syscall"
 	"time"
 )
@@ -52,21 +52,26 @@ func StartUp(Version, LoadFileType string) {
 
 	dr, err := conf.GetDataLoc()
 	if err != nil {
-		logger.Panic(err)
+		panic(err)
 	} else {
 		logger.Infof("db location is %s", dr)
 		dataDir = dr
 	}
 
-	ruleProcessor = processor.NewRuleProcessor(dataDir)
-	streamProcessor = processor.NewStreamProcessor(path.Join(dataDir, "stream"))
+	err = sqlkv.Setup(dataDir)
+	if err != nil {
+		panic(err)
+	}
+
+	ruleProcessor = processor.NewRuleProcessor()
+	streamProcessor = processor.NewStreamProcessor()
 	pluginManager, err = plugin.NewPluginManager()
 	if err != nil {
-		logger.Panic(err)
+		panic(err)
 	}
 	serviceManager, err = service.GetServiceManager()
 	if err != nil {
-		logger.Panic(err)
+		panic(err)
 	}
 	xsql.InitFuncRegisters(serviceManager, pluginManager)
 
@@ -180,5 +185,6 @@ 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 2037b7cd..a9f18696 100644
--- a/internal/service/manager.go
+++ b/internal/service/manager.go
@@ -20,6 +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/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/kv"
 	"io/ioutil"
@@ -59,17 +60,11 @@ func GetServiceManager() (*Manager, error) {
 		if err != nil {
 			return nil, fmt.Errorf("cannot find etc/services folder: %s", err)
 		}
-		dbDir, err := kconf.GetDataLoc()
-		if err != nil {
-			return nil, fmt.Errorf("cannot find db folder: %s", err)
-		}
-		sdb := kv.GetDefaultKVStore(path.Join(dbDir, "services"))
-		fdb := kv.GetDefaultKVStore(path.Join(dbDir, "serviceFuncs"))
-		err = sdb.Open()
+		sdb, err := sqlkv.GetKVStore("services")
 		if err != nil {
 			return nil, fmt.Errorf("cannot open service db: %s", err)
 		}
-		err = fdb.Open()
+		fdb, err := sqlkv.GetKVStore("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 e83e6c73..078a92a8 100644
--- a/internal/topo/node/sink_cache.go
+++ b/internal/topo/node/sink_cache.go
@@ -18,10 +18,10 @@ 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/topo/checkpoint"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/kv"
-	"io"
 	"path"
 	"sort"
 	"strconv"
@@ -100,12 +100,11 @@ func (c *Cache) initStore(ctx api.StreamContext) {
 		Data: make(map[int]interface{}),
 		Tail: 0,
 	}
-	dbDir, err := conf.GetDataLoc()
-	logger.Debugf("cache saved to %s", dbDir)
+	var err error
+	c.store, err = sqlkv.GetKVStore(path.Join("sink", ctx.GetRuleId()))
 	if err != nil {
 		c.drainError(err)
 	}
-	c.store = kv.GetDefaultKVStore(path.Join(dbDir, "sink", ctx.GetRuleId()))
 	c.key = ctx.GetOpId() + strconv.Itoa(ctx.GetInstanceId())
 	logger.Debugf("cache saved to key %s", c.key)
 	//load cache
@@ -172,52 +171,35 @@ func (c *Cache) timebasedRun(ctx api.StreamContext, saveInterval int) {
 
 func (c *Cache) loadCache() error {
 	gob.Register(c.pending)
-	err := c.store.Open()
-	if err != nil && err != io.EOF {
-		return err
-	}
-	defer c.store.Close()
-	if err == nil {
-		mt := new(LinkedQueue)
-		if f, err := c.store.Get(c.key, &mt); f {
-			if nil != err {
-				return fmt.Errorf("load malform cache, found %v(%v)", c.key, mt)
-			}
-			c.pending = mt
-			c.changed = true
-			// To store the keys in slice in sorted order
-			var keys []int
-			for k := range mt.Data {
-				keys = append(keys, k)
-			}
-			sort.Ints(keys)
-			for _, k := range keys {
-				t := &CacheTuple{
-					index: k,
-					data:  mt.Data[k],
-				}
-				c.Out <- t
+	mt := new(LinkedQueue)
+	if f, err := c.store.Get(c.key, &mt); f {
+		if nil != err {
+			return fmt.Errorf("load malform cache, found %v(%v)", c.key, mt)
+		}
+		c.pending = mt
+		c.changed = true
+		// To store the keys in slice in sorted order
+		var keys []int
+		for k := range mt.Data {
+			keys = append(keys, k)
+		}
+		sort.Ints(keys)
+		for _, k := range keys {
+			t := &CacheTuple{
+				index: k,
+				data:  mt.Data[k],
 			}
-			return nil
+			c.Out <- t
 		}
+		return nil
 	}
 	return nil
 }
 
 func (c *Cache) saveCache(logger api.Logger, p *LinkedQueue) error {
-	err := c.store.Open()
-	if err != nil {
-		logger.Errorf("save cache error while opening cache store: %s", err)
-		logger.Infof("clean the cache and reopen")
-		c.store.Close()
-		c.store.Clean()
-		err = c.store.Open()
-		if err != nil {
-			logger.Errorf("save cache error after reset the cache store: %s", err)
-			return err
-		}
-	}
-	defer c.store.Close()
+	logger.Infof("clean the cache and reopen")
+	c.store.Clean()
+
 	return c.store.Set(c.key, p)
 }
 
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index cfc7f4c5..d2ea8181 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -107,6 +107,12 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 				m.mutex.Unlock()
 				buffer = si.dataCh
 
+				defer func() {
+					logger.Infof("source %s done", m.name)
+					m.close(ctx, logger)
+					buffer.Close()
+				}()
+
 				stats, err := NewStatManager("source", ctx)
 				if err != nil {
 					m.drainError(errCh, err, ctx, logger)
@@ -119,9 +125,6 @@ func (m *SourceNode) Open(ctx api.StreamContext, errCh chan<- error) {
 				for {
 					select {
 					case <-ctx.Done():
-						logger.Infof("source %s done", m.name)
-						m.close(ctx, logger)
-						buffer.Close()
 						return
 					case err := <-si.errorCh:
 						m.drainError(errCh, err, ctx, logger)
@@ -183,8 +186,8 @@ func doGetSource(t string) (api.Source, error) {
 func (m *SourceNode) drainError(errCh chan<- error, err error, ctx api.StreamContext, logger api.Logger) {
 	select {
 	case errCh <- err:
+		logger.Debugf("sent error: %v", err)
 	case <-ctx.Done():
-		m.close(ctx, logger)
 	}
 	return
 }
diff --git a/internal/topo/node/source_pool.go b/internal/topo/node/source_pool.go
index b6e669ee..31f13cc4 100644
--- a/internal/topo/node/source_pool.go
+++ b/internal/topo/node/source_pool.go
@@ -137,7 +137,8 @@ func (p *sourcePool) deleteInstance(k string, node *SourceNode, index int) {
 		end := s.detach(instanceKey)
 		if end {
 			s.cancel()
-			s.source.Close(s.ctx)
+			_ = s.source.Close(s.ctx)
+			s.dataCh.Close()
 			delete(p.registry, k)
 		}
 	}
@@ -228,7 +229,7 @@ func (ss *sourceSingleton) broadcastError(err error) {
 			wg.Done()
 		}(n, out.errorCh)
 	}
-	ss.RLock()
+	ss.RUnlock()
 	logger.Debugf("broadcasting from source pool")
 	wg.Wait()
 }
diff --git a/internal/topo/planner/planner.go b/internal/topo/planner/planner.go
index ac702db2..3dca4ec4 100644
--- a/internal/topo/planner/planner.go
+++ b/internal/topo/planner/planner.go
@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
 	"github.com/lf-edge/ekuiper/internal/topo"
 	"github.com/lf-edge/ekuiper/internal/topo/node"
 	"github.com/lf-edge/ekuiper/internal/topo/operator"
@@ -25,15 +26,14 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/kv"
-	"path"
 )
 
-func Plan(rule *api.Rule, storePath string) (*topo.Topo, error) {
-	return PlanWithSourcesAndSinks(rule, storePath, nil, nil)
+func Plan(rule *api.Rule) (*topo.Topo, error) {
+	return PlanWithSourcesAndSinks(rule, nil, nil)
 }
 
 // For test only
-func PlanWithSourcesAndSinks(rule *api.Rule, storePath string, sources []*node.SourceNode, sinks []*node.SinkNode) (*topo.Topo, error) {
+func PlanWithSourcesAndSinks(rule *api.Rule, sources []*node.SourceNode, sinks []*node.SinkNode) (*topo.Topo, error) {
 	sql := rule.Sql
 
 	conf.Log.Infof("Init rule with options %+v", rule.Options)
@@ -49,12 +49,10 @@ func PlanWithSourcesAndSinks(rule *api.Rule, storePath string, sources []*node.S
 	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 := kv.GetDefaultKVStore(path.Join(storePath, "stream"))
-	err = store.Open()
+	store, err := sqlkv.GetKVStore("stream")
 	if err != nil {
 		return nil, err
 	}
-	defer store.Close()
 	// Create logical plan and optimize. Logical plans are a linked list
 	lp, err := createLogicalPlan(stmt, rule.Options, store)
 	if err != nil {
diff --git a/internal/topo/topo.go b/internal/topo/topo.go
index b972c3bf..ec5e04fe 100644
--- a/internal/topo/topo.go
+++ b/internal/topo/topo.go
@@ -24,6 +24,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"strconv"
+	"sync"
 )
 
 type PrintableTopo struct {
@@ -44,6 +45,7 @@ type Topo struct {
 	store              api.Store
 	coordinator        *checkpoint.Coordinator
 	topo               *PrintableTopo
+	mu                 sync.Mutex
 }
 
 func NewWithNameAndQos(name string, qos api.Qos, checkpointInterval int) (*Topo, error) {
@@ -65,6 +67,8 @@ func (s *Topo) GetContext() api.StreamContext {
 
 // Cancel may be called multiple times so must be idempotent
 func (s *Topo) Cancel() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	// completion signal
 	s.drainErr(nil)
 	s.cancel()
@@ -146,6 +150,8 @@ func (s *Topo) Open() <-chan error {
 	log.Infoln("Opening stream")
 	// open stream
 	go func() {
+		s.mu.Lock()
+		defer s.mu.Unlock()
 		var err error
 		if s.store, err = state.CreateStore(s.name, s.qos); err != nil {
 			fmt.Println(err)
diff --git a/internal/xsql/stmtx.go b/internal/xsql/stmtx.go
index e6e56abc..7dbb69cd 100644
--- a/internal/xsql/stmtx.go
+++ b/internal/xsql/stmtx.go
@@ -62,11 +62,6 @@ func GetDataSourceStatement(m kv.KeyValue, name string) (*StreamInfo, error) {
 		v  string
 		vs = &StreamInfo{}
 	)
-	err := m.Open()
-	if err != nil {
-		return nil, fmt.Errorf("error when opening db: %v", err)
-	}
-	defer m.Close()
 	if ok, _ := m.Get(name, &v); ok {
 		if err := json.Unmarshal([]byte(v), vs); err != nil {
 			return nil, fmt.Errorf("error unmarshall %s, the data in db may be corrupted", name)
diff --git a/pkg/kv/kv.go b/pkg/kv/kv.go
index 1cc24c5b..b1d91072 100644
--- a/pkg/kv/kv.go
+++ b/pkg/kv/kv.go
@@ -15,8 +15,6 @@
 package kv
 
 type KeyValue interface {
-	Open() error
-	Close() error
 	// Set key to hold string value if key does not exist otherwise return an error
 	Setnx(key string, value interface{}) error
 	// Set key to hold the string value. If key already holds a value, it is overwritten
@@ -27,7 +25,3 @@ type KeyValue interface {
 	Keys() (keys []string, err error)
 	Clean() error
 }
-
-func GetDefaultKVStore(fpath string) (ret KeyValue) {
-	return GetSqliteKVStore(fpath)
-}
diff --git a/pkg/kv/sqliteKV.go b/pkg/kv/sqliteKV.go
deleted file mode 100644
index 39414103..00000000
--- a/pkg/kv/sqliteKV.go
+++ /dev/null
@@ -1,159 +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 kv
-
-import (
-	"bytes"
-	"database/sql"
-	"encoding/gob"
-	"fmt"
-	"github.com/lf-edge/ekuiper/pkg/errorx"
-	_ "github.com/mattn/go-sqlite3"
-	"os"
-	"path"
-	"path/filepath"
-	"strings"
-)
-
-type SqliteKVStore struct {
-	db    *sql.DB
-	table string
-	path  string
-}
-
-func init() {
-	gob.Register(make(map[string]interface{}))
-}
-
-func GetSqliteKVStore(fpath string) (ret *SqliteKVStore) {
-	dir, file := filepath.Split(fpath)
-	if _, err := os.Stat(dir); os.IsNotExist(err) {
-		os.MkdirAll(dir, os.ModePerm)
-	}
-	ret = new(SqliteKVStore)
-	ret.path = path.Join(dir, "sqliteKV.db")
-	ret.table = file
-	return ret
-}
-
-func (m *SqliteKVStore) Open() error {
-	db, err := sql.Open("sqlite3", m.path)
-	if nil != err {
-		return err
-	}
-	m.db = db
-	sql := fmt.Sprintf("CREATE TABLE IF NOT EXISTS '%s'('key' VARCHAR(255) PRIMARY KEY, 'val' BLOB);", m.table)
-	_, err = m.db.Exec(sql)
-	return err
-}
-
-func (m *SqliteKVStore) Close() error {
-	if nil != m.db {
-		return m.db.Close()
-	}
-	return nil
-}
-
-func (m *SqliteKVStore) 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 (m *SqliteKVStore) Setnx(key string, value interface{}) error {
-	b, err := m.encode(value)
-	if nil != err {
-		return err
-	}
-	sql := fmt.Sprintf("INSERT INTO %s(key,val) values(?,?);", m.table)
-	stmt, err := m.db.Prepare(sql)
-	_, err = stmt.Exec(key, b)
-	stmt.Close()
-	if nil != err && strings.Contains(err.Error(), "UNIQUE constraint failed") {
-		return fmt.Errorf(`Item %s already exists`, key)
-	}
-	return err
-}
-
-func (m *SqliteKVStore) Set(key string, value interface{}) error {
-	b, err := m.encode(value)
-	if nil != err {
-		return err
-	}
-	sql := fmt.Sprintf("REPLACE INTO %s(key,val) values(?,?);", m.table)
-	stmt, err := m.db.Prepare(sql)
-	_, err = stmt.Exec(key, b)
-	stmt.Close()
-	return err
-}
-
-func (m *SqliteKVStore) Get(key string, value interface{}) (bool, error) {
-	sql := fmt.Sprintf("SELECT val FROM %s WHERE key='%s';", m.table, key)
-	row := m.db.QueryRow(sql)
-	var tmp []byte
-	err := row.Scan(&tmp)
-	if nil != err {
-		return false, nil
-	}
-
-	dec := gob.NewDecoder(bytes.NewBuffer(tmp))
-	if err := dec.Decode(value); err != nil {
-		return false, err
-	}
-	return true, nil
-}
-
-func (m *SqliteKVStore) Delete(key string) error {
-	sql := fmt.Sprintf("SELECT key FROM %s WHERE key='%s';", m.table, key)
-	row := m.db.QueryRow(sql)
-	var tmp []byte
-	err := row.Scan(&tmp)
-	if nil != err || 0 == len(tmp) {
-		return errorx.NewWithCode(errorx.NOT_FOUND, fmt.Sprintf("%s is not found", key))
-	}
-	sql = fmt.Sprintf("DELETE FROM %s WHERE key='%s';", m.table, key)
-	_, err = m.db.Exec(sql)
-	return err
-}
-
-func (m *SqliteKVStore) Keys() ([]string, error) {
-	keys := make([]string, 0)
-	sql := fmt.Sprintf("SELECT key FROM %s", m.table)
-	row, err := m.db.Query(sql)
-	if nil != err {
-		return nil, err
-	}
-	defer row.Close()
-	for row.Next() {
-		var val string
-		err = row.Scan(&val)
-		if nil != err {
-			return nil, err
-		} else {
-			keys = append(keys, val)
-		}
-	}
-	return keys, nil
-}
-
-func (m *SqliteKVStore) Clean() error {
-	sql := fmt.Sprintf("DELETE FROM %s", m.table)
-	_, err := m.db.Exec(sql)
-	return err
-}
diff --git a/tools/migration/util/migration.go b/tools/migration/util/migration.go
index d58b39af..620dab67 100644
--- a/tools/migration/util/migration.go
+++ b/tools/migration/util/migration.go
@@ -36,11 +36,10 @@ func migration(dir string) error {
 		return err
 	}
 
-	store := kv.GetDefaultKVStore(dir)
-	if err := store.Open(); nil != err {
+	err, store := kv.GetKVStore(dir)
+	if err != nil {
 		return err
 	}
-	defer store.Close()
 
 	for _, k := range keys {
 		if value, ok := c.Get(k); !ok {
@@ -74,6 +73,7 @@ func DataMigration(dir string) error {
 			}
 		}
 	}
+	kv.CloseAll()
 	return nil
 }
 


Test patch:
diff --git a/pkg/kv/sqliteKV_test.go b/internal/pkg/sqlkv/sqlKv_test.go
similarity index 87%
rename from pkg/kv/sqliteKV_test.go
rename to internal/pkg/sqlkv/sqlKv_test.go
index f627f8c8..538dac59 100644
--- a/pkg/kv/sqliteKV_test.go
+++ b/internal/pkg/sqlkv/sqlKv_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package kv
+package sqlkv
 
 import (
 	"os"
@@ -22,17 +22,16 @@ import (
 	"testing"
 )
 
-func TestSqliteKVStore_Funcs(t *testing.T) {
+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 := GetSqliteKVStore(abs)
-	if e := ks.Open(); e != nil {
-		t.Errorf("Failed to open data %s.", e)
-	}
-
+	ks, _ := GetKVStore("test")
 	if err := ks.Setnx("foo", "bar"); nil != err {
 		t.Error(err)
 	}
@@ -71,14 +70,6 @@ func TestSqliteKVStore_Funcs(t *testing.T) {
 		}
 	}
 
-	if e2 := ks.Close(); e2 != nil {
-		t.Errorf("Failed to close data: %s.", e2)
-	}
-
-	if err := ks.Open(); nil != err {
-		t.Error(err)
-	}
-
 	var v2 string
 	if ok, _ := ks.Get("foo", &v2); ok {
 		if !reflect.DeepEqual("bar", v2) {
@@ -108,6 +99,7 @@ func TestSqliteKVStore_Funcs(t *testing.T) {
 		reflect.DeepEqual(0, len(keys))
 	}
 
+	database.Disconnect()
 	dir, _ := filepath.Split(abs)
 	abs = path.Join(dir, "sqliteKV.db")
 	os.Remove(abs)
diff --git a/internal/plugin/manager_test.go b/internal/plugin/manager_test.go
index a4bf81bc..084efb58 100644
--- a/internal/plugin/manager_test.go
+++ b/internal/plugin/manager_test.go
@@ -17,6 +17,7 @@ package plugin
 import (
 	"errors"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"net/http"
 	"net/http/httptest"
@@ -31,6 +32,7 @@ var manager *Manager
 
 func init() {
 	var err error
+	testx.InitEnv()
 	manager, err = NewPluginManager()
 	if err != nil {
 		panic(err)
diff --git a/internal/processor/rule_test.go b/internal/processor/rule_test.go
index e2a611af..04d0f48a 100644
--- a/internal/processor/rule_test.go
+++ b/internal/processor/rule_test.go
@@ -142,7 +142,7 @@ func TestRuleActionParse_Apply(t *testing.T) {
 		},
 	}
 
-	p := NewRuleProcessor(DbDir)
+	p := NewRuleProcessor()
 	for i, tt := range tests {
 		r, err := p.getRuleByJson(tt.result.Id, tt.ruleStr)
 		if err != nil {
diff --git a/internal/processor/stream_test.go b/internal/processor/stream_test.go
index 81e0c641..ffd3b95b 100644
--- a/internal/processor/stream_test.go
+++ b/internal/processor/stream_test.go
@@ -17,14 +17,13 @@ package processor
 import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/testx"
-	"path"
 	"reflect"
 	"testing"
 )
 
-var (
-	DbDir = testx.GetDbDir()
-)
+func init() {
+	testx.InitEnv()
+}
 
 func TestStreamCreateProcessor(t *testing.T) {
 	var tests = []struct {
@@ -102,9 +101,9 @@ func TestStreamCreateProcessor(t *testing.T) {
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 
-	streamDB := path.Join(DbDir, "streamTest")
+	p := NewStreamProcessor()
 	for i, tt := range tests {
-		results, err := NewStreamProcessor(streamDB).ExecStmt(tt.s)
+		results, err := p.ExecStmt(tt.s)
 		if !reflect.DeepEqual(tt.err, testx.Errstring(err)) {
 			t.Errorf("%d. %q: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.s, tt.err, err)
 		} else if tt.err == "" {
@@ -190,10 +189,9 @@ func TestTableProcessor(t *testing.T) {
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
-
-	streamDB := path.Join(testx.GetDbDir(), "streamTest")
+	p := NewStreamProcessor()
 	for i, tt := range tests {
-		results, err := NewStreamProcessor(streamDB).ExecStmt(tt.s)
+		results, err := p.ExecStmt(tt.s)
 		if !reflect.DeepEqual(tt.err, testx.Errstring(err)) {
 			t.Errorf("%d. %q: error mismatch:\n  exp=%s\n  got=%s\n\n", i, tt.s, tt.err, err)
 		} else if tt.err == "" {
diff --git a/internal/service/manager_test.go b/internal/service/manager_test.go
index e4fc9cbc..de6869cb 100644
--- a/internal/service/manager_test.go
+++ b/internal/service/manager_test.go
@@ -258,12 +258,6 @@ func TestInitByFiles(t *testing.T) {
 		},
 	}
 
-	err := m.serviceKV.Open()
-	if err != nil {
-		t.Error(err)
-		t.FailNow()
-	}
-	defer m.serviceKV.Close()
 	actualService := &serviceInfo{}
 	ok, err := m.serviceKV.Get(name, actualService)
 	if err != nil {
@@ -278,12 +272,6 @@ func TestInitByFiles(t *testing.T) {
 		t.Errorf("service info mismatch, expect %v but got %v", info, actualService)
 	}
 
-	err = m.functionKV.Open()
-	if err != nil {
-		t.Error(err)
-		t.FailNow()
-	}
-	defer m.functionKV.Close()
 	actualKeys, _ := m.functionKV.Keys()
 	if len(funcs) != len(actualKeys) {
 		t.Errorf("functions info mismatch: expect %d funcs but got %v", len(funcs), actualKeys)
diff --git a/internal/testx/testUtil.go b/internal/testx/testUtil.go
index 53745ab3..f5229c7c 100644
--- a/internal/testx/testUtil.go
+++ b/internal/testx/testUtil.go
@@ -16,6 +16,7 @@ package testx
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
+	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
 )
 
 // errstring returns the string representation of an error.
@@ -26,11 +27,12 @@ func Errstring(err error) string {
 	return ""
 }
 
-func GetDbDir() string {
+func InitEnv() string {
 	conf.InitConf()
 	dbDir, err := conf.GetDataLoc()
 	if err != nil {
 		conf.Log.Fatal(err)
 	}
+	sqlkv.Setup(dbDir)
 	return dbDir
 }
diff --git a/internal/topo/planner/analyzer_test.go b/internal/topo/planner/analyzer_test.go
index 94c6cfa4..f35ca1c8 100644
--- a/internal/topo/planner/analyzer_test.go
+++ b/internal/topo/planner/analyzer_test.go
@@ -17,12 +17,11 @@ package planner
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
 	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
-	"github.com/lf-edge/ekuiper/pkg/kv"
-	"path"
 	"reflect"
 	"strings"
 	"testing"
@@ -116,13 +115,11 @@ var tests = []struct {
 }
 
 func Test_validation(t *testing.T) {
-	store := kv.GetDefaultKVStore(path.Join(DbDir, "stream"))
-	err := store.Open()
+	store, err := sqlkv.GetKVStore("stream")
 	if err != nil {
 		t.Error(err)
 		return
 	}
-	defer store.Close()
 	streamSqls := map[string]string{
 		"src1": `CREATE STREAM src1 (
 					id1 BIGINT,
@@ -179,13 +176,11 @@ func Test_validation(t *testing.T) {
 }
 
 func Test_validationSchemaless(t *testing.T) {
-	store := kv.GetDefaultKVStore(path.Join(DbDir, "stream"))
-	err := store.Open()
+	store, err := sqlkv.GetKVStore("stream")
 	if err != nil {
 		t.Error(err)
 		return
 	}
-	defer store.Close()
 	streamSqls := map[string]string{
 		"src1": `CREATE STREAM src1 (
 				) WITH (DATASOURCE="src1", FORMAT="json", KEY="ts");`,
diff --git a/internal/topo/planner/planner_test.go b/internal/topo/planner/planner_test.go
index 96e126ea..45be2c83 100644
--- a/internal/topo/planner/planner_test.go
+++ b/internal/topo/planner/planner_test.go
@@ -18,29 +18,26 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/gdexlab/go-render/render"
+	"github.com/lf-edge/ekuiper/internal/pkg/sqlkv"
 	"github.com/lf-edge/ekuiper/internal/testx"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
-	"github.com/lf-edge/ekuiper/pkg/kv"
-	"path"
 	"reflect"
 	"strings"
 	"testing"
 )
 
-var (
-	DbDir = testx.GetDbDir()
-)
+func init() {
+	testx.InitEnv()
+}
 
 func Test_createLogicalPlan(t *testing.T) {
-	store := kv.GetDefaultKVStore(path.Join(DbDir, "stream"))
-	err := store.Open()
+	store, err := sqlkv.GetKVStore("stream")
 	if err != nil {
 		t.Error(err)
 		return
 	}
-	defer store.Close()
 	streamSqls := map[string]string{
 		"src1": `CREATE STREAM src1 (
 					id1 BIGINT,
@@ -1053,13 +1050,11 @@ func Test_createLogicalPlan(t *testing.T) {
 }
 
 func Test_createLogicalPlanSchemaless(t *testing.T) {
-	store := kv.GetDefaultKVStore(path.Join(DbDir, "stream"))
-	err := store.Open()
+	store, err := sqlkv.GetKVStore("stream")
 	if err != nil {
 		t.Error(err)
 		return
 	}
-	defer store.Close()
 	streamSqls := map[string]string{
 		"src1": `CREATE STREAM src1 (
 				) WITH (DATASOURCE="src1", FORMAT="json", KEY="ts");`,
diff --git a/internal/topo/topotest/mock_topo.go b/internal/topo/topotest/mock_topo.go
index 3dc4cf70..a9e9083f 100644
--- a/internal/topo/topotest/mock_topo.go
+++ b/internal/topo/topotest/mock_topo.go
@@ -29,13 +29,16 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
-	"path"
 	"reflect"
 	"strings"
 	"testing"
 	"time"
 )
 
+func init() {
+	testx.InitEnv()
+}
+
 const POSTLEAP = 1000 // Time change after all data sends out
 type RuleTest struct {
 	Name string
@@ -46,10 +49,6 @@ type RuleTest struct {
 	W    int                    // wait time for each data sending, in milli
 }
 
-var (
-	DbDir = testx.GetDbDir()
-)
-
 func compareMetrics(tp *topo.Topo, m map[string]interface{}) (err error) {
 	keys, values := tp.GetMetrics()
 	for k, v := range m {
@@ -252,7 +251,7 @@ func createStream(t *testing.T, tt RuleTest, j int, opt *api.RuleOption, sinkPro
 	}
 	mockSink := mocknode.NewMockSink()
 	sink := node.NewSinkNodeWithSink("mockSink", mockSink, sinkProps)
-	tp, err := planner.PlanWithSourcesAndSinks(&api.Rule{Id: fmt.Sprintf("%s_%d", tt.Name, j), Sql: tt.Sql, Options: opt}, DbDir, sources, []*node.SinkNode{sink})
+	tp, err := planner.PlanWithSourcesAndSinks(&api.Rule{Id: fmt.Sprintf("%s_%d", tt.Name, j), Sql: tt.Sql, Options: opt}, sources, []*node.SinkNode{sink})
 	if err != nil {
 		t.Error(err)
 		return nil, 0, nil, nil, nil
@@ -263,7 +262,7 @@ func createStream(t *testing.T, tt RuleTest, j int, opt *api.RuleOption, sinkPro
 
 // Create or drop streams
 func HandleStream(createOrDrop bool, names []string, t *testing.T) {
-	p := processor.NewStreamProcessor(path.Join(DbDir, "stream"))
+	p := processor.NewStreamProcessor()
 	for _, name := range names {
 		var sql string
 		if createOrDrop {
@@ -445,7 +444,7 @@ func DoCheckpointRuleTest(t *testing.T, tests []RuleCheckpointTest, j int, opt *
 }
 
 func CreateRule(name, sql string) (*api.Rule, error) {
-	p := processor.NewRuleProcessor(DbDir)
+	p := processor.NewRuleProcessor()
 	p.ExecDrop(name)
 	return p.ExecCreate(name, sql)
 }
diff --git a/internal/topo/topotest/plugin_rule_test.go b/internal/topo/topotest/plugin_rule_test.go
index 0b1f3073..edeabb98 100644
--- a/internal/topo/topotest/plugin_rule_test.go
+++ b/internal/topo/topotest/plugin_rule_test.go
@@ -81,7 +81,7 @@ func TestExtensions(t *testing.T) {
 			t.Errorf("failed to create rule: %s.", err)
 			continue
 		}
-		tp, err := planner.Plan(rs, DbDir)
+		tp, err := planner.Plan(rs)
 		if err != nil {
 			t.Errorf("fail to init rule: %v", err)
 			continue
diff --git a/test/select_aggr_rule.jmx b/test/select_aggr_rule.jmx
index 2fdbc3da..dd583103 100644
--- a/test/select_aggr_rule.jmx
+++ b/test/select_aggr_rule.jmx
@@ -131,7 +131,7 @@
                 <elementProp name="" elementType="HTTPArgument">
                   <boolProp name="HTTPArgument.always_encode">false</boolProp>
                   <stringProp name="Argument.value">{&#xd;
-&quot;sql&quot; : &quot;create stream demo (temperature float, humidity bigint) WITH (FORMAT=\&quot;JSON\&quot;, DATASOURCE=\&quot;devices/+/messages\&quot; )&quot;&#xd;
+&quot;sql&quot; : &quot;create stream ademo (temperature float, humidity bigint) WITH (FORMAT=\&quot;JSON\&quot;, DATASOURCE=\&quot;devices/+/messages\&quot; )&quot;&#xd;
 }</stringProp>
                   <stringProp name="Argument.metadata">=</stringProp>
                 </elementProp>
@@ -154,7 +154,7 @@
           <hashTree>
             <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
               <collectionProp name="Asserion.test_strings">
-                <stringProp name="-1754954177">Stream demo is created.</stringProp>
+                <stringProp name="-1754954177">Stream ademo is created.</stringProp>
               </collectionProp>
               <stringProp name="Assertion.custom_message"></stringProp>
               <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
@@ -170,15 +170,15 @@
                 <elementProp name="" elementType="HTTPArgument">
                   <boolProp name="HTTPArgument.always_encode">false</boolProp>
                   <stringProp name="Argument.value">{&#xd;
-  &quot;id&quot;: &quot;rule1&quot;,&#xd;
-  &quot;sql&quot;: &quot;SELECT avg(temperature) AS t_av, max(temperature) AS t_max, min(temperature) AS t_min, COUNT(*) As t_count, split_value(mqtt(topic), \&quot;/\&quot;, 1) AS device_id FROM demo GROUP BY device_id, TUMBLINGWINDOW(ss, 5)&quot;,&#xd;
+  &quot;id&quot;: &quot;arule&quot;,&#xd;
+  &quot;sql&quot;: &quot;SELECT avg(temperature) AS t_av, max(temperature) AS t_max, min(temperature) AS t_min, COUNT(*) As t_count, split_value(mqtt(topic), \&quot;/\&quot;, 1) AS device_id FROM ademo GROUP BY device_id, TUMBLINGWINDOW(ss, 5)&quot;,&#xd;
   &quot;actions&quot;: [&#xd;
     {&#xd;
       &quot;mqtt&quot;: {&#xd;
         &quot;server&quot;: &quot;tcp://${mqtt_srv}:1883&quot;,&#xd;
         &quot;topic&quot;: &quot;devices/result&quot;,&#xd;
         &quot;qos&quot;: 1,&#xd;
-        &quot;clientId&quot;: &quot;demo_001&quot;&#xd;
+        &quot;clientId&quot;: &quot;ademo_001&quot;&#xd;
       }&#xd;
     }&#xd;
   ]&#xd;
@@ -204,7 +204,7 @@
           <hashTree>
             <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
               <collectionProp name="Asserion.test_strings">
-                <stringProp name="-2022196798">Rule rule1 was created</stringProp>
+                <stringProp name="-2022196798">Rule arule was created</stringProp>
               </collectionProp>
               <stringProp name="Assertion.custom_message"></stringProp>
               <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
@@ -228,7 +228,7 @@
             <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
             <stringProp name="HTTPSampler.protocol"></stringProp>
             <stringProp name="HTTPSampler.contentEncoding"></stringProp>
-            <stringProp name="HTTPSampler.path">/rules/rule1/status</stringProp>
+            <stringProp name="HTTPSampler.path">/rules/arule/status</stringProp>
             <stringProp name="HTTPSampler.method">GET</stringProp>
             <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
             <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
@@ -240,7 +240,7 @@
           </HTTPSamplerProxy>
           <hashTree>
             <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
-              <stringProp name="JSON_PATH">$.source_demo_0_records_in_total</stringProp>
+              <stringProp name="JSON_PATH">$.source_ademo_0_records_in_total</stringProp>
               <stringProp name="EXPECTED_VALUE">0</stringProp>
               <boolProp name="JSONVALIDATION">true</boolProp>
               <boolProp name="EXPECT_NULL">false</boolProp>
@@ -291,7 +291,7 @@
             <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
             <stringProp name="HTTPSampler.protocol"></stringProp>
             <stringProp name="HTTPSampler.contentEncoding"></stringProp>
-            <stringProp name="HTTPSampler.path">/rules/rule1/status</stringProp>
+            <stringProp name="HTTPSampler.path">/rules/arule/status</stringProp>
             <stringProp name="HTTPSampler.method">GET</stringProp>
             <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
             <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
@@ -303,7 +303,7 @@
           </HTTPSamplerProxy>
           <hashTree>
             <JSONPathAssertion guiclass="JSONPathAssertionGui" testclass="JSONPathAssertion" testname="JSON Assertion" enabled="true">
-              <stringProp name="JSON_PATH">$.source_demo_0_records_in_total</stringProp>
+              <stringProp name="JSON_PATH">$.source_ademo_0_records_in_total</stringProp>
               <stringProp name="EXPECTED_VALUE">10</stringProp>
               <boolProp name="JSONVALIDATION">true</boolProp>
               <boolProp name="EXPECT_NULL">false</boolProp>
@@ -340,7 +340,7 @@
             <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
             <stringProp name="HTTPSampler.protocol"></stringProp>
             <stringProp name="HTTPSampler.contentEncoding"></stringProp>
-            <stringProp name="HTTPSampler.path">/rules/rule1</stringProp>
+            <stringProp name="HTTPSampler.path">/rules/arule</stringProp>
             <stringProp name="HTTPSampler.method">DELETE</stringProp>
             <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
             <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
@@ -353,7 +353,7 @@
           <hashTree>
             <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
               <collectionProp name="Asserion.test_strings">
-                <stringProp name="717250485">Rule rule1 is dropped.</stringProp>
+                <stringProp name="717250485">Rule arule is dropped.</stringProp>
               </collectionProp>
               <stringProp name="Assertion.custom_message"></stringProp>
               <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>
@@ -377,7 +377,7 @@
             <stringProp name="HTTPSampler.port">${rest_port}</stringProp>
             <stringProp name="HTTPSampler.protocol"></stringProp>
             <stringProp name="HTTPSampler.contentEncoding"></stringProp>
-            <stringProp name="HTTPSampler.path">/streams/demo</stringProp>
+            <stringProp name="HTTPSampler.path">/streams/ademo</stringProp>
             <stringProp name="HTTPSampler.method">DELETE</stringProp>
             <boolProp name="HTTPSampler.follow_redirects">true</boolProp>
             <boolProp name="HTTPSampler.auto_redirects">false</boolProp>
@@ -390,7 +390,7 @@
           <hashTree>
             <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="Response Assertion" enabled="true">
               <collectionProp name="Asserion.test_strings">
-                <stringProp name="287881319">Stream demo is dropped.</stringProp>
+                <stringProp name="287881319">Stream ademo is dropped.</stringProp>
               </collectionProp>
               <stringProp name="Assertion.custom_message"></stringProp>
               <stringProp name="Assertion.test_field">Assertion.response_data</stringProp>