Index: lucene/CHANGES.txt
===================================================================
--- lucene/CHANGES.txt	(revision 1718443)
+++ lucene/CHANGES.txt	(working copy)
@@ -6,6 +6,12 @@
 
 ======================= Lucene 5.5.0 =======================
 
+New Features
+
+* LUCENE-5868: JoinUtil.createJoinQuery(..,NumericType,..) query-time join 
+  for LONG and INT fields with NUMERIC and SORTED_NUMERIC doc values.
+  (Alexey Zelin via Mikhail Khludnev) 
+
 API Changes
 
 * LUCENE-6900: Grouping sortWithinGroup variables used to allow null to mean

Property changes on: lucene/CHANGES.txt
___________________________________________________________________
Modified: svn:mergeinfo
   Merged /lucene/dev/trunk/lucene/CHANGES.txt:r1718443
Index: lucene/join/src/java/org/apache/lucene/search/join/DocValuesTermsCollector.java
===================================================================
--- lucene/join/src/java/org/apache/lucene/search/join/DocValuesTermsCollector.java	(revision 1718443)
+++ lucene/join/src/java/org/apache/lucene/search/join/DocValuesTermsCollector.java	(working copy)
@@ -1,7 +1,6 @@
 package org.apache.lucene.search.join;
 
 import java.io.IOException;
-import java.util.function.LongConsumer;
 
 import org.apache.lucene.document.FieldType.NumericType;
 import org.apache.lucene.index.BinaryDocValues;
@@ -35,11 +34,14 @@
 
 abstract class DocValuesTermsCollector<DV> extends SimpleCollector {
   
-  @FunctionalInterface
   static interface Function<R> {
       R apply(LeafReader t) throws IOException  ;
   }
   
+  static interface LongConsumer {
+    void accept(long value);
+  }
+  
   protected DV docValues;
   private final Function<DV> docValuesCall;
   
@@ -52,37 +54,62 @@
     docValues = docValuesCall.apply(context.reader());
   }
   
