MB-32053: Make Metadata Repo GetIndexDefnByName collection aware

Metadata repo's GetIndexDefnByName should consider scope name
and collection name while retrieving index definition from its
definitions cache

Change-Id: I4be6e62499970ae33bdeabc761cbcbfc0eb54285

secondary/common/util.go

@@ -4,7 +4,6 @@ import (
 	"errors"
 	"expvar"
 	"fmt"
-	"github.com/couchbase/indexing/secondary/common/collections"
 	"hash/crc64"
 	"io"
 	"io/ioutil"
@@ -25,6 +24,7 @@ import (
 
 	"github.com/couchbase/cbauth"
 	"github.com/couchbase/cbauth/cbauthimpl"
+	"github.com/couchbase/indexing/secondary/common/collections"
 	"github.com/couchbase/indexing/secondary/dcp"
 	"github.com/couchbase/indexing/secondary/dcp/transport/client"
 	"github.com/couchbase/indexing/secondary/logging"
@@ -1224,3 +1224,15 @@ func ExpvarHandler(rw http.ResponseWriter, r *http.Request) {
 	})
 	fmt.Fprintf(rw, "\n}\n")
 }
+
+func GetCollectionDefaults(scope, collection string) (string, string) {
+	if scope == "" {
+		scope = DEFAULT_SCOPE
+	}
+
+	if collection == "" {
+		collection = DEFAULT_COLLECTION
+	}
+
+	return scope, collection
+}

secondary/manager/client/metadata_provider.go

@@ -4191,6 +4191,9 @@ func (r *metadataRepo) unmarshallAndUpdateTopology(content []byte, indexerId c.I
 	if err != nil {
 		return err
 	}
+
+	// TODO (Collections): If indexer is old and query is new, scope/collection name will be empty
+	// Populate it with defaults
 	r.updateTopology(topology, indexerId)
 	return nil
 }

secondary/manager/lifecycle.go

@@ -445,6 +445,9 @@ func (m *LifecycleMgr) handlePrepareCreateIndex(content []byte) ([]byte, error)
 		return nil, err
 	}
 
+	prepareCreateIndex.Scope, prepareCreateIndex.Collection = common.GetCollectionDefaults(prepareCreateIndex.Scope,
+		prepareCreateIndex.Collection)
+
 	if prepareCreateIndex.Op == client.PREPARE {
 		if m.prepareLock != nil {
 			if m.prepareLock.RequesterId != prepareCreateIndex.RequesterId ||
@@ -469,8 +472,8 @@ func (m *LifecycleMgr) handlePrepareCreateIndex(content []byte) ([]byte, error)
 		if prepareCreateIndex.Name != "" && prepareCreateIndex.Bucket != "" {
 			// Check for duplicate index name only if name and bucket name
 			// are specified in the request
-			existDefn, err := m.repo.GetIndexDefnByName(prepareCreateIndex.Bucket,
-				prepareCreateIndex.Name)
+			existDefn, err := m.repo.GetIndexDefnByName(prepareCreateIndex.Bucket, prepareCreateIndex.Scope,
+				prepareCreateIndex.Collection, prepareCreateIndex.Name)
 			if err != nil {
 				logging.Infof("LifecycleMgr.handlePrepareCreateIndex() : Reject "+
 					"%v because of error (%v) in GetIndexDefnByName", prepareCreateIndex.DefnId, err)
@@ -1324,7 +1327,7 @@ func (m *LifecycleMgr) setPartition(defn *common.IndexDefn) ([]common.PartitionI
 
 func (m *LifecycleMgr) verifyDuplicateInstance(defn *common.IndexDefn, reqCtx *common.MetadataRequestContext) error {
 
-	existDefn, err := m.repo.GetIndexDefnByName(defn.Bucket, defn.Name)
+	existDefn, err := m.repo.GetIndexDefnByName(defn.Bucket, defn.Scope, defn.Collection, defn.Name)
 	if err != nil {
 		logging.Errorf("LifecycleMgr.CreateIndexInstance() : createIndex fails. Reason = %v", err)
 		return err
@@ -1358,7 +1361,7 @@ func (m *LifecycleMgr) verifyDuplicateInstance(defn *common.IndexDefn, reqCtx *c
 
 func (m *LifecycleMgr) verifyDuplicateDefn(defn *common.IndexDefn, reqCtx *common.MetadataRequestContext) (*common.IndexDefn, error) {
 
-	existDefn, err := m.repo.GetIndexDefnByName(defn.Bucket, defn.Name)
+	existDefn, err := m.repo.GetIndexDefnByName(defn.Bucket, defn.Scope, defn.Collection, defn.Name)
 	if err != nil {
 		logging.Errorf("LifecycleMgr.verifyDuplicateDefn() : createIndex fails. Reason = %v", err)
 		return nil, err
@@ -2328,7 +2331,7 @@ func (m *LifecycleMgr) verifyOverlapPartition(defn *common.IndexDefn, reqCtx *co
 			return err
 		}
 
-		existDefn, err := m.repo.GetIndexDefnByName(defn.Bucket, defn.Name)
+		existDefn, err := m.repo.GetIndexDefnByName(defn.Bucket, defn.Scope, defn.Collection, defn.Name)
 		if err != nil {
 			logging.Errorf("LifecycleMgr.CreateIndexInstance() : createIndex fails. Reason = %v", err)
 			return err

secondary/manager/meta_repo.go

@@ -269,13 +269,14 @@ func (c *MetadataRepo) GetIndexDefnById(id common.IndexDefnId) (*common.IndexDef
 	return defn, nil
 }
 
-func (c *MetadataRepo) GetIndexDefnByName(bucket string, name string) (*common.IndexDefn, error) {
+func (c *MetadataRepo) GetIndexDefnByName(bucket, scope, collection, name string) (*common.IndexDefn, error) {
 
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
 	for _, defn := range c.defnCache {
-		if defn.Name == name && defn.Bucket == bucket {
+		if defn.Name == name && defn.Bucket == bucket &&
+			defn.Scope == scope && defn.Collection == collection {
 			return defn, nil
 		}
 	}