[FAB-6053] Enhance transient store retrieval api

This CR enhances the transient store retrieval api with an additional
parameter `filter` that acts as a projection for the results and retains
only namespace/collection specified in the filter

Change-Id: I2dd20f4382f1566a65c0bd23689c77a0ec60de24
Signed-off-by: manish <manish.sethi@gmail.com>
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: manish-sethi

core/endorser/endorser.go

@@ -9,6 +9,8 @@ package endorser
 import (
 	"fmt"
 
+	"github.com/hyperledger/fabric/protos/ledger/rwset"
+
 	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/common/flogging"
 	"golang.org/x/net/context"
@@ -52,7 +54,7 @@ var endorserLogger = flogging.MustGetLogger("endorser")
 // The Jira issue that documents Endorser flow along with its relationship to
 // the lifecycle chaincode - https://jira.hyperledger.org/browse/FAB-181
 
-type privateDataDistributor func(channel string, txID string, privateData []byte) error
+type privateDataDistributor func(channel string, txID string, privateData *rwset.TxPvtReadWriteSet) error
 
 // Endorser provides the Endorser service ProcessProposal
 type Endorser struct {
@@ -261,7 +263,6 @@ func (e *Endorser) simulateProposal(ctx context.Context, chainID string, txid st
 	//---3. execute the proposal and get simulation results
 	var simResult *ledger.TxSimulationResults
 	var pubSimResBytes []byte
-	var prvtSimResBytes []byte
 	var res *pb.Response
 	var ccevent *pb.ChaincodeEvent
 	res, ccevent, err = e.callChaincode(ctx, chainID, version, txid, signedProp, prop, cis, cid, txsim)
@@ -275,16 +276,11 @@ func (e *Endorser) simulateProposal(ctx context.Context, chainID string, txid st
 			return nil, nil, nil, nil, err
 		}
 
-		if prvtSimResBytes, err = simResult.GetPvtSimulationBytes(); err != nil {
-			return nil, nil, nil, nil, err
-		}
-
-		if len(prvtSimResBytes) > 0 {
-			if err := e.distributePrivateData(chainID, txid, prvtSimResBytes); err != nil {
+		if simResult.PvtSimulationResults != nil {
+			if err := e.distributePrivateData(chainID, txid, simResult.PvtSimulationResults); err != nil {
 				return nil, nil, nil, nil, err
 			}
 		}
-
 		if pubSimResBytes, err = simResult.GetPubSimulationBytes(); err != nil {
 			return nil, nil, nil, nil, err
 		}

core/endorser/endorser_test.go

@@ -30,6 +30,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hyperledger/fabric/protos/ledger/rwset"
+
 	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/bccsp"
 	"github.com/hyperledger/fabric/bccsp/factory"
@@ -746,7 +748,7 @@ func TestMain(m *testing.M) {
 		return
 	}
 
-	endorserServer = NewEndorserServer(func(channel string, txID string, privateData []byte) error {
+	endorserServer = NewEndorserServer(func(channel string, txID string, privateData *rwset.TxPvtReadWriteSet) error {
 		return nil
 	})
 

core/ledger/kvledger/example/committer.go

@@ -1,17 +1,7 @@
 /*
-Copyright IBM Corp. 2016 All Rights Reserved.
+Copyright IBM Corp. 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.
+SPDX-License-Identifier: Apache-2.0
 */
 
 package example
@@ -35,7 +25,7 @@ func ConstructCommitter(ledger ledger.PeerLedger) *Committer {
 // Commit commits the block
 func (c *Committer) Commit(rawBlock *common.Block) error {
 	logger.Debugf("Committer validating the block...")
-	if err := c.ledger.Commit(rawBlock); err != nil {
+	if err := c.ledger.CommitWithPvtData(&ledger.BlockAndPvtData{Block: rawBlock}); err != nil {
 		return err
 	}
 	return nil

core/ledger/pvtdatastorage/store_impl.go

@@ -167,7 +167,7 @@ func (s *store) GetPvtDataByBlockNum(blockNum uint64, filter ledger.PvtNsCollFil
 			return nil, err
 		}
 		logger.Debugf("Retrieving pvtdata for bNum=%d, tNum=%d", bNum, tNum)
-		filteredWSet := trimPvtWSet(pvtWSet, filter)
+		filteredWSet := TrimPvtWSet(pvtWSet, filter)
 		pvtData = append(pvtData, &ledger.TxPvtData{SeqInBlock: tNum, WriteSet: filteredWSet})
 	}
 	return pvtData, nil
@@ -246,7 +246,9 @@ func (s *store) getLastCommittedBlockNum() (bool, uint64, error) {
 	return false, decodeBlockNum(v), nil
 }
 
-func trimPvtWSet(pvtWSet *rwset.TxPvtReadWriteSet, filter ledger.PvtNsCollFilter) *rwset.TxPvtReadWriteSet {
+// TrimPvtWSet returns a `TxPvtReadWriteSet` that retains only list of 'ns/collections' supplied in the filter
+// A nil filter does not filter any results and returns the original `pvtWSet` as is
+func TrimPvtWSet(pvtWSet *rwset.TxPvtReadWriteSet, filter ledger.PvtNsCollFilter) *rwset.TxPvtReadWriteSet {
 	if filter == nil {
 		return pvtWSet
 	}

core/transientstore/store.go

@@ -1,26 +1,20 @@
 /*
-Copyright IBM Corp. 2017 All Rights Reserved.
+Copyright IBM Corp. 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.
+SPDX-License-Identifier: Apache-2.0
 */
 
 package transientstore
 
 import (
 	"errors"
 
-	commonledger "github.com/hyperledger/fabric/common/ledger"
+	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/protos/ledger/rwset"
+
 	"github.com/hyperledger/fabric/common/ledger/util/leveldbhelper"
+	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/pvtdatastorage"
 	"github.com/syndtr/goleveldb/leveldb/iterator"
 )
 
@@ -44,10 +38,10 @@ type StoreProvider interface {
 // the permanent storage or the pruning of some data items is enforced by the policy
 type Store interface {
 	// Persist stores the private read-write set of a transaction in the transient store
-	Persist(txid string, endorserid string, endorsementBlkHt uint64, privateSimulationResults []byte) error
+	Persist(txid string, endorserid string, endorsementBlkHt uint64, privateSimulationResults *rwset.TxPvtReadWriteSet) error
 	// GetTxPvtRWSetByTxid returns an iterator due to the fact that the txid may have multiple private
 	// RWSets persisted from different endorsers (via Gossip)
-	GetTxPvtRWSetByTxid(txid string) (commonledger.ResultsIterator, error)
+	GetTxPvtRWSetByTxid(txid string, filter ledger.PvtNsCollFilter) (*rwsetScanner, error)
 	// GetSelfSimulatedTxPvtRWSetByTxid returns the private read-write set generated from the simulation
 	// performed by the peer itself
 	GetSelfSimulatedTxPvtRWSetByTxid(txid string) (*EndorserPvtSimulationResults, error)
@@ -64,7 +58,7 @@ type Store interface {
 type EndorserPvtSimulationResults struct {
 	EndorserID             string
 	EndorsementBlockHeight uint64
-	PvtSimulationResults   []byte
+	PvtSimulationResults   *rwset.TxPvtReadWriteSet
 }
 
 //////////////////////////////////////////////
@@ -85,8 +79,9 @@ type store struct {
 }
 
 type rwsetScanner struct {
-	txid  string
-	dbItr iterator.Iterator
+	txid   string
+	dbItr  iterator.Iterator
+	filter ledger.PvtNsCollFilter
 }
 
 // NewStoreProvider instantiates TransientStoreProvider
@@ -95,16 +90,6 @@ func NewStoreProvider() StoreProvider {
 	return &storeProvider{dbProvider: dbProvider}
 }
 
-// NewCustomPathStoreProvider constructs a StoreProvider at the given path
-// This function is used by ledger to construct a dedicated store till the time the store is managed
-// by the transient coordinator. A separate store location by the ledger is desired so that
-// the coordinator can be developed independently to handle the management of the transient store at default location.
-// Once the coordinator is developed, the ledger can stop using the transient store and this function can be removed.
-func NewCustomPathStoreProvider(path string) StoreProvider {
-	dbProvider := leveldbhelper.NewProvider(&leveldbhelper.Conf{DBPath: path})
-	return &storeProvider{dbProvider: dbProvider}
-}
-
 // OpenStore returns a handle to a ledgerId in Store
 func (provider *storeProvider) OpenStore(ledgerID string) (Store, error) {
 	dbHandle := provider.dbProvider.GetDBHandle(ledgerID)
@@ -118,12 +103,16 @@ func (provider *storeProvider) Close() {
 
 // Persist stores the private read-write set of a transaction in the transient store
 func (s *store) Persist(txid string, endorserid string,
-	endorsementBlkHt uint64, privateSimulationResults []byte) error {
+	endorsementBlkHt uint64, privateSimulationResults *rwset.TxPvtReadWriteSet) error {
 	dbBatch := leveldbhelper.NewUpdateBatch()
 
 	// Create compositeKey with appropriate prefix, txid, endorserid and endorsementBlkHt
 	compositeKey := createCompositeKeyForPvtRWSet(txid, endorserid, endorsementBlkHt)
-	dbBatch.Put(compositeKey, privateSimulationResults)
+	privateSimulationResultsBytes, err := proto.Marshal(privateSimulationResults)
+	if err != nil {
+		return err
+	}
+	dbBatch.Put(compositeKey, privateSimulationResultsBytes)
 
 	// Create compositeKey with appropriate prefix, endorsementBlkHt, txid, endorserid & Store
 	// the compositeKey (purge index) a null byte as value.
@@ -136,34 +125,29 @@ func (s *store) Persist(txid string, endorserid string,
 // GetTxPvtRWSetByTxid returns an iterator due to the fact that the txid may have multiple private
 // RWSets persisted from different endorserids. Eventually, we may pass a set of collections name
 // (filters) along with txid.
-func (s *store) GetTxPvtRWSetByTxid(txid string) (commonledger.ResultsIterator, error) {
+func (s *store) GetTxPvtRWSetByTxid(txid string, filter ledger.PvtNsCollFilter) (*rwsetScanner, error) {
 	// Construct startKey and endKey to do an range query
 	startKey := createTxidRangeStartKey(txid)
 	endKey := createTxidRangeEndKey(txid)
 
 	iter := s.db.GetIterator(startKey, endKey)
-	return &rwsetScanner{txid, iter}, nil
+	return &rwsetScanner{txid, iter, filter}, nil
 }
 
 // GetSelfSimulatedTxPvtRWSetByTxid returns the private write set generated from the simulation performed
 // by the peer itself. Eventually, we may pass a set of collections name (filters) along with txid.
 func (s *store) GetSelfSimulatedTxPvtRWSetByTxid(txid string) (*EndorserPvtSimulationResults, error) {
 	var err error
-	var itr commonledger.ResultsIterator
-	if itr, err = s.GetTxPvtRWSetByTxid(txid); err != nil {
+	var itr *rwsetScanner
+	if itr, err = s.GetTxPvtRWSetByTxid(txid, nil); err != nil {
 		return nil, err
 	}
 	defer itr.Close()
-	var temp commonledger.QueryResult
 	var res *EndorserPvtSimulationResults
 	for {
-		if temp, err = itr.Next(); err != nil {
+		if res, err = itr.Next(); res == nil || err != nil {
 			return nil, err
 		}
-		if temp == nil {
-			return nil, nil
-		}
-		res = temp.(*EndorserPvtSimulationResults)
 		if selfSimulated(res) {
 			return res, nil
 		}
@@ -219,17 +203,24 @@ func (s *store) Shutdown() {
 
 // Next moves the iterator to the next key/value pair.
 // It returns whether the iterator is exhausted.
-func (scanner *rwsetScanner) Next() (commonledger.QueryResult, error) {
+func (scanner *rwsetScanner) Next() (*EndorserPvtSimulationResults, error) {
 	if !scanner.dbItr.Next() {
 		return nil, nil
 	}
 	dbKey := scanner.dbItr.Key()
 	dbVal := scanner.dbItr.Value()
 	endorserid, endorsementBlkHt := splitCompositeKeyOfPvtRWSet(dbKey)
+
+	txPvtRWSet := &rwset.TxPvtReadWriteSet{}
+	if err := proto.Unmarshal(dbVal, txPvtRWSet); err != nil {
+		return nil, err
+	}
+	filteredTxPvtRWSet := pvtdatastorage.TrimPvtWSet(txPvtRWSet, scanner.filter)
+
 	return &EndorserPvtSimulationResults{
 		EndorserID:             endorserid,
 		EndorsementBlockHeight: endorsementBlkHt,
-		PvtSimulationResults:   dbVal,
+		PvtSimulationResults:   filteredTxPvtRWSet,
 	}, nil
 }