Fixing the double fetch is the shard server is accessed directly and the beginnings of the sorting feature has been added.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/3b36b782 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/3b36b782 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/3b36b782 Branch: refs/heads/apache-blur-0.2 Commit: 3b36b782ecdc0d3dd2a579f01b903d9d52093c11 Parents: c18ed74 Author: Aaron McCurry <[email protected]> Authored: Sun Jan 26 14:58:24 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Sun Jan 26 14:58:24 2014 -0500 ---------------------------------------------------------------------- .../org/apache/blur/manager/IndexManager.java | 56 ++++++++++- .../results/BlurResultIterableSearcher.java | 68 +++---------- .../org/apache/blur/thrift/BlurShardServer.java | 5 +- .../blur/lucene/search/IterablePaging.java | 21 +++- .../lucene/search/TestingPagingCollector.java | 100 +++++++++++++++++-- .../apache/blur/lucene/codec/Blur022Codec.java | 2 +- 6 files changed, 185 insertions(+), 67 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3b36b782/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 e70d095..83f2b25 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 @@ -71,6 +71,7 @@ import org.apache.blur.thrift.MutationHelper; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.BlurQuery; import org.apache.blur.thrift.generated.BlurQueryStatus; +import org.apache.blur.thrift.generated.BlurResult; import org.apache.blur.thrift.generated.ErrorType; import org.apache.blur.thrift.generated.Facet; import org.apache.blur.thrift.generated.FetchResult; @@ -89,6 +90,7 @@ import org.apache.blur.trace.Trace; import org.apache.blur.trace.Tracer; import org.apache.blur.utils.BlurExecutorCompletionService; import org.apache.blur.utils.BlurExecutorCompletionService.Cancel; +import org.apache.blur.utils.BlurIterator; import org.apache.blur.utils.BlurUtil; import org.apache.blur.utils.ForkJoin; import org.apache.blur.utils.ForkJoin.Merger; @@ -507,7 +509,7 @@ public class IndexManager { if (executor != null) { executor.processFacets(_facetExecutor); } - return merge; + return fetchDataIfNeeded(merge, table, blurQuery.getSelector()); } catch (StopExecutionCollectorException e) { BlurQueryStatus queryStatus = status.getQueryStatus(); QueryState state = queryStatus.getState(); @@ -531,6 +533,58 @@ public class IndexManager { } } + private BlurResultIterable fetchDataIfNeeded(final BlurResultIterable iterable, final String table, final Selector selector) { + if (selector == null) { + return iterable; + } + return new BlurResultIterable() { + + @Override + public BlurIterator<BlurResult, BlurException> iterator() throws BlurException { + final BlurIterator<BlurResult, BlurException> iterator = iterable.iterator(); + return new BlurIterator<BlurResult, BlurException>() { + + @Override + public BlurResult next() throws BlurException { + BlurResult result = iterator.next(); + String locationId = result.getLocationId(); + FetchResult fetchResult = new FetchResult(); + Selector s = new Selector(selector); + s.setLocationId(locationId); + fetchRow(table, s, fetchResult); + result.setFetchResult(fetchResult); + return result; + } + + @Override + public boolean hasNext() throws BlurException { + return iterator.hasNext(); + } + }; + } + + @Override + public void close() throws IOException { + iterable.close(); + } + + @Override + public void skipTo(long skipTo) { + iterable.skipTo(skipTo); + } + + @Override + public long getTotalResults() { + return iterable.getTotalResults(); + } + + @Override + public Map<String, Long> getShardInfo() { + return iterable.getShardInfo(); + } + }; + } + private long[] getFacetMinimums(List<Facet> facets) { long[] mins = new long[facets.size()]; boolean smallerThanMaxLong = false; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3b36b782/blur-core/src/main/java/org/apache/blur/manager/results/BlurResultIterableSearcher.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/results/BlurResultIterableSearcher.java b/blur-core/src/main/java/org/apache/blur/manager/results/BlurResultIterableSearcher.java index 7e5b371..6f669ab 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/results/BlurResultIterableSearcher.java +++ b/blur-core/src/main/java/org/apache/blur/manager/results/BlurResultIterableSearcher.java @@ -24,13 +24,10 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.apache.blur.lucene.search.IterablePaging; import org.apache.blur.lucene.search.IterablePaging.ProgressRef; import org.apache.blur.lucene.search.IterablePaging.TotalHitsRef; -import org.apache.blur.manager.IndexManager; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.TableContext; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.BlurResult; -import org.apache.blur.thrift.generated.ErrorType; -import org.apache.blur.thrift.generated.FetchResult; import org.apache.blur.thrift.generated.Selector; import org.apache.blur.utils.BlurIterator; import org.apache.blur.utils.Converter; @@ -41,73 +38,44 @@ import org.apache.lucene.search.ScoreDoc; public class BlurResultIterableSearcher implements BlurResultIterable { - private Map<String, Long> _shardInfo = new TreeMap<String, Long>(); - private String _shard; - private long _skipTo; - private String _table; + private final Map<String, Long> _shardInfo = new TreeMap<String, Long>(); + private final String _shard; private final int _fetchCount; - - private IteratorConverter<ScoreDoc, BlurResult, BlurException> _iterator; - private final Selector _selector; + private final IteratorConverter<ScoreDoc, BlurResult, BlurException> _iteratorConverter; private final Query _query; - private IndexSearcherClosable _searcher; private final TotalHitsRef _totalHitsRef = new TotalHitsRef(); private final ProgressRef _progressRef = new ProgressRef(); private final AtomicBoolean _running; private final boolean _closeSearcher; private final boolean _runSlow; - private final int _maxHeapPerRowFetch; - private final TableContext _context; - private final Filter _filter; + private final IterablePaging _iterablePaging; + + private IndexSearcherClosable _searcher; + private long _skipTo; public BlurResultIterableSearcher(AtomicBoolean running, Query query, String table, String shard, IndexSearcherClosable searcher, Selector selector, boolean closeSearcher, boolean runSlow, int fetchCount, int maxHeapPerRowFetch, TableContext context, Filter filter) throws BlurException { _running = running; - _table = table; _query = query; _shard = shard; _searcher = searcher; - _selector = selector; _closeSearcher = closeSearcher; _runSlow = runSlow; _fetchCount = fetchCount; - _maxHeapPerRowFetch = maxHeapPerRowFetch; - _context = context; - _filter = filter; - performSearch(); - } - - private void performSearch() throws BlurException { - IterablePaging iterablePaging = new IterablePaging(_running, _searcher, _query, _fetchCount, _totalHitsRef, - _progressRef, _runSlow); - _iterator = new IteratorConverter<ScoreDoc, BlurResult, BlurException>(iterablePaging.iterator(), + _iterablePaging = new IterablePaging(_running, _searcher, _query, _fetchCount, _totalHitsRef, _progressRef, + _runSlow); + _iteratorConverter = new IteratorConverter<ScoreDoc, BlurResult, BlurException>(_iterablePaging.iterator(), new Converter<ScoreDoc, BlurResult, BlurException>() { @Override public BlurResult convert(ScoreDoc scoreDoc) throws BlurException { - String resolveId = resolveId(scoreDoc.doc); - return new BlurResult(resolveId, scoreDoc.score, getFetchResult(resolveId)); + String resolveId = resolveId(scoreDoc); + return new BlurResult(resolveId, scoreDoc.score, null); } }); _shardInfo.put(_shard, (long) _totalHitsRef.totalHits()); } - private FetchResult getFetchResult(String resolveId) throws BlurException { - if (_selector == null) { - return null; - } - FetchResult fetchResult = new FetchResult(); - _selector.setLocationId(resolveId); - IndexManager.validSelector(_selector); - try { - IndexManager.fetchRow(_searcher.getIndexReader(), _table, _shard, _selector, fetchResult, null, - _maxHeapPerRowFetch, _context, _filter); - } catch (IOException e) { - throw new BlurException("Unknown IO error", null, ErrorType.UNKNOWN); - } - return fetchResult; - } - @Override public Map<String, Long> getShardInfo() { return _shardInfo; @@ -125,16 +93,12 @@ public class BlurResultIterableSearcher implements BlurResultIterable { @Override public BlurIterator<BlurResult, BlurException> iterator() throws BlurException { - long start = 0; - while (_iterator.hasNext() && start < _skipTo) { - _iterator.next(); - start++; - } - return _iterator; + _iterablePaging.skipTo((int) _skipTo); + return _iteratorConverter; } - private String resolveId(int docId) { - return _shard + "/" + docId; + private String resolveId(ScoreDoc scoreDoc) { + return _shard + "/" + scoreDoc.doc; } @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3b36b782/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java index ecd01c0..3532429 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java @@ -125,8 +125,9 @@ public class BlurShardServer extends TableAdmin implements Iface { try { AtomicLongArray facetCounts = BlurUtil.getAtomicLongArraySameLengthAsList(blurQuery.facets); hitsIterable = _indexManager.query(table, blurQuery, facetCounts); - BlurResults blurResults = BlurUtil.convertToHits(hitsIterable, blurQuery, facetCounts, _dataFetch, - blurQuery.selector, this, table); + // Data will be fetch by IndexManager if selector is provided. + // This should only happen if the Shard server is accessed directly. + BlurResults blurResults = BlurUtil.convertToHits(hitsIterable, blurQuery, facetCounts, null, null, this, table); if (selector != null) { return blurResults; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3b36b782/blur-query/src/main/java/org/apache/blur/lucene/search/IterablePaging.java ---------------------------------------------------------------------- diff --git a/blur-query/src/main/java/org/apache/blur/lucene/search/IterablePaging.java b/blur-query/src/main/java/org/apache/blur/lucene/search/IterablePaging.java index 8df4a99..15fd094 100644 --- a/blur-query/src/main/java/org/apache/blur/lucene/search/IterablePaging.java +++ b/blur-query/src/main/java/org/apache/blur/lucene/search/IterablePaging.java @@ -29,10 +29,14 @@ import org.apache.blur.thrift.generated.ErrorType; import org.apache.blur.utils.BlurIterable; import org.apache.blur.utils.BlurIterator; import org.apache.lucene.search.Collector; +import org.apache.lucene.search.FieldDoc; import org.apache.lucene.search.IndexSearcher; import org.apache.lucene.search.Query; import org.apache.lucene.search.ScoreDoc; +import org.apache.lucene.search.Sort; import org.apache.lucene.search.TopDocs; +import org.apache.lucene.search.TopDocsCollector; +import org.apache.lucene.search.TopFieldCollector; import org.apache.lucene.search.TopScoreDocCollector; /** @@ -45,6 +49,7 @@ public class IterablePaging implements BlurIterable<ScoreDoc, BlurException> { private final AtomicBoolean _running; private final int _numHitsToCollect; private final boolean _runSlow; + private final Sort _sort; private TotalHitsRef _totalHitsRef; private ProgressRef _progressRef; @@ -53,7 +58,13 @@ public class IterablePaging implements BlurIterable<ScoreDoc, BlurException> { public IterablePaging(AtomicBoolean running, IndexSearcher searcher, Query query, int numHitsToCollect, TotalHitsRef totalHitsRef, ProgressRef progressRef, boolean runSlow) throws BlurException { + this(running, searcher, query, numHitsToCollect, totalHitsRef, progressRef, runSlow, null); + } + + public IterablePaging(AtomicBoolean running, IndexSearcher searcher, Query query, int numHitsToCollect, + TotalHitsRef totalHitsRef, ProgressRef progressRef, boolean runSlow, Sort sort) throws BlurException { _running = running; + _sort = sort; try { _query = searcher.rewrite(query); } catch (IOException e) { @@ -184,7 +195,13 @@ public class IterablePaging implements BlurIterable<ScoreDoc, BlurException> { long s = System.currentTimeMillis(); _progressRef.searchesPerformed.incrementAndGet(); try { - TopScoreDocCollector collector = TopScoreDocCollector.create(_numHitsToCollect, lastScoreDoc, true); + TopDocsCollector<?> collector; + if (_sort == null) { + collector = TopScoreDocCollector.create(_numHitsToCollect, lastScoreDoc, true); + } else { + collector = TopFieldCollector.create(_sort, _numHitsToCollect, (FieldDoc) lastScoreDoc, true, true, false, + true); + } Collector col = new StopExecutionCollector(collector, _running); if (_runSlow) { col = new SlowCollector(col); @@ -230,7 +247,7 @@ public class IterablePaging implements BlurIterable<ScoreDoc, BlurException> { } } - private BlurIterator<ScoreDoc, BlurException> skipHits(BlurIterator<ScoreDoc, BlurException> iterator) + private BlurIterator<ScoreDoc, BlurException> skipHits(PagingIterator iterator) throws BlurException { _progressRef.skipTo.set(skipTo); for (int i = 0; i < skipTo && iterator.hasNext(); i++) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3b36b782/blur-query/src/test/java/org/apache/blur/lucene/search/TestingPagingCollector.java ---------------------------------------------------------------------- diff --git a/blur-query/src/test/java/org/apache/blur/lucene/search/TestingPagingCollector.java b/blur-query/src/test/java/org/apache/blur/lucene/search/TestingPagingCollector.java index 48884af..078b1f9 100644 --- a/blur-query/src/test/java/org/apache/blur/lucene/search/TestingPagingCollector.java +++ b/blur-query/src/test/java/org/apache/blur/lucene/search/TestingPagingCollector.java @@ -18,7 +18,10 @@ package org.apache.blur.lucene.search; */ import static org.apache.blur.lucene.LuceneVersionConstant.LUCENE_VERSION; +import static org.junit.Assert.assertEquals; +import java.io.IOException; +import java.lang.management.ManagementFactory; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.blur.lucene.search.IterablePaging.ProgressRef; @@ -27,17 +30,26 @@ import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.utils.BlurIterator; import org.apache.lucene.analysis.core.KeywordAnalyzer; import org.apache.lucene.document.Document; -import org.apache.lucene.document.Field; -import org.apache.lucene.document.Field.Index; import org.apache.lucene.document.Field.Store; +import org.apache.lucene.document.IntDocValuesField; +import org.apache.lucene.document.IntField; +import org.apache.lucene.document.StringField; +import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.IndexReader; import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.IndexWriterConfig; import org.apache.lucene.index.Term; +import org.apache.lucene.search.FieldComparator; +import org.apache.lucene.search.FieldComparatorSource; import org.apache.lucene.search.IndexSearcher; import org.apache.lucene.search.ScoreDoc; +import org.apache.lucene.search.Sort; +import org.apache.lucene.search.SortField; +import org.apache.lucene.search.SortField.Type; import org.apache.lucene.search.TermQuery; +import org.apache.lucene.store.Directory; import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.util.BytesRef; import org.junit.Test; /** @@ -47,39 +59,109 @@ import org.junit.Test; public class TestingPagingCollector { @Test - public void testNothingYet() { + public void testSimpleSearchPaging() throws Exception { + IndexReader reader = getReaderFlatScore(13245); + IndexSearcher searcher = new IndexSearcher(reader); + + TotalHitsRef totalHitsRef = new TotalHitsRef(); + ProgressRef progressRef = new ProgressRef(); + TermQuery query = new TermQuery(new Term("f1", "value")); + IterablePaging paging = new IterablePaging(new AtomicBoolean(true), searcher, query, 100, null, null, false); + IterablePaging itPaging = paging.skipTo(90).gather(20).totalHits(totalHitsRef).progress(progressRef); + BlurIterator<ScoreDoc, BlurException> iterator = itPaging.iterator(); + int position = 90; + int searches = 1; + while (iterator.hasNext()) { + ScoreDoc sd = iterator.next(); + assertEquals(position, progressRef.currentHitPosition()); + assertEquals(searches, progressRef.searchesPerformed()); + System.out.println("time [" + progressRef.queryTime() + "] " + "total hits [" + totalHitsRef.totalHits() + "] " + + "searches [" + progressRef.searchesPerformed() + "] " + "position [" + progressRef.currentHitPosition() + + "] " + "doc id [" + sd.doc + "] " + "score [" + sd.score + "]"); + position++; + if (position == 100) { + searches++; + } + } } - public static void main(String[] args) throws Exception { + @Test + public void testSimpleSearchPagingWithSorting() throws Exception { IndexReader reader = getReaderFlatScore(13245); IndexSearcher searcher = new IndexSearcher(reader); TotalHitsRef totalHitsRef = new TotalHitsRef(); ProgressRef progressRef = new ProgressRef(); + printHeapSize(); TermQuery query = new TermQuery(new Term("f1", "value")); - IterablePaging paging = new IterablePaging(new AtomicBoolean(true), searcher, query, 100, null, null, false); +// SortField sortfield = new SortField("index", Type.INT, true); + FieldComparatorSource comparator = getFieldComparatorSource(); + SortField sortfield = new SortField("index", comparator); + Sort sort = new Sort(sortfield); + IterablePaging paging = new IterablePaging(new AtomicBoolean(true), searcher, query, 100, null, null, false, sort); IterablePaging itPaging = paging.skipTo(90).gather(20).totalHits(totalHitsRef).progress(progressRef); BlurIterator<ScoreDoc, BlurException> iterator = itPaging.iterator(); + int position = 90; + int searches = 1; while (iterator.hasNext()) { - ScoreDoc sd = iterator.next(); + ScoreDoc sd = iterator.next(); + assertEquals(position, progressRef.currentHitPosition()); + assertEquals(searches, progressRef.searchesPerformed()); System.out.println("time [" + progressRef.queryTime() + "] " + "total hits [" + totalHitsRef.totalHits() + "] " + "searches [" + progressRef.searchesPerformed() + "] " + "position [" + progressRef.currentHitPosition() + "] " + "doc id [" + sd.doc + "] " + "score [" + sd.score + "]"); + position++; + if (position == 100) { + searches++; + } } + printHeapSize(); + } + + private FieldComparatorSource getFieldComparatorSource() { + return new FieldComparatorSource() { + @Override + public FieldComparator<?> newComparator(String fieldname, int numHits, int sortPos, boolean reversed) + throws IOException { + // TODO Auto-generated method stub + return null; + } + }; + } + + private void printHeapSize() { + System.gc(); + System.gc(); + System.out.println("heap size=" + ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed()); } private static IndexReader getReaderFlatScore(int length) throws Exception { - RAMDirectory directory = new RAMDirectory(); + Directory directory = new RAMDirectory(); IndexWriter indexWriter = new IndexWriter(directory, new IndexWriterConfig(LUCENE_VERSION, new KeywordAnalyzer())); for (int i = 0; i < length; i++) { Document document = new Document(); - document.add(new Field("f1", "value", Store.NO, Index.ANALYZED_NO_NORMS)); + document.add(new StringField("f1", "value", Store.NO)); + document.add(new IntDocValuesField("index", i)); + document.add(new IntField("index", i, Store.YES)); indexWriter.addDocument(document); } indexWriter.close(); - return IndexReader.open(directory); + return DirectoryReader.open(directory); + } + + static byte[] toBytes(int val) { + byte[] b = new byte[4]; + b[3] = (byte) (val); + b[2] = (byte) (val >>> 8); + b[1] = (byte) (val >>> 16); + b[0] = (byte) (val >>> 24); + return b; + } + + static int toInt(byte[] b) { + return ((b[3] & 0xFF)) + ((b[2] & 0xFF) << 8) + ((b[1] & 0xFF) << 16) + ((b[0]) << 24); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/3b36b782/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur022Codec.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur022Codec.java b/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur022Codec.java index deba9ef..b1e8330 100644 --- a/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur022Codec.java +++ b/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur022Codec.java @@ -145,7 +145,7 @@ public class Blur022Codec extends Codec { } private final PostingsFormat defaultFormat = PostingsFormat.forName("Lucene41"); - private final DocValuesFormat defaultDVFormat = DocValuesFormat.forName("Lucene42"); + private final DocValuesFormat defaultDVFormat = DocValuesFormat.forName("Disk"); private final NormsFormat normsFormat = new Lucene42NormsFormat();
