Added support for LEFT OUTER JOIN ... ON queries through an outerJoin()

variant that comes with a source and a comparison lambda.

api/src/org/jinq/orm/internal/QueryComposer.java

@@ -36,6 +36,7 @@ public <V extends Comparable<V>> QueryComposer<T> sortedBy(
    public <U> QueryComposer<Pair<T, U>> joinIterable(JinqStream.JoinToIterable<T,U> join);
    public <U> QueryComposer<Pair<T, U>> leftOuterJoin(JinqStream.Join<T,U> join);
    public <U> QueryComposer<Pair<T, U>> leftOuterJoinIterable(JinqStream.JoinToIterable<T,U> join);
+   public <U> QueryComposer<Pair<T, U>> leftOuterJoinWithSource(JinqStream.JoinWithSource<T,U> join, JinqStream.WhereForOn<T, U> on);
 //   public <U, V> QueryComposer<Pair<U, V>> group(JinqStream.Select<T, U> select, JinqStream.AggregateGroup<U, T, V> aggregate);
 
    // returns null if the aggregates cannot be calculated

api/src/org/jinq/orm/stream/JinqStream.java

@@ -273,6 +273,50 @@ public static interface JoinToIterable<U, V> extends Serializable
     */
    public <U> JinqStream<Pair<T, U>> leftOuterJoinList(JoinToIterable<T, U> join);
 
+   @FunctionalInterface
+   public static interface WhereForOn<U, V> extends Serializable
+   {
+      public boolean where(U obj1, V obj2);
+   }
+
+   /**
+    * Pairs up each entry of the stream with a stream of related elements. Uses
+    * a left outer join during the pairing up process, so even if an element is
+    * not joined with anything, a pair will still be created in the output
+    * stream consisting of the element paired with null. This version also passes 
+    * an {@link InQueryStreamSource} to the join function so that the function 
+    * can join elements with unrelated streams of entities from a database
+    * and an ON clause can be specified that will determine which elements from
+    * the two streams will be joined together. 
+    * 
+    * <pre>
+    * {@code JinqStream<Country> stream = ...;
+    * JinqStream<Pair<Country, Mountain>> result = 
+    *    stream.leftOuterJoin(
+    *            (c, source) -> source.stream(Mountain.class), 
+    *            (country, mountain) -> country.getName().equals(mountain.getCountry()));
+    * }
+    * </pre>
+    * 
+    * @see #leftOuterJoin(Join)
+    * @param join
+    *           function applied to the elements of the stream. When passed an
+    *           element from the stream, the function should return a stream of
+    *           values that should be paired up with that stream element. The
+    *           function must use a JPA association or navigational link as the
+    *           base for the stream returned. Both singular or plural
+    *           associations are allowed.
+    * @param on
+    *           this is a comparison function that returns true if the elements
+    *           from the two streams should be joined together. It is similar to
+    *           a standard where() clause except that the elements from the two
+    *           streams are passed in as separate parameters for convenience
+    *           (as opposed to being passed in as a pair)
+    * @return a new stream with the paired up elements
+    */
+   public <U> JinqStream<Pair<T, U>> leftOuterJoin(JoinWithSource<T, U> join, WhereForOn<T, U> on);
+
+
    @FunctionalInterface
    public static interface AggregateGroup<W, U, V> extends Serializable
    {

api/src/org/jinq/orm/stream/NonQueryJinqStream.java

@@ -12,11 +12,13 @@
 import java.util.Spliterator;
 import java.util.Spliterators;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
-
+import org.jinq.orm.stream.JinqStream.JoinWithSource;
+import org.jinq.orm.stream.JinqStream.WhereForOn;
 import org.jinq.tuples.Pair;
 import org.jinq.tuples.Tuple;
 import org.jinq.tuples.Tuple3;
@@ -183,6 +185,26 @@ public <U> JinqStream<Pair<T, U>> leftOuterJoinList(
       return wrap(streamBuilder.build());
    }
 
+   @Override
+   public <U> JinqStream<Pair<T, U>> leftOuterJoin(JoinWithSource<T, U> join, WhereForOn<T, U> on)
+   {
+      // TODO: This stream should be constructed on the fly
+      final Stream.Builder<Pair<T,U>> streamBuilder = Stream.builder();
+      forEach( left -> {
+         AtomicBoolean wasMatched = new AtomicBoolean();
+         join.join(left, inQueryStreamSource).forEach( right -> {
+            if (on.where(left, right))
+            {
+               wasMatched.set(true);
+               streamBuilder.accept(new Pair<>(left, right));
+            }
+         });
+         if (!wasMatched.get())
+            streamBuilder.accept(new Pair<>(left, null));
+         });
+      return wrap(streamBuilder.build());
+   }
+
    protected <U, W extends Tuple> JinqStream<W> groupToTuple(Select<T, U> select, AggregateGroup<U, T, ?>[] aggregates)
    {
       Map<U, List<T>> groups = collect(Collectors.groupingBy(in -> select.select(in)));

api/src/org/jinq/orm/stream/QueryJinqStream.java

@@ -127,7 +127,7 @@ public <U> JinqStream<Pair<T, U>> leftOuterJoin(Join<T,U> join)
       if (newComposer != null) return makeQueryStream(newComposer, inQueryStreamSource);
       return super.leftOuterJoin(join);
    }
-   
+
    @Override
    public <U> JinqStream<Pair<T, U>> leftOuterJoinList(JoinToIterable<T, U> join)
    {
@@ -136,6 +136,14 @@ public <U> JinqStream<Pair<T, U>> leftOuterJoinList(JoinToIterable<T, U> join)
       return super.leftOuterJoinList(join);
    }
 
+   @Override
+   public <U> JinqStream<Pair<T, U>> leftOuterJoin(JoinWithSource<T, U> join, WhereForOn<T, U> on)
+   {
+      QueryComposer<Pair<T, U>> newComposer = queryComposer.leftOuterJoinWithSource(join, on);
+      if (newComposer != null) return makeQueryStream(newComposer, inQueryStreamSource);
+      return super.leftOuterJoin(join, on);
+   }
+
    @Override
    protected <U, W extends Tuple> JinqStream<W> groupToTuple(Select<T, U> select, AggregateGroup<U, T, ?>[] aggregates)
    {

api/test/org/jinq/test/query/NonQueryJinqStreamTest.java

@@ -96,4 +96,22 @@ public void testFindOneException()
    {
       new NonQueryJinqStream<>( Stream.of(1, 2) ).findOne();
    }
+
+   @Test
+   public void testLeftOuterJoinOn()
+   {
+      Pair<Integer, Integer>[] vals = 
+            new NonQueryJinqStream<>( Stream.of(1, 2) )
+                  .leftOuterJoin(
+                        (val, source) -> new NonQueryJinqStream<>( Stream.of(1, 3) ),
+                        (a, b) -> a == b
+                  )
+                  .toList()
+                  .toArray(new Pair[0]);
+      assertArrayEquals(
+            new Pair[] {
+                  new Pair<>(1, 1),
+                  new Pair<>(2, null)
+            }, vals);
+   }
 }

jinq-jpa/main/org/jinq/jpa/JPAJinqStreamWrapper.java

@@ -248,6 +248,15 @@ public <U> JPAJinqStream<Pair<T, U>> leftOuterJoinList(
    {
       return wrap(wrapped.leftOuterJoinList(join));
    }
+
+   @Override
+   public <U> JinqStream<Pair<T, U>> leftOuterJoin(
+         org.jinq.orm.stream.JinqStream.JoinWithSource<T, U> join, 
+         org.jinq.orm.stream.JinqStream.WhereForOn<T, U> on)
+   {
+      return wrap(wrapped.leftOuterJoin(join, on));
+   }
+
 
    @Override
    public <U, V> JPAJinqStream<Pair<U, V>> group(

jinq-jpa/main/org/jinq/jpa/JPAQueryComposer.java

@@ -34,6 +34,7 @@
 import org.jinq.jpa.transform.LimitSkipTransform;
 import org.jinq.jpa.transform.MetamodelUtil;
 import org.jinq.jpa.transform.MultiAggregateTransform;
+import org.jinq.jpa.transform.OuterJoinOnTransform;
 import org.jinq.jpa.transform.OuterJoinTransform;
 import org.jinq.jpa.transform.QueryTransformException;
 import org.jinq.jpa.transform.SelectTransform;
@@ -43,6 +44,7 @@
 import org.jinq.orm.stream.JinqStream.AggregateGroup;
 import org.jinq.orm.stream.JinqStream.JoinToIterable;
 import org.jinq.orm.stream.JinqStream.Select;
+import org.jinq.orm.stream.JinqStream;
 import org.jinq.orm.stream.NextOnlyIterator;
 import org.jinq.tuples.Pair;
 import org.jinq.tuples.Tuple;
@@ -296,6 +298,43 @@ public <U> JPAQueryComposer<U> applyTransformWithLambda(JPQLOneLambdaQueryTransf
       return new JPAQueryComposer<>(this, (JPQLQuery<U>)cachedQuery.get(), lambdas, lambdaInfo);
    }
 
+   public <U> JPAQueryComposer<U> applyTransformWithTwoLambdas(OuterJoinOnTransform transform, Object lambda1, Object lambda2)
+   {
+      LambdaInfo lambdaInfo1 = lambdaAnalyzer.extractSurfaceInfo(lambda1, lambdas.size(), hints.dieOnError);
+      if (lambdaInfo1 == null) { translationFail(); return null; }
+      LambdaInfo lambdaInfo2 = lambdaAnalyzer.extractSurfaceInfo(lambda2, lambdas.size() + 1, hints.dieOnError);
+      if (lambdaInfo2 == null) { translationFail(); return null; }
+      Optional<JPQLQuery<?>> cachedQuery = hints.useCaching ?
+            cachedQueries.findInCache(query, transform.getTransformationTypeCachingTag(), new String[] {lambdaInfo1.getLambdaSourceString(), lambdaInfo2.getLambdaSourceString()}) : null;
+      if (cachedQuery == null)
+      {
+         cachedQuery = Optional.empty();
+         JPQLQuery<U> newQuery = null;
+         try {
+            LambdaAnalysis lambdaAnalysis1 = lambdaInfo1.fullyAnalyze(metamodel, hints.lambdaClassLoader, hints.isObjectEqualsSafe, hints.isAllEqualsSafe, hints.isCollectionContainsSafe, hints.dieOnError);
+            if (lambdaAnalysis1 == null) { translationFail(); return null; }
+            LambdaAnalysis lambdaAnalysis2 = lambdaInfo2.fullyAnalyze(metamodel, hints.lambdaClassLoader, hints.isObjectEqualsSafe, hints.isAllEqualsSafe, hints.isCollectionContainsSafe, hints.dieOnError);
+            if (lambdaAnalysis2 == null) { translationFail(); return null; }
+            getConfig().checkLambdaSideEffects(lambdaAnalysis1);
+            getConfig().checkLambdaSideEffects(lambdaAnalysis1);
+            newQuery = transform.apply(query, lambdaAnalysis1, lambdaAnalysis2, null);
+         }
+         catch (QueryTransformException e)
+         {
+            translationFail(e);
+         }
+         finally 
+         {
+            // Always cache the resulting query, even if it is an error
+            cachedQuery = Optional.ofNullable(newQuery);
+            if (hints.useCaching)
+               cachedQuery = cachedQueries.cacheQuery(query, transform.getTransformationTypeCachingTag(), new String[] {lambdaInfo1.getLambdaSourceString(), lambdaInfo2.getLambdaSourceString()}, cachedQuery);
+         }
+      }
+      if (!cachedQuery.isPresent()) { translationFail(); return null; }
+      return new JPAQueryComposer<>(this, (JPQLQuery<U>)cachedQuery.get(), lambdas, lambdaInfo1, lambdaInfo2);
+   }
+
    public <U> JPAQueryComposer<U> applyTransformWithLambdas(JPQLMultiLambdaQueryTransform transform, Object [] groupingLambdas)
    {
       LambdaInfo[] lambdaInfos = new LambdaInfo[groupingLambdas.length];
@@ -471,6 +510,13 @@ public <U> QueryComposer<Pair<T, U>> leftOuterJoinIterable(
    {
       return applyTransformWithLambda(new OuterJoinTransform(getConfig()).setIsExpectingStream(false), joinLambda);
    }
+
+   @Override
+   public <U> QueryComposer<Pair<T, U>> leftOuterJoinWithSource(JinqStream.JoinWithSource<T,U> join, JinqStream.WhereForOn<T, U> on)
+   {
+      return applyTransformWithTwoLambdas(new OuterJoinOnTransform(getConfig()).setIsExpectingStream(true), join, on);
+   }
+
 
    public <U> JPAQueryComposer<T> leftOuterJoinFetch(
          org.jinq.orm.stream.JinqStream.Join<T, U> joinLambda)

jinq-jpa/main/org/jinq/jpa/jpqlquery/From.java

@@ -2,7 +2,7 @@
 
 public abstract class From implements JPQLFragment
 {
-   abstract void generateFromString(QueryGenerationState queryState, boolean isFirst);
+   abstract void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst);
    public boolean isPrecededByComma() { return true; }
 
    protected void prepareQueryGeneration(Expression.QueryGenerationPreparationPhase preparePhase,
@@ -29,6 +29,23 @@ public static From forNavigationalLinks(Expression linksExpr)
       return from;
    }
 
+   public static FromLeftOuterJoinSettableOn forLeftOuterJoinOn(From from)
+   {
+      if (from instanceof FromEntity)
+      {
+         FromEntityLeftOuterJoinOn toReturn = new FromEntityLeftOuterJoinOn();
+         toReturn.entityName = ((FromEntity)from).entityName;
+         return toReturn;
+      }
+      else if (from instanceof FromNavigationalLinks)
+      {
+         FromNavigationalLinksLeftOuterJoinOn toReturn = new FromNavigationalLinksLeftOuterJoinOn();
+         toReturn.links = ((FromNavigationalLinks)from).links;
+         return toReturn;
+      }
+      throw new IllegalArgumentException("Creating a LEFT OUTER JOIN using unexpected parameters");
+   }
+
    public static FromNavigationalLinksLeftOuterJoin forNavigationalLinksLeftOuterJoin(FromNavigationalLinks link)
    {
       FromNavigationalLinksLeftOuterJoin from = new FromNavigationalLinksLeftOuterJoin();
@@ -55,9 +72,10 @@ public static class FromEntity extends From
       public String entityName;
 
       @Override
-      void generateFromString(QueryGenerationState queryState, boolean isFirst)
+      void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst)
       {
          queryState.queryString += entityName;
+         queryState.queryString += " " + alias;
       }
    }
 
@@ -69,47 +87,96 @@ public static abstract class FromNavigationalLinksGeneric extends From
    public static class FromNavigationalLinks extends FromNavigationalLinksGeneric
    {
       @Override
-      void generateFromString(QueryGenerationState queryState, boolean isFirst)
+      void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst)
       {
          if (!isFirst)
             queryState.queryString += " JOIN ";
          links.generateQuery(queryState, OperatorPrecedenceLevel.JPQL_UNRESTRICTED_OPERATOR_PRECEDENCE);
+         queryState.queryString += " " + alias;
       }
       @Override public boolean isPrecededByComma() { return false; }
    }
 
    public static class FromNavigationalLinksLeftOuterJoin extends FromNavigationalLinksGeneric
    {
       @Override
-      void generateFromString(QueryGenerationState queryState, boolean isFirst)
+      void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst)
       {
          if (!isFirst)
             queryState.queryString += " LEFT OUTER JOIN ";
          links.generateQuery(queryState, OperatorPrecedenceLevel.JPQL_UNRESTRICTED_OPERATOR_PRECEDENCE);
+         queryState.queryString += " " + alias;
       }
       @Override public boolean isPrecededByComma() { return false; }
    }
+
+   public static interface FromLeftOuterJoinSettableOn
+   {
+      void setOn(Expression onExpr);
+   }
 
+   public static class FromEntityLeftOuterJoinOn extends FromEntity implements FromLeftOuterJoinSettableOn
+   {
+      Expression onExpr;
+      @Override
+      void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst)
+      {
+         if (!isFirst)
+            queryState.queryString += " LEFT OUTER JOIN ";
+         queryState.queryString += entityName;
+         queryState.queryString += " " + alias;
+         if (onExpr != null)
+         {
+            queryState.queryString += " ON ";
+            onExpr.generateQuery(queryState, OperatorPrecedenceLevel.JPQL_UNRESTRICTED_OPERATOR_PRECEDENCE);
+         }
+      }
+      @Override public boolean isPrecededByComma() { return false; }
+      @Override public void setOn(Expression onExpr) { this.onExpr = onExpr; }
+   }
+
+   public static class FromNavigationalLinksLeftOuterJoinOn extends FromNavigationalLinksGeneric implements FromLeftOuterJoinSettableOn
+   {
+      Expression onExpr;
+      @Override
+      void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst)
+      {
+         if (!isFirst)
+            queryState.queryString += " LEFT OUTER JOIN ";
+         links.generateQuery(queryState, OperatorPrecedenceLevel.JPQL_UNRESTRICTED_OPERATOR_PRECEDENCE);
+         queryState.queryString += " " + alias;
+         if (onExpr != null)
+         {
+            queryState.queryString += " ON ";
+            onExpr.generateQuery(queryState, OperatorPrecedenceLevel.JPQL_UNRESTRICTED_OPERATOR_PRECEDENCE);
+         }
+      }
+      @Override public boolean isPrecededByComma() { return false; }
+      @Override public void setOn(Expression onExpr) { this.onExpr = onExpr; }
+   }
+
    public static class FromNavigationalLinksLeftOuterJoinFetch extends FromNavigationalLinksGeneric
    {
       @Override
-      void generateFromString(QueryGenerationState queryState, boolean isFirst)
+      void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst)
       {
          if (!isFirst)
             queryState.queryString += " LEFT OUTER JOIN FETCH ";
          links.generateQuery(queryState, OperatorPrecedenceLevel.JPQL_UNRESTRICTED_OPERATOR_PRECEDENCE);
+         queryState.queryString += " " + alias;
       }
       @Override public boolean isPrecededByComma() { return false; }
    }
 
    public static class FromNavigationalLinksJoinFetch extends FromNavigationalLinksGeneric
    {
       @Override
-      void generateFromString(QueryGenerationState queryState, boolean isFirst)
+      void generateFromString(QueryGenerationState queryState, String alias, boolean isFirst)
       {
          if (!isFirst)
             queryState.queryString += " JOIN FETCH ";
          links.generateQuery(queryState, OperatorPrecedenceLevel.JPQL_UNRESTRICTED_OPERATOR_PRECEDENCE);
+         queryState.queryString += " " + alias;
       }
       @Override public boolean isPrecededByComma() { return false; }
    }

jinq-jpa/main/org/jinq/jpa/jpqlquery/SelectFromWhere.java

@@ -98,9 +98,8 @@ protected void generateSelectFromWhere(QueryGenerationState queryState)
                if (from.isPrecededByComma())
                   queryState.queryString += ", ";
             }
-            from.generateFromString(queryState, isFirst);
+            from.generateFromString(queryState, queryState.getFromAlias(from), isFirst);
             isFirst = false;
-            queryState.queryString += " " + queryState.getFromAlias(from);
          }
       }
       if (where != null)