FAB-1405 Ledger History query framework

Ledger History queries and API implementation.
The initial implementation will be to get the
history of transactions for a given key.

Change-Id: Ic2eb4bb6d4a53a534f434c0e4d81fed8f919a7d0
Signed-off-by: Mari Wade <[email protected]>

Author: mariwade

core/ledger/history/couchdb_histmgr.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"strconv"
 
+	"github.com/hyperledger/fabric/core/ledger"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/rwset"
 	"github.com/hyperledger/fabric/core/ledger/util/couchdb"
 	"github.com/hyperledger/fabric/protos/common"
@@ -29,6 +30,8 @@ import (
 
 var logger = logging.MustGetLogger("history")
 
+var compositeKeySep = []byte{0x00}
+
 // CouchDBHistMgr a simple implementation of interface `histmgmt.HistMgr'.
 // TODO This implementation does not currently use a lock but may need one to ensure query's are consistent
 type CouchDBHistMgr struct {
@@ -49,8 +52,14 @@ func NewCouchDBHistMgr(couchDBConnectURL string, dbName string, id string, pw st
 	return &CouchDBHistMgr{couchDB: couchDB}
 }
 
+// NewHistoryQueryExecutor implements method in interface `histmgmt.HistMgr'.
+func (histmgr *CouchDBHistMgr) NewHistoryQueryExecutor() (ledger.HistoryQueryExecutor, error) {
+	return &CouchDBHistQueryExecutor{histmgr}, nil
+}
+
 // Commit implements method in interface `histmgmt.HistMgr`
 // This writes to a separate history database.
+// TODO dpending on how invalid transactions are handled may need to filter what history commits.
 func (histmgr *CouchDBHistMgr) Commit(block *common.Block) error {
 	logger.Debugf("===HISTORYDB=== Entering CouchDBHistMgr.Commit()")
 
@@ -124,6 +133,22 @@ func (histmgr *CouchDBHistMgr) Commit(block *common.Block) error {
 	return nil
 }
 
+//getTransactionsForNsKey contructs composite start and end keys based on the namespace and key then calls the CouchDB range scanner
+func (histmgr *CouchDBHistMgr) getTransactionsForNsKey(namespace string, key string, includeValues bool) (*histScanner, error) {
+	var compositeStartKey []byte
+	var compositeEndKey []byte
+	if key != "" {
+		compositeStartKey = constructPartialCompositeKey(namespace, key, false)
+		compositeEndKey = constructPartialCompositeKey(namespace, key, true)
+	}
+
+	//TODO the limit should not be hardcoded.  Need the config.
+	//TODO Implement includeValues so that values are not returned in the readDocRange
+	queryResult, _ := histmgr.couchDB.ReadDocRange(string(compositeStartKey), string(compositeEndKey), 1000, 0)
+
+	return newHistScanner(compositeStartKey, *queryResult), nil
+}
+
 func constructCompositeKey(ns string, key string, blocknum uint64, trannum uint64) string {
 	//History Key is:  "namespace key blocknum trannum"", with namespace being the chaincode id
 
@@ -139,3 +164,53 @@ func constructCompositeKey(ns string, key string, blocknum uint64, trannum uint6
 
 	return buffer.String()
 }
+
+func constructPartialCompositeKey(ns string, key string, endkey bool) []byte {
+	compositeKey := []byte(ns)
+	compositeKey = append(compositeKey, compositeKeySep...)
+	compositeKey = append(compositeKey, []byte(key)...)
+	if endkey {
+		compositeKey = append(compositeKey, []byte("1")...)
+	}
+	return compositeKey
+}
+
+func splitCompositeKey(compositePartialKey []byte, compositeKey []byte) (string, string) {
+	split := bytes.SplitN(compositeKey, compositePartialKey, 2)
+	return string(split[0]), string(split[1])
+}
+
+type histScanner struct {
+	cursor              int
+	compositePartialKey []byte
+	results             []couchdb.QueryResult
+}
+
+type historicValue struct {
+	blockNumTranNum string
+	value           []byte
+}
+
+func newHistScanner(compositePartialKey []byte, queryResults []couchdb.QueryResult) *histScanner {
+	return &histScanner{-1, compositePartialKey, queryResults}
+}
+
+func (scanner *histScanner) next() (*historicValue, error) {
+
+	scanner.cursor++
+
+	if scanner.cursor >= len(scanner.results) {
+		return nil, nil
+	}
+
+	selectedValue := scanner.results[scanner.cursor]
+
+	_, blockNumTranNum := splitCompositeKey(scanner.compositePartialKey, []byte(selectedValue.ID))
+
+	return &historicValue{blockNumTranNum, selectedValue.Value}, nil
+
+}
+
+func (scanner *histScanner) close() {
+	scanner = nil
+}

core/ledger/history/couchdb_histmgr_test.go

@@ -35,7 +35,7 @@ func TestHistoryDatabaseAutoCreate(t *testing.T) {
 
 	//call a helper method to load the core.yaml
 	testutil.SetupCoreYAMLConfig("./../../../peer")
-	logger.Debugf("===HISTORYDB=== TestHistoryDatabaseAutoCreate  IsCouchDBEnabled()value: %v , IsHistoryDBEnabled()value: %v\n",
+	logger.Debugf("===HISTORYDB=== TestHistoryDatabaseAutoCreate IsCouchDBEnabled()value: %v , IsHistoryDBEnabled()value: %v\n",
 		ledgerconfig.IsCouchDBEnabled(), ledgerconfig.IsHistoryDBEnabled())
 
 	if ledgerconfig.IsHistoryDBEnabled() == true {
@@ -82,3 +82,17 @@ func TestConstructCompositeKey(t *testing.T) {
 
 	testutil.AssertEquals(t, compositeKey, "ns1"+strKeySep+"key1"+strKeySep+"1"+strKeySep+"1")
 }
+
+//History Database commit and read is being tested with kv_ledger_test.go.
+//This test will push some of the testing down into history itself
+func TestHistoryDatabaseCommit(t *testing.T) {
+	//call a helper method to load the core.yaml
+	testutil.SetupCoreYAMLConfig("./../../../peer")
+	logger.Debugf("===HISTORYDB=== TestHistoryDatabaseCommit  IsCouchDBEnabled()value: %v , IsHistoryDBEnabled()value: %v\n",
+		ledgerconfig.IsCouchDBEnabled(), ledgerconfig.IsHistoryDBEnabled())
+
+	if ledgerconfig.IsHistoryDBEnabled() == true {
+		//TODO Build the necessary infrastructure so that history can be tested iwthout ledger
+
+	}
+}

core/ledger/history/couchdb_history_query_executer.go

@@ -0,0 +1,56 @@
+/*
+Copyright IBM Corp. 2016 All Rights Reserved.
+
+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 history
+
+import "github.com/hyperledger/fabric/core/ledger"
+
+// CouchDBHistQueryExecutor is a query executor used in `CouchDBHistMgr`
+type CouchDBHistQueryExecutor struct {
+	histmgr *CouchDBHistMgr
+}
+
+// GetTransactionsForKey implements method in interface `ledger.HistoryQueryExecutor`
+func (q *CouchDBHistQueryExecutor) GetTransactionsForKey(namespace string, key string, includeValues bool, includeTransactions bool) (ledger.ResultsIterator, error) {
+	//TODO  includeTransactions has not been implemented yet.
+	scanner, err := q.histmgr.getTransactionsForNsKey(namespace, key, includeValues)
+	if err != nil {
+		return nil, err
+	}
+	return &qHistoryItr{scanner}, nil
+}
+
+type qHistoryItr struct {
+	q *histScanner
+}
+
+// Next implements Next() method in ledger.ResultsIterator
+func (itr *qHistoryItr) Next() (ledger.QueryResult, error) {
+	historicValue, err := itr.q.next()
+	if err != nil {
+		return nil, err
+	}
+	if historicValue == nil {
+		return nil, nil
+	}
+	//TODO Returning blockNumTrannum as TxID for now but eventually will return txID instead
+	return &ledger.KeyModification{TxID: historicValue.blockNumTranNum, Value: historicValue.value}, nil
+}
+
+// Close implements Close() method in ledger.ResultsIterator
+func (itr *qHistoryItr) Close() {
+	itr.q.close()
+}

core/ledger/history/histmgmt.go

@@ -17,8 +17,10 @@ limitations under the License.
 package history
 
 import "github.com/hyperledger/fabric/protos/common"
+import "github.com/hyperledger/fabric/core/ledger"
 
 // HistMgr - an interface that a history manager should implement
 type HistMgr interface {
+	NewHistoryQueryExecutor() (ledger.HistoryQueryExecutor, error)
 	Commit(block *common.Block) error
 }

core/ledger/kvledger/kv_ledger.go

@@ -204,13 +204,20 @@ func (l *KVLedger) NewTxSimulator() (ledger.TxSimulator, error) {
 	return l.txtmgmt.NewTxSimulator()
 }
 
-// NewQueryExecutor gives handle to a query executer.
+// NewQueryExecutor gives handle to a query executor.
 // A client can obtain more than one 'QueryExecutor's for parallel execution.
 // Any synchronization should be performed at the implementation level if required
 func (l *KVLedger) NewQueryExecutor() (ledger.QueryExecutor, error) {
 	return l.txtmgmt.NewQueryExecutor()
 }
 
+// NewHistoryQueryExecutor gives handle to a history query executor.
+// A client can obtain more than one 'HistoryQueryExecutor's for parallel execution.
+// Any synchronization should be performed at the implementation level if required
+func (l *KVLedger) NewHistoryQueryExecutor() (ledger.HistoryQueryExecutor, error) {
+	return l.historymgmt.NewHistoryQueryExecutor()
+}
+
 // Commit commits the valid block (returned in the method RemoveInvalidTransactionsAndPrepare) and related state changes
 func (l *KVLedger) Commit(block *common.Block) error {
 	var err error
@@ -231,8 +238,10 @@ func (l *KVLedger) Commit(block *common.Block) error {
 		panic(fmt.Errorf(`Error during commit to txmgr:%s`, err))
 	}
 
-	//TODO future will want to run async with state db writes.  History needs to wait for chain (FSBlock) to write but not the state db
-	logger.Debugf("===HISTORYDB=== Commit() will write to hisotry if enabled else will be by-passed if not enabled: vledgerconfig.IsHistoryDBEnabled(): %v\n", ledgerconfig.IsHistoryDBEnabled())
+	//TODO There are still some decisions to be made regarding how consistent history
+	//needs to stay with the state.  At min will want this to run async with state db writes.
+	//(History needs to wait for the block (FSBlock) to write but not the state db)
+	logger.Debugf("===HISTORYDB=== Commit() will write to history if enabled else will be by-passed if not enabled: vledgerconfig.IsHistoryDBEnabled(): %v\n", ledgerconfig.IsHistoryDBEnabled())
 	if ledgerconfig.IsHistoryDBEnabled() == true {
 		logger.Debugf("Committing transactions to history database")
 		if err := l.historymgmt.Commit(block); err != nil {

core/ledger/kvledger/kv_ledger_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package kvledger
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/hyperledger/fabric/core/ledger/ledgerconfig"
@@ -170,6 +171,8 @@ func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {
 	ledger, _ := NewKVLedger(env.conf)
 	defer ledger.Close()
 
+	//testNs := "ns1"
+
 	bcInfo, _ := ledger.GetBlockchainInfo()
 	testutil.AssertEquals(t, bcInfo, &pb.BlockchainInfo{
 		Height: 0, CurrentBlockHash: nil, PreviousBlockHash: nil})
@@ -204,6 +207,7 @@ func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {
 	simulationResults = append(simulationResults, simRes)
 	//add a 2nd transaction
 	simulator2, _ := ledger.NewTxSimulator()
+	simulator2.SetState("ns1", "key7", []byte("{\"shipmentID\":\"161003PKC7600\",\"customsInvoice\":{\"methodOfTransport\":\"TRAIN\",\"invoiceNumber\":\"00091624\"},\"weightUnitOfMeasure\":\"KGM\",\"volumeUnitOfMeasure\": \"CO\",\"dimensionUnitOfMeasure\":\"CM\",\"currency\":\"USD\"}"))
 	simulator2.SetState("ns1", "key9", []byte("value5"))
 	simulator2.SetState("ns1", "key10", []byte("{\"shipmentID\":\"261003PKC8000\",\"customsInvoice\":{\"methodOfTransport\":\"DONKEY\",\"invoiceNumber\":\"00091626\"},\"weightUnitOfMeasure\":\"KGM\",\"volumeUnitOfMeasure\": \"CO\",\"dimensionUnitOfMeasure\":\"CM\",\"currency\":\"USD\"}"))
 	simulator2.Done()
@@ -230,7 +234,27 @@ func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {
 	b2, _ = ledger.GetBlockByNumber(2)
 	testutil.AssertEquals(t, b2, block2)
 
+	//TODO move this test to history.
 	if ledgerconfig.IsHistoryDBEnabled() == true {
-		//TODO history specific test once the query api's are in and we can validate content
+		qhistory, err := ledger.NewHistoryQueryExecutor()
+		testutil.AssertNoError(t, err, fmt.Sprintf("Error when trying to retrieve history database executor"))
+
+		itr, err2 := qhistory.GetTransactionsForKey("ns1", "key7", true, false)
+		testutil.AssertNoError(t, err2, fmt.Sprintf("Error when trying to retrieve history database executor"))
+
+		count := 0
+		for {
+			kmod, _ := itr.Next()
+			if kmod == nil {
+				break
+			}
+			//TODO TEST CONTENT - need to point to ledger import and not the KVledger
+			//TODO MOVE TEST TO HISTORY
+			//bt := kmod.(*ledger.KeyModification).TxID
+			//v := kmod.(*ledger.KeyModification).Value
+			//t.Logf("Retrieved for ns=%s, key=key7  : k=%s, v=%s at count=%d start=%s end=%s", testNs, bt, v, count)
+			count++
+		}
+		testutil.AssertEquals(t, count, 3)
 	}
 }

core/ledger/kvledger/txmgmt/couchdbtxmgmt/couchdb_query_executer.go

@@ -62,11 +62,6 @@ func (q *CouchDBQueryExecutor) GetStateRangeScanIterator(namespace string, start
 	return &qKVItr{scanner}, nil
 }
 
-// GetTransactionsForKey - implements method in interface `ledger.QueryExecutor`
-func (q *CouchDBQueryExecutor) GetTransactionsForKey(namespace string, key string) (ledger.ResultsIterator, error) {
-	return nil, errors.New("Not yet implemented")
-}
-
 // ExecuteQuery implements method in interface `ledger.QueryExecutor`
 func (q *CouchDBQueryExecutor) ExecuteQuery(query string) (ledger.ResultsIterator, error) {
 	return nil, errors.New("Not supported by KV data model")

core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_query_executer.go

@@ -17,8 +17,6 @@ limitations under the License.
 package lockbasedtxmgr
 
 import (
-	"errors"
-
 	"github.com/hyperledger/fabric/core/ledger"
 	"github.com/hyperledger/fabric/core/util"
 )
@@ -54,11 +52,6 @@ func (q *lockBasedQueryExecutor) GetStateRangeScanIterator(namespace string, sta
 	return q.helper.getStateRangeScanIterator(namespace, startKey, endKey)
 }
 
-// GetTransactionsForKey - implements method in interface `ledger.QueryExecutor`
-func (q *lockBasedQueryExecutor) GetTransactionsForKey(namespace string, key string) (ledger.ResultsIterator, error) {
-	return nil, errors.New("Not yet implemented")
-}
-
 // ExecuteQuery implements method in interface `ledger.QueryExecutor`
 func (q *lockBasedQueryExecutor) ExecuteQuery(query string) (ledger.ResultsIterator, error) {
 	return q.helper.executeQuery(query)

core/ledger/ledger_interface.go

@@ -57,10 +57,14 @@ type ValidatedLedger interface {
 	// A client can obtain more than one 'TxSimulator's for parallel execution.
 	// Any snapshoting/synchronization should be performed at the implementation level if required
 	NewTxSimulator() (TxSimulator, error)
-	// NewQueryExecuter gives handle to a query executor.
+	// NewQueryExecutor gives handle to a query executor.
 	// A client can obtain more than one 'QueryExecutor's for parallel execution.
 	// Any synchronization should be performed at the implementation level if required
 	NewQueryExecutor() (QueryExecutor, error)
+	// NewHistoryQueryExecutor gives handle to a history query executor.
+	// A client can obtain more than one 'HistoryQueryExecutor's for parallel execution.
+	// Any synchronization should be performed at the implementation level if required
+	NewHistoryQueryExecutor() (HistoryQueryExecutor, error)
 	// Commits block into the ledger
 	Commit(block *common.Block) error
 }
@@ -78,15 +82,18 @@ type QueryExecutor interface {
 	// GetStateRangeScanIterator returns an iterator that contains all the key-values between given key ranges.
 	// The returned ResultsIterator contains results of type *KV
 	GetStateRangeScanIterator(namespace string, startKey string, endKey string) (ResultsIterator, error)
-	// GetTransactionsForKey returns an iterator that contains all the transactions that modified the given key.
-	// The returned ResultsIterator contains results of type *msgs.Transaction
-	GetTransactionsForKey(namespace string, key string) (ResultsIterator, error)
 	// ExecuteQuery executes the given query and returns an iterator that contains results of type specific to the underlying data store.
 	ExecuteQuery(query string) (ResultsIterator, error)
 	// Done releases resources occupied by the QueryExecutor
 	Done()
 }
 
+// HistoryQueryExecutor executes the history queries
+type HistoryQueryExecutor interface {
+	// GetTransactionsForKey retrieves the set of transactons that updated this key by doing a key range query.
+	GetTransactionsForKey(namespace string, key string, includeValues bool, includeTransactions bool) (ResultsIterator, error)
+}
+
 // TxSimulator simulates a transaction on a consistent snapshot of the 'as recent state as possible'
 // Set* methods are for supporting KV-based data model. ExecuteUpdate method is for supporting a rich datamodel and query support
 type TxSimulator interface {
@@ -128,6 +135,13 @@ type KV struct {
 	Value []byte
 }
 
+// KeyModification - QueryResult for History.
+type KeyModification struct {
+	TxID        string
+	Value       []byte
+	Transaction *pb.Transaction
+}
+
 // BlockHolder holds block returned by the iterator in GetBlocksIterator.
 // The sole purpose of this holder is to avoid desrialization if block is desired in raw bytes form (e.g., for transfer)
 type BlockHolder interface {