-  static Function<BinaryDocValues> binaryDocValues(String field) {
-      return (ctx) -> DocValues.getBinary(ctx, field);
-  }
-  static Function<SortedSetDocValues> sortedSetDocValues(String field) {
-    return (ctx) -> DocValues.getSortedSet(ctx, field);
-  }
-  
-  static Function<BinaryDocValues> numericAsBinaryDocValues(String field, NumericType numTyp) {
-    return (ctx) -> {
-      final NumericDocValues numeric = DocValues.getNumeric(ctx, field);
-      final BytesRefBuilder bytes = new BytesRefBuilder();
-      
-      final LongConsumer coder = coder(bytes, numTyp, field);
-      
-      return new BinaryDocValues() {
+  static Function<BinaryDocValues> binaryDocValues(final String field) {
+      return new Function<BinaryDocValues>() 
+      {
         @Override
-        public BytesRef get(int docID) {
-          final long lVal = numeric.get(docID);
-          coder.accept(lVal);
-          return bytes.get();
+        public BinaryDocValues apply(LeafReader ctx) throws IOException {
+          return DocValues.getBinary(ctx, field);
         }
       };
+  }
+  static Function<SortedSetDocValues> sortedSetDocValues(final String field) {
+    return new Function<SortedSetDocValues>() 
+    {
+      @Override
+      public SortedSetDocValues apply(LeafReader ctx) throws IOException {
+        return DocValues.getSortedSet(ctx, field);
+      }
     };
   }
   
-  static LongConsumer coder(BytesRefBuilder bytes, NumericType type, String fieldName){
+  static Function<BinaryDocValues> numericAsBinaryDocValues(final String field, final NumericType numTyp) {
+    return new Function<BinaryDocValues>() {
+      @Override
+      public BinaryDocValues apply(LeafReader ctx) throws IOException {
+        final NumericDocValues numeric = DocValues.getNumeric(ctx, field);
+        final BytesRefBuilder bytes = new BytesRefBuilder();
+        
+        final LongConsumer coder = coder(bytes, numTyp, field);
+        
+        return new BinaryDocValues() {
+          @Override
+          public BytesRef get(int docID) {
+            final long lVal = numeric.get(docID);
+            coder.accept(lVal);
+            return bytes.get();
+          }
+        };
+      }
+    };
+  }
+  
+  static LongConsumer coder(final BytesRefBuilder bytes, NumericType type, String fieldName){
     switch(type){
       case INT: 
-        return (l) -> NumericUtils.intToPrefixCoded((int)l, 0, bytes);
+        return new LongConsumer() {
+          @Override
+          public void accept(long value) {
+            NumericUtils.intToPrefixCoded((int)value, 0, bytes);
+          }
+        };
       case LONG: 
-        return (l) -> NumericUtils.longToPrefixCoded(l, 0, bytes);
+        return new LongConsumer() {
+          @Override
+          public void accept(long value) {
+            NumericUtils.longToPrefixCoded((int)value, 0, bytes);
+          }
+        };
       default:
         throw new IllegalArgumentException("Unsupported "+type+
             ". Only "+NumericType.INT+" and "+NumericType.LONG+" are supported."
@@ -91,46 +118,49 @@
   }
   
   /** this adapter is quite weird. ords are per doc index, don't use ords across different docs*/
-  static Function<SortedSetDocValues> sortedNumericAsSortedSetDocValues(String field, NumericType numTyp) {
-    return (ctx) -> {
-      final SortedNumericDocValues numerics = DocValues.getSortedNumeric(ctx, field);
-      final BytesRefBuilder bytes = new BytesRefBuilder();
-      
-      final LongConsumer coder = coder(bytes, numTyp, field);
-      
-      return new SortedSetDocValues() {
-
-        private int index = Integer.MIN_VALUE;
-
-        @Override
-        public long nextOrd() {
-          return index < numerics.count()-1 ? ++index : NO_MORE_ORDS;
-        }
-
-        @Override
-        public void setDocument(int docID) {
-          numerics.setDocument(docID);
-          index=-1;
-        }
-
-        @Override
-        public BytesRef lookupOrd(long ord) {
-          assert ord>=0 && ord<numerics.count();
-          final long value = numerics.valueAt((int)ord);
-          coder.accept(value);
-          return bytes.get();
-        }
-
-        @Override
-        public long getValueCount() {
-          throw new UnsupportedOperationException("it's just number encoding wrapper");
-        }
+  static Function<SortedSetDocValues> sortedNumericAsSortedSetDocValues(final String field, final NumericType numTyp) {
+    return new Function<SortedSetDocValues>() {
+      @Override
+      public SortedSetDocValues apply(LeafReader ctx) throws IOException {
+        final SortedNumericDocValues numerics = DocValues.getSortedNumeric(ctx, field);
+        final BytesRefBuilder bytes = new BytesRefBuilder();
         
-        @Override
-        public long lookupTerm(BytesRef key) {
-          throw new UnsupportedOperationException("it's just number encoding wrapper");
-        }
-      };
+        final LongConsumer coder = coder(bytes, numTyp, field);
+        
+        return new SortedSetDocValues() {
+          
+          private int index = Integer.MIN_VALUE;
+          
+          @Override
+          public long nextOrd() {
+            return index < numerics.count() - 1 ? ++index : NO_MORE_ORDS;
+          }
+          
+          @Override
+          public void setDocument(int docID) {
+            numerics.setDocument(docID);
+            index = -1;
+          }
+          
+          @Override
+          public BytesRef lookupOrd(long ord) {
+            assert ord >= 0 && ord < numerics.count();
+            final long value = numerics.valueAt((int) ord);
+            coder.accept(value);
+            return bytes.get();
+          }
+          
+          @Override
+          public long getValueCount() {
+            throw new UnsupportedOperationException("it's just number encoding wrapper");
+          }
+          
+          @Override
+          public long lookupTerm(BytesRef key) {
+            throw new UnsupportedOperationException("it's just number encoding wrapper");
+          }
+        };
+      }
     };
   }
 }
Index: lucene/join/src/java/org/apache/lucene/search/join/GenericTermsCollector.java
===================================================================
--- lucene/join/src/java/org/apache/lucene/search/join/GenericTermsCollector.java	(revision 1718443)
+++ lucene/join/src/java/org/apache/lucene/search/join/GenericTermsCollector.java	(working copy)
@@ -1,17 +1,6 @@
 package org.apache.lucene.search.join;
 
-import java.io.IOException;
-import java.io.PrintStream;
-
-import org.apache.lucene.index.BinaryDocValues;
-import org.apache.lucene.index.LeafReaderContext;
-import org.apache.lucene.index.SortedSetDocValues;
 import org.apache.lucene.search.Collector;
-import org.apache.lucene.search.LeafCollector;
-import org.apache.lucene.search.join.DocValuesTermsCollector.Function;
-import org.apache.lucene.search.join.TermsWithScoreCollector.MV;
-import org.apache.lucene.search.join.TermsWithScoreCollector.SV;
-import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefHash;
 
 /*
@@ -37,87 +26,4 @@
   
   float[] getScoresPerTerm();
   
-  static GenericTermsCollector createCollectorMV(Function<SortedSetDocValues> mvFunction,
-      ScoreMode mode) {
-    
-    switch (mode) {
-      case None:
-        return wrap(new TermsCollector.MV(mvFunction));
-      case Avg:
-          return new MV.Avg(mvFunction);
-      default:
-          return new MV(mvFunction, mode);
-    }
-  }
-
-  static Function<SortedSetDocValues> verbose(PrintStream out, Function<SortedSetDocValues> mvFunction){
-    return (ctx) -> {
-      final SortedSetDocValues target = mvFunction.apply(ctx);
-      return new SortedSetDocValues() {
-        
-        @Override
-        public void setDocument(int docID) {
-          target.setDocument(docID);
-          out.println("\ndoc# "+docID);
-        }
-        
-        @Override
-        public long nextOrd() {
-          return target.nextOrd();
-        }
-        
-        @Override
-        public BytesRef lookupOrd(long ord) {
-          final BytesRef val = target.lookupOrd(ord);
-          out.println(val.toString()+", ");
-          return val;
-        }
-        
-        @Override
-        public long getValueCount() {
-          return target.getValueCount();
-        }
-      };
-      
-    };
-  }
-
-  static GenericTermsCollector createCollectorSV(Function<BinaryDocValues> svFunction,
-      ScoreMode mode) {
-    
-    switch (mode) {
-      case None:
-        return wrap(new TermsCollector.SV(svFunction));
-      case Avg:
-        return new SV.Avg(svFunction);
-      default:
-        return new SV(svFunction, mode);  
-    }
-  }
-  
-  static GenericTermsCollector wrap(final TermsCollector<?> collector) {
-    return new GenericTermsCollector() {
-
-      
-      @Override
-      public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
-        return collector.getLeafCollector(context);
-      }
-
-      @Override
-      public boolean needsScores() {
-        return collector.needsScores();
-      }
-
-      @Override
-      public BytesRefHash getCollectedTerms() {
-        return collector.getCollectorTerms();
-      }
-
-      @Override
-      public float[] getScoresPerTerm() {
-        throw new UnsupportedOperationException("scores are not available for "+collector);
-      }
-    };
-  }
 }
Index: lucene/join/src/java/org/apache/lucene/search/join/GenericTermsCollectorFactory.java
===================================================================
--- lucene/join/src/java/org/apache/lucene/search/join/GenericTermsCollectorFactory.java	(revision 0)
+++ lucene/join/src/java/org/apache/lucene/search/join/GenericTermsCollectorFactory.java	(working copy)
@@ -0,0 +1,86 @@
+package org.apache.lucene.search.join;
+
+import java.io.IOException;
+
+import org.apache.lucene.index.BinaryDocValues;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.SortedSetDocValues;
+import org.apache.lucene.search.LeafCollector;
+import org.apache.lucene.search.join.DocValuesTermsCollector.Function;
+import org.apache.lucene.search.join.TermsWithScoreCollector.MV;
+import org.apache.lucene.search.join.TermsWithScoreCollector.SV;
+import org.apache.lucene.util.BytesRefHash;
+
+/*
+ * 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.
+ */
+
+final class GenericTermsCollectorFactory {
+  
+  private GenericTermsCollectorFactory() {}
+ 
+  static GenericTermsCollector createCollectorMV(Function<SortedSetDocValues> mvFunction,
+      ScoreMode mode) {
+    
+    switch (mode) {
+      case None:
+        return wrap(new TermsCollector.MV(mvFunction));
+      case Avg:
+          return new MV.Avg(mvFunction);
+      default:
+          return new MV(mvFunction, mode);
+    }
+  }
+
+  static GenericTermsCollector createCollectorSV(Function<BinaryDocValues> svFunction,
+      ScoreMode mode) {
+    
+    switch (mode) {
+      case None:
+        return wrap(new TermsCollector.SV(svFunction));
+      case Avg:
+        return new SV.Avg(svFunction);
+      default:
+        return new SV(svFunction, mode);  
+    }
+  }
+  
+  static GenericTermsCollector wrap(final TermsCollector<?> collector) {
+    return new GenericTermsCollector() {
+
+      
+      @Override
+      public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
+        return collector.getLeafCollector(context);
+      }
+
+      @Override
+      public boolean needsScores() {
+        return collector.needsScores();
+      }
+
+      @Override
+      public BytesRefHash getCollectedTerms() {
+        return collector.getCollectorTerms();
+      }
+
+      @Override
+      public float[] getScoresPerTerm() {
+        throw new UnsupportedOperationException("scores are not available for "+collector);
+      }
+    };
+  }
+}

Property changes on: lucene/join/src/java/org/apache/lucene/search/join/GenericTermsCollectorFactory.java
___________________________________________________________________
Added: svn:eol-style
## -0,0 +1 ##
+native
\ No newline at end of property
Index: lucene/join/src/java/org/apache/lucene/search/join/JoinUtil.java
===================================================================
--- lucene/join/src/java/org/apache/lucene/search/join/JoinUtil.java	(revision 1718443)
+++ lucene/join/src/java/org/apache/lucene/search/join/JoinUtil.java	(working copy)
@@ -1,5 +1,14 @@
 package org.apache.lucene.search.join;
 
+import java.io.IOException;
+import java.util.Locale;
+
+import org.apache.lucene.document.FieldType.NumericType;
+import org.apache.lucene.document.IntField;
+import org.apache.lucene.document.LongField;
+import org.apache.lucene.index.BinaryDocValues;
+import org.apache.lucene.index.DocValuesType;
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
@@ -24,9 +33,11 @@
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.MultiDocValues;
 import org.apache.lucene.index.SortedDocValues;
+import org.apache.lucene.index.SortedSetDocValues;
 import org.apache.lucene.search.IndexSearcher;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
+import org.apache.lucene.search.join.DocValuesTermsCollector.Function;
 
 /**
  * Utility for query time joining.
@@ -67,28 +78,87 @@
    * @throws IOException If I/O related errors occur
    */
   public static Query createJoinQuery(String fromField,
-                                      boolean multipleValuesPerDocument,
-                                      String toField,
-                                      Query fromQuery,
-                                      IndexSearcher fromSearcher,
-                                      ScoreMode scoreMode) throws IOException {
+      boolean multipleValuesPerDocument,
+      String toField,
+      Query fromQuery,
+      IndexSearcher fromSearcher,
+      ScoreMode scoreMode) throws IOException {
+    
+    final GenericTermsCollector termsWithScoreCollector;
+     
+    if (multipleValuesPerDocument) {
+      Function<SortedSetDocValues> mvFunction = DocValuesTermsCollector.sortedSetDocValues(fromField);
+      termsWithScoreCollector = GenericTermsCollectorFactory.createCollectorMV(mvFunction, scoreMode);
+    } else {
+      Function<BinaryDocValues> svFunction = DocValuesTermsCollector.binaryDocValues(fromField);
+      termsWithScoreCollector =  GenericTermsCollectorFactory.createCollectorSV(svFunction, scoreMode);
+    }
+    
+    return createJoinQuery(multipleValuesPerDocument, toField, fromQuery, fromSearcher, scoreMode,
+        termsWithScoreCollector);
+    
+  }
+  
+  /**
+   * Method for query time joining for numeric fields. It supports multi- and single- values longs and ints. 
+   * All considerations from {@link JoinUtil#createJoinQuery(String, boolean, String, Query, IndexSearcher, ScoreMode)} are applicable here too,
+   * though memory consumption might be higher.
+   * <p>
+   *
+   * @param fromField                 The from field to join from
+   * @param multipleValuesPerDocument Whether the from field has multiple terms per document
+   *                                  when true fromField might be {@link DocValuesType#SORTED_NUMERIC},
+   *                                  otherwise fromField should be {@link DocValuesType#NUMERIC}
+   * @param toField                   The to field to join to, should be {@link IntField} or {@link LongField}
+   * @param numericType               either {@link NumericType#INT} or {@link NumericType#LONG}, it should correspond to fromField and toField types
+   * @param fromQuery                 The query to match documents on the from side
+   * @param fromSearcher              The searcher that executed the specified fromQuery
+   * @param scoreMode                 Instructs how scores from the fromQuery are mapped to the returned query
+   * @return a {@link Query} instance that can be used to join documents based on the
+   *         terms in the from and to field
+   * @throws IOException If I/O related errors occur
+   */
+  
+  public static Query createJoinQuery(String fromField,
+      boolean multipleValuesPerDocument,
+      String toField, NumericType numericType,
+      Query fromQuery,
+      IndexSearcher fromSearcher,
+      ScoreMode scoreMode) throws IOException {
+    
+    final GenericTermsCollector termsCollector;
+     
+    if (multipleValuesPerDocument) {
+      Function<SortedSetDocValues> mvFunction = DocValuesTermsCollector.sortedNumericAsSortedSetDocValues(fromField,numericType);
+      termsCollector = GenericTermsCollectorFactory.createCollectorMV(mvFunction, scoreMode);
+    } else {
+      Function<BinaryDocValues> svFunction = DocValuesTermsCollector.numericAsBinaryDocValues(fromField,numericType);
+      termsCollector =  GenericTermsCollectorFactory.createCollectorSV(svFunction, scoreMode);
+    }
+    
+    return createJoinQuery(multipleValuesPerDocument, toField, fromQuery, fromSearcher, scoreMode,
+        termsCollector);
+    
+  }
+  
+  private static Query createJoinQuery(boolean multipleValuesPerDocument, String toField, Query fromQuery,
+      IndexSearcher fromSearcher, ScoreMode scoreMode, final GenericTermsCollector collector)
+          throws IOException {
+    
+    fromSearcher.search(fromQuery, collector);
+    
     switch (scoreMode) {
       case None:
-        TermsCollector termsCollector = TermsCollector.create(fromField, multipleValuesPerDocument);
-        fromSearcher.search(fromQuery, termsCollector);
-        return new TermsQuery(toField, fromQuery, termsCollector.getCollectorTerms());
+        return new TermsQuery(toField, fromQuery, collector.getCollectedTerms());
       case Total:
       case Max:
       case Min:
       case Avg:
-        TermsWithScoreCollector termsWithScoreCollector =
-            TermsWithScoreCollector.create(fromField, multipleValuesPerDocument, scoreMode);
-        fromSearcher.search(fromQuery, termsWithScoreCollector);
         return new TermsIncludingScoreQuery(
             toField,
             multipleValuesPerDocument,
-            termsWithScoreCollector.getCollectedTerms(),
-            termsWithScoreCollector.getScoresPerTerm(),
+            collector.getCollectedTerms(),
+            collector.getScoresPerTerm(),
             fromQuery
         );
       default:
@@ -96,6 +166,7 @@
     }
   }
 
+
   /**
    * Delegates to {@link #createJoinQuery(String, Query, Query, IndexSearcher, ScoreMode, MultiDocValues.OrdinalMap, int, int)},
    * but disables the min and max filtering.
Index: lucene/join/src/java/org/apache/lucene/search/join/TermsCollector.java
===================================================================
--- lucene/join/src/java/org/apache/lucene/search/join/TermsCollector.java	(revision 1718443)
+++ lucene/join/src/java/org/apache/lucene/search/join/TermsCollector.java	(working copy)
@@ -19,11 +19,8 @@
 
 import java.io.IOException;
 
-import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.BinaryDocValues;
-import org.apache.lucene.index.DocValues;
 import org.apache.lucene.index.SortedSetDocValues;
-import org.apache.lucene.search.SimpleCollector;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefHash;
 
@@ -32,19 +29,19 @@
  *
  * @lucene.experimental
  */
-abstract class TermsCollector extends SimpleCollector {
+abstract class TermsCollector<DV> extends DocValuesTermsCollector<DV> {
 
-  final String field;
+  TermsCollector(Function<DV> docValuesCall) {
+    super(docValuesCall);
+  }
+
   final BytesRefHash collectorTerms = new BytesRefHash();
 
-  TermsCollector(String field) {
-    this.field = field;
-  }
-
   public BytesRefHash getCollectorTerms() {
     return collectorTerms;
   }
 
+  
   /**
    * Chooses the right {@link TermsCollector} implementation.
    *
@@ -52,55 +49,42 @@
    * @param multipleValuesPerDocument Whether the field to collect terms for has multiple values per document.
    * @return a {@link TermsCollector} instance
    */
-  static TermsCollector create(String field, boolean multipleValuesPerDocument) {
-    return multipleValuesPerDocument ? new MV(field) : new SV(field);
+  static TermsCollector<?> create(String field, boolean multipleValuesPerDocument) {
+    return multipleValuesPerDocument 
+        ? new MV(sortedSetDocValues(field))
+        : new SV(binaryDocValues(field));
   }
-
+  
   // impl that works with multiple values per document
-  static class MV extends TermsCollector {
-    final BytesRef scratch = new BytesRef();
-    private SortedSetDocValues docTermOrds;
-
-    MV(String field) {
-      super(field);
+  static class MV extends TermsCollector<SortedSetDocValues> {
+    
+    MV(Function<SortedSetDocValues> docValuesCall) {
+      super(docValuesCall);
     }
 
     @Override
     public void collect(int doc) throws IOException {
-      docTermOrds.setDocument(doc);
       long ord;
-      while ((ord = docTermOrds.nextOrd()) != SortedSetDocValues.NO_MORE_ORDS) {
-        final BytesRef term = docTermOrds.lookupOrd(ord);
+      docValues.setDocument(doc);
+      while ((ord = docValues.nextOrd()) != SortedSetDocValues.NO_MORE_ORDS) {
+        final BytesRef term = docValues.lookupOrd(ord);
         collectorTerms.add(term);
       }
     }
-
-    @Override
-    protected void doSetNextReader(LeafReaderContext context) throws IOException {
-      docTermOrds = DocValues.getSortedSet(context.reader(), field);
-    }
   }
 
   // impl that works with single value per document
-  static class SV extends TermsCollector {
+  static class SV extends TermsCollector<BinaryDocValues> {
 
-    final BytesRef spare = new BytesRef();
-    private BinaryDocValues fromDocTerms;
-
-    SV(String field) {
-      super(field);
+    SV(Function<BinaryDocValues> docValuesCall) {
+      super(docValuesCall);
     }
 
     @Override
     public void collect(int doc) throws IOException {
-      final BytesRef term = fromDocTerms.get(doc);
+      final BytesRef term = docValues.get(doc);
       collectorTerms.add(term);
     }
-
-    @Override
-    protected void doSetNextReader(LeafReaderContext context) throws IOException {
-      fromDocTerms = DocValues.getBinary(context.reader(), field);
-    }
   }
 
   @Override
Index: lucene/join/src/java/org/apache/lucene/search/join/TermsIncludingScoreQuery.java
===================================================================
--- lucene/join/src/java/org/apache/lucene/search/join/TermsIncludingScoreQuery.java	(revision 1718443)
+++ lucene/join/src/java/org/apache/lucene/search/join/TermsIncludingScoreQuery.java	(working copy)
@@ -18,6 +18,7 @@
  */
 
 import java.io.IOException;
+import java.io.PrintStream;
 import java.util.Locale;
 import java.util.Set;
 
@@ -37,6 +38,7 @@
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefHash;
 import org.apache.lucene.util.FixedBitSet;
+import org.apache.lucene.util.NumericUtils;
 
 class TermsIncludingScoreQuery extends Query {
 
@@ -271,5 +273,23 @@
       }
     }
   }
-
+  
+  void dump(PrintStream out){
+    out.println(field+":");
+    final BytesRef ref = new BytesRef();
+    for (int i = 0; i < terms.size(); i++) {
+      terms.get(ords[i], ref);
+      out.print(ref+" "+ref.utf8ToString()+" ");
+      try {
+        out.print(Long.toHexString(NumericUtils.prefixCodedToLong(ref))+"L");
+      } catch (Exception e) {
+        try {
+          out.print(Integer.toHexString(NumericUtils.prefixCodedToInt(ref))+"i");
+        } catch (Exception ee) {
+        }
+      }
+      out.println(" score="+scores[ords[i]]);
+      out.println("");
+    }
+  }
 }
Index: lucene/join/src/java/org/apache/lucene/search/join/TermsWithScoreCollector.java
===================================================================
--- lucene/join/src/java/org/apache/lucene/search/join/TermsWithScoreCollector.java	(revision 1718443)
+++ lucene/join/src/java/org/apache/lucene/search/join/TermsWithScoreCollector.java	(working copy)
@@ -1,5 +1,8 @@
 package org.apache.lucene.search.join;
 
+import java.io.IOException;
+import java.util.Arrays;
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
@@ -18,22 +21,16 @@
  */
 
 import org.apache.lucene.index.BinaryDocValues;
-import org.apache.lucene.index.DocValues;
-import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.SortedSetDocValues;
 import org.apache.lucene.search.Scorer;
-import org.apache.lucene.search.SimpleCollector;
 import org.apache.lucene.util.ArrayUtil;
 import org.apache.lucene.util.BytesRefHash;
 
-import java.io.IOException;
-import java.util.Arrays;
+abstract class TermsWithScoreCollector<DV> extends DocValuesTermsCollector<DV> 
+                                    implements GenericTermsCollector {
 
-abstract class TermsWithScoreCollector extends SimpleCollector {
-
   private final static int INITIAL_ARRAY_SIZE = 0;
 
-  final String field;
   final BytesRefHash collectedTerms = new BytesRefHash();
   final ScoreMode scoreMode;
 
@@ -40,8 +37,8 @@
   Scorer scorer;
   float[] scoreSums = new float[INITIAL_ARRAY_SIZE];
 
-  TermsWithScoreCollector(String field, ScoreMode scoreMode) {
-    this.field = field;
+  TermsWithScoreCollector(Function<DV> docValuesCall, ScoreMode scoreMode) {
+    super(docValuesCall);
     this.scoreMode = scoreMode;
     if (scoreMode == ScoreMode.Min) {
       Arrays.fill(scoreSums, Float.POSITIVE_INFINITY);
@@ -50,10 +47,12 @@
     }
   }
 
+  @Override
   public BytesRefHash getCollectedTerms() {
     return collectedTerms;
   }
-
+ 
+  @Override
   public float[] getScoresPerTerm() {
     return scoreSums;
   }
@@ -70,36 +69,34 @@
    * @param multipleValuesPerDocument Whether the field to collect terms for has multiple values per document.
    * @return a {@link TermsWithScoreCollector} instance
    */
-  static TermsWithScoreCollector create(String field, boolean multipleValuesPerDocument, ScoreMode scoreMode) {
+  static TermsWithScoreCollector<?> create(String field, boolean multipleValuesPerDocument, ScoreMode scoreMode) {
     if (multipleValuesPerDocument) {
       switch (scoreMode) {
         case Avg:
-          return new MV.Avg(field);
+          return new MV.Avg(sortedSetDocValues(field));
         default:
-          return new MV(field, scoreMode);
+          return new MV(sortedSetDocValues(field), scoreMode);
       }
     } else {
       switch (scoreMode) {
         case Avg:
-          return new SV.Avg(field);
+          return new SV.Avg(binaryDocValues(field));
         default:
-          return new SV(field, scoreMode);
+          return new SV(binaryDocValues(field), scoreMode);
       }
     }
   }
-
+ 
   // impl that works with single value per document
-  static class SV extends TermsWithScoreCollector {
+  static class SV extends TermsWithScoreCollector<BinaryDocValues> {
 
-    BinaryDocValues fromDocTerms;
-
-    SV(String field, ScoreMode scoreMode) {
-      super(field, scoreMode);
+    SV(Function<BinaryDocValues> docValuesCall, ScoreMode scoreMode) {
+      super(docValuesCall, scoreMode);
     }
 
     @Override
     public void collect(int doc) throws IOException {
-      int ord = collectedTerms.add(fromDocTerms.get(doc));
+      int ord = collectedTerms.add(docValues.get(doc));
       if (ord < 0) {
         ord = -ord - 1;
       } else {
@@ -133,26 +130,23 @@
               scoreSums[ord] = current;
             }
             break;
+          default:
+            throw new AssertionError("unexpected: " + scoreMode);
         }
       }
     }
 
-    @Override
-    protected void doSetNextReader(LeafReaderContext context) throws IOException {
-      fromDocTerms = DocValues.getBinary(context.reader(), field);
-    }
-
     static class Avg extends SV {
 
       int[] scoreCounts = new int[INITIAL_ARRAY_SIZE];
 
-      Avg(String field) {
-        super(field, ScoreMode.Avg);
+      Avg(Function<BinaryDocValues> docValuesCall) {
+        super(docValuesCall, ScoreMode.Avg);
       }
 
       @Override
       public void collect(int doc) throws IOException {
-        int ord = collectedTerms.add(fromDocTerms.get(doc));
+        int ord = collectedTerms.add(docValues.get(doc));
         if (ord < 0) {
           ord = -ord - 1;
         } else {
@@ -187,20 +181,18 @@
   }
 
   // impl that works with multiple values per document
-  static class MV extends TermsWithScoreCollector {
+  static class MV extends TermsWithScoreCollector<SortedSetDocValues> {
 
-    SortedSetDocValues fromDocTermOrds;
-
-    MV(String field, ScoreMode scoreMode) {
-      super(field, scoreMode);
+    MV(Function<SortedSetDocValues> docValuesCall, ScoreMode scoreMode) {
+      super(docValuesCall, scoreMode);
     }
 
     @Override
     public void collect(int doc) throws IOException {
-      fromDocTermOrds.setDocument(doc);
+      docValues.setDocument(doc);
       long ord;
-      while ((ord = fromDocTermOrds.nextOrd()) != SortedSetDocValues.NO_MORE_ORDS) {
-        int termID = collectedTerms.add(fromDocTermOrds.lookupOrd(ord));
+      while ((ord = docValues.nextOrd()) != SortedSetDocValues.NO_MORE_ORDS) {
+        int termID = collectedTerms.add(docValues.lookupOrd(ord));
         if (termID < 0) {
           termID = -termID - 1;
         } else {
@@ -225,29 +217,26 @@
           case Max:
             scoreSums[termID] = Math.max(scoreSums[termID], scorer.score());
             break;
+          default:
+            throw new AssertionError("unexpected: " + scoreMode);
         }
       }
     }
 
-    @Override
-    protected void doSetNextReader(LeafReaderContext context) throws IOException {
-      fromDocTermOrds = DocValues.getSortedSet(context.reader(), field);
-    }
-
     static class Avg extends MV {
 
       int[] scoreCounts = new int[INITIAL_ARRAY_SIZE];
 
-      Avg(String field) {
-        super(field, ScoreMode.Avg);
+      Avg(Function<SortedSetDocValues> docValuesCall) {
+        super(docValuesCall, ScoreMode.Avg);
       }
 
       @Override
       public void collect(int doc) throws IOException {
-        fromDocTermOrds.setDocument(doc);
+        docValues.setDocument(doc);
         long ord;
-        while ((ord = fromDocTermOrds.nextOrd()) != SortedSetDocValues.NO_MORE_ORDS) {
-          int termID = collectedTerms.add(fromDocTermOrds.lookupOrd(ord));
+        while ((ord = docValues.nextOrd()) != SortedSetDocValues.NO_MORE_ORDS) {
+          int termID = collectedTerms.add(docValues.lookupOrd(ord));
           if (termID < 0) {
             termID = -termID - 1;
           } else {
Index: lucene/join/src/test/org/apache/lucene/search/join/TestJoinUtil.java
===================================================================
--- lucene/join/src/test/org/apache/lucene/search/join/TestJoinUtil.java	(revision 1718443)
+++ lucene/join/src/test/org/apache/lucene/search/join/TestJoinUtil.java	(working copy)
@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -26,6 +27,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Random;
 import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeSet;
@@ -37,8 +39,12 @@
 import org.apache.lucene.analysis.MockTokenizer;
 import org.apache.lucene.document.Document;
 import org.apache.lucene.document.Field;
+import org.apache.lucene.document.FieldType.NumericType;
+import org.apache.lucene.document.IntField;
+import org.apache.lucene.document.LongField;
 import org.apache.lucene.document.NumericDocValuesField;
 import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.document.SortedNumericDocValuesField;
 import org.apache.lucene.document.SortedSetDocValuesField;
 import org.apache.lucene.document.StringField;
 import org.apache.lucene.document.TextField;
@@ -850,10 +856,18 @@
         }
 
         final Query joinQuery;
-        if (from) {
-          joinQuery = JoinUtil.createJoinQuery("from", multipleValuesPerDocument, "to", actualQuery, indexSearcher, scoreMode);
-        } else {
-          joinQuery = JoinUtil.createJoinQuery("to", multipleValuesPerDocument, "from", actualQuery, indexSearcher, scoreMode);
+        {
+          // single val can be handled by multiple-vals 
+          final boolean muliValsQuery = multipleValuesPerDocument || random().nextBoolean();
+          final String fromField = from ? "from":"to"; 
+          final String toField = from ? "to":"from"; 
+          
+          if (random().nextBoolean()) { // numbers
+            final NumericType numType = random().nextBoolean() ? NumericType.INT: NumericType.LONG ;
+            joinQuery = JoinUtil.createJoinQuery(fromField+numType, muliValsQuery, toField+numType, numType, actualQuery, indexSearcher, scoreMode);
+          } else {
+            joinQuery = JoinUtil.createJoinQuery(fromField, muliValsQuery, toField, actualQuery, indexSearcher, scoreMode);
+          }
         }
         if (VERBOSE) {
           System.out.println("joinQuery=" + joinQuery);
@@ -897,7 +911,6 @@
       return;
     }
 
-    assertEquals(expectedTopDocs.getMaxScore(), actualTopDocs.getMaxScore(), 0.0f);
     if (VERBOSE) {
       for (int i = 0; i < expectedTopDocs.scoreDocs.length; i++) {
         System.out.printf(Locale.ENGLISH, "Expected doc: %d | Actual doc: %d\n", expectedTopDocs.scoreDocs[i].doc, actualTopDocs.scoreDocs[i].doc);
@@ -904,6 +917,7 @@
         System.out.printf(Locale.ENGLISH, "Expected score: %f | Actual score: %f\n", expectedTopDocs.scoreDocs[i].score, actualTopDocs.scoreDocs[i].score);
       }
     }
+    assertEquals(expectedTopDocs.getMaxScore(), actualTopDocs.getMaxScore(), 0.0f);
 
     for (int i = 0; i < expectedTopDocs.scoreDocs.length; i++) {
       assertEquals(expectedTopDocs.scoreDocs[i].doc, actualTopDocs.scoreDocs[i].doc);
@@ -919,14 +933,15 @@
     }
 
     Directory dir = newDirectory();
+    final Random random = random();
     RandomIndexWriter w = new RandomIndexWriter(
-        random(),
+        random,
         dir,
-        newIndexWriterConfig(new MockAnalyzer(random(), MockTokenizer.KEYWORD, false))
+        newIndexWriterConfig(new MockAnalyzer(random, MockTokenizer.KEYWORD, false))
     );
 
     IndexIterationContext context = new IndexIterationContext();
-    int numRandomValues = nDocs / RandomInts.randomIntBetween(random(), 2, 10);
+    int numRandomValues = nDocs / RandomInts.randomIntBetween(random, 1, 4);
     context.randomUniqueValues = new String[numRandomValues];
     Set<String> trackSet = new HashSet<>();
     context.randomFrom = new boolean[numRandomValues];
@@ -933,32 +948,46 @@
     for (int i = 0; i < numRandomValues; i++) {
       String uniqueRandomValue;
       do {
-//        uniqueRandomValue = TestUtil.randomRealisticUnicodeString(random());
-        uniqueRandomValue = TestUtil.randomSimpleString(random());
+        // the trick is to generate values which will be ordered similarly for string, ints&longs, positive nums makes it easier 
+        final int nextInt = random.nextInt(Integer.MAX_VALUE);
+        uniqueRandomValue = String.format(Locale.ROOT, "%08x", nextInt);
+        assert nextInt == Integer.parseUnsignedInt(uniqueRandomValue,16);
       } while ("".equals(uniqueRandomValue) || trackSet.contains(uniqueRandomValue));
+     
       // Generate unique values and empty strings aren't allowed.
       trackSet.add(uniqueRandomValue);
-      context.randomFrom[i] = random().nextBoolean();
+      
+      context.randomFrom[i] = random.nextBoolean();
       context.randomUniqueValues[i] = uniqueRandomValue;
+      
     }
 
+    List<String> randomUniqueValuesReplica = new ArrayList<>(Arrays.asList(context.randomUniqueValues));
+        
     RandomDoc[] docs = new RandomDoc[nDocs];
     for (int i = 0; i < nDocs; i++) {
       String id = Integer.toString(i);
-      int randomI = random().nextInt(context.randomUniqueValues.length);
+      int randomI = random.nextInt(context.randomUniqueValues.length);
       String value = context.randomUniqueValues[randomI];
       Document document = new Document();
-      document.add(newTextField(random(), "id", id, Field.Store.YES));
-      document.add(newTextField(random(), "value", value, Field.Store.NO));
+      document.add(newTextField(random, "id", id, Field.Store.YES));
+      document.add(newTextField(random, "value", value, Field.Store.NO));
 
       boolean from = context.randomFrom[randomI];
-      int numberOfLinkValues = multipleValuesPerDocument ? 2 + random().nextInt(10) : 1;
+      int numberOfLinkValues = multipleValuesPerDocument ? Math.min(2 + random.nextInt(10), context.randomUniqueValues.length) : 1;
       docs[i] = new RandomDoc(id, numberOfLinkValues, value, from);
       if (globalOrdinalJoin) {
         document.add(newStringField("type", from ? "from" : "to", Field.Store.NO));
       }
-      for (int j = 0; j < numberOfLinkValues; j++) {
-        String linkValue = context.randomUniqueValues[random().nextInt(context.randomUniqueValues.length)];
+      final List<String> subValues;
+      {
+      int start = randomUniqueValuesReplica.size()==numberOfLinkValues? 0 : random.nextInt(randomUniqueValuesReplica.size()-numberOfLinkValues);
+      subValues = randomUniqueValuesReplica.subList(start, start+numberOfLinkValues);
+      Collections.shuffle(subValues, random);
+      }
+      for (String linkValue : subValues) {
+        
+        assert !docs[i].linkValues.contains(linkValue);
         docs[i].linkValues.add(linkValue);
         if (from) {
           if (!context.fromDocuments.containsKey(linkValue)) {
@@ -970,15 +999,8 @@
 
           context.fromDocuments.get(linkValue).add(docs[i]);
           context.randomValueFromDocs.get(value).add(docs[i]);
-          document.add(newTextField(random(), "from", linkValue, Field.Store.NO));
-          if (multipleValuesPerDocument) {
-            document.add(new SortedSetDocValuesField("from", new BytesRef(linkValue)));
-          } else {
-            document.add(new SortedDocValuesField("from", new BytesRef(linkValue)));
-          }
-          if (globalOrdinalJoin) {
-            document.add(new SortedDocValuesField("join_field", new BytesRef(linkValue)));
-          }
+          addLinkFields(random, document,  "from", linkValue, multipleValuesPerDocument, globalOrdinalJoin);
+          
         } else {
           if (!context.toDocuments.containsKey(linkValue)) {
             context.toDocuments.put(linkValue, new ArrayList<RandomDoc>());
@@ -989,20 +1011,12 @@
 
           context.toDocuments.get(linkValue).add(docs[i]);
           context.randomValueToDocs.get(value).add(docs[i]);
-          document.add(newTextField(random(), "to", linkValue, Field.Store.NO));
-          if (multipleValuesPerDocument) {
-            document.add(new SortedSetDocValuesField("to", new BytesRef(linkValue)));
-          } else {
-            document.add(new SortedDocValuesField("to", new BytesRef(linkValue)));
-          }
-          if (globalOrdinalJoin) {
-            document.add(new SortedDocValuesField("join_field", new BytesRef(linkValue)));
-          }
+          addLinkFields(random, document,  "to", linkValue, multipleValuesPerDocument, globalOrdinalJoin);
         }
       }
 
       w.addDocument(document);
-      if (random().nextInt(10) == 4) {
+      if (random.nextInt(10) == 4) {
         w.commit();
       }
       if (VERBOSE) {
@@ -1010,7 +1024,7 @@
       }
     }
 
-    if (random().nextBoolean()) {
+    if (random.nextBoolean()) {
       w.forceMerge(1);
     }
     w.close();
@@ -1185,6 +1199,30 @@
     return context;
   }
 
+  private void addLinkFields(final Random random, Document document, final String fieldName, String linkValue,
+      boolean multipleValuesPerDocument, boolean globalOrdinalJoin) {
+    document.add(newTextField(random, fieldName, linkValue, Field.Store.NO));
+
+    final int linkInt = Integer.parseUnsignedInt(linkValue,16);
+    document.add(new IntField(fieldName+NumericType.INT, linkInt, Field.Store.NO));
+
+    final long linkLong = linkInt<<32 | linkInt;
+    document.add(new LongField(fieldName+NumericType.LONG, linkLong, Field.Store.NO));
+
+    if (multipleValuesPerDocument) {
+      document.add(new SortedSetDocValuesField(fieldName, new BytesRef(linkValue)));
+      document.add(new SortedNumericDocValuesField(fieldName+NumericType.INT, linkInt));
+      document.add(new SortedNumericDocValuesField(fieldName+NumericType.LONG, linkLong));
+    } else {
+      document.add(new SortedDocValuesField(fieldName, new BytesRef(linkValue)));
+      document.add(new NumericDocValuesField(fieldName+NumericType.INT, linkInt));
+      document.add(new NumericDocValuesField(fieldName+NumericType.LONG, linkLong));
+    }
+    if (globalOrdinalJoin) {
+      document.add(new SortedDocValuesField("join_field", new BytesRef(linkValue)));
+    }
+  }
+
   private TopDocs createExpectedTopDocs(String queryValue,
                                         final boolean from,
                                         final ScoreMode scoreMode,
Index: lucene/join
===================================================================
--- lucene/join	(revision 1718443)
+++ lucene/join	(working copy)

Property changes on: lucene/join
___________________________________________________________________
Modified: svn:mergeinfo
   Merged /lucene/dev/trunk/lucene/join:r1718443
Index: lucene
===================================================================
--- lucene	(revision 1718443)
+++ lucene	(working copy)

Property changes on: lucene
___________________________________________________________________
Modified: svn:mergeinfo
   Merged /lucene/dev/trunk/lucene:r1718443
Index: .
===================================================================
--- .	(revision 1718443)
+++ .	(working copy)

Property changes on: .
___________________________________________________________________
Modified: svn:mergeinfo
   Merged /lucene/dev/trunk:r1718443
