SOLR-17923: Add fullOuterJoin stream function (#3676)

Author: andywebb1975

solr/CHANGES.txt

@@ -277,6 +277,8 @@ New Features
 ---------------------
 * SOLR-17915: shards.preference=replica.location now supports the "host" option for routing to replicas on the same host. (Houston Putman)
 
+* SOLR-17923: Add fullOuterJoin stream function (Andy Webb)
+
 Improvements
 ---------------------
 * SOLR-17860: DocBasedVersionConstraintsProcessorFactory now supports PULL replicas. (Houston Putman)

solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc

@@ -786,73 +786,42 @@ fetch(addresses,
 
 The example above fetches addresses for users by matching the username in the tuple with the userId field in the addresses collection.
 
-== having
-
-The `having` expression wraps a stream and applies a boolean operation to each tuple.
-It emits only tuples for which the boolean operation returns *true*.
-
-=== having Parameters
-
-* `StreamExpression`: (Mandatory) The stream source for the having function.
-* `booleanEvaluator`: (Mandatory) The following boolean operations are supported: `eq` (equals), `gt` (greater than), `lt` (less than), `gteq` (greater than or equal to), `lteq` (less than or equal to), `and`, `or`, `eor` (exclusive or), and `not`.
-Boolean evaluators can be nested with other evaluators to form complex boolean logic.
-
-The comparison evaluators compare the value in a specific field with a value, whether a string, number, or boolean.
-For example: `eq(field1, 10)`, returns `true` if `field1` is equal to 10.
-
-=== having Syntax
-
-[source,text]
-----
-having(rollup(over=a_s,
-              sum(a_i),
-              search(collection1,
-                     q="*:*",
-                     qt="/export",
-                     fl="id,a_s,a_i,a_f",
-                     sort="a_s asc")),
-       and(gt(sum(a_i), 100), lt(sum(a_i), 110)))
-
-----
-
-In this example, the `having` expression iterates the aggregated tuples from the `rollup` expression and emits all tuples where the field `sum(a_i)` is greater than 100 and less than 110.
-
-== leftOuterJoin
+== fullOuterJoin
 
-The `leftOuterJoin` function wraps two streams, Left and Right, and emits tuples from Left.
-If there is a tuple in Right equal (as defined by `on`) then the values in that tuple will be included in the emitted tuple.
-An equal tuple in Right *need not* exist for the Left tuple to be emitted.
-This supports one-to-one, one-to-many, many-to-one, and many-to-many left outer join scenarios.
-The tuples are emitted in the order in which they appear in the Left stream.
+The `fullOuterJoin` function wraps two streams, Left and Right, and emits tuples from both.
+If there is a tuple in Right equal to that in the Left (as defined by `on`) then the values in that tuple will be included in the emitted tuple.
+An equal tuple in one stream *need not* exist for a tuple in the other to be emitted.
+This supports one-to-one, one-to-many, many-to-one, and many-to-many full outer join scenarios.
+The tuples are emitted in the order in which they appear in the streams.
 Both streams must be sorted by the fields being used to determine equality (using the `on` parameter).
 If both tuples contain a field of the same name then the value from the Right stream will be used in the emitted tuple.
 
 You can wrap the incoming streams with a `select` function to be specific about which field values are included in the emitted tuple.
 
-=== leftOuterJoin Parameters
+=== fullOuterJoin Parameters
 
 * `StreamExpression for StreamLeft`
 * `StreamExpression for StreamRight`
 * `on`: Fields to be used for checking equality of tuples between Left and Right.
 Can be of the format `on="fieldName"`, `on="fieldNameInLeft=fieldNameInRight"`, or `on="fieldName, otherFieldName=rightOtherFieldName"`.
 
-=== leftOuterJoin Syntax
+=== fullOuterJoin Syntax
 
 [source,text]
 ----
-leftOuterJoin(
+fullOuterJoin(
   search(people, q="*:*", qt="/export", fl="personId,name", sort="personId asc"),
   search(pets, q="type:cat", qt="/export", fl="personId,petName", sort="personId asc"),
   on="personId"
 )
 
-leftOuterJoin(
+fullOuterJoin(
   search(people, q="*:*", qt="/export", fl="personId,name", sort="personId asc"),
   search(pets, q="type:cat", qt="/export", fl="ownerId,petName", sort="ownerId asc"),
   on="personId=ownerId"
 )
 
-leftOuterJoin(
+fullOuterJoin(
   search(people, q="*:*", qt="/export", fl="personId,name", sort="personId asc"),
   select(
     search(pets, q="type:cat", qt="/export", fl="ownerId,name", sort="ownerId asc"),
@@ -863,6 +832,103 @@ leftOuterJoin(
 )
 ----
 
+=== Reciprocal Rank Fusion (RRF) using fullOuterJoin
+
+The `fullOuterJoin` function can be used to construct an RRF algorithm for merging together result sets, as illustrated below.
+
+(In a real-world example the two `sort/list/tuple` blocks would likely use `search` instead.)
+
+[source,text]
+----
+top(
+    n=10,
+    sort(
+        select(
+            fullOuterJoin(
+                sort(
+                    select(
+                        sort(
+                            list(
+                                tuple(id=1, title="L 1", left="a", score=4.5),
+                                tuple(id=2, title="L 2", left="b", score=3.5),
+                                tuple(id=3, title="L 3", left="c", score=2.5),
+                                tuple(id=4, title="L 4", left="d", score=2.5),
+                                tuple(id=5, title="L 5", left="e", score=2.5),
+                                tuple(id=6, title="L 6", left="f", score=2.5),
+                            ),
+                            by="score desc"
+                        ),
+                        *,
+                        score as scoreL,
+                        add(recNum(),1) as rankL,
+                        div(1,add(rankL,60)) as rrL
+                    ),
+                    by="id asc"
+                ),
+                sort(
+                    select(
+                        sort(
+                            list(
+                                tuple(id=3, title="R 3", right="g", score=0.9),
+                                tuple(id=2, title="R 2", right="h", score=0.8),
+                                tuple(id=4, title="R 4", right="i", score=0.7),
+                                tuple(id=7, title="R 7", right="j", score=0.6),
+                                tuple(id=8, title="R 8", right="k", score=0.5),
+                                tuple(id=9, title="R 9", right="l", score=0.5),
+                            ),
+                            by="score desc"
+                        ),
+                        *,
+                        score as scoreR,
+                        add(recNum(),1) as rankR,
+                        div(1,add(rankR,60)) as rrR
+                    ),
+                    by="id asc"
+                ),
+                on="id"
+            ),
+            *,
+            replace(rrL,null,withValue=0),
+            replace(rrR,null,withValue=0),
+            add(rrL,rrR) as rrf,
+        ),
+        by="rrf desc"
+    ),
+    sort="rrf desc"
+)
+----
+
+== having
+
+The `having` expression wraps a stream and applies a boolean operation to each tuple.
+It emits only tuples for which the boolean operation returns *true*.
+
+=== having Parameters
+
+* `StreamExpression`: (Mandatory) The stream source for the having function.
+* `booleanEvaluator`: (Mandatory) The following boolean operations are supported: `eq` (equals), `gt` (greater than), `lt` (less than), `gteq` (greater than or equal to), `lteq` (less than or equal to), `and`, `or`, `eor` (exclusive or), and `not`.
+Boolean evaluators can be nested with other evaluators to form complex boolean logic.
+
+The comparison evaluators compare the value in a specific field with a value, whether a string, number, or boolean.
+For example: `eq(field1, 10)`, returns `true` if `field1` is equal to 10.
+
+=== having Syntax
+
+[source,text]
+----
+having(rollup(over=a_s,
+              sum(a_i),
+              search(collection1,
+                     q="*:*",
+                     qt="/export",
+                     fl="id,a_s,a_i,a_f",
+                     sort="a_s asc")),
+       and(gt(sum(a_i), 100), lt(sum(a_i), 110)))
+
+----
+
+In this example, the `having` expression iterates the aggregated tuples from the `rollup` expression and emits all tuples where the field `sum(a_i)` is greater than 100 and less than 110.
+
 == hashJoin
 
 The `hashJoin` function wraps two streams, Left and Right, and for every tuple in Left which exists in Right will emit a tuple containing the fields of both tuples.
@@ -986,6 +1052,52 @@ intersect(
 )
 ----
 
+== leftOuterJoin
+
+The `leftOuterJoin` function wraps two streams, Left and Right, and emits tuples from Left.
+If there is a tuple in Right equal (as defined by `on`) then the values in that tuple will be included in the emitted tuple.
+An equal tuple in Right *need not* exist for the Left tuple to be emitted.
+This supports one-to-one, one-to-many, many-to-one, and many-to-many left outer join scenarios.
+The tuples are emitted in the order in which they appear in the Left stream.
+Both streams must be sorted by the fields being used to determine equality (using the `on` parameter).
+If both tuples contain a field of the same name then the value from the Right stream will be used in the emitted tuple.
+
+You can wrap the incoming streams with a `select` function to be specific about which field values are included in the emitted tuple.
+
+=== leftOuterJoin Parameters
+
+* `StreamExpression for StreamLeft`
+* `StreamExpression for StreamRight`
+* `on`: Fields to be used for checking equality of tuples between Left and Right.
+Can be of the format `on="fieldName"`, `on="fieldNameInLeft=fieldNameInRight"`, or `on="fieldName, otherFieldName=rightOtherFieldName"`.
+
+=== leftOuterJoin Syntax
+
+[source,text]
+----
+leftOuterJoin(
+  search(people, q="*:*", qt="/export", fl="personId,name", sort="personId asc"),
+  search(pets, q="type:cat", qt="/export", fl="personId,petName", sort="personId asc"),
+  on="personId"
+)
+
+leftOuterJoin(
+  search(people, q="*:*", qt="/export", fl="personId,name", sort="personId asc"),
+  search(pets, q="type:cat", qt="/export", fl="ownerId,petName", sort="ownerId asc"),
+  on="personId=ownerId"
+)
+
+leftOuterJoin(
+  search(people, q="*:*", qt="/export", fl="personId,name", sort="personId asc"),
+  select(
+    search(pets, q="type:cat", qt="/export", fl="ownerId,name", sort="ownerId asc"),
+    ownerId,
+    name as petName
+  ),
+  on="personId=ownerId"
+)
+----
+
 [#list_expression]
 == list
 

solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/Lang.java

@@ -272,6 +272,7 @@
 import org.apache.solr.client.solrj.io.stream.FacetStream;
 import org.apache.solr.client.solrj.io.stream.FeaturesSelectionStream;
 import org.apache.solr.client.solrj.io.stream.FetchStream;
+import org.apache.solr.client.solrj.io.stream.FullOuterJoinStream;
 import org.apache.solr.client.solrj.io.stream.GetStream;
 import org.apache.solr.client.solrj.io.stream.HashJoinStream;
 import org.apache.solr.client.solrj.io.stream.HashRollupStream;
@@ -355,6 +356,7 @@ public static void register(StreamFactory streamFactory) {
         .withFunctionName("stats", StatsStream.class)
         .withFunctionName("innerJoin", InnerJoinStream.class)
         .withFunctionName("leftOuterJoin", LeftOuterJoinStream.class)
+        .withFunctionName("fullOuterJoin", FullOuterJoinStream.class)
         .withFunctionName("hashJoin", HashJoinStream.class)
         .withFunctionName("outerHashJoin", OuterHashJoinStream.class)
         .withFunctionName("intersect", IntersectStream.class)

solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/FullOuterJoinStream.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.client.solrj.io.stream;
+
+import java.io.IOException;
+import java.util.LinkedList;
+import org.apache.solr.client.solrj.io.Tuple;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
+import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
+
+/**
+ * Joins leftStream with rightStream based on an Equalitor. Both streams must be sorted by the
+ * fields being joined on. Resulting stream is sorted by the equalitor.
+ *
+ * @since 9.10.0
+ */
+public class FullOuterJoinStream extends BiJoinStream {
+
+  @SuppressWarnings("JdkObsolete")
+  private final LinkedList<Tuple> joinedTuples = new LinkedList<>();
+
+  @SuppressWarnings("JdkObsolete")
+  private final LinkedList<Tuple> leftTupleGroup = new LinkedList<>();
+
+  @SuppressWarnings("JdkObsolete")
+  private final LinkedList<Tuple> rightTupleGroup = new LinkedList<>();
+
+  public FullOuterJoinStream(TupleStream leftStream, TupleStream rightStream, StreamEqualitor eq)
+      throws IOException {
+    super(leftStream, rightStream, eq);
+  }
+
+  public FullOuterJoinStream(StreamExpression expression, StreamFactory factory)
+      throws IOException {
+    super(expression, factory);
+  }
+
+  @Override
+  public Tuple read() throws IOException {
+    // if we've already figured out the next joined tuple then just return it
+    if (joinedTuples.size() > 0) {
+      return joinedTuples.removeFirst();
+    }
+
+    // keep going until we find something to return or both streams are empty
+    while (true) {
+
+      // load next set of equal tuples from leftStream into leftTupleGroup
+      if (0 == leftTupleGroup.size()) {
+        loadEqualTupleGroup(leftStream, leftTupleGroup, leftStreamComparator);
+      }
+
+      // same for right
+      if (0 == rightTupleGroup.size()) {
+        loadEqualTupleGroup(rightStream, rightTupleGroup, rightStreamComparator);
+      }
+
+      Boolean leftFinished = (0 == leftTupleGroup.size() || leftTupleGroup.get(0).EOF);
+      Boolean rightFinished = (0 == rightTupleGroup.size() || rightTupleGroup.get(0).EOF);
+
+      // If both streams are at EOF, we're done
+      if (leftFinished && rightFinished) {
+        return Tuple.EOF();
+      }
+
+      // If the left stream is at the EOF, we just return the next element from the right stream
+      if (leftFinished) {
+        return rightTupleGroup.removeFirst();
+      }
+
+      // If the right stream is at the EOF, we just return the next element from the left stream
+      if (rightFinished) {
+        return leftTupleGroup.removeFirst();
+      }
+
+      // At this point we know both left and right groups have at least 1 member
+      if (eq.test(leftTupleGroup.get(0), rightTupleGroup.get(0))) {
+        // The groups are equal. Join em together and build the joinedTuples
+        for (Tuple left : leftTupleGroup) {
+          for (Tuple right : rightTupleGroup) {
+            Tuple clone = left.clone();
+            clone.merge(right);
+            joinedTuples.add(clone);
+          }
+        }
+
+        // Cause each to advance next time we need to look
+        leftTupleGroup.clear();
+        rightTupleGroup.clear();
+
+        return joinedTuples.removeFirst();
+      } else {
+        int c = iterationComparator.compare(leftTupleGroup.get(0), rightTupleGroup.get(0));
+        if (c < 0) {
+          // If there's no match, we still advance the left stream while returning every element.
+          // Because it's an outer join we still return the left tuple if no match on right.
+          return leftTupleGroup.removeFirst();
+        } else {
+          // return right item as it didn't match left and this is a full outer join
+          return rightTupleGroup.removeFirst();
+        }
+      }
+    }
+  }
+
+  @Override
+  public StreamComparator getStreamSort() {
+    return iterationComparator;
+  }
+}