feat: Add support for aggregate filters on inline arrays (sourcenetwo…

…rk#622)

* Remove out of date comment

* Remove unuseful 0 check

* Remove filter type restrictions

This needs to be able to accomodate inline arrays shortly

* Add support for inline array aggregate filters

connor/connor.go

@@ -2,13 +2,11 @@ package connor
 
 import (
 	"fmt"
-
-	"github.com/sourcenetwork/defradb/core"
 )
 
 // Match is the default method used in Connor to match some data to a
 // set of conditions.
-func Match(conditions map[FilterKey]interface{}, data core.Doc) (bool, error) {
+func Match(conditions map[FilterKey]interface{}, data interface{}) (bool, error) {
 	return eq(conditions, data)
 }
 

query/graphql/mapper/mapper.go

@@ -134,12 +134,15 @@ func resolveAggregates(
 				fieldDesc, isField := desc.GetField(target.hostExternalName)
 				if isField && !fieldDesc.IsObject() {
 					// If the hostExternalName matches a non-object field
-					// we can just take it as a field-requestable as only
-					// objects are targetable-requestables.
+					// we don't have to search for it and can just construct the
+					// targeting info here.
 					hasHost = true
-					host = &Field{
-						Index: int(fieldDesc.ID),
-						Name:  target.hostExternalName,
+					host = &Targetable{
+						Field: Field{
+							Index: int(fieldDesc.ID),
+							Name:  target.hostExternalName,
+						},
+						Filter: ToFilter(target.filter, mapping),
 					}
 				} else {
 					childObjectIndex := mapping.FirstIndexOfName(target.hostExternalName)
@@ -828,7 +831,7 @@ func toOrderBy(source *parserTypes.OrderBy, mapping *core.DocumentMapping) *Orde
 
 // RunFilter runs the given filter expression
 // using the document, and evaluates.
-func RunFilter(doc core.Doc, filter *Filter) (bool, error) {
+func RunFilter(doc interface{}, filter *Filter) (bool, error) {
 	if filter == nil {
 		return true, nil
 	}

query/graphql/mapper/targetable.go

@@ -151,3 +151,7 @@ func (t *Targetable) cloneTo(index int) *Targetable {
 		OrderBy: t.OrderBy,
 	}
 }
+
+func (t *Targetable) AsTargetable() (*Targetable, bool) {
+	return t, true
+}

query/graphql/planner/count.go

@@ -98,14 +98,54 @@ func (n *countNode) Next() (bool, error) {
 		switch v.Kind() {
 		// v.Len will panic if v is not one of these types, we don't want it to panic
 		case reflect.Array, reflect.Chan, reflect.Map, reflect.Slice, reflect.String:
-			length := v.Len()
-			// For now, we only support count filters internally to support averages
-			// so this is fine here now, but may need to be moved later once external
-			// count filter support is added.
-			if count > 0 && source.Filter != nil {
-				docArray, isDocArray := property.([]core.Doc)
-				if isDocArray {
-					for _, doc := range docArray {
+			if source.Filter != nil {
+				switch array := property.(type) {
+				case []core.Doc:
+					for _, doc := range array {
+						passed, err := mapper.RunFilter(doc, source.Filter)
+						if err != nil {
+							return false, err
+						}
+						if passed {
+							count += 1
+						}
+					}
+
+				case []bool:
+					for _, doc := range array {
+						passed, err := mapper.RunFilter(doc, source.Filter)
+						if err != nil {
+							return false, err
+						}
+						if passed {
+							count += 1
+						}
+					}
+
+				case []int64:
+					for _, doc := range array {
+						passed, err := mapper.RunFilter(doc, source.Filter)
+						if err != nil {
+							return false, err
+						}
+						if passed {
+							count += 1
+						}
+					}
+
+				case []float64:
+					for _, doc := range array {
+						passed, err := mapper.RunFilter(doc, source.Filter)
+						if err != nil {
+							return false, err
+						}
+						if passed {
+							count += 1
+						}
+					}
+
+				case []string:
+					for _, doc := range array {
 						passed, err := mapper.RunFilter(doc, source.Filter)
 						if err != nil {
 							return false, err
@@ -116,7 +156,7 @@ func (n *countNode) Next() (bool, error) {
 					}
 				}
 			} else {
-				count = count + length
+				count = count + v.Len()
 			}
 		}
 	}

query/graphql/planner/sum.go

@@ -223,10 +223,24 @@ func (n *sumNode) Next() (bool, error) {
 			}
 		case []int64:
 			for _, childItem := range childCollection {
+				passed, err := mapper.RunFilter(childItem, source.Filter)
+				if err != nil {
+					return false, err
+				}
+				if !passed {
+					continue
+				}
 				sum += float64(childItem)
 			}
 		case []float64:
 			for _, childItem := range childCollection {
+				passed, err := mapper.RunFilter(childItem, source.Filter)
+				if err != nil {
+					return false, err
+				}
+				if !passed {
+					continue
+				}
 				sum += childItem
 			}
 		}

query/graphql/schema/generate.go

@@ -291,15 +291,28 @@ func (g *Generator) createExpandedFieldAggregate(
 ) {
 	for _, aggregateTarget := range f.Args {
 		target := aggregateTarget.Name()
-		var targetType string
+		var filterTypeName string
 		if target == parserTypes.GroupFieldName {
-			targetType = obj.Name()
+			filterTypeName = obj.Name() + "FilterArg"
 		} else {
-			targetType = obj.Fields()[target].Type.Name()
+			filterType := obj.Fields()[target].Type
+			if list, isList := filterType.(*gql.List); isList && gql.IsLeafType(list.OfType) {
+				// If it is a list of leaf types - the filter is just the set of OperatorBlocks
+				// that are supported by this type - there can be no field selections.
+				if notNull, isNotNull := list.OfType.(*gql.NonNull); isNotNull {
+					// GQL does not support '!' in type names, and so we have to manipulate the
+					// underlying name like this if it is a nullable type.
+					filterTypeName = fmt.Sprintf("NotNull%sOperatorBlock", notNull.OfType.Name())
+				} else {
+					filterTypeName = genTypeName(list.OfType, "OperatorBlock")
+				}
+			} else {
+				filterTypeName = filterType.Name() + "FilterArg"
+			}
 		}
 
 		expandedField := &gql.InputObjectFieldConfig{
-			Type: g.manager.schema.TypeMap()[targetType+"FilterArg"],
+			Type: g.manager.schema.TypeMap()[filterTypeName],
 		}
 		aggregateTarget.Type.(*gql.InputObject).AddFieldConfig("filter", expandedField)
 	}

query/graphql/schema/manager.go

@@ -146,11 +146,15 @@ func defaultTypes() []gql.Type {
 
 		// Filter scalar blocks
 		booleanOperatorBlock,
+		notNullBooleanOperatorBlock,
 		dateTimeOperatorBlock,
 		floatOperatorBlock,
+		notNullFloatOperatorBlock,
 		idOperatorBlock,
 		intOperatorBlock,
+		notNullIntOperatorBlock,
 		stringOperatorBlock,
+		notNullstringOperatorBlock,
 
 		schemaTypes.CommitLinkObject,
 		schemaTypes.CommitObject,

query/graphql/schema/root.go

@@ -49,6 +49,25 @@ var booleanOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	},
 })
 
+// notNullBooleanOperatorBlock filter block for boolean! types.
+var notNullBooleanOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
+	Name: "NotNullBooleanOperatorBlock",
+	Fields: gql.InputObjectConfigFieldMap{
+		"_eq": &gql.InputObjectFieldConfig{
+			Type: gql.Boolean,
+		},
+		"_ne": &gql.InputObjectFieldConfig{
+			Type: gql.Boolean,
+		},
+		"_in": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.Boolean)),
+		},
+		"_nin": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.Boolean)),
+		},
+	},
+})
+
 // dateTimeOperatorBlock filter block for DateTime types.
 var dateTimeOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	Name: "DateTimeOperatorBlock",
@@ -111,6 +130,37 @@ var floatOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	},
 })
 
