diff --git a/core/chaincode/lifecycle/mock/legacy_ccinfo.go b/core/chaincode/lifecycle/mock/legacy_ccinfo.go
index de11fc7776f..f12201fd0f6 100644
--- a/core/chaincode/lifecycle/mock/legacy_ccinfo.go
+++ b/core/chaincode/lifecycle/mock/legacy_ccinfo.go
@@ -71,6 +71,17 @@ type LegacyDeployedCCInfoProvider struct {
 		result1 *peer.StaticCollectionConfig
 		result2 error
 	}
+	GenerateImplicitCollectionForOrgStub        func(string) *peer.StaticCollectionConfig
+	generateImplicitCollectionForOrgMutex       sync.RWMutex
+	generateImplicitCollectionForOrgArgsForCall []struct {
+		arg1 string
+	}
+	generateImplicitCollectionForOrgReturns struct {
+		result1 *peer.StaticCollectionConfig
+	}
+	generateImplicitCollectionForOrgReturnsOnCall map[int]struct {
+		result1 *peer.StaticCollectionConfig
+	}
 	ImplicitCollectionsStub        func(string, string, ledger.SimpleQueryExecutor) ([]*peer.StaticCollectionConfig, error)
 	implicitCollectionsMutex       sync.RWMutex
 	implicitCollectionsArgsForCall []struct {
@@ -373,6 +384,66 @@ func (fake *LegacyDeployedCCInfoProvider) CollectionInfoReturnsOnCall(i int, res
 	}{result1, result2}
 }
 
