fix: Compound filter operators with relations (#1855)

## Relevant issue(s)

Resolves #1808 

## Description

Any time a filter with `_and` or `_or` operator that includes relations
is applied it will fail.
There were 2 problems:
- filter is not recursively analysed during mapping phase which would
result in some relational fields not being mapped and later panic upon
attempt to access them.
- the filter is not properly split between the scan and select nodes. It
would just analyse and split top-level fields.
These problems are fixed.

client/request/filter.go

@@ -10,6 +10,12 @@
 
 package request
 
+const (
+	FilterOpOr  = "_or"
+	FilterOpAnd = "_and"
+	FilterOpNot = "_not"
+)
+
 // Filter contains the parsed condition map to be
 // run by the Filter Evaluator.
 // @todo: Cache filter structure for faster condition

planner/filter/complex.go

@@ -0,0 +1,64 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+package filter
+
+import (
+	"github.com/sourcenetwork/defradb/client/request"
+	"github.com/sourcenetwork/defradb/connor"
+	"github.com/sourcenetwork/defradb/planner/mapper"
+)
+
+// IsComplex returns true if the provided filter is complex.
+// A filter is considered complex if it contains a relation
+// object withing an _or operator not necessarily being
+// its direct  child.
+func IsComplex(filter *mapper.Filter) bool {
+	if filter == nil {
+		return false
+	}
+	return isComplex(filter.Conditions, false)
+}
+
+func isComplex(conditions any, seekRelation bool) bool {
+	switch typedCond := conditions.(type) {
+	case map[connor.FilterKey]any:
+		for k, v := range typedCond {
+			if op, ok := k.(*mapper.Operator); ok {
+				if (op.Operation == request.FilterOpOr && len(v.([]any)) > 1) ||
+					op.Operation == request.FilterOpNot {
+					if isComplex(v, true) {
+						return true
+					}
+					continue
+				}
+			}
+			if _, isProp := k.(*mapper.PropertyIndex); isProp && seekRelation {
+				objMap := v.(map[connor.FilterKey]any)
+				for objK := range objMap {
+					if _, isRelation := objK.(*mapper.PropertyIndex); isRelation {
+						return true
+					}
+				}
+			}
+			if isComplex(v, seekRelation) {
+				return true
+			}
+		}
+	case []any:
+		for _, v := range typedCond {
+			if isComplex(v, seekRelation) {
+				return true
+			}
+		}
+	default:
+		return false
+	}
+	return false
+}

planner/filter/complex_test.go

@@ -0,0 +1,175 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+package filter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/sourcenetwork/defradb/client/request"
+	"github.com/sourcenetwork/defradb/planner/mapper"
+)
+
+func TestIsComplex(t *testing.T) {
+	tests := []struct {
+		name        string
+		inputFilter map[string]any
+		isComplex   bool
+	}{
+		{
+			name: "flat structure",
+			inputFilter: map[string]any{
+				"name": m("_eq", "John"),
+				"age":  m("_gt", 55),
+			},
+			isComplex: false,
+		},
+		{
+			name: "fields within _and",
+			inputFilter: r("_and",
+				m("name", m("_eq", "John")),
+				m("age", m("_gt", 55)),
+			),
+			isComplex: false,
+		},
+		{
+			name: "fields within _not",
+			inputFilter: r("_and",
+				m("_not", m("name", m("_eq", "John"))),
+				m("age", m("_gt", 55)),
+			),
+			isComplex: false,
+		},
+		{
+			name: "fields within _or and _and (with _and root)",
+			inputFilter: r("_and",
+				r("_or",
+					r("_and",
+						m("name", m("_eq", "John")),
+						m("age", m("_gt", 30)),
+					),
+				),
+				r("_or",
+					m("name", m("_eq", "Islam")),
+					m("age", m("_lt", 55)),
+				),
+			),
+			isComplex: false,
+		},
+		{
+			name: "fields within _or and _and (with _or root)",
+			inputFilter: r("_or",
+				r("_and",
+					m("name", m("_eq", "John")),
+					m("age", m("_gt", 30)),
+				),
+				m("verified", m("_eq", true)),
+			),
+			isComplex: false,
+		},
+		{
+			name: "only 1 relation within _or",
+			inputFilter: r("_or",
+				m("published", m("rating", m("_gt", 4.0))),
+			),
+			isComplex: false,
+		},
+		{
+			name: "relation inside _or",
+			inputFilter: r("_or",
+				m("published", m("rating", m("_gt", 4.0))),
+				m("age", m("_gt", 30)),
+				m("verified", m("_eq", true)),
+			),
+			isComplex: true,
+		},
+		{
+			name: "relation not inside _or",
+			inputFilter: r("_and",
+				r("_or",
+					m("age", m("_lt", 30)),
+					m("verified", m("_eq", false)),
+				),
+				r("_or",
+					r("_and",
+						m("age", m("_gt", 30)),
+					),
+					m("name", m("_eq", "John")),
+				),
+				r("_and",
+					m("name", m("_eq", "Islam")),
+					m("published", m("rating", m("_gt", 4.0))),
+				),
+			),
+			isComplex: false,
+		},
+		{
+			name: "relation inside _and and _or",
+			inputFilter: r("_and",
+				r("_or",
+					m("age", m("_lt", 30)),
+					m("verified", m("_eq", false)),
+				),
+				r("_or",
+					r("_and",
+						m("published", m("rating", m("_gt", 4.0))),
+						m("age", m("_gt", 30)),
+					),
+					m("name", m("_eq", "John")),
+				),
+			),
+			isComplex: true,
+		},
+		{
+			name: "relation within _not",
+			inputFilter: m("_not",
+				m("published", m("rating", m("_gt", 4.0))),
+			),
+			isComplex: true,
+		},
+		{
+			name: "field inside long _or/_and/_not chain",
+			inputFilter: m("_not", r("_and", r("_or", m("_not", r("_or", r("_and",
+				m("name", m("_eq", "John")))),
+			)))),
+			isComplex: false,
+		},
+		{
+			name: "relation inside _and/_or and _not",
+			inputFilter: r("_and",
+				r("_or",
+					m("age", m("_lt", 30)),
+					m("verified", m("_eq", false)),
+				),
+				r("_or",
+					m("_not",
+						m("published", m("rating", m("_gt", 4.0))),
+					),
+					m("name", m("_eq", "John")),
+				),
+			),
+			isComplex: true,
+		},
+	}
+
+	mapping := getDocMapping()
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			inputFilter := mapper.ToFilter(request.Filter{Conditions: test.inputFilter}, mapping)
+			actual := IsComplex(inputFilter)
+			assert.Equal(t, test.isComplex, actual)
+		})
+	}
+}
+
+func TestIsComplexNullFilter(t *testing.T) {
+	assert.False(t, IsComplex(nil))
+}

