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

Added a bit of docs

This commit is contained in:
Justin Borromeo 2019-03-22 15:19:52 -07:00
parent 49472162b7
commit 1b46b58aec
1 changed files with 11 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
processing/src/main/java/org/apache/druid/query/scan/ScanQueryRunnerFactory.java
View File

@ -164,6 +164,9 @@ public class ScanQueryRunnerFactory implements QueryRunnerFactory<ScanResultValu
if (maxNumPartitionsInSegment <= scanQueryConfig.getMaxSegmentPartitionsOrderedInMemory()) { if (maxNumPartitionsInSegment <= scanQueryConfig.getMaxSegmentPartitionsOrderedInMemory()) {
// Use n-way merge strategy // Use n-way merge strategy
// Create a list of grouped runner lists (i.e. each sublist/"runner group" corresponds to an interval) ->
// there should be no interval overlap
List<List<QueryRunner<ScanResultValue>>> groupedRunners = List<List<QueryRunner<ScanResultValue>>> groupedRunners =
partitionsGroupedByInterval.entrySet() partitionsGroupedByInterval.entrySet()
.stream() .stream()
@ -172,14 +175,16 @@ public class ScanQueryRunnerFactory implements QueryRunnerFactory<ScanResultValu
.map(segQueryRunnerPair -> segQueryRunnerPair.rhs) .map(segQueryRunnerPair -> segQueryRunnerPair.rhs)
.collect(Collectors.toList())) .collect(Collectors.toList()))
.collect(Collectors.toList()); .collect(Collectors.toList());
return Sequences.concat(
Sequences.map(
Sequences.simple(groupedRunners), // Sequence of runnerGroups return Sequences.concat( // 5) Join all the results into a single sequence
Sequences.map( // 4) Create a sequence of results from each runner group
Sequences.simple(groupedRunners),
runnerGroup -> runnerGroup ->
Sequences.map( Sequences.map( // 3) Create a sequence of results from each runner in the group and flatmerge based on timestamp
Sequences.simple(runnerGroup), Sequences.simple(runnerGroup),
(input) -> Sequences.concat( (input) -> Sequences.concat( // 2) Combine the deaggregated ScanResultValues into a single sequence
Sequences.map( Sequences.map( // 1) Deaggregate each ScanResultValue returned by the query runners
input.run(queryPlus, responseContext), input.run(queryPlus, responseContext),
srv -> Sequences.simple(srv.toSingleEventScanResultValues()) srv -> Sequences.simple(srv.toSingleEventScanResultValues())
) )
@ -278,8 +283,6 @@ public class ScanQueryRunnerFactory implements QueryRunnerFactory<ScanResultValu
return Sequences.simple(sortedElements); return Sequences.simple(sortedElements);
} }
@VisibleForTesting
@Override @Override
public QueryToolChest<ScanResultValue, ScanQuery> getToolchest() public QueryToolChest<ScanResultValue, ScanQuery> getToolchest()