SOLR-12505: fetch() returns the fetched fields when the target handler's default query parser is not lucene

changelog/unreleased/SOLR-12505-fetch-stream-deftype.yml

@@ -0,0 +1,8 @@
+title: >
+  The fetch() streaming expression now sends defType=lucene with its batch query, so it no longer fails to enrich tuples when the target request handler defaults to a different query parser, and it rejects a join field name that would break that query.
+type: fixed
+authors:
+  - name: Nick Shanin
+links:
+  - name: SOLR-12505
+    url: https://issues.apache.org/jira/browse/SOLR-12505

solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc

@@ -768,6 +768,7 @@ This is the approach for executing batches of streaming expressions from a `topi
 The `fetch` function iterates a stream and fetches additional fields and adds them to the tuples.
 The `fetch` function fetches in batches to limit the number of calls back to Solr.
 Tuples streamed from the `fetch` function will contain the original fields and the additional fields that were fetched.
+The batch lookup uses `defType=lucene`, even if the target collection's `/select` handler defaults to a different query parser.
 The `fetch` function supports one-to-one fetches.
 Many-to-one fetches, where the stream source contains duplicate keys, will also work, but one-to-many fetches are currently not supported by this function.
 
@@ -778,6 +779,7 @@ Many-to-one fetches, where the stream source contains duplicate keys, will also
 * `fl`: (Mandatory) The fields to be fetched.
 * `on`: Fields to be used for checking equality of tuples between stream source and fetched records.
 Formatted as `on="fieldNameInTuple=fieldNameInCollection"`.
+Because the collection-side field name is inserted into a local-parameters query, it cannot contain whitespace or `}`, or start with `'`, `"`, or `$`.
 * `batchSize`: (Optional) The batch fetch size.
 
 === fetch Syntax

solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/FetchStream.java

@@ -150,6 +150,14 @@ private void init(
       leftKey = rightKey = on;
     }
 