+func (fake *LegacyDeployedCCInfoProvider) GenerateImplicitCollectionForOrg(arg1 string) *peer.StaticCollectionConfig {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	ret, specificReturn := fake.generateImplicitCollectionForOrgReturnsOnCall[len(fake.generateImplicitCollectionForOrgArgsForCall)]
+	fake.generateImplicitCollectionForOrgArgsForCall = append(fake.generateImplicitCollectionForOrgArgsForCall, struct {
+		arg1 string
+	}{arg1})
+	fake.recordInvocation("GenerateImplicitCollectionForOrg", []interface{}{arg1})
+	fake.generateImplicitCollectionForOrgMutex.Unlock()
+	if fake.GenerateImplicitCollectionForOrgStub != nil {
+		return fake.GenerateImplicitCollectionForOrgStub(arg1)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	fakeReturns := fake.generateImplicitCollectionForOrgReturns
+	return fakeReturns.result1
+}
+
+func (fake *LegacyDeployedCCInfoProvider) GenerateImplicitCollectionForOrgCallCount() int {
+	fake.generateImplicitCollectionForOrgMutex.RLock()
+	defer fake.generateImplicitCollectionForOrgMutex.RUnlock()
+	return len(fake.generateImplicitCollectionForOrgArgsForCall)
+}
+
+func (fake *LegacyDeployedCCInfoProvider) GenerateImplicitCollectionForOrgCalls(stub func(string) *peer.StaticCollectionConfig) {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	defer fake.generateImplicitCollectionForOrgMutex.Unlock()
+	fake.GenerateImplicitCollectionForOrgStub = stub
+}
+
+func (fake *LegacyDeployedCCInfoProvider) GenerateImplicitCollectionForOrgArgsForCall(i int) string {
+	fake.generateImplicitCollectionForOrgMutex.RLock()
+	defer fake.generateImplicitCollectionForOrgMutex.RUnlock()
+	argsForCall := fake.generateImplicitCollectionForOrgArgsForCall[i]
+	return argsForCall.arg1
+}
+
+func (fake *LegacyDeployedCCInfoProvider) GenerateImplicitCollectionForOrgReturns(result1 *peer.StaticCollectionConfig) {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	defer fake.generateImplicitCollectionForOrgMutex.Unlock()
+	fake.GenerateImplicitCollectionForOrgStub = nil
+	fake.generateImplicitCollectionForOrgReturns = struct {
+		result1 *peer.StaticCollectionConfig
+	}{result1}
+}
+
+func (fake *LegacyDeployedCCInfoProvider) GenerateImplicitCollectionForOrgReturnsOnCall(i int, result1 *peer.StaticCollectionConfig) {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	defer fake.generateImplicitCollectionForOrgMutex.Unlock()
+	fake.GenerateImplicitCollectionForOrgStub = nil
+	if fake.generateImplicitCollectionForOrgReturnsOnCall == nil {
+		fake.generateImplicitCollectionForOrgReturnsOnCall = make(map[int]struct {
+			result1 *peer.StaticCollectionConfig
+		})
+	}
+	fake.generateImplicitCollectionForOrgReturnsOnCall[i] = struct {
+		result1 *peer.StaticCollectionConfig
+	}{result1}
+}
+
 func (fake *LegacyDeployedCCInfoProvider) ImplicitCollections(arg1 string, arg2 string, arg3 ledger.SimpleQueryExecutor) ([]*peer.StaticCollectionConfig, error) {
 	fake.implicitCollectionsMutex.Lock()
 	ret, specificReturn := fake.implicitCollectionsReturnsOnCall[len(fake.implicitCollectionsArgsForCall)]
@@ -564,6 +635,8 @@ func (fake *LegacyDeployedCCInfoProvider) Invocations() map[string][][]interface
 	defer fake.chaincodeInfoMutex.RUnlock()
 	fake.collectionInfoMutex.RLock()
 	defer fake.collectionInfoMutex.RUnlock()
+	fake.generateImplicitCollectionForOrgMutex.RLock()
+	defer fake.generateImplicitCollectionForOrgMutex.RUnlock()
 	fake.implicitCollectionsMutex.RLock()
 	defer fake.implicitCollectionsMutex.RUnlock()
 	fake.namespacesMutex.RLock()
diff --git a/core/ledger/kvledger/channelinfo_provider.go b/core/ledger/kvledger/channelinfo_provider.go
index 9e174a3bef2..8dde617af38 100644
--- a/core/ledger/kvledger/channelinfo_provider.go
+++ b/core/ledger/kvledger/channelinfo_provider.go
@@ -8,20 +8,76 @@ package kvledger
 
 import (
 	cb "github.com/hyperledger/fabric-protos-go/common"
+	"github.com/hyperledger/fabric-protos-go/ledger/queryresult"
 	"github.com/hyperledger/fabric/bccsp/factory"
 	"github.com/hyperledger/fabric/common/channelconfig"
+	commonledger "github.com/hyperledger/fabric/common/ledger"
 	"github.com/hyperledger/fabric/common/ledger/blkstorage"
+	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
 	"github.com/hyperledger/fabric/protoutil"
+	"github.com/pkg/errors"
 )
 
 type channelInfoProvider struct {
 	channelName string
 	blockStore  *blkstorage.BlockStore
+	ledger.DeployedChaincodeInfoProvider
 }
 
-// GetAllMSPIDs retrieves the MSPIDs of application organizations in all the channel configurations,
+// NamespacesAndCollections returns namespaces and their collections for the channel.
+func (p *channelInfoProvider) NamespacesAndCollections(vdb statedb.VersionedDB) (map[string][]string, error) {
+	mspIDs, err := p.getAllMSPIDs()
+	if err != nil {
+		return nil, err
+	}
+	implicitCollNames := make([]string, len(mspIDs))
+	for i, mspID := range mspIDs {
+		implicitCollNames[i] = p.GenerateImplicitCollectionForOrg(mspID).Name
+	}
+	chaincodesInfo, err := p.AllChaincodesInfo(p.channelName, &simpleQueryExecutor{vdb})
+	if err != nil {
+		return nil, err
+	}
+
+	retNamespaces := map[string][]string{}
+	// iterate each chaincode, add implicit collections and explicit collections
+	for _, ccInfo := range chaincodesInfo {
+		ccName := ccInfo.Name
+		retNamespaces[ccName] = []string{}
+		for _, implicitCollName := range implicitCollNames {
+			retNamespaces[ccName] = append(retNamespaces[ccName], implicitCollName)
+		}
+		if ccInfo.ExplicitCollectionConfigPkg == nil {
+			continue
+		}
+		for _, config := range ccInfo.ExplicitCollectionConfigPkg.Config {
+			collConfig := config.GetStaticCollectionConfig()
+			if collConfig != nil {
+				retNamespaces[ccName] = append(retNamespaces[ccName], collConfig.Name)
+			}
+		}
+	}
+
+	// add lifecycle management namespaces with implicit collections (not applicable to legacy lifecycle)
+	for _, ns := range p.Namespaces() {
+		retNamespaces[ns] = []string{}
+		if ns == "lscc" {
+			continue
+		}
+		for _, implicitCollName := range implicitCollNames {
+			retNamespaces[ns] = append(retNamespaces[ns], implicitCollName)
+		}
+	}
+
+	// add namespace ""
+	retNamespaces[""] = []string{}
+	return retNamespaces, nil
+}
+
+// getAllMSPIDs retrieves the MSPIDs of application organizations in all the channel configurations,
 // including current and previous channel configurations.
-func (p *channelInfoProvider) GetAllMSPIDs() ([]string, error) {
+func (p *channelInfoProvider) getAllMSPIDs() ([]string, error) {
 	blockchainInfo, err := p.blockStore.GetBlockchainInfo()
 	if err != nil {
 		return nil, err
@@ -74,3 +130,58 @@ func (p *channelInfoProvider) mostRecentConfigBlockAsOf(blockNum uint64) (*cb.Bl
 	}
 	return p.blockStore.RetrieveBlockByNumber(configBlockNum)
 }
+
+// simpleQueryExecutor implements ledger.SimpleQueryExecutor interface
+type simpleQueryExecutor struct {
+	statedb.VersionedDB
+}
+
+func (sqe *simpleQueryExecutor) GetState(ns string, key string) ([]byte, error) {
+	versionedValue, err := sqe.VersionedDB.GetState(ns, key)
+	if err != nil {
+		return nil, err
+	}
+	var value []byte
+	if versionedValue != nil {
+		value = versionedValue.Value
+	}
+	return value, nil
+}
+
+func (sqe *simpleQueryExecutor) GetStateRangeScanIterator(ns string, startKey string, endKey string) (commonledger.ResultsIterator, error) {
+	dbItr, err := sqe.VersionedDB.GetStateRangeScanIterator(ns, startKey, endKey)
+	if err != nil {
+		return nil, err
+	}
+	itr := &resultsItr{ns: ns, dbItr: dbItr}
+	return itr, nil
+}
+
+// GetPrivateDataHash is not implemented and should not be called
+func (sqe *simpleQueryExecutor) GetPrivateDataHash(namespace, collection, key string) ([]byte, error) {
+	return nil, errors.New("not implemented yet")
+}
+
+type resultsItr struct {
+	ns    string
+	dbItr statedb.ResultsIterator
+}
+
+// Next implements method in interface ledger.ResultsIterator
+func (itr *resultsItr) Next() (commonledger.QueryResult, error) {
+	queryResult, err := itr.dbItr.Next()
+	if err != nil {
+		return nil, err
+	}
+	// itr.updateRangeQueryInfo(queryResult)
+	if queryResult == nil {
+		return nil, nil
+	}
+	versionedKV := queryResult.(*statedb.VersionedKV)
+	return &queryresult.KV{Namespace: versionedKV.Namespace, Key: versionedKV.Key, Value: versionedKV.Value}, nil
+}
+
+// Close implements method in interface ledger.ResultsIterator
+func (itr *resultsItr) Close() {
+	itr.dbItr.Close()
+}
diff --git a/core/ledger/kvledger/channelinfo_provider_test.go b/core/ledger/kvledger/channelinfo_provider_test.go
index d7205704954..cb8768b0476 100644
--- a/core/ledger/kvledger/channelinfo_provider_test.go
+++ b/core/ledger/kvledger/channelinfo_provider_test.go
@@ -8,6 +8,7 @@ package kvledger
 
 import (
 	"bytes"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"testing"
@@ -16,16 +17,80 @@ import (
 	"github.com/hyperledger/fabric-config/protolator"
 	"github.com/hyperledger/fabric-protos-go/common"
 	cb "github.com/hyperledger/fabric-protos-go/common"
+	pb "github.com/hyperledger/fabric-protos-go/peer"
 	"github.com/hyperledger/fabric/common/channelconfig"
 	"github.com/hyperledger/fabric/common/configtx/test"
 	"github.com/hyperledger/fabric/common/ledger/blkstorage"
 	"github.com/hyperledger/fabric/common/ledger/testutil"
 	"github.com/hyperledger/fabric/common/metrics/disabled"
+	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/mock"
 	"github.com/hyperledger/fabric/protoutil"
 	"github.com/stretchr/testify/require"
 )
 
-// TestGetAllMSPIDs verifies GetAllMSPIDs by adding and removing organizations to the channel config.
+func TestNamespacesAndCollections(t *testing.T) {
+	channelName := "testnamespacesandcollections"
+	basePath, err := ioutil.TempDir("", "testchannelinfoprovider")
+	require.NoError(t, err)
+	defer os.RemoveAll(basePath)
+	blkStoreProvider, blkStore := openBlockStorage(t, channelName, basePath)
+	defer blkStoreProvider.Close()
+
+	// add genesis block and another config block so that we can retrieve MSPIDs
+	// 3 orgs/MSPIDs are added: SampleOrg/SampleOrg, org1/Org1MSP, org2/Org2MSP
+	genesisBlock, err := test.MakeGenesisBlock(channelName)
+	require.NoError(t, err)
+	require.NoError(t, blkStore.AddBlock(genesisBlock))
+	orgGroups := createTestOrgGroups(t)
+	config := getConfigFromBlock(genesisBlock)
+	config.ChannelGroup.Groups[channelconfig.ApplicationGroupKey].Groups["org1"] = orgGroups["org1"]
+	config.ChannelGroup.Groups[channelconfig.ApplicationGroupKey].Groups["org2"] = orgGroups["org2"]
+	configEnv := getEnvelopeFromConfig(channelName, config)
+	configBlock := newBlock([]*cb.Envelope{configEnv}, 1, 1, protoutil.BlockHeaderHash(genesisBlock.Header))
+	require.NoError(t, blkStore.AddBlock(configBlock))
+
+	// prepare fakeDeployedCCInfoProvider to create mocked test data
+	deployedccInfo := map[string]*ledger.DeployedChaincodeInfo{
+		"cc1": {
+			Name:                        "cc1",
+			Version:                     "version",
+			Hash:                        []byte("cc1-hash"),
+			ExplicitCollectionConfigPkg: prepareCollectionConfigPackage([]string{"collectionA", "collectionB"}),
+		},
+		"cc2": {
+			Name:    "cc2",
+			Version: "version",
+			Hash:    []byte("cc2-hash"),
+		},
+	}
+	fakeDeployedCCInfoProvider := &mock.DeployedChaincodeInfoProvider{}
+	fakeDeployedCCInfoProvider.NamespacesReturns([]string{"_lifecycle", "lscc"})
+	fakeDeployedCCInfoProvider.AllChaincodesInfoReturns(deployedccInfo, nil)
+	fakeDeployedCCInfoProvider.GenerateImplicitCollectionForOrgStub = func(mspID string) *pb.StaticCollectionConfig {
+		return &pb.StaticCollectionConfig{
+			Name: fmt.Sprintf("_implicit_org_%s", mspID),
+		}
+	}
+
+	// verify NamespacesAndCollections
+	channelInfoProvider := &channelInfoProvider{channelName, blkStore, fakeDeployedCCInfoProvider}
+	expectedNamespacesAndColls := map[string][]string{
+		"cc1":        {"_implicit_org_Org1MSP", "_implicit_org_Org2MSP", "_implicit_org_SampleOrg", "collectionA", "collectionB"},
+		"cc2":        {"_implicit_org_Org1MSP", "_implicit_org_Org2MSP", "_implicit_org_SampleOrg"},
+		"_lifecycle": {"_implicit_org_Org1MSP", "_implicit_org_Org2MSP", "_implicit_org_SampleOrg"},
+		"lscc":       {},
+		"":           {},
+	}
+	namespacesAndColls, err := channelInfoProvider.NamespacesAndCollections(nil)
+	require.NoError(t, err)
+	require.Equal(t, len(expectedNamespacesAndColls), len(namespacesAndColls))
+	for ns, colls := range expectedNamespacesAndColls {
+		require.ElementsMatch(t, colls, namespacesAndColls[ns])
+	}
+}
+
+// TestGetAllMSPIDs verifies getAllMSPIDs by adding and removing organizations to the channel config.
 func TestGetAllMSPIDs(t *testing.T) {
 	channelName := "testgetallmspids"
 	basePath, err := ioutil.TempDir("", "testchannelinfoprovider")
@@ -34,7 +99,7 @@ func TestGetAllMSPIDs(t *testing.T) {
 
 	blkStoreProvider, blkStore := openBlockStorage(t, channelName, basePath)
 	defer blkStoreProvider.Close()
-	channelInfoProvider := &channelInfoProvider{channelName, blkStore}
+	channelInfoProvider := &channelInfoProvider{channelName, blkStore, nil}
 
 	var block *cb.Block
 	var configBlock *cb.Block
@@ -112,7 +177,7 @@ func TestGetAllMSPIDs_NegativeTests(t *testing.T) {
 
 	blkStoreProvider, blkStore := openBlockStorage(t, channelName, basePath)
 	defer blkStoreProvider.Close()
-	channelInfoProvider := &channelInfoProvider{channelName, blkStore}
+	channelInfoProvider := &channelInfoProvider{channelName, blkStore, nil}
 
 	var configBlock *cb.Block
 	var lastBlockNum = uint64(0)
@@ -128,7 +193,7 @@ func TestGetAllMSPIDs_NegativeTests(t *testing.T) {
 	lastConfigBlockNum++
 	configBlock = newBlock([]*cb.Envelope{}, lastBlockNum, lastConfigBlockNum, protoutil.BlockHeaderHash(configBlock.Header))
 	require.NoError(t, blkStore.AddBlock(configBlock))
-	_, err = channelInfoProvider.GetAllMSPIDs()
+	_, err = channelInfoProvider.getAllMSPIDs()
 	require.EqualError(t, err, "malformed configuration block: envelope index out of bounds")
 
 	// test RetrieveBlockByNumber error by using a non-existent block num for config block index
@@ -136,7 +201,7 @@ func TestGetAllMSPIDs_NegativeTests(t *testing.T) {
 	lastConfigBlockNum++
 	configBlock = newBlock(nil, lastBlockNum, lastBlockNum+1, protoutil.BlockHeaderHash(configBlock.Header))
 	require.NoError(t, blkStore.AddBlock(configBlock))
-	_, err = channelInfoProvider.GetAllMSPIDs()
+	_, err = channelInfoProvider.getAllMSPIDs()
 	require.EqualError(t, err, "Entry not found in index")
 
 	// test GetLastConfigIndexFromBlock error by using invalid bytes for LastConfig metadata value
@@ -145,12 +210,12 @@ func TestGetAllMSPIDs_NegativeTests(t *testing.T) {
 	configBlock = newBlock(nil, lastBlockNum, lastConfigBlockNum, protoutil.BlockHeaderHash(configBlock.Header))
 	configBlock.Metadata.Metadata[cb.BlockMetadataIndex_SIGNATURES] = []byte("invalid_bytes")
 	require.NoError(t, blkStore.AddBlock(configBlock))
-	_, err = channelInfoProvider.GetAllMSPIDs()
+	_, err = channelInfoProvider.getAllMSPIDs()
 	require.EqualError(t, err, "failed to retrieve metadata: error unmarshaling metadata at index [SIGNATURES]: unexpected EOF")
 
 	// test RetrieveBlockByNumber error (before calling GetLastConfigIndexFromBlock) by closing block store provider
 	blkStoreProvider.Close()
-	_, err = channelInfoProvider.GetAllMSPIDs()
+	_, err = channelInfoProvider.getAllMSPIDs()
 	require.Contains(t, err.Error(), "leveldb: closed")
 }
 
@@ -167,7 +232,7 @@ func openBlockStorage(t *testing.T, channelName string, basePath string) (*blkst
 }
 
 func verifyGetAllMSPIDs(t *testing.T, channelInfoProvider *channelInfoProvider, expectedMSPIDs []string) {
-	mspids, err := channelInfoProvider.GetAllMSPIDs()
+	mspids, err := channelInfoProvider.getAllMSPIDs()
 	require.NoError(t, err)
 	require.ElementsMatch(t, expectedMSPIDs, mspids)
 }
@@ -230,3 +295,17 @@ func createTestOrgGroups(t *testing.T) map[string]*cb.ConfigGroup {
 	config := getConfigFromBlock(block)
 	return config.ChannelGroup.Groups[channelconfig.ApplicationGroupKey].Groups
 }
+
+func prepareCollectionConfigPackage(collNames []string) *pb.CollectionConfigPackage {
+	collConfigs := make([]*pb.CollectionConfig, len(collNames))
+	for i, name := range collNames {
+		collConfigs[i] = &pb.CollectionConfig{
+			Payload: &pb.CollectionConfig_StaticCollectionConfig{
+				StaticCollectionConfig: &pb.StaticCollectionConfig{
+					Name: name,
+				},
+			},
+		}
+	}
+	return &pb.CollectionConfigPackage{Config: collConfigs}
+}
diff --git a/core/ledger/kvledger/kv_ledger_provider.go b/core/ledger/kvledger/kv_ledger_provider.go
index 30505865069..8a75b78c610 100644
--- a/core/ledger/kvledger/kv_ledger_provider.go
+++ b/core/ledger/kvledger/kv_ledger_provider.go
@@ -333,7 +333,8 @@ func (p *Provider) open(ledgerID string) (ledger.PeerLedger, error) {
 	p.collElgNotifier.registerListener(ledgerID, pvtdataStore)
 
 	// Get the versioned database (state database) for a chain/ledger
-	db, err := p.dbProvider.GetDBHandle(ledgerID)
+	channelInfoProvider := &channelInfoProvider{ledgerID, blockStore, p.collElgNotifier.deployedChaincodeInfoProvider}
+	db, err := p.dbProvider.GetDBHandle(ledgerID, channelInfoProvider)
 	if err != nil {
 		return nil, err
 	}
diff --git a/core/ledger/kvledger/txmgmt/privacyenabledstate/db.go b/core/ledger/kvledger/txmgmt/privacyenabledstate/db.go
index 5b69103fbf6..d36d5a0ff38 100644
--- a/core/ledger/kvledger/txmgmt/privacyenabledstate/db.go
+++ b/core/ledger/kvledger/txmgmt/privacyenabledstate/db.go
@@ -47,7 +47,7 @@ type StateDBConfig struct {
 // DBProvider encapsulates other providers such as VersionedDBProvider and
 // BookeepingProvider which are required to create DB for a channel
 type DBProvider struct {
-	statedb.VersionedDBProvider
+	VersionedDBProvider statedb.VersionedDBProvider
 	HealthCheckRegistry ledger.HealthCheckRegistry
 	bookkeepingProvider bookkeeping.Provider
 }
@@ -95,8 +95,8 @@ func (p *DBProvider) RegisterHealthChecker() error {
 }
 
 // GetDBHandle gets a handle to DB for a given id, i.e., a channel
-func (p *DBProvider) GetDBHandle(id string) (*DB, error) {
-	vdb, err := p.VersionedDBProvider.GetDBHandle(id)
+func (p *DBProvider) GetDBHandle(id string, chInfoProvider channelInfoProvider) (*DB, error) {
+	vdb, err := p.VersionedDBProvider.GetDBHandle(id, &namespaceProvider{chInfoProvider})
 	if err != nil {
 		return nil, err
 	}
@@ -429,3 +429,41 @@ func getIndexInfo(indexPath string) *indexInfo {
 	}
 	return indexInfo
 }
+
+// channelInfoProvider interface enables the privateenabledstate package to retrieve all the config blocks
+// and  namespaces and collections.
+type channelInfoProvider interface {
+	// NamespacesAndCollections returns namespaces and collections for the channel.
+	NamespacesAndCollections(vdb statedb.VersionedDB) (map[string][]string, error)
+}
+
+// namespaceProvider implements statedb.NamespaceProvider interface
+type namespaceProvider struct {
+	channelInfoProvider
+}
+
+// PossibleNamespaces returns all possible namespaces for a channel. In ledger, a private data namespace is
+// created only if the peer is a member of the collection or owns the implicit collection. However, this function
+// adopts a simple implementation that always adds private data namespace for a collection without checking
+// peer membership/ownership. As a result, it returns a superset of namespaces that may be created.
+// However, it will not cause any inconsistent issue because the caller in statecouchdb will check if any
+// existing database matches the namespace and filter out all extra namespaces if no databases match them.
+// Checking peer membership is complicated because it requires retrieving all the collection configs from
+// the collection config store. Because this is a temporary function needed to retroactively build namespaces
+// when upgrading v2.0/2.1 peers to a newer v2.x version and because returning extra private data namespaces
+// does not cause inconsistence, it makes sense to use the simple implementation.
+func (p *namespaceProvider) PossibleNamespaces(vdb statedb.VersionedDB) ([]string, error) {
+	retNamespaces := []string{}
+	nsCollMap, err := p.NamespacesAndCollections(vdb)
+	if err != nil {
+		return nil, err
+	}
+	for ns, collections := range nsCollMap {
+		retNamespaces = append(retNamespaces, ns)
+		for _, collection := range collections {
+			retNamespaces = append(retNamespaces, deriveHashedDataNs(ns, collection))
+			retNamespaces = append(retNamespaces, derivePvtDataNs(ns, collection))
+		}
+	}
+	return retNamespaces, nil
+}
diff --git a/core/ledger/kvledger/txmgmt/privacyenabledstate/db_test.go b/core/ledger/kvledger/txmgmt/privacyenabledstate/db_test.go
index 2c27523e3a2..27542ade60f 100644
--- a/core/ledger/kvledger/txmgmt/privacyenabledstate/db_test.go
+++ b/core/ledger/kvledger/txmgmt/privacyenabledstate/db_test.go
@@ -18,6 +18,7 @@ import (
 	"github.com/hyperledger/fabric/core/common/ccprovider"
 	"github.com/hyperledger/fabric/core/ledger/cceventmgmt"
 	"github.com/hyperledger/fabric/core/ledger/internal/version"
+	testmock "github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/privacyenabledstate/mock"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb/statecouchdb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb/stateleveldb"
@@ -587,3 +588,50 @@ func generateLedgerID(t *testing.T) string {
 	assert.NoError(t, err)
 	return fmt.Sprintf("x%s", hex.EncodeToString(bytes))
 }
+
+//go:generate counterfeiter -o mock/channelinfo_provider.go -fake-name ChannelInfoProvider . channelInfoProviderWrapper
+
+// define this interface to break circular dependency
+type channelInfoProviderWrapper interface {
+	channelInfoProvider
+}
+
+func TestPossibleNamespaces(t *testing.T) {
+	namespacesAndCollections := map[string][]string{
+		"cc1":        {"_implicit_org_Org1MSP", "_implicit_org_Org2MSP", "collectionA", "collectionB"},
+		"cc2":        {"_implicit_org_Org1MSP", "_implicit_org_Org2MSP"},
+		"_lifecycle": {"_implicit_org_Org1MSP", "_implicit_org_Org2MSP"},
+		"lscc":       {},
+		"":           {},
+	}
+	expectedNamespaces := []string{
+		"cc1",
+		"cc1$$p_implicit_org_Org1MSP",
+		"cc1$$h_implicit_org_Org1MSP",
+		"cc1$$p_implicit_org_Org2MSP",
+		"cc1$$h_implicit_org_Org2MSP",
+		"cc1$$pcollectionA",
+		"cc1$$hcollectionA",
+		"cc1$$pcollectionB",
+		"cc1$$hcollectionB",
+		"cc2",
+		"cc2$$p_implicit_org_Org1MSP",
+		"cc2$$h_implicit_org_Org1MSP",
+		"cc2$$p_implicit_org_Org2MSP",
+		"cc2$$h_implicit_org_Org2MSP",
+		"_lifecycle",
+		"_lifecycle$$p_implicit_org_Org1MSP",
+		"_lifecycle$$h_implicit_org_Org1MSP",
+		"_lifecycle$$p_implicit_org_Org2MSP",
+		"_lifecycle$$h_implicit_org_Org2MSP",
+		"lscc",
+		"",
+	}
+
+	fakeChannelInfoProvider := &testmock.ChannelInfoProvider{}
+	fakeChannelInfoProvider.NamespacesAndCollectionsReturns(namespacesAndCollections, nil)
+	nsProvider := &namespaceProvider{fakeChannelInfoProvider}
+	namespaces, err := nsProvider.PossibleNamespaces(&statecouchdb.VersionedDB{})
+	require.NoError(t, err)
+	require.ElementsMatch(t, expectedNamespaces, namespaces)
+}
diff --git a/core/ledger/kvledger/txmgmt/privacyenabledstate/mock/channelinfo_provider.go b/core/ledger/kvledger/txmgmt/privacyenabledstate/mock/channelinfo_provider.go
new file mode 100644
index 00000000000..bda33ce7d80
--- /dev/null
+++ b/core/ledger/kvledger/txmgmt/privacyenabledstate/mock/channelinfo_provider.go
@@ -0,0 +1,113 @@
+// Code generated by counterfeiter. DO NOT EDIT.
+package mock
+
+import (
+	"sync"
+
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
+)
+
+type ChannelInfoProvider struct {
+	NamespacesAndCollectionsStub        func(statedb.VersionedDB) (map[string][]string, error)
+	namespacesAndCollectionsMutex       sync.RWMutex
+	namespacesAndCollectionsArgsForCall []struct {
+		arg1 statedb.VersionedDB
+	}
+	namespacesAndCollectionsReturns struct {
+		result1 map[string][]string
+		result2 error
+	}
+	namespacesAndCollectionsReturnsOnCall map[int]struct {
+		result1 map[string][]string
+		result2 error
+	}
+	invocations      map[string][][]interface{}
+	invocationsMutex sync.RWMutex
+}
+
+func (fake *ChannelInfoProvider) NamespacesAndCollections(arg1 statedb.VersionedDB) (map[string][]string, error) {
+	fake.namespacesAndCollectionsMutex.Lock()
+	ret, specificReturn := fake.namespacesAndCollectionsReturnsOnCall[len(fake.namespacesAndCollectionsArgsForCall)]
+	fake.namespacesAndCollectionsArgsForCall = append(fake.namespacesAndCollectionsArgsForCall, struct {
+		arg1 statedb.VersionedDB
+	}{arg1})
+	fake.recordInvocation("NamespacesAndCollections", []interface{}{arg1})
+	fake.namespacesAndCollectionsMutex.Unlock()
+	if fake.NamespacesAndCollectionsStub != nil {
+		return fake.NamespacesAndCollectionsStub(arg1)
+	}
+	if specificReturn {
+		return ret.result1, ret.result2
+	}
+	fakeReturns := fake.namespacesAndCollectionsReturns
+	return fakeReturns.result1, fakeReturns.result2
+}
+
+func (fake *ChannelInfoProvider) NamespacesAndCollectionsCallCount() int {
+	fake.namespacesAndCollectionsMutex.RLock()
+	defer fake.namespacesAndCollectionsMutex.RUnlock()
+	return len(fake.namespacesAndCollectionsArgsForCall)
+}
+
+func (fake *ChannelInfoProvider) NamespacesAndCollectionsCalls(stub func(statedb.VersionedDB) (map[string][]string, error)) {
+	fake.namespacesAndCollectionsMutex.Lock()
+	defer fake.namespacesAndCollectionsMutex.Unlock()
+	fake.NamespacesAndCollectionsStub = stub
+}
+
+func (fake *ChannelInfoProvider) NamespacesAndCollectionsArgsForCall(i int) statedb.VersionedDB {
+	fake.namespacesAndCollectionsMutex.RLock()
+	defer fake.namespacesAndCollectionsMutex.RUnlock()
+	argsForCall := fake.namespacesAndCollectionsArgsForCall[i]
+	return argsForCall.arg1
+}
+
+func (fake *ChannelInfoProvider) NamespacesAndCollectionsReturns(result1 map[string][]string, result2 error) {
+	fake.namespacesAndCollectionsMutex.Lock()
+	defer fake.namespacesAndCollectionsMutex.Unlock()
+	fake.NamespacesAndCollectionsStub = nil
+	fake.namespacesAndCollectionsReturns = struct {
+		result1 map[string][]string
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *ChannelInfoProvider) NamespacesAndCollectionsReturnsOnCall(i int, result1 map[string][]string, result2 error) {
+	fake.namespacesAndCollectionsMutex.Lock()
+	defer fake.namespacesAndCollectionsMutex.Unlock()
+	fake.NamespacesAndCollectionsStub = nil
+	if fake.namespacesAndCollectionsReturnsOnCall == nil {
+		fake.namespacesAndCollectionsReturnsOnCall = make(map[int]struct {
+			result1 map[string][]string
+			result2 error
+		})
+	}
+	fake.namespacesAndCollectionsReturnsOnCall[i] = struct {
+		result1 map[string][]string
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *ChannelInfoProvider) Invocations() map[string][][]interface{} {
+	fake.invocationsMutex.RLock()
+	defer fake.invocationsMutex.RUnlock()
+	fake.namespacesAndCollectionsMutex.RLock()
+	defer fake.namespacesAndCollectionsMutex.RUnlock()
+	copiedInvocations := map[string][][]interface{}{}
+	for key, value := range fake.invocations {
+		copiedInvocations[key] = value
+	}
+	return copiedInvocations
+}
+
+func (fake *ChannelInfoProvider) recordInvocation(key string, args []interface{}) {
+	fake.invocationsMutex.Lock()
+	defer fake.invocationsMutex.Unlock()
+	if fake.invocations == nil {
+		fake.invocations = map[string][][]interface{}{}
+	}
+	if fake.invocations[key] == nil {
+		fake.invocations[key] = [][]interface{}{}
+	}
+	fake.invocations[key] = append(fake.invocations[key], args)
+}
diff --git a/core/ledger/kvledger/txmgmt/privacyenabledstate/test_exports.go b/core/ledger/kvledger/txmgmt/privacyenabledstate/test_exports.go
index 94b5cb2f9f7..547ac15036e 100644
--- a/core/ledger/kvledger/txmgmt/privacyenabledstate/test_exports.go
+++ b/core/ledger/kvledger/txmgmt/privacyenabledstate/test_exports.go
@@ -15,6 +15,7 @@ import (
 	"github.com/hyperledger/fabric/common/metrics/disabled"
 	"github.com/hyperledger/fabric/core/ledger"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/bookkeeping"
+	testmock "github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/privacyenabledstate/mock"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb/statecouchdb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb/stateleveldb"
@@ -84,7 +85,7 @@ func (env *LevelDBTestEnv) StopExternalResource() {
 
 // GetDBHandle implements corresponding function from interface TestEnv
 func (env *LevelDBTestEnv) GetDBHandle(id string) *DB {
-	db, err := env.provider.GetDBHandle(id)
+	db, err := env.provider.GetDBHandle(id, nil)
 	assert.NoError(env.t, err)
 	return db
 }
@@ -185,7 +186,7 @@ func (env *CouchDBTestEnv) Init(t testing.TB) {
 
 // GetDBHandle implements corresponding function from interface TestEnv
 func (env *CouchDBTestEnv) GetDBHandle(id string) *DB {
-	db, err := env.provider.GetDBHandle(id)
+	db, err := env.provider.GetDBHandle(id, &testmock.ChannelInfoProvider{})
 	assert.NoError(env.t, err)
 	return db
 }
diff --git a/core/ledger/kvledger/txmgmt/statedb/commontests/test_common.go b/core/ledger/kvledger/txmgmt/statedb/commontests/test_common.go
index e843545494d..2ccf30070f2 100644
--- a/core/ledger/kvledger/txmgmt/statedb/commontests/test_common.go
+++ b/core/ledger/kvledger/txmgmt/statedb/commontests/test_common.go
@@ -19,7 +19,7 @@ import (
 
 // TestGetStateMultipleKeys tests read for given multiple keys
 func TestGetStateMultipleKeys(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testgetmultiplekeys")
+	db, err := dbProvider.GetDBHandle("testgetmultiplekeys", nil)
 	assert.NoError(t, err)
 
 	// Test that savepoint is nil for a new state db
@@ -48,7 +48,7 @@ func TestGetStateMultipleKeys(t *testing.T, dbProvider statedb.VersionedDBProvid
 
 // TestBasicRW tests basic read-write
 func TestBasicRW(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testbasicrw")
+	db, err := dbProvider.GetDBHandle("testbasicrw", nil)
 	assert.NoError(t, err)
 
 	// Test that savepoint is nil for a new state db
@@ -93,10 +93,10 @@ func TestBasicRW(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 
 // TestMultiDBBasicRW tests basic read-write on multiple dbs
 func TestMultiDBBasicRW(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db1, err := dbProvider.GetDBHandle("testmultidbbasicrw")
+	db1, err := dbProvider.GetDBHandle("testmultidbbasicrw", nil)
 	assert.NoError(t, err)
 
-	db2, err := dbProvider.GetDBHandle("testmultidbbasicrw2")
+	db2, err := dbProvider.GetDBHandle("testmultidbbasicrw2", nil)
 	assert.NoError(t, err)
 
 	batch1 := statedb.NewUpdateBatch()
@@ -132,7 +132,7 @@ func TestMultiDBBasicRW(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 
 // TestDeletes tests deletes
 func TestDeletes(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testdeletes")
+	db, err := dbProvider.GetDBHandle("testdeletes", nil)
 	assert.NoError(t, err)
 
 	batch := statedb.NewUpdateBatch()
@@ -167,7 +167,7 @@ func TestDeletes(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 
 // TestIterator tests the iterator
 func TestIterator(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testiterator")
+	db, err := dbProvider.GetDBHandle("testiterator", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -209,7 +209,7 @@ func testItr(t *testing.T, itr statedb.ResultsIterator, expectedKeys []string) {
 
 // TestQuery tests queries
 func TestQuery(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testquery")
+	db, err := dbProvider.GetDBHandle("testquery", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -489,7 +489,7 @@ func TestQuery(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 // TestGetVersion tests retrieving the version by namespace and key
 func TestGetVersion(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 
-	db, err := dbProvider.GetDBHandle("testgetversion")
+	db, err := dbProvider.GetDBHandle("testgetversion", nil)
 	assert.NoError(t, err)
 
 	batch := statedb.NewUpdateBatch()
@@ -546,7 +546,7 @@ func TestGetVersion(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 
 // TestSmallBatchSize tests multiple update batches
 func TestSmallBatchSize(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testsmallbatchsize")
+	db, err := dbProvider.GetDBHandle("testsmallbatchsize", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -616,7 +616,7 @@ func TestSmallBatchSize(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 // TestBatchWithIndividualRetry tests a single failure in a batch
 func TestBatchWithIndividualRetry(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 
-	db, err := dbProvider.GetDBHandle("testbatchretry")
+	db, err := dbProvider.GetDBHandle("testbatchretry", nil)
 	assert.NoError(t, err)
 
 	batch := statedb.NewUpdateBatch()
@@ -724,7 +724,7 @@ func TestBatchWithIndividualRetry(t *testing.T, dbProvider statedb.VersionedDBPr
 
 // TestValueAndMetadataWrites tests statedb for value and metadata read-writes
 func TestValueAndMetadataWrites(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testvalueandmetadata")
+	db, err := dbProvider.GetDBHandle("testvalueandmetadata", nil)
 	assert.NoError(t, err)
 	batch := statedb.NewUpdateBatch()
 
@@ -754,7 +754,7 @@ func TestValueAndMetadataWrites(t *testing.T, dbProvider statedb.VersionedDBProv
 
 // TestPaginatedRangeQuery tests range queries with pagination
 func TestPaginatedRangeQuery(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testpaginatedrangequery")
+	db, err := dbProvider.GetDBHandle("testpaginatedrangequery", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -881,7 +881,7 @@ func TestPaginatedRangeQuery(t *testing.T, dbProvider statedb.VersionedDBProvide
 
 // TestRangeQuerySpecialCharacters tests range queries for keys with special characters and/or non-English characters
 func TestRangeQuerySpecialCharacters(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("testrangequeryspecialcharacters")
+	db, err := dbProvider.GetDBHandle("testrangequeryspecialcharacters", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -965,7 +965,7 @@ func TestItrWithoutClose(t *testing.T, itr statedb.ResultsIterator, expectedKeys
 }
 
 func TestApplyUpdatesWithNilHeight(t *testing.T, dbProvider statedb.VersionedDBProvider) {
-	db, err := dbProvider.GetDBHandle("test-apply-updates-with-nil-height")
+	db, err := dbProvider.GetDBHandle("test-apply-updates-with-nil-height", nil)
 	assert.NoError(t, err)
 
 	batch1 := statedb.NewUpdateBatch()
@@ -989,7 +989,7 @@ func TestFullScanIterator(
 	valueFormat byte,
 	dbValueDeserializer func(b []byte) (*statedb.VersionedValue, error)) {
 
-	db, err := dbProvider.GetDBHandle("test-full-scan-iterator")
+	db, err := dbProvider.GetDBHandle("test-full-scan-iterator", nil)
 	assert.NoError(t, err)
 
 	// generateSampleData returns a slice of KVs. The returned value contains five KVs for each of the namespaces
diff --git a/core/ledger/kvledger/txmgmt/statedb/mock/namespace_provider.go b/core/ledger/kvledger/txmgmt/statedb/mock/namespace_provider.go
new file mode 100644
index 00000000000..4bf98740c73
--- /dev/null
+++ b/core/ledger/kvledger/txmgmt/statedb/mock/namespace_provider.go
@@ -0,0 +1,115 @@
+// Code generated by counterfeiter. DO NOT EDIT.
+package mock
+
+import (
+	"sync"
+
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
+)
+
+type NamespaceProvider struct {
+	PossibleNamespacesStub        func(statedb.VersionedDB) ([]string, error)
+	possibleNamespacesMutex       sync.RWMutex
+	possibleNamespacesArgsForCall []struct {
+		arg1 statedb.VersionedDB
+	}
+	possibleNamespacesReturns struct {
+		result1 []string
+		result2 error
+	}
+	possibleNamespacesReturnsOnCall map[int]struct {
+		result1 []string
+		result2 error
+	}
+	invocations      map[string][][]interface{}
+	invocationsMutex sync.RWMutex
+}
+
+func (fake *NamespaceProvider) PossibleNamespaces(arg1 statedb.VersionedDB) ([]string, error) {
+	fake.possibleNamespacesMutex.Lock()
+	ret, specificReturn := fake.possibleNamespacesReturnsOnCall[len(fake.possibleNamespacesArgsForCall)]
+	fake.possibleNamespacesArgsForCall = append(fake.possibleNamespacesArgsForCall, struct {
+		arg1 statedb.VersionedDB
+	}{arg1})
+	fake.recordInvocation("PossibleNamespaces", []interface{}{arg1})
+	fake.possibleNamespacesMutex.Unlock()
+	if fake.PossibleNamespacesStub != nil {
+		return fake.PossibleNamespacesStub(arg1)
+	}
+	if specificReturn {
+		return ret.result1, ret.result2
+	}
+	fakeReturns := fake.possibleNamespacesReturns
+	return fakeReturns.result1, fakeReturns.result2
+}
+
+func (fake *NamespaceProvider) PossibleNamespacesCallCount() int {
+	fake.possibleNamespacesMutex.RLock()
+	defer fake.possibleNamespacesMutex.RUnlock()
+	return len(fake.possibleNamespacesArgsForCall)
+}
+
+func (fake *NamespaceProvider) PossibleNamespacesCalls(stub func(statedb.VersionedDB) ([]string, error)) {
+	fake.possibleNamespacesMutex.Lock()
+	defer fake.possibleNamespacesMutex.Unlock()
+	fake.PossibleNamespacesStub = stub
+}
+
+func (fake *NamespaceProvider) PossibleNamespacesArgsForCall(i int) statedb.VersionedDB {
+	fake.possibleNamespacesMutex.RLock()
+	defer fake.possibleNamespacesMutex.RUnlock()
+	argsForCall := fake.possibleNamespacesArgsForCall[i]
+	return argsForCall.arg1
+}
+
+func (fake *NamespaceProvider) PossibleNamespacesReturns(result1 []string, result2 error) {
+	fake.possibleNamespacesMutex.Lock()
+	defer fake.possibleNamespacesMutex.Unlock()
+	fake.PossibleNamespacesStub = nil
+	fake.possibleNamespacesReturns = struct {
+		result1 []string
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *NamespaceProvider) PossibleNamespacesReturnsOnCall(i int, result1 []string, result2 error) {
+	fake.possibleNamespacesMutex.Lock()
+	defer fake.possibleNamespacesMutex.Unlock()
+	fake.PossibleNamespacesStub = nil
+	if fake.possibleNamespacesReturnsOnCall == nil {
+		fake.possibleNamespacesReturnsOnCall = make(map[int]struct {
+			result1 []string
+			result2 error
+		})
+	}
+	fake.possibleNamespacesReturnsOnCall[i] = struct {
+		result1 []string
+		result2 error
+	}{result1, result2}
+}
+
+func (fake *NamespaceProvider) Invocations() map[string][][]interface{} {
+	fake.invocationsMutex.RLock()
+	defer fake.invocationsMutex.RUnlock()
+	fake.possibleNamespacesMutex.RLock()
+	defer fake.possibleNamespacesMutex.RUnlock()
+	copiedInvocations := map[string][][]interface{}{}
+	for key, value := range fake.invocations {
+		copiedInvocations[key] = value
+	}
+	return copiedInvocations
+}
+
+func (fake *NamespaceProvider) recordInvocation(key string, args []interface{}) {
+	fake.invocationsMutex.Lock()
+	defer fake.invocationsMutex.Unlock()
+	if fake.invocations == nil {
+		fake.invocations = map[string][][]interface{}{}
+	}
+	if fake.invocations[key] == nil {
+		fake.invocations[key] = [][]interface{}{}
+	}
+	fake.invocations[key] = append(fake.invocations[key], args)
+}
+
+var _ statedb.NamespaceProvider = new(NamespaceProvider)
diff --git a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/commit_handling_test.go b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/commit_handling_test.go
index cc2b425ca7a..0f6e70ca3a3 100644
--- a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/commit_handling_test.go
+++ b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/commit_handling_test.go
@@ -19,7 +19,7 @@ func TestGetRevision(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-get-revisions")
+	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-get-revisions", nil)
 	assert.NoError(t, err)
 	db := versionedDB.(*VersionedDB)
 
@@ -85,7 +85,7 @@ func TestBuildCommittersForNs(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-build-committers-for-ns")
+	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-build-committers-for-ns", nil)
 	assert.NoError(t, err)
 	db := versionedDB.(*VersionedDB)
 
@@ -119,7 +119,7 @@ func TestBuildCommitters(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-build-committers")
+	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-build-committers", nil)
 	assert.NoError(t, err)
 	db := versionedDB.(*VersionedDB)
 
@@ -152,7 +152,7 @@ func TestExecuteCommitter(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-execute-committer")
+	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-execute-committer", nil)
 	assert.NoError(t, err)
 	db := versionedDB.(*VersionedDB)
 
@@ -212,7 +212,7 @@ func TestCommitUpdates(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-commitupdates")
+	versionedDB, err := vdbEnv.DBProvider.GetDBHandle("test-commitupdates", nil)
 	assert.NoError(t, err)
 	db := versionedDB.(*VersionedDB)
 
diff --git a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/redolog_test.go b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/redolog_test.go
index 5ec02464bc2..744272f6c16 100644
--- a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/redolog_test.go
+++ b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/redolog_test.go
@@ -70,7 +70,7 @@ func TestCouchdbRedoLogger(t *testing.T) {
 	commitToRedologAndRestart := func(newVal string, version *version.Height) {
 		batch := statedb.NewUpdateBatch()
 		batch.Put("ns1", "key1", []byte(newVal), version)
-		db, err := vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger")
+		db, err := vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger", nil)
 		assert.NoError(t, err)
 		vdb := db.(*VersionedDB)
 		assert.NoError(t,
@@ -85,7 +85,7 @@ func TestCouchdbRedoLogger(t *testing.T) {
 	}
 	// verifyExpectedVal - a helper function that verifies the statedb contents
 	verifyExpectedVal := func(expectedVal string, expectedSavepoint *version.Height) {
-		db, err := vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger")
+		db, err := vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger", nil)
 		assert.NoError(t, err)
 		vdb := db.(*VersionedDB)
 		vv, err := vdb.GetState("ns1", "key1")
@@ -97,7 +97,7 @@ func TestCouchdbRedoLogger(t *testing.T) {
 	}
 
 	// initialize statedb with initial set of writes
-	db, err := vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger")
+	db, err := vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger", nil)
 	if err != nil {
 		t.Fatalf("Failed to get database handle: %s", err)
 	}
@@ -119,7 +119,7 @@ func TestCouchdbRedoLogger(t *testing.T) {
 	verifyExpectedVal("value2", version.NewHeight(2, 1))
 
 	// A nil height should cause skipping the writing of redo-record
-	db, _ = vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger")
+	db, _ = vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger", nil)
 	vdb = db.(*VersionedDB)
 	vdb.ApplyUpdates(batch1, nil)
 	record, err := vdb.redoLogger.load()
@@ -128,7 +128,7 @@ func TestCouchdbRedoLogger(t *testing.T) {
 	assert.Equal(t, []byte("value3"), record.UpdateBatch.Get("ns1", "key1").Value)
 
 	// A batch that does not contain PostOrderWrites should cause skipping the writing of redo-record
-	db, _ = vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger")
+	db, _ = vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger", nil)
 	vdb = db.(*VersionedDB)
 	batchWithNoGeneratedWrites := batch1
 	batchWithNoGeneratedWrites.ContainsPostOrderWrites = false
@@ -139,7 +139,7 @@ func TestCouchdbRedoLogger(t *testing.T) {
 	assert.Equal(t, []byte("value3"), record.UpdateBatch.Get("ns1", "key1").Value)
 
 	// A batch that contains PostOrderWrites should cause writing of redo-record
-	db, _ = vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger")
+	db, _ = vdbEnv.DBProvider.GetDBHandle("testcouchdbredologger", nil)
 	vdb = db.(*VersionedDB)
 	batchWithGeneratedWrites := batch1
 	batchWithGeneratedWrites.ContainsPostOrderWrites = true
diff --git a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb.go b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb.go
index f75c6d3e5a8..e6c1a8efbb9 100644
--- a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb.go
+++ b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb.go
@@ -138,7 +138,7 @@ func writeDataFormatVersion(couchInstance *couchInstance, dataformatVersion stri
 }
 
 // GetDBHandle gets the handle to a named database
-func (provider *VersionedDBProvider) GetDBHandle(dbName string) (statedb.VersionedDB, error) {
+func (provider *VersionedDBProvider) GetDBHandle(dbName string, nsProvider statedb.NamespaceProvider) (statedb.VersionedDB, error) {
 	provider.mux.Lock()
 	defer provider.mux.Unlock()
 	vdb := provider.databases[dbName]
@@ -149,6 +149,7 @@ func (provider *VersionedDBProvider) GetDBHandle(dbName string) (statedb.Version
 			provider.redoLoggerProvider.newRedoLogger(dbName),
 			dbName,
 			provider.cache,
+			nsProvider,
 		)
 		if err != nil {
 			return nil, err
@@ -184,7 +185,7 @@ type VersionedDB struct {
 }
 
 // newVersionedDB constructs an instance of VersionedDB
-func newVersionedDB(couchInstance *couchInstance, redoLogger *redoLogger, dbName string, cache *cache) (*VersionedDB, error) {
+func newVersionedDB(couchInstance *couchInstance, redoLogger *redoLogger, dbName string, cache *cache, nsProvider statedb.NamespaceProvider) (*VersionedDB, error) {
 	// CreateCouchDatabase creates a CouchDB database object, as well as the underlying database if it does not exist
 	chainName := dbName
 	dbName = constructMetadataDBName(dbName)
@@ -204,20 +205,6 @@ func newVersionedDB(couchInstance *couchInstance, redoLogger *redoLogger, dbName
 		cache:              cache,
 	}
 
-	vdb.channelMetadata, err = vdb.readChannelMetadata()
-	if err != nil {
-		return nil, err
-	}
-	if vdb.channelMetadata == nil {
-		vdb.channelMetadata = &channelMetadata{
-			ChannelName:      chainName,
-			NamespaceDBsInfo: make(map[string]*namespaceDBInfo),
-		}
-		if err = vdb.writeChannelMetadata(); err != nil {
-			return nil, err
-		}
-	}
-
 	logger.Debugf("chain [%s]: checking for redolog record", chainName)
 	redologRecord, err := redoLogger.load()
 	if err != nil {
@@ -228,6 +215,11 @@ func newVersionedDB(couchInstance *couchInstance, redoLogger *redoLogger, dbName
 		return nil, err
 	}
 
+	isNewDB := savepoint == nil
+	if err = vdb.initChannelMetadata(isNewDB, nsProvider); err != nil {
+		return nil, err
+	}
+
 	// in normal circumstances, redolog is expected to be either equal to the last block
 	// committed to the statedb or one ahead (in the event of a crash). However, either of
 	// these or both could be nil on first time start (fresh start/rebuild)
@@ -727,19 +719,6 @@ func (vdb *VersionedDB) Close() {
 	// no need to close db since a shared couch instance is used
 }
 
-// writeChannelMetadata saves channel metadata to metadataDB
-func (vdb *VersionedDB) writeChannelMetadata() error {
-	couchDoc, err := encodeChannelMetadata(vdb.channelMetadata)
-	if err != nil {
-		return err
-	}
-	if _, err := vdb.metadataDB.saveDoc(channelMetadataDocID, "", couchDoc); err != nil {
-		return err
-	}
-	_, err = vdb.metadataDB.ensureFullCommit()
-	return err
-}
-
 // ensureFullCommitAndRecordSavepoint flushes all the dbs (corresponding to `namespaces`) to disk
 // and Record a savepoint in the metadata db.
 // Couch parallelizes writes in cluster or sharded setup and ordering is not guaranteed.
@@ -819,6 +798,56 @@ func (vdb *VersionedDB) GetLatestSavePoint() (*version.Height, error) {
 	return decodeSavepoint(couchDoc)
 }
 
+// initChannelMetadata initizlizes channelMetadata and build NamespaceDBInfo mapping if not present
+func (vdb *VersionedDB) initChannelMetadata(isNewDB bool, namespaceProvider statedb.NamespaceProvider) error {
+	// create channelMetadata with empty NamespaceDBInfo mapping for a new DB
+	if isNewDB {
+		vdb.channelMetadata = &channelMetadata{
+			ChannelName:      vdb.chainName,
+			NamespaceDBsInfo: make(map[string]*namespaceDBInfo),
+		}
+		return vdb.writeChannelMetadata()
+	}
+
+	// read stored channelMetadata from an existing DB
+	var err error
+	vdb.channelMetadata, err = vdb.readChannelMetadata()
+	if vdb.channelMetadata != nil || err != nil {
+		return err
+	}
+
+	// channelMetadata is not present - this is the case when opening older dbs (e.g., v2.0/v2.1) for the first time
+	// create channelMetadata and build NamespaceDBInfo mapping retroactively
+	vdb.channelMetadata = &channelMetadata{
+		ChannelName:      vdb.chainName,
+		NamespaceDBsInfo: make(map[string]*namespaceDBInfo),
+	}
+	// retrieve existing DB names
+	dbNames, err := vdb.couchInstance.retrieveApplicationDBNames()
+	if err != nil {
+		return err
+	}
+	existingDBNames := make(map[string]struct{}, len(dbNames))
+	for _, dbName := range dbNames {
+		existingDBNames[dbName] = struct{}{}
+	}
+	// get namespaces and add a namespace to channelMetadata only if its DB name already exists
+	namespaces, err := namespaceProvider.PossibleNamespaces(vdb)
+	if err != nil {
+		return err
+	}
+	for _, ns := range namespaces {
+		dbName := constructNamespaceDBName(vdb.chainName, ns)
+		if _, ok := existingDBNames[dbName]; ok {
+			vdb.channelMetadata.NamespaceDBsInfo[ns] = &namespaceDBInfo{
+				Namespace: ns,
+				DBName:    dbName,
+			}
+		}
+	}
+	return vdb.writeChannelMetadata()
+}
+
 // readChannelMetadata returns channel metadata stored in metadataDB
 func (vdb *VersionedDB) readChannelMetadata() (*channelMetadata, error) {
 	var err error
@@ -834,6 +863,19 @@ func (vdb *VersionedDB) readChannelMetadata() (*channelMetadata, error) {
 	return decodeChannelMetadata(couchDoc)
 }
 
+// writeChannelMetadata saves channel metadata to metadataDB
+func (vdb *VersionedDB) writeChannelMetadata() error {
+	couchDoc, err := encodeChannelMetadata(vdb.channelMetadata)
+	if err != nil {
+		return err
+	}
+	if _, err := vdb.metadataDB.saveDoc(channelMetadataDocID, "", couchDoc); err != nil {
+		return err
+	}
+	_, err = vdb.metadataDB.ensureFullCommit()
+	return err
+}
+
 // GetFullScanIterator implements method in VersionedDB interface. This function returns a
 // FullScanIterator that can be used to iterate over entire data in the statedb for a channel.
 // `skipNamespace` parameter can be used to control if the consumer wants the FullScanIterator
diff --git a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb_test.go b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb_test.go
index 7df101f3584..f0a9b487721 100644
--- a/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb_test.go
+++ b/core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb_test.go
@@ -22,6 +22,8 @@ import (
 	"github.com/hyperledger/fabric/core/ledger/internal/version"
 	"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/statedb/mock"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -151,7 +153,7 @@ func TestGetStateFromCache(t *testing.T) {
 	defer vdbEnv.cleanup()
 
 	chainID := "testgetstatefromcache"
-	db, err := vdbEnv.DBProvider.GetDBHandle(chainID)
+	db, err := vdbEnv.DBProvider.GetDBHandle(chainID, nil)
 	require.NoError(t, err)
 
 	// scenario 1: get state would receives a
@@ -214,7 +216,7 @@ func TestGetVersionFromCache(t *testing.T) {
 	defer vdbEnv.cleanup()
 
 	chainID := "testgetstatefromcache"
-	db, err := vdbEnv.DBProvider.GetDBHandle(chainID)
+	db, err := vdbEnv.DBProvider.GetDBHandle(chainID, nil)
 	require.NoError(t, err)
 
 	// scenario 1: get version would receives a
@@ -277,7 +279,7 @@ func TestGetMultipleStatesFromCache(t *testing.T) {
 	defer vdbEnv.cleanup()
 
 	chainID := "testgetmultiplestatesfromcache"
-	db, err := vdbEnv.DBProvider.GetDBHandle(chainID)
+	db, err := vdbEnv.DBProvider.GetDBHandle(chainID, nil)
 	require.NoError(t, err)
 
 	// scenario: given 5 keys, get multiple states find
@@ -337,7 +339,7 @@ func TestCacheUpdatesAfterCommit(t *testing.T) {
 	defer vdbEnv.cleanup()
 
 	chainID := "testcacheupdatesaftercommit"
-	db, err := vdbEnv.DBProvider.GetDBHandle(chainID)
+	db, err := vdbEnv.DBProvider.GetDBHandle(chainID, nil)
 	require.NoError(t, err)
 
 	// scenario: cache has 4 keys while the commit operation
@@ -506,7 +508,7 @@ func TestUtilityFunctions(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	db, err := vdbEnv.DBProvider.GetDBHandle("testutilityfunctions")
+	db, err := vdbEnv.DBProvider.GetDBHandle("testutilityfunctions", nil)
 	assert.NoError(t, err)
 
 	// BytesKeySupported should be false for CouchDB
@@ -556,7 +558,7 @@ func TestInvalidJSONFields(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	db, err := vdbEnv.DBProvider.GetDBHandle("testinvalidfields")
+	db, err := vdbEnv.DBProvider.GetDBHandle("testinvalidfields", nil)
 	assert.NoError(t, err)
 
 	db.Open()
@@ -612,7 +614,7 @@ func TestHandleChaincodeDeploy(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	db, err := vdbEnv.DBProvider.GetDBHandle("testinit")
+	db, err := vdbEnv.DBProvider.GetDBHandle("testinit", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -710,7 +712,7 @@ func TestIndexDeploymentWithOrderAndBadSyntax(t *testing.T) {
 	channelName := "ch1"
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
-	db, err := vdbEnv.DBProvider.GetDBHandle(channelName)
+	db, err := vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -786,7 +788,7 @@ func TestPaginatedQuery(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 
-	db, err := vdbEnv.DBProvider.GetDBHandle("testpaginatedquery")
+	db, err := vdbEnv.DBProvider.GetDBHandle("testpaginatedquery", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -986,7 +988,7 @@ func TestApplyUpdatesWithNilHeight(t *testing.T) {
 func TestRangeScanWithCouchInternalDocsPresent(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
-	db, err := vdbEnv.DBProvider.GetDBHandle("testrangescanfiltercouchinternaldocs")
+	db, err := vdbEnv.DBProvider.GetDBHandle("testrangescanfiltercouchinternaldocs", nil)
 	assert.NoError(t, err)
 	couchDatabse, err := db.(*VersionedDB).getNamespaceDBHandle("ns")
 	assert.NoError(t, err)
@@ -1131,7 +1133,7 @@ func testFormatCheck(t *testing.T, dataFormat string, dataExists bool, expectedE
 
 	// create preconditions for test
 	if dataExists {
-		db, err := dbProvider.GetDBHandle("testns")
+		db, err := dbProvider.GetDBHandle("testns", nil)
 		require.NoError(t, err)
 		batch := statedb.NewUpdateBatch()
 		batch.Put("testns", "testkey", []byte("testVal"), version.NewHeight(1, 1))
@@ -1188,7 +1190,7 @@ func TestLoadCommittedVersion(t *testing.T) {
 	defer vdbEnv.cleanup()
 
 	chainID := "testloadcommittedversion"
-	db, err := vdbEnv.DBProvider.GetDBHandle(chainID)
+	db, err := vdbEnv.DBProvider.GetDBHandle(chainID, nil)
 	require.NoError(t, err)
 
 	// scenario: state cache has (ns1, key1), (ns1, key2),
@@ -1283,7 +1285,7 @@ func TestMissingRevisionRetrievalFromDB(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 	chainID := "testmissingrevisionfromdb"
-	db, err := vdbEnv.DBProvider.GetDBHandle(chainID)
+	db, err := vdbEnv.DBProvider.GetDBHandle(chainID, nil)
 	require.NoError(t, err)
 
 	// store key1, key2, key3 to the DB
@@ -1325,7 +1327,7 @@ func TestMissingRevisionRetrievalFromCache(t *testing.T) {
 	defer vdbEnv.cleanup()
 
 	chainID := "testmissingrevisionfromcache"
-	db, err := vdbEnv.DBProvider.GetDBHandle(chainID)
+	db, err := vdbEnv.DBProvider.GetDBHandle(chainID, nil)
 	require.NoError(t, err)
 
 	// scenario 1: missing from cache.
@@ -1358,7 +1360,7 @@ func TestChannelMetadata(t *testing.T) {
 	defer vdbEnv.cleanup()
 	channelName := "testchannelmetadata"
 
-	db, err := vdbEnv.DBProvider.GetDBHandle(channelName)
+	db, err := vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	require.NoError(t, err)
 	vdb := db.(*VersionedDB)
 	expectedChannelMetadata := &channelMetadata{
@@ -1414,12 +1416,12 @@ func TestChannelMetadata_NegativeTests(t *testing.T) {
 	vdbEnv.config.Address = "127.0.0.1:1"
 	expectedErrMsg := fmt.Sprintf("http error calling couchdb: Get \"http://%s/testchannelmetadata-errorpropagation_\": dial tcp %s: connect: connection refused",
 		vdbEnv.config.Address, vdbEnv.config.Address)
-	_, err := vdbEnv.DBProvider.GetDBHandle(channelName)
+	_, err := vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	require.EqualError(t, err, expectedErrMsg)
 	vdbEnv.config.Address = origCouchAddress
 
 	// simulate db connection error by setting an invalid address before getNamespaceDBHandle, verify error is propagated
-	db, err := vdbEnv.DBProvider.GetDBHandle(channelName)
+	db, err := vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	require.NoError(t, err)
 	vdb := db.(*VersionedDB)
 	vdbEnv.config.Address = "127.0.0.1:1"
@@ -1444,7 +1446,7 @@ func TestChannelMetadata_NegativeTests(t *testing.T) {
 	require.NoError(t, err)
 	require.Nil(t, savedChannelMetadata)
 
-	db, err = vdbEnv.DBProvider.GetDBHandle(channelName)
+	db, err = vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	require.NoError(t, err)
 	vdb = db.(*VersionedDB)
 	expectedChannelMetadata := &channelMetadata{
@@ -1479,6 +1481,81 @@ func TestChannelMetadata_NegativeTests(t *testing.T) {
 	require.Equal(t, expectedChannelMetadata, vdb.channelMetadata)
 }
 
+func TestInitChannelMetadta(t *testing.T) {
+	vdbEnv.init(t, sysNamespaces)
+	defer vdbEnv.cleanup()
+	channelName1 := "testinithannelmetadata"
+	channelName2 := "testinithannelmetadata_anotherchannel"
+
+	// create versioned DBs for channelName1 and channelName2
+	db, err := vdbEnv.DBProvider.GetDBHandle(channelName1, nil)
+	require.NoError(t, err)
+	vdb := db.(*VersionedDB)
+	db2, err := vdbEnv.DBProvider.GetDBHandle(channelName2, nil)
+	require.NoError(t, err)
+	vdb2 := db2.(*VersionedDB)
+
+	// prepare test data:
+	// create dbs for channelName1: "ns1" and "ns3", which should match channelName1 namespaces
+	// create dbs for channelName2: "ns2" and "ns4", which should not match any channelName1 namespaces
+	_, err = vdb.getNamespaceDBHandle("ns1")
+	require.NoError(t, err)
+	_, err = vdb.getNamespaceDBHandle("ns3")
+	require.NoError(t, err)
+	_, err = vdb2.getNamespaceDBHandle("ns2")
+	require.NoError(t, err)
+	_, err = vdb2.getNamespaceDBHandle("ns4")
+	require.NoError(t, err)
+
+	namespaces := []string{"ns1", "ns2", "ns3", "ns4"}
+	fakeNsProvider := &mock.NamespaceProvider{}
+	fakeNsProvider.PossibleNamespacesReturns(namespaces, nil)
+	expectedDBsInfo := map[string]*namespaceDBInfo{
+		"ns1": {Namespace: "ns1", DBName: constructNamespaceDBName(channelName1, "ns1")},
+		"ns3": {Namespace: "ns3", DBName: constructNamespaceDBName(channelName1, "ns3")},
+	}
+	expectedChannelMetadata := &channelMetadata{
+		ChannelName:      channelName1,
+		NamespaceDBsInfo: expectedDBsInfo,
+	}
+
+	// test an existing DB with channelMetadata, namespace provider should not be called
+	require.NoError(t, vdb.initChannelMetadata(false, fakeNsProvider))
+	require.Equal(t, expectedChannelMetadata, vdb.channelMetadata)
+	require.Equal(t, 0, fakeNsProvider.PossibleNamespacesCallCount())
+
+	// test an existing DB with no channelMetadata by deleting channelMetadata, namespace provider should be called
+	require.NoError(t, vdb.metadataDB.deleteDoc(channelMetadataDocID, ""))
+	_, err = vdb.metadataDB.ensureFullCommit()
+	require.NoError(t, err)
+	require.NoError(t, vdb.initChannelMetadata(false, fakeNsProvider))
+	require.Equal(t, expectedChannelMetadata, vdb.channelMetadata)
+	require.Equal(t, 1, fakeNsProvider.PossibleNamespacesCallCount())
+	savedChannelMetadata, err := vdb.readChannelMetadata()
+	require.NoError(t, err)
+	require.Equal(t, expectedChannelMetadata, savedChannelMetadata)
+
+	// test namespaceProvider error
+	fakeNsProvider.PossibleNamespacesReturns(nil, errors.New("fake-namespaceprivder-error"))
+	require.NoError(t, vdb.metadataDB.deleteDoc(channelMetadataDocID, ""))
+	_, err = vdb.metadataDB.ensureFullCommit()
+	require.NoError(t, err)
+	err = vdb.initChannelMetadata(false, fakeNsProvider)
+	require.EqualError(t, err, "fake-namespaceprivder-error")
+
+	// test db error
+	origCouchAddress := vdbEnv.config.Address
+	vdbEnv.config.Address = "127.0.0.1:1"
+	vdbEnv.config.MaxRetries = 1
+	vdbEnv.config.MaxRetriesOnStartup = 1
+	expectedErrMsg := fmt.Sprintf("http error calling couchdb: Get \"http://%s/testinithannelmetadata_/channel_metadata?attachments=true\": dial tcp %s: connect: connection refused",
+		vdbEnv.config.Address, vdbEnv.config.Address)
+	vdb.channelMetadata = nil
+	err = vdb.initChannelMetadata(false, fakeNsProvider)
+	require.EqualError(t, err, expectedErrMsg)
+	vdbEnv.config.Address = origCouchAddress
+}
+
 func TestRangeQueryWithInternalLimitAndPageSize(t *testing.T) {
 	// generateSampleData returns a slice of KVs. The returned value contains 12 KVs for a namespace ns1
 	generateSampleData := func() []*statedb.VersionedKV {
@@ -1514,7 +1591,7 @@ func TestRangeQueryWithInternalLimitAndPageSize(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 	channelName := "ch1"
-	vdb, err := vdbEnv.DBProvider.GetDBHandle(channelName)
+	vdb, err := vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	require.NoError(t, err)
 	db := vdb.(*VersionedDB)
 
@@ -1676,7 +1753,7 @@ func TestFullScanIteratorDeterministicJSONOutput(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 	channelName := "ch1"
-	vdb, err := vdbEnv.DBProvider.GetDBHandle(channelName)
+	vdb, err := vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	require.NoError(t, err)
 	db := vdb.(*VersionedDB)
 
@@ -1740,7 +1817,7 @@ func TestFullScanIteratorSkipInternalKeys(t *testing.T) {
 	vdbEnv.init(t, nil)
 	defer vdbEnv.cleanup()
 	channelName := "ch1"
-	vdb, err := vdbEnv.DBProvider.GetDBHandle(channelName)
+	vdb, err := vdbEnv.DBProvider.GetDBHandle(channelName, nil)
 	require.NoError(t, err)
 	db := vdb.(*VersionedDB)
 
diff --git a/core/ledger/kvledger/txmgmt/statedb/statedb.go b/core/ledger/kvledger/txmgmt/statedb/statedb.go
index 37c928be3a2..45ffd55fb85 100644
--- a/core/ledger/kvledger/txmgmt/statedb/statedb.go
+++ b/core/ledger/kvledger/txmgmt/statedb/statedb.go
@@ -15,11 +15,12 @@ import (
 
 //go:generate counterfeiter -o mock/results_iterator.go -fake-name ResultsIterator . ResultsIterator
 //go:generate counterfeiter -o mock/versioned_db.go -fake-name VersionedDB . VersionedDB
+//go:generate counterfeiter -o mock/namespace_provider.go -fake-name NamespaceProvider . NamespaceProvider
 
 // VersionedDBProvider provides an instance of an versioned DB
 type VersionedDBProvider interface {
 	// GetDBHandle returns a handle to a VersionedDB
-	GetDBHandle(id string) (VersionedDB, error)
+	GetDBHandle(id string, namespaceProvider NamespaceProvider) (VersionedDB, error)
 	// Close closes all the VersionedDB instances and releases any resources held by VersionedDBProvider
 	Close()
 }
@@ -77,6 +78,16 @@ type VersionedDB interface {
 	Close()
 }
 
+// NamespaceProvider provides a mean for statedb to get all the possible namespaces for a channel.
+// The intended use is for statecouchdb to retroactively build channel metadata when it is missing,
+// e.g., when opening a statecouchdb from v2.0/2.1 version.
+type NamespaceProvider interface {
+	// PossibleNamespaces returns all possible namespaces for the statedb. Note that it is a superset
+	// of the actual namespaces. Therefore, the caller should compare with the existing databases to
+	// filter out the namespaces that have no matched databases.
+	PossibleNamespaces(vdb VersionedDB) ([]string, error)
+}
+
 //BulkOptimizable interface provides additional functions for
 //databases capable of batch operations
 type BulkOptimizable interface {
diff --git a/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb.go b/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb.go
index bcb59629fcc..844f89e5713 100644
--- a/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb.go
+++ b/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb.go
@@ -49,7 +49,7 @@ func NewVersionedDBProvider(dbPath string) (*VersionedDBProvider, error) {
 }
 
 // GetDBHandle gets the handle to a named database
-func (provider *VersionedDBProvider) GetDBHandle(dbName string) (statedb.VersionedDB, error) {
+func (provider *VersionedDBProvider) GetDBHandle(dbName string, namespaceProvider statedb.NamespaceProvider) (statedb.VersionedDB, error) {
 	return newVersionedDB(provider.dbProvider.GetDBHandle(dbName), dbName), nil
 }
 
diff --git a/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb_test.go b/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb_test.go
index 14010ea7225..3a59a294279 100644
--- a/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb_test.go
+++ b/core/ledger/kvledger/txmgmt/statedb/stateleveldb/stateleveldb_test.go
@@ -57,7 +57,7 @@ func testDataKeyEncoding(t *testing.T, dbName string, ns string, key string) {
 func TestQueryOnLevelDB(t *testing.T) {
 	env := NewTestVDBEnv(t)
 	defer env.Cleanup()
-	db, err := env.DBProvider.GetDBHandle("testquery")
+	db, err := env.DBProvider.GetDBHandle("testquery", nil)
 	assert.NoError(t, err)
 	db.Open()
 	defer db.Close()
@@ -92,7 +92,7 @@ func TestUtilityFunctions(t *testing.T) {
 	env := NewTestVDBEnv(t)
 	defer env.Cleanup()
 
-	db, err := env.DBProvider.GetDBHandle("testutilityfunctions")
+	db, err := env.DBProvider.GetDBHandle("testutilityfunctions", nil)
 	assert.NoError(t, err)
 
 	// BytesKeySupported should be true for goleveldb
@@ -149,7 +149,7 @@ func TestFullScanIteratorErrorPropagation(t *testing.T) {
 	initEnv := func() {
 		env = NewTestVDBEnv(t)
 		vdbProvider = env.DBProvider
-		db, err := vdbProvider.GetDBHandle("TestFullScanIteratorErrorPropagation")
+		db, err := vdbProvider.GetDBHandle("TestFullScanIteratorErrorPropagation", nil)
 		require.NoError(t, err)
 		vdb = db.(*versionedDB)
 		cleanup = func() {
diff --git a/core/ledger/kvledger/txmgmt/validation/combined_iterator_test.go b/core/ledger/kvledger/txmgmt/validation/combined_iterator_test.go
index f3513210bb5..e013b2ac81e 100644
--- a/core/ledger/kvledger/txmgmt/validation/combined_iterator_test.go
+++ b/core/ledger/kvledger/txmgmt/validation/combined_iterator_test.go
@@ -19,7 +19,7 @@ func TestCombinedIterator(t *testing.T) {
 	testDBEnv := stateleveldb.NewTestVDBEnv(t)
 	defer testDBEnv.Cleanup()
 
-	db, err := testDBEnv.DBProvider.GetDBHandle("TestDB")
+	db, err := testDBEnv.DBProvider.GetDBHandle("TestDB", nil)
 	assert.NoError(t, err)
 
 	// populate db with initial data
diff --git a/core/ledger/kvledger/txmgmt/validation/rangequery_validator_test.go b/core/ledger/kvledger/txmgmt/validation/rangequery_validator_test.go
index d38e413493c..bf6a0f3cb43 100644
--- a/core/ledger/kvledger/txmgmt/validation/rangequery_validator_test.go
+++ b/core/ledger/kvledger/txmgmt/validation/rangequery_validator_test.go
@@ -57,7 +57,7 @@ func testRangeQuery(t *testing.T, testcase string, stateData *statedb.UpdateBatc
 	t.Run(testcase, func(t *testing.T) {
 		testDBEnv := stateleveldb.NewTestVDBEnv(t)
 		defer testDBEnv.Cleanup()
-		db, err := testDBEnv.DBProvider.GetDBHandle("TestDB")
+		db, err := testDBEnv.DBProvider.GetDBHandle("TestDB", nil)
 		assert.NoError(t, err)
 		if stateData != nil {
 			db.ApplyUpdates(stateData, savepoint)
diff --git a/core/ledger/ledger_interface.go b/core/ledger/ledger_interface.go
index 958dd356f37..38e283d625d 100644
--- a/core/ledger/ledger_interface.go
+++ b/core/ledger/ledger_interface.go
@@ -606,6 +606,8 @@ type DeployedChaincodeInfoProvider interface {
 	CollectionInfo(channelName, chaincodeName, collectionName string, qe SimpleQueryExecutor) (*peer.StaticCollectionConfig, error)
 	// ImplicitCollections returns a slice that contains one proto msg for each of the implicit collections
 	ImplicitCollections(channelName, chaincodeName string, qe SimpleQueryExecutor) ([]*peer.StaticCollectionConfig, error)
+	// GenerateImplicitCollectionForOrg generates implicit collection for the org
+	GenerateImplicitCollectionForOrg(mspid string) *peer.StaticCollectionConfig
 	// AllCollectionsConfigPkg returns a combined collection config pkg that contains both explicit and implicit collections
 	AllCollectionsConfigPkg(channelName, chaincodeName string, qe SimpleQueryExecutor) (*peer.CollectionConfigPackage, error)
 }
diff --git a/core/ledger/mock/deployed_ccinfo_provider.go b/core/ledger/mock/deployed_ccinfo_provider.go
index 1db98ce1586..10c9b40f87d 100644
--- a/core/ledger/mock/deployed_ccinfo_provider.go
+++ b/core/ledger/mock/deployed_ccinfo_provider.go
@@ -70,6 +70,17 @@ type DeployedChaincodeInfoProvider struct {
 		result1 *peer.StaticCollectionConfig
 		result2 error
 	}
+	GenerateImplicitCollectionForOrgStub        func(string) *peer.StaticCollectionConfig
+	generateImplicitCollectionForOrgMutex       sync.RWMutex
+	generateImplicitCollectionForOrgArgsForCall []struct {
+		arg1 string
+	}
+	generateImplicitCollectionForOrgReturns struct {
+		result1 *peer.StaticCollectionConfig
+	}
+	generateImplicitCollectionForOrgReturnsOnCall map[int]struct {
+		result1 *peer.StaticCollectionConfig
+	}
 	ImplicitCollectionsStub        func(string, string, ledger.SimpleQueryExecutor) ([]*peer.StaticCollectionConfig, error)
 	implicitCollectionsMutex       sync.RWMutex
 	implicitCollectionsArgsForCall []struct {
@@ -372,6 +383,66 @@ func (fake *DeployedChaincodeInfoProvider) CollectionInfoReturnsOnCall(i int, re
 	}{result1, result2}
 }
 
+func (fake *DeployedChaincodeInfoProvider) GenerateImplicitCollectionForOrg(arg1 string) *peer.StaticCollectionConfig {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	ret, specificReturn := fake.generateImplicitCollectionForOrgReturnsOnCall[len(fake.generateImplicitCollectionForOrgArgsForCall)]
+	fake.generateImplicitCollectionForOrgArgsForCall = append(fake.generateImplicitCollectionForOrgArgsForCall, struct {
+		arg1 string
+	}{arg1})
+	fake.recordInvocation("GenerateImplicitCollectionForOrg", []interface{}{arg1})
+	fake.generateImplicitCollectionForOrgMutex.Unlock()
+	if fake.GenerateImplicitCollectionForOrgStub != nil {
+		return fake.GenerateImplicitCollectionForOrgStub(arg1)
+	}
+	if specificReturn {
+		return ret.result1
+	}
+	fakeReturns := fake.generateImplicitCollectionForOrgReturns
+	return fakeReturns.result1
+}
+
+func (fake *DeployedChaincodeInfoProvider) GenerateImplicitCollectionForOrgCallCount() int {
+	fake.generateImplicitCollectionForOrgMutex.RLock()
+	defer fake.generateImplicitCollectionForOrgMutex.RUnlock()
+	return len(fake.generateImplicitCollectionForOrgArgsForCall)
+}
+
+func (fake *DeployedChaincodeInfoProvider) GenerateImplicitCollectionForOrgCalls(stub func(string) *peer.StaticCollectionConfig) {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	defer fake.generateImplicitCollectionForOrgMutex.Unlock()
+	fake.GenerateImplicitCollectionForOrgStub = stub
+}
+
+func (fake *DeployedChaincodeInfoProvider) GenerateImplicitCollectionForOrgArgsForCall(i int) string {
+	fake.generateImplicitCollectionForOrgMutex.RLock()
+	defer fake.generateImplicitCollectionForOrgMutex.RUnlock()
+	argsForCall := fake.generateImplicitCollectionForOrgArgsForCall[i]
+	return argsForCall.arg1
+}
+
+func (fake *DeployedChaincodeInfoProvider) GenerateImplicitCollectionForOrgReturns(result1 *peer.StaticCollectionConfig) {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	defer fake.generateImplicitCollectionForOrgMutex.Unlock()
+	fake.GenerateImplicitCollectionForOrgStub = nil
+	fake.generateImplicitCollectionForOrgReturns = struct {
+		result1 *peer.StaticCollectionConfig
+	}{result1}
+}
+
+func (fake *DeployedChaincodeInfoProvider) GenerateImplicitCollectionForOrgReturnsOnCall(i int, result1 *peer.StaticCollectionConfig) {
+	fake.generateImplicitCollectionForOrgMutex.Lock()
+	defer fake.generateImplicitCollectionForOrgMutex.Unlock()
+	fake.GenerateImplicitCollectionForOrgStub = nil
+	if fake.generateImplicitCollectionForOrgReturnsOnCall == nil {
+		fake.generateImplicitCollectionForOrgReturnsOnCall = make(map[int]struct {
+			result1 *peer.StaticCollectionConfig
+		})
+	}
+	fake.generateImplicitCollectionForOrgReturnsOnCall[i] = struct {
+		result1 *peer.StaticCollectionConfig
+	}{result1}
+}
+
 func (fake *DeployedChaincodeInfoProvider) ImplicitCollections(arg1 string, arg2 string, arg3 ledger.SimpleQueryExecutor) ([]*peer.StaticCollectionConfig, error) {
 	fake.implicitCollectionsMutex.Lock()
 	ret, specificReturn := fake.implicitCollectionsReturnsOnCall[len(fake.implicitCollectionsArgsForCall)]
@@ -563,6 +634,8 @@ func (fake *DeployedChaincodeInfoProvider) Invocations() map[string][][]interfac
 	defer fake.chaincodeInfoMutex.RUnlock()
 	fake.collectionInfoMutex.RLock()
 	defer fake.collectionInfoMutex.RUnlock()
+	fake.generateImplicitCollectionForOrgMutex.RLock()
+	defer fake.generateImplicitCollectionForOrgMutex.RUnlock()
 	fake.implicitCollectionsMutex.RLock()
 	defer fake.implicitCollectionsMutex.RUnlock()
 	fake.namespacesMutex.RLock()
diff --git a/core/scc/lscc/deployedcc_infoprovider.go b/core/scc/lscc/deployedcc_infoprovider.go
index a85c24e1802..144aceaa850 100644
--- a/core/scc/lscc/deployedcc_infoprovider.go
+++ b/core/scc/lscc/deployedcc_infoprovider.go
@@ -62,6 +62,11 @@ func (p *DeployedCCInfoProvider) ImplicitCollections(channelName, chaincodeName
 	return nil, nil
 }
 
+// GenerateImplicitCollectionForOrg is not implemented for legacy chaincodes
+func (p *DeployedCCInfoProvider) GenerateImplicitCollectionForOrg(mspid string) *peer.StaticCollectionConfig {
+	return nil
+}
+
 // ChaincodeInfo implements function in interface ledger.DeployedChaincodeInfoProvider
 func (p *DeployedCCInfoProvider) ChaincodeInfo(channelName, chaincodeName string, qe ledger.SimpleQueryExecutor) (*ledger.DeployedChaincodeInfo, error) {
 	chaincodeDataBytes, err := qe.GetState(lsccNamespace, chaincodeName)