PROPOSAL - rework property transform stuff

query/graphql/parser/commit.go

@@ -45,8 +45,6 @@ type CommitSelect struct {
 
 	Limit   *Limit
 	OrderBy *OrderBy
-	Counts  []PropertyTransformation
-	Sums    []PropertyTransformation
 
 	Fields []Selection
 
@@ -73,21 +71,12 @@ func (c CommitSelect) GetSelections() []Selection {
 	return c.Fields
 }
 
-func (s *CommitSelect) AddCount(transformationDefinition PropertyTransformation) {
-	s.Counts = append(s.Counts, transformationDefinition)
-}
-
-func (s *CommitSelect) AddSum(transformationDefinition PropertyTransformation) {
-	s.Sums = append(s.Sums, transformationDefinition)
-}
-
 func (c CommitSelect) ToSelect() *Select {
 	return &Select{
 		Name:      c.Name,
 		Alias:     c.Alias,
 		Limit:     c.Limit,
 		OrderBy:   c.OrderBy,
-		Counts:    c.Counts,
 		Statement: c.Statement,
 		Fields:    c.Fields,
 		Root:      CommitSelection,
@@ -131,7 +120,5 @@ func parseCommitSelect(field *ast.Field) (*CommitSelect, error) {
 	var err error
 	commit.Fields, err = parseSelectFields(commit.GetRoot(), field.SelectionSet)
 
-	parseAggregates(commit)
-
 	return commit, err
 }

query/graphql/parser/query.go

@@ -47,6 +47,11 @@ var ReservedFields = map[string]bool{
 	DocKeyFieldName:  true,
 }
 
+var aggregates = map[string]struct{}{
+	CountFieldName: {},
+	SumFieldName:   {},
+}
+
 type Query struct {
 	Queries   []*OperationDefinition
 	Mutations []*OperationDefinition
@@ -80,12 +85,6 @@ type Selection interface {
 	GetRoot() SelectionType
 }
 
-type baseSelect interface {
-	Selection
-	AddCount(transformationDefinition PropertyTransformation)
-	AddSum(transformationDefinition PropertyTransformation)
-}
-
 // Select is a complex Field with strong typing
 // It used for sub types in a query. Includes
 // fields, and query arguments like filters,
@@ -105,8 +104,6 @@ type Select struct {
 	Limit   *Limit
 	OrderBy *OrderBy
 	GroupBy *GroupBy
-	Counts  []PropertyTransformation
-	Sums    []PropertyTransformation
 
 	Fields []Selection
 
@@ -134,14 +131,6 @@ func (s Select) GetAlias() string {
 	return s.Alias
 }
 
-func (s *Select) AddCount(transformationDefinition PropertyTransformation) {
-	s.Counts = append(s.Counts, transformationDefinition)
-}
-
-func (s *Select) AddSum(transformationDefinition PropertyTransformation) {
-	s.Sums = append(s.Sums, transformationDefinition)
-}
-
 // Field implements Selection
 type Field struct {
 	Name  string
@@ -178,14 +167,6 @@ type GroupBy struct {
 	Fields []string
 }
 
-// Contains mapping information between a source and destination properties
-type PropertyTransformation struct {
-	// Where the result of transformation should be written to
-	Destination string
-	// Where the data to be transformed should be read from
-	Source []string
-}
-
 type SortDirection string
 
 const (
@@ -398,11 +379,6 @@ func parseSelect(rootType SelectionType, field *ast.Field) (*Select, error) {
 		return nil, err
 	}
 
-	err = parseAggregates(slct)
-	if err != nil {
-		return nil, err
-	}
-
 	return slct, err
 }
 
@@ -413,7 +389,10 @@ func parseSelectFields(root SelectionType, fields *ast.SelectionSet) ([]Selectio
 		switch node := selection.(type) {
 		case *ast.Field:
 			if node.SelectionSet == nil { // regular field
-				f := parseField(root, node)
+				f, err := parseField(i, root, node)
+				if err != nil {
+					return nil, err
+				}
 				selections[i] = f
 			} else { // sub type with extra fields
 				subroot := root
@@ -435,16 +414,31 @@ func parseSelectFields(root SelectionType, fields *ast.SelectionSet) ([]Selectio
 
 // parseField simply parses the Name/Alias
 // into a Field type
-func parseField(root SelectionType, field *ast.Field) *Field {
+func parseField(i int, root SelectionType, field *ast.Field) (*Field, error) {
+	var name string
+	var alias string
+
+	if _, isAggregate := aggregates[field.Name.Value]; isAggregate {
+		name = fmt.Sprintf("_agg%v", i)
+		if field.Alias == nil {
+			alias = field.Name.Value
+		} else {
+			alias = field.Alias.Value
+		}
+	} else {
+		name = field.Name.Value
+		if field.Alias != nil {
+			alias = field.Alias.Value
+		}
+	}
+
 	f := &Field{
 		Root:      root,
-		Name:      field.Name.Value,
+		Name:      name,
 		Statement: field,
+		Alias:     alias,
 	}
-	if field.Alias != nil {
-		f.Alias = field.Alias.Value
-	}
-	return f
+	return f, nil
 }
 
 func parseAPIQuery(field *ast.Field) (Selection, error) {
@@ -456,49 +450,19 @@ func parseAPIQuery(field *ast.Field) (Selection, error) {
 	}
 }
 
-// Parses requested aggregates, creating a virtual name (alias) if an alias is not provided to allow for multiple aggregate
-// fields.  Mutates any aggregate field properties on the select, and adds the result onto the given select object.
-func parseAggregates(slct baseSelect) error {
-	for i, field := range slct.GetSelections() {
-		switch field.GetName() {
-		case CountFieldName:
-			err, transformation := parseAggregate(i, field)
-			if err != nil {
-				return err
-			}
-			slct.AddCount(transformation)
-		case SumFieldName:
-			err, transformation := parseAggregate(i, field)
-			if err != nil {
-				return err
-			}
-			slct.AddSum(transformation)
-		}
-	}
-
-	return nil
-}
-
-// Parses the given aggregate field selector, mutating the given field and returning the resultant PropertyTransformation
-func parseAggregate(i int, field Selection) (error, PropertyTransformation) {
-	virtualName := fmt.Sprintf("_agg%v", i)
-	f := field.(*Field)
-	if f.Alias == "" {
-		f.Alias = f.Name
-	}
-	f.Name = virtualName
-
+// Returns the source of the aggregate as requested by the consumer
+func (field Field) GetAggregateSource() ([]string, error) {
 	var path []string
-	if len(f.Statement.Arguments) == 0 {
+
+	if len(field.Statement.Arguments) == 0 {
 		path = []string{}
 	} else {
-		switch arguementValue := f.Statement.Arguments[0].Value.GetValue().(type) {
+		switch arguementValue := field.Statement.Arguments[0].Value.GetValue().(type) {
 		case string:
 			path = []string{arguementValue}
 		case []*ast.ObjectField:
 			if len(arguementValue) == 0 {
-				//Note: Scalar arrays will hit this clause and should be handled appropriately (not adding now as not testable in a time-efficient manner)
-				return fmt.Errorf("Unexpected error: aggregate field contained no child field selector"), PropertyTransformation{}
+				return []string{}, fmt.Errorf("Unexpected error: aggregate field contained no child field selector")
 			}
 			innerPath := arguementValue[0].Value.GetValue()
 			if innerPathStringValue, isString := innerPath.(string); isString {
@@ -510,5 +474,5 @@ func parseAggregate(i int, field Selection) (error, PropertyTransformation) {
 		}
 	}
 
-	return nil, PropertyTransformation{Destination: virtualName, Source: path}
+	return path, nil
 }

query/graphql/planner/count.go

@@ -28,18 +28,23 @@ type countNode struct {
 	virtualFieldId string
 }
 
-func (p *Planner) Count(c *parser.PropertyTransformation) (*countNode, error) {
+func (p *Planner) Count(field *parser.Field) (*countNode, error) {
+	source, err := field.GetAggregateSource()
+	if err != nil {
+		return nil, err
+	}
+
 	var sourceProperty string
-	if len(c.Source) == 1 {
-		sourceProperty = c.Source[0]
+	if len(source) == 1 {
+		sourceProperty = source[0]
 	} else {
 		sourceProperty = ""
 	}
 
 	return &countNode{
 		p:              p,
 		sourceProperty: sourceProperty,
-		virtualFieldId: c.Destination,
+		virtualFieldId: field.Name,
 	}, nil
 }
 
@@ -74,3 +79,5 @@ func (n *countNode) Values() map[string]interface{} {
 func (n *countNode) Next() (bool, error) {
 	return n.plan.Next()
 }
+
+func (n *countNode) SetPlan(p planNode) { n.plan = p }

query/graphql/planner/planner.go

@@ -224,15 +224,15 @@ func (p *Planner) expandSelectTopNodePlan(plan *selectTopNode, parentPlan *selec
 	return nil
 }
 
-func (p *Planner) expandAggregatePlans(plan *selectTopNode) {
-	for _, countPlan := range plan.countPlans {
-		countPlan.plan = plan.plan
-		plan.plan = countPlan
-	}
+type aggregateNode interface {
+	planNode
+	SetPlan(plan planNode)
+}
 
-	for _, sumPlan := range plan.sumPlans {
-		sumPlan.plan = plan.plan
-		plan.plan = sumPlan
+func (p *Planner) expandAggregatePlans(plan *selectTopNode) {
+	for _, aggregate := range plan.aggregates {
+		aggregate.SetPlan(plan.plan)
+		plan.plan = aggregate
 	}
 }
 
@@ -305,7 +305,7 @@ func (p *Planner) expandLimitPlan(plan *selectTopNode, parentPlan *selectTopNode
 		// if this is a child node, and the parent select has an aggregate then we need to
 		// replace the hard limit with a render limit to allow the full set of child records
 		// to be aggregated
-		if parentPlan != nil && (len(parentPlan.countPlans) > 0 || len(parentPlan.sumPlans) > 0) {
+		if parentPlan != nil && len(parentPlan.aggregates) > 0 {
 			renderLimit, err := p.RenderLimit(&parser.Limit{
 				Offset: l.offset,
 				Limit:  l.limit,