Made some small improvements to indexing performance and remove the TimeBasedIndexDeletionPolicy.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/4d530d95 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/4d530d95 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/4d530d95 Branch: refs/heads/master Commit: 4d530d95df932d1c30da30a2b2f7ded43f6323de Parents: b0ef526 Author: Aaron McCurry <[email protected]> Authored: Sat Aug 25 16:04:59 2012 -0400 Committer: Aaron McCurry <[email protected]> Committed: Sat Aug 25 16:04:59 2012 -0400 ---------------------------------------------------------------------- .../lucene/index/TimeBasedIndexDeletionPolicy.java | 55 --- .../nearinfinity/blur/manager/IndexManager.java | 35 ++- .../blur/manager/stats/LoadFactorProcessor.java | 26 +- .../blur/manager/stats/WeightedAvg.java | 36 ++- .../manager/writer/DirectoryReferenceCounter.java | 8 - .../blur/manager/writer/TransactionRecorder.java | 22 +- .../nearinfinity/blur/thrift/BlurShardServer.java | 3 + .../blur/thrift/ThriftBlurShardServer.java | 8 +- .../index/TimeBasedIndexDeletionPolicyTest.java | 58 --- .../writer/DirectoryReferenceCounterTest.java | 324 +++++++++++++++ .../nearinfinity/blur/lucene/LuceneConstant.java | 2 +- .../apache/lucene/index/WarmUpByFieldBounds.java | 3 +- .../src/main/resources/blur-default.properties | 1 - src/pom.xml | 1 + 14 files changed, 428 insertions(+), 154 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicy.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicy.java b/src/blur-core/src/main/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicy.java deleted file mode 100644 index 075ab6b..0000000 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicy.java +++ /dev/null @@ -1,55 +0,0 @@ -package com.nearinfinity.blur.lucene.index; - -import java.io.IOException; -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.apache.lucene.index.IndexCommit; -import org.apache.lucene.index.IndexDeletionPolicy; - -public class TimeBasedIndexDeletionPolicy implements IndexDeletionPolicy { - - private static final Log LOG = LogFactory.getLog(TimeBasedIndexDeletionPolicy.class); - - private long maxAge; - - public TimeBasedIndexDeletionPolicy(long maxAge) { - this.maxAge = maxAge; - } - - @Override - public void onInit(List<? extends IndexCommit> commits) throws IOException { - onCommit(commits); - } - - @Override - public void onCommit(List<? extends IndexCommit> commits) throws IOException { - IndexCommit current = commits.get(commits.size() - 1); - int length; - if (isTooOld(current.getTimestamp())) { - // the current index is old enough that following generation can be - // removed - length = commits.size() - 1; - } else { - // the current index is NOT old enough, so therefore the next generation - // (no matter the old can be removed) - length = commits.size() - 2; - } - for (int i = 0; i < length; i++) { - IndexCommit commit = commits.get(i); - if (isTooOld(commit.getTimestamp())) { - LOG.info("Removing old generation [" + commit.getGeneration() + "] for directory [" + commit.getDirectory() + "]"); - commit.delete(); - } - } - } - - private boolean isTooOld(long timestamp) { - if (timestamp + maxAge < System.currentTimeMillis()) { - return true; - } - return false; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/manager/IndexManager.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/IndexManager.java b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/IndexManager.java index c4afef0..474f0d4 100644 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/IndexManager.java +++ b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/IndexManager.java @@ -128,7 +128,8 @@ public class IndexManager { public void init() { _executor = Executors.newThreadPool("index-manager", _threadCount); - _mutateExecutor = Executors.newSingleThreadExecutor("index-manager-mutate"); + //@TODO give the mutate it's own thread pool + _mutateExecutor = Executors.newThreadPool("index-manager-mutate", _threadCount); _statusManager.init(); LOG.info("Init Complete"); } @@ -647,7 +648,10 @@ public class IndexManager { Future<Void> future = _mutateExecutor.submit(new Callable<Void>() { @Override public Void call() throws Exception { + long s = System.nanoTime(); doMutates(mutations); + long e = System.nanoTime(); + LOG.debug("doMutates took [" + (e - s) / 1000000.0 + " ms] to complete"); return null; } }); @@ -667,8 +671,8 @@ public class IndexManager { } } - private void doMutates(String table, List<RowMutation> mutations) throws IOException, BlurException { - Map<String, BlurIndex> indexes = _indexServer.getIndexes(table); + private void doMutates(final String table, List<RowMutation> mutations) throws IOException, BlurException { + final Map<String, BlurIndex> indexes = _indexServer.getIndexes(table); Map<String, List<RowMutation>> mutationsByShard = new HashMap<String, List<RowMutation>>(); @@ -683,12 +687,33 @@ public class IndexManager { list.add(mutation); } + List<Future<Void>> futures = new ArrayList<Future<Void>>(); + for (Entry<String, List<RowMutation>> entry : mutationsByShard.entrySet()) { - executeMutates(table, entry.getKey(), indexes, entry.getValue()); + final String shard = entry.getKey(); + final List<RowMutation> value = entry.getValue(); + futures.add(_mutateExecutor.submit(new Callable<Void>() { + @Override + public Void call() throws Exception { + executeMutates(table, shard, indexes, value); + return null; + } + })); + } + + for (Future<Void> future : futures) { + try { + future.get(); + } catch (InterruptedException e) { + throw new BException("Unknown error during mutation", e); + } catch (ExecutionException e) { + throw new BException("Unknown error during mutation", e.getCause()); + } } } private void executeMutates(String table, String shard, Map<String, BlurIndex> indexes, List<RowMutation> mutations) throws BlurException, IOException { + long s = System.nanoTime(); boolean waitToBeVisible = false; for (int i = 0; i < mutations.size(); i++) { RowMutation mutation = mutations.get(i); @@ -720,6 +745,8 @@ public class IndexManager { throw new RuntimeException("Not supported [" + type + "]"); } } + long e = System.nanoTime(); + LOG.debug("executeMutates took [" + (e - s) / 1000000.0 + " ms] to complete"); } private Map<String, List<RowMutation>> getMutatesPerTable(List<RowMutation> mutations) { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/LoadFactorProcessor.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/LoadFactorProcessor.java b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/LoadFactorProcessor.java index 53e2aa6..0d54ba7 100644 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/LoadFactorProcessor.java +++ b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/LoadFactorProcessor.java @@ -4,10 +4,10 @@ import java.util.concurrent.TimeUnit; public class LoadFactorProcessor { - private Sampler _sampler; - private WeightedAvg _one; - private WeightedAvg _five; - private WeightedAvg _fifteen; + private final Sampler _sampler; + private final WeightedAvg _one; + private final WeightedAvg _five; + private final WeightedAvg _fifteen; public LoadFactorProcessor(Sampler sampler) { _sampler = sampler; @@ -35,4 +35,22 @@ public class LoadFactorProcessor { return _fifteen.getAvg(); } + public Sampler getSampler() { + return _sampler; + } + + public WeightedAvg getOne() { + return _one; + } + + public WeightedAvg getFive() { + return _five; + } + + public WeightedAvg getFifteen() { + return _fifteen; + } + + + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/WeightedAvg.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/WeightedAvg.java b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/WeightedAvg.java index cad8b32..7e4bea3 100644 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/WeightedAvg.java +++ b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/stats/WeightedAvg.java @@ -1,14 +1,13 @@ package com.nearinfinity.blur.manager.stats; -import java.util.concurrent.atomic.AtomicLong; public class WeightedAvg { - private int _maxSize; - private long[] _values; + private final int _maxSize; + private final long[] _values; private int _numberOfAdds; private int _currentPosition; - private AtomicLong _totalValue = new AtomicLong(0); + private long _totalValue = 0; public WeightedAvg(int maxSize) { _maxSize = maxSize; @@ -23,17 +22,38 @@ public class WeightedAvg { } long currentValue = _values[_currentPosition]; _values[_currentPosition] = value; - _totalValue.addAndGet(value - currentValue); + _totalValue += value - currentValue; _numberOfAdds++; _currentPosition++; } public double getAvg() { - long v = _totalValue.get(); - if (v == 0) { + if (_totalValue == 0) { return 0; } - return (double) v / (double) Math.min(_numberOfAdds, _maxSize); + return (double) _totalValue / (double) Math.min(_numberOfAdds, _maxSize); } + public int getMaxSize() { + return _maxSize; + } + + public long[] getValues() { + return _values; + } + + public int getNumberOfAdds() { + return _numberOfAdds; + } + + public int getCurrentPosition() { + return _currentPosition; + } + + public long getTotalValue() { + return _totalValue; + } + + + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounter.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounter.java b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounter.java index 0a13b42..30c02dd 100644 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounter.java +++ b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounter.java @@ -231,10 +231,6 @@ public class DirectoryReferenceCounter extends Directory { directory.copy(to, src, dest); } - public boolean equals(Object obj) { - return directory.equals(obj); - } - public boolean fileExists(String name) throws IOException { return directory.fileExists(name); } @@ -252,10 +248,6 @@ public class DirectoryReferenceCounter extends Directory { return directory.getLockFactory(); } - public int hashCode() { - return directory.hashCode(); - } - public String[] listAll() throws IOException { return directory.listAll(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/TransactionRecorder.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/TransactionRecorder.java b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/TransactionRecorder.java index 43ac7d4..2d9e0b4 100644 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/TransactionRecorder.java +++ b/src/blur-core/src/main/java/com/nearinfinity/blur/manager/writer/TransactionRecorder.java @@ -111,7 +111,7 @@ public class TransactionRecorder { switch (lookup) { case ROW: Row row = readRow(dataInputStream); - writer.updateDocuments(ROW_ID.createTerm(row.id), getDocs(row)); + writer.updateDocuments(ROW_ID.createTerm(row.id), getDocs(row, analyzer)); updateCount++; continue; case DELETE: @@ -247,8 +247,8 @@ public class TransactionRecorder { } public long replaceRow(boolean wal, Row row, TrackingIndexWriter writer) throws IOException { - synchronized (running) { - if (wal) { + if (wal) { + synchronized (running) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream outputStream = new DataOutputStream(baos); outputStream.writeByte(TYPE.ROW.value()); @@ -256,13 +256,15 @@ public class TransactionRecorder { outputStream.close(); sync(baos.toByteArray()); } - return writer.updateDocuments(ROW_ID.createTerm(row.id), getDocs(row)); } + Term term = ROW_ID.createTerm(row.id); + List<Document> docs = getDocs(row, analyzer); + return writer.updateDocuments(term, docs); } public long deleteRow(boolean wal, String rowId, TrackingIndexWriter writer) throws IOException { - synchronized (running) { - if (wal) { + if (wal) { + synchronized (running) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream outputStream = new DataOutputStream(baos); outputStream.writeByte(TYPE.DELETE.value()); @@ -270,8 +272,8 @@ public class TransactionRecorder { outputStream.close(); sync(baos.toByteArray()); } - return writer.deleteDocuments(ROW_ID.createTerm(rowId)); } + return writer.deleteDocuments(ROW_ID.createTerm(rowId)); } public void setWalPath(Path walPath) { @@ -294,14 +296,14 @@ public class TransactionRecorder { } } - private List<Document> getDocs(Row row) { + public static List<Document> getDocs(Row row, BlurAnalyzer analyzer) { List<Record> records = row.records; int size = records.size(); final String rowId = row.id; final StringBuilder builder = new StringBuilder(); List<Document> docs = new ArrayList<Document>(size); for (int i = 0; i < size; i++) { - Document document = convert(rowId, records.get(i), builder); + Document document = convert(rowId, records.get(i), builder, analyzer); if (i == 0) { document.add(BlurConstants.PRIME_DOC_FIELD); } @@ -310,7 +312,7 @@ public class TransactionRecorder { return docs; } - private Document convert(String rowId, Record record, StringBuilder builder) { + public static Document convert(String rowId, Record record, StringBuilder builder, BlurAnalyzer analyzer) { Document document = new Document(); document.add(new Field(BlurConstants.ROW_ID, rowId, Store.YES, Index.NOT_ANALYZED_NO_NORMS)); document.add(new Field(BlurConstants.RECORD_ID, record.recordId, Store.YES, Index.NOT_ANALYZED_NO_NORMS)); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/BlurShardServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/BlurShardServer.java b/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/BlurShardServer.java index c6d8195..b623d09 100644 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/BlurShardServer.java +++ b/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/BlurShardServer.java @@ -235,6 +235,7 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public void mutateBatch(List<RowMutation> mutations) throws BlurException, TException { + long s = System.nanoTime(); for (RowMutation mutation : mutations) { checkTable(_cluster, mutation.table); checkForUpdates(_cluster, mutation.table); @@ -246,6 +247,8 @@ public class BlurShardServer extends TableAdmin implements Iface { LOG.error("Unknown error during processing of [mutations={0}]", e, mutations); throw new BException(e.getMessage(), e); } + long e = System.nanoTime(); + LOG.debug("mutateBatch took [" + (e-s) / 1000000.0 + " ms] to complete"); } public long getMaxTimeToLive() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/ThriftBlurShardServer.java b/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/ThriftBlurShardServer.java index 6a51274..bb1243d 100644 --- a/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/ThriftBlurShardServer.java +++ b/src/blur-core/src/main/java/com/nearinfinity/blur/thrift/ThriftBlurShardServer.java @@ -16,7 +16,8 @@ package com.nearinfinity.blur.thrift; -import static com.nearinfinity.blur.utils.BlurConstants.*; +import static com.nearinfinity.blur.utils.BlurConstants.BLUR_CLUSTER_NAME; +import static com.nearinfinity.blur.utils.BlurConstants.BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_MAX_CLAUSE_COUNT; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_BIND_ADDRESS; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_BIND_PORT; @@ -27,7 +28,6 @@ import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_CACHE_MAX_TIM import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_DATA_FETCH_THREAD_COUNT; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_FILTER_CACHE_CLASS; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_HOSTNAME; -import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_INDEX_DELETION_POLICY_MAXAGE; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_INDEX_WARMUP_CLASS; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_OPENER_THREAD_COUNT; import static com.nearinfinity.blur.utils.BlurConstants.BLUR_SHARD_SAFEMODEDELAY; @@ -42,6 +42,7 @@ import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.lucene.index.IndexDeletionPolicy; +import org.apache.lucene.index.KeepOnlyLastCommitDeletionPolicy; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; import org.apache.zookeeper.ZooKeeper; @@ -51,7 +52,6 @@ import com.nearinfinity.blur.concurrent.SimpleUncaughtExceptionHandler; import com.nearinfinity.blur.concurrent.ThreadWatcher; import com.nearinfinity.blur.log.Log; import com.nearinfinity.blur.log.LogFactory; -import com.nearinfinity.blur.lucene.index.TimeBasedIndexDeletionPolicy; import com.nearinfinity.blur.manager.BlurFilterCache; import com.nearinfinity.blur.manager.BlurQueryChecker; import com.nearinfinity.blur.manager.DefaultBlurFilterCache; @@ -155,7 +155,7 @@ public class ThriftBlurShardServer extends ThriftServer { BlurFilterCache filterCache = getFilterCache(configuration); BlurIndexWarmup indexWarmup = getIndexWarmup(configuration); - IndexDeletionPolicy indexDeletionPolicy = new TimeBasedIndexDeletionPolicy(configuration.getLong(BLUR_SHARD_INDEX_DELETION_POLICY_MAXAGE, 60000)); + IndexDeletionPolicy indexDeletionPolicy = new KeepOnlyLastCommitDeletionPolicy(); final DistributedIndexServer indexServer = new DistributedIndexServer(); indexServer.setBlurMetrics(blurMetrics); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/test/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicyTest.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/test/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicyTest.java b/src/blur-core/src/test/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicyTest.java deleted file mode 100644 index de3ea77..0000000 --- a/src/blur-core/src/test/java/com/nearinfinity/blur/lucene/index/TimeBasedIndexDeletionPolicyTest.java +++ /dev/null @@ -1,58 +0,0 @@ -package com.nearinfinity.blur.lucene.index; - -import static org.junit.Assert.assertEquals; - -import java.io.IOException; - -import org.apache.lucene.analysis.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.index.CorruptIndexException; -import org.apache.lucene.index.IndexDeletionPolicy; -import org.apache.lucene.index.IndexReader; -import org.apache.lucene.index.IndexWriterConfig; -import org.apache.lucene.store.LockObtainFailedException; -import org.apache.lucene.store.RAMDirectory; -import org.apache.lucene.util.Version; -import org.junit.Test; - -import com.nearinfinity.blur.index.IndexWriter; - -public class TimeBasedIndexDeletionPolicyTest { - - @Test - public void testTimeBasedIndexDeletionPolicy() throws IOException, InterruptedException { - TimeBasedIndexDeletionPolicy indexDeletionPolicy = new TimeBasedIndexDeletionPolicy(3000); - RAMDirectory directory = new RAMDirectory(); - addAndCommit(directory, indexDeletionPolicy); - addAndCommit(directory, indexDeletionPolicy); - addAndCommit(directory, indexDeletionPolicy); - Thread.sleep(1000); - assertEquals(3, IndexReader.listCommits(directory).size()); - Thread.sleep(4000); - addAndCommit(directory, indexDeletionPolicy); - assertEquals(2, IndexReader.listCommits(directory).size()); - Thread.sleep(4000); - openClose(directory, indexDeletionPolicy); - assertEquals(1, IndexReader.listCommits(directory).size()); - } - - private void openClose(RAMDirectory directory, TimeBasedIndexDeletionPolicy indexDeletionPolicy) throws CorruptIndexException, LockObtainFailedException, IOException { - IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_35, new KeywordAnalyzer()); - conf.setIndexDeletionPolicy(indexDeletionPolicy); - new IndexWriter(directory, conf).close(); - } - - private void addAndCommit(RAMDirectory directory, IndexDeletionPolicy indexDeletionPolicy) throws CorruptIndexException, LockObtainFailedException, IOException { - IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_35, new KeywordAnalyzer()); - conf.setIndexDeletionPolicy(indexDeletionPolicy); - IndexWriter writer = new IndexWriter(directory, conf); - Document doc = new Document(); - doc.add(new Field("id", "1", Store.YES, Index.ANALYZED_NO_NORMS)); - writer.addDocument(doc); - writer.close(); - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-core/src/test/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounterTest.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/test/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounterTest.java b/src/blur-core/src/test/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounterTest.java new file mode 100644 index 0000000..310c575 --- /dev/null +++ b/src/blur-core/src/test/java/com/nearinfinity/blur/manager/writer/DirectoryReferenceCounterTest.java @@ -0,0 +1,324 @@ +package com.nearinfinity.blur.manager.writer; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.LinkedBlockingQueue; + +import org.apache.lucene.analysis.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.index.CorruptIndexException; +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.IndexWriter; +import org.apache.lucene.index.IndexWriterConfig; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.IndexInput; +import org.apache.lucene.store.IndexOutput; +import org.apache.lucene.store.Lock; +import org.apache.lucene.store.LockFactory; +import org.apache.lucene.store.LockObtainFailedException; +import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.util.Version; +import org.junit.Test; + +public class DirectoryReferenceCounterTest { + + @Test + public void testDirectoryReferenceCounterTestError() throws CorruptIndexException, IOException { + Directory directory = wrap(new RAMDirectory()); + IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_36, new KeywordAnalyzer()); + IndexWriter writer = new IndexWriter(directory, conf); + int size = 100; + IndexReader[] readers = new IndexReader[size]; + for (int i = 0; i < size; i++) { + writer.addDocument(getDoc()); + readers[i] = IndexReader.open(writer, true); + writer.forceMerge(1); + } + + try { + for (int i = 0; i < size; i++) { + checkReader(readers[i], i); + } + fail(); + } catch (Exception e) { + //should error + } + } + + @Test + public void testDirectoryReferenceCounter() throws CorruptIndexException, LockObtainFailedException, IOException, InterruptedException { + Directory directory = wrap(new RAMDirectory()); + DirectoryReferenceFileGC gc = new DirectoryReferenceFileGC(); + gc.init(); + DirectoryReferenceCounter counter = new DirectoryReferenceCounter(directory, gc); + IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_36, new KeywordAnalyzer()); + IndexWriter writer = new IndexWriter(counter, conf); + int size = 100; + IndexReader[] readers = new IndexReader[size]; + for (int i = 0; i < size; i++) { + writer.addDocument(getDoc()); + writer.forceMerge(1); + readers[i] = IndexReader.open(writer, true); + } + + for (int i = 0; i < size; i++) { + assertEquals(i + 1, readers[i].numDocs()); + checkReader(readers[i], i); + } + + String[] listAll = directory.listAll(); + + for (int i = 0; i < size - 1; i++) { + readers[i].close(); + } + + for (int i = 0; i < 1000; i++) { + gc.run(); + Thread.sleep(1); + } + + IndexReader last = readers[size - 1]; + + assertEquals(100, last.numDocs()); + + assertTrue(listAll.length > directory.listAll().length); + + last.close(); + writer.close(); + gc.close(); + } + + private Document getDoc() { + Document document = new Document(); + document.add(new Field("id", "value", Store.YES, Index.NOT_ANALYZED_NO_NORMS)); + return document; + } + + private void checkReader(IndexReader indexReader, int size) throws CorruptIndexException, IOException { + for (int i = 0; i < size; i++) { + Document document = indexReader.document(i); + String value = document.get("id"); + assertEquals(value, "value"); + } + } + + // This class is use simulate what would happen with a directory that will forcefully delete files even if they are still in use. e.g. HDFSDirectory + public static Directory wrap(final RAMDirectory ramDirectory) { + return new Directory() { + private Directory d = ramDirectory; + private Collection<String> deletedFiles = new LinkedBlockingQueue<String>(); + + @SuppressWarnings("deprecation") + public void touchFile(String name) throws IOException { + d.touchFile(name); + } + + public void deleteFile(String name) throws IOException { + deletedFiles.add(name); + d.deleteFile(name); + } + + public IndexOutput createOutput(String name) throws IOException { + return d.createOutput(name); + } + + @SuppressWarnings("deprecation") + public void sync(String name) throws IOException { + d.sync(name); + } + + public void sync(Collection<String> names) throws IOException { + d.sync(names); + } + + public IndexInput openInput(String name) throws IOException { + return wrap(d.openInput(name), deletedFiles, name); + } + + public IndexInput openInput(String name, int bufferSize) throws IOException { + return wrap(d.openInput(name, bufferSize), deletedFiles, name); + } + + public void clearLock(String name) throws IOException { + d.clearLock(name); + } + + public void close() throws IOException { + d.close(); + } + + public void setLockFactory(LockFactory lockFactory) throws IOException { + d.setLockFactory(lockFactory); + } + + public String getLockID() { + return d.getLockID(); + } + + public void copy(Directory to, String src, String dest) throws IOException { + d.copy(to, src, dest); + } + + public boolean equals(Object arg0) { + return d.equals(arg0); + } + + public boolean fileExists(String name) throws IOException { + return d.fileExists(name); + } + + @SuppressWarnings("deprecation") + public long fileModified(String name) throws IOException { + return d.fileModified(name); + } + + public long fileLength(String name) throws IOException { + return d.fileLength(name); + } + + public LockFactory getLockFactory() { + return d.getLockFactory(); + } + + public int hashCode() { + return d.hashCode(); + } + + public String[] listAll() throws IOException { + return d.listAll(); + } + + public Lock makeLock(String name) { + return d.makeLock(name); + } + + public String toString() { + return d.toString(); + } + }; + } + + @SuppressWarnings("deprecation") + public static IndexInput wrap(final IndexInput input, final Collection<String> deletedFiles, final String name) { + return new IndexInput() { + private IndexInput in = input; + + private void checkForDeleted() throws IOException { + if (deletedFiles.contains(name)) { + throw new IOException("File [" + name + "] does not exist"); + } + } + + public void skipChars(int length) throws IOException { + checkForDeleted(); + in.skipChars(length); + } + + public void setModifiedUTF8StringsMode() { + in.setModifiedUTF8StringsMode(); + } + + public void close() throws IOException { + checkForDeleted(); + in.close(); + } + + public short readShort() throws IOException { + checkForDeleted(); + return in.readShort(); + } + + public void seek(long pos) throws IOException { + checkForDeleted(); + in.seek(pos); + } + + public int readInt() throws IOException { + checkForDeleted(); + return in.readInt(); + } + + public void copyBytes(IndexOutput out, long numBytes) throws IOException { + checkForDeleted(); + in.copyBytes(out, numBytes); + } + + public int readVInt() throws IOException { + checkForDeleted(); + return in.readVInt(); + } + + public String toString() { + return in.toString(); + } + + public long readLong() throws IOException { + checkForDeleted(); + return in.readLong(); + } + + public long readVLong() throws IOException { + checkForDeleted(); + return in.readVLong(); + } + + public String readString() throws IOException { + checkForDeleted(); + return in.readString(); + } + + public Object clone() { + return super.clone(); + } + + public boolean equals(Object obj) { + return in.equals(obj); + } + + public long getFilePointer() { + return in.getFilePointer(); + } + + public int hashCode() { + return in.hashCode(); + } + + public byte readByte() throws IOException { + checkForDeleted(); + return in.readByte(); + } + + public void readBytes(byte[] b, int offset, int len) throws IOException { + checkForDeleted(); + in.readBytes(b, offset, len); + } + + public void readBytes(byte[] b, int offset, int len, boolean useBuffer) throws IOException { + checkForDeleted(); + in.readBytes(b, offset, len, useBuffer); + } + + public long length() { + return in.length(); + } + + public void readChars(char[] buffer, int start, int length) throws IOException { + checkForDeleted(); + in.readChars(buffer, start, length); + } + + public Map<String, String> readStringStringMap() throws IOException { + checkForDeleted(); + return in.readStringStringMap(); + } + }; + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-store/src/main/java/com/nearinfinity/blur/lucene/LuceneConstant.java ---------------------------------------------------------------------- diff --git a/src/blur-store/src/main/java/com/nearinfinity/blur/lucene/LuceneConstant.java b/src/blur-store/src/main/java/com/nearinfinity/blur/lucene/LuceneConstant.java index 28c6269..781663a 100644 --- a/src/blur-store/src/main/java/com/nearinfinity/blur/lucene/LuceneConstant.java +++ b/src/blur-store/src/main/java/com/nearinfinity/blur/lucene/LuceneConstant.java @@ -4,6 +4,6 @@ import org.apache.lucene.util.Version; public class LuceneConstant { - public static final Version LUCENE_VERSION = Version.LUCENE_35; + public static final Version LUCENE_VERSION = Version.LUCENE_36; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-store/src/main/java/org/apache/lucene/index/WarmUpByFieldBounds.java ---------------------------------------------------------------------- diff --git a/src/blur-store/src/main/java/org/apache/lucene/index/WarmUpByFieldBounds.java b/src/blur-store/src/main/java/org/apache/lucene/index/WarmUpByFieldBounds.java index ba09662..1fa099e 100644 --- a/src/blur-store/src/main/java/org/apache/lucene/index/WarmUpByFieldBounds.java +++ b/src/blur-store/src/main/java/org/apache/lucene/index/WarmUpByFieldBounds.java @@ -23,6 +23,7 @@ import org.apache.lucene.util.Version; import com.nearinfinity.blur.log.Log; import com.nearinfinity.blur.log.LogFactory; +import com.nearinfinity.blur.lucene.LuceneConstant; public class WarmUpByFieldBounds { @@ -54,7 +55,7 @@ public class WarmUpByFieldBounds { private static Directory getDir() throws CorruptIndexException, LockObtainFailedException, IOException { RAMDirectory dir = new RAMDirectory(); - IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_35, new KeywordAnalyzer()); + IndexWriterConfig conf = new IndexWriterConfig(LuceneConstant.LUCENE_VERSION, new KeywordAnalyzer()); TieredMergePolicy mergePolicy = (TieredMergePolicy) conf.getMergePolicy(); mergePolicy.setUseCompoundFile(false); IndexWriter writer = new IndexWriter(dir, conf); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/blur-util/src/main/resources/blur-default.properties ---------------------------------------------------------------------- diff --git a/src/blur-util/src/main/resources/blur-default.properties b/src/blur-util/src/main/resources/blur-default.properties index 2b0784a..fcdeb5f 100644 --- a/src/blur-util/src/main/resources/blur-default.properties +++ b/src/blur-util/src/main/resources/blur-default.properties @@ -8,7 +8,6 @@ blur.shard.cache.max.querycache.elements=128 blur.shard.cache.max.timetolive=60000 blur.shard.filter.cache.class=com.nearinfinity.blur.manager.DefaultBlurFilterCache blur.shard.index.warmup.class=com.nearinfinity.blur.manager.indexserver.DefaultBlurIndexWarmup -blur.shard.index.deletion.policy.maxage=300000 blur.shard.blockcache.direct.memory.allocation=true blur.shard.blockcache.slab.count=1 blur.shard.buffercache.1024=8192 http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/4d530d95/src/pom.xml ---------------------------------------------------------------------- diff --git a/src/pom.xml b/src/pom.xml index 6759613..c48402a 100644 --- a/src/pom.xml +++ b/src/pom.xml @@ -15,6 +15,7 @@ <module>blur-mapred</module> <module>blur-util</module> <module>blur-jdbc</module> + <module>blur-testsuite</module> </modules> <dependencies>
