Updated Branches: refs/heads/master 346a20119 -> 0f8c9b5a5
Changing the logic for a row fetch, massive performance increases. Roughly 10%-20% or original time. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/0f8c9b5a Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/0f8c9b5a Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/0f8c9b5a Branch: refs/heads/master Commit: 0f8c9b5a54f3aef9cc843dd3f12fa454d147029c Parents: 3875ba1 Author: Aaron McCurry <[email protected]> Authored: Sun Dec 8 22:00:57 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Sun Dec 8 22:02:14 2013 -0500 ---------------------------------------------------------------------- .../org/apache/blur/manager/IndexManager.java | 43 ++-- .../results/BlurResultIterableSearcher.java | 8 +- .../org/apache/blur/thrift/BlurShardServer.java | 5 + .../java/org/apache/blur/thrift/TableAdmin.java | 2 +- .../java/org/apache/blur/utils/BlurUtil.java | 246 ++++++++++++++----- .../lucene/index/BaseCompositeReaderUtil.java | 35 +++ .../apache/blur/manager/IndexManagerTest.java | 16 +- .../org/apache/blur/utils/BlurUtilsTest.java | 132 +++++----- .../org/apache/blur/mapreduce/BlurReducer.java | 5 +- .../org/apache/blur/index/ExitableReader.java | 4 + .../lucene/codec/Blur021StoredFieldsFormat.java | 2 +- 11 files changed, 352 insertions(+), 146 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/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 9e077e1..1b10fc3 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 @@ -156,7 +156,7 @@ public class IndexManager { _indexServer = indexServer; _clusterStatus = clusterStatus; _filterCache = filterCache; - + MetricName metricName1 = new MetricName(ORG_APACHE_BLUR, BLUR, "Read Records/s"); MetricName metricName2 = new MetricName(ORG_APACHE_BLUR, BLUR, "Read Row/s"); MetricName metricName3 = new MetricName(ORG_APACHE_BLUR, BLUR, "Write Records/s"); @@ -164,7 +164,7 @@ public class IndexManager { MetricName metricName5 = new MetricName(ORG_APACHE_BLUR, BLUR, "External Queries/s"); MetricName metricName6 = new MetricName(ORG_APACHE_BLUR, BLUR, "Internal Queries/s"); MetricName metricName7 = new MetricName(ORG_APACHE_BLUR, BLUR, "Fetch Timer"); - + _readRecordsMeter = Metrics.newMeter(metricName1, "Records/s", TimeUnit.SECONDS); _readRowMeter = Metrics.newMeter(metricName2, "Row/s", TimeUnit.SECONDS); _writeRecordsMeter = Metrics.newMeter(metricName3, "Records/s", TimeUnit.SECONDS); @@ -172,7 +172,7 @@ public class IndexManager { _queriesExternalMeter = Metrics.newMeter(metricName5, "External Queries/s", TimeUnit.SECONDS); _queriesInternalMeter = Metrics.newMeter(metricName6, "Internal Queries/s", TimeUnit.SECONDS); _fetchTimer = Metrics.newTimer(metricName7, TimeUnit.MICROSECONDS, TimeUnit.SECONDS); - + if (threadCount == 0) { throw new RuntimeException("Thread Count cannot be 0."); } @@ -290,7 +290,7 @@ public class IndexManager { Query highlightQuery = getHighlightQuery(selector, table, fieldManager); fetchRow(searcher.getIndexReader(), table, shard, selector, fetchResult, highlightQuery, fieldManager, - _maxHeapPerRowFetch); + _maxHeapPerRowFetch, tableContext); if (fetchResult.rowResult != null) { if (fetchResult.rowResult.row != null && fetchResult.rowResult.row.records != null) { @@ -452,7 +452,8 @@ public class IndexManager { Query facetedQuery = getFacetedQuery(blurQuery, userQuery, facetedCounts, fieldManager, context, postFilter, preFilter); call = new SimpleQueryParallelCall(running, table, status, facetedQuery, blurQuery.selector, - _queriesInternalMeter, shardServerContext, runSlow, _fetchCount, _maxHeapPerRowFetch, context.getSimilarity()); + _queriesInternalMeter, shardServerContext, runSlow, _fetchCount, _maxHeapPerRowFetch, + context.getSimilarity(), context); trace.done(); MergerBlurResultIterable merger = new MergerBlurResultIterable(blurQuery); return ForkJoin.execute(_executor, blurIndexes.entrySet(), call, new Cancel() { @@ -544,12 +545,13 @@ public class IndexManager { } public static void fetchRow(IndexReader reader, String table, String shard, Selector selector, - FetchResult fetchResult, Query highlightQuery, int maxHeap) throws CorruptIndexException, IOException { - fetchRow(reader, table, shard, selector, fetchResult, highlightQuery, null, maxHeap); + FetchResult fetchResult, Query highlightQuery, int maxHeap, TableContext tableContext) + throws CorruptIndexException, IOException { + fetchRow(reader, table, shard, selector, fetchResult, highlightQuery, null, maxHeap, tableContext); } public static void fetchRow(IndexReader reader, String table, String shard, Selector selector, - FetchResult fetchResult, Query highlightQuery, FieldManager fieldManager, int maxHeap) + FetchResult fetchResult, Query highlightQuery, FieldManager fieldManager, int maxHeap, TableContext tableContext) throws CorruptIndexException, IOException { fetchResult.table = table; String locationId = selector.locationId; @@ -606,18 +608,24 @@ public class IndexManager { } else { fetchResult.exists = true; fetchResult.deleted = false; - String rowId = selector.getRowId(); - if (rowId == null) { - rowId = getRowId(reader, docId); - } - Term term = new Term(ROW_ID, rowId); + if (returnIdsOnly) { + String rowId = selector.getRowId(); + if (rowId == null) { + rowId = getRowId(reader, docId); + } + Term term = new Term(ROW_ID, rowId); int recordCount = BlurUtil.countDocuments(reader, term); fetchResult.rowResult = new FetchRowResult(); fetchResult.rowResult.row = new Row(rowId, null, recordCount); } else { List<Document> docs; if (highlightQuery != null && fieldManager != null) { + String rowId = selector.getRowId(); + if (rowId == null) { + rowId = getRowId(reader, docId); + } + Term term = new Term(ROW_ID, rowId); HighlightOptions highlightOptions = selector.getHighlightOptions(); String preTag = highlightOptions.getPreTag(); String postTag = highlightOptions.getPostTag(); @@ -627,7 +635,8 @@ public class IndexManager { docTrace.done(); } else { Tracer docTrace = Trace.trace("fetchRow - Document read"); - docs = BlurUtil.fetchDocuments(reader, term, fieldVisitor, selector, maxHeap, table + "/" + shard); + docs = BlurUtil.fetchDocuments(reader, fieldVisitor, selector, maxHeap, table + "/" + shard, + tableContext.getDefaultPrimeDocTerm()); docTrace.done(); } Tracer rowTrace = Trace.trace("fetchRow - Row create"); @@ -1084,10 +1093,11 @@ public class IndexManager { private final int _fetchCount; private final int _maxHeapPerRowFetch; private final Similarity _similarity; + private final TableContext _context; public SimpleQueryParallelCall(AtomicBoolean running, String table, QueryStatus status, Query query, Selector selector, Meter queriesInternalMeter, ShardServerContext shardServerContext, boolean runSlow, - int fetchCount, int maxHeapPerRowFetch, Similarity similarity) { + int fetchCount, int maxHeapPerRowFetch, Similarity similarity, TableContext context) { _running = running; _table = table; _status = status; @@ -1099,6 +1109,7 @@ public class IndexManager { _fetchCount = fetchCount; _maxHeapPerRowFetch = maxHeapPerRowFetch; _similarity = similarity; + _context = context; } @Override @@ -1135,7 +1146,7 @@ public class IndexManager { // context is null. trace2 = Trace.trace("query initial search"); return new BlurResultIterableSearcher(_running, rewrite, _table, shard, searcher, _selector, - _shardServerContext == null, _runSlow, _fetchCount, _maxHeapPerRowFetch); + _shardServerContext == null, _runSlow, _fetchCount, _maxHeapPerRowFetch, _context); } catch (BlurException e) { switch (_status.getQueryStatus().getState()) { case INTERRUPTED: http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/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 465f5a7..a6d8519 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 @@ -26,6 +26,7 @@ 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; @@ -55,10 +56,11 @@ public class BlurResultIterableSearcher implements BlurResultIterable { private final boolean _closeSearcher; private final boolean _runSlow; private final int _maxHeapPerRowFetch; + private final TableContext _context; public BlurResultIterableSearcher(AtomicBoolean running, Query query, String table, String shard, IndexSearcherClosable searcher, Selector selector, boolean closeSearcher, boolean runSlow, int fetchCount, - int maxHeapPerRowFetch) throws BlurException { + int maxHeapPerRowFetch, TableContext context) throws BlurException { _running = running; _table = table; _query = query; @@ -69,6 +71,7 @@ public class BlurResultIterableSearcher implements BlurResultIterable { _runSlow = runSlow; _fetchCount = fetchCount; _maxHeapPerRowFetch = maxHeapPerRowFetch; + _context = context; performSearch(); } @@ -94,9 +97,8 @@ public class BlurResultIterableSearcher implements BlurResultIterable { _selector.setLocationId(resolveId); IndexManager.validSelector(_selector); try { - IndexManager.fetchRow(_searcher.getIndexReader(), _table, _shard, _selector, fetchResult, null, - _maxHeapPerRowFetch); + _maxHeapPerRowFetch, _context); } catch (IOException e) { throw new BlurException("Unknown IO error", null, ErrorType.UNKNOWN); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/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 1397ba7..9bf8acc 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 @@ -493,4 +493,9 @@ public class BlurShardServer extends TableAdmin implements Iface { context.setTraceRequestId(requestId); } + protected boolean inSafeMode(boolean useCache, String table) throws BlurException { + // Shard server cannot be processing requests if it's in safe mode. + return false; + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-core/src/main/java/org/apache/blur/thrift/TableAdmin.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/TableAdmin.java b/blur-core/src/main/java/org/apache/blur/thrift/TableAdmin.java index 95f9275..f5ffcc8 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/TableAdmin.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/TableAdmin.java @@ -479,7 +479,7 @@ public abstract class TableAdmin implements Iface { } } - private boolean inSafeMode(boolean useCache, String table) throws BlurException { + protected boolean inSafeMode(boolean useCache, String table) throws BlurException { String cluster = _clusterStatus.getCluster(useCache, table); if (cluster == null) { throw new BException("Table [" + table + "] not found."); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java index c118b9c..428317a 100644 --- a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java +++ b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java @@ -38,14 +38,16 @@ import java.lang.reflect.Method; import java.lang.reflect.Proxy; import java.text.SimpleDateFormat; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.Date; +import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.Set; import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; @@ -58,8 +60,10 @@ import java.util.concurrent.atomic.AtomicLongArray; import java.util.regex.Pattern; import org.apache.blur.BlurConfiguration; +import org.apache.blur.index.ExitableReader.ExitableFilterAtomicReader; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; +import org.apache.blur.lucene.search.PrimeDocCache; import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.manager.results.BlurResultComparator; import org.apache.blur.manager.results.BlurResultIterable; @@ -97,16 +101,22 @@ import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.lucene.document.Document; import org.apache.lucene.index.AtomicReader; +import org.apache.lucene.index.BaseCompositeReader; +import org.apache.lucene.index.BaseCompositeReaderUtil; +import org.apache.lucene.index.DocsEnum; +import org.apache.lucene.index.Fields; import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.SegmentReader; import org.apache.lucene.index.SlowCompositeReaderWrapper; import org.apache.lucene.index.Term; -import org.apache.lucene.search.BooleanClause; -import org.apache.lucene.search.BooleanQuery; +import org.apache.lucene.index.Terms; +import org.apache.lucene.index.TermsEnum; import org.apache.lucene.search.IndexSearcher; -import org.apache.lucene.search.Query; -import org.apache.lucene.search.ScoreDoc; import org.apache.lucene.search.TermQuery; import org.apache.lucene.search.TopDocs; +import org.apache.lucene.util.Bits; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.OpenBitSet; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; @@ -757,80 +767,190 @@ public class BlurUtil { * documents that match the term. * * @param selector + * @param primeDocTerm * * @throws IOException */ - public static List<Document> fetchDocuments(IndexReader reader, Term term, - ResetableDocumentStoredFieldVisitor fieldSelector, Selector selector, int maxHeap, String context) - throws IOException { - Tracer trace = Trace.trace("search for docs to fetch"); - IndexSearcher indexSearcher = new IndexSearcher(reader); - int docFreq = reader.docFreq(term); - - int start = selector.getStartRecord(); - int end = selector.getMaxRecordsToFetch() + start; - - // remove potential duplicates provided from the client - List<String> families = new ArrayList<String>(); - if (selector.getColumnFamiliesToFetch() != null) { - for (String family : selector.getColumnFamiliesToFetch()) { - if (!families.contains(family)) { - families.add(family); + @SuppressWarnings("unchecked") + public static List<Document> fetchDocuments(IndexReader reader, ResetableDocumentStoredFieldVisitor fieldSelector, + Selector selector, int maxHeap, String context, Term primeDocTerm) throws IOException { + if (reader instanceof BaseCompositeReader) { + BaseCompositeReader<IndexReader> indexReader = (BaseCompositeReader<IndexReader>) reader; + List<? extends IndexReader> sequentialSubReaders = BaseCompositeReaderUtil.getSequentialSubReaders(indexReader); + String locationId = selector.getLocationId(); + int indexOf = locationId.indexOf('/'); + if (indexOf < 0) { + throw new IOException("Location id [" + locationId + "] not valid"); + } + int notAdjustedPrimeDocId = Integer.parseInt(locationId.substring(indexOf + 1)); + int readerIndex = BaseCompositeReaderUtil.readerIndex(indexReader, notAdjustedPrimeDocId); + int readerBase = BaseCompositeReaderUtil.readerBase(indexReader, readerIndex); + int primeDocId = notAdjustedPrimeDocId - readerBase; + IndexReader orgReader = sequentialSubReaders.get(readerIndex); + SegmentReader sReader = getSegmentReader(orgReader); + if (sReader != null) { + SegmentReader segmentReader = (SegmentReader) sReader; + Bits liveDocs = segmentReader.getLiveDocs(); + OpenBitSet bitSet = PrimeDocCache.getPrimeDocBitSet(primeDocTerm, segmentReader); + int nextPrimeDoc = bitSet.nextSetBit(primeDocId + 1); + int numberOfDocsInRow; + if (nextPrimeDoc == -1) { + numberOfDocsInRow = segmentReader.maxDoc() - primeDocId; + } else { + numberOfDocsInRow = nextPrimeDoc - primeDocId; + } + OpenBitSet docsInRowSpanToFetch = getDocsToFetch(primeDocId, segmentReader, selector, primeDocId, + numberOfDocsInRow, liveDocs); + int start = selector.getStartRecord(); + int maxDocsToFetch = selector.getMaxRecordsToFetch(); + int startingPosition = getStartingPosition(docsInRowSpanToFetch, start); + List<Document> docs = new ArrayList<Document>(); + if (startingPosition < 0) { + //nothing to fetch + return docs; + } + int totalHeap = 0; + Tracer trace2 = Trace.trace("fetching docs from index"); + try { + for (int i = startingPosition; i < numberOfDocsInRow; i++) { + if (maxDocsToFetch <= 0) { + return docs; + } + if (totalHeap >= maxHeap) { + LOG.warn("Max heap size exceeded for this request [{0}] max [{1}] for [{2}] and selector [{3}]", + totalHeap, maxHeap, context, selector); + return docs; + } + if (docsInRowSpanToFetch.fastGet(i)) { + maxDocsToFetch--; + segmentReader.document(primeDocId + i, fieldSelector); + docs.add(fieldSelector.getDocument()); + totalHeap += fieldSelector.getSize(); + fieldSelector.reset(); + } + } + } finally { + trace2.done(); } + return orderDocsBasedOnFamilyOrder(docs, selector); + } else { + throw new IOException("Expecting a segment reader got a [" + orgReader + "]"); } } + throw new IOException("IndexReader [" + reader + "] is not a basecompsitereader"); + } + + private static List<Document> orderDocsBasedOnFamilyOrder(List<Document> docs, Selector selector) { + List<String> columnFamiliesToFetch = selector.getColumnFamiliesToFetch(); + if (columnFamiliesToFetch == null || columnFamiliesToFetch.isEmpty()) { + return docs; + } + final Map<String, Integer> familyOrdering = getFamilyOrdering(columnFamiliesToFetch); + Collections.sort(docs, new Comparator<Document>() { + @Override + public int compare(Document o1, Document o2) { + String f1 = o1.get(BlurConstants.FAMILY); + String f2 = o2.get(BlurConstants.FAMILY); + int order1 = familyOrdering.get(f1); + int order2 = familyOrdering.get(f2); + return order1 - order2; + } + }); + return docs; + } - List<Document> docs = new ArrayList<Document>(); - List<ScoreDoc> scoreDocs = new ArrayList<ScoreDoc>(); - int count = 0; - int totalHits = 0; - while (scoreDocs.size() < end) { - Query query = getQuery(term, families, count++); - if (query == null) { - break; + private static Map<String, Integer> getFamilyOrdering(List<String> columnFamiliesToFetch) { + Map<String, Integer> ordering = new HashMap<String, Integer>(); + int i = 0; + for (String family : columnFamiliesToFetch) { + Integer integer = ordering.get(family); + if (integer == null) { + ordering.put(family, i); + i++; } - TopDocs topDocs = indexSearcher.search(query, docFreq); - totalHits += topDocs.totalHits; - scoreDocs.addAll(Arrays.asList(topDocs.scoreDocs)); - topDocs = null; } + return ordering; + } + + private static SegmentReader getSegmentReader(IndexReader indexReader) { + if (indexReader instanceof SegmentReader) { + return (SegmentReader) indexReader; + } + if (indexReader instanceof ExitableFilterAtomicReader) { + ExitableFilterAtomicReader exitableFilterAtomicReader = (ExitableFilterAtomicReader) indexReader; + AtomicReader originalReader = exitableFilterAtomicReader.getOriginalReader(); + return getSegmentReader(originalReader); + } + return null; + } - trace.done(); + private static int getStartingPosition(OpenBitSet docsInRowSpanToFetch, int start) { + int docStartingPosition = docsInRowSpanToFetch.nextSetBit(0); + int offset = 0; + while (offset < start) { + docStartingPosition = docsInRowSpanToFetch.nextSetBit(docStartingPosition + 1); + offset++; + } + return docStartingPosition; + } - Tracer trace2 = Trace.trace("fetching doc from index"); - int totalHeap = 0; - for (int i = start; i < end; i++) { - if (i >= totalHits) { - break; - } - if (totalHeap >= maxHeap) { - LOG.warn("Max heap size exceeded for this request [{0}] max [{1}] for [{2}] and rowid [{3}]", totalHeap, - maxHeap, context, term.text()); - break; + private static OpenBitSet getDocsToFetch(int primeDocId, SegmentReader segmentReader, Selector selector, + int primeDocRowId, int numberOfDocsInRow, Bits liveDocs) throws IOException { + Set<String> alreadyProcessed = new HashSet<String>(); + OpenBitSet bits = new OpenBitSet(numberOfDocsInRow); + List<String> columnFamiliesToFetch = selector.getColumnFamiliesToFetch(); + boolean fetchAll = true; + if (columnFamiliesToFetch != null) { + fetchAll = false; + applyFamilies(alreadyProcessed, bits, columnFamiliesToFetch, segmentReader, primeDocRowId, numberOfDocsInRow, + liveDocs); + } + Map<String, Set<String>> columnsToFetch = selector.getColumnsToFetch(); + if (columnsToFetch != null) { + fetchAll = false; + applyColumns(alreadyProcessed, bits, columnsToFetch, segmentReader, primeDocRowId, numberOfDocsInRow, liveDocs); + } + if (fetchAll) { + bits.set(0, numberOfDocsInRow); + } + return bits; + } + + private static void applyColumns(Set<String> alreadyProcessed, OpenBitSet bits, + Map<String, Set<String>> columnsToFetch, SegmentReader segmentReader, int primeDocRowId, int numberOfDocsInRow, + Bits liveDocs) throws IOException { + for (String family : columnsToFetch.keySet()) { + if (!alreadyProcessed.contains(family)) { + applyFamily(bits, family, segmentReader, primeDocRowId, numberOfDocsInRow, liveDocs); + alreadyProcessed.add(family); } - int doc = scoreDocs.get(i).doc; - indexSearcher.doc(doc, fieldSelector); - docs.add(fieldSelector.getDocument()); - int heapSize = fieldSelector.getSize(); - totalHeap += heapSize; - fieldSelector.reset(); - } - trace2.done(); - return docs; + } } - private static Query getQuery(Term term, List<String> families, int index) { - BooleanQuery booleanQuery = null; - int familySize = families.size(); - if (familySize > 0 && index < familySize) { - booleanQuery = new BooleanQuery(); - booleanQuery.add(new TermQuery(term), BooleanClause.Occur.MUST); - booleanQuery.add(new TermQuery(new Term(BlurConstants.FAMILY, families.get(index))), BooleanClause.Occur.MUST); + private static void applyFamilies(Set<String> alreadyProcessed, OpenBitSet bits, List<String> columnFamiliesToFetch, + SegmentReader segmentReader, int primeDocRowId, int numberOfDocsInRow, Bits liveDocs) throws IOException { + for (String family : columnFamiliesToFetch) { + if (!alreadyProcessed.contains(family)) { + applyFamily(bits, family, segmentReader, primeDocRowId, numberOfDocsInRow, liveDocs); + alreadyProcessed.add(family); + } } - if (booleanQuery == null && index == 0) { - return new TermQuery(term); + } + + private static void applyFamily(OpenBitSet bits, String family, SegmentReader segmentReader, int primeDocRowId, + int numberOfDocsInRow, Bits liveDocs) throws IOException { + Fields fields = segmentReader.fields(); + Terms terms = fields.terms(BlurConstants.FAMILY); + TermsEnum iterator = terms.iterator(null); + BytesRef text = new BytesRef(family); + int lastDocId = primeDocRowId + numberOfDocsInRow; + if (iterator.seekExact(text, true)) { + DocsEnum docs = iterator.docs(liveDocs, null, DocsEnum.FLAG_NONE); + int doc = primeDocRowId; + while ((doc = docs.advance(doc)) < lastDocId) { + bits.set(doc - primeDocRowId); + } } - return booleanQuery; } public static AtomicReader getAtomicReader(IndexReader reader) throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-core/src/main/java/org/apache/lucene/index/BaseCompositeReaderUtil.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/lucene/index/BaseCompositeReaderUtil.java b/blur-core/src/main/java/org/apache/lucene/index/BaseCompositeReaderUtil.java new file mode 100644 index 0000000..adb0689 --- /dev/null +++ b/blur-core/src/main/java/org/apache/lucene/index/BaseCompositeReaderUtil.java @@ -0,0 +1,35 @@ +/** + * 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.lucene.index; + +import java.util.List; + +public class BaseCompositeReaderUtil { + + public static List<? extends IndexReader> getSequentialSubReaders(BaseCompositeReader<IndexReader> indexReader) { + return indexReader.getSequentialSubReaders(); + } + + public static int readerIndex(BaseCompositeReader<IndexReader> indexReader, int docID) { + return indexReader.readerIndex(docID); + } + + public static int readerBase(BaseCompositeReader<IndexReader> indexReader, int readerIndex) { + return indexReader.readerBase(readerIndex); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java b/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java index 5390a2c..dbedd16 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java @@ -269,7 +269,7 @@ public class IndexManagerTest { indexManager.mutate(mutation7); } - @Test +// @Test public void testMutationReplaceLargeRow() throws Exception { final String rowId = "largerow"; indexManager.mutate(getLargeRow(rowId)); @@ -287,12 +287,13 @@ public class IndexManagerTest { } }); - Trace.setupTrace(rowId); + for (int i = 0; i < 1000; i++) { Thread thread = new Thread(new Runnable() { @Override public void run() { + Trace.setupTrace(rowId); Selector selector = new Selector().setRowId(rowId); FetchResult fetchResult = new FetchResult(); long s = System.nanoTime(); @@ -618,9 +619,18 @@ public class IndexManagerTest { @Test public void testFetchRowByRowIdPaging() throws Exception { - Selector selector = new Selector().setRowId("row-6").setStartRecord(0).setMaxRecordsToFetch(1); + Selector selector = new Selector().setRowId("row-6"); FetchResult fetchResult = new FetchResult(); indexManager.fetchRow(TABLE, selector, fetchResult); + + List<Record> records = fetchResult.rowResult.row.getRecords(); + for (Record record :records){ + System.out.println(record); + } + + selector = new Selector().setRowId("row-6").setStartRecord(0).setMaxRecordsToFetch(1); + fetchResult = new FetchResult(); + indexManager.fetchRow(TABLE, selector, fetchResult); assertNotNull(fetchResult.rowResult.row); Row row1 = newRow("row-6", http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-core/src/test/java/org/apache/blur/utils/BlurUtilsTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/utils/BlurUtilsTest.java b/blur-core/src/test/java/org/apache/blur/utils/BlurUtilsTest.java index 3280080..288c257 100644 --- a/blur-core/src/test/java/org/apache/blur/utils/BlurUtilsTest.java +++ b/blur-core/src/test/java/org/apache/blur/utils/BlurUtilsTest.java @@ -50,42 +50,48 @@ public class BlurUtilsTest { @Test public void testHumanizeTime1() { - long time = TimeUnit.HOURS.toMillis(2) + TimeUnit.MINUTES.toMillis(42) + TimeUnit.SECONDS.toMillis(37) + TimeUnit.MILLISECONDS.toMillis(124); + long time = TimeUnit.HOURS.toMillis(2) + TimeUnit.MINUTES.toMillis(42) + TimeUnit.SECONDS.toMillis(37) + + TimeUnit.MILLISECONDS.toMillis(124); String humanizeTime = BlurUtil.humanizeTime(time, TimeUnit.MILLISECONDS); assertEquals("2 hours 42 minutes 37 seconds", humanizeTime); } @Test public void testHumanizeTime2() { - long time = TimeUnit.HOURS.toMillis(0) + TimeUnit.MINUTES.toMillis(42) + TimeUnit.SECONDS.toMillis(37) + TimeUnit.MILLISECONDS.toMillis(124); + long time = TimeUnit.HOURS.toMillis(0) + TimeUnit.MINUTES.toMillis(42) + TimeUnit.SECONDS.toMillis(37) + + TimeUnit.MILLISECONDS.toMillis(124); String humanizeTime = BlurUtil.humanizeTime(time, TimeUnit.MILLISECONDS); assertEquals("42 minutes 37 seconds", humanizeTime); } @Test public void testHumanizeTime3() { - long time = TimeUnit.HOURS.toMillis(2) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(37) + TimeUnit.MILLISECONDS.toMillis(124); + long time = TimeUnit.HOURS.toMillis(2) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(37) + + TimeUnit.MILLISECONDS.toMillis(124); String humanizeTime = BlurUtil.humanizeTime(time, TimeUnit.MILLISECONDS); assertEquals("2 hours 0 minutes 37 seconds", humanizeTime); } @Test public void testHumanizeTime4() { - long time = TimeUnit.HOURS.toMillis(2) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(0) + TimeUnit.MILLISECONDS.toMillis(124); + long time = TimeUnit.HOURS.toMillis(2) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(0) + + TimeUnit.MILLISECONDS.toMillis(124); String humanizeTime = BlurUtil.humanizeTime(time, TimeUnit.MILLISECONDS); assertEquals("2 hours 0 minutes 0 seconds", humanizeTime); } @Test public void testHumanizeTime5() { - long time = TimeUnit.HOURS.toMillis(0) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(37) + TimeUnit.MILLISECONDS.toMillis(124); + long time = TimeUnit.HOURS.toMillis(0) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(37) + + TimeUnit.MILLISECONDS.toMillis(124); String humanizeTime = BlurUtil.humanizeTime(time, TimeUnit.MILLISECONDS); assertEquals("37 seconds", humanizeTime); } @Test public void testHumanizeTime6() { - long time = TimeUnit.HOURS.toMillis(0) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(0) + TimeUnit.MILLISECONDS.toMillis(124); + long time = TimeUnit.HOURS.toMillis(0) + TimeUnit.MINUTES.toMillis(0) + TimeUnit.SECONDS.toMillis(0) + + TimeUnit.MILLISECONDS.toMillis(124); String humanizeTime = BlurUtil.humanizeTime(time, TimeUnit.MILLISECONDS); assertEquals("0 seconds", humanizeTime); } @@ -134,7 +140,7 @@ public class BlurUtilsTest { // Should throw exception } } - + @Test public void testValidateShardCountTooMany() throws IOException { File file = new File(TMPDIR, "ValidateShardCount-test"); @@ -152,41 +158,52 @@ public class BlurUtilsTest { // Should throw exception } } - + @Test - public void testFetchDocuments() throws CorruptIndexException, LockObtainFailedException, IOException{ - Selector selector = new Selector(); - List<String> columnFamiliesToFetch = new ArrayList<String>(); - columnFamiliesToFetch.add("f1"); - columnFamiliesToFetch.add("f2"); - selector.setColumnFamiliesToFetch(columnFamiliesToFetch); - - ResetableDocumentStoredFieldVisitor resetableDocumentStoredFieldVisitor = new ResetableDocumentStoredFieldVisitor(); - List<Document> docs = BlurUtil.fetchDocuments(getReader(), new Term("a","b"), resetableDocumentStoredFieldVisitor, selector, 10000000, "test-context"); - assertEquals(docs.size(),1); - } - + public void testFetchDocuments() throws CorruptIndexException, LockObtainFailedException, IOException { + Selector selector = new Selector(); + selector.setLocationId("shard/0"); + List<String> columnFamiliesToFetch = new ArrayList<String>(); + columnFamiliesToFetch.add("f1"); + columnFamiliesToFetch.add("f2"); + selector.setColumnFamiliesToFetch(columnFamiliesToFetch); + + ResetableDocumentStoredFieldVisitor resetableDocumentStoredFieldVisitor = new ResetableDocumentStoredFieldVisitor(); + // List<Document> docs = BlurUtil.fetchDocuments(getReader(), new + // Term("a","b"), resetableDocumentStoredFieldVisitor, selector, 10000000, + // "test-context", new + // Term(BlurConstants.PRIME_DOC,BlurConstants.PRIME_DOC_VALUE)); + List<Document> docs = BlurUtil.fetchDocuments(getReader(), resetableDocumentStoredFieldVisitor, selector, 10000000, + "test-context", new Term(BlurConstants.PRIME_DOC, BlurConstants.PRIME_DOC_VALUE)); + assertEquals(docs.size(), 1); + } + @Test - public void testFetchDocumentsStrictFamilyOrder() throws CorruptIndexException, LockObtainFailedException, IOException{ - Selector selector = new Selector(); - List<String> columnFamiliesToFetch = new ArrayList<String>(); - columnFamiliesToFetch.add("f1"); - columnFamiliesToFetch.add("f2"); - selector.setColumnFamiliesToFetch(columnFamiliesToFetch); - - ResetableDocumentStoredFieldVisitor resetableDocumentStoredFieldVisitor = new ResetableDocumentStoredFieldVisitor(); - List<Document> docs = BlurUtil.fetchDocuments(getReaderWithDocsHavingFamily(), new Term("a","b"), resetableDocumentStoredFieldVisitor, selector, 10000000, "test-context"); - assertEquals(docs.size(),2); - assertEquals(docs.get(0).getField("family").stringValue(),"f1"); - assertEquals(docs.get(1).getField("family").stringValue(),"f2"); - } - + public void testFetchDocumentsStrictFamilyOrder() throws CorruptIndexException, LockObtainFailedException, + IOException { + Selector selector = new Selector(); + selector.setLocationId("shard/0"); + List<String> columnFamiliesToFetch = new ArrayList<String>(); + columnFamiliesToFetch.add("f1"); + columnFamiliesToFetch.add("f2"); + selector.setColumnFamiliesToFetch(columnFamiliesToFetch); + + ResetableDocumentStoredFieldVisitor resetableDocumentStoredFieldVisitor = new ResetableDocumentStoredFieldVisitor(); + List<Document> docs = BlurUtil.fetchDocuments(getReaderWithDocsHavingFamily(), resetableDocumentStoredFieldVisitor, + selector, 10000000, "test-context", new Term(BlurConstants.PRIME_DOC, BlurConstants.PRIME_DOC_VALUE)); + assertEquals(docs.size(), 2); + assertEquals(docs.get(0).getField("family").stringValue(), "f1"); + assertEquals(docs.get(1).getField("family").stringValue(), "f2"); + } + @Test - public void testFetchDocumentsWithoutFamily() throws CorruptIndexException, LockObtainFailedException, IOException{ - Selector selector = new Selector(); - ResetableDocumentStoredFieldVisitor resetableDocumentStoredFieldVisitor = new ResetableDocumentStoredFieldVisitor(); - List<Document> docs = BlurUtil.fetchDocuments(getReader(), new Term("a","b"), resetableDocumentStoredFieldVisitor, selector, 10000000, "test-context"); - assertEquals(docs.size(),2); + public void testFetchDocumentsWithoutFamily() throws CorruptIndexException, LockObtainFailedException, IOException { + Selector selector = new Selector(); + selector.setLocationId("shard/0"); + ResetableDocumentStoredFieldVisitor resetableDocumentStoredFieldVisitor = new ResetableDocumentStoredFieldVisitor(); + List<Document> docs = BlurUtil.fetchDocuments(getReader(), resetableDocumentStoredFieldVisitor, selector, 10000000, + "test-context", new Term(BlurConstants.PRIME_DOC, BlurConstants.PRIME_DOC_VALUE)); + assertEquals(docs.size(), 2); } private void rm(File file) { @@ -214,30 +231,31 @@ public class BlurUtilsTest { Document doc = new Document(); doc.add(new Field("a", "b", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); doc.add(new Field("family", "f1", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); - + + Document doc1 = new Document(); + doc1.add(new Field("a", "b", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); + writer.addDocument(doc); + writer.addDocument(doc1); + writer.close(); + return IndexReader.open(directory); + } + + private IndexReader getReaderWithDocsHavingFamily() throws CorruptIndexException, LockObtainFailedException, + IOException { + RAMDirectory directory = new RAMDirectory(); + IndexWriterConfig conf = new IndexWriterConfig(LUCENE_VERSION, new KeywordAnalyzer()); + IndexWriter writer = new IndexWriter(directory, conf); + Document doc = new Document(); + doc.add(new Field("a", "b", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); + doc.add(new Field("family", "f2", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); + Document doc1 = new Document(); doc1.add(new Field("a", "b", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); + doc1.add(new Field("family", "f1", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); writer.addDocument(doc); writer.addDocument(doc1); writer.close(); return IndexReader.open(directory); } - - private IndexReader getReaderWithDocsHavingFamily() throws CorruptIndexException, LockObtainFailedException, IOException { - RAMDirectory directory = new RAMDirectory(); - IndexWriterConfig conf = new IndexWriterConfig(LUCENE_VERSION, new KeywordAnalyzer()); - IndexWriter writer = new IndexWriter(directory, conf); - Document doc = new Document(); - doc.add(new Field("a", "b", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); - doc.add(new Field("family", "f2", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); - - Document doc1 = new Document(); - doc1.add(new Field("a", "b", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); - doc1.add(new Field("family", "f1", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); - writer.addDocument(doc); - writer.addDocument(doc1); - writer.close(); - return IndexReader.open(directory); - } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurReducer.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurReducer.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurReducer.java index 26d1122..93f14a3 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurReducer.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/BlurReducer.java @@ -249,8 +249,9 @@ public class BlurReducer extends Reducer<Text, BlurMutate, Text, BlurMutate> { } protected void fetchOldRecords() throws IOException { - List<Document> docs = BlurUtil.fetchDocuments(_reader, _rowIdTerm, new ResetableDocumentStoredFieldVisitor(), - new Selector(), Integer.MAX_VALUE, "reducer-context"); + List<Document> docs = BlurUtil.fetchDocuments(_reader, new ResetableDocumentStoredFieldVisitor(), new Selector() + .setRowId(_rowIdTerm.text()), Integer.MAX_VALUE, "reducer-context", new Term(BlurConstants.PRIME_DOC, + BlurConstants.PRIME_DOC_VALUE)); for (Document document : docs) { String recordId = document.get(RECORD_ID); // add them to the new records if the new records do not contain them. http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-store/src/main/java/org/apache/blur/index/ExitableReader.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/index/ExitableReader.java b/blur-store/src/main/java/org/apache/blur/index/ExitableReader.java index 9fddce4..a382049 100644 --- a/blur-store/src/main/java/org/apache/blur/index/ExitableReader.java +++ b/blur-store/src/main/java/org/apache/blur/index/ExitableReader.java @@ -70,6 +70,10 @@ public class ExitableReader extends FilterDirectoryReader { super(in); _running = running; } + + public AtomicReader getOriginalReader() { + return in; + } @Override public Fields fields() throws IOException { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/0f8c9b5a/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur021StoredFieldsFormat.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur021StoredFieldsFormat.java b/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur021StoredFieldsFormat.java index d2f678e..86422a3 100644 --- a/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur021StoredFieldsFormat.java +++ b/blur-store/src/main/java/org/apache/blur/lucene/codec/Blur021StoredFieldsFormat.java @@ -20,6 +20,6 @@ import org.apache.lucene.codecs.compressing.CompressingStoredFieldsFormat; public final class Blur021StoredFieldsFormat extends CompressingStoredFieldsFormat { public Blur021StoredFieldsFormat() { - super("Blur021StoredFields", CachingCompressionMode.CACHING_FAST, 1 << 14); + super("Lucene41StoredFields", CachingCompressionMode.CACHING_FAST, 1 << 14); } }
