You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by is...@apache.org on 2020/03/03 06:11:00 UTC
[lucene-solr] 01/02: SOLR-13350: Collector manager used for
multi-threaded search
This is an automated email from the ASF dual-hosted git repository.
ishan pushed a commit to branch jira/solr-13350-new
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit be9b8b092606df55f8c1823ef4323afc65ce79a2
Author: Ishan Chattopadhyaya <is...@apache.org>
AuthorDate: Mon Mar 2 21:43:19 2020 +0530
SOLR-13350: Collector manager used for multi-threaded search
---
.../org/apache/lucene/search/MultiCollector.java | 6 +
.../org/apache/solr/search/SolrIndexSearcher.java | 174 +++++++++++++++++----
.../test/org/apache/solr/search/TestFiltering.java | 18 ++-
3 files changed, 161 insertions(+), 37 deletions(-)
diff --git a/lucene/core/src/java/org/apache/lucene/search/MultiCollector.java b/lucene/core/src/java/org/apache/lucene/search/MultiCollector.java
index 82251e4..56965e6 100644
--- a/lucene/core/src/java/org/apache/lucene/search/MultiCollector.java
+++ b/lucene/core/src/java/org/apache/lucene/search/MultiCollector.java
@@ -20,6 +20,7 @@ package org.apache.lucene.search;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import org.apache.lucene.index.LeafReaderContext;
@@ -115,6 +116,11 @@ public class MultiCollector implements Collector {
return scoreMode;
}
+ // nocommit: need to raise a LUCENE jira for this?
+ public List<Collector> getCollectors() {
+ return Collections.unmodifiableList(Arrays.asList(collectors));
+ }
+
@Override
public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
final List<LeafCollector> leafCollectors = new ArrayList<>();
diff --git a/solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java b/solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java
index ae47aed..e671b89 100644
--- a/solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java
+++ b/solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java
@@ -31,6 +31,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -1611,24 +1612,39 @@ public class SolrIndexSearcher extends IndexSearcher implements Closeable, SolrI
// no docs on this page, so cursor doesn't change
qr.setNextCursorMark(cmd.getCursorMark());
} else {
- final TopDocsCollector topCollector = buildTopDocsCollector(len, cmd);
- MaxScoreCollector maxScoreCollector = null;
- Collector collector = topCollector;
- if ((cmd.getFlags() & GET_SCORES) != 0) {
- maxScoreCollector = new MaxScoreCollector();
- collector = MultiCollector.wrap(topCollector, maxScoreCollector);
+ TopDocs topDocs;
+ log.info("calling from 2, query: "+query.getClass()); // nocommit
+ if (pf.postFilter != null || cmd.getSegmentTerminateEarly() || cmd.getTimeAllowed() > 0
+ || query instanceof RankQuery) {
+ log.debug("skipping collector manager");
+ final TopDocsCollector topCollector = buildTopDocsCollector(len, cmd);
+ MaxScoreCollector maxScoreCollector = null;
+ Collector collector = topCollector;
+ if ((cmd.getFlags() & GET_SCORES) != 0) {
+ maxScoreCollector = new MaxScoreCollector();
+ collector = MultiCollector.wrap(topCollector, maxScoreCollector);
+ }
+ buildAndRunCollectorChain(qr, query, collector, cmd, pf.postFilter);
+ totalHits = topCollector.getTotalHits();
+ topDocs = topCollector.topDocs(0, len);
+
+ maxScore = totalHits > 0 ? (maxScoreCollector == null ? Float.NaN : maxScoreCollector.getMaxScore()) : 0.0f;
+ nDocsReturned = topDocs.scoreDocs.length;
+ } else {
+ log.debug("using collectormanager");
+ CollectorManagerResult result = searchCollectorManagers(len, cmd, query, true, true, false); // nocommit: need docset should be false
+ totalHits = result.totalHits;
+
+ maxScore = result.maxScore;
+ nDocsReturned = result.topDocs.scoreDocs.length;
+ topDocs = result.topDocs;
}
- buildAndRunCollectorChain(qr, query, collector, cmd, pf.postFilter);
- totalHits = topCollector.getTotalHits();
- TopDocs topDocs = topCollector.topDocs(0, len);
if (cmd.getSort() != null && query instanceof RankQuery == false && (cmd.getFlags() & GET_SCORES) != 0) {
TopFieldCollector.populateScores(topDocs.scoreDocs, this, query);
}
populateNextCursorMarkFromTopDocs(qr, cmd, topDocs);
- maxScore = totalHits > 0 ? (maxScoreCollector == null ? Float.NaN : maxScoreCollector.getMaxScore()) : 0.0f;
- nDocsReturned = topDocs.scoreDocs.length;
ids = new int[nDocsReturned];
scores = (cmd.getFlags() & GET_SCORES) != 0 ? new float[nDocsReturned] : null;
for (int i = 0; i < nDocsReturned; i++) {
@@ -1643,6 +1659,83 @@ public class SolrIndexSearcher extends IndexSearcher implements Closeable, SolrI
qr.setDocList(new DocSlice(0, sliceLen, ids, scores, totalHits, maxScore));
}
+ CollectorManagerResult searchCollectorManagers(int len, QueryCommand cmd, Query query,
+ boolean needTopDocs, boolean needMaxScore, boolean needDocSet) throws IOException {
+ CollectorManager<MultiCollector, CollectorManagerResult> manager = new CollectorManager<MultiCollector, CollectorManagerResult>() {
+ @Override
+ public MultiCollector newCollector() throws IOException {
+ // nocommit: Here, creating a MultiCollector for every segment (correctness > speed).
+ // Need to explore sharing a single MultiCollector with every segment. Are these
+ // sub-collectors thread-safe? DocSetCollector seems like not thread-safe, does someone know?
+ Collection<Collector> collectors = new ArrayList<Collector>();
+ if (needTopDocs) collectors.add(buildTopDocsCollector(len, cmd));
+ if (needMaxScore) collectors.add(new MaxScoreCollector());
+ if (needDocSet) collectors.add(new DocSetCollector(maxDoc()));
+ return (MultiCollector) MultiCollector.wrap(collectors);
+ }
+
+ @Override
+ public CollectorManagerResult reduce(Collection<MultiCollector> multiCollectors) throws IOException {
+ final TopDocs[] topDocs = new TopDocs[multiCollectors.size()];
+ float maxScore = 0.0f;
+ DocSet docSet = new BitDocSet(new FixedBitSet(maxDoc())); // TODO: if docset is not needed, avoid this initialization
+ int i = 0;
+ for (MultiCollector multiCollector: multiCollectors) {
+ int c = 0;
+ List<Collector> subCollectors = multiCollector.getCollectors();
+ TopDocsCollector topDocsCollector = needTopDocs? ((TopDocsCollector) subCollectors.get(c++)): null;
+ MaxScoreCollector maxScoreCollector = needMaxScore? ((MaxScoreCollector) subCollectors.get(c++)): null;
+ DocSetCollector docSetCollector = needDocSet? ((DocSetCollector) subCollectors.get(c++)): null;
+
+ if (needTopDocs) topDocs[i++] = (topDocsCollector instanceof TopFieldCollector)? ((TopFieldCollector)topDocsCollector).topDocs(0, len): topDocsCollector.topDocs(0, len);
+ if (needMaxScore)
+ if (!Float.isNaN(maxScoreCollector.getMaxScore()))
+ maxScore = Math.max(maxScore, maxScoreCollector.getMaxScore());
+ if (needDocSet) docSet = docSet.union(docSetCollector.getDocSet());
+ }
+ TopDocs mergedTopDocs;
+ if (topDocs != null && topDocs.length>0 && topDocs[0] instanceof TopFieldDocs) {
+ TopFieldDocs[] topFieldDocs = Arrays.copyOf(topDocs, topDocs.length, TopFieldDocs[].class);
+ mergedTopDocs = TopFieldDocs.merge(weightSort(cmd.getSort()), len, topFieldDocs);
+ } else {
+ mergedTopDocs = needTopDocs? TopDocs.merge(0, len, topDocs): null;
+ }
+ int totalHits = needTopDocs? (int)mergedTopDocs.totalHits.value: -1;
+ maxScore = totalHits > 0 ? maxScore : 0.0f;
+ return new CollectorManagerResult(mergedTopDocs, docSet, maxScore, totalHits);
+ }
+
+ };
+
+ CollectorManagerResult ret;
+ try {
+ ret = super.search(query, manager);
+ } catch (Exception ex) {
+ if (ex instanceof RuntimeException &&
+ ex.getCause() != null & ex.getCause() instanceof ExecutionException
+ && ex.getCause().getCause() != null && ex.getCause().getCause() instanceof RuntimeException) {
+ throw (RuntimeException)ex.getCause().getCause();
+ } else {
+ throw ex;
+ }
+ }
+ return ret;
+ }
+
+ class CollectorManagerResult {
+ TopDocs topDocs;
+ DocSet docSet;
+ float maxScore;
+ int totalHits;
+
+ public CollectorManagerResult(TopDocs topDocs, DocSet docSet, float maxScore, int totalHits) {
+ this.topDocs = topDocs;
+ this.docSet = docSet;
+ this.maxScore = maxScore;
+ this.totalHits = totalHits;
+ }
+ }
+
// any DocSet returned is for the query only, without any filtering... that way it may
// be cached if desired.
private DocSet getDocListAndSetNC(QueryResult qr, QueryCommand cmd) throws IOException {
@@ -1699,7 +1792,7 @@ public class SolrIndexSearcher extends IndexSearcher implements Closeable, SolrI
collector = MultiCollector.wrap(setCollector, topScoreCollector);
}
-
+ log.info("calling from 3"); // nocommit
buildAndRunCollectorChain(qr, query, collector, cmd, pf.postFilter);
set = DocSetUtil.getDocSet(setCollector, this);
@@ -1712,33 +1805,46 @@ public class SolrIndexSearcher extends IndexSearcher implements Closeable, SolrI
// no docs on this page, so cursor doesn't change
qr.setNextCursorMark(cmd.getCursorMark());
} else {
-
- final TopDocsCollector topCollector = buildTopDocsCollector(len, cmd);
- DocSetCollector setCollector = new DocSetCollector(maxDoc);
- MaxScoreCollector maxScoreCollector = null;
- List<Collector> collectors = new ArrayList<>(Arrays.asList(topCollector, setCollector));
-
- if ((cmd.getFlags() & GET_SCORES) != 0) {
- maxScoreCollector = new MaxScoreCollector();
- collectors.add(maxScoreCollector);
+ TopDocs topDocs;
+ if (pf.postFilter != null || cmd.getSegmentTerminateEarly() || cmd.getTimeAllowed() > 0 || query instanceof RankQuery) {
+
+ final TopDocsCollector topCollector = buildTopDocsCollector(len, cmd);
+ DocSetCollector setCollector = new DocSetCollector(maxDoc);
+ MaxScoreCollector maxScoreCollector = null;
+ List<Collector> collectors = new ArrayList<>(Arrays.asList(topCollector, setCollector));
+
+ if ((cmd.getFlags() & GET_SCORES) != 0) {
+ maxScoreCollector = new MaxScoreCollector();
+ collectors.add(maxScoreCollector);
+ }
+
+ Collector collector = MultiCollector.wrap(collectors);
+ log.info("calling from 4"); // nocommit
+ buildAndRunCollectorChain(qr, query, collector, cmd, pf.postFilter);
+
+ set = DocSetUtil.getDocSet(setCollector, this);
+
+ totalHits = topCollector.getTotalHits();
+ assert (totalHits == set.size()) || qr.isPartialResults();
+
+ topDocs = topCollector.topDocs(0, len);
+ maxScore = totalHits > 0 ? (maxScoreCollector == null ? Float.NaN : maxScoreCollector.getMaxScore()) : 0.0f;
+ nDocsReturned = topDocs.scoreDocs.length;
+ } else {
+ log.debug("using collectormanager");
+ CollectorManagerResult result = searchCollectorManagers(len, cmd, query, true, true, true);
+ set = result.docSet;
+ totalHits = result.totalHits;
+ assert (totalHits == set.size()) || qr.isPartialResults();
+ topDocs = result.topDocs;
+ maxScore = result.maxScore;
+ nDocsReturned = result.topDocs.scoreDocs.length;
}
-
- Collector collector = MultiCollector.wrap(collectors);
-
- buildAndRunCollectorChain(qr, query, collector, cmd, pf.postFilter);
-
- set = DocSetUtil.getDocSet(setCollector, this);
-
- totalHits = topCollector.getTotalHits();
- assert (totalHits == set.size()) || qr.isPartialResults();
-
- TopDocs topDocs = topCollector.topDocs(0, len);
+
if (cmd.getSort() != null && query instanceof RankQuery == false && (cmd.getFlags() & GET_SCORES) != 0) {
TopFieldCollector.populateScores(topDocs.scoreDocs, this, query);
}
populateNextCursorMarkFromTopDocs(qr, cmd, topDocs);
- maxScore = totalHits > 0 ? (maxScoreCollector == null ? Float.NaN : maxScoreCollector.getMaxScore()) : 0.0f;
- nDocsReturned = topDocs.scoreDocs.length;
ids = new int[nDocsReturned];
scores = (cmd.getFlags() & GET_SCORES) != 0 ? new float[nDocsReturned] : null;
diff --git a/solr/core/src/test/org/apache/solr/search/TestFiltering.java b/solr/core/src/test/org/apache/solr/search/TestFiltering.java
index b2ad9cf..6d5e081 100644
--- a/solr/core/src/test/org/apache/solr/search/TestFiltering.java
+++ b/solr/core/src/test/org/apache/solr/search/TestFiltering.java
@@ -24,6 +24,7 @@ import java.util.Locale;
import org.apache.lucene.search.DocIdSetIterator;
import org.apache.lucene.search.Query;
+import org.apache.lucene.util.Bits;
import org.apache.lucene.util.FixedBitSet;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.common.SolrException;
@@ -77,7 +78,7 @@ public class TestFiltering extends SolrTestCaseJ4 {
if (live == null) {
live = searcher.getLiveDocSet();
}
- assertTrue( set == live);
+ assertTrue( set.equals(live) );
QueryCommand cmd = new QueryCommand();
cmd.setQuery( QParser.getParser(qstr, null, req).getQuery() );
@@ -86,14 +87,19 @@ public class TestFiltering extends SolrTestCaseJ4 {
QueryResult res = new QueryResult();
searcher.search(res, cmd);
set = res.getDocSet();
- assertTrue( set == live );
+ System.out.println("Live: "+bitsString(live.getFixedBitSet()));
+ System.out.println("Set: "+bitsString(set.getFixedBitSet()));
+ FixedBitSet xor = live.getFixedBitSet().clone();
+ xor.xor(set.getFixedBitSet());
+ System.out.println("xor: "+bitsString(xor));
+ assertTrue( set.equals(live) );
cmd.setQuery( QParser.getParser(qstr + " OR id:0", null, req).getQuery() );
cmd.setFilterList( QParser.getParser(qstr + " OR id:1", null, req).getQuery() );
res = new QueryResult();
searcher.search(res, cmd);
set = res.getDocSet();
- assertTrue( set == live );
+ assertTrue( set.equals(live) );
}
} finally {
@@ -101,6 +107,12 @@ public class TestFiltering extends SolrTestCaseJ4 {
}
}
+ private String bitsString(Bits bits) {
+ String s = "";
+ for (int i=0; i<bits.length(); i++)
+ s+=bits.get(i)? 1: 0;
+ return s;
+ }
public void testCaching() throws Exception {
clearIndex();
assertU(adoc("id","4", "val_i","1"));