+    if (containsBatchBreakingChar(rightKey)) {
+      throw new IOException(
+          String.format(
+              Locale.ROOT,
+              "fetch on parameter has an invalid join field name [%s]: whitespace, '}', a leading quote and a leading '$' are not allowed",
+              rightKey));
+    }
+
     for (int i = 0; i < fields.length; i++) {
       fields[i] = fields[i].trim();
       if (fields[i].equals(VERSION_FIELD)) {
@@ -162,6 +170,22 @@ private void init(
     }
   }
 
+  // The batch query embeds the join field name as an unquoted local-params value. That reader
+  // ends an unquoted literal at whitespace or '}', a leading quote starts a quoted string, and
+  // a leading $ dereferences a request parameter, so reject exactly those shapes.
+  private static boolean containsBatchBreakingChar(String name) {
+    for (int i = 0; i < name.length(); i++) {
+      char c = name.charAt(i);
+      if (Character.isWhitespace(c) || c == '}') {
+        return true;
+      }
+      if (i == 0 && (c == '\'' || c == '"' || c == '$')) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   @Override
   public StreamExpression toExpression(StreamFactory factory) throws IOException {
     return toExpression(factory, true);
@@ -244,6 +268,9 @@ private void fetchBatch() throws IOException {
 
       ModifiableSolrParams params = new ModifiableSolrParams();
       params.add("q", buf.toString());
+      // The local-params prefix above is only honored by the lucene parser; do not inherit another
+      // defType (e.g. edismax) from the target request handler defaults.
+      params.add("defType", "lucene");
       params.add("fl", fieldList + appendFields());
       params.add("rows", Integer.toString(batchSize));
       params.add(SORT, "_version_ desc");

solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java

@@ -29,6 +29,7 @@
 import org.apache.lucene.tests.util.LuceneTestCase;
 import org.apache.solr.SolrTestCaseJ4;
 import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrRequest;
 import org.apache.solr.client.solrj.io.SolrClientCache;
 import org.apache.solr.client.solrj.io.Tuple;
 import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
@@ -60,6 +61,7 @@
 import org.apache.solr.client.solrj.io.stream.metrics.SumMetric;
 import org.apache.solr.client.solrj.request.CollectionAdminRequest;
 import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.client.solrj.request.V2Request;
 import org.apache.solr.client.solrj.response.QueryResponse;
 import org.apache.solr.cloud.AbstractFullDistribZkTestBase;
 import org.apache.solr.cloud.SolrCloudTestCase;
@@ -93,6 +95,13 @@ public static void setupCluster() throws Exception {
                 .resolve("configsets")
                 .resolve("streaming")
                 .resolve("conf"))
+        .addConfig(
+            "conf-edismax",
+            getFile("solrj")
+                .resolve("solr")
+                .resolve("configsets")
+                .resolve("streaming")
+                .resolve("conf"))
         .addConfig(
             "ml",
             getFile("solrj").resolve("solr").resolve("configsets").resolve("ml").resolve("conf"))
@@ -941,6 +950,251 @@ public void testParallelHavingStream() throws Exception {
     solrClientCache.close();
   }
 
+  /**
+   * SOLR-12505: fetch() sends its batch query as "{! df=.. q.op=OR}.." and must not depend on the
+   * target handler's default defType.
+   */
+  @Test
+  public void testFetchStreamWithNonLuceneDefaultDefType() throws Exception {
+    final String edismaxCollection = "fetchEdismax";
+    // The Config API overlay is stored with the configset, not the collection, so the /select
+    // overlay set below would outlive this collection and leak into every other test in this
+    // class that uses the shared "conf" configset. This collection uses its own copy instead.
+    CollectionAdminRequest.createCollection(edismaxCollection, "conf-edismax", 1, 1)
+        .process(cluster.getSolrClient());
+    cluster.waitForActiveCollection(edismaxCollection, 1, 1);
+    SolrClientCache solrClientCache = new SolrClientCache();
+    try {
+      // make the target handler default to a parser that ignores the {! df=.. q.op=OR} local params
+      cluster
+          .getSolrClient()
+          .request(
+              new V2Request.Builder("/c/" + edismaxCollection + "/config")
+                  .withMethod(SolrRequest.METHOD.POST)
+                  .withPayload(
+                      "{'update-requesthandler': {'name': '/select',"
+                          + " 'class': 'solr.SearchHandler', 'defaults': {'defType': 'edismax'}}}")
+                  .build());
+
+      // The subject values deliberately share no token with the join keys (0, 1, 2): the schema
+      // copyFields subject into text, the handler default field, so a token in common would let
+      // the default edismax parser find the documents by accident and hide the defType bug.
+      new UpdateRequest()
+          .add(id, "0", "a_s", "hello0", "a_i", "0", "subject", "alpha")
+          .add(id, "1", "a_s", "hello1", "a_i", "1", "subject", "bravo")
+          .add(id, "2", "a_s", "hello2", "a_i", "2", "subject", "charlie")
+          .commit(cluster.getSolrClient(), edismaxCollection);
+
+      StreamFactory factory =
+          new StreamFactory()
+              .withCollectionUseThisConnection(edismaxCollection, getSolrConnection())
+              .withFunctionName("search", CloudSolrStream.class)
+              .withFunctionName("fetch", FetchStream.class);
+
+      TupleStream stream =
+          factory.constructStream(
+              "fetch("
+                  + edismaxCollection
+                  + ", search("
+                  + edismaxCollection
+                  + ", q=*:*, fl=\"id,a_s,a_i\", sort=\"a_i asc\"), on=\"id=a_i\","
+                  + " batchSize=\"2\", fl=\"subject\")");
+      StreamContext context = new StreamContext();
+      context.setSolrClientCache(solrClientCache);
+      stream.setStreamContext(context);
+      List<Tuple> tuples = getTuples(stream);
+
+      assertEquals(3, tuples.size());
+      final String[] expectedSubjects = {"alpha", "bravo", "charlie"};
+      for (int i = 0; i < 3; i++) {
+        assertEquals(expectedSubjects[i], tuples.get(i).getString("subject"));
+      }
+    } finally {
+      solrClientCache.close();
+      CollectionAdminRequest.deleteCollection(edismaxCollection).process(cluster.getSolrClient());
+    }
+  }
+
+  @Test
+  public void testFetchStreamRejectsJoinFieldNameThatBreaksLocalParams() throws Exception {
+    SolrClientCache solrClientCache = new SolrClientCache();
+    new UpdateRequest()
+        .add(id, "0", "a_s", "hello0", "a_i", "0")
+        .add(id, "1", "a_s", "hello1", "a_i", "1")
+        .commit(cluster.getSolrClient(), COLLECTIONORALIAS);
+    try {
+      StreamFactory factory =
+          new StreamFactory()
+              .withCollectionUseThisConnection(COLLECTIONORALIAS, getSolrConnection())
+              .withFunctionName("search", CloudSolrStream.class)
+              .withFunctionName("fetch", FetchStream.class);
+
+      String expression =
+          "fetch("
+              + COLLECTIONORALIAS
+              + ", search("
+              + COLLECTIONORALIAS
+              + ", q=*:*, fl=\"id,a_s,a_i\", sort=\"a_i asc\"), on=\"id=a i\","
+              + " batchSize=\"2\", fl=\"subject\")";
+      IOException e =
+          assertThrows(
+              IOException.class, () -> drainFetchStream(factory, solrClientCache, expression));
+      assertTrue(e.getMessage(), causeChainContains(e, "invalid join field name [a i]"));
+    } finally {
+      solrClientCache.close();
+    }
+  }
+
+  @Test
+  public void testFetchStreamRejectsJoinFieldNameWithUnicodeSpace() throws Exception {
+    SolrClientCache solrClientCache = new SolrClientCache();
+    new UpdateRequest()
+        .add(id, "0", "a_s", "hello0", "a_i", "0")
+        .add(id, "1", "a_s", "hello1", "a_i", "1")
+        .commit(cluster.getSolrClient(), COLLECTIONORALIAS);
+    try {
+      StreamFactory factory =
+          new StreamFactory()
+              .withCollectionUseThisConnection(COLLECTIONORALIAS, getSolrConnection())
+              .withFunctionName("search", CloudSolrStream.class)
+              .withFunctionName("fetch", FetchStream.class);
+
+      // The join field name carries an em space (U+2003). The local-params reader splits on
+      // Character.isWhitespace, so this name breaks the batch query the same way "a i" does,
+      // even though it has no ASCII whitespace.
+      String expression =
+          "fetch("
+              + COLLECTIONORALIAS
+              + ", search("
+              + COLLECTIONORALIAS
+              + ", q=*:*, fl=\"id,a_s,a_i\", sort=\"a_i asc\"), on=\"id=a\u2003i\","
+              + " batchSize=\"2\", fl=\"subject\")";
+      IOException e =
+          assertThrows(
+              IOException.class, () -> drainFetchStream(factory, solrClientCache, expression));
+      assertTrue(e.getMessage(), causeChainContains(e, "invalid join field name [a\u2003i]"));
+    } finally {
+      solrClientCache.close();
+    }
+  }
+
+  @Test
+  public void testFetchStreamRejectsJoinFieldNameStartingWithDollarSign() throws Exception {
+    SolrClientCache solrClientCache = new SolrClientCache();
+    new UpdateRequest()
+        .add(id, "0", "a_s", "hello0", "a_i", "0")
+        .add(id, "1", "a_s", "hello1", "a_i", "1")
+        .commit(cluster.getSolrClient(), COLLECTIONORALIAS);
+    try {
+      StreamFactory factory =
+          new StreamFactory()
+              .withCollectionUseThisConnection(COLLECTIONORALIAS, getSolrConnection())
+              .withFunctionName("search", CloudSolrStream.class)
+              .withFunctionName("fetch", FetchStream.class);
+
+      // A leading $ in the df value is read as a request parameter dereference, so the batch
+      // would look up the field named by that parameter instead of the join field itself.
+      String expression =
+          "fetch("
+              + COLLECTIONORALIAS
+              + ", search("
+              + COLLECTIONORALIAS
+              + ", q=*:*, fl=\"id,a_s,a_i\", sort=\"a_i asc\"), on=\"id=$a_i\","
+              + " batchSize=\"2\", fl=\"subject\")";
+      IOException e =
+          assertThrows(
+              IOException.class, () -> drainFetchStream(factory, solrClientCache, expression));
+      assertTrue(e.getMessage(), causeChainContains(e, "invalid join field name [$a_i]"));
+    } finally {
+      solrClientCache.close();
+    }
+  }
+
+  @Test
+  public void testFetchStreamRejectsJoinFieldNameWithClosingBraceOrLeadingQuote() throws Exception {
+    SolrClientCache solrClientCache = new SolrClientCache();
+    new UpdateRequest()
+        .add(id, "0", "a_s", "hello0", "a_i", "0")
+        .add(id, "1", "a_s", "hello1", "a_i", "1")
+        .commit(cluster.getSolrClient(), COLLECTIONORALIAS);
+    try {
+      StreamFactory factory =
+          new StreamFactory()
+              .withCollectionUseThisConnection(COLLECTIONORALIAS, getSolrConnection())
+              .withFunctionName("search", CloudSolrStream.class)
+              .withFunctionName("fetch", FetchStream.class);
+
+      // A } in the name ends the unquoted local-params value early, so the rest of the name
+      // lands outside the local params and changes the batch query.
+      String braceExpression =
+          "fetch("
+              + COLLECTIONORALIAS
+              + ", search("
+              + COLLECTIONORALIAS
+              + ", q=*:*, fl=\"id,a_s,a_i\", sort=\"a_i asc\"), on=\"id=a}i\","
+              + " batchSize=\"2\", fl=\"subject\")";
+      IOException braceFailure =
+          assertThrows(
+              IOException.class, () -> drainFetchStream(factory, solrClientCache, braceExpression));
+      assertTrue(
+          braceFailure.getMessage(),
+          causeChainContains(braceFailure, "invalid join field name [a}i]"));
+
+      // A leading quote makes the local-params reader treat the rest of the name as a quoted
+      // string instead of the field name itself.
+      String singleQuoteExpression =
+          "fetch("
+              + COLLECTIONORALIAS
+              + ", search("
+              + COLLECTIONORALIAS
+              + ", q=*:*, fl=\"id,a_s,a_i\", sort=\"a_i asc\"), on=\"id='a_i\","
+              + " batchSize=\"2\", fl=\"subject\")";
+      IOException singleQuoteFailure =
+          assertThrows(
+              IOException.class,
+              () -> drainFetchStream(factory, solrClientCache, singleQuoteExpression));
+      assertTrue(
+          singleQuoteFailure.getMessage(),
+          causeChainContains(singleQuoteFailure, "invalid join field name ['a_i]"));
+
+      String doubleQuoteExpression =
+          "fetch("
+              + COLLECTIONORALIAS
+              + ", search("
+              + COLLECTIONORALIAS
+              + ", q=*:*, fl=\"id,a_s,a_i\", sort=\"a_i asc\"), on=\"id=\\\"a_i\","
+              + " batchSize=\"2\", fl=\"subject\")";
+      IOException doubleQuoteFailure =
+          assertThrows(
+              IOException.class,
+              () -> drainFetchStream(factory, solrClientCache, doubleQuoteExpression));
+      assertTrue(
+          doubleQuoteFailure.getMessage(),
+          causeChainContains(doubleQuoteFailure, "invalid join field name [\"a_i]"));
+    } finally {
+      solrClientCache.close();
+    }
+  }
+
+  private List<Tuple> drainFetchStream(
+      StreamFactory factory, SolrClientCache solrClientCache, String expression)
+      throws IOException {
+    TupleStream stream = factory.constructStream(expression);
+    StreamContext context = new StreamContext();
+    context.setSolrClientCache(solrClientCache);
+    stream.setStreamContext(context);
+    return getTuples(stream);
+  }
+
+  private static boolean causeChainContains(Throwable t, String text) {
+    for (; t != null; t = t.getCause()) {
+      if (t.getMessage() != null && t.getMessage().contains(text)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   @Test
   public void testFetchStream() throws Exception {