lucene-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbern...@apache.org
Subject svn commit: r1692325 - in /lucene/dev/branches/branch_5x: ./ solr/ solr/core/ solr/core/src/java/org/apache/solr/handler/ solr/solrj/ solr/solrj/src/java/org/apache/solr/client/solrj/ solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/ solr/solrj...
Date Wed, 22 Jul 2015 20:00:22 GMT
Author: jbernste
Date: Wed Jul 22 20:00:22 2015
New Revision: 1692325

URL: http://svn.apache.org/r1692325
Log:
SOLR-7513, SOLR-7528: Add Equalitors to Streaming Expressions

Added:
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/eq/
      - copied from r1686481, lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/eq/
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/Expressible.java
      - copied unchanged from r1686481, lucene/dev/trunk/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/Expressible.java
Removed:
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/ExpressibleComparator.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/FieldComparator.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExpressibleStream.java
Modified:
    lucene/dev/branches/branch_5x/   (props changed)
    lucene/dev/branches/branch_5x/solr/   (props changed)
    lucene/dev/branches/branch_5x/solr/core/   (props changed)
    lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/SQLHandler.java
    lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/StreamHandler.java
    lucene/dev/branches/branch_5x/solr/solrj/   (props changed)
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/MultiComp.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/StreamComparator.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/MergeStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/PushBackStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RankStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ReducerStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/UniqueStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/StreamFactory.java
    lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/package-info.java
    lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CountStream.java
    lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java
    lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java

Modified: lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/SQLHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/SQLHandler.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/SQLHandler.java (original)
+++ lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/SQLHandler.java Wed Jul 22 20:00:22 2015
@@ -32,7 +32,7 @@ import com.google.common.base.Strings;
 import com.google.common.collect.Iterables;
 import org.apache.solr.client.solrj.io.Tuple;
 import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
 import org.apache.solr.client.solrj.io.comp.MultiComp;
 import org.apache.solr.client.solrj.io.stream.CloudSolrStream;
 import org.apache.solr.client.solrj.io.stream.ParallelStream;
