@Test public void testSearch() throws Exception { kickSearchRequests(2); prepareTestData(); segmentStore.doMergeFragments(Lists.newArrayList(segmentStore.getAllFragments())); Thread.sleep(5000); cleanupData(); }
@Override public void search(final StreamingSearchContext searchContext, ResultCollector collector) throws IOException { SegmentMemoryStore searchMemoryStore; List<DataSegmentFragment> searchFragments; mergeReadLock.lock(); collector.addCloseListener(new CloseListener() { @Override public void onClose() { mergeReadLock.unlock(); } }); persistReadLock.lock(); try { searchFragments = getAllFragments(); if (persisting) { searchMemoryStore = persistingMemoryStore; } else { searchMemoryStore = activeMemoryStore; } } finally { persistReadLock.unlock(); } new ColumnarSegmentStoreFilesSearcher(segmentName, searchFragments).search(searchContext, collector); searchMemoryStore.search(searchContext, collector); }
ColumnarSegmentStore implements IStreamingSegmentStore { @Override public void search(final StreamingSearchContext searchContext, ResultCollector collector) throws IOException { SegmentMemoryStore searchMemoryStore; List<DataSegmentFragment> searchFragments; mergeReadLock.lock(); collector.addCloseListener(new CloseListener() { @Override public void onClose() { mergeReadLock.unlock(); } }); persistReadLock.lock(); try { searchFragments = getAllFragments(); if (persisting) { searchMemoryStore = persistingMemoryStore; } else { searchMemoryStore = activeMemoryStore; } } finally { persistReadLock.unlock(); } new ColumnarSegmentStoreFilesSearcher(segmentName, searchFragments).search(searchContext, collector); searchMemoryStore.search(searchContext, collector); } }
ColumnarSegmentStore implements IStreamingSegmentStore { @Override public void search(final StreamingSearchContext searchContext, ResultCollector collector) throws IOException { SegmentMemoryStore searchMemoryStore; List<DataSegmentFragment> searchFragments; mergeReadLock.lock(); collector.addCloseListener(new CloseListener() { @Override public void onClose() { mergeReadLock.unlock(); } }); persistReadLock.lock(); try { searchFragments = getAllFragments(); if (persisting) { searchMemoryStore = persistingMemoryStore; } else { searchMemoryStore = activeMemoryStore; } } finally { persistReadLock.unlock(); } new ColumnarSegmentStoreFilesSearcher(segmentName, searchFragments).search(searchContext, collector); searchMemoryStore.search(searchContext, collector); } ColumnarSegmentStore(String baseStorePath, CubeInstance cubeInstance, String segmentName); }
ColumnarSegmentStore implements IStreamingSegmentStore { @Override public void search(final StreamingSearchContext searchContext, ResultCollector collector) throws IOException { SegmentMemoryStore searchMemoryStore; List<DataSegmentFragment> searchFragments; mergeReadLock.lock(); collector.addCloseListener(new CloseListener() { @Override public void onClose() { mergeReadLock.unlock(); } }); persistReadLock.lock(); try { searchFragments = getAllFragments(); if (persisting) { searchMemoryStore = persistingMemoryStore; } else { searchMemoryStore = activeMemoryStore; } } finally { persistReadLock.unlock(); } new ColumnarSegmentStoreFilesSearcher(segmentName, searchFragments).search(searchContext, collector); searchMemoryStore.search(searchContext, collector); } ColumnarSegmentStore(String baseStorePath, CubeInstance cubeInstance, String segmentName); @Override void init(); int addEvent(StreamingMessage event); @Override void addExternalDict(Map<TblColRef, Dictionary<String>> dictMap); @Override File getStorePath(); @Override Object checkpoint(); @Override void persist(); @Override void purge(); @Override void restoreFromCheckpoint(Object checkpoint); @Override String getSegmentName(); @Override StreamingCubeSegment.State getSegmentState(); @Override void setSegmentState(StreamingCubeSegment.State state); DataSegmentFragment createNewFragment(); List<DataSegmentFragment> getAllFragments(); @Override SegmentStoreStats getStoreStats(); SegmentMemoryStore getActiveMemoryStore(); @Override void search(final StreamingSearchContext searchContext, ResultCollector collector); void close(); }
ColumnarSegmentStore implements IStreamingSegmentStore { @Override public void search(final StreamingSearchContext searchContext, ResultCollector collector) throws IOException { SegmentMemoryStore searchMemoryStore; List<DataSegmentFragment> searchFragments; mergeReadLock.lock(); collector.addCloseListener(new CloseListener() { @Override public void onClose() { mergeReadLock.unlock(); } }); persistReadLock.lock(); try { searchFragments = getAllFragments(); if (persisting) { searchMemoryStore = persistingMemoryStore; } else { searchMemoryStore = activeMemoryStore; } } finally { persistReadLock.unlock(); } new ColumnarSegmentStoreFilesSearcher(segmentName, searchFragments).search(searchContext, collector); searchMemoryStore.search(searchContext, collector); } ColumnarSegmentStore(String baseStorePath, CubeInstance cubeInstance, String segmentName); @Override void init(); int addEvent(StreamingMessage event); @Override void addExternalDict(Map<TblColRef, Dictionary<String>> dictMap); @Override File getStorePath(); @Override Object checkpoint(); @Override void persist(); @Override void purge(); @Override void restoreFromCheckpoint(Object checkpoint); @Override String getSegmentName(); @Override StreamingCubeSegment.State getSegmentState(); @Override void setSegmentState(StreamingCubeSegment.State state); DataSegmentFragment createNewFragment(); List<DataSegmentFragment> getAllFragments(); @Override SegmentStoreStats getStoreStats(); SegmentMemoryStore getActiveMemoryStore(); @Override void search(final StreamingSearchContext searchContext, ResultCollector collector); void close();  }