+// notNullFloatOperatorBlock filter block for Float! types.
+var notNullFloatOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
+	Name: "NotNullFloatOperatorBlock",
+	Fields: gql.InputObjectConfigFieldMap{
+		"_eq": &gql.InputObjectFieldConfig{
+			Type: gql.Float,
+		},
+		"_ne": &gql.InputObjectFieldConfig{
+			Type: gql.Float,
+		},
+		"_gt": &gql.InputObjectFieldConfig{
+			Type: gql.Float,
+		},
+		"_ge": &gql.InputObjectFieldConfig{
+			Type: gql.Float,
+		},
+		"_lt": &gql.InputObjectFieldConfig{
+			Type: gql.Float,
+		},
+		"_le": &gql.InputObjectFieldConfig{
+			Type: gql.Float,
+		},
+		"_in": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.Float)),
+		},
+		"_nin": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.Float)),
+		},
+	},
+})
+
 // intOperatorBlock filter block for Int types.
 var intOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	Name: "IntOperatorBlock",
@@ -142,6 +192,37 @@ var intOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	},
 })
 
+// notNullIntOperatorBlock filter block for Int! types.
+var notNullIntOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
+	Name: "NotNullIntOperatorBlock",
+	Fields: gql.InputObjectConfigFieldMap{
+		"_eq": &gql.InputObjectFieldConfig{
+			Type: gql.Int,
+		},
+		"_ne": &gql.InputObjectFieldConfig{
+			Type: gql.Int,
+		},
+		"_gt": &gql.InputObjectFieldConfig{
+			Type: gql.Int,
+		},
+		"_ge": &gql.InputObjectFieldConfig{
+			Type: gql.Int,
+		},
+		"_lt": &gql.InputObjectFieldConfig{
+			Type: gql.Int,
+		},
+		"_le": &gql.InputObjectFieldConfig{
+			Type: gql.Int,
+		},
+		"_in": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.Int)),
+		},
+		"_nin": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.Int)),
+		},
+	},
+})
+
 // stringOperatorBlock filter block for string types.
 var stringOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	Name: "StringOperatorBlock",
@@ -164,6 +245,25 @@ var stringOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	},
 })
 
+// notNullstringOperatorBlock filter block for string! types.
+var notNullstringOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
+	Name: "NotNullStringOperatorBlock",
+	Fields: gql.InputObjectConfigFieldMap{
+		"_eq": &gql.InputObjectFieldConfig{
+			Type: gql.String,
+		},
+		"_ne": &gql.InputObjectFieldConfig{
+			Type: gql.String,
+		},
+		"_in": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.String)),
+		},
+		"_nin": &gql.InputObjectFieldConfig{
+			Type: gql.NewList(gql.NewNonNull(gql.String)),
+		},
+	},
+})
+
 // idOperatorBlock filter block for ID types.
 var idOperatorBlock = gql.NewInputObject(gql.InputObjectConfig{
 	Name: "IDOperatorBlock",