You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by jb...@apache.org on 2015/07/23 00:02:16 UTC
svn commit: r1692330 - 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...
Author: jbernste
Date: Wed Jul 22 22:02:15 2015
New Revision: 1692330
URL: http://svn.apache.org/r1692330
Log:
SOLR-7513, SOLR-7528: Revert 5x commit
Added:
lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/ExpressibleComparator.java
- copied unchanged from r1692324, 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
- copied unchanged from r1692324, 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
- copied unchanged from r1692324, lucene/dev/branches/branch_5x/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExpressibleStream.java
Removed:
lucene/dev/branches/branch_5x/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
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=1692330&r1=1692329&r2=1692330&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 22:02:15 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.StreamComparator;
+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.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 StreamComparator(stripQuotes(sortKey), comparatorOrder);
+ comps[i] = new FieldComparator(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 StreamComparator(stripQuotes(sortKey), comparatorOrder);
+ comps[i] = new FieldComparator(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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -17,13 +17,14 @@
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;
@@ -31,10 +32,8 @@ 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;
@@ -42,6 +41,7 @@ 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(), Expressible.class);
+ Class<?> clazz = core.getResourceLoader().findClass((String)functionMapping.getValue(), ExpressibleStream.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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -22,7 +22,6 @@ 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;
@@ -32,7 +31,7 @@ import org.apache.solr.client.solrj.io.s
* Wraps multiple Comparators to provide sub-sorting.
**/
-public class MultiComp implements Comparator<Tuple>, Expressible, Serializable {
+public class MultiComp implements Comparator<Tuple>, ExpressibleComparator, Serializable {
private static final long serialVersionUID = 1;
@@ -57,9 +56,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 Expressible){
+ if(comp instanceof ExpressibleComparator){
if(sb.length() > 0){ sb.append(","); }
- sb.append(((Expressible)comp).toExpression(factory));
+ sb.append(((ExpressibleComparator)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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -1,3 +1,12 @@
+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
@@ -15,103 +24,22 @@
* 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;
-
/**
- * An equality field Comparator which compares a field of two Tuples and determines sort order.
- **/
-public class StreamComparator implements Comparator<Tuple>, Expressible, Serializable {
+ * 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;
- 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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -38,9 +38,8 @@ 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;
@@ -61,7 +60,7 @@ import org.apache.solr.common.util.Solrj
* to iterate and merge Tuples from each SolrStream.
**/
-public class CloudSolrStream extends TupleStream implements Expressible {
+public class CloudSolrStream extends TupleStream implements ExpressibleStream {
private static final long serialVersionUID = 1;
@@ -269,7 +268,7 @@ public class CloudSolrStream extends Tup
fieldName = fieldMappings.get(fieldName);
}
- comps[i] = new StreamComparator(fieldName, order.equalsIgnoreCase("asc") ? ComparatorOrder.ASCENDING : ComparatorOrder.DESCENDING);
+ comps[i] = new FieldComparator(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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -18,14 +18,15 @@
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.StreamComparator;
-import org.apache.solr.client.solrj.io.stream.expr.Expressible;
+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.stream.expr.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
@@ -37,7 +38,7 @@ import org.apache.solr.client.solrj.io.s
**/
-public class MergeStream extends TupleStream implements Expressible {
+public class MergeStream extends TupleStream implements ExpressibleStream {
private static final long serialVersionUID = 1;
@@ -53,7 +54,7 @@ public class MergeStream extends TupleSt
public MergeStream(StreamExpression expression,StreamFactory factory) throws IOException {
// grab all parameters out
- List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
+ List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
StreamExpressionNamedParameter onExpression = factory.getNamedOperand(expression, "on");
// validate expression contains only what we want.
@@ -72,7 +73,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(), StreamComparator.class);
+ this.comp = factory.constructComparator(((StreamExpressionValue)onExpression.getParameter()).getValue(), FieldComparator.class);
}
@Override
@@ -85,8 +86,8 @@ public class MergeStream extends TupleSt
expression.addParameter(streamB.toExpression(factory));
// on
- if(comp instanceof Expressible){
- expression.addParameter(new StreamExpressionNamedParameter("on",((Expressible)comp).toExpression(factory)));
+ if(comp instanceof ExpressibleComparator){
+ expression.addParameter(new StreamExpressionNamedParameter("on",((ExpressibleComparator)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=1692330&r1=1692329&r2=1692330&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 22:02:15 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.Comparator;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
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.StreamComparator;
-import org.apache.solr.client.solrj.io.stream.expr.Expressible;
+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.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 Expressible {
+public class ParallelStream extends CloudSolrStream implements ExpressibleStream {
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, Expressible.class, TupleStream.class);
+ List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.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(), StreamComparator.class);
+ Comparator<Tuple> comp = factory.constructComparator(((StreamExpressionValue)sortExpression.getParameter()).getValue(), FieldComparator.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 Expressible)){
+ if(!objectSerialize && !(tupleStream instanceof ExpressibleStream)){
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 Expressible){
- expression.addParameter(((Expressible)tupleStream).toExpression(factory));
+ if(tupleStream instanceof ExpressibleStream){
+ expression.addParameter(((ExpressibleStream)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 Expressible){
- expression.addParameter(new StreamExpressionNamedParameter("sort",((Expressible)comp).toExpression(factory)));
+ if(comp instanceof ExpressibleComparator){
+ expression.addParameter(new StreamExpressionNamedParameter("sort",((ExpressibleComparator)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 = ((Expressible) tupleStream).toExpression(streamFactory);
+ pushStream = ((ExpressibleStream) 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=1692330&r1=1692329&r2=1692330&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 22:02:15 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.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;
@@ -44,8 +44,8 @@ public class PushBackStream extends Tupl
}
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException{
- if(stream instanceof Expressible){
- return ((Expressible)stream).toExpression(factory);
+ if(stream instanceof ExpressibleStream){
+ return ((ExpressibleStream)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=1692330&r1=1692329&r2=1692330&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 22:02:15 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.StreamComparator;
-import org.apache.solr.client.solrj.io.stream.expr.Expressible;
+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.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 Expressible {
+public class RankStream extends TupleStream implements ExpressibleStream {
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, Expressible.class, TupleStream.class);
+ List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.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(), StreamComparator.class);
+ Comparator<Tuple> comp = factory.constructComparator(((StreamExpressionValue)sortExpression.getParameter()).getValue(), FieldComparator.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 Expressible){
- expression.addParameter(((Expressible)tupleStream).toExpression(factory));
+ if(tupleStream instanceof ExpressibleStream){
+ expression.addParameter(((ExpressibleStream)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 Expressible){
- expression.addParameter(new StreamExpressionNamedParameter("sort",((Expressible)comp).toExpression(factory)));
+ if(comp instanceof ExpressibleComparator){
+ expression.addParameter(new StreamExpressionNamedParameter("sort",((ExpressibleComparator)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=1692330&r1=1692329&r2=1692330&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 22:02:15 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.StreamComparator;
-import org.apache.solr.client.solrj.io.stream.expr.Expressible;
+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.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 Expressible {
+public class ReducerStream extends TupleStream implements ExpressibleStream {
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, Expressible.class, TupleStream.class);
+ List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.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(), StreamComparator.class);
+ this.comp = factory.constructComparator(((StreamExpressionValue)byExpression.getParameter()).getValue(), FieldComparator.class);
}
@Override
@@ -95,8 +95,8 @@ public class ReducerStream extends Tuple
expression.addParameter(tupleStream.toExpression(factory));
// over
- if(comp instanceof Expressible){
- expression.addParameter(new StreamExpressionNamedParameter("by",((Expressible)comp).toExpression(factory)));
+ if(comp instanceof ExpressibleComparator){
+ expression.addParameter(new StreamExpressionNamedParameter("by",((ExpressibleComparator)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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -24,10 +24,8 @@ import java.util.List;
import java.util.Locale;
import org.apache.solr.client.solrj.io.Tuple;
-import org.apache.solr.client.solrj.io.comp.StreamComparator;
-import org.apache.solr.client.solrj.io.eq.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.comp.FieldComparator;
+import org.apache.solr.client.solrj.io.comp.ExpressibleComparator;
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;
@@ -40,22 +38,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 Expressible {
+public class UniqueStream extends TupleStream implements ExpressibleStream {
private static final long serialVersionUID = 1;
private TupleStream tupleStream;
- private Equalitor<Tuple> eq;
+ private Comparator<Tuple> comp;
private transient Tuple currentTuple;
- public UniqueStream(TupleStream tupleStream, Equalitor<Tuple> eq) {
+ public UniqueStream(TupleStream tupleStream, Comparator<Tuple> comp) {
this.tupleStream = tupleStream;
- this.eq = eq;
+ this.comp = comp;
}
public UniqueStream(StreamExpression expression,StreamFactory factory) throws IOException {
// grab all parameters out
- List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
+ List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
StreamExpressionNamedParameter overExpression = factory.getNamedOperand(expression, "over");
// validate expression contains only what we want.
@@ -73,7 +71,7 @@ public class UniqueStream extends TupleS
}
// Uniqueness is always done over equality, so always use an EqualTo comparator
- this.eq = factory.constructEqualitor(((StreamExpressionValue)overExpression.getParameter()).getValue(), StreamEqualitor.class);
+ this.comp = factory.constructComparator(((StreamExpressionValue)overExpression.getParameter()).getValue(), FieldComparator.class);
}
@Override
@@ -82,24 +80,28 @@ public class UniqueStream extends TupleS
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// streams
- if(tupleStream instanceof Expressible){
- expression.addParameter(((Expressible)tupleStream).toExpression(factory));
+ if(tupleStream instanceof ExpressibleStream){
+ expression.addParameter(((ExpressibleStream)tupleStream).toExpression(factory));
}
else{
throw new IOException("This UniqueStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
// over
- if(eq instanceof Expressible){
- expression.addParameter(new StreamExpressionNamedParameter("over",((Expressible)eq).toExpression(factory)));
+ if(comp instanceof ExpressibleComparator){
+ expression.addParameter(new StreamExpressionNamedParameter("over",((ExpressibleComparator)comp).toExpression(factory)));
}
else{
- throw new IOException("This UniqueStream contains a non-expressible equalitor - it cannot be converted to an expression");
+ throw new IOException("This UniqueStream contains a non-expressible comparator - 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);
}
@@ -129,7 +131,8 @@ public class UniqueStream extends TupleS
return tuple;
} else {
while(true) {
- if(eq.test(currentTuple, tuple)){
+ int i = comp.compare(currentTuple, tuple);
+ if(i == 0) {
//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=1692330&r1=1692329&r2=1692330&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 22:02:15 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.eq.Equalitor;
-import org.apache.solr.client.solrj.io.eq.MultiEqualitor;
+import org.apache.solr.client.solrj.io.stream.ExpressibleStream;
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(Expressible.class.isAssignableFrom(clazz) && TupleStream.class.isAssignableFrom(clazz)){
+ if(ExpressibleStream.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,36 +201,6 @@ 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=1692330&r1=1692329&r2=1692330&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 22:02:15 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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -23,12 +23,11 @@ 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 Expressible, Serializable {
+public class CountStream extends TupleStream implements ExpressibleStream, Serializable {
private TupleStream stream;
private int count;
@@ -38,7 +37,7 @@ public class CountStream extends TupleSt
}
public CountStream(StreamExpression expression, StreamFactory factory) throws IOException{
- List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);
+ List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, ExpressibleStream.class, TupleStream.class);
// validate expression contains only what we want.
if(expression.getParameters().size() != streamExpressions.size()){
@@ -58,8 +57,8 @@ public class CountStream extends TupleSt
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// stream
- if(stream instanceof Expressible){
- expression.addParameter(((Expressible)stream).toExpression(factory));
+ if(stream instanceof ExpressibleStream){
+ expression.addParameter(((ExpressibleStream)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=1692330&r1=1692329&r2=1692330&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 22:02:15 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\")");
+ 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\")");
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\")");
+ 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\")");
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, a_i\")");
+ 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\")");
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, a_i\")");
+ 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\")");
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\"),"
+ + "over=\"a_f desc\"),"
+ "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\"),"
+ + "over=\"a_f asc\"),"
+ "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\"), 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 asc\"), 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=1692330&r1=1692329&r2=1692330&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 22:02:15 2015
@@ -19,19 +19,25 @@ package org.apache.solr.client.solrj.io.
import java.io.File;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
+import java.io.Serializable;
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.comp.StreamComparator;
-import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
+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.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.metrics.Bucket;
import org.apache.solr.client.solrj.io.stream.metrics.CountMetric;
@@ -49,6 +55,9 @@ 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.
@@ -139,7 +148,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 StreamEqualitor("a_f"));
+ UniqueStream ustream = new UniqueStream(stream, new FieldComparator("a_f",ComparatorOrder.ASCENDING));
List<Tuple> tuples = getTuples(ustream);
assert(tuples.size() == 4);
assertOrder(tuples, 0,1,3,4);
@@ -188,7 +197,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 StreamComparator("a_s",ComparatorOrder.ASCENDING));
+ ParallelStream pstream = new ParallelStream(zkHost, "collection1", stream, 2, new FieldComparator("a_s",ComparatorOrder.ASCENDING));
List<Tuple> tuples = getTuples(pstream);
@@ -221,8 +230,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 StreamEqualitor("a_f"));
- ParallelStream pstream = new ParallelStream(zkHost, "collection1", ustream, 2, new StreamComparator("a_f",ComparatorOrder.ASCENDING));
+ 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));
List<Tuple> tuples = getTuples(pstream);
assert(tuples.size() == 5);
assertOrder(tuples, 0,1,3,4,6);
@@ -255,7 +264,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 StreamComparator("a_i",ComparatorOrder.DESCENDING));
+ RankStream rstream = new RankStream(stream, 3, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
List<Tuple> tuples = getTuples(rstream);
@@ -287,8 +296,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 StreamComparator("a_i",ComparatorOrder.DESCENDING));
- ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
+ 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));
List<Tuple> tuples = getTuples(pstream);
assert(tuples.size() == 10);
@@ -354,7 +363,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 StreamComparator("a_s",ComparatorOrder.ASCENDING));
+ ReducerStream rstream = new ReducerStream(stream, new FieldComparator("a_s",ComparatorOrder.ASCENDING));
List<Tuple> tuples = getTuples(rstream);
@@ -401,7 +410,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 StreamComparator("a_s", ComparatorOrder.ASCENDING));
+ ReducerStream rstream = new ReducerStream(stream, new FieldComparator("a_s", ComparatorOrder.ASCENDING));
List<Tuple> tuples = getTuples(rstream);
@@ -432,8 +441,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 StreamComparator("a_s",ComparatorOrder.ASCENDING));
- ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_s",ComparatorOrder.ASCENDING));
+ 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));
List<Tuple> tuples = getTuples(pstream);
@@ -456,8 +465,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 StreamComparator("a_s",ComparatorOrder.DESCENDING));
- pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_s",ComparatorOrder.DESCENDING));
+ rstream = new ReducerStream(stream, new FieldComparator("a_s",ComparatorOrder.DESCENDING));
+ pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new FieldComparator("a_s",ComparatorOrder.DESCENDING));
tuples = getTuples(pstream);
@@ -639,7 +648,7 @@ public class StreamingTest extends Abstr
new CountMetric()};
RollupStream rollupStream = new RollupStream(stream, buckets, metrics);
- ParallelStream parallelStream = new ParallelStream(zkHost, "collection1", rollupStream, 2, new StreamComparator("a_s", ComparatorOrder.ASCENDING));
+ ParallelStream parallelStream = new ParallelStream(zkHost, "collection1", rollupStream, 2, new FieldComparator("a_s", ComparatorOrder.ASCENDING));
List<Tuple> tuples = getTuples(parallelStream);
assert(tuples.size() == 3);
@@ -739,8 +748,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 StreamComparator("a_s", ComparatorOrder.ASCENDING));
- ParallelStream pstream = new ParallelStream(zkHost, "collection1", rstream, 2, new StreamComparator("a_s", ComparatorOrder.ASCENDING));
+ 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));
List<Tuple> tuples = getTuples(pstream);
assert(tuples.size() == 0);
@@ -809,7 +818,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 StreamComparator("a_i",ComparatorOrder.ASCENDING));
+ MergeStream mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
List<Tuple> tuples = getTuples(mstream);
assert(tuples.size() == 5);
@@ -822,7 +831,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 StreamComparator("a_i",ComparatorOrder.DESCENDING));
+ mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
tuples = getTuples(mstream);
assert(tuples.size() == 5);
@@ -836,7 +845,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 StreamComparator("a_f",ComparatorOrder.ASCENDING),new StreamComparator("a_i",ComparatorOrder.ASCENDING)));
+ mstream = new MergeStream(streamA, streamB, new MultiComp(new FieldComparator("a_f",ComparatorOrder.ASCENDING),new FieldComparator("a_i",ComparatorOrder.ASCENDING)));
tuples = getTuples(mstream);
assert(tuples.size() == 5);
@@ -848,7 +857,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 StreamComparator("a_f",ComparatorOrder.ASCENDING),new StreamComparator("a_i",ComparatorOrder.DESCENDING)));
+ mstream = new MergeStream(streamA, streamB, new MultiComp(new FieldComparator("a_f",ComparatorOrder.ASCENDING),new FieldComparator("a_i",ComparatorOrder.DESCENDING)));
tuples = getTuples(mstream);
assert(tuples.size() == 5);
@@ -884,8 +893,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 StreamComparator("a_i",ComparatorOrder.ASCENDING));
- ParallelStream pstream = new ParallelStream(zkHost, "collection1", mstream, 2, new StreamComparator("a_i",ComparatorOrder.ASCENDING));
+ 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));
List<Tuple> tuples = getTuples(pstream);
assert(tuples.size() == 9);
@@ -898,8 +907,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 StreamComparator("a_i",ComparatorOrder.DESCENDING));
- pstream = new ParallelStream(zkHost, "collection1", mstream, 2, new StreamComparator("a_i",ComparatorOrder.DESCENDING));
+ mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
+ pstream = new ParallelStream(zkHost, "collection1", mstream, 2, new FieldComparator("a_i",ComparatorOrder.DESCENDING));
tuples = getTuples(pstream);
assert(tuples.size() == 8);
@@ -934,9 +943,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 StreamComparator("a_i",ComparatorOrder.ASCENDING));
+ MergeStream mstream = new MergeStream(streamA, streamB, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
CountStream cstream = new CountStream(mstream);
- ParallelStream pstream = new ParallelStream(zkHost, "collection1", cstream, 2, new StreamComparator("a_i",ComparatorOrder.ASCENDING));
+ ParallelStream pstream = new ParallelStream(zkHost, "collection1", cstream, 2, new FieldComparator("a_i",ComparatorOrder.ASCENDING));
List<Tuple> tuples = getTuples(pstream);
assert(tuples.size() == 9);