You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by va...@apache.org on 2018/08/19 19:15:31 UTC
lucene-solr:master: SOLR-12674: RollupStream should not use the
HashQueryParser for 1 worker
Repository: lucene-solr
Updated Branches:
refs/heads/master d6e1d4a4b -> 3e4545219
SOLR-12674: RollupStream should not use the HashQueryParser for 1 worker
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/3e454521
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/3e454521
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/3e454521
Branch: refs/heads/master
Commit: 3e4545219ef72e89d227b46a26a951f621989d40
Parents: d6e1d4a
Author: Varun Thacker <va...@apache.org>
Authored: Sun Aug 19 11:58:14 2018 -0700
Committer: Varun Thacker <va...@apache.org>
Committed: Sun Aug 19 11:59:38 2018 -0700
----------------------------------------------------------------------
solr/CHANGES.txt | 2 +
.../apache/solr/search/HashQParserPlugin.java | 4 ++
.../solr/search/TestHashQParserPlugin.java | 4 +-
.../solr/client/solrj/io/stream/SolrStream.java | 6 +--
.../client/solrj/io/stream/StreamingTest.java | 57 ++++++++++++++++++++
5 files changed, 67 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3e454521/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 70f1d8a..785dada 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -245,6 +245,8 @@ Bug Fixes
* SOLR-12475: Fix MaxSizeAutoCommitTest failures (Rupa Shankar, Anshum Gupta)
+* SOLR-12674: RollupStream should not use the HashQueryParser for 1 worker. (Varun Thacker)
+
Optimizations
----------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3e454521/solr/core/src/java/org/apache/solr/search/HashQParserPlugin.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/search/HashQParserPlugin.java b/solr/core/src/java/org/apache/solr/search/HashQParserPlugin.java
index bd8f425..a499a10 100644
--- a/solr/core/src/java/org/apache/solr/search/HashQParserPlugin.java
+++ b/solr/core/src/java/org/apache/solr/search/HashQParserPlugin.java
@@ -39,6 +39,7 @@ import org.apache.lucene.util.BytesRef;
import org.apache.lucene.util.CharsRef;
import org.apache.lucene.util.CharsRefBuilder;
import org.apache.lucene.util.FixedBitSet;
+import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.request.SolrQueryRequest;
import org.apache.solr.schema.FieldType;
@@ -68,6 +69,9 @@ public class HashQParserPlugin extends QParserPlugin {
public Query parse() {
int workers = localParams.getInt("workers", 0);
+ if (workers < 2) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "workers needs to be more than 1");
+ }
int worker = localParams.getInt("worker", 0);
String keys = params.get("partitionKeys");
keys = keys.replace(" ", "");
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3e454521/solr/core/src/test/org/apache/solr/search/TestHashQParserPlugin.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/search/TestHashQParserPlugin.java b/solr/core/src/test/org/apache/solr/search/TestHashQParserPlugin.java
index 6f68906..06c2097 100644
--- a/solr/core/src/test/org/apache/solr/search/TestHashQParserPlugin.java
+++ b/solr/core/src/test/org/apache/solr/search/TestHashQParserPlugin.java
@@ -66,7 +66,7 @@ public class TestHashQParserPlugin extends SolrTestCaseJ4 {
//Test with string hash
ModifiableSolrParams params = new ModifiableSolrParams();
params.add("q", "*:*");
- params.add("fq", "{!hash worker=0 workers=1 cost="+getCost(random())+"}");
+ params.add("fq", "{!hash worker=0 workers=2 cost="+getCost(random())+"}");
params.add("partitionKeys", "a_s");
params.add("wt", "xml");
String response = h.query(req(params));
@@ -75,7 +75,7 @@ public class TestHashQParserPlugin extends SolrTestCaseJ4 {
//Test with int hash
params = new ModifiableSolrParams();
params.add("q", "*:*");
- params.add("fq", "{!hash worker=0 workers=1 cost="+getCost(random())+"}");
+ params.add("fq", "{!hash worker=0 workers=2 cost="+getCost(random())+"}");
params.add("partitionKeys", "a_i");
params.add("wt", "xml");
response = h.query(req(params));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3e454521/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
index a126aa9..74c6bfd 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/SolrStream.java
@@ -131,14 +131,12 @@ public class SolrStream extends TupleStream {
private SolrParams loadParams(SolrParams paramsIn) throws IOException {
ModifiableSolrParams solrParams = new ModifiableSolrParams(paramsIn);
if (params.get("partitionKeys") != null) {
- if(!params.get("partitionKeys").equals("none")) {
+ if(!params.get("partitionKeys").equals("none") && numWorkers > 1) {
String partitionFilter = getPartitionFilter();
solrParams.add("fq", partitionFilter);
}
- } else {
- if(numWorkers > 1) {
+ } else if(numWorkers > 1) {
throw new IOException("When numWorkers > 1 partitionKeys must be set. Set partitionKeys=none to send the entire stream to each worker.");
- }
}
if(checkpoint > 0) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/3e454521/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java
index 9ee6688..c444b6c 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/io/stream/StreamingTest.java
@@ -1785,6 +1785,63 @@ public void testParallelRankStream() throws Exception {
}
@Test
+ public void testRollupWithNoParallel() throws Exception {
+
+ new UpdateRequest()
+ .add(id, "0", "a_s", "hello0", "a_i", "0", "a_f", "1")
+ .add(id, "2", "a_s", "hello0", "a_i", "2", "a_f", "2")
+ .add(id, "3", "a_s", "hello3", "a_i", "3", "a_f", "3")
+ .add(id, "4", "a_s", "hello4", "a_i", "4", "a_f", "4")
+ .add(id, "1", "a_s", "hello0", "a_i", "1", "a_f", "5")
+ .add(id, "5", "a_s", "hello3", "a_i", "10", "a_f", "6")
+ .add(id, "6", "a_s", "hello4", "a_i", "11", "a_f", "7")
+ .add(id, "7", "a_s", "hello3", "a_i", "12", "a_f", "8")
+ .add(id, "8", "a_s", "hello3", "a_i", "13", "a_f", "9")
+ .add(id, "9", "a_s", "hello0", "a_i", "14", "a_f", "10")
+ .commit(cluster.getSolrClient(), COLLECTIONORALIAS);
+
+ StreamContext streamContext = new StreamContext();
+ SolrClientCache solrClientCache = new SolrClientCache();
+ streamContext.setSolrClientCache(solrClientCache);
+
+ try {
+ //Intentionally adding partitionKeys to trigger SOLR-12674
+ SolrParams sParamsA = mapParams("q", "*:*", "fl", "a_s,a_i,a_f", "sort", "a_s asc", "partitionKeys", "a_s");
+ CloudSolrStream stream = new CloudSolrStream(zkHost, COLLECTIONORALIAS, sParamsA);
+
+ Bucket[] buckets = {new Bucket("a_s")};
+
+ Metric[] metrics = {new SumMetric("a_i"),
+ new SumMetric("a_f"),
+ new MinMetric("a_i"),
+ new MinMetric("a_f"),
+ new MaxMetric("a_i"),
+ new MaxMetric("a_f"),
+ new MeanMetric("a_i"),
+ new MeanMetric("a_f"),
+ new CountMetric()};
+
+ RollupStream rollupStream = new RollupStream(stream, buckets, metrics);
+ rollupStream.setStreamContext(streamContext);
+ List<Tuple> tuples = getTuples(rollupStream);
+ assertEquals(3, tuples.size());
+
+
+ List<String> shardUrls = TupleStream.getShards(cluster.getZkServer().getZkAddress(), COLLECTIONORALIAS, streamContext);
+ ModifiableSolrParams solrParams = new ModifiableSolrParams();
+ solrParams.add("qt", "/stream");
+ solrParams.add("expr", "rollup(search(" + COLLECTIONORALIAS + ",q=\"*:*\",fl=\"a_s,a_i,a_f\",sort=\"a_s desc\",partitionKeys=\"a_s\"),over=\"a_s\")\n");
+ SolrStream solrStream = new SolrStream(shardUrls.get(0), solrParams);
+ streamContext = new StreamContext();
+ solrStream.setStreamContext(streamContext);
+ tuples = getTuples(solrStream);
+ assert (tuples.size() == 3);
+ } finally {
+ solrClientCache.close();
+ }
+ }
+
+ @Test
public void testParallelRollupStream() throws Exception {
new UpdateRequest()