[FAB-9746] Ledger: Metadata - level-statedb support

This CR includes
 - Enhancement in statedb data struct to encapsulate metadata
 - Introducing a proto message for encapsulating version, metadata,
   and value
 - Enhancement to the encoding/decoding so as to handle both
   the old and new format of encoding the value field in the leveldb

Change-Id: Iac530f3045fef80c4dbb5c09bfe1de41ccc4bb97
Signed-off-by: manish <manish.sethi@gmail.com>

Author: manish-sethi

core/ledger/kvledger/txmgmt/statedb/statedb.go

@@ -1,5 +1,6 @@
 /*
 Copyright IBM Corp. All Rights Reserved.
+
 SPDX-License-Identifier: Apache-2.0
 */
 
@@ -81,8 +82,9 @@ type CompositeKey struct {
 
 // VersionedValue encloses value and corresponding version
 type VersionedValue struct {
-	Value   []byte
-	Version *version.Height
+	Value    []byte
+	Metadata []byte
+	Version  *version.Height
 }
 
 // VersionedKV encloses key and corresponding VersionedValue
@@ -131,17 +133,17 @@ func (batch *UpdateBatch) Get(ns string, key string) *VersionedValue {
 	return vv
 }
 
-// Put adds a VersionedKV
+// Put adds a key with value only. The metadata is assumed to be nil
 func (batch *UpdateBatch) Put(ns string, key string, value []byte, version *version.Height) {
 	if value == nil {
-		panic("Nil value not allowed")
+		panic("Nil value not allowed. Instead call 'Delete' function")
 	}
-	batch.Update(ns, key, &VersionedValue{value, version})
+	batch.Update(ns, key, &VersionedValue{value, nil, version})
 }
 
 // Delete deletes a Key and associated value
 func (batch *UpdateBatch) Delete(ns string, key string, version *version.Height) {
-	batch.Update(ns, key, &VersionedValue{nil, version})
+	batch.Update(ns, key, &VersionedValue{nil, nil, version})
 }
 
 // Exists checks whether the given key exists in the batch
@@ -236,7 +238,7 @@ func (itr *nsIterator) Next() (QueryResult, error) {
 	key := itr.sortedKeys[itr.nextIndex]
 	vv := itr.nsUpdates.m[key]
 	itr.nextIndex++
-	return &VersionedKV{CompositeKey{itr.ns, key}, VersionedValue{vv.Value, vv.Version}}, nil
+	return &VersionedKV{CompositeKey{itr.ns, key}, VersionedValue{vv.Value, vv.Metadata, vv.Version}}, nil
 }
 
 // Close implements the method from QueryResult interface

core/ledger/kvledger/txmgmt/statedb/statedb_test.go

@@ -105,19 +105,19 @@ func TestUpdateBatchIterator(t *testing.T) {
 	batch.Put("ns2", "key4", []byte("value4"), version.NewHeight(2, 1))
 
 	checkItrResults(t, batch.GetRangeScanIterator("ns1", "key2", "key3"), []*VersionedKV{
-		{CompositeKey{"ns1", "key2"}, VersionedValue{[]byte("value2"), version.NewHeight(1, 2)}},
+		{CompositeKey{"ns1", "key2"}, VersionedValue{[]byte("value2"), nil, version.NewHeight(1, 2)}},
 	})
 
 	checkItrResults(t, batch.GetRangeScanIterator("ns2", "key0", "key8"), []*VersionedKV{
-		{CompositeKey{"ns2", "key4"}, VersionedValue{[]byte("value4"), version.NewHeight(2, 1)}},
-		{CompositeKey{"ns2", "key5"}, VersionedValue{[]byte("value5"), version.NewHeight(2, 2)}},
-		{CompositeKey{"ns2", "key6"}, VersionedValue{[]byte("value6"), version.NewHeight(2, 3)}},
+		{CompositeKey{"ns2", "key4"}, VersionedValue{[]byte("value4"), nil, version.NewHeight(2, 1)}},
+		{CompositeKey{"ns2", "key5"}, VersionedValue{[]byte("value5"), nil, version.NewHeight(2, 2)}},
+		{CompositeKey{"ns2", "key6"}, VersionedValue{[]byte("value6"), nil, version.NewHeight(2, 3)}},
 	})
 
 	checkItrResults(t, batch.GetRangeScanIterator("ns2", "", ""), []*VersionedKV{
-		{CompositeKey{"ns2", "key4"}, VersionedValue{[]byte("value4"), version.NewHeight(2, 1)}},
-		{CompositeKey{"ns2", "key5"}, VersionedValue{[]byte("value5"), version.NewHeight(2, 2)}},
-		{CompositeKey{"ns2", "key6"}, VersionedValue{[]byte("value6"), version.NewHeight(2, 3)}},
+		{CompositeKey{"ns2", "key4"}, VersionedValue{[]byte("value4"), nil, version.NewHeight(2, 1)}},
+		{CompositeKey{"ns2", "key5"}, VersionedValue{[]byte("value5"), nil, version.NewHeight(2, 2)}},
+		{CompositeKey{"ns2", "key6"}, VersionedValue{[]byte("value6"), nil, version.NewHeight(2, 3)}},
 	})
 
 	checkItrResults(t, batch.GetRangeScanIterator("non-existing-ns", "", ""), nil)

core/ledger/kvledger/txmgmt/statedb/stateleveldb/msgs/storage.pb.go

@@ -0,0 +1,17 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+syntax = "proto3";
+
+option go_package = "github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb/stateleveldb/msgs";
+
+package msgs;
+
+message VersionedValueProto {
+    bytes version_bytes = 1;
+    bytes value = 2;
+    bytes metadata = 3;
+}

core/ledger/kvledger/txmgmt/statedb/stateleveldb/msgs/storage.proto

@@ -89,8 +89,7 @@ func (vdb *versionedDB) GetState(namespace string, key string) (*statedb.Version
 	if dbVal == nil {
 		return nil, nil
 	}
-	val, ver := DecodeValue(dbVal)
-	return &statedb.VersionedValue{Value: val, Version: ver}, nil
+	return decodeValue(dbVal)
 }
 
 // GetVersion implements method in VersionedDB interface
@@ -149,7 +148,11 @@ func (vdb *versionedDB) ApplyUpdates(batch *statedb.UpdateBatch, height *version
 			if vv.Value == nil {
 				dbBatch.Delete(compositeKey)
 			} else {
-				dbBatch.Put(compositeKey, EncodeValue(vv.Value, vv.Version))
+				encodedVal, err := encodeValue(vv)
+				if err != nil {
+					return err
+				}
+				dbBatch.Put(compositeKey, encodedVal)
 			}
 		}
 	}
@@ -201,10 +204,15 @@ func (scanner *kvScanner) Next() (statedb.QueryResult, error) {
 	dbValCopy := make([]byte, len(dbVal))
 	copy(dbValCopy, dbVal)
 	_, key := splitCompositeKey(dbKey)
-	value, version := DecodeValue(dbValCopy)
+	vv, err := decodeValue(dbValCopy)
+	if err != nil {
+		return nil, err
+	}
 	return &statedb.VersionedKV{
-		CompositeKey:   statedb.CompositeKey{Namespace: scanner.namespace, Key: key},
-		VersionedValue: statedb.VersionedValue{Value: value, Version: version}}, nil
+		CompositeKey: statedb.CompositeKey{Namespace: scanner.namespace, Key: key},
+		// TODO remove dereferrencing below by changing the type of the field
+		// `VersionedValue` in `statedb.VersionedKV` to a pointer
+		VersionedValue: *vv}, nil
 }
 
 func (scanner *kvScanner) Close() {

core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb.go

@@ -56,18 +56,6 @@ func TestIterator(t *testing.T) {
 	commontests.TestIterator(t, env.DBProvider)
 }
 
-func TestEncodeDecodeValueAndVersion(t *testing.T) {
-	testValueAndVersionEncoding(t, []byte("value1"), version.NewHeight(1, 2))
-	testValueAndVersionEncoding(t, []byte{}, version.NewHeight(50, 50))
-}
-
-func testValueAndVersionEncoding(t *testing.T, value []byte, version *version.Height) {
-	encodedValue := EncodeValue(value, version)
-	val, ver := DecodeValue(encodedValue)
-	testutil.AssertEquals(t, val, value)
-	testutil.AssertEquals(t, ver, version)
-}
-
 func TestCompositeKey(t *testing.T) {
 	testCompositeKey(t, "ledger1", "ns", "key")
 	testCompositeKey(t, "ledger2", "ns", "")