Updated Branches: refs/heads/apache-blur-0.2 179d53864 -> 36c5fd263
Made the performance of the faceting implementation much faster. The unit test sample of 1 million docs with 200 facets and a query that hit all 1 million and each of the facets hits all 1 million. The improvement went from ~5 seconds to 1.2 seconds with a single thread, and down to 750 ms with a threaded model. Also fixed the facets so that the minimums passed in actually work. So with minimums (set at 1000) in place the time drops to about 120 ms. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/36c5fd26 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/36c5fd26 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/36c5fd26 Branch: refs/heads/apache-blur-0.2 Commit: 36c5fd263bc7694b9ee701b70c042dc9075e896c Parents: 179d538 Author: Aaron McCurry <[email protected]> Authored: Sat Dec 21 15:43:26 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Sat Dec 21 15:43:26 2013 -0500 ---------------------------------------------------------------------- .../org/apache/blur/manager/IndexManager.java | 50 ++++- .../blur/lucene/search/FacetExecutor.java | 222 +++++++++++++++++++ .../apache/blur/lucene/search/FacetQuery.java | 114 ++++------ .../blur/lucene/search/FacetQueryTest.java | 163 +++++++++++--- .../blur/lucene/search/SuperQueryTest.java | 6 +- 5 files changed, 450 insertions(+), 105 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/36c5fd26/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java index 9915483..e9ed231 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java +++ b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java @@ -50,6 +50,7 @@ import org.apache.blur.index.ExitableReader; import org.apache.blur.index.ExitableReader.ExitingReaderException; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; +import org.apache.blur.lucene.search.FacetExecutor; import org.apache.blur.lucene.search.FacetQuery; import org.apache.blur.lucene.search.StopExecutionCollector.StopExecutionCollectorException; import org.apache.blur.manager.clusterstatus.ClusterStatus; @@ -69,6 +70,7 @@ import org.apache.blur.thrift.generated.BlurQuery; import org.apache.blur.thrift.generated.BlurQueryStatus; import org.apache.blur.thrift.generated.Column; import org.apache.blur.thrift.generated.ErrorType; +import org.apache.blur.thrift.generated.Facet; import org.apache.blur.thrift.generated.FetchResult; import org.apache.blur.thrift.generated.FetchRowResult; import org.apache.blur.thrift.generated.HighlightOptions; @@ -469,20 +471,35 @@ public class IndexManager { context); Query userQuery = QueryParserUtil.parseQuery(simpleQuery.query, simpleQuery.rowQuery, fieldManager, postFilter, preFilter, getScoreType(simpleQuery.scoreType), context); - Query facetedQuery = getFacetedQuery(blurQuery, userQuery, facetedCounts, fieldManager, context, postFilter, - preFilter); + + Query facetedQuery; + FacetExecutor executor = null; + if (blurQuery.facets != null) { + long[] facetMinimums = getFacetMinimums(blurQuery.facets); + executor = new FacetExecutor(blurQuery.facets.size(), facetMinimums, facetedCounts); + facetedQuery = new FacetQuery(userQuery, getFacetQueries(blurQuery, fieldManager, context, postFilter, + preFilter), executor); + } else { + facetedQuery = userQuery; + } + ReadInterceptor interceptor = context.getReadInterceptor(); call = new SimpleQueryParallelCall(running, table, status, facetedQuery, interceptor.getFilter(), blurQuery.selector, _queriesInternalMeter, shardServerContext, runSlow, _fetchCount, _maxHeapPerRowFetch, context.getSimilarity(), context); trace.done(); MergerBlurResultIterable merger = new MergerBlurResultIterable(blurQuery); - return ForkJoin.execute(_executor, blurIndexes.entrySet(), call, new Cancel() { + BlurResultIterable merge = ForkJoin.execute(_executor, blurIndexes.entrySet(), call, new Cancel() { @Override public void cancel() { running.set(false); } }).merge(merger); + + if (executor != null) { + executor.processFacets(null); + } + return merge; } catch (StopExecutionCollectorException e) { BlurQueryStatus queryStatus = status.getQueryStatus(); QueryState state = queryStatus.getState(); @@ -506,6 +523,25 @@ public class IndexManager { } } + private long[] getFacetMinimums(List<Facet> facets) { + long[] mins = new long[facets.size()]; + boolean smallerThanMaxLong = false; + for (int i = 0; i < facets.size(); i++) { + Facet facet = facets.get(i); + if (facet != null) { + long minimumNumberOfBlurResults = facet.getMinimumNumberOfBlurResults(); + mins[i] = minimumNumberOfBlurResults; + if (minimumNumberOfBlurResults < Long.MAX_VALUE) { + smallerThanMaxLong = true; + } + } + } + if (smallerThanMaxLong) { + return mins; + } + return null; + } + public String parseQuery(String table, org.apache.blur.thrift.generated.Query simpleQuery) throws ParseException, BlurException { TableContext context = getTableContext(table); @@ -523,14 +559,6 @@ public class IndexManager { return TableContext.create(_clusterStatus.getTableDescriptor(true, _clusterStatus.getCluster(true, table), table)); } - private Query getFacetedQuery(BlurQuery blurQuery, Query userQuery, AtomicLongArray counts, - FieldManager fieldManager, TableContext context, Filter postFilter, Filter preFilter) throws ParseException { - if (blurQuery.facets == null) { - return userQuery; - } - return new FacetQuery(userQuery, getFacetQueries(blurQuery, fieldManager, context, postFilter, preFilter), counts); - } - private Query[] getFacetQueries(BlurQuery blurQuery, FieldManager fieldManager, TableContext context, Filter postFilter, Filter preFilter) throws ParseException { int size = blurQuery.facets.size(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/36c5fd26/blur-query/src/main/java/org/apache/blur/lucene/search/FacetExecutor.java ---------------------------------------------------------------------- diff --git a/blur-query/src/main/java/org/apache/blur/lucene/search/FacetExecutor.java b/blur-query/src/main/java/org/apache/blur/lucene/search/FacetExecutor.java new file mode 100644 index 0000000..16cdba4 --- /dev/null +++ b/blur-query/src/main/java/org/apache/blur/lucene/search/FacetExecutor.java @@ -0,0 +1,222 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.blur.lucene.search; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLongArray; + +import org.apache.lucene.index.AtomicReader; +import org.apache.lucene.index.AtomicReaderContext; +import org.apache.lucene.search.Collector; +import org.apache.lucene.search.Scorer; +import org.apache.lucene.util.OpenBitSet; + +public class FacetExecutor { + + static Comparator<Entry<Object, Info>> COMPARATOR = new Comparator<Entry<Object, Info>>() { + @Override + public int compare(Entry<Object, Info> o1, Entry<Object, Info> o2) { + AtomicReader r1 = o1.getValue()._reader; + AtomicReader r2 = o2.getValue()._reader; + return r2.maxDoc() - r1.maxDoc(); + } + }; + + static class SimpleCollector extends Collector { + + int _hits; + final OpenBitSet _bitSet; + + SimpleCollector(OpenBitSet bitSet) { + _bitSet = bitSet; + } + + @Override + public void collect(int doc) throws IOException { + if (_bitSet.fastGet(doc)) { + _hits++; + } + } + + @Override + public void setScorer(Scorer scorer) throws IOException { + + } + + @Override + public void setNextReader(AtomicReaderContext context) throws IOException { + + } + + @Override + public boolean acceptsDocsOutOfOrder() { + return false; + } + + } + + static class Info { + final OpenBitSet _bitSet; + final Scorer[] _scorers; + final AtomicReader _reader; + + Info(AtomicReaderContext context, Scorer[] scorers) { + AtomicReader reader = context.reader(); + _bitSet = new OpenBitSet(reader.maxDoc()); + _scorers = scorers; + _reader = reader; + } + + void process(AtomicLongArray counts, long[] minimumsBeforeReturning) throws IOException { + SimpleCollector col = new SimpleCollector(_bitSet); + if (minimumsBeforeReturning == null) { + for (int i = 0; i < _scorers.length; i++) { + Scorer scorer = _scorers[i]; + if (scorer != null) { + scorer.score(col); + counts.addAndGet(i, col._hits); + } + col._hits = 0; + } + } else { + for (int i = 0; i < _scorers.length; i++) { + long min = minimumsBeforeReturning[i]; + long currentCount = counts.get(i); + if (currentCount < min) { + Scorer scorer = _scorers[i]; + if (scorer != null) { + scorer.score(col); + counts.addAndGet(i, col._hits); + } + counts.addAndGet(i, col._hits); + col._hits = 0; + } + } + } + } + } + + private final Map<Object, Info> _infoMap = new ConcurrentHashMap<Object, FacetExecutor.Info>(); + private final int _length; + private final AtomicLongArray _counts; + private final long[] _minimumsBeforeReturning; + private boolean _processed; + + public FacetExecutor(int length) { + this(length, null, new AtomicLongArray(length)); + } + + public FacetExecutor(int length, long[] minimumsBeforeReturning, AtomicLongArray counts) { + _length = length; + _counts = counts; + _minimumsBeforeReturning = minimumsBeforeReturning; + } + + public FacetExecutor(int length, long[] minimumsBeforeReturning) { + this(length, minimumsBeforeReturning, new AtomicLongArray(length)); + } + + public void addScorers(AtomicReaderContext context, Scorer[] scorers) throws IOException { + if (scorers.length != _length) { + throw new IOException("Scorer length is not correct expecting [" + _length + "] actual [" + scorers.length + "]"); + } + Object key = getKey(context); + Info info = _infoMap.get(key); + if (info == null) { + info = new Info(context, scorers); + _infoMap.put(key, info); + } else { + throw new IOException("Info about reader context [" + context + "] alread created."); + } + } + + private Object getKey(AtomicReaderContext context) { + return context.reader().getCoreCacheKey(); + } + + public OpenBitSet getBitSet(AtomicReaderContext context) throws IOException { + Info info = _infoMap.get(getKey(context)); + if (info == null) { + throw new IOException("Info object missing."); + } + return info._bitSet; + } + + public int length() { + return _length; + } + + public long get(int i) throws IOException { + if (!_processed) { + throw new IOException("Has not been processed."); + } + return _counts.get(i); + } + + public void processFacets(ExecutorService executor) throws IOException { + if (!_processed) { + processInternal(executor); + _processed = true; + } + } + + private void processInternal(ExecutorService executor) throws IOException { + List<Entry<Object, Info>> entries = new ArrayList<Entry<Object, Info>>(_infoMap.entrySet()); + Collections.sort(entries, COMPARATOR); + if (executor == null) { + for (Entry<Object, Info> e : entries) { + e.getValue().process(_counts, _minimumsBeforeReturning); + } + } else { + final AtomicInteger finished = new AtomicInteger(); + for (Entry<Object, Info> e : entries) { + final Entry<Object, Info> entry = e; + executor.submit(new Runnable() { + @Override + public void run() { + try { + entry.getValue().process(_counts, _minimumsBeforeReturning); + } catch (IOException e) { + throw new RuntimeException(e); + } finally { + finished.incrementAndGet(); + } + } + }); + } + + while (finished.get() < entries.size()) { + synchronized (this) { + try { + wait(1); + } catch (InterruptedException e) { + throw new IOException(e); + } + } + } + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/36c5fd26/blur-query/src/main/java/org/apache/blur/lucene/search/FacetQuery.java ---------------------------------------------------------------------- diff --git a/blur-query/src/main/java/org/apache/blur/lucene/search/FacetQuery.java b/blur-query/src/main/java/org/apache/blur/lucene/search/FacetQuery.java index 06fb712..aa9abda 100644 --- a/blur-query/src/main/java/org/apache/blur/lucene/search/FacetQuery.java +++ b/blur-query/src/main/java/org/apache/blur/lucene/search/FacetQuery.java @@ -17,7 +17,6 @@ package org.apache.blur.lucene.search; * limitations under the License. */ import java.io.IOException; -import java.util.concurrent.atomic.AtomicLongArray; import org.apache.lucene.index.AtomicReaderContext; import org.apache.lucene.index.IndexReader; @@ -27,22 +26,23 @@ import org.apache.lucene.search.Query; import org.apache.lucene.search.Scorer; import org.apache.lucene.search.Weight; import org.apache.lucene.util.Bits; +import org.apache.lucene.util.OpenBitSet; public class FacetQuery extends AbstractWrapperQuery { - private Query[] facets; - private AtomicLongArray counts; + private final Query[] _facets; + private final FacetExecutor _executor; - public FacetQuery(Query query, Query[] facets, AtomicLongArray counts) { + public FacetQuery(Query query, Query[] facets, FacetExecutor executor) { super(query, false); - this.facets = facets; - this.counts = counts; + _facets = facets; + _executor = executor; } - public FacetQuery(Query query, Query[] facets, AtomicLongArray counts, boolean rewritten) { + public FacetQuery(Query query, Query[] facets, FacetExecutor executor, boolean rewritten) { super(query, rewritten); - this.facets = facets; - this.counts = counts; + _facets = facets; + _executor = executor; } public String toString() { @@ -55,7 +55,7 @@ public class FacetQuery extends AbstractWrapperQuery { @Override public Query clone() { - return new FacetQuery((Query) _query.clone(), facets, counts, _rewritten); + return new FacetQuery((Query) _query.clone(), _facets, _executor, _rewritten); } @Override @@ -63,143 +63,127 @@ public class FacetQuery extends AbstractWrapperQuery { if (_rewritten) { return this; } - for (int i = 0; i < facets.length; i++) { - facets[i] = facets[i].rewrite(reader); + for (int i = 0; i < _facets.length; i++) { + _facets[i] = _facets[i].rewrite(reader); } - return new FacetQuery(_query.rewrite(reader), facets, counts, true); + return new FacetQuery(_query.rewrite(reader), _facets, _executor, true); } @Override public Weight createWeight(IndexSearcher searcher) throws IOException { Weight weight = _query.createWeight(searcher); - return new FacetWeight(weight, getWeights(searcher), counts); + return new FacetWeight(weight, getWeights(searcher), _executor); } private Weight[] getWeights(IndexSearcher searcher) throws IOException { - Weight[] weights = new Weight[facets.length]; + Weight[] weights = new Weight[_facets.length]; for (int i = 0; i < weights.length; i++) { - weights[i] = facets[i].createWeight(searcher); + weights[i] = _facets[i].createWeight(searcher); } return weights; } public static class FacetWeight extends Weight { - private Weight weight; - private Weight[] facets; - private AtomicLongArray counts; + private final Weight _weight; + private final Weight[] _facets; + private FacetExecutor _executor; - public FacetWeight(Weight weight, Weight[] facets, AtomicLongArray counts) { - this.weight = weight; - this.facets = facets; - this.counts = counts; + public FacetWeight(Weight weight, Weight[] facets, FacetExecutor executor) { + _weight = weight; + _facets = facets; + _executor = executor; } @Override public Explanation explain(AtomicReaderContext reader, int doc) throws IOException { - return weight.explain(reader, doc); + return _weight.explain(reader, doc); } @Override public Query getQuery() { - return weight.getQuery(); + return _weight.getQuery(); } @Override public void normalize(float norm, float topLevelBoost) { - weight.normalize(norm, topLevelBoost); + _weight.normalize(norm, topLevelBoost); } @Override - public Scorer scorer(AtomicReaderContext context, boolean scoreDocsInOrder, boolean topScorer, Bits acceptDocs) throws IOException { - Scorer scorer = weight.scorer(context, true, topScorer, acceptDocs); + public Scorer scorer(AtomicReaderContext context, boolean scoreDocsInOrder, boolean topScorer, Bits acceptDocs) + throws IOException { + Scorer scorer = _weight.scorer(context, true, topScorer, acceptDocs); if (scorer == null) { return null; } - return new FacetScorer(scorer, getScorers(context, true, topScorer, acceptDocs), counts); + Scorer[] scorers = getScorers(context, true, topScorer, acceptDocs); + _executor.addScorers(context, scorers); + return new FacetScorer(scorer, _executor, context); } - private Scorer[] getScorers(AtomicReaderContext context, boolean scoreDocsInOrder, boolean topScorer, Bits acceptDocs) throws IOException { - Scorer[] scorers = new Scorer[facets.length]; + private Scorer[] getScorers(AtomicReaderContext context, boolean scoreDocsInOrder, boolean topScorer, + Bits acceptDocs) throws IOException { + Scorer[] scorers = new Scorer[_facets.length]; for (int i = 0; i < scorers.length; i++) { - scorers[i] = facets[i].scorer(context, scoreDocsInOrder, topScorer, acceptDocs); + scorers[i] = _facets[i].scorer(context, scoreDocsInOrder, topScorer, acceptDocs); } return scorers; } @Override public float getValueForNormalization() throws IOException { - return weight.getValueForNormalization(); + return _weight.getValueForNormalization(); } } public static class FacetScorer extends Scorer { - private Scorer baseScorer; - private Scorer[] facets; - private AtomicLongArray counts; - private int facetLength; + private final Scorer _baseScorer; + private final OpenBitSet _hit; - public FacetScorer(Scorer scorer, Scorer[] facets, AtomicLongArray counts) { + public FacetScorer(Scorer scorer, FacetExecutor executor, AtomicReaderContext context) throws IOException { super(scorer.getWeight()); - this.baseScorer = scorer; - this.facets = facets; - this.counts = counts; - this.facetLength = facets.length; + _baseScorer = scorer; + _hit = executor.getBitSet(context); } private int processFacets(int doc) throws IOException { if (doc == NO_MORE_DOCS) { return doc; } - for (int i = 0; i < facetLength; i++) { - Scorer facet = facets[i]; - if (facet == null) { - continue; - } - int docID = facet.docID(); - if (docID == NO_MORE_DOCS) { - continue; - } - if (docID == doc) { - counts.incrementAndGet(i); - } else if (docID < doc) { - if (facet.advance(doc) == doc) { - counts.incrementAndGet(i); - } - } - } + _hit.fastSet(doc); return doc; } @Override public float score() throws IOException { - return baseScorer.score(); + return _baseScorer.score(); } @Override public int advance(int target) throws IOException { - return processFacets(baseScorer.advance(target)); + return processFacets(_baseScorer.advance(target)); } @Override public int docID() { - return baseScorer.docID(); + return _baseScorer.docID(); } @Override public int nextDoc() throws IOException { - return processFacets(baseScorer.nextDoc()); + return processFacets(_baseScorer.nextDoc()); } @Override public int freq() throws IOException { - return baseScorer.freq(); + return _baseScorer.freq(); } @Override public long cost() { - return baseScorer.cost(); + return _baseScorer.cost(); } } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/36c5fd26/blur-query/src/test/java/org/apache/blur/lucene/search/FacetQueryTest.java ---------------------------------------------------------------------- diff --git a/blur-query/src/test/java/org/apache/blur/lucene/search/FacetQueryTest.java b/blur-query/src/test/java/org/apache/blur/lucene/search/FacetQueryTest.java index f0f7cd7..28e2c35 100644 --- a/blur-query/src/test/java/org/apache/blur/lucene/search/FacetQueryTest.java +++ b/blur-query/src/test/java/org/apache/blur/lucene/search/FacetQueryTest.java @@ -18,9 +18,14 @@ package org.apache.blur.lucene.search; */ import static org.apache.blur.lucene.LuceneVersionConstant.LUCENE_VERSION; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import java.io.File; import java.io.IOException; -import java.util.concurrent.atomic.AtomicLongArray; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.apache.lucene.analysis.core.KeywordAnalyzer; import org.apache.lucene.document.Document; @@ -37,29 +42,17 @@ import org.apache.lucene.search.BooleanQuery; import org.apache.lucene.search.IndexSearcher; import org.apache.lucene.search.Query; import org.apache.lucene.search.TermQuery; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FSDirectory; import org.apache.lucene.store.LockObtainFailedException; import org.apache.lucene.store.RAMDirectory; -import org.junit.After; -import org.junit.Before; import org.junit.Test; - public class FacetQueryTest { - private IndexReader reader; - - @Before - public void setup() throws CorruptIndexException, LockObtainFailedException, IOException { - reader = createIndex(); - } - - @After - public void tearDown() { - - } - @Test - public void testFacetQueryNoSuper() throws IOException { + public void testFacetQueryNoSuper() throws IOException, InterruptedException { + IndexReader reader = createIndex(10, 0, true); BooleanQuery bq = new BooleanQuery(); bq.add(new TermQuery(new Term("f1", "value")), Occur.SHOULD); bq.add(new TermQuery(new Term("f2", "v3")), Occur.SHOULD); @@ -72,35 +65,153 @@ public class FacetQueryTest { Query[] facets = new Query[] { f1, f2 }; - AtomicLongArray counts = new AtomicLongArray(facets.length); - FacetQuery facetQuery = new FacetQuery(bq, facets, counts); + FacetExecutor facetExecutor = new FacetExecutor(facets.length); + FacetQuery facetQuery = new FacetQuery(bq, facets, facetExecutor); IndexSearcher indexSearcher = new IndexSearcher(reader); indexSearcher.search(facetQuery, 10); - //@TODO add actual assertion - for (int i = 0; i < counts.length(); i++) { - System.out.println(counts.get(i)); + ExecutorService executor = Executors.newCachedThreadPool(); + facetExecutor.processFacets(executor); + executor.shutdown(); + executor.awaitTermination(10, TimeUnit.SECONDS); + + for (int i = 0; i < facetExecutor.length(); i++) { + assertEquals(1L, facetExecutor.get(i)); + } + } + + @Test + public void testFacetQueryPerformance() throws IOException, InterruptedException { + int facetCount = 200; + int docCount = 1000000; + IndexReader reader = createIndex(docCount, facetCount, false); + + Query[] facets = new Query[facetCount]; + for (int i = 0; i < facetCount; i++) { + facets[i] = new TermQuery(new Term("facet" + i, "value")); + } + + ExecutorService executor = null; + try { + for (int t = 0; t < 5; t++) { + executor = Executors.newCachedThreadPool(); + IndexSearcher indexSearcher = new IndexSearcher(reader, executor); + FacetExecutor facetExecutor = new FacetExecutor(facets.length); + FacetQuery facetQuery = new FacetQuery(new TermQuery(new Term("f1", "value")), facets, facetExecutor); + long t1 = System.nanoTime(); + indexSearcher.search(facetQuery, 10); + facetExecutor.processFacets(executor); + executor.shutdown(); + executor.awaitTermination(10, TimeUnit.SECONDS); + long t2 = System.nanoTime(); + System.out.println((t2 - t1) / 1000000.0); + + for (int i = 0; i < facetExecutor.length(); i++) { + assertEquals((long) docCount, facetExecutor.get(i)); + } + } + } finally { + executor.shutdownNow(); } } - private IndexReader createIndex() throws CorruptIndexException, LockObtainFailedException, IOException { - RAMDirectory directory = new RAMDirectory(); + @Test + public void testFacetQueryPerformanceWithMins() throws IOException, InterruptedException { + int facetCount = 200; + int docCount = 1000000; + IndexReader reader = createIndex(docCount, facetCount, false); + + Query[] facets = new Query[facetCount]; + for (int i = 0; i < facetCount; i++) { + facets[i] = new TermQuery(new Term("facet" + i, "value")); + } + + long min = 1000; + long[] minimumsBeforeReturning = new long[facets.length]; + for (int i = 0; i < minimumsBeforeReturning.length; i++) { + minimumsBeforeReturning[i] = min; + } + + ExecutorService executor = null; + try { + for (int t = 0; t < 5; t++) { + executor = Executors.newCachedThreadPool(); + IndexSearcher indexSearcher = new IndexSearcher(reader, executor); + FacetExecutor facetExecutor = new FacetExecutor(facets.length, minimumsBeforeReturning); + FacetQuery facetQuery = new FacetQuery(new TermQuery(new Term("f1", "value")), facets, facetExecutor); + long t1 = System.nanoTime(); + indexSearcher.search(facetQuery, 10); + facetExecutor.processFacets(executor); + executor.shutdown(); + executor.awaitTermination(10, TimeUnit.SECONDS); + long t2 = System.nanoTime(); + System.out.println((t2 - t1) / 1000000.0); + + for (int i = 0; i < facetExecutor.length(); i++) { + + assertTrue(facetExecutor.get(i) >= min); + } + } + } finally { + executor.shutdownNow(); + } + } + + private IndexReader createIndex(int docCount, int facetFields, boolean ram) throws CorruptIndexException, + LockObtainFailedException, IOException { + Directory directory; + if (ram) { + directory = new RAMDirectory(); + } else { + File dir = new File("./target/tmp/facet_tmp"); + if (dir.exists()) { + directory = FSDirectory.open(dir); + DirectoryReader reader = DirectoryReader.open(directory); + if (reader.numDocs() == docCount) { + return reader; + } + reader.close(); + directory.close(); + } + rmr(dir); + directory = FSDirectory.open(dir); + } IndexWriterConfig conf = new IndexWriterConfig(LUCENE_VERSION, new KeywordAnalyzer()); IndexWriter writer = new IndexWriter(directory, conf); FieldType fieldType = new FieldType(); fieldType.setStored(true); fieldType.setIndexed(true); fieldType.setOmitNorms(true); - for (int i = 0; i < 10; i++) { + long start = System.nanoTime(); + for (int i = 0; i < docCount; i++) { + long now = System.nanoTime(); + if (start + TimeUnit.SECONDS.toNanos(5) < now) { + System.out.println("Indexing doc " + i + " of " + docCount); + start = System.nanoTime(); + } Document document = new Document(); - document.add(new Field("f1", "value", fieldType)); document.add(new Field("f2", "v" + i, fieldType)); + for (int f = 0; f < facetFields; f++) { + document.add(new Field("facet" + f, "value", fieldType)); + } writer.addDocument(document); } writer.close(); return DirectoryReader.open(directory); } + private void rmr(File file) { + if (!file.exists()) { + return; + } + if (file.isDirectory()) { + for (File f : file.listFiles()) { + rmr(f); + } + } + file.delete(); + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/36c5fd26/blur-query/src/test/java/org/apache/blur/lucene/search/SuperQueryTest.java ---------------------------------------------------------------------- diff --git a/blur-query/src/test/java/org/apache/blur/lucene/search/SuperQueryTest.java b/blur-query/src/test/java/org/apache/blur/lucene/search/SuperQueryTest.java index 54ee490..8cf4bde 100644 --- a/blur-query/src/test/java/org/apache/blur/lucene/search/SuperQueryTest.java +++ b/blur-query/src/test/java/org/apache/blur/lucene/search/SuperQueryTest.java @@ -23,7 +23,6 @@ import static org.apache.blur.lucene.LuceneVersionConstant.LUCENE_VERSION; import java.io.IOException; import java.util.Arrays; import java.util.List; -import java.util.concurrent.atomic.AtomicLongArray; import org.apache.blur.thrift.generated.ScoreType; import org.apache.lucene.analysis.standard.StandardAnalyzer; @@ -141,8 +140,9 @@ public class SuperQueryTest { f1.add(new TermQuery(new Term(PERSON_NAME, NAME2)), Occur.MUST); Query[] facets = new Query[] { new SuperQuery(f1, ScoreType.CONSTANT, new Term(PRIME_DOC, PRIME_DOC_VALUE)) }; - AtomicLongArray counts = new AtomicLongArray(facets.length); - FacetQuery query = new FacetQuery(booleanQuery, facets, counts); + FacetExecutor executor = new FacetExecutor(facets.length); + FacetQuery query = new FacetQuery(booleanQuery, facets, executor); + executor.processFacets(null); TopDocs topDocs = searcher.search(query, 10); assertEquals(3, topDocs.totalHits);