@@ -316,7 +316,7 @@ public class SQLHandler extends RequestH
     for(int i=0; i<buckets.length; i++) {
       ComparatorOrder comparatorOrder = ascDescComp(dir);
       String sortKey = buckets[i].toString();
-      comps[i] = new FieldComparator(stripQuotes(sortKey), comparatorOrder);
+      comps[i] = new StreamComparator(stripQuotes(sortKey), comparatorOrder);
     }
 
     if(comps.length == 1) {
@@ -333,7 +333,7 @@ public class SQLHandler extends RequestH
       String ordering = sortItem.getOrdering().toString();
       ComparatorOrder comparatorOrder = ascDescComp(ordering);
       String sortKey = sortItem.getSortKey().toString();
-      comps[i] = new FieldComparator(stripQuotes(sortKey), comparatorOrder);
+      comps[i] = new StreamComparator(stripQuotes(sortKey), comparatorOrder);
     }
 
     if(comps.length == 1) {

Modified: lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/StreamHandler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/StreamHandler.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/StreamHandler.java (original)
+++ lucene/dev/branches/branch_5x/solr/core/src/java/org/apache/solr/handler/StreamHandler.java Wed Jul 22 20:00:22 2015
@@ -17,14 +17,13 @@
 
 package org.apache.solr.handler;
 
-import java.util.Map.Entry;
-import java.net.URLDecoder;
 import java.io.ByteArrayInputStream;
 import java.io.ObjectInputStream;
+import java.net.URLDecoder;
+import java.util.Map.Entry;
 
 import org.apache.solr.client.solrj.io.SolrClientCache;
 import org.apache.solr.client.solrj.io.stream.CloudSolrStream;
-import org.apache.solr.client.solrj.io.stream.ExpressibleStream;
 import org.apache.solr.client.solrj.io.stream.MergeStream;
 import org.apache.solr.client.solrj.io.stream.ParallelStream;
 import org.apache.solr.client.solrj.io.stream.RankStream;
@@ -32,8 +31,10 @@ import org.apache.solr.client.solrj.io.s
 import org.apache.solr.client.solrj.io.stream.StreamContext;
 import org.apache.solr.client.solrj.io.stream.TupleStream;
 import org.apache.solr.client.solrj.io.stream.UniqueStream;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.Base64;
 import org.apache.solr.common.util.NamedList;
 import org.apache.solr.core.CloseHook;
 import org.apache.solr.core.CoreContainer;
@@ -41,7 +42,6 @@ import org.apache.solr.core.SolrCore;
 import org.apache.solr.request.SolrQueryRequest;
 import org.apache.solr.response.SolrQueryResponse;
 import org.apache.solr.util.plugin.SolrCoreAware;
-import org.apache.solr.common.util.Base64;
 
 public class StreamHandler extends RequestHandlerBase implements SolrCoreAware {
 
@@ -84,7 +84,7 @@ public class StreamHandler extends Reque
     if(null != functionMappingsObj){
       NamedList<?> functionMappings = (NamedList<?>)functionMappingsObj;
       for(Entry<String,?> functionMapping : functionMappings){
-        Class<?> clazz = core.getResourceLoader().findClass((String)functionMapping.getValue(), ExpressibleStream.class);
+        Class<?> clazz = core.getResourceLoader().findClass((String)functionMapping.getValue(), Expressible.class);
         streamFactory.withStreamFunction(functionMapping.getKey(), clazz);
       }
     }

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/MultiComp.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/MultiComp.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/MultiComp.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/MultiComp.java Wed Jul 22 20:00:22 2015
@@ -22,6 +22,7 @@ import java.io.Serializable;
 import java.util.Comparator;
 
 import org.apache.solr.client.solrj.io.Tuple;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
@@ -31,7 +32,7 @@ import org.apache.solr.client.solrj.io.s
  *  Wraps multiple Comparators to provide sub-sorting.
  **/
 
-public class MultiComp implements Comparator<Tuple>, ExpressibleComparator, Serializable {
+public class MultiComp implements Comparator<Tuple>, Expressible, Serializable {
 
   private static final long serialVersionUID = 1;
 
@@ -56,9 +57,9 @@ public class MultiComp implements Compar
   public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
     StringBuilder sb = new StringBuilder();
     for(Comparator<Tuple> comp : comps){
-      if(comp instanceof ExpressibleComparator){
+      if(comp instanceof Expressible){
         if(sb.length() > 0){ sb.append(","); }
-        sb.append(((ExpressibleComparator)comp).toExpression(factory));
+        sb.append(((Expressible)comp).toExpression(factory));
       }
       else{
         throw new IOException("This MultiComp contains a non-expressible comparator - it cannot be converted to an expression");

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/StreamComparator.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/StreamComparator.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/StreamComparator.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/StreamComparator.java Wed Jul 22 20:00:22 2015
@@ -1,12 +1,3 @@
-package org.apache.solr.client.solrj.io.comp;
-
-import java.io.Serializable;
-import java.util.Comparator;
-
-import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
-import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
-
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
@@ -24,22 +15,103 @@ import org.apache.solr.client.solrj.io.s
  * limitations under the License.
  */
 
+package org.apache.solr.client.solrj.io.comp;
+
+import java.io.Serializable;
+import java.util.Comparator;
+
+import org.apache.solr.client.solrj.io.Tuple;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
+import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
+import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
+
 /**
- * Defines a comparator that can be expressed in an expression
- */
-public abstract class StreamComparator implements Comparator<Tuple>, Serializable {
-  protected String leftField;
-  protected String rightField;
-  protected final ComparatorOrder order;
+ *  An equality field Comparator which compares a field of two Tuples and determines sort order.
+ **/
+public class StreamComparator implements Comparator<Tuple>, Expressible, Serializable {
 
+  private static final long serialVersionUID = 1;
+  
+  private String leftField;
+  private String rightField;
+  private final ComparatorOrder order;
+  private ComparatorLambda comparator;
+  
   public StreamComparator(String field, ComparatorOrder order) {
     this.leftField = field;
     this.rightField = field;
     this.order = order;
+    assignComparator();
   }
   public StreamComparator(String leftField, String rightField, ComparatorOrder order){
     this.leftField = leftField;
     this.rightField = rightField;
     this.order = order;
+    assignComparator();
+  }
+  
+  public StreamExpressionParameter toExpression(StreamFactory factory){
+    StringBuilder sb = new StringBuilder();
+    
+    sb.append(leftField);
+    
+    if(!leftField.equals(rightField)){
+      sb.append("=");
+      sb.append(rightField); 
+    }
+    
+    sb.append(" ");
+    sb.append(order);
+    
+    return new StreamExpressionValue(sb.toString());
+  }
+  
+  /*
+   * What're we doing here messing around with lambdas for the comparator logic?
+   * We want the compare(...) function to run as fast as possible because it will be called many many
+   * times over the lifetime of this object. For that reason we want to limit the number of comparisons
+   * taking place in the compare(...) function. Because this class supports both ascending and
+   * descending comparisons and the logic for each is slightly different, we want to do the 
+   *   if(ascending){ compare like this } else { compare like this }
+   * check only once - we can do that in the constructor of this class, create a lambda, and then execute 
+   * that lambda in the compare function. A little bit of branch prediction savings right here.
+   */
+  private void assignComparator(){
+    if(ComparatorOrder.DESCENDING == order){
+      comparator = new ComparatorLambda() {
+        @Override
+        public int compare(Tuple leftTuple, Tuple rightTuple) {
+          Comparable leftComp = (Comparable)leftTuple.get(leftField);
+          Comparable rightComp = (Comparable)rightTuple.get(rightField);
+          
+          if(leftComp == rightComp){ return 0; } // if both null then they are equal. if both are same ref then are equal
+          if(null == leftComp){ return 1; }
+          if(null == rightComp){ return -1; }
+          
+          return rightComp.compareTo(leftComp);
+        }
+      };
+    }
+    else{
+      // See above for black magic reasoning.
+      comparator = new ComparatorLambda() {
+        @Override
+        public int compare(Tuple leftTuple, Tuple rightTuple) {
+          Comparable leftComp = (Comparable)leftTuple.get(leftField);
+          Comparable rightComp = (Comparable)rightTuple.get(rightField);
+          
+          if(leftComp == rightComp){ return 0; } // if both null then they are equal. if both are same ref then are equal
+          if(null == leftComp){ return -1; }
+          if(null == rightComp){ return 1; }
+          
+          return leftComp.compareTo(rightComp);
+        }
+      };
+    }
+  }
+
+  public int compare(Tuple leftTuple, Tuple rightTuple) {
+    return comparator.compare(leftTuple, rightTuple); 
   }
-}
+}
\ No newline at end of file

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java Wed Jul 22 20:00:22 2015
@@ -38,8 +38,9 @@ import org.apache.solr.client.solrj.impl
 import org.apache.solr.client.solrj.io.SolrClientCache;
 import org.apache.solr.client.solrj.io.Tuple;
 import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
 import org.apache.solr.client.solrj.io.comp.MultiComp;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
@@ -60,7 +61,7 @@ import org.apache.solr.common.util.Solrj
  * to iterate and merge Tuples from each SolrStream.
  **/
 
-public class CloudSolrStream extends TupleStream implements ExpressibleStream {
+public class CloudSolrStream extends TupleStream implements Expressible {
 
   private static final long serialVersionUID = 1;
 
@@ -268,7 +269,7 @@ public class CloudSolrStream extends Tup
         fieldName = fieldMappings.get(fieldName);
       }
       
-      comps[i] = new FieldComparator(fieldName, order.equalsIgnoreCase("asc") ? ComparatorOrder.ASCENDING : ComparatorOrder.DESCENDING);
+      comps[i] = new StreamComparator(fieldName, order.equalsIgnoreCase("asc") ? ComparatorOrder.ASCENDING : ComparatorOrder.DESCENDING);
     }
 
     if(comps.length > 1) {

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/MergeStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/MergeStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/MergeStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/MergeStream.java Wed Jul 22 20:00:22 2015
@@ -18,15 +18,14 @@
 package org.apache.solr.client.solrj.io.stream;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.List;
-import java.util.ArrayList;
 import java.util.Locale;
-import java.util.Map.Entry;
 
 import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
-import org.apache.solr.client.solrj.io.comp.ExpressibleComparator;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
@@ -38,7 +37,7 @@ import org.apache.solr.client.solrj.io.s
 **/
 
 
-public class MergeStream extends TupleStream implements ExpressibleStream {
+public class MergeStream extends TupleStream implements Expressible {
 
   private static final long serialVersionUID = 1;
 
@@ -54,7 +53,7 @@ public class MergeStream extends TupleSt
   
   public MergeStream(StreamExpression expression,StreamFactory factory) throws IOException {
     // grab all parameters out
-    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
+    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
     StreamExpressionNamedParameter onExpression = factory.getNamedOperand(expression, "on");
     
     // validate expression contains only what we want.
@@ -73,7 +72,7 @@ public class MergeStream extends TupleSt
     }
     
     // Merge is always done over equality, so always use an EqualTo comparator
-    this.comp = factory.constructComparator(((StreamExpressionValue)onExpression.getParameter()).getValue(), FieldComparator.class);
+    this.comp = factory.constructComparator(((StreamExpressionValue)onExpression.getParameter()).getValue(), StreamComparator.class);
   }
   
   @Override
@@ -86,8 +85,8 @@ public class MergeStream extends TupleSt
     expression.addParameter(streamB.toExpression(factory));
     
     // on
-    if(comp instanceof ExpressibleComparator){
-      expression.addParameter(new StreamExpressionNamedParameter("on",((ExpressibleComparator)comp).toExpression(factory)));
+    if(comp instanceof Expressible){
+      expression.addParameter(new StreamExpressionNamedParameter("on",((Expressible)comp).toExpression(factory)));
     }
     else{
       throw new IOException("This MergeStream contains a non-expressible comparator - it cannot be converted to an expression");

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelStream.java Wed Jul 22 20:00:22 2015
@@ -17,25 +17,25 @@
 
 package org.apache.solr.client.solrj.io.stream;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.ObjectOutputStream;
 import java.net.URLEncoder;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.io.ByteArrayOutputStream;
 import java.util.Random;
 
 import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
-import org.apache.solr.client.solrj.io.comp.ExpressibleComparator;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
@@ -54,7 +54,7 @@ import org.apache.solr.common.util.Base6
  **/
 
 
-public class ParallelStream extends CloudSolrStream implements ExpressibleStream {
+public class ParallelStream extends CloudSolrStream implements Expressible {
 
   private TupleStream tupleStream;
   private int workers;
@@ -85,7 +85,7 @@ public class ParallelStream extends Clou
     objectSerialize = false;
     String collectionName = factory.getValueOperand(expression, 0);
     StreamExpressionNamedParameter workersParam = factory.getNamedOperand(expression, "workers");
-    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
+    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
     StreamExpressionNamedParameter sortExpression = factory.getNamedOperand(expression, "sort");
     StreamExpressionNamedParameter zkHostExpression = factory.getNamedOperand(expression, "zkHost");
     
@@ -140,7 +140,7 @@ public class ParallelStream extends Clou
     
     // We've got all the required items    
     TupleStream stream = factory.constructStream(streamExpressions.get(0));
-    Comparator<Tuple> comp = factory.constructComparator(((StreamExpressionValue)sortExpression.getParameter()).getValue(), FieldComparator.class);
+    Comparator<Tuple> comp = factory.constructComparator(((StreamExpressionValue)sortExpression.getParameter()).getValue(), StreamComparator.class);
     streamFactory = factory;
     init(zkHost,collectionName,stream,workersInt,comp);
   }
@@ -153,7 +153,7 @@ public class ParallelStream extends Clou
     this.tupleStream = tupleStream;
 
     // requires Expressible stream and comparator
-    if(!objectSerialize && !(tupleStream instanceof ExpressibleStream)){
+    if(!objectSerialize && !(tupleStream instanceof Expressible)){
       throw new IOException("Unable to create ParallelStream with a non-expressible TupleStream.");
     }
   }
@@ -171,16 +171,16 @@ public class ParallelStream extends Clou
     expression.addParameter(new StreamExpressionNamedParameter("workers", Integer.toString(workers)));
     
     // stream
-    if(tupleStream instanceof ExpressibleStream){
-      expression.addParameter(((ExpressibleStream)tupleStream).toExpression(factory));
+    if(tupleStream instanceof Expressible){
+      expression.addParameter(((Expressible)tupleStream).toExpression(factory));
     }
     else{
       throw new IOException("This ParallelStream contains a non-expressible TupleStream - it cannot be converted to an expression");
     }
         
     // sort
-    if(comp instanceof ExpressibleComparator){
-      expression.addParameter(new StreamExpressionNamedParameter("sort",((ExpressibleComparator)comp).toExpression(factory)));
+    if(comp instanceof Expressible){
+      expression.addParameter(new StreamExpressionNamedParameter("sort",((Expressible)comp).toExpression(factory)));
     }
     else{
       throw new IOException("This ParallelStream contains a non-expressible comparator - it cannot be converted to an expression");
@@ -241,7 +241,7 @@ public class ParallelStream extends Clou
         String encoded = Base64.byteArrayToBase64(bytes, 0, bytes.length);
         pushStream = URLEncoder.encode(encoded, "UTF-8");
       } else {
-        pushStream = ((ExpressibleStream) tupleStream).toExpression(streamFactory);
+        pushStream = ((Expressible) tupleStream).toExpression(streamFactory);
       }
 
       ZkStateReader zkStateReader = cloudSolrClient.getZkStateReader();

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/PushBackStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/PushBackStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/PushBackStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/PushBackStream.java Wed Jul 22 20:00:22 2015
@@ -22,7 +22,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 
@@ -44,8 +44,8 @@ public class PushBackStream extends Tupl
   }
   
   public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException{
-    if(stream instanceof ExpressibleStream){
-      return ((ExpressibleStream)stream).toExpression(factory);
+    if(stream instanceof Expressible){
+      return ((Expressible)stream).toExpression(factory);
     }
     
     throw new IOException("This PushBackStream contains a non-expressible TupleStream - it cannot be converted to an expression");

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RankStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RankStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RankStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RankStream.java Wed Jul 22 20:00:22 2015
@@ -27,8 +27,8 @@ import java.util.Locale;
 import java.util.PriorityQueue;
 
 import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
-import org.apache.solr.client.solrj.io.comp.ExpressibleComparator;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
@@ -39,7 +39,7 @@ import org.apache.solr.client.solrj.io.s
 *  Iterates over a TupleStream and Ranks the topN tuples based on a Comparator.
 **/
 
-public class RankStream extends TupleStream implements ExpressibleStream {
+public class RankStream extends TupleStream implements Expressible {
 
   private static final long serialVersionUID = 1;
 
@@ -56,7 +56,7 @@ public class RankStream extends TupleStr
   
   public RankStream(StreamExpression expression, StreamFactory factory) throws IOException {
     // grab all parameters out
-    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
+    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
     StreamExpressionNamedParameter nParam = factory.getNamedOperand(expression, "n");
     StreamExpressionNamedParameter sortExpression = factory.getNamedOperand(expression, "sort");
     
@@ -87,7 +87,7 @@ public class RankStream extends TupleStr
     }
     
     TupleStream stream = factory.constructStream(streamExpressions.get(0));
-    Comparator<Tuple> comp = factory.constructComparator(((StreamExpressionValue)sortExpression.getParameter()).getValue(), FieldComparator.class);
+    Comparator<Tuple> comp = factory.constructComparator(((StreamExpressionValue)sortExpression.getParameter()).getValue(), StreamComparator.class);
     
     init(stream,nInt,comp);    
   }
@@ -107,16 +107,16 @@ public class RankStream extends TupleStr
     expression.addParameter(new StreamExpressionNamedParameter("n", Integer.toString(size)));
     
     // stream
-    if(tupleStream instanceof ExpressibleStream){
-      expression.addParameter(((ExpressibleStream)tupleStream).toExpression(factory));
+    if(tupleStream instanceof Expressible){
+      expression.addParameter(((Expressible)tupleStream).toExpression(factory));
     }
     else{
       throw new IOException("This RankStream contains a non-expressible TupleStream - it cannot be converted to an expression");
     }
         
     // sort
-    if(comp instanceof ExpressibleComparator){
-      expression.addParameter(new StreamExpressionNamedParameter("sort",((ExpressibleComparator)comp).toExpression(factory)));
+    if(comp instanceof Expressible){
+      expression.addParameter(new StreamExpressionNamedParameter("sort",((Expressible)comp).toExpression(factory)));
     }
     else{
       throw new IOException("This RankStream contains a non-expressible comparator - it cannot be converted to an expression");

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ReducerStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ReducerStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ReducerStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ReducerStream.java Wed Jul 22 20:00:22 2015
@@ -18,16 +18,16 @@
 package org.apache.solr.client.solrj.io.stream;
 
 import java.io.IOException;
-import java.util.List;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
-import java.util.HashMap;
 
 import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
-import org.apache.solr.client.solrj.io.comp.ExpressibleComparator;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
@@ -48,7 +48,7 @@ import org.apache.solr.client.solrj.io.s
  *
  **/
 
-public class ReducerStream extends TupleStream implements ExpressibleStream {
+public class ReducerStream extends TupleStream implements Expressible {
 
   private static final long serialVersionUID = 1;
 
@@ -65,7 +65,7 @@ public class ReducerStream extends Tuple
   
   public ReducerStream(StreamExpression expression, StreamFactory factory) throws IOException{
     // grab all parameters out
-    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
+    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
     StreamExpressionNamedParameter byExpression = factory.getNamedOperand(expression, "by");
     
     // validate expression contains only what we want.
@@ -83,7 +83,7 @@ public class ReducerStream extends Tuple
     }
     
     // Reducing is always done over equality, so always use an EqualTo comparator
-    this.comp = factory.constructComparator(((StreamExpressionValue)byExpression.getParameter()).getValue(), FieldComparator.class);
+    this.comp = factory.constructComparator(((StreamExpressionValue)byExpression.getParameter()).getValue(), StreamComparator.class);
   }
 
   @Override
@@ -95,8 +95,8 @@ public class ReducerStream extends Tuple
     expression.addParameter(tupleStream.toExpression(factory));
     
     // over
-    if(comp instanceof ExpressibleComparator){
-      expression.addParameter(new StreamExpressionNamedParameter("by",((ExpressibleComparator)comp).toExpression(factory)));
+    if(comp instanceof Expressible){
+      expression.addParameter(new StreamExpressionNamedParameter("by",((Expressible)comp).toExpression(factory)));
     }
     else{
       throw new IOException("This ReducerStream contains a non-expressible comparator - it cannot be converted to an expression");

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/UniqueStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/UniqueStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/UniqueStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/UniqueStream.java Wed Jul 22 20:00:22 2015
@@ -24,8 +24,10 @@ import java.util.List;
 import java.util.Locale;
 
 import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
-import org.apache.solr.client.solrj.io.comp.ExpressibleComparator;
+import org.apache.solr.client.solrj.io.comp.StreamComparator;
+import org.apache.solr.client.solrj.io.eq.Equalitor;
+import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
@@ -38,22 +40,22 @@ import org.apache.solr.client.solrj.io.s
  * Note: The sort order of the underlying stream must match the Comparator.
  **/
 
-public class UniqueStream extends TupleStream implements ExpressibleStream {
+public class UniqueStream extends TupleStream implements Expressible {
 
   private static final long serialVersionUID = 1;
 
   private TupleStream tupleStream;
-  private Comparator<Tuple> comp;
+  private Equalitor<Tuple> eq;
   private transient Tuple currentTuple;
 
-  public UniqueStream(TupleStream tupleStream, Comparator<Tuple> comp) {
+  public UniqueStream(TupleStream tupleStream, Equalitor<Tuple> eq) {
     this.tupleStream = tupleStream;
-    this.comp = comp;
+    this.eq = eq;
   }
   
   public UniqueStream(StreamExpression expression,StreamFactory factory) throws IOException {
     // grab all parameters out
-    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
+    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
     StreamExpressionNamedParameter overExpression = factory.getNamedOperand(expression, "over");
     
     // validate expression contains only what we want.
@@ -71,7 +73,7 @@ public class UniqueStream extends TupleS
     }
     
     // Uniqueness is always done over equality, so always use an EqualTo comparator
-    this.comp = factory.constructComparator(((StreamExpressionValue)overExpression.getParameter()).getValue(), FieldComparator.class);
+    this.eq = factory.constructEqualitor(((StreamExpressionValue)overExpression.getParameter()).getValue(), StreamEqualitor.class);
   }
 
   @Override
@@ -80,28 +82,24 @@ public class UniqueStream extends TupleS
     StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
     
     // streams
-    if(tupleStream instanceof ExpressibleStream){
-      expression.addParameter(((ExpressibleStream)tupleStream).toExpression(factory));
+    if(tupleStream instanceof Expressible){
+      expression.addParameter(((Expressible)tupleStream).toExpression(factory));
     }
     else{
       throw new IOException("This UniqueStream contains a non-expressible TupleStream - it cannot be converted to an expression");
     }
     
     // over
-    if(comp instanceof ExpressibleComparator){
-      expression.addParameter(new StreamExpressionNamedParameter("over",((ExpressibleComparator)comp).toExpression(factory)));
+    if(eq instanceof Expressible){
+      expression.addParameter(new StreamExpressionNamedParameter("over",((Expressible)eq).toExpression(factory)));
     }
     else{
-      throw new IOException("This UniqueStream contains a non-expressible comparator - it cannot be converted to an expression");
+      throw new IOException("This UniqueStream contains a non-expressible equalitor - it cannot be converted to an expression");
     }
     
     return expression;   
   }
-  
-  public void setComp(Comparator<Tuple> comp) {
-    this.comp = comp;
-  }
-  
+    
   public void setStreamContext(StreamContext context) {
     this.tupleStream.setStreamContext(context);
   }
@@ -131,8 +129,7 @@ public class UniqueStream extends TupleS
       return tuple;
     } else {
       while(true) {
-        int i = comp.compare(currentTuple, tuple);
-        if(i == 0) {
+        if(eq.test(currentTuple, tuple)){
           //We have duplicate tuple so read the next tuple from the stream.
           tuple = tupleStream.read();
           if(tuple.EOF) {

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/StreamFactory.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/StreamFactory.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/StreamFactory.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/StreamFactory.java Wed Jul 22 20:00:22 2015
@@ -14,9 +14,9 @@ import java.util.Map.Entry;
 
 import org.apache.solr.client.solrj.io.Tuple;
 import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
-import org.apache.solr.client.solrj.io.comp.ExpressibleComparator;
 import org.apache.solr.client.solrj.io.comp.MultiComp;
-import org.apache.solr.client.solrj.io.stream.ExpressibleStream;
+import org.apache.solr.client.solrj.io.eq.Equalitor;
+import org.apache.solr.client.solrj.io.eq.MultiEqualitor;
 import org.apache.solr.client.solrj.io.stream.TupleStream;
 
 /*
@@ -170,7 +170,7 @@ public class StreamFactory implements Se
     String function = expression.getFunctionName();
     if(streamFunctions.containsKey(function)){
       Class clazz = streamFunctions.get(function);
-      if(ExpressibleStream.class.isAssignableFrom(clazz) && TupleStream.class.isAssignableFrom(clazz)){
+      if(Expressible.class.isAssignableFrom(clazz) && TupleStream.class.isAssignableFrom(clazz)){
         TupleStream stream = (TupleStream)createInstance(streamFunctions.get(function), new Class[]{ StreamExpression.class, StreamFactory.class }, new Object[]{ expression, this});
         return stream;
       }
@@ -201,6 +201,36 @@ public class StreamFactory implements Se
     }
   }
     
+  public Equalitor<Tuple> constructEqualitor(String equalitorString, Class equalitorType) throws IOException {
+    if(equalitorString.contains(",")){
+      String[] parts = equalitorString.split(",");
+      Equalitor[] eqs = new Equalitor[parts.length];
+      for(int idx = 0; idx < parts.length; ++idx){
+        eqs[idx] = constructEqualitor(parts[idx].trim(), equalitorType);
+      }
+      return new MultiEqualitor(eqs);
+    }
+    else{
+      String leftFieldName;
+      String rightFieldName;
+      
+      if(equalitorString.contains("=")){
+        String[] parts = equalitorString.split("=");
+        if(2 != parts.length){
+          throw new IOException(String.format(Locale.ROOT,"Invalid equalitor expression %s - expecting fieldName=fieldName",equalitorString));
+        }
+        
+        leftFieldName = parts[0].trim();
+        rightFieldName = parts[1].trim();
+      }
+      else{
+        leftFieldName = rightFieldName = equalitorString.trim();
+      }
+      
+      return (Equalitor)createInstance(equalitorType, new Class[]{ String.class, String.class }, new Object[]{ leftFieldName, rightFieldName });
+    }
+  }
+
   public <T> T createInstance(Class<T> clazz, Class<?>[] paramTypes, Object[] params) throws IOException{
     // This should use SolrResourceLoader - TODO
     // This is adding a restriction that the class has a public constructor - we may not want to do that

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/package-info.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/package-info.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/package-info.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/package-info.java Wed Jul 22 20:00:22 2015
@@ -15,9 +15,9 @@
  * limitations under the License.
  */
  
-/** 
+/**
  * Primary APIs for communicating with a Solr Server from a Java client.
- */
+ **/
 package org.apache.solr.client.solrj;
 
 

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CountStream.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CountStream.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CountStream.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/CountStream.java Wed Jul 22 20:00:22 2015
@@ -23,11 +23,12 @@ import java.util.List;
 import java.util.Locale;
 
 import org.apache.solr.client.solrj.io.Tuple;
+import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 
-public class CountStream extends TupleStream implements ExpressibleStream, Serializable {
+public class CountStream extends TupleStream implements Expressible, Serializable {
 
   private TupleStream stream;
   private int count;
@@ -37,7 +38,7 @@ public class CountStream extends TupleSt
   }
   
   public CountStream(StreamExpression expression, StreamFactory factory) throws IOException{
-    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
+    List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
     
     // validate expression contains only what we want.
     if(expression.getParameters().size() != streamExpressions.size()){
@@ -57,8 +58,8 @@ public class CountStream extends TupleSt
     StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
     
     // stream
-    if(stream instanceof ExpressibleStream){
-      expression.addParameter(((ExpressibleStream)stream).toExpression(factory));
+    if(stream instanceof Expressible){
+      expression.addParameter(((Expressible)stream).toExpression(factory));
     }
     else{
       throw new IOException("This CountStream contains a non-expressible TupleStream - it cannot be converted to an expression");

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java Wed Jul 22 20:00:22 2015
@@ -236,7 +236,7 @@ public class StreamExpressionTest extend
       .withStreamFunction("unique", UniqueStream.class);
     
     // Basic test
-    expression = StreamExpressionParser.parse("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), over=\"a_f asc\")");
+    expression = StreamExpressionParser.parse("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), over=\"a_f\")");
     stream = new UniqueStream(expression, factory);
     tuples = getTuples(stream);
     
@@ -244,7 +244,7 @@ public class StreamExpressionTest extend
     assertOrder(tuples, 0, 1, 3, 4);
 
     // Basic test desc
-    expression = StreamExpressionParser.parse("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f desc, a_i desc\"), over=\"a_f desc\")");
+    expression = StreamExpressionParser.parse("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f desc, a_i desc\"), over=\"a_f\")");
     stream = new UniqueStream(expression, factory);
     tuples = getTuples(stream);
     
@@ -252,7 +252,7 @@ public class StreamExpressionTest extend
     assertOrder(tuples, 4,3,1,2);
     
     // Basic w/multi comp
-    expression = StreamExpressionParser.parse("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), over=\"a_f asc, a_i asc\")");
+    expression = StreamExpressionParser.parse("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), over=\"a_f, a_i\")");
     stream = new UniqueStream(expression, factory);
     tuples = getTuples(stream);
     
@@ -260,7 +260,7 @@ public class StreamExpressionTest extend
     assertOrder(tuples, 0,2,1,3,4);
     
     // full factory w/multi comp
-    stream = factory.constructStream("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), over=\"a_f asc, a_i asc\")");
+    stream = factory.constructStream("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), over=\"a_f, a_i\")");
     tuples = getTuples(stream);
     
     assert(tuples.size() == 5);
@@ -371,7 +371,7 @@ public class StreamExpressionTest extend
                                               + "n=2,"
                                               + "unique("
                                               +   "search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f desc\"),"
-                                              +   "over=\"a_f desc\"),"
+                                              +   "over=\"a_f\"),"
                                               + "sort=\"a_f desc\")");
     stream = new RankStream(expression, factory);
     tuples = getTuples(stream);
@@ -384,7 +384,7 @@ public class StreamExpressionTest extend
                                     + "n=4,"
                                     + "unique("
                                     +   "search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"),"
-                                    +   "over=\"a_f asc\"),"
+                                    +   "over=\"a_f\"),"
                                     + "sort=\"a_f asc\")");
     tuples = getTuples(stream);
     
@@ -491,7 +491,7 @@ public class StreamExpressionTest extend
         .withStreamFunction("group", ReducerStream.class)
         .withStreamFunction("parallel", ParallelStream.class);
 
-    ParallelStream pstream = (ParallelStream)streamFactory.constructStream("parallel(collection1, unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\", partitionKeys=\"a_f\"), over=\"a_f asc\"), workers=\"2\", zkHost=\""+zkHost+"\", sort=\"a_f asc\")");
+    ParallelStream pstream = (ParallelStream)streamFactory.constructStream("parallel(collection1, unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\", partitionKeys=\"a_f\"), over=\"a_f\"), workers=\"2\", zkHost=\""+zkHost+"\", sort=\"a_f asc\")");
 
     List<Tuple> tuples = getTuples(pstream);
     assert(tuples.size() == 5);

Modified: lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java?rev=1692325&r1=1692324&r2=1692325&view=diff
==============================================================================
--- lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java (original)
+++ lucene/dev/branches/branch_5x/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java Wed Jul 22 20:00:22 2015
@@ -19,25 +19,19 @@ package org.apache.solr.client.solrj.io.
 
 import java.io.File;
 import java.io.IOException;
-import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.HashMap;
 
 import org.apache.lucene.util.LuceneTestCase;
 import org.apache.lucene.util.LuceneTestCase.Slow;
 import org.apache.solr.client.solrj.io.Tuple;
 import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
-import org.apache.solr.client.solrj.io.comp.FieldComparator;
 import org.apache.solr.client.solrj.io.comp.MultiComp;
-import org.apache.solr.client.solrj.io.stream.CloudSolrStream;
-import org.apache.solr.client.solrj.io.stream.MergeStream;
-import org.apache.solr.client.solrj.io.stream.ParallelStream;
-import org.apache.solr.client.solrj.io.stream.RankStream;
-import org.apache.solr.client.solrj.io.stream.ReducerStream;
-import org.apache.solr.client.solrj.io.stream.TupleStream;
-import org.apache.solr.client.solrj.io.stream.UniqueStream;
+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.StreamFactory;
 import org.apache.solr.client.solrj.io.stream.metrics.Bucket;
 import org.apache.solr.client.solrj.io.stream.metrics.CountMetric;
@@ -55,9 +49,6 @@ import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
-import java.util.List;
-import java.util.ArrayList;
-
 /**
  *  All base tests will be done with CloudSolrStream. Under the covers CloudSolrStream uses SolrStream so
  *  SolrStream will get fully exercised through these tests.
@@ -148,7 +139,7 @@ public class StreamingTest extends Abstr
 
     Map params = mapParams("q","*:*","fl","id,a_s,a_i,a_f","sort", "a_f asc,a_i asc");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", params);
-    UniqueStream ustream = new UniqueStream(stream, new FieldComparator("a_f",ComparatorOrder.ASCENDING));
+    UniqueStream ustream = new UniqueStream(stream, new StreamEqualitor("a_f"));
     List<Tuple> tuples = getTuples(ustream);
     assert(tuples.size() == 4);
     assertOrder(tuples, 0,1,3,4);
@@ -197,7 +188,7 @@ public class StreamingTest extends Abstr
 
     Map paramsA = mapParams("q", "*:*", "fl", "id,a_s,a_i,a_f", "sort", "a_s asc,a_f asc", "partitionKeys", "none");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", paramsA);
-    ParallelStream pstream = new ParallelStream(zkHost, "collection1", stream, 2, new FieldComparator("a_s",ComparatorOrder.ASCENDING));
+    ParallelStream pstream = new ParallelStream(zkHost, "collection1", stream, 2, new StreamComparator("a_s",ComparatorOrder.ASCENDING));
 
     List<Tuple> tuples = getTuples(pstream);
 
@@ -230,8 +221,8 @@ public class StreamingTest extends Abstr
 
     Map params = mapParams("q","*:*","fl","id,a_s,a_i,a_f","sort", "a_f asc,a_i asc", "partitionKeys", "a_f");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", params);
-    UniqueStream ustream = new UniqueStream(stream, new FieldComparator("a_f",ComparatorOrder.ASCENDING));
-    ParallelStream pstream = new ParallelStream(zkHost, "collection1", ustream, 2, new FieldComparator("a_f",ComparatorOrder.ASCENDING));
+    UniqueStream ustream = new UniqueStream(stream, new StreamEqualitor("a_f"));
+    ParallelStream pstream = new ParallelStream(zkHost, "collection1", ustream, 2, new StreamComparator("a_f",ComparatorOrder.ASCENDING));
     List<Tuple> tuples = getTuples(pstream);
     assert(tuples.size() == 5);
     assertOrder(tuples, 0,1,3,4,6);
@@ -264,7 +255,7 @@ public class StreamingTest extends Abstr
 
     Map params = mapParams("q", "*:*", "fl", "id,a_s,a_i", "sort", "a_i asc");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", params);
-    RankStream rstream = new RankStream(stream, 3, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
+    RankStream rstream = new RankStream(stream, 3, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
     List<Tuple> tuples = getTuples(rstream);
 
 
@@ -296,8 +287,8 @@ public class StreamingTest extends Abstr
 
     Map params = mapParams("q", "*:*", "fl", "id,a_s,a_i", "sort", "a_i asc", "partitionKeys", "a_i");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", params);
-    RankStream rstream = new RankStream(stream, 11, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
-    ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
+    RankStream rstream = new RankStream(stream, 11, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
+    ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
     List<Tuple> tuples = getTuples(pstream);
 
     assert(tuples.size() == 10);
@@ -363,7 +354,7 @@ public class StreamingTest extends Abstr
     //Test with spaces in the parameter lists.
     Map paramsA = mapParams("q","*:*","fl","id,a_s, a_i,  a_f","sort", "a_s asc  ,  a_f   asc");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", paramsA);
-    ReducerStream rstream = new ReducerStream(stream, new FieldComparator("a_s",ComparatorOrder.ASCENDING));
+    ReducerStream rstream = new ReducerStream(stream, new StreamComparator("a_s",ComparatorOrder.ASCENDING));
 
     List<Tuple> tuples = getTuples(rstream);
 
@@ -410,7 +401,7 @@ public class StreamingTest extends Abstr
     //Test with spaces in the parameter lists.
     Map paramsA = mapParams("q", "blah", "fl", "id,a_s, a_i,  a_f", "sort", "a_s asc  ,  a_f   asc");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", paramsA);
-    ReducerStream rstream = new ReducerStream(stream, new FieldComparator("a_s", ComparatorOrder.ASCENDING));
+    ReducerStream rstream = new ReducerStream(stream, new StreamComparator("a_s", ComparatorOrder.ASCENDING));
 
     List<Tuple> tuples = getTuples(rstream);
 
@@ -441,8 +432,8 @@ public class StreamingTest extends Abstr
 
     Map paramsA = mapParams("q","*:*","fl","id,a_s,a_i,a_f","sort", "a_s asc,a_f asc", "partitionKeys", "a_s");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", paramsA);
-    ReducerStream rstream = new ReducerStream(stream, new FieldComparator("a_s",ComparatorOrder.ASCENDING));
-    ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new FieldComparator("a_s",ComparatorOrder.ASCENDING));
+    ReducerStream rstream = new ReducerStream(stream, new StreamComparator("a_s",ComparatorOrder.ASCENDING));
+    ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_s",ComparatorOrder.ASCENDING));
 
     List<Tuple> tuples = getTuples(pstream);
 
@@ -465,8 +456,8 @@ public class StreamingTest extends Abstr
 
     paramsA = mapParams("q","*:*","fl","id,a_s,a_i,a_f","sort", "a_s desc,a_f asc", "partitionKeys", "a_s");
     stream = new CloudSolrStream(zkHost, "collection1", paramsA);
-    rstream = new ReducerStream(stream, new FieldComparator("a_s",ComparatorOrder.DESCENDING));
-    pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new FieldComparator("a_s",ComparatorOrder.DESCENDING));
+    rstream = new ReducerStream(stream, new StreamComparator("a_s",ComparatorOrder.DESCENDING));
+    pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_s",ComparatorOrder.DESCENDING));
 
     tuples = getTuples(pstream);
 
@@ -648,7 +639,7 @@ public class StreamingTest extends Abstr
                         new CountMetric()};
 
     RollupStream rollupStream = new RollupStream(stream, buckets, metrics);
-    ParallelStream parallelStream = new ParallelStream(zkHost, "collection1", rollupStream, 2, new FieldComparator("a_s", ComparatorOrder.ASCENDING));
+    ParallelStream parallelStream = new ParallelStream(zkHost, "collection1", rollupStream, 2, new StreamComparator("a_s", ComparatorOrder.ASCENDING));
     List<Tuple> tuples = getTuples(parallelStream);
 
     assert(tuples.size() == 3);
@@ -748,8 +739,8 @@ public class StreamingTest extends Abstr
 
     Map paramsA = mapParams("q","blah","fl","id,a_s,a_i,a_f","sort", "a_s asc,a_f asc", "partitionKeys", "a_s");
     CloudSolrStream stream = new CloudSolrStream(zkHost, "collection1", paramsA);
-    ReducerStream rstream = new ReducerStream(stream, new FieldComparator("a_s", ComparatorOrder.ASCENDING));
-    ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new FieldComparator("a_s", ComparatorOrder.ASCENDING));
+    ReducerStream rstream = new ReducerStream(stream, new StreamComparator("a_s", ComparatorOrder.ASCENDING));
+    ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_s", ComparatorOrder.ASCENDING));
 
     List<Tuple> tuples = getTuples(pstream);
     assert(tuples.size() == 0);
@@ -818,7 +809,7 @@ public class StreamingTest extends Abstr
     Map paramsB = mapParams("q","id:(0 2 3)","fl","id,a_s,a_i","sort", "a_i asc");
     CloudSolrStream streamB = new CloudSolrStream(zkHost, "collection1", paramsB);
 
-    MergeStream mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
+    MergeStream mstream = new MergeStream(streamA, streamB, new StreamComparator("a_i",ComparatorOrder.ASCENDING));
     List<Tuple> tuples = getTuples(mstream);
 
     assert(tuples.size() == 5);
@@ -831,7 +822,7 @@ public class StreamingTest extends Abstr
     paramsB = mapParams("q","id:(0 2 3)","fl","id,a_s,a_i","sort", "a_i desc");
     streamB = new CloudSolrStream(zkHost, "collection1", paramsB);
 
-    mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
+    mstream = new MergeStream(streamA, streamB, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
     tuples = getTuples(mstream);
 
     assert(tuples.size() == 5);
@@ -845,7 +836,7 @@ public class StreamingTest extends Abstr
     paramsB = mapParams("q","id:(0 3)","fl","id,a_s,a_i,a_f","sort", "a_f asc,a_i asc");
     streamB = new CloudSolrStream(zkHost, "collection1", paramsB);
 
-    mstream = new MergeStream(streamA, streamB, new MultiComp(new FieldComparator("a_f",ComparatorOrder.ASCENDING),new FieldComparator("a_i",ComparatorOrder.ASCENDING)));
+    mstream = new MergeStream(streamA, streamB, new MultiComp(new StreamComparator("a_f",ComparatorOrder.ASCENDING),new StreamComparator("a_i",ComparatorOrder.ASCENDING)));
     tuples = getTuples(mstream);
 
     assert(tuples.size() == 5);
@@ -857,7 +848,7 @@ public class StreamingTest extends Abstr
     paramsB = mapParams("q","id:(0 3)","fl","id,a_s,a_i,a_f","sort", "a_f asc,a_i desc");
     streamB = new CloudSolrStream(zkHost, "collection1", paramsB);
 
-    mstream = new MergeStream(streamA, streamB, new MultiComp(new FieldComparator("a_f",ComparatorOrder.ASCENDING),new FieldComparator("a_i",ComparatorOrder.DESCENDING)));
+    mstream = new MergeStream(streamA, streamB, new MultiComp(new StreamComparator("a_f",ComparatorOrder.ASCENDING),new StreamComparator("a_i",ComparatorOrder.DESCENDING)));
     tuples = getTuples(mstream);
 
     assert(tuples.size() == 5);
@@ -893,8 +884,8 @@ public class StreamingTest extends Abstr
     Map paramsB = mapParams("q","id:(0 2 3 6)","fl","id,a_s,a_i","sort", "a_i asc", "partitionKeys", "a_i");
     CloudSolrStream streamB = new CloudSolrStream(zkHost, "collection1", paramsB);
 
-    MergeStream mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
-    ParallelStream pstream = new ParallelStream(zkHost, "collection1", mstream, 2, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
+    MergeStream mstream = new MergeStream(streamA, streamB, new StreamComparator("a_i",ComparatorOrder.ASCENDING));
+    ParallelStream pstream = new ParallelStream(zkHost, "collection1", mstream, 2, new StreamComparator("a_i",ComparatorOrder.ASCENDING));
     List<Tuple> tuples = getTuples(pstream);
 
     assert(tuples.size() == 9);
@@ -907,8 +898,8 @@ public class StreamingTest extends Abstr
     paramsB = mapParams("q","id:(0 2 3 6)","fl","id,a_s,a_i","sort", "a_i desc", "partitionKeys", "a_i");
     streamB = new CloudSolrStream(zkHost, "collection1", paramsB);
 
-    mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
-    pstream = new ParallelStream(zkHost, "collection1", mstream, 2, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
+    mstream = new MergeStream(streamA, streamB, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
+    pstream = new ParallelStream(zkHost, "collection1", mstream, 2, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
     tuples = getTuples(pstream);
 
     assert(tuples.size() == 8);
@@ -943,9 +934,9 @@ public class StreamingTest extends Abstr
     Map paramsB = mapParams("q","id:(0 2 3 6)","fl","id,a_s,a_i","sort", "a_i asc", "partitionKeys", "a_i");
     CloudSolrStream streamB = new CloudSolrStream(zkHost, "collection1", paramsB);
 
-    MergeStream mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
+    MergeStream mstream = new MergeStream(streamA, streamB, new StreamComparator("a_i",ComparatorOrder.ASCENDING));
     CountStream cstream = new CountStream(mstream);
-    ParallelStream pstream = new ParallelStream(zkHost, "collection1", cstream, 2, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
+    ParallelStream pstream = new ParallelStream(zkHost, "collection1", cstream, 2, new StreamComparator("a_i",ComparatorOrder.ASCENDING));
     List<Tuple> tuples = getTuples(pstream);
 
     assert(tuples.size() == 9);



Mime
View raw message