Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

address cr

This commit is contained in:
fjy 2014-11-14 13:28:32 -08:00
parent bbc079b880
commit 7736c3fc27
1 changed files with 26 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
processing/src/main/java/io/druid/query/RetryQueryRunner.java
View File

@ -61,8 +61,7 @@ public class RetryQueryRunner<T> implements QueryRunner<T>
public Sequence<T> run(final Query<T> query, final Map<String, Object> context)
{
final List<Sequence<T>> listOfSequences = Lists.newArrayList();
final Sequence<T> returningSeq = baseRunner.run(query, context);
listOfSequences.add(returningSeq);
listOfSequences.add(baseRunner.run(query, context));
return new YieldingSequenceBase<T>()
{
@ -73,10 +72,7 @@ public class RetryQueryRunner<T> implements QueryRunner<T>
{
final List<SegmentDescriptor> missingSegments = getMissingSegments(context);
if (missingSegments.isEmpty()) {
return returningSeq.toYielder(initValue, accumulator);
}
if (!missingSegments.isEmpty()) {
for (int i = 0; i < config.numTries(); i++) {
log.info("[%,d] missing segments found. Retry attempt [%,d]", missingSegments.size(), i);
@ -97,8 +93,17 @@ public class RetryQueryRunner<T> implements QueryRunner<T>
if (!config.returnPartialResults() && !finalMissingSegs.isEmpty()) {
throw new SegmentMissingException("No results found for segments[%s]", finalMissingSegs);
}
}
return toolChest.mergeSequencesUnordered(Sequences.simple(listOfSequences)).toYielder(initValue, accumulator);
final Sequence<T> retSeq;
if (listOfSequences.size() == 1) {
retSeq = listOfSequences.get(0);
} else {
retSeq = toolChest.mergeSequencesUnordered(Sequences.simple(listOfSequences));
}
return retSeq.toYielder(initValue, accumulator);
}
};
}