WIP - Replace planner.getCollectionDesc with db.GetCollectionByName

They both do pretty much the same thing, only the db is a public func and returns a collection instead of just its description.

planner/datasource.go

@@ -11,11 +11,7 @@
 package planner
 
 import (
-	"encoding/json"
-
 	"github.com/sourcenetwork/defradb/client"
-	"github.com/sourcenetwork/defradb/core"
-	"github.com/sourcenetwork/defradb/errors"
 	"github.com/sourcenetwork/defradb/planner/mapper"
 )
 
@@ -36,7 +32,7 @@ func (p *Planner) getSource(parsed *mapper.Select) (planSource, error) {
 }
 
 func (p *Planner) getCollectionScanPlan(parsed *mapper.Select) (planSource, error) {
-	colDesc, err := p.getCollectionDesc(parsed.CollectionName)
+	col, err := p.db.GetCollectionByName(p.ctx, parsed.CollectionName)
 	if err != nil {
 		return planSource{}, err
 	}
@@ -49,30 +45,7 @@ func (p *Planner) getCollectionScanPlan(parsed *mapper.Select) (planSource, erro
 	return planSource{
 		plan: scan,
 		info: sourceInfo{
-			collectionDescription: colDesc,
+			collectionDescription: col.Description(),
 		},
 	}, nil
 }
-
-func (p *Planner) getCollectionDesc(name string) (client.CollectionDescription, error) {
-	collectionKey := core.NewCollectionKey(name)
-	var desc client.CollectionDescription
-	schemaVersionIdBytes, err := p.txn.Systemstore().Get(p.ctx, collectionKey.ToDS())
-	if err != nil {
-		return desc, errors.Wrap("failed to get collection description", err)
-	}
-
-	schemaVersionId := string(schemaVersionIdBytes)
-	schemaVersionKey := core.NewCollectionSchemaVersionKey(schemaVersionId)
-	buf, err := p.txn.Systemstore().Get(p.ctx, schemaVersionKey.ToDS())
-	if err != nil {
-		return desc, err
-	}
-
-	err = json.Unmarshal(buf, &desc)
-	if err != nil {
-		return desc, err
-	}
-
-	return desc, nil
-}

planner/select.go

@@ -424,12 +424,12 @@ func (p *Planner) SelectFromSource(
 	}
 
 	if fromCollection {
-		desc, err := p.getCollectionDesc(selectReq.Name)
+		col, err := p.db.GetCollectionByName(p.ctx, selectReq.Name)
 		if err != nil {
 			return nil, err
 		}
 
-		s.sourceInfo = sourceInfo{desc}
+		s.sourceInfo = sourceInfo{col.Description()}
 	}
 
 	aggregates, err := s.initFields(selectReq)

planner/sum.go

@@ -77,12 +77,12 @@ func (p *Planner) isValueFloat(
 	}
 
 	if !source.ChildTarget.HasValue {
-		parentDescription, err := p.getCollectionDesc(parent.CollectionName)
+		parentCollection, err := p.db.GetCollectionByName(p.ctx, parent.CollectionName)
 		if err != nil {
 			return false, err
 		}
 
-		fieldDescription, fieldDescriptionFound := parentDescription.Schema.GetField(source.Name)
+		fieldDescription, fieldDescriptionFound := parentCollection.Schema().GetField(source.Name)
 		if !fieldDescriptionFound {
 			return false, client.NewErrFieldNotExist(source.Name)
 		}
@@ -125,12 +125,12 @@ func (p *Planner) isValueFloat(
 		return false, nil
 	}
 
-	childCollectionDescription, err := p.getCollectionDesc(child.CollectionName)
+	childCollection, err := p.db.GetCollectionByName(p.ctx, child.CollectionName)
 	if err != nil {
 		return false, err
 	}
 
-	fieldDescription, fieldDescriptionFound := childCollectionDescription.Schema.GetField(source.ChildTarget.Name)
+	fieldDescription, fieldDescriptionFound := childCollection.Schema().GetField(source.ChildTarget.Name)
 	if !fieldDescriptionFound {
 		return false, client.NewErrFieldNotExist(source.ChildTarget.Name)
 	}

planner/type_join.go

@@ -254,16 +254,17 @@ func (p *Planner) makeTypeJoinOne(
 	// check if the field we're querying is the primary side of the relation
 	isPrimary := subTypeFieldDesc.RelationType.IsSet(client.Relation_Type_Primary)
 
-	subTypeCollectionDesc, err := p.getCollectionDesc(subType.CollectionName)
+	subTypeCollection, err := p.db.GetCollectionByName(p.ctx, subType.CollectionName)
 	if err != nil {
 		return nil, err
 	}
 
-	subTypeField, subTypeFieldNameFound := subTypeCollectionDesc.GetFieldByRelation(
+	subTypeSchema := subTypeCollection.Schema()
+	subTypeField, subTypeFieldNameFound := subTypeCollection.Description().GetFieldByRelation(
 		subTypeFieldDesc.RelationName,
 		parent.sourceInfo.collectionDescription.Name,
 		subTypeFieldDesc.Name,
-		&subTypeCollectionDesc.Schema,
+		&subTypeSchema,
 	)
 	if !subTypeFieldNameFound {
 		return nil, client.NewErrFieldNotExist(subTypeFieldDesc.RelationName)
@@ -481,16 +482,17 @@ func (p *Planner) makeTypeJoinMany(
 		return nil, client.NewErrFieldNotExist(subType.Name)
 	}
 
-	subTypeCollectionDesc, err := p.getCollectionDesc(subType.CollectionName)
+	subTypeCollection, err := p.db.GetCollectionByName(p.ctx, subType.CollectionName)
 	if err != nil {
 		return nil, err
 	}
 
-	rootField, rootNameFound := subTypeCollectionDesc.GetFieldByRelation(
+	subTypeSchema := subTypeCollection.Schema()
+	rootField, rootNameFound := subTypeCollection.Description().GetFieldByRelation(
 		subTypeFieldDesc.RelationName,
 		parent.sourceInfo.collectionDescription.Name,
 		subTypeFieldDesc.Name,
-		&subTypeCollectionDesc.Schema,
+		&subTypeSchema,
 	)
 
 	if !rootNameFound {