new ArrayList<Future<ResultObjectProvider>>();
final List<Executor> usedExecutors = new ArrayList<Executor>();
final List<ResultObjectProvider> rops =
new ArrayList<ResultObjectProvider>();
List<SliceStoreManager> targets = findTargets();
QueryContext ctx = q.getContext();
boolean isReplicated = containsReplicated(ctx);
ExecutorService threadPool = SliceThread.getPool();
for (int i = 0; i < owner._queries.size(); i++) {
// if replicated, then execute only on single slice
if (isReplicated && !usedExecutors.isEmpty()) {
break;
}
StoreManager sm = owner.getDistributedStore().getSlice(i);
if (!targets.contains(sm))
continue;
StoreQuery query = owner._queries.get(i);
Executor executor = executors.get(i);
if (!targets.contains(sm))
continue;
usedExecutors.add(executor);
QueryExecutor call = new QueryExecutor();
call.executor = executor;
call.query = query;
call.params = params;
call.range = range;
futures.add(threadPool.submit(call));
}
for (Future<ResultObjectProvider> future : futures) {
try {
rops.add(future.get());
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new StoreException(e.getCause());
}
}
ResultObjectProvider[] tmp = rops
.toArray(new ResultObjectProvider[rops.size()]);
ResultObjectProvider result = null;
boolean[] ascending = getAscending(q);
boolean isAscending = ascending.length > 0;
boolean isAggregate = ctx.isAggregate();
boolean hasRange = ctx.getEndRange() != Long.MAX_VALUE;
if (isAggregate) {
result = new UniqueResultObjectProvider(tmp, q,
getQueryExpressions());
} else if (isAscending) {
result = new OrderingMergedResultObjectProvider(tmp, ascending,
usedExecutors.toArray(new Executor[usedExecutors.size()]),
q, params);
} else {
result = new MergedResultObjectProvider(tmp);
}
if (hasRange) {
result = new RangeResultObjectProvider(result,
ctx.getStartRange(), ctx.getEndRange());
}
return result;
}