[FAB-6176] Couchdb index management state database

Add couchdb index management capabilities to the state database layer.

Add create and delete index functions to statecouchdb.

Add HandleChaincodeDefinition to statecouchdb.

Add unit tests for create and save index functions.

Add unit tests for HandleChaincodeDefinition.

Change-Id: I82f31ced8eb666e607c549f995d0c83fc7364664
Signed-off-by: Chris Elder <chris.elder@us.ibm.com>

Author: Chris Elder

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

@@ -91,6 +91,7 @@ func TestBasicRW(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 	sp, err = db.GetLatestSavePoint()
 	testutil.AssertNoError(t, err, "")
 	testutil.AssertEquals(t, sp, savePoint)
+
 }
 
 // TestMultiDBBasicRW tests basic read-write on multiple dbs

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

@@ -6,16 +6,21 @@ SPDX-License-Identifier: Apache-2.0
 package statecouchdb
 
 import (
+	"archive/tar"
 	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"io/ioutil"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
 	"unicode/utf8"
 
 	"github.com/hyperledger/fabric/common/flogging"
+	"github.com/hyperledger/fabric/core/ledger/cceventmgmt"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/version"
 	"github.com/hyperledger/fabric/core/ledger/ledgerconfig"
@@ -67,6 +72,68 @@ func NewVersionedDBProvider() (*VersionedDBProvider, error) {
 	return &VersionedDBProvider{couchInstance, make(map[string]*VersionedDB), sync.Mutex{}, 0}, nil
 }
 
+//HandleChaincodeDeploy initializes database artifacts for the database associated with the namespace
+func (vdb *VersionedDB) HandleChaincodeDeploy(chaincodeDefinition *cceventmgmt.ChaincodeDefinition, dbArtifactsTar []byte) error {
+
+	logger.Debugf("Entering HandleChaincodeDeploy")
+
+	if chaincodeDefinition == nil {
+		return fmt.Errorf("chaincodeDefinition must not be nil")
+	}
+
+	db, err := vdb.getNamespaceDBHandle(chaincodeDefinition.Name)
+	if err != nil {
+		return err
+	}
+
+	//initialize a reader for the artifacts tar file
+	artifactReader := bytes.NewReader(dbArtifactsTar)
+	tarReader := tar.NewReader(artifactReader)
+
+	for {
+
+		//read the next header from the tar
+		tarHeader, err := tarReader.Next()
+
+		//if the EOF is detected, then exit
+		if err == io.EOF {
+			// end of tar archive
+			break
+		}
+		if err != nil {
+			return err
+		}
+
+		logger.Debugf("Reading artifact from file: %s", tarHeader.Name)
+
+		//Ensure that this is not a directory
+		if !tarHeader.FileInfo().IsDir() {
+
+			//split the filename into directory and file name
+			dir, file := filepath.Split(tarHeader.Name)
+
+			if dir == "META-INF/statedb/couchdb/indexes/" {
+
+				logger.Debugf("Creating index from file file: %s", file)
+
+				//read the tar entry into a byte array
+				indexData, err := ioutil.ReadAll(tarReader)
+				if err != nil {
+					return err
+				}
+
+				//create the index from the tar entry
+				err = db.CreateIndex(string(indexData))
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
 // GetDBHandle gets the handle to a named database
 func (provider *VersionedDBProvider) GetDBHandle(dbName string) (statedb.VersionedDB, error) {
 

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

@@ -17,11 +17,15 @@ limitations under the License.
 package statecouchdb
 
 import (
+	"archive/tar"
+	"bytes"
+	"log"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/hyperledger/fabric/common/ledger/testutil"
+	"github.com/hyperledger/fabric/core/ledger/cceventmgmt"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb/commontests"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/version"
@@ -58,33 +62,51 @@ func TestMain(m *testing.M) {
 
 func TestBasicRW(t *testing.T) {
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testbasicrw")
-	defer env.Cleanup("testbasicrw")
+	env.Cleanup("testbasicrw_")
+	env.Cleanup("testbasicrw_ns")
+	env.Cleanup("testbasicrw_ns1")
+	env.Cleanup("testbasicrw_ns2")
+	defer env.Cleanup("testbasicrw_")
+	defer env.Cleanup("testbasicrw_ns")
+	defer env.Cleanup("testbasicrw_ns1")
+	defer env.Cleanup("testbasicrw_ns2")
 	commontests.TestBasicRW(t, env.DBProvider)
 
 }
 
 func TestMultiDBBasicRW(t *testing.T) {
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testmultidbbasicrw")
-	env.Cleanup("testmultidbbasicrw2")
-	defer env.Cleanup("testmultidbbasicrw")
-	defer env.Cleanup("testmultidbbasicrw2")
+	env.Cleanup("testmultidbbasicrw_")
+	env.Cleanup("testmultidbbasicrw_ns1")
+	env.Cleanup("testmultidbbasicrw2_")
+	env.Cleanup("testmultidbbasicrw2_ns1")
+	defer env.Cleanup("testmultidbbasicrw_")
+	defer env.Cleanup("testmultidbbasicrw_ns1")
+	defer env.Cleanup("testmultidbbasicrw2_")
+	defer env.Cleanup("testmultidbbasicrw2_ns1")
 	commontests.TestMultiDBBasicRW(t, env.DBProvider)
 
 }
 
 func TestDeletes(t *testing.T) {
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testdeletes")
-	defer env.Cleanup("testdeletes")
+	env.Cleanup("testdeletes_")
+	env.Cleanup("testdeletes_ns")
+	defer env.Cleanup("testdeletes_")
+	defer env.Cleanup("testdeletes_ns")
 	commontests.TestDeletes(t, env.DBProvider)
 }
 
 func TestIterator(t *testing.T) {
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testiterator")
-	defer env.Cleanup("testiterator")
+	env.Cleanup("testiterator_")
+	env.Cleanup("testiterator_ns1")
+	env.Cleanup("testiterator_ns2")
+	env.Cleanup("testiterator_ns3")
+	defer env.Cleanup("testiterator_")
+	defer env.Cleanup("testiterator_ns1")
+	defer env.Cleanup("testiterator_ns2")
+	defer env.Cleanup("testiterator_ns3")
 	commontests.TestIterator(t, env.DBProvider)
 }
 
@@ -104,48 +126,68 @@ func testValueAndVersionEncoding(t *testing.T, value []byte, version *version.He
 //  query test
 func TestQuery(t *testing.T) {
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testquery")
-	defer env.Cleanup("testquery")
+	env.Cleanup("testquery_")
+	env.Cleanup("testquery_ns1")
+	env.Cleanup("testquery_ns2")
+	env.Cleanup("testquery_ns3")
+	defer env.Cleanup("testquery_")
+	defer env.Cleanup("testquery_ns1")
+	defer env.Cleanup("testquery_ns2")
+	defer env.Cleanup("testquery_ns3")
 	commontests.TestQuery(t, env.DBProvider)
 }
 
 func TestGetStateMultipleKeys(t *testing.T) {
 
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testgetmultiplekeys")
-	defer env.Cleanup("testgetmultiplekeys")
+	env.Cleanup("testgetmultiplekeys_")
+	env.Cleanup("testgetmultiplekeys_ns1")
+	env.Cleanup("testgetmultiplekeys_ns2")
+	defer env.Cleanup("testgetmultiplekeys_")
+	defer env.Cleanup("testgetmultiplekeys_ns1")
+	defer env.Cleanup("testgetmultiplekeys_ns2")
 	commontests.TestGetStateMultipleKeys(t, env.DBProvider)
 }
 
 func TestGetVersion(t *testing.T) {
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testgetversion")
-	defer env.Cleanup("testgetversion")
+	env.Cleanup("testgetversion_")
+	env.Cleanup("testgetversion_ns")
+	env.Cleanup("testgetversion_ns2")
+	defer env.Cleanup("testgetversion_")
+	defer env.Cleanup("testgetversion_ns")
+	defer env.Cleanup("testgetversion_ns2")
 	commontests.TestGetVersion(t, env.DBProvider)
 }
 
 func TestSmallBatchSize(t *testing.T) {
 	viper.Set("ledger.state.couchDBConfig.maxBatchUpdateSize", 2)
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testsmallbatchsize")
-	defer env.Cleanup("testsmallbatchsize")
+	env.Cleanup("testsmallbatchsize_")
+	env.Cleanup("testsmallbatchsize_ns1")
+	defer env.Cleanup("testsmallbatchsize_")
+	defer env.Cleanup("testsmallbatchsize_ns1")
 	defer viper.Set("ledger.state.couchDBConfig.maxBatchUpdateSize", 1000)
 	commontests.TestSmallBatchSize(t, env.DBProvider)
 }
 
 func TestBatchRetry(t *testing.T) {
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testbatchretry")
-	defer env.Cleanup("testbatchretry")
+	env.Cleanup("testbatchretry_")
+	env.Cleanup("testbatchretry_ns")
+	env.Cleanup("testbatchretry_ns1")
+	defer env.Cleanup("testbatchretry_")
+	defer env.Cleanup("testbatchretry_ns")
+	defer env.Cleanup("testbatchretry_ns1")
 	commontests.TestBatchWithIndividualRetry(t, env.DBProvider)
 }
 
 // TestUtilityFunctions tests utility functions
 func TestUtilityFunctions(t *testing.T) {
 
 	env := NewTestVDBEnv(t)
-	env.Cleanup("testutilityfunctions")
-	defer env.Cleanup("testutilityfunctions")
+	env.Cleanup("testutilityfunctions_")
+	defer env.Cleanup("testutilityfunctions_")
 
 	db, err := env.DBProvider.GetDBHandle("testutilityfunctions")
 	testutil.AssertNoError(t, err, "")
@@ -177,3 +219,134 @@ func TestDebugFunctions(t *testing.T) {
 	testutil.AssertEquals(t, printCompositeKeys(loadKeys), "[ns,key4],[ns,key4]")
 
 }
+
+func TestHandleChaincodeDeploy(t *testing.T) {
+
+	env := NewTestVDBEnv(t)
+	env.Cleanup("testinit_")
+	env.Cleanup("testinit_ns1")
+	env.Cleanup("testinit_ns2")
+	defer env.Cleanup("testinit_")
+	defer env.Cleanup("testinit_ns1")
+	defer env.Cleanup("testinit_ns2")
+
+	db, err := env.DBProvider.GetDBHandle("testinit")
+	testutil.AssertNoError(t, err, "")
+	db.Open()
+	defer db.Close()
+	batch := statedb.NewUpdateBatch()
+
+	jsonValue1 := "{\"asset_name\": \"marble1\",\"color\": \"blue\",\"size\": 1,\"owner\": \"tom\"}"
+	batch.Put("ns1", "key1", []byte(jsonValue1), version.NewHeight(1, 1))
+	jsonValue2 := "{\"asset_name\": \"marble2\",\"color\": \"blue\",\"size\": 2,\"owner\": \"jerry\"}"
+	batch.Put("ns1", "key2", []byte(jsonValue2), version.NewHeight(1, 2))
+	jsonValue3 := "{\"asset_name\": \"marble3\",\"color\": \"blue\",\"size\": 3,\"owner\": \"fred\"}"
+	batch.Put("ns1", "key3", []byte(jsonValue3), version.NewHeight(1, 3))
+	jsonValue4 := "{\"asset_name\": \"marble4\",\"color\": \"blue\",\"size\": 4,\"owner\": \"martha\"}"
+	batch.Put("ns1", "key4", []byte(jsonValue4), version.NewHeight(1, 4))
+	jsonValue5 := "{\"asset_name\": \"marble5\",\"color\": \"blue\",\"size\": 5,\"owner\": \"fred\"}"
+	batch.Put("ns1", "key5", []byte(jsonValue5), version.NewHeight(1, 5))
+	jsonValue6 := "{\"asset_name\": \"marble6\",\"color\": \"blue\",\"size\": 6,\"owner\": \"elaine\"}"
+	batch.Put("ns1", "key6", []byte(jsonValue6), version.NewHeight(1, 6))
+	jsonValue7 := "{\"asset_name\": \"marble7\",\"color\": \"blue\",\"size\": 7,\"owner\": \"fred\"}"
+	batch.Put("ns1", "key7", []byte(jsonValue7), version.NewHeight(1, 7))
+	jsonValue8 := "{\"asset_name\": \"marble8\",\"color\": \"blue\",\"size\": 8,\"owner\": \"elaine\"}"
+	batch.Put("ns1", "key8", []byte(jsonValue8), version.NewHeight(1, 8))
+	jsonValue9 := "{\"asset_name\": \"marble9\",\"color\": \"green\",\"size\": 9,\"owner\": \"fred\"}"
+	batch.Put("ns1", "key9", []byte(jsonValue9), version.NewHeight(1, 9))
+	jsonValue10 := "{\"asset_name\": \"marble10\",\"color\": \"green\",\"size\": 10,\"owner\": \"mary\"}"
+	batch.Put("ns1", "key10", []byte(jsonValue10), version.NewHeight(1, 10))
+	jsonValue11 := "{\"asset_name\": \"marble11\",\"color\": \"cyan\",\"size\": 1000007,\"owner\": \"joe\"}"
+	batch.Put("ns1", "key11", []byte(jsonValue11), version.NewHeight(1, 11))
+
+	//add keys for a separate namespace
+	batch.Put("ns2", "key1", []byte(jsonValue1), version.NewHeight(1, 12))
+	batch.Put("ns2", "key2", []byte(jsonValue2), version.NewHeight(1, 13))
+	batch.Put("ns2", "key3", []byte(jsonValue3), version.NewHeight(1, 14))
+	batch.Put("ns2", "key4", []byte(jsonValue4), version.NewHeight(1, 15))
+	batch.Put("ns2", "key5", []byte(jsonValue5), version.NewHeight(1, 16))
+	batch.Put("ns2", "key6", []byte(jsonValue6), version.NewHeight(1, 17))
+	batch.Put("ns2", "key7", []byte(jsonValue7), version.NewHeight(1, 18))
+	batch.Put("ns2", "key8", []byte(jsonValue8), version.NewHeight(1, 19))
+	batch.Put("ns2", "key9", []byte(jsonValue9), version.NewHeight(1, 20))
+	batch.Put("ns2", "key10", []byte(jsonValue10), version.NewHeight(1, 21))
+
+	savePoint := version.NewHeight(2, 22)
+	db.ApplyUpdates(batch, savePoint)
+
+	//Create a buffer for the tar file
+	buffer := new(bytes.Buffer)
+	tarWriter := tar.NewWriter(buffer)
+
+	//Add 2 index definitions
+	var files = []struct {
+		Name, Body string
+	}{
+		{"META-INF/statedb/couchdb/indexes/indexColorSortName.json", "{\"index\":{\"fields\":[{\"data.color\":\"desc\"}]},\"ddoc\":\"indexColorSortName\",\"name\":\"indexColorSortName\",\"type\":\"json\"}"},
+		{"META-INF/statedb/couchdb/indexes/indexSizeSortName.json", "{\"index\":{\"fields\":[{\"data.size\":\"desc\"}]},\"ddoc\":\"indexSizeSortName\",\"name\":\"indexSizeSortName\",\"type\":\"json\"}"},
+	}
+	for _, file := range files {
+		tarHeader := &tar.Header{
+			Name: file.Name,
+			Mode: 0600,
+			Size: int64(len(file.Body)),
+		}
+		err := tarWriter.WriteHeader(tarHeader)
+		testutil.AssertNoError(t, err, "")
+
+		_, err = tarWriter.Write([]byte(file.Body))
+		testutil.AssertNoError(t, err, "")
+
+	}
+	// Make sure to check the error on Close.
+	if err := tarWriter.Close(); err != nil {
+		log.Fatalln(err)
+	}
+
+	//Create a query
+	queryString := `{"selector":{"owner":"fred"}}`
+
+	_, err = db.ExecuteQuery("ns1", queryString)
+	testutil.AssertNoError(t, err, "")
+
+	//Create a query with a sort
+	queryString = `{"selector":{"owner":"fred"}, "sort": [{"size": "desc"}]}`
+
+	_, err = db.ExecuteQuery("ns1", queryString)
+	testutil.AssertError(t, err, "Error should have been thrown for a missing index")
+
+	if handleDefinition, ok := db.(cceventmgmt.ChaincodeLifecycleEventListener); ok {
+
+		chaincodeDef := &cceventmgmt.ChaincodeDefinition{Name: "ns1", Hash: nil, Version: ""}
+
+		//Test HandleChaincodeDefinition with a valid tar file
+		err := handleDefinition.HandleChaincodeDeploy(chaincodeDef, buffer.Bytes())
+		testutil.AssertNoError(t, err, "")
+
+		//Test HandleChaincodeDefinition with a nil tar file
+		err = handleDefinition.HandleChaincodeDeploy(chaincodeDef, nil)
+		testutil.AssertNoError(t, err, "")
+
+		//Test HandleChaincodeDefinition with a bad tar file
+		err = handleDefinition.HandleChaincodeDeploy(chaincodeDef, []byte(`This is a really bad tar file`))
+		testutil.AssertError(t, err, "Error should have been thrown for a bad tar file")
+
+		//Test HandleChaincodeDefinition with a nil chaincodeDef
+		err = handleDefinition.HandleChaincodeDeploy(nil, []byte(`This is a really bad tar file`))
+		testutil.AssertError(t, err, "Error should have been thrown for a nil chaincodeDefinition")
+
+		//Sleep to allow time for index creation
+		time.Sleep(100 * time.Millisecond)
+		//Create a query with a sort
+		queryString = `{"selector":{"owner":"fred"}, "sort": [{"size": "desc"}]}`
+
+		//Query should complete without error
+		_, err = db.ExecuteQuery("ns1", queryString)
+		testutil.AssertNoError(t, err, "")
+
+		//Query namespace "ns2", index is only created in "ns1".  This should return an error.
+		_, err = db.ExecuteQuery("ns2", queryString)
+		testutil.AssertError(t, err, "Error should have been thrown for a missing index")
+
+	}
+}