[FAB-6619] purge pvtdata from pvt block store

This CR includes
 - Make the  pvtdata key based on collection so that purge does not
   have to perform a read and can simply issue delete for the keys
 - Store the expiry schedule based on the expirying block for collections
 - Purge the expired pvt data periodically (default every 500 block)

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

Author: manish-sethi

core/ledger/ledgerconfig/ledger_config.go

@@ -108,6 +108,16 @@ func GetMaxBatchUpdateSize() int {
 	return maxBatchUpdateSize
 }
 
+// GetPvtdataStorePurgeInterval returns the interval in the terms of number of blocks
+// when the purge for the expired data would be performed
+func GetPvtdataStorePurgeInterval() uint64 {
+	purgeInterval := viper.GetInt("ledger.pvtdataStore.purgeInterval")
+	if purgeInterval <= 0 {
+		purgeInterval = 100
+	}
+	return uint64(purgeInterval)
+}
+
 //IsHistoryDBEnabled exposes the historyDatabase variable
 func IsHistoryDBEnabled() bool {
 	return viper.GetBool(confEnableHistoryDatabase)

core/ledger/pvtdatastorage/expiry_data.pb.go

@@ -0,0 +1,23 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+syntax = "proto3";
+
+option go_package = "github.com/hyperledger/fabric/core/ledger/pvtdatastorage";
+
+package pvtdatastorage;
+
+message ExpiryData {
+    map<string, Collections>  map = 1;
+}
+
+message Collections {
+    map<string, TxNums> map = 1;
+}
+
+message TxNums {
+    repeated uint64 list = 1;
+}

core/ledger/pvtdatastorage/expiry_data.proto

@@ -0,0 +1,29 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package pvtdatastorage
+
+func newExpiryData() *ExpiryData {
+	return &ExpiryData{make(map[string]*Collections)}
+}
+
+func newCollections() *Collections {
+	return &Collections{make(map[string]*TxNums)}
+}
+
+func (e *ExpiryData) add(ns, coll string, txNum uint64) {
+	collections, ok := e.Map[ns]
+	if !ok {
+		collections = newCollections()
+		e.Map[ns] = collections
+	}
+	txNums, ok := collections.Map[coll]
+	if !ok {
+		txNums = &TxNums{}
+		collections.Map[coll] = txNums
+	}
+	txNums.List = append(txNums.List, txNum)
+}

core/ledger/pvtdatastorage/expiry_data_helper.go

@@ -0,0 +1,122 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package pvtdatastorage
+
+import (
+	"math"
+
+	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/pvtdatapolicy"
+	"github.com/hyperledger/fabric/protos/ledger/rwset"
+)
+
+func prepareStoreEntries(blockNum uint64, pvtdata []*ledger.TxPvtData, btlPolicy pvtdatapolicy.BTLPolicy) ([]*dataEntry, []*expiryEntry) {
+	dataEntries := prepareDataEntries(blockNum, pvtdata)
+	expiryEntries := prepareExpiryEntries(blockNum, dataEntries, btlPolicy)
+	return dataEntries, expiryEntries
+}
+
+func prepareDataEntries(blockNum uint64, pvtData []*ledger.TxPvtData) []*dataEntry {
+	var dataEntries []*dataEntry
+	for _, txPvtdata := range pvtData {
+		for _, nsPvtdata := range txPvtdata.WriteSet.NsPvtRwset {
+			for _, collPvtdata := range nsPvtdata.CollectionPvtRwset {
+				txnum := txPvtdata.SeqInBlock
+				ns := nsPvtdata.Namespace
+				coll := collPvtdata.CollectionName
+				dataKey := &dataKey{blockNum, txnum, ns, coll}
+				dataEntries = append(dataEntries, &dataEntry{key: dataKey, value: collPvtdata})
+			}
+		}
+	}
+	return dataEntries
+}
+
+func prepareExpiryEntries(committingBlk uint64, dataEntries []*dataEntry, btlPolicy pvtdatapolicy.BTLPolicy) []*expiryEntry {
+	mapByExpiringBlk := make(map[uint64]*ExpiryData)
+	for _, dataEntry := range dataEntries {
+		expiringBlk := btlPolicy.GetExpiringBlock(dataEntry.key.ns, dataEntry.key.coll, dataEntry.key.blkNum)
+		if neverExpires(expiringBlk) {
+			continue
+		}
+		expiryData, ok := mapByExpiringBlk[expiringBlk]
+		if !ok {
+			expiryData = newExpiryData()
+			mapByExpiringBlk[expiringBlk] = expiryData
+		}
+		expiryData.add(dataEntry.key.ns, dataEntry.key.coll, dataEntry.key.txNum)
+	}
+	var expiryEntries []*expiryEntry
+	for expiryBlk, expiryData := range mapByExpiringBlk {
+		expiryKey := &expiryKey{expiringBlk: expiryBlk, committingBlk: committingBlk}
+		expiryEntries = append(expiryEntries, &expiryEntry{key: expiryKey, value: expiryData})
+	}
+	return expiryEntries
+}
+
+func deriveDataKeys(expiryEntry *expiryEntry) []*dataKey {
+	var dataKeys []*dataKey
+	for ns, colls := range expiryEntry.value.Map {
+		for coll, txNums := range colls.Map {
+			for _, txNum := range txNums.List {
+				dataKeys = append(dataKeys, &dataKey{expiryEntry.key.committingBlk, txNum, ns, coll})
+			}
+		}
+	}
+	return dataKeys
+}
+
+func passesFilter(dataKey *dataKey, filter ledger.PvtNsCollFilter) bool {
+	return filter == nil || filter.Has(dataKey.ns, dataKey.coll)
+}
+
+func isExpired(dataKey *dataKey, btl pvtdatapolicy.BTLPolicy, latestBlkNum uint64) bool {
+	return latestBlkNum >= btl.GetExpiringBlock(dataKey.ns, dataKey.coll, dataKey.blkNum)
+}
+
+func neverExpires(expiringBlkNum uint64) bool {
+	return expiringBlkNum == math.MaxUint64
+}
+
+type txPvtdataAssembler struct {
+	blockNum, txNum uint64
+	txWset          *rwset.TxPvtReadWriteSet
+	currentNsWSet   *rwset.NsPvtReadWriteSet
+	firstCall       bool
+}
+
+func newTxPvtdataAssembler(blockNum, txNum uint64) *txPvtdataAssembler {
+	return &txPvtdataAssembler{blockNum, txNum, &rwset.TxPvtReadWriteSet{}, nil, true}
+}
+
+func (a *txPvtdataAssembler) add(ns string, collPvtWset *rwset.CollectionPvtReadWriteSet) {
+	// start a NsWset
+	if a.firstCall {
+		a.currentNsWSet = &rwset.NsPvtReadWriteSet{Namespace: ns}
+		a.firstCall = false
+	}
+
+	// if a new ns started, add the existing NsWset to TxWset and start a new one
+	if a.currentNsWSet.Namespace != ns {
+		a.txWset.NsPvtRwset = append(a.txWset.NsPvtRwset, a.currentNsWSet)
+		a.currentNsWSet = &rwset.NsPvtReadWriteSet{Namespace: ns}
+	}
+	// add the collWset to the current NsWset
+	a.currentNsWSet.CollectionPvtRwset = append(a.currentNsWSet.CollectionPvtRwset, collPvtWset)
+}
+
+func (a *txPvtdataAssembler) done() {
+	if a.currentNsWSet != nil {
+		a.txWset.NsPvtRwset = append(a.txWset.NsPvtRwset, a.currentNsWSet)
+	}
+	a.currentNsWSet = nil
+}
+
+func (a *txPvtdataAssembler) getTxPvtdata() *ledger.TxPvtData {
+	a.done()
+	return &ledger.TxPvtData{SeqInBlock: a.txNum, WriteSet: a.txWset}
+}