@Test public void testTerms() throws Exception { List<String> terms = indexManager.terms(TABLE, FAMILY, "testcol1", "", (short) 100); assertEquals(Arrays.asList("value1", "value13", "value16", "value4", "value7"), terms); }
public List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size) throws Exception { Map<String, BlurIndex> blurIndexes; try { blurIndexes = _indexServer.getIndexes(table); } catch (IOException e) { LOG.error("Unknown error while trying to fetch index readers.", e); throw new BException(e.getMessage(), e); } return ForkJoin.execute(_executor, blurIndexes.entrySet(), new ParallelCall<Entry<String, BlurIndex>, List<String>>() { @Override public List<String> call(Entry<String, BlurIndex> input) throws Exception { BlurIndex index = input.getValue(); IndexReader reader = index.getIndexReader(); try { return terms(reader, columnFamily, columnName, startWith, size); } finally { reader.decRef(); } } }).merge(new Merger<List<String>>() { @Override public List<String> merge(BlurExecutorCompletionService<List<String>> service) throws BlurException { TreeSet<String> terms = new TreeSet<String>(); while (service.getRemainingCount() > 0) { Future<List<String>> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, columnFamily, columnName, startWith, size); terms.addAll(service.getResultThrowException(future, table, columnFamily, columnName, startWith, size)); } return new ArrayList<String>(terms).subList(0, Math.min(size, terms.size())); } }); }
IndexManager { public List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size) throws Exception { Map<String, BlurIndex> blurIndexes; try { blurIndexes = _indexServer.getIndexes(table); } catch (IOException e) { LOG.error("Unknown error while trying to fetch index readers.", e); throw new BException(e.getMessage(), e); } return ForkJoin.execute(_executor, blurIndexes.entrySet(), new ParallelCall<Entry<String, BlurIndex>, List<String>>() { @Override public List<String> call(Entry<String, BlurIndex> input) throws Exception { BlurIndex index = input.getValue(); IndexReader reader = index.getIndexReader(); try { return terms(reader, columnFamily, columnName, startWith, size); } finally { reader.decRef(); } } }).merge(new Merger<List<String>>() { @Override public List<String> merge(BlurExecutorCompletionService<List<String>> service) throws BlurException { TreeSet<String> terms = new TreeSet<String>(); while (service.getRemainingCount() > 0) { Future<List<String>> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, columnFamily, columnName, startWith, size); terms.addAll(service.getResultThrowException(future, table, columnFamily, columnName, startWith, size)); } return new ArrayList<String>(terms).subList(0, Math.min(size, terms.size())); } }); } }
IndexManager { public List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size) throws Exception { Map<String, BlurIndex> blurIndexes; try { blurIndexes = _indexServer.getIndexes(table); } catch (IOException e) { LOG.error("Unknown error while trying to fetch index readers.", e); throw new BException(e.getMessage(), e); } return ForkJoin.execute(_executor, blurIndexes.entrySet(), new ParallelCall<Entry<String, BlurIndex>, List<String>>() { @Override public List<String> call(Entry<String, BlurIndex> input) throws Exception { BlurIndex index = input.getValue(); IndexReader reader = index.getIndexReader(); try { return terms(reader, columnFamily, columnName, startWith, size); } finally { reader.decRef(); } } }).merge(new Merger<List<String>>() { @Override public List<String> merge(BlurExecutorCompletionService<List<String>> service) throws BlurException { TreeSet<String> terms = new TreeSet<String>(); while (service.getRemainingCount() > 0) { Future<List<String>> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, columnFamily, columnName, startWith, size); terms.addAll(service.getResultThrowException(future, table, columnFamily, columnName, startWith, size)); } return new ArrayList<String>(terms).subList(0, Math.min(size, terms.size())); } }); }  }
IndexManager { public List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size) throws Exception { Map<String, BlurIndex> blurIndexes; try { blurIndexes = _indexServer.getIndexes(table); } catch (IOException e) { LOG.error("Unknown error while trying to fetch index readers.", e); throw new BException(e.getMessage(), e); } return ForkJoin.execute(_executor, blurIndexes.entrySet(), new ParallelCall<Entry<String, BlurIndex>, List<String>>() { @Override public List<String> call(Entry<String, BlurIndex> input) throws Exception { BlurIndex index = input.getValue(); IndexReader reader = index.getIndexReader(); try { return terms(reader, columnFamily, columnName, startWith, size); } finally { reader.decRef(); } } }).merge(new Merger<List<String>>() { @Override public List<String> merge(BlurExecutorCompletionService<List<String>> service) throws BlurException { TreeSet<String> terms = new TreeSet<String>(); while (service.getRemainingCount() > 0) { Future<List<String>> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, columnFamily, columnName, startWith, size); terms.addAll(service.getResultThrowException(future, table, columnFamily, columnName, startWith, size)); } return new ArrayList<String>(terms).subList(0, Math.min(size, terms.size())); } }); }  void setMaxClauseCount(int maxClauseCount); void init(); synchronized void close(); void fetchRow(String table, Selector selector, FetchResult fetchResult); static void validSelector(Selector selector); BlurResultIterable query(final String table, final BlurQuery blurQuery, AtomicLongArray facetedCounts); void cancelQuery(String table, long uuid); List<BlurQueryStatus> currentQueries(String table); BlurQueryStatus queryStatus(String table, long uuid); List<Long> queryStatusIdList(String table); static void fetchRow(IndexReader reader, String table, Selector selector, FetchResult fetchResult); static FieldSelector getFieldSelector(final Selector selector); IndexServer getIndexServer(); void setIndexServer(IndexServer indexServer); long recordFrequency(final String table, final String columnFamily, final String columnName, final String value); List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size); static long recordFrequency(IndexReader reader, String columnFamily, String columnName, String value); static List<String> terms(IndexReader reader, String columnFamily, String columnName, String startWith, short size); Schema schema(String table); void setStatusCleanupTimerDelay(long delay); void mutate(final RowMutation mutation); void mutate(final List<RowMutation> mutations); void setThreadCount(int threadCount); void setBlurMetrics(BlurMetrics blurMetrics); void setFilterCache(BlurFilterCache filterCache); void optimize(String table, int numberOfSegmentsPerShard); }
IndexManager { public List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size) throws Exception { Map<String, BlurIndex> blurIndexes; try { blurIndexes = _indexServer.getIndexes(table); } catch (IOException e) { LOG.error("Unknown error while trying to fetch index readers.", e); throw new BException(e.getMessage(), e); } return ForkJoin.execute(_executor, blurIndexes.entrySet(), new ParallelCall<Entry<String, BlurIndex>, List<String>>() { @Override public List<String> call(Entry<String, BlurIndex> input) throws Exception { BlurIndex index = input.getValue(); IndexReader reader = index.getIndexReader(); try { return terms(reader, columnFamily, columnName, startWith, size); } finally { reader.decRef(); } } }).merge(new Merger<List<String>>() { @Override public List<String> merge(BlurExecutorCompletionService<List<String>> service) throws BlurException { TreeSet<String> terms = new TreeSet<String>(); while (service.getRemainingCount() > 0) { Future<List<String>> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, columnFamily, columnName, startWith, size); terms.addAll(service.getResultThrowException(future, table, columnFamily, columnName, startWith, size)); } return new ArrayList<String>(terms).subList(0, Math.min(size, terms.size())); } }); }  void setMaxClauseCount(int maxClauseCount); void init(); synchronized void close(); void fetchRow(String table, Selector selector, FetchResult fetchResult); static void validSelector(Selector selector); BlurResultIterable query(final String table, final BlurQuery blurQuery, AtomicLongArray facetedCounts); void cancelQuery(String table, long uuid); List<BlurQueryStatus> currentQueries(String table); BlurQueryStatus queryStatus(String table, long uuid); List<Long> queryStatusIdList(String table); static void fetchRow(IndexReader reader, String table, Selector selector, FetchResult fetchResult); static FieldSelector getFieldSelector(final Selector selector); IndexServer getIndexServer(); void setIndexServer(IndexServer indexServer); long recordFrequency(final String table, final String columnFamily, final String columnName, final String value); List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size); static long recordFrequency(IndexReader reader, String columnFamily, String columnName, String value); static List<String> terms(IndexReader reader, String columnFamily, String columnName, String startWith, short size); Schema schema(String table); void setStatusCleanupTimerDelay(long delay); void mutate(final RowMutation mutation); void mutate(final List<RowMutation> mutations); void setThreadCount(int threadCount); void setBlurMetrics(BlurMetrics blurMetrics); void setFilterCache(BlurFilterCache filterCache); void optimize(String table, int numberOfSegmentsPerShard);  }