planner/filter/copy.go

@@ -0,0 +1,38 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+package filter
+
+import (
+	"github.com/sourcenetwork/defradb/connor"
+)
+
+// Copy performs a deep copy of the provided filter.
+func Copy(filter map[connor.FilterKey]any) map[connor.FilterKey]any {
+	return copyFilterConditions(filter).(map[connor.FilterKey]any)
+}
+
+func copyFilterConditions(conditions any) any {
+	switch typedCond := conditions.(type) {
+	case map[connor.FilterKey]any:
+		result := make(map[connor.FilterKey]any)
+		for key, clause := range typedCond {
+			result[key] = copyFilterConditions(clause)
+		}
+		return result
+	case []any:
+		resultArr := make([]any, len(typedCond))
+		for i, elementClause := range typedCond {
+			resultArr[i] = copyFilterConditions(elementClause)
+		}
+		return resultArr
+	default:
+		return conditions
+	}
+}

planner/filter/copy_field.go

@@ -0,0 +1,73 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+package filter
+
+import (
+	"github.com/sourcenetwork/defradb/connor"
+	"github.com/sourcenetwork/defradb/planner/mapper"
+)
+
+// copyField copies the given field from the provided filter.
+// The result filter preserves the structure of the original filter.
+func copyField(filter *mapper.Filter, field mapper.Field) *mapper.Filter {
+	if filter == nil {
+		return nil
+	}
+	conditionKey := &mapper.PropertyIndex{
+		Index: field.Index,
+	}
+
+	resultFilter := &mapper.Filter{}
+	conditionMap := traverseFilterByProperty(conditionKey, filter.Conditions, false)
+	if len(conditionMap) > 0 {
+		resultFilter.Conditions = conditionMap
+		return resultFilter
+	}
+	return nil
+}
+
+func traverseFilterByProperty(
+	key *mapper.PropertyIndex,
+	conditions map[connor.FilterKey]any,
+	shouldDelete bool,
+) map[connor.FilterKey]any {
+	result := conditions
+	if !shouldDelete {
+		result = make(map[connor.FilterKey]any)
+	}
+	for targetKey, clause := range conditions {
+		if targetKey.Equal(key) {
+			if shouldDelete {
+				delete(result, targetKey)
+			} else {
+				result[key] = clause
+			}
+		} else if opKey, isOpKey := targetKey.(*mapper.Operator); isOpKey {
+			clauseArr, isArr := clause.([]any)
+			if isArr {
+				resultArr := make([]any, 0)
+				for _, elementClause := range clauseArr {
+					elementMap, ok := elementClause.(map[connor.FilterKey]any)
+					if !ok {
+						continue
+					}
+					compoundCond := traverseFilterByProperty(key, elementMap, shouldDelete)
+					if len(compoundCond) > 0 {
+						resultArr = append(resultArr, compoundCond)
+					}
+				}
+				if len(resultArr) > 0 {
+					result[opKey] = resultArr
+				}
+			}
+		}
+	}
+	return result
+}