refactor(kv): move kv to a standalone package to prevent loading sqlite at startup

common/kv/kv.go

@@ -0,0 +1,19 @@
+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
+	Set(key string, value interface{}) error
+	Get(key string, val interface{}) (bool, error)
+	//Must return *common.Error with NOT_FOUND error
+	Delete(key string) error
+	Keys() (keys []string, err error)
+	Clean() error
+}
+
+func GetDefaultKVStore(fpath string) (ret KeyValue) {
+	return GetSqliteKVStore(fpath)
+}

common/kv.go

@@ -1,10 +1,11 @@
-package common
+package kv
 
 import (
 	"bytes"
 	"database/sql"
 	"encoding/gob"
 	"fmt"
+	"github.com/emqx/kuiper/common"
 	_ "github.com/mattn/go-sqlite3"
 	"os"
 	"path"
@@ -12,20 +13,6 @@ import (
 	"strings"
 )
 
-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
-	Set(key string, value interface{}) error
-	Get(key string, val interface{}) (bool, error)
-	//Must return *common.Error with NOT_FOUND error
-	Delete(key string) error
-	Keys() (keys []string, err error)
-	Clean() error
-}
-
 type SqliteKVStore struct {
 	db    *sql.DB
 	table string
@@ -120,7 +107,7 @@ func (m *SqliteKVStore) Delete(key string) error {
 	var tmp []byte
 	err := row.Scan(&tmp)
 	if nil != err || 0 == len(tmp) {
-		return NewErrorWithCode(NOT_FOUND, fmt.Sprintf("%s is not found", key))
+		return common.NewErrorWithCode(common.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)

common/kv/sqliteKV_test.go

@@ -0,0 +1,101 @@
+package kv
+
+import (
+	"os"
+	"path"
+	"path/filepath"
+	"reflect"
+	"testing"
+)
+
+func TestSqliteKVStore_Funcs(t *testing.T) {
+	abs, _ := filepath.Abs("test")
+	if f, _ := os.Stat(abs); f != nil {
+		os.Remove(abs)
+	}
+
+	ks := GetSqliteKVStore(abs)
+	if e := ks.Open(); e != nil {
+		t.Errorf("Failed to open data %s.", e)
+	}
+
+	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))
+		}
+	}
+
+	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) {
+			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))
+	}
+
+	dir, _ := filepath.Split(abs)
+	abs = path.Join(dir, "sqliteKV.db")
+	os.Remove(abs)
+
+}

common/util_test.go

@@ -1,106 +1,11 @@
 package common
 
 import (
-	"os"
-	"path"
-	"path/filepath"
 	"reflect"
 	"strings"
 	"testing"
 )
 
