SOLR-9128: Fix error handling issues in Streaming classes

solr/core/src/java/org/apache/solr/handler/GraphHandler.java

@@ -113,6 +113,7 @@ public class GraphHandler extends RequestHandlerBase implements SolrCoreAware, P
         .withFunctionName("update", UpdateStream.class)
         .withFunctionName("jdbc", JDBCStream.class)
         .withFunctionName("intersect", IntersectStream.class)
+        .withFunctionName("select", SelectStream.class)
         .withFunctionName("complement", ComplementStream.class)
         .withFunctionName("daemon", DaemonStream.class)
         .withFunctionName("topic", TopicStream.class)

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

@@ -275,7 +275,7 @@ public class GatherNodesStream extends TupleStream implements Expressible {
     // collection
     expression.addParameter(collection);
 
-    if(includeStreams){
+    if(includeStreams && !(tupleStream instanceof NodeStream)){
       if(tupleStream instanceof Expressible){
         expression.addParameter(((Expressible)tupleStream).toExpression(factory));
       }
@@ -311,7 +311,14 @@ public class GatherNodesStream extends TupleStream implements Expressible {
     if(maxDocFreq > -1) {
       expression.addParameter(new StreamExpressionNamedParameter("maxDocFreq", Integer.toString(maxDocFreq)));
     }
+    if(tupleStream instanceof NodeStream) {
+      NodeStream nodeStream = (NodeStream)tupleStream;
+      expression.addParameter(new StreamExpressionNamedParameter("walk", nodeStream.toString() + "->" + traverseTo));
+
+    } else {
       expression.addParameter(new StreamExpressionNamedParameter("walk", traverseFrom + "->" + traverseTo));
+    }
+
     expression.addParameter(new StreamExpressionNamedParameter("trackTraversal", Boolean.toString(trackTraversal)));
 
     StringBuilder buf = new StringBuilder();
@@ -642,6 +649,19 @@ public class GatherNodesStream extends TupleStream implements Expressible {
       }
     }
 
+    public String toString() {
+      StringBuilder builder = new StringBuilder();
+      boolean comma = false;
+      for(String s : ids) {
+        if(comma) {
+          builder.append(",");
+        }
+        builder.append(s);
+        comma = true;
+      }
+      return builder.toString();
+    }
+    
     @Override
     public Explanation toExplanation(StreamFactory factory) throws IOException {
 

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

@@ -126,16 +126,16 @@ public class CloudSolrStream extends TupleStream implements Expressible {
     StreamExpressionNamedParameter aliasExpression = factory.getNamedOperand(expression, "aliases");
     StreamExpressionNamedParameter zkHostExpression = factory.getNamedOperand(expression, "zkHost");
 
-    // Validate there are no unknown parameters - zkHost and alias are namedParameter so we don't need to count it twice
-    if(expression.getParameters().size() != 1 + namedParams.size()){
-      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - unknown operands found",expression));
-    }
-    
     // Collection Name
     if(null == collectionName){
       throw new IOException(String.format(Locale.ROOT,"invalid expression %s - collectionName expected as first operand",expression));
     }
 
+    // Validate there are no unknown parameters - zkHost and alias are namedParameter so we don't need to count it twice
+    if(expression.getParameters().size() != 1 + namedParams.size()){
+      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - unknown operands found",expression));
+    }
+    
     // Named parameters - passed directly to solr as solrparams
     if(0 == namedParams.size()){
       throw new IOException(String.format(Locale.ROOT,"invalid expression %s - at least one named parameter expected. eg. 'q=*:*'",expression));
@@ -257,15 +257,20 @@ public class CloudSolrStream extends TupleStream implements Expressible {
     // If the comparator is null then it was not explicitly set so we will create one using the sort parameter
     // of the query. While doing this we will also take into account any aliases such that if we are sorting on
     // fieldA but fieldA is aliased to alias.fieldA then the comparater will be against alias.fieldA.
-    String fls = String.join(",", params.getParams("fl"));
-    if (fls == null) {
-      throw new IOException("fl param expected for a stream");
+
+    if (params.get("q") == null) {
+      throw new IOException("q param expected for search function");
     }
 
-    String sorts = String.join(",", params.getParams("sort"));
-    if (sorts == null) {
-      throw new IOException("sort param expected for a stream");
+    if (params.getParams("fl") == null) {
+      throw new IOException("fl param expected for search function");
     }
+    String fls = String.join(",", params.getParams("fl"));
+
+    if (params.getParams("sort") == null) {
+      throw new IOException("sort param expected for search function");
+    }
+    String sorts = String.join(",", params.getParams("sort"));
     this.comp = parseComp(sorts, fls);
   }
   

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

@@ -477,6 +477,9 @@ public class FacetStream extends TupleStream implements Expressible  {
 
     String bucketName = _buckets[level].toString();
     NamedList nl = (NamedList)facets.get(bucketName);
+    if(nl == null) {
+      return;
+    }
     List allBuckets = (List)nl.get("buckets");
     for(int b=0; b<allBuckets.size(); b++) {
       NamedList bucket = (NamedList)allBuckets.get(b);

solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamExpressionTest.java

@@ -129,6 +129,24 @@ public class StreamExpressionTest extends SolrCloudTestCase {
     assertOrder(tuples, 0, 3, 4);
     assertLong(tuples.get(1), "a_i", 3);
 
+    try {
+      expression = StreamExpressionParser.parse("search(" + COLLECTION + ", fl=\"id,a_s,a_i,a_f\", sort=\"a_f asc, a_i asc\")");
+      stream = new CloudSolrStream(expression, factory);
+      tuples = getTuples(stream);
+      throw new Exception("Should be an exception here");
+    } catch(Exception e) {
+      assertTrue(e.getMessage().contains("q param expected for search function"));
+    }
+
+    try {
+      expression = StreamExpressionParser.parse("search(" + COLLECTION + ", q=\"blah\", sort=\"a_f asc, a_i asc\")");
+      stream = new CloudSolrStream(expression, factory);
+      tuples = getTuples(stream);
+      throw new Exception("Should be an exception here");
+    } catch(Exception e) {
+      assertTrue(e.getMessage().contains("fl param expected for search function"));
+    }
+
   }
 
   @Test
@@ -2131,6 +2149,27 @@ public class StreamExpressionTest extends SolrCloudTestCase {
     assertTrue(avgf.doubleValue() == 5.5D);
     assertTrue(count.doubleValue() == 2);
 
+    //Test zero result facets
+    clause = "facet("
+        +   "collection1, "
+        +   "q=\"blahhh\", "
+        +   "fl=\"a_s,a_i,a_f\", "
+        +   "sort=\"a_s asc\", "
+        +   "buckets=\"a_s\", "
+        +   "bucketSorts=\"a_s 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(*)"
+        + ")";
+
+    stream = factory.constructStream(clause);
+    tuples = getTuples(stream);
+
+    assert(tuples.size() == 0);
+
   }
 
   @Test