Apache
/
lucene
mirror of https://github.com/apache/lucene.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SOLR-9009: Adds ability to get an Explanation of a Streaming Expression

This commit is contained in:
Dennis Gove 2016-04-18 21:34:36 -04:00
parent 2342290816
commit 2e95a54a52
47 changed files with 1678 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
solr/CHANGES.txt
View File

@ -98,6 +98,7 @@ New Features
* SOLR-8349: Allow sharing of large in memory data structures across cores (Gus Heck, noble)
* SOLR-9009: Adds ability to get an Explanation of a Streaming Expression (Dennis Gove)
Bug Fixes
----------------------

72
solr/core/src/java/org/apache/solr/handler/SQLHandler.java
View File

@ -49,7 +49,10 @@ 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.ExceptionStream;
import org.apache.solr.client.solrj.io.stream.UniqueStream;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.metrics.*;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ZkStateReader;
@ -1290,6 +1293,19 @@ public class SQLHandler extends RequestHandlerBase implements SolrCoreAware , Pe
stream.setStreamContext(context);
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName("SQL LIMIT")
.withExpression("--non-expressible--")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR);
}
public Tuple read() throws IOException {
++count;
if(count > limit) {
@ -1349,6 +1365,19 @@ public class SQLHandler extends RequestHandlerBase implements SolrCoreAware , Pe
return children;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName("SQL HAVING")
.withExpression("--non-expressible--")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR);
}
public void setStreamContext(StreamContext context) {
stream.setStreamContext(context);
}
@ -1386,6 +1415,16 @@ public class SQLHandler extends RequestHandlerBase implements SolrCoreAware , Pe
this.catalogs.add(this.zkHost);
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("SQL CATALOG")
.withExpression("--non-expressible--")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR);
}
public Tuple read() throws IOException {
Map<String, String> fields = new HashMap<>();
if (this.currentIndex < this.catalogs.size()) {
@ -1426,6 +1465,16 @@ public class SQLHandler extends RequestHandlerBase implements SolrCoreAware , Pe
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("SQL SCHEMA")
.withExpression("--non-expressible--")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR);
}
public Tuple read() throws IOException {
Map<String, String> fields = new HashMap<>();
fields.put("EOF", "true");
@ -1471,6 +1520,16 @@ public class SQLHandler extends RequestHandlerBase implements SolrCoreAware , Pe
Collections.sort(this.tables);
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("SQL TABLE")
.withExpression("--non-expressible--")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR);
}
public Tuple read() throws IOException {
Map<String, String> fields = new HashMap<>();
if (this.currentIndex < this.tables.size()) {
@ -1518,6 +1577,19 @@ public class SQLHandler extends RequestHandlerBase implements SolrCoreAware , Pe
this.stream.open();
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName("SQL METADATA")
.withExpression("--non-expressible--")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR);
}
// Return a metadata tuple as the first tuple and then pass through to the underlying stream.
public Tuple read() throws IOException {
if(firstTuple) {

49
solr/core/src/java/org/apache/solr/handler/StreamHandler.java
View File

@ -35,8 +35,11 @@ import org.apache.solr.client.solrj.io.ops.DistinctOperation;
import org.apache.solr.client.solrj.io.ops.GroupOperation;
import org.apache.solr.client.solrj.io.ops.ReplaceOperation;
import org.apache.solr.client.solrj.io.stream.*;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.metrics.CountMetric;
import org.apache.solr.client.solrj.io.stream.metrics.MaxMetric;
import org.apache.solr.client.solrj.io.stream.metrics.MeanMetric;
@ -191,6 +194,12 @@ public class StreamHandler extends RequestHandlerBase implements SolrCoreAware,
context.setSolrClientCache(clientCache);
context.put("core", this.coreName);
tupleStream.setStreamContext(context);
// if asking for explanation then go get it
if(params.getBool("explain", false)){
rsp.add("explanation", tupleStream.toExplanation(this.streamFactory));
}
if(tupleStream instanceof DaemonStream) {
DaemonStream daemonStream = (DaemonStream)tupleStream;
if(daemons.containsKey(daemonStream.getId())) {
@ -272,6 +281,16 @@ public class StreamHandler extends RequestHandlerBase implements SolrCoreAware,
return null;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("error")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression("--non-expressible--");
}
public Tuple read() {
String msg = e.getMessage();
@ -312,6 +331,16 @@ public class StreamHandler extends RequestHandlerBase implements SolrCoreAware,
return null;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("daemon-collection")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression("--non-expressible--");
}
public Tuple read() {
if(it.hasNext()) {
return it.next().getInfo();
@ -347,6 +376,16 @@ public class StreamHandler extends RequestHandlerBase implements SolrCoreAware,
return null;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("daemon-response")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression("--non-expressible--");
}
public Tuple read() {
if (sendEOF) {
Map m = new HashMap();
@ -391,6 +430,16 @@ public class StreamHandler extends RequestHandlerBase implements SolrCoreAware,
return this.tupleStream.children();
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("timer")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression("--non-expressible--");
}
public Tuple read() throws IOException {
Tuple tuple = this.tupleStream.read();
if(tuple.EOF) {

7
solr/core/src/java/org/apache/solr/response/TextResponseWriter.java
View File

@ -30,6 +30,7 @@ import org.apache.lucene.index.IndexableField;
import org.apache.lucene.util.BytesRef;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.TupleStream;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.common.EnumFieldValue;
import org.apache.solr.common.SolrDocument;
import org.apache.solr.common.SolrDocumentList;
@ -165,6 +166,8 @@ public abstract class TextResponseWriter {
writeNamedList(name, (NamedList)val);
} else if (val instanceof TupleStream) {
writeTupleStream((TupleStream) val);
} else if (val instanceof Explanation){
writeExplanation((Explanation) val);
} else if (val instanceof Path) {
writeStr(name, ((Path) val).toAbsolutePath().toString(), true);
} else if (val instanceof Iterable) {
@ -317,6 +320,10 @@ public abstract class TextResponseWriter {
tupleStream.close();
}
public void writeExplanation(Explanation explanation) throws IOException {
writeMap(null, explanation.toMap(), false, true);
}
/** if this form of the method is called, val is the Java string form of a double */
public abstract void writeDouble(String name, String val) throws IOException;

13
solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/FieldComparator.java
View File

@ -16,9 +16,13 @@
*/
package org.apache.solr.client.solrj.io.comp;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
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;
@ -29,6 +33,7 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class FieldComparator implements StreamComparator {
private static final long serialVersionUID = 1;
private UUID comparatorNodeId = UUID.randomUUID();
private String leftFieldName;
private String rightFieldName;
@ -85,6 +90,14 @@ public class FieldComparator implements StreamComparator {
return new StreamExpressionValue(sb.toString());
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(comparatorNodeId.toString())
.withExpressionType(ExpressionType.SORTER)
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).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

12
solr/solrj/src/java/org/apache/solr/client/solrj/io/comp/MultipleFieldComparator.java
View File

@ -19,12 +19,15 @@ package org.apache.solr.client.solrj.io.comp;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
/**
@ -34,6 +37,7 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class MultipleFieldComparator implements StreamComparator {
private static final long serialVersionUID = 1;
private UUID comparatorNodeId = UUID.randomUUID();
private StreamComparator[] comps;
@ -92,6 +96,14 @@ public class MultipleFieldComparator implements StreamComparator {
return false;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(comparatorNodeId.toString())
.withExpressionType(ExpressionType.SORTER)
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
@Override
public MultipleFieldComparator copyAliased(Map<String,String> aliases){
StreamComparator[] aliasedComps = new StreamComparator[comps.length];

17
solr/solrj/src/java/org/apache/solr/client/solrj/io/eq/FieldEqualitor.java
View File

@ -16,15 +16,15 @@
*/
package org.apache.solr.client.solrj.io.eq;
import java.io.Serializable;
import java.util.Comparator;
import java.io.IOException;
import java.util.UUID;
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.MultipleFieldComparator;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
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;
@ -35,6 +35,7 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class FieldEqualitor implements StreamEqualitor {
private static final long serialVersionUID = 1;
private UUID equalitorNodeId = UUID.randomUUID();
private String leftFieldName;
private String rightFieldName;
@ -64,6 +65,14 @@ public class FieldEqualitor implements StreamEqualitor {
return new StreamExpressionValue(sb.toString());
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(equalitorNodeId.toString())
.withExpressionType(ExpressionType.EQUALITOR)
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
public boolean test(Tuple leftTuple, Tuple rightTuple) {
Comparable leftComp = (Comparable)leftTuple.get(leftFieldName);

12
solr/solrj/src/java/org/apache/solr/client/solrj/io/eq/MultipleFieldEqualitor.java
View File

@ -17,14 +17,17 @@
package org.apache.solr.client.solrj.io.eq;
import java.io.IOException;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.MultipleFieldComparator;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
/**
@ -34,6 +37,7 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class MultipleFieldEqualitor implements StreamEqualitor {
private static final long serialVersionUID = 1;
private UUID equalitorNodeId = UUID.randomUUID();
private StreamEqualitor[] eqs;
@ -61,6 +65,14 @@ public class MultipleFieldEqualitor implements StreamEqualitor {
return new StreamExpressionValue(sb.toString());
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(equalitorNodeId.toString())
.withExpressionType(ExpressionType.EQUALITOR)
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
@Override
public boolean isDerivedFrom(StreamEqualitor base){
if(null == base){ return false; }

60
solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/GatherNodesStream.java
View File

@ -29,6 +29,7 @@ import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.io.eq.MultipleFieldEqualitor;
import org.apache.solr.client.solrj.io.stream.*;
@ -36,12 +37,15 @@ import org.apache.solr.client.solrj.io.stream.metrics.*;
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.FieldEqualitor;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.SolrjNamedThreadFactory;
@ -50,7 +54,7 @@ public class GatherNodesStream extends TupleStream implements Expressible {
private String zkHost;
private String collection;
private StreamContext streamContext;
private Map queryParams;
private Map<String,String> queryParams;
private String traverseFrom;
private String traverseTo;
private String gather;
@ -250,21 +254,30 @@ public class GatherNodesStream extends TupleStream implements Expressible {
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// collection
expression.addParameter(collection);
if(includeStreams){
if(tupleStream instanceof Expressible){
expression.addParameter(((Expressible)tupleStream).toExpression(factory));
}
else{
throw new IOException("This GatherNodesStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
}
Set<Map.Entry> entries = queryParams.entrySet();
Set<Map.Entry<String,String>> entries = queryParams.entrySet();
// parameters
for(Map.Entry param : entries){
String value = param.getValue().toString();
@ -301,6 +314,37 @@ public class GatherNodesStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.GRAPH_SOURCE);
explanation.setExpression(toExpression(factory).toString());
// one child is a stream
explanation.addChild(tupleStream.toExplanation(factory));
// one child is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName("solr (graph)");
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
child.setExpression(queryParams.entrySet().stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), e.getValue())).collect(Collectors.joining(",")));
explanation.addChild(child);
if(null != metrics){
for(Metric metric : metrics){
explanation.addHelper(metric.toExplanation(factory));
}
}
return explanation;
}
public void setStreamContext(StreamContext context) {
this.traversal = (Traversal) context.get("traversal");
if (traversal == null) {
@ -576,5 +620,15 @@ public class GatherNodesStream extends TupleStream implements Expressible {
return new Tuple(map);
}
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("non-expressible")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_SOURCE)
.withExpression("non-expressible");
}
}
}

29
solr/solrj/src/java/org/apache/solr/client/solrj/io/graph/ShortestPathStream.java
View File

@ -30,18 +30,22 @@ import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.io.eq.MultipleFieldEqualitor;
import org.apache.solr.client.solrj.io.stream.*;
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.FieldEqualitor;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.SolrjNamedThreadFactory;
@ -61,7 +65,7 @@ public class ShortestPathStream extends TupleStream implements Expressible {
private boolean found;
private StreamContext streamContext;
private int threads;
private Map queryParams;
private Map<String,String> queryParams;
public ShortestPathStream(String zkHost,
String collection,
@ -221,7 +225,7 @@ public class ShortestPathStream extends TupleStream implements Expressible {
// collection
expression.addParameter(collection);
Set<Map.Entry> entries = queryParams.entrySet();
Set<Map.Entry<String,String>> entries = queryParams.entrySet();
// parameters
for(Map.Entry param : entries){
String value = param.getValue().toString();
@ -244,6 +248,27 @@ public class ShortestPathStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.GRAPH_SOURCE);
explanation.setExpression(toExpression(factory).toString());
// child is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName("solr (graph)");
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
child.setExpression(queryParams.entrySet().stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), e.getValue())).collect(Collectors.joining(",")));
explanation.addChild(child);
return explanation;
}
public void setStreamContext(StreamContext context) {
this.streamContext = context;
}

13
solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/ConcatOperation.java
View File

@ -18,8 +18,11 @@ package org.apache.solr.client.solrj.io.ops;
import java.io.IOException;
import java.util.Locale;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
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;
@ -33,6 +36,7 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class ConcatOperation implements StreamOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
private String[] fields;
private String as;
@ -94,4 +98,13 @@ public class ConcatOperation implements StreamOperation {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

39
solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/DistinctOperation.java
View File

@ -16,32 +16,20 @@
*/
package org.apache.solr.client.solrj.io.ops;
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.StreamComparator;
import org.apache.solr.client.solrj.io.eq.FieldEqualitor;
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.StreamExpressionParameter;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import java.io.IOException;
import java.io.Serializable;
import java.util.Comparator;
import java.util.List;
import java.util.ArrayList;
import java.util.Locale;
import java.util.Map;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.PriorityQueue;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
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 DistinctOperation implements ReduceOperation {
private static final long serialVersionUID = 1L;
private UUID operationNodeId = UUID.randomUUID();
private Tuple current;
public DistinctOperation(StreamExpression expression, StreamFactory factory) throws IOException {
@ -60,6 +48,15 @@ public class DistinctOperation implements ReduceOperation {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
public Tuple reduce() {
// Return the tuple after setting current to null. This will ensure the next call to
// operate stores that tuple

42
solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/GroupOperation.java
View File

@ -16,29 +16,34 @@
*/
package org.apache.solr.client.solrj.io.ops;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.UUID;
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.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import java.io.IOException;
import java.io.Serializable;
import java.util.Comparator;
import java.util.List;
import java.util.ArrayList;
import java.util.Locale;
import java.util.Map;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.PriorityQueue;
public class GroupOperation implements ReduceOperation {
private static final long serialVersionUID = 1L;
private UUID operationNodeId = UUID.randomUUID();
private PriorityQueue<Tuple> priorityQueue;
private Comparator comp;
private StreamComparator streamComparator;
@ -86,6 +91,18 @@ public class GroupOperation implements ReduceOperation {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString())
.withHelpers(new Explanation[]{
streamComparator.toExplanation(factory)
});
}
public Tuple reduce() {
Map map = new HashMap();
List<Map> list = new ArrayList();
@ -122,6 +139,7 @@ public class GroupOperation implements ReduceOperation {
}
public int compare(Tuple t1, Tuple t2) {
// Couldn't this be comp.compare(t2,t1) ?
return comp.compare(t1, t2)*(-1);
}
}

6
solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/ReplaceOperation.java
View File

@ -20,6 +20,7 @@ import java.io.IOException;
import java.util.Locale;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
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;
@ -80,4 +81,9 @@ public class ReplaceOperation implements StreamOperation {
return replacer.toExpression(factory);
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException{
return replacer.toExplanation(factory);
}
}

12
solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/ReplaceWithFieldOperation.java
View File

@ -18,13 +18,16 @@ package org.apache.solr.client.solrj.io.ops;
import java.io.IOException;
import java.util.Locale;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
/**
* Implementation of replace(...., withField=fieldName)
@ -33,6 +36,7 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class ReplaceWithFieldOperation implements StreamOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
private boolean wasBuiltWithFieldName;
private String originalFieldName;
@ -108,4 +112,12 @@ public class ReplaceWithFieldOperation implements StreamOperation {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

12
solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/ReplaceWithValueOperation.java
View File

@ -18,13 +18,16 @@ package org.apache.solr.client.solrj.io.ops;
import java.io.IOException;
import java.util.Locale;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
/**
* Implementation of replace(...., withValue="some value")
@ -33,6 +36,7 @@ import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public class ReplaceWithValueOperation implements StreamOperation {
private static final long serialVersionUID = 1;
private UUID operationNodeId = UUID.randomUUID();
private boolean wasBuiltWithFieldName;
private String fieldName;
@ -113,4 +117,12 @@ public class ReplaceWithValueOperation implements StreamOperation {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(operationNodeId.toString())
.withExpressionType(ExpressionType.OPERATION)
.withFunctionName(factory.getFunctionName(getClass()))
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString());
}
}

3
solr/solrj/src/java/org/apache/solr/client/solrj/io/ops/StreamOperation.java
View File

@ -16,13 +16,10 @@
*/
package org.apache.solr.client.solrj.io.ops;
import java.io.IOException;
import java.io.Serializable;
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.StreamFactory;
/**
* Interface for any operation one can perform on a tuple in a TupleStream

32
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/CloudSolrStream.java
View File

@ -26,12 +26,13 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.CloudSolrClient.Builder;
@ -41,10 +42,12 @@ 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.MultipleFieldComparator;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.common.cloud.ClusterState;
@ -154,7 +157,7 @@ public class CloudSolrStream extends TupleStream implements Expressible {
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
public StreamExpression toExpression(StreamFactory factory) throws IOException {
// functionName(collectionName, param1, param2, ..., paramN, sort="comp", [aliases="field=alias,..."])
// function name
@ -194,6 +197,29 @@ public class CloudSolrStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_SOURCE);
explanation.setExpression(toExpression(factory).toString());
// child is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName(String.format(Locale.ROOT, "solr (%s)", collection));
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
if(null != params){
child.setExpression(params.entrySet().stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), e.getValue())).collect(Collectors.joining(",")));
}
explanation.addChild(child);
return explanation;
}
private void init(String collectionName, String zkHost, Map params) throws IOException {
this.zkHost = zkHost;
this.collection = collectionName;

33
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ComplementStream.java
View File

@ -22,12 +22,13 @@ 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.StreamComparator;
import org.apache.solr.client.solrj.io.eq.FieldEqualitor;
import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
import org.apache.solr.client.solrj.io.ops.DistinctOperation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -89,9 +90,14 @@ public class ComplementStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// streams
if(streamA instanceof Expressible){
expression.addParameter(((Expressible)streamA).toExpression(factory));
@ -106,6 +112,11 @@ public class ComplementStream extends TupleStream implements Expressible {
else{
throw new IOException("This IntersectStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
expression.addParameter("<stream>");
}
// on
expression.addParameter(new StreamExpressionNamedParameter("on",eq.toExpression(factory)));
@ -113,13 +124,29 @@ public class ComplementStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
streamA.toExplanation(factory),
originalStreamB.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
.withHelper(eq.toExplanation(factory))
;
}
public void setStreamContext(StreamContext context) {
this.streamA.setStreamContext(context);
this.streamB.setStreamContext(context);
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
List<TupleStream> l = new ArrayList<TupleStream>();
l.add(streamA);
l.add(streamB);
return l;

45
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/DaemonStream.java
View File

@ -15,23 +15,26 @@
* limitations under the License.
*/
package org.apache.solr.client.solrj.io.stream;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.lang.invoke.MethodHandles;
import java.util.Locale;
import java.util.Map;
import java.util.HashMap;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -101,15 +104,24 @@ public class DaemonStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// streams
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");
}
}
else{
expression.addParameter("<stream>");
}
expression.addParameter(new StreamExpressionNamedParameter("id", id));
expression.addParameter(new StreamExpressionNamedParameter("runInterval", Long.toString(runInterval)));
@ -118,6 +130,19 @@ public class DaemonStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[] {
tupleStream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString());
}
public int remainingCapacity() {
return this.queue.remainingCapacity();
}

14
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ExceptionStream.java
View File

@ -24,6 +24,10 @@ import java.util.Map;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.common.SolrException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -71,6 +75,16 @@ public class ExceptionStream extends TupleStream {
}
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("non-expressible")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_SOURCE)
.withExpression("non-expressible");
}
public StreamComparator getStreamSort() {
return this.stream.getStreamSort();
}

34
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/FacetStream.java
View File

@ -17,13 +17,14 @@
package org.apache.solr.client.solrj.io.stream;
import java.io.IOException;
import java.util.HashMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Collections;
import java.util.Map.Entry;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.CloudSolrClient.Builder;
@ -33,7 +34,10 @@ 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.MultipleFieldComparator;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -58,7 +62,7 @@ public class FacetStream extends TupleStream implements Expressible {
private Metric[] metrics;
private int bucketSizeLimit;
private FieldComparator[] bucketSorts;
private List<Tuple> tuples = new ArrayList();
private List<Tuple> tuples = new ArrayList<Tuple>();
private int index;
private String zkHost;
private Map<String, String> props;
@ -267,6 +271,30 @@ public class FacetStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_SOURCE);
explanation.setExpression(toExpression(factory).toString());
// child is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName(String.format(Locale.ROOT, "solr (%s)", collection));
// TODO: fix this so we know the # of workers - check with Joel about a Topic's ability to be in a
// parallel stream.
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
child.setExpression(props.entrySet().stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), e.getValue())).collect(Collectors.joining(",")));
explanation.addChild(child);
return explanation;
}
public void setStreamContext(StreamContext context) {
cache = context.getSolrClientCache();

29
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/HashJoinStream.java
View File

@ -24,7 +24,10 @@ 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.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -100,9 +103,14 @@ public class HashJoinStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// streams
if(hashStream instanceof Expressible && fullStream instanceof Expressible){
expression.addParameter(((Expressible)fullStream).toExpression(factory));
@ -111,6 +119,11 @@ public class HashJoinStream extends TupleStream implements Expressible {
else{
throw new IOException("This HashJoinStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
expression.addParameter("hashed=<stream>");
}
// on
StringBuilder sb = new StringBuilder();
@ -123,13 +136,27 @@ public class HashJoinStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
fullStream.toExplanation(factory),
hashStream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString());
}
public void setStreamContext(StreamContext context) {
this.hashStream.setStreamContext(context);
this.fullStream.setStreamContext(context);
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
List<TupleStream> l = new ArrayList<TupleStream>();
l.add(hashStream);
l.add(fullStream);
return l;

32
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/IntersectStream.java
View File

@ -22,12 +22,13 @@ 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.StreamComparator;
import org.apache.solr.client.solrj.io.eq.FieldEqualitor;
import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
import org.apache.solr.client.solrj.io.ops.DistinctOperation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -89,9 +90,14 @@ public class IntersectStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// streams
if(streamA instanceof Expressible){
expression.addParameter(((Expressible)streamA).toExpression(factory));
@ -106,6 +112,11 @@ public class IntersectStream extends TupleStream implements Expressible {
else{
throw new IOException("This IntersectStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
expression.addParameter("<stream>");
}
// on
expression.addParameter(new StreamExpressionNamedParameter("on",eq.toExpression(factory)));
@ -113,13 +124,28 @@ public class IntersectStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
streamA.toExplanation(factory),
originalStreamB.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
.withHelper(eq.toExplanation(factory));
}
public void setStreamContext(StreamContext context) {
this.streamA.setStreamContext(context);
this.streamB.setStreamContext(context);
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
List<TupleStream> l = new ArrayList<TupleStream>();
l.add(streamA);
l.add(streamB);
return l;

37
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/JDBCStream.java
View File

@ -34,7 +34,10 @@ import java.util.Properties;
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.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -348,6 +351,40 @@ public class JDBCStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_SOURCE);
StreamExpression expression = (StreamExpression)toExpression(factory);
explanation.setExpression(expression.toString());
String driverClassName = this.driverClassName;
if(null == driverClassName){
try{
driverClassName = DriverManager.getDriver(connectionUrl).getClass().getName();
}
catch(Exception e){
driverClassName = String.format(Locale.ROOT, "Failed to find driver for connectionUrl='%s'", connectionUrl);
}
}
// child is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName("jdbc-source");
child.setImplementingClass(driverClassName);
child.setExpressionType(ExpressionType.DATASTORE);
child.setExpression(sqlQuery);
explanation.addChild(child);
return explanation;
}
@Override
public List<TupleStream> children() {
return new ArrayList<TupleStream>();

29
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/JoinStream.java
View File

@ -26,7 +26,10 @@ 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.FieldEqualitor;
import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -88,13 +91,22 @@ public abstract class JoinStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// streams
for (PushBackStream stream : streams) {
if(includeStreams){
expression.addParameter(stream.toExpression(factory));
}
else{
expression.addParameter("<stream>");
}
}
// on
if (eq instanceof Expressible) {
@ -107,6 +119,23 @@ public abstract class JoinStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_DECORATOR);
explanation.setExpression(toExpression(factory, false).toString());
explanation.addHelper(eq.toExplanation(factory));
for(TupleStream stream : streams){
explanation.addChild(stream.toExplanation(factory));
}
return explanation;
}
public void setStreamContext(StreamContext context) {
for (PushBackStream stream : streams) {
stream.setStreamContext(context);

31
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/MergeStream.java
View File

@ -24,7 +24,10 @@ 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.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -96,13 +99,22 @@ public class MergeStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// streams
for(PushBackStream stream : streams){
if(includeStreams){
expression.addParameter(stream.toExpression(factory));
}
else{
expression.addParameter("<stream>");
}
}
// on
expression.addParameter(new StreamExpressionNamedParameter("on",comp.toExpression(factory)));
@ -110,6 +122,23 @@ public class MergeStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_DECORATOR);
explanation.setExpression(toExpression(factory, false).toString());
explanation.addHelper(comp.toExplanation(factory));
for(PushBackStream stream : streams){
explanation.addChild(stream.toExplanation(factory));
}
return explanation;
}
public void setStreamContext(StreamContext context) {
for(PushBackStream stream : streams){
stream.setStreamContext(context);
@ -117,7 +146,7 @@ public class MergeStream extends TupleStream implements Expressible {
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
List<TupleStream> l = new ArrayList<TupleStream>();
for(PushBackStream stream : streams){
l.add(stream);
}

31
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ParallelStream.java
View File

@ -30,16 +30,20 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
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.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
@ -164,7 +168,10 @@ public class ParallelStream extends CloudSolrStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
@ -174,13 +181,17 @@ public class ParallelStream extends CloudSolrStream implements Expressible {
// workers
expression.addParameter(new StreamExpressionNamedParameter("workers", Integer.toString(workers)));
// stream
if(includeStreams){
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");
}
}
else{
expression.addParameter("<stream>");
}
// sort
expression.addParameter(new StreamExpressionNamedParameter("sort",comp.toExpression(factory)));
@ -191,6 +202,24 @@ public class ParallelStream extends CloudSolrStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_DECORATOR);
explanation.setExpression(toExpression(factory, false).toString());
// add a child for each worker
for(int idx = 0; idx < workers; ++idx){
explanation.addChild(tupleStream.toExplanation(factory));
}
return explanation;
}
public List<TupleStream> children() {
List l = new ArrayList();
l.add(tupleStream);

7
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/PushBackStream.java
View File

@ -22,6 +22,7 @@ import java.util.List;
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.Explanation;
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;
@ -51,12 +52,16 @@ public class PushBackStream extends TupleStream implements Expressible {
throw new IOException("This PushBackStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
public Explanation toExplanation(StreamFactory factory) throws IOException{
return stream.toExplanation(factory);
}
public void setStreamContext(StreamContext context) {
this.stream.setStreamContext(context);
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
List<TupleStream> l = new ArrayList<TupleStream>();
l.add(stream);
return l;
}

26
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RandomStream.java
View File

@ -24,6 +24,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
import java.util.Random;
import java.util.Iterator;
@ -31,12 +32,15 @@ import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.io.SolrClientCache;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.request.QueryRequest;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.SolrDocument;
@ -132,6 +136,28 @@ public class RandomStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_SOURCE);
explanation.setExpression(toExpression(factory).toString());
// child is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName(String.format(Locale.ROOT, "solr (%s)", collection));
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
if(null != props){
child.setExpression(props.entrySet().stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), e.getValue())).collect(Collectors.joining(",")));
}
explanation.addChild(child);
return explanation;
}
public void setStreamContext(StreamContext context) {
cache = context.getSolrClientCache();

33
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RankStream.java
View File

@ -28,7 +28,10 @@ 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.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -102,12 +105,17 @@ public class RankStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// n
expression.addParameter(new StreamExpressionNamedParameter("n", Integer.toString(size)));
if(includeStreams){
// stream
if(stream instanceof Expressible){
expression.addParameter(((Expressible)stream).toExpression(factory));
@ -115,6 +123,10 @@ public class RankStream extends TupleStream implements Expressible {
else{
throw new IOException("This RankStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
}
// sort
expression.addParameter(new StreamExpressionNamedParameter("sort",comp.toExpression(factory)));
@ -122,19 +134,33 @@ public class RankStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
.withHelper(comp.toExplanation(factory));
}
public void setStreamContext(StreamContext context) {
this.stream.setStreamContext(context);
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
List<TupleStream> l = new ArrayList<TupleStream>();
l.add(stream);
return l;
}
public void open() throws IOException {
this.top = new PriorityQueue(size, new ReverseComp(comp));
this.topList = new LinkedList();
this.top = new PriorityQueue<Tuple>(size, new ReverseComp(comp));
this.topList = new LinkedList<Tuple>();
stream.open();
}
@ -187,6 +213,7 @@ public class RankStream extends TupleStream implements Expressible {
class ReverseComp implements Comparator<Tuple>, Serializable {
private static final long serialVersionUID = 1L;
private StreamComparator comp;
public ReverseComp(StreamComparator comp) {

33
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/ReducerStream.java
View File

@ -18,10 +18,8 @@ package org.apache.solr.client.solrj.io.stream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.FieldComparator;
@ -32,7 +30,10 @@ import org.apache.solr.client.solrj.io.eq.MultipleFieldEqualitor;
import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
import org.apache.solr.client.solrj.io.ops.ReduceOperation;
import org.apache.solr.client.solrj.io.ops.StreamOperation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -135,11 +136,20 @@ public class ReducerStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// stream
if(includeStreams){
expression.addParameter(stream.toExpression(factory));
}
else{
expression.addParameter("<stream>");
}
// over
if(eq instanceof Expressible){
@ -158,12 +168,29 @@ public class ReducerStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
.withHelpers(new Explanation[]{
eq.toExplanation(factory),
op.toExplanation(factory)
});
}
public void setStreamContext(StreamContext context) {
this.stream.setStreamContext(context);
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
List<TupleStream> l = new ArrayList<TupleStream>();
l.add(stream);
return l;
}

41
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/RollupStream.java
View File

@ -17,23 +17,24 @@
package org.apache.solr.client.solrj.io.stream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.ArrayList;
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.HashKey;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
import org.apache.solr.client.solrj.io.eq.FieldEqualitor;
import org.apache.solr.client.solrj.io.eq.MultipleFieldEqualitor;
import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.metrics.Bucket;
@ -121,12 +122,21 @@ public class RollupStream extends TupleStream implements Expressible {
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
// stream
if(includeStreams){
expression.addParameter(tupleStream.toExpression(factory));
}
else{
expression.addParameter("<stream>");
}
// over
StringBuilder overBuilder = new StringBuilder();
@ -144,6 +154,25 @@ public class RollupStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
Explanation explanation = new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
tupleStream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString());
for(Metric metric : metrics){
explanation.withHelper(metric.toExplanation(factory));
}
return explanation;
}
public void setStreamContext(StreamContext context) {
this.tupleStream.setStreamContext(context);
}
@ -176,7 +205,7 @@ public class RollupStream extends TupleStream implements Expressible {
return tuple;
}
Map map = new HashMap();
Map<String,Object> map = new HashMap<String,Object>();
for(Metric metric : currentMetrics) {
map.put(metric.getIdentifier(), metric.getValue());
}
@ -207,7 +236,7 @@ public class RollupStream extends TupleStream implements Expressible {
} else {
Tuple t = null;
if(currentMetrics != null) {
Map map = new HashMap();
Map<String,Object> map = new HashMap<String,Object>();
for(Metric metric : currentMetrics) {
map.put(metric.getIdentifier(), metric.getValue());
}

31
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SelectStream.java
View File

@ -26,7 +26,10 @@ import java.util.Map;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
import org.apache.solr.client.solrj.io.ops.StreamOperation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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.StreamExpressionValue;
@ -111,9 +114,14 @@ public class SelectStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// stream
if(stream instanceof Expressible){
expression.addParameter(((Expressible)stream).toExpression(factory));
@ -121,6 +129,10 @@ public class SelectStream extends TupleStream implements Expressible {
else{
throw new IOException("This SelectStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
}
// selects
for(Map.Entry<String, String> selectField : selectedFields.entrySet()) {
@ -139,6 +151,25 @@ public class SelectStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
Explanation explanation = new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString());
for(StreamOperation operation : operations){
explanation.addHelper(operation.toExplanation(factory));
}
return explanation;
}
public void setStreamContext(StreamContext context) {
this.stream.setStreamContext(context);
}

18
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
View File

@ -19,15 +19,19 @@ package org.apache.solr.client.solrj.io.stream;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
import java.util.List;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.io.SolrClientCache;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
import org.slf4j.Logger;
@ -149,6 +153,16 @@ public class SolrStream extends TupleStream {
return buf.toString();
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withFunctionName("non-expressible")
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_SOURCE)
.withExpression("non-expressible");
}
/**
* Closes the Stream to a single Solr Instance
* */

26
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SortStream.java
View File

@ -25,11 +25,14 @@ 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.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
/**
@ -107,9 +110,14 @@ public class SortStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// streams
if(stream instanceof Expressible){
expression.addParameter(((Expressible)stream).toExpression(factory));
@ -117,6 +125,10 @@ public class SortStream extends TupleStream implements Expressible {
else{
throw new IOException("This SortStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
}
// by
if(comparator instanceof Expressible){
@ -129,6 +141,20 @@ public class SortStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
.withHelper(comparator.toExplanation(factory));
}
public void setStreamContext(StreamContext context) {
this.stream.setStreamContext(context);
}

27
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/StatsStream.java
View File

@ -23,13 +23,17 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.CloudSolrClient.Builder;
import org.apache.solr.client.solrj.io.SolrClientCache;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -150,6 +154,29 @@ public class StatsStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_SOURCE);
explanation.setExpression(toExpression(factory).toString());
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName(String.format(Locale.ROOT, "solr (worker ? of ?)"));
// TODO: fix this so we know the # of workers - check with Joel about a Stat's ability to be in a
// parallel stream.
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
child.setExpression(props.entrySet().stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), e.getValue())).collect(Collectors.joining(",")));
explanation.addChild(child);
return explanation;
}
public void setStreamContext(StreamContext context) {
cache = context.getSolrClientCache();
}

52
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TopicStream.java
View File

@ -23,33 +23,36 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Random;
import java.util.Set;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.impl.CloudSolrClient.Builder;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
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.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.common.SolrDocument;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkCoreNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.ExecutorUtil;
@ -61,7 +64,6 @@ public class TopicStream extends CloudSolrStream implements Expressible {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final long serialVersionUID = 1;
private long count;
@ -69,7 +71,7 @@ public class TopicStream extends CloudSolrStream implements Expressible {
private String id;
protected long checkpointEvery;
private Map<String, Long> checkpoints = new HashMap();
private Map<String, Long> checkpoints = new HashMap<String, Long>();
private String checkpointCollection;
public TopicStream(String zkHost,
@ -178,7 +180,7 @@ public class TopicStream extends CloudSolrStream implements Expressible {
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
public StreamExpression toExpression(StreamFactory factory) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
@ -205,6 +207,42 @@ public class TopicStream extends CloudSolrStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
explanation.setFunctionName(factory.getFunctionName(this.getClass()));
explanation.setImplementingClass(this.getClass().getName());
explanation.setExpressionType(ExpressionType.STREAM_SOURCE);
explanation.setExpression(toExpression(factory).toString());
{
// child 1 is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
child.setFunctionName(String.format(Locale.ROOT, "solr (%s)", collection));
// TODO: fix this so we know the # of workers - check with Joel about a Topic's ability to be in a
// parallel stream.
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
child.setExpression(params.entrySet().stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), e.getValue())).collect(Collectors.joining(",")));
explanation.addChild(child);
}
{
// child 2 is a place where we store and read checkpoint info from
StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-checkpoint");
child.setFunctionName(String.format(Locale.ROOT, "solr (checkpoint store)"));
child.setImplementingClass("Solr/Lucene");
child.setExpressionType(ExpressionType.DATASTORE);
child.setExpression(String.format(Locale.ROOT, "id=%s, collection=%s, checkpointEvery=%d", id, checkpointCollection, checkpointEvery));
explanation.addChild(child);
}
return explanation;
}
public List<TupleStream> children() {
List<TupleStream> l = new ArrayList();
return l;

10
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/TupleStream.java
View File

@ -21,15 +21,20 @@ import java.io.IOException;
import java.io.Serializable;
import java.io.Writer;
import java.util.List;
import java.util.UUID;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public abstract class TupleStream implements Closeable, Serializable {
private static final long serialVersionUID = 1;
private UUID streamNodeId = UUID.randomUUID();
public TupleStream() {
}
@ -54,8 +59,13 @@ public abstract class TupleStream implements Closeable, Serializable {
public abstract StreamComparator getStreamSort();
public abstract Explanation toExplanation(StreamFactory factory) throws IOException;
public int getCost() {
return 0;
}
public UUID getStreamNodeId(){
return streamNodeId;
}
}

27
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/UniqueStream.java
View File

@ -26,7 +26,10 @@ import org.apache.solr.client.solrj.io.comp.StreamComparator;
import org.apache.solr.client.solrj.io.eq.FieldEqualitor;
import org.apache.solr.client.solrj.io.eq.StreamEqualitor;
import org.apache.solr.client.solrj.io.ops.DistinctOperation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
@ -86,9 +89,14 @@ public class UniqueStream extends TupleStream implements Expressible {
@Override
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// streams
if(originalStream instanceof Expressible){
expression.addParameter(((Expressible)originalStream).toExpression(factory));
@ -96,6 +104,10 @@ public class UniqueStream extends TupleStream implements Expressible {
else{
throw new IOException("This UniqueStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
}
// over
if(originalEqualitor instanceof Expressible){
@ -108,6 +120,21 @@ public class UniqueStream extends TupleStream implements Expressible {
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[] {
originalStream.toExplanation(factory)
// we're not including that this is wrapped with a ReducerStream stream because that's just an implementation detail
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
.withHelper(originalEqualitor.toExplanation(factory));
}
public void setStreamContext(StreamContext context) {
this.originalStream.setStreamContext(context);
this.reducerStream.setStreamContext(context);

40
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/UpdateStream.java
View File

@ -30,10 +30,12 @@ import org.apache.solr.client.solrj.impl.CloudSolrClient.Builder;
import org.apache.solr.client.solrj.io.SolrClientCache;
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.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.common.SolrInputDocument;
@ -143,21 +145,55 @@ public class UpdateStream extends TupleStream implements Expressible {
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
expression.addParameter(collection);
expression.addParameter(new StreamExpressionNamedParameter("zkHost", zkHost));
expression.addParameter(new StreamExpressionNamedParameter("batchSize", Integer.toString(updateBatchSize)));
if(includeStreams){
if(tupleSource instanceof Expressible){
expression.addParameter(((Expressible)tupleSource).toExpression(factory));
} else {
throw new IOException("This ParallelStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
}
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
// An update stream is backward wrt the order in the explanation. This stream is the "child"
// while the collection we're updating is the parent.
StreamExplanation explanation = new StreamExplanation(getStreamNodeId() + "-datastore");
explanation.setFunctionName(String.format(Locale.ROOT, "solr (%s)", collection));
explanation.setImplementingClass("Solr/Lucene");
explanation.setExpressionType(ExpressionType.DATASTORE);
explanation.setExpression("Update into " + collection);
// child is a datastore so add it at this point
StreamExplanation child = new StreamExplanation(getStreamNodeId().toString());
child.setFunctionName(String.format(Locale.ROOT, factory.getFunctionName(getClass())));
child.setImplementingClass(getClass().getName());
child.setExpressionType(ExpressionType.STREAM_DECORATOR);
child.setExpression(toExpression(factory, false).toString());
explanation.addChild(child);
return explanation;
}
@Override
public void setStreamContext(StreamContext context) {
this.cache = context.getSolrClientCache();

158
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/Explanation.java Normal file
View File

@ -0,0 +1,158 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.solr.client.solrj.io.stream.expr;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Explanation containing details about a expression
*/
public class Explanation {
private String expressionNodeId;
private String expressionType;
private String functionName;
private String implementingClass;
private String expression;
private String note;
private List<Explanation> helpers;
public Explanation(String expressionNodeId){
this.expressionNodeId = expressionNodeId;
}
public String getExpressionNodeId(){
return expressionNodeId;
}
public String getExpressionType(){
return expressionType;
}
public void setExpressionType(String expressionType){
this.expressionType = expressionType;
}
public Explanation withExpressionType(String expressionType){
setExpressionType(expressionType);
return this;
}
public String getFunctionName(){
return functionName;
}
public void setFunctionName(String functionName){
this.functionName = functionName;
}
public Explanation withFunctionName(String functionName){
setFunctionName(functionName);
return this;
}
public String getImplementingClass(){
return implementingClass;
}
public void setImplementingClass(String implementingClass){
this.implementingClass = implementingClass;
}
public Explanation withImplementingClass(String implementingClass){
setImplementingClass(implementingClass);
return this;
}
public String getExpression(){
return expression;
}
public void setExpression(String expression){
this.expression = expression;
}
public Explanation withExpression(String expression){
setExpression(expression);
return this;
}
public String getNote(){
return note;
}
public void setNote(String note){
this.note = note;
}
public Explanation withNote(String note){
setNote(note);
return this;
}
public List<Explanation> getHelpers(){
return helpers;
}
public void setHelpers(List<Explanation> helpers){
this.helpers = helpers;
}
public Explanation withHelpers(List<Explanation> helpers){
setHelpers(helpers);
return this;
}
public Explanation withHelpers(Explanation[] helpers){
for(Explanation helper : helpers){
addHelper(helper);
}
return this;
}
public Explanation withHelper(Explanation helper){
addHelper(helper);
return this;
}
public void addHelper(Explanation helper){
if(null == helpers){
helpers = new ArrayList<Explanation>(1);
}
helpers.add(helper);
}
public Map<String,Object> toMap(){
Map<String,Object> map = new HashMap<String,Object>();
if(null != expressionNodeId){ map.put("expressionNodeId",expressionNodeId); }
if(null != expressionType){ map.put("expressionType",expressionType); }
if(null != functionName){ map.put("functionName",functionName); }
if(null != implementingClass){ map.put("implementingClass",implementingClass); }
if(null != expression){ map.put("expression",expression); }
if(null != note){ map.put("note",note); }
if(null != helpers && 0 != helpers.size()){
List<Map<String,Object>> helperMaps = new ArrayList<Map<String,Object>>();
for(Explanation helper : helpers){
helperMaps.add(helper.toMap());
}
map.put("helpers", helperMaps);
}
return map;
}
public static interface ExpressionType{
public static final String GRAPH_SOURCE = "graph-source";
public static final String STREAM_SOURCE = "stream-source";
public static final String STREAM_DECORATOR = "stream-decorator";
public static final String DATASTORE = "datastore";
public static final String METRIC = "metric";
public static final String OPERATION = "operation";
public static final String EQUALITOR = "equalitor";
public static final String SORTER = "sorter";
}
}

9
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/Expressible.java
View File

@ -25,4 +25,13 @@ public interface Expressible {
// public String getFunctionName();
// public void setFunctionName(String functionName);
StreamExpressionParameter toExpression(StreamFactory factory) throws IOException;
/**
* Returns an explanation about the stream object
* @param factory Stream factory for this, contains information about the function name
* @return Explanation about this stream object containing explanations of any child stream objects
* @throws IOException throw on any error
*/
Explanation toExplanation(StreamFactory factory) throws IOException;
}

72
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/expr/StreamExplanation.java Normal file
View File

@ -0,0 +1,72 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.solr.client.solrj.io.stream.expr;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
/**
* Explanation containing details about a stream expression
*/
public class StreamExplanation extends Explanation {
private List<Explanation> children;
public StreamExplanation(String expressionNodeId){
super(expressionNodeId);
}
public List<Explanation> getChildren(){
return children;
}
public void setChildren(List<Explanation> children){
this.children = children;
}
public StreamExplanation withChildren(List<Explanation> children){
setChildren(children);
return this;
}
public StreamExplanation withChildren(Explanation[] children){
for(Explanation child : children){
addChild(child);
}
return this;
}
public void addChild(Explanation child){
if(null == children){
children = new ArrayList<Explanation>(1);
}
children.add(child);
}
public Map<String,Object> toMap(){
Map<String,Object> map = super.toMap();
if(null != children && 0 != children.size()){
List<Map<String,Object>> childrenMaps = new ArrayList<Map<String,Object>>();
for(Explanation child : children){
childrenMaps.add(child.toMap());
}
map.put("children", childrenMaps);
}
return map;
}
}

19
solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/metrics/Metric.java
View File

@ -16,14 +16,20 @@
*/
package org.apache.solr.client.solrj.io.stream.metrics;
import java.io.IOException;
import java.io.Serializable;
import java.util.UUID;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
public abstract class Metric implements Serializable, Expressible {
private static final long serialVersionUID = 1L;
private UUID metricNodeId = UUID.randomUUID();
private String functionName;
private String identifier;
@ -51,6 +57,19 @@ public abstract class Metric implements Serializable, Expressible {
this.identifier = sb.toString();
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new Explanation(metricNodeId.toString())
.withFunctionName(functionName)
.withImplementingClass(getClass().getName())
.withExpression(toExpression(factory).toString())
.withExpressionType(ExpressionType.METRIC);
}
public UUID getMetricNodeId(){
return metricNodeId;
}
public abstract double getValue();
public abstract void update(Tuple tuple);
public abstract Metric newInstance();

28
solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/RecordCountStream.java
View File

@ -24,10 +24,13 @@ 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.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
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;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
public class RecordCountStream extends TupleStream implements Expressible, Serializable {
@ -54,10 +57,15 @@ public class RecordCountStream extends TupleStream implements Expressible, Seria
}
@Override
public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {
public StreamExpression toExpression(StreamFactory factory) throws IOException{
return toExpression(factory, true);
}
private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
// function name
StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
if(includeStreams){
// stream
if(stream instanceof Expressible){
expression.addParameter(((Expressible)stream).toExpression(factory));
@ -65,10 +73,28 @@ public class RecordCountStream extends TupleStream implements Expressible, Seria
else{
throw new IOException("This RecordCountStream contains a non-expressible TupleStream - it cannot be converted to an expression");
}
}
else{
expression.addParameter("<stream>");
}
return expression;
}
@Override
public Explanation toExplanation(StreamFactory factory) throws IOException {
return new StreamExplanation(getStreamNodeId().toString())
.withChildren(new Explanation[]{
stream.toExplanation(factory)
})
.withFunctionName(factory.getFunctionName(this.getClass()))
.withImplementingClass(this.getClass().getName())
.withExpressionType(ExpressionType.STREAM_DECORATOR)
.withExpression(toExpression(factory, false).toString())
;
}
public void close() throws IOException {
this.stream.close();
}

290
solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionToExplanationTest.java Normal file
View File

@ -0,0 +1,290 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.solr.client.solrj.io.stream;
import junit.framework.Assert;
import org.apache.lucene.util.LuceneTestCase;
import org.apache.solr.client.solrj.io.ops.GroupOperation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParser;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.metrics.CountMetric;
import org.apache.solr.client.solrj.io.stream.metrics.MaxMetric;
import org.apache.solr.client.solrj.io.stream.metrics.MeanMetric;
import org.apache.solr.client.solrj.io.stream.metrics.Metric;
import org.apache.solr.client.solrj.io.stream.metrics.MinMetric;
import org.apache.solr.client.solrj.io.stream.metrics.SumMetric;
import org.junit.Test;
/**
**/
public class StreamExpressionToExplanationTest extends LuceneTestCase {
private StreamFactory factory;
public StreamExpressionToExplanationTest() {
super();
factory = new StreamFactory()
.withCollectionZkHost("collection1", "testhost:1234")
.withCollectionZkHost("collection2", "testhost:1234")
.withFunctionName("search", CloudSolrStream.class)
.withFunctionName("select", SelectStream.class)
.withFunctionName("merge", MergeStream.class)
.withFunctionName("unique", UniqueStream.class)
.withFunctionName("top", RankStream.class)
.withFunctionName("reduce", ReducerStream.class)
.withFunctionName("group", GroupOperation.class)
.withFunctionName("update", UpdateStream.class)
.withFunctionName("stats", StatsStream.class)
.withFunctionName("facet", FacetStream.class)
.withFunctionName("jdbc", JDBCStream.class)
.withFunctionName("intersect", IntersectStream.class)
.withFunctionName("complement", ComplementStream.class)
.withFunctionName("count", CountMetric.class)
.withFunctionName("sum", SumMetric.class)
.withFunctionName("min", MinMetric.class)
.withFunctionName("max", MaxMetric.class)
.withFunctionName("avg", MeanMetric.class)
.withFunctionName("daemon", DaemonStream.class)
.withFunctionName("topic", TopicStream.class)
;
}
@Test
public void testCloudSolrStream() throws Exception {
CloudSolrStream stream;
// Basic test
stream = new CloudSolrStream(StreamExpressionParser.parse("search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("search", explanation.getFunctionName());
Assert.assertEquals(CloudSolrStream.class.getName(), explanation.getImplementingClass());
}
@Test
public void testSelectStream() throws Exception {
SelectStream stream;
// Basic test
stream = new SelectStream(StreamExpressionParser.parse("select(\"a_s as fieldA\", search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"))"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("select", explanation.getFunctionName());
Assert.assertEquals(SelectStream.class.getName(), explanation.getImplementingClass());
}
@Test
public void testDaemonStream() throws Exception {
DaemonStream stream;
// Basic test
stream = new DaemonStream(StreamExpressionParser.parse("daemon(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), id=\"blah\", runInterval=\"1000\", queueSize=\"100\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("daemon", explanation.getFunctionName());
Assert.assertEquals(DaemonStream.class.getName(), explanation.getImplementingClass());
}
@Test
public void testTopicStream() throws Exception {
TopicStream stream;
// Basic test
stream = new TopicStream(StreamExpressionParser.parse("topic(collection2, collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", id=\"blah\", checkpointEvery=1000)"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("topic", explanation.getFunctionName());
Assert.assertEquals(TopicStream.class.getName(), explanation.getImplementingClass());
}
@Test
public void testStatsStream() throws Exception {
StatsStream stream;
// Basic test
stream = new StatsStream(StreamExpressionParser.parse("stats(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\", sum(a_i), avg(a_i), count(*), min(a_i), max(a_i))"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("stats", explanation.getFunctionName());
Assert.assertEquals(StatsStream.class.getName(), explanation.getImplementingClass());
}
@Test
public void testUniqueStream() throws Exception {
UniqueStream stream;
// Basic test
stream = new UniqueStream(StreamExpressionParser.parse("unique(search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\"), over=\"a_f\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("unique", explanation.getFunctionName());
Assert.assertEquals(UniqueStream.class.getName(), explanation.getImplementingClass());
}
@Test
public void testMergeStream() throws Exception {
MergeStream stream;
// Basic test
stream = new MergeStream(StreamExpressionParser.parse("merge("
+ "search(collection1, q=\"id:(0 3 4)\", fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_s asc\"),"
+ "search(collection1, q=\"id:(1 2)\", fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_s asc\"),"
+ "on=\"a_f asc, a_s asc\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("merge", explanation.getFunctionName());
Assert.assertEquals(MergeStream.class.getName(), explanation.getImplementingClass());
Assert.assertEquals(2, ((StreamExplanation)explanation).getChildren().size());
}
@Test
public void testRankStream() throws Exception {
RankStream stream;
String expressionString;
// Basic test
stream = new RankStream(StreamExpressionParser.parse("top("
+ "n=3,"
+ "search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc,a_i asc\"),"
+ "sort=\"a_f asc, a_i asc\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("top", explanation.getFunctionName());
Assert.assertEquals(RankStream.class.getName(), explanation.getImplementingClass());
Assert.assertEquals(1, ((StreamExplanation)explanation).getChildren().size());
}
@Test
public void testReducerStream() throws Exception {
ReducerStream stream;
String expressionString;
// Basic test
stream = new ReducerStream(StreamExpressionParser.parse("reduce("
+ "search(collection1, q=*:*, fl=\"id,a_s,a_i,a_f\", sort=\"a_s desc, a_f asc\"),"
+ "by=\"a_s\", group(sort=\"a_i desc\", n=\"5\"))"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("reduce", explanation.getFunctionName());
Assert.assertEquals(ReducerStream.class.getName(), explanation.getImplementingClass());
Assert.assertEquals(1, ((StreamExplanation)explanation).getChildren().size());
}
@Test
public void testUpdateStream() throws Exception {
StreamExpression expression = StreamExpressionParser.parse("update("
+ "collection2, "
+ "batchSize=5, "
+ "search("
+ "collection1, "
+ "q=*:*, "
+ "fl=\"id,a_s,a_i,a_f\", "
+ "sort=\"a_f asc, a_i asc\"))");
UpdateStream updateStream = new UpdateStream(expression, factory);
Explanation explanation = updateStream.toExplanation(factory);
Assert.assertEquals("solr (collection2)", explanation.getFunctionName());
Assert.assertEquals("Solr/Lucene", explanation.getImplementingClass());
StreamExplanation updateExplanation = (StreamExplanation)explanation;
Assert.assertEquals(1, updateExplanation.getChildren().size());
Assert.assertEquals("update", updateExplanation.getChildren().get(0).getFunctionName());
Assert.assertEquals(UpdateStream.class.getName(), updateExplanation.getChildren().get(0).getImplementingClass());
}
@Test
public void testFacetStream() throws Exception {
FacetStream stream;
String expressionString;
// Basic test
stream = new FacetStream(StreamExpressionParser.parse("facet("
+ "collection1, "
+ "q=\"*:*\", "
+ "buckets=\"a_s\", "
+ "bucketSorts=\"sum(a_i) asc\", "
+ "bucketSizeLimit=100, "
+ "sum(a_i), sum(a_f), "
+ "min(a_i), min(a_f), "
+ "max(a_i), max(a_f), "
+ "avg(a_i), avg(a_f), "
+ "count(*)"
+ ")"), factory);
expressionString = stream.toExpression(factory).toString();
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("facet", explanation.getFunctionName());
Assert.assertEquals(FacetStream.class.getName(), explanation.getImplementingClass());
Assert.assertEquals(1, ((StreamExplanation)explanation).getChildren().size());
}
@Test
public void testJDBCStream() throws Exception {
JDBCStream stream;
String expressionString;
// Basic test
stream = new JDBCStream(StreamExpressionParser.parse("jdbc(connection=\"jdbc:hsqldb:mem:.\", sql=\"select PEOPLE.ID, PEOPLE.NAME, COUNTRIES.COUNTRY_NAME from PEOPLE inner join COUNTRIES on PEOPLE.COUNTRY_CODE = COUNTRIES.CODE order by PEOPLE.ID\", sort=\"ID asc\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("jdbc", explanation.getFunctionName());
Assert.assertEquals(JDBCStream.class.getName(), explanation.getImplementingClass());
Assert.assertEquals(1, ((StreamExplanation)explanation).getChildren().size());
}
@Test
public void testIntersectStream() throws Exception {
IntersectStream stream;
String expressionString;
// Basic test
stream = new IntersectStream(StreamExpressionParser.parse("intersect("
+ "search(collection1, q=\"id:(0 3 4)\", fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_s asc\"),"
+ "search(collection1, q=\"id:(1 2)\", fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_s asc\"),"
+ "on=\"a_f, a_s\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("intersect", explanation.getFunctionName());
Assert.assertEquals(IntersectStream.class.getName(), explanation.getImplementingClass());
Assert.assertEquals(2, ((StreamExplanation)explanation).getChildren().size());
}
@Test
public void testComplementStream() throws Exception {
ComplementStream stream;
String expressionString;
// Basic test
stream = new ComplementStream(StreamExpressionParser.parse("complement("
+ "search(collection1, q=\"id:(0 3 4)\", fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_s asc\"),"
+ "search(collection1, q=\"id:(1 2)\", fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_s asc\"),"
+ "on=\"a_f, a_s\")"), factory);
Explanation explanation = stream.toExplanation(factory);
Assert.assertEquals("complement", explanation.getFunctionName());
Assert.assertEquals(ComplementStream.class.getName(), explanation.getImplementingClass());
Assert.assertEquals(2, ((StreamExplanation)explanation).getChildren().size());
}
}