diff --git a/changelog/unreleased/SOLR-12505-fetch-stream-deftype.yml b/changelog/unreleased/SOLR-12505-fetch-stream-deftype.yml new file mode 100644 index 000000000000..36ecebc53daf --- /dev/null +++ b/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 diff --git a/solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc b/solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc index 120f871f5f60..e3a3ec3fb0fc 100644 --- a/solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc +++ b/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 diff --git a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/FetchStream.java b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/FetchStream.java index 4f740539f806..b39629146e44 100644 --- a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/FetchStream.java +++ b/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"); diff --git a/solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java b/solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java index 6337f6261894..d777f65c7b4b 100644 --- a/solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java +++ b/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 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 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 {