-func TestSqliteKVStore_Funcs(t *testing.T) {
-	abs, _ := filepath.Abs("test")
-	if f, _ := os.Stat(abs); f != nil {
-		os.Remove(abs)
-	}
-
-	ks := GetSqliteKVStore(abs)
-	if e := ks.Open(); e != nil {
-		t.Errorf("Failed to open data %s.", e)
-	}
-
-	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))
-		}
-	}
-
-	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) {
-			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))
-	}
-
-	dir, _ := filepath.Split(abs)
-	abs = path.Join(dir, "sqliteKV.db")
-	os.Remove(abs)
-
-}
-
 func TestMapConvert_Funcs(t *testing.T) {
 	source := map[interface{}]interface{}{
 		"QUERY_TABLE": "VBAP",

plugins/manager.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/emqx/kuiper/xstream/api"
 	"io"
 	"io/ioutil"
@@ -281,7 +282,7 @@ type Manager struct {
 	pluginDir string
 	etcDir    string
 	registry  *Registry
-	db        common.KeyValue
+	db        kv.KeyValue
 }
 
 func NewPluginManager() (*Manager, error) {
@@ -302,7 +303,7 @@ func NewPluginManager() (*Manager, error) {
 			outerErr = fmt.Errorf("cannot find db folder: %s", err)
 			return
 		}
-		db := common.GetSqliteKVStore(path.Join(dbDir, "pluginFuncs"))
+		db := kv.GetDefaultKVStore(path.Join(dbDir, "pluginFuncs"))
 		err = db.Open()
 		if err != nil {
 			outerErr = fmt.Errorf("error when opening db: %v.", err)

tools/migration/util/migration.go

@@ -2,7 +2,7 @@ package util
 
 import (
 	"fmt"
-	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/patrickmn/go-cache"
 	"io/ioutil"
 	"os"
@@ -22,7 +22,7 @@ func migration(dir string) error {
 		return err
 	}
 
-	store := common.GetSqliteKVStore(dir)
+	store := kv.GetDefaultKVStore(dir)
 	if err := store.Open(); nil != err {
 		return err
 	}

tools/migration/util/migration_test.go

@@ -1,7 +1,7 @@
 package util
 
 import (
-	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/patrickmn/go-cache"
 	"os"
 	"path"
@@ -47,7 +47,7 @@ func TestDataMigration(t *testing.T) {
 		return
 	}
 
-	store := common.GetSqliteKVStore(dir)
+	store := kv.GetDefaultKVStore(dir)
 	if err := store.Open(); nil != err {
 		t.Error(err)
 		return

xsql/processors/xsql_processor.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/emqx/kuiper/xsql"
 	"github.com/emqx/kuiper/xstream"
 	"github.com/emqx/kuiper/xstream/api"
@@ -18,13 +19,13 @@ import (
 var log = common.Log
 
 type StreamProcessor struct {
-	db common.KeyValue
+	db kv.KeyValue
 }
 
 //@params d : the directory of the DB to save the stream info
 func NewStreamProcessor(d string) *StreamProcessor {
 	processor := &StreamProcessor{
-		db: common.GetSqliteKVStore(d),
+		db: kv.GetDefaultKVStore(d),
 	}
 	return processor
 }
@@ -208,13 +209,13 @@ func (p *StreamProcessor) DropStream(name string) (string, error) {
 }
 
 type RuleProcessor struct {
-	db        common.KeyValue
+	db        kv.KeyValue
 	rootDbDir string
 }
 
 func NewRuleProcessor(d string) *RuleProcessor {
 	processor := &RuleProcessor{
-		db:        common.GetSqliteKVStore(path.Join(d, "rule")),
+		db:        kv.GetDefaultKVStore(path.Join(d, "rule")),
 		rootDbDir: d,
 	}
 	return processor
@@ -456,7 +457,7 @@ func cleanSinkCache(rule *api.Rule) error {
 	if err != nil {
 		return err
 	}
-	store := common.GetSqliteKVStore(path.Join(dbDir, "sink"))
+	store := kv.GetDefaultKVStore(path.Join(dbDir, "sink"))
 	err = store.Open()
 	if err != nil {
 		return err

xstream/nodes/sink_cache.go

@@ -4,6 +4,7 @@ import (
 	"encoding/gob"
 	"fmt"
 	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/emqx/kuiper/xstream/api"
 	"github.com/emqx/kuiper/xstream/checkpoints"
 	"io"
@@ -65,7 +66,7 @@ type Cache struct {
 	changed bool
 	//serialize
 	key   string //the key for current cache
-	store common.KeyValue
+	store kv.KeyValue
 }
 
 func NewTimebasedCache(in <-chan interface{}, limit int, saveInterval int, errCh chan<- error, ctx api.StreamContext) *Cache {
@@ -90,7 +91,7 @@ func (c *Cache) initStore(ctx api.StreamContext) {
 	if err != nil {
 		c.drainError(err)
 	}
-	c.store = common.GetSqliteKVStore(path.Join(dbDir, "sink", ctx.GetRuleId()))
+	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

xstream/planner/planner.go

@@ -3,6 +3,7 @@ package planner
 import (
 	"fmt"
 	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/emqx/kuiper/xsql"
 	"github.com/emqx/kuiper/xstream"
 	"github.com/emqx/kuiper/xstream/api"
@@ -33,7 +34,7 @@ func PlanWithSourcesAndSinks(rule *api.Rule, storePath string, sources []*nodes.
 	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 := common.GetSqliteKVStore(path.Join(storePath, "stream"))
+	store := kv.GetDefaultKVStore(path.Join(storePath, "stream"))
 	err = store.Open()
 	if err != nil {
 		return nil, err
@@ -162,7 +163,7 @@ func buildOps(lp LogicalPlan, tp *xstream.TopologyNew, options *api.RuleOption,
 	return op, newIndex, nil
 }
 
-func getStream(m common.KeyValue, name string) (stmt *xsql.StreamStmt, err error) {
+func getStream(m kv.KeyValue, name string) (stmt *xsql.StreamStmt, err error) {
 	var s string
 	f, err := m.Get(name, &s)
 	if !f || err != nil {
@@ -177,7 +178,7 @@ func getStream(m common.KeyValue, name string) (stmt *xsql.StreamStmt, err error
 	return
 }
 
-func createLogicalPlan(stmt *xsql.SelectStatement, opt *api.RuleOption, store common.KeyValue) (LogicalPlan, error) {
+func createLogicalPlan(stmt *xsql.SelectStatement, opt *api.RuleOption, store kv.KeyValue) (LogicalPlan, error) {
 	streamsFromStmt := xsql.GetStreams(stmt)
 	dimensions := stmt.Dimensions
 	var (

xstream/planner/planner_test.go

@@ -3,6 +3,7 @@ package planner
 import (
 	"fmt"
 	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/emqx/kuiper/xsql"
 	"github.com/emqx/kuiper/xstream/api"
 	"path"
@@ -26,7 +27,7 @@ func getDbDir() string {
 }
 
 func Test_createLogicalPlan(t *testing.T) {
-	store := common.GetSqliteKVStore(path.Join(DbDir, "stream"))
+	store := kv.GetDefaultKVStore(path.Join(DbDir, "stream"))
 	err := store.Open()
 	if err != nil {
 		t.Error(err)

xstream/states/kv_store.go

@@ -4,6 +4,7 @@ import (
 	"encoding/gob"
 	"fmt"
 	"github.com/emqx/kuiper/common"
+	"github.com/emqx/kuiper/common/kv"
 	"github.com/emqx/kuiper/xstream/checkpoints"
 	"path"
 	"sync"
@@ -20,7 +21,7 @@ func init() {
 ***   { "checkpoint1", "checkpoint2" ... "checkpointn" : The complete or incomplete snapshot
  */
 type KVStore struct {
-	db          common.KeyValue
+	db          kv.KeyValue
 	mapStore    *sync.Map //The current root store of a rule
 	checkpoints []int64
 	max         int
@@ -33,7 +34,7 @@ type KVStore struct {
 //Assume each operator only has one instance
 func getKVStore(ruleId string) (*KVStore, error) {
 	dr, _ := common.GetDataLoc()
-	db := common.GetSqliteKVStore(path.Join(dr, "checkpoints", ruleId))
+	db := kv.GetDefaultKVStore(path.Join(dr, "checkpoints", ruleId))
 	s := &KVStore{db: db, max: 3, mapStore: &sync.Map{}}
 	//read data from badger db
 	if err := s.restore(); err != nil {