[db] KvWithVersion to handle both versioned and non-versioned namespace

db/db_versioned.go

@@ -46,6 +46,9 @@ type (
 
 		// Version returns the key's most recent version
 		Version(string, []byte) (uint64, error)
+
+		// CommitToDB writes a batch to the underlying DB
+		CommitToDB(uint64, map[string]bool, batch.KVStoreBatch) error
 	}
 
 	// BoltDBVersioned is KvVersioned implementation based on bolt DB

db/kvstore_versioned.go

@@ -7,6 +7,11 @@
 package db
 
 import (
+	"context"
+
+	"github.com/pkg/errors"
+
+	"github.com/iotexproject/iotex-core/v2/db/batch"
 	"github.com/iotexproject/iotex-core/v2/pkg/lifecycle"
 )
 
@@ -47,4 +52,107 @@ type (
 		// SetVersion sets the version, and returns a KVStore to call Put()/Get()
 		SetVersion(uint64) KVStore
 	}
+
+	// KvWithVersion wraps the versioned DB implementation with a certain version
+	KvWithVersion struct {
+		db        VersionedDB
+		kvBase    KVStore
+		versioned map[string]bool // map of versioned buckets
+		version   uint64          // the current version
+	}
 )
+
+// Option sets an option
+type Option func(*KvWithVersion)
+
+func VersionedNamespaceOption(ns ...string) Option {
+	return func(k *KvWithVersion) {
+		k.versioned = make(map[string]bool)
+		for _, ns := range ns {
+			k.versioned[ns] = true
+		}
+	}
+}
+
+// NewKVStoreWithVersion implements a KVStore that can handle both versioned
+// and non-versioned namespace
+func NewKVStoreWithVersion(cfg Config, opts ...Option) *KvWithVersion {
+	db := NewBoltDBVersioned(cfg)
+	kv := KvWithVersion{
+		db:     db,
+		kvBase: db.Base(),
+	}
+	for _, opt := range opts {
+		opt(&kv)
+	}
+	return &kv
+}
+
+// Start starts the DB
+func (b *KvWithVersion) Start(ctx context.Context) error {
+	return b.kvBase.Start(ctx)
+}
+
+// Stop stops the DB
+func (b *KvWithVersion) Stop(ctx context.Context) error {
+	return b.kvBase.Stop(ctx)
+}
+
+// Put writes a <key, value> record
+func (b *KvWithVersion) Put(ns string, key, value []byte) error {
+	if b.versioned[ns] {
+		return b.db.Put(b.version, ns, key, value)
+	}
+	return b.kvBase.Put(ns, key, value)
+}
+
+// Get retrieves a key's value
+func (b *KvWithVersion) Get(ns string, key []byte) ([]byte, error) {
+	if b.versioned[ns] {
+		return b.db.Get(b.version, ns, key)
+	}
+	return b.kvBase.Get(ns, key)
+}
+
+// Delete deletes a key
+func (b *KvWithVersion) Delete(ns string, key []byte) error {
+	if b.versioned[ns] {
+		return b.db.Delete(b.version, ns, key)
+	}
+	return b.kvBase.Delete(ns, key)
+}
+
+// Filter returns <k, v> pair in a bucket that meet the condition
+func (b *KvWithVersion) Filter(ns string, cond Condition, minKey, maxKey []byte) ([][]byte, [][]byte, error) {
+	if b.versioned[ns] {
+		panic("Filter not supported for versioned DB")
+	}
+	return b.kvBase.Filter(ns, cond, minKey, maxKey)
+}
+
+// WriteBatch commits a batch
+func (b *KvWithVersion) WriteBatch(kvsb batch.KVStoreBatch) error {
+	return b.db.CommitToDB(b.version, b.versioned, kvsb)
+}
+
+// Version returns the key's most recent version
+func (b *KvWithVersion) Version(ns string, key []byte) (uint64, error) {
+	if b.versioned[ns] {
+		return b.db.Version(ns, key)
+	}
+	return 0, errors.Errorf("namespace %s is non-versioned", ns)
+}
+
+// SetVersion sets the version, and returns a KVStore to call Put()/Get()
+func (b *KvWithVersion) SetVersion(v uint64) KVStore {
+	kv := KvWithVersion{
+		db:        b.db,
+		kvBase:    b.kvBase,
+		versioned: make(map[string]bool),
+		version:   v,
+	}
+	for k := range b.versioned {
+		kv.versioned[k] = true
+	}
+	return &kv
+}