fix: Handle compound filters on related indexed fields (#2575)

## Relevant issue(s)

Resolves #2572

## Description

Handles compound filters targeting related indexed fields, and one-many
joins from the many side.

The invertableJoin issue may be affecting non indexed joins.

There is another issue in this space not solved by this PR:
#2574

planner/filter/complex.go

@@ -33,9 +33,6 @@ func isComplex(conditions any, seekRelation bool) bool {
 			if op, ok := k.(*mapper.Operator); ok {
 				switch op.Operation {
 				case request.FilterOpOr, request.FilterOpAnd, request.FilterOpNot:
-					if v, ok := v.([]any); ok && len(v) == 1 {
-						continue
-					}
 					if isComplex(v, true) {
 						return true
 					}

planner/filter/complex_test.go

@@ -80,7 +80,7 @@ func TestIsComplex(t *testing.T) {
 			inputFilter: r("_or",
 				m("published", m("rating", m("_gt", 4.0))),
 			),
-			isComplex: false,
+			isComplex: true,
 		},
 		{
 			name: "relation inside _or",

planner/planner.go

@@ -361,7 +361,7 @@ func (p *Planner) tryOptimizeJoinDirection(node *invertibleTypeJoin, parentPlan
 	desc := slct.collection.Description()
 	for subFieldName, subFieldInd := range filteredSubFields {
 		indexes := desc.GetIndexesOnField(subFieldName)
-		if len(indexes) > 0 {
+		if len(indexes) > 0 && !filter.IsComplex(parentPlan.selectNode.filter) {
 			subInd := node.documentMapping.FirstIndexOfName(node.subTypeName)
 			relatedField := mapper.Field{Name: node.subTypeName, Index: subInd}
 			fieldFilter := filter.UnwrapRelation(filter.CopyField(

planner/type_join.go

@@ -473,7 +473,6 @@ func (dir *joinDirection) invert() {
 }
 
 type invertibleTypeJoin struct {
-	documentIterator
 	docMapper
 
 	root        planNode
@@ -487,6 +486,9 @@ type invertibleTypeJoin struct {
 	secondaryFieldIndex immutable.Option[int]
 	secondaryFetchLimit uint
 
+	// docsToYield contains documents read and ready to be yielded by this node.
+	docsToYield []core.Doc
+
 	dir joinDirection
 }
 
@@ -556,6 +558,17 @@ func (join *invertibleTypeJoin) processSecondResult(secondDocs []core.Doc) (any,
 }
 
 func (join *invertibleTypeJoin) Next() (bool, error) {
+	if len(join.docsToYield) > 0 {
+		// If there is one or more documents in the queue, drop the first one -
+		// it will have been yielded by the last `Next()` call.
+		join.docsToYield = join.docsToYield[1:]
+		if len(join.docsToYield) > 0 {
+			// If there are still documents in the queue, return true yielding the next
+			// one in the queue.
+			return true, nil
+		}
+	}
+
 	hasFirstValue, err := join.dir.firstNode.Next()
 
 	if err != nil || !hasFirstValue {
@@ -577,7 +590,14 @@ func (join *invertibleTypeJoin) Next() (bool, error) {
 			return false, err
 		}
 		if join.dir.secondNode == join.root {
-			join.root.Value().Fields[join.subSelect.Index] = join.subType.Value()
+			if len(secondDocs) == 0 {
+				return false, nil
+			}
+			for i := range secondDocs {
+				secondDocs[i].Fields[join.subSelect.Index] = join.subType.Value()
+			}
+			join.docsToYield = append(join.docsToYield, secondDocs...)
+			return true, nil
 		} else {
 			secondResult, secondIDResult := join.processSecondResult(secondDocs)
 			join.dir.firstNode.Value().Fields[join.subSelect.Index] = secondResult
@@ -596,11 +616,18 @@ func (join *invertibleTypeJoin) Next() (bool, error) {
 		}
 	}
 
-	join.currentValue = join.root.Value()
+	join.docsToYield = append(join.docsToYield, join.root.Value())
 
 	return true, nil
 }
 
+func (join *invertibleTypeJoin) Value() core.Doc {
+	if len(join.docsToYield) == 0 {
+		return core.Doc{}
+	}
+	return join.docsToYield[0]
+}
+
 func (join *invertibleTypeJoin) invertJoinDirectionWithIndex(
 	fieldFilter *mapper.Filter,
 	index client.IndexDescription,