processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ParallelCombiner.java
/**
   * Recursively build a combining tree in a bottom-up manner.  Each node of the tree is a task that combines input
   * iterators asynchronously.
   *
   * @param childIterators     all iterators of the child level
   * @param bufferSupplier     combining buffer supplier
   * @param combiningFactories array of combining aggregator factories
   * @param combineDegree      combining degree for the current level
   * @param dictionary         merged dictionary
   *
   * @return a pair of a list of iterators of the current level in the combining tree and a list of futures of all
   * executed combining tasks
   */
Recursively build a combining tree in a bottom-up manner.  Each node of the tree is a task that combines input
iterators asynchronously.



@return a pair of a list of iterators of the current level in the combining tree and a list of futures of all
executed combining tasks
Recursively build a combining tree in a bottom - up manner . Each node of the tree is a task that combines input iterators asynchronously . @return a pair of a list of iterators of the current level in the combining tree and a list of futures of all executed combining tasks
private Pair<List<CloseableIterator<Entry<KeyType>>>, List<Future>> buildCombineTree(
      List<? extends CloseableIterator<Entry<KeyType>>> childIterators,
      Supplier<ByteBuffer> bufferSupplier,
      AggregatorFactory[] combiningFactories,
      int combineDegree,
      List<String> dictionary
  )
  {
    final int numChildLevelIterators = childIterators.size();
    final List<CloseableIterator<Entry<KeyType>>> childIteratorsOfNextLevel = new ArrayList<>();
    final List<Future> combineFutures = new ArrayList<>();
          o
         / \
        o   \
       / \   \
      o   o   o
     / \ / \ / \
     o o o o o o
     We can expect that the aggregates can be combined as early as possible because the tree is built in a bottom-up
     manner.
    for (int i = 0; i < numChildLevelIterators; i += combineDegree) {
      if (i < numChildLevelIterators - 1) {
        final List<? extends CloseableIterator<Entry<KeyType>>> subIterators = childIterators.subList(
            i,
            Math.min(i + combineDegree, numChildLevelIterators)
        );
        final Pair<CloseableIterator<Entry<KeyType>>, Future> iteratorAndFuture = runCombiner(
            subIterators,
            bufferSupplier.get(),
            combiningFactories,
            dictionary
        );
        childIteratorsOfNextLevel.add(iteratorAndFuture.lhs);
        combineFutures.add(iteratorAndFuture.rhs);
      } else {
         If there remains one child, it can be directly connected to a node of the parent level.
        childIteratorsOfNextLevel.add(childIterators.get(i));
      }
    }
    if (childIteratorsOfNextLevel.size() == 1) {
       This is the root
      return Pair.of(childIteratorsOfNextLevel, combineFutures);
    } else {
       Build the parent level iterators
      final Pair<List<CloseableIterator<Entry<KeyType>>>, List<Future>> parentIteratorsAndFutures =
          buildCombineTree(
              childIteratorsOfNextLevel,
              bufferSupplier,
              combiningFactories,
              intermediateCombineDegree,
              dictionary
          );
      combineFutures.addAll(parentIteratorsAndFutures.rhs);
      return Pair.of(parentIteratorsAndFutures.lhs, combineFutures);
    }
  }
private Pair < List < CloseableIterator < Entry < KeyType > > > , List < Future > > buildCombineTree ( List < ? extends CloseableIterator < Entry < KeyType > > > childIterators , Supplier < ByteBuffer > bufferSupplier , AggregatorFactory [ ] combiningFactories , int combineDegree , List < String > dictionary ) { final int numChildLevelIterators = childIterators . size ( ) ; final List < CloseableIterator < Entry < KeyType > > > childIteratorsOfNextLevel = new ArrayList < > ( ) ; final List < Future > combineFutures = new ArrayList < > ( ) ; for ( int i = 0 ; i < numChildLevelIterators ; i += combineDegree ) { if ( i < numChildLevelIterators - 1 ) { final List < ? extends CloseableIterator < Entry < KeyType > > > subIterators = childIterators . subList ( i , Math . min ( i + combineDegree , numChildLevelIterators ) ) ; final Pair < CloseableIterator < Entry < KeyType > > , Future > iteratorAndFuture = runCombiner ( subIterators , bufferSupplier . get ( ) , combiningFactories , dictionary ) ; childIteratorsOfNextLevel . add ( iteratorAndFuture . lhs ) ; combineFutures . add ( iteratorAndFuture . rhs ) ; } else { childIteratorsOfNextLevel . add ( childIterators . get ( i ) ) ; } } if ( childIteratorsOfNextLevel . size ( ) == 1 ) { return Pair . of ( childIteratorsOfNextLevel , combineFutures ) ; } else { final Pair < List < CloseableIterator < Entry < KeyType > > > , List < Future > > parentIteratorsAndFutures = buildCombineTree ( childIteratorsOfNextLevel , bufferSupplier , combiningFactories , intermediateCombineDegree , dictionary ) ; combineFutures . addAll ( parentIteratorsAndFutures . rhs ) ; return Pair . of ( parentIteratorsAndFutures . lhs , combineFutures ) ; } }
Recursively build a combining tree in a bottom-up manner.
Recursively build a combining tree in a bottom - up manner .
// The below algorithm creates the combining nodes of the current level. It first checks that the number of children // to be combined together is 1. If it is, the intermediate combining node for that child is not needed. Instead, it // can be directly connected to a node of the parent level. Here is an example of generated tree when // numLeafNodes = 6 and leafCombineDegree = intermediateCombineDegree = 2. See the description of // MINIMUM_LEAF_COMBINE_DEGREE for more details about leafCombineDegree and intermediateCombineDegree. // //      o //     / \ //    o   \ //   / \   \ //  o   o   o // / \ / \ / \ // o o o o o o // // We can expect that the aggregates can be combined as early as possible because the tree is built in a bottom-up // manner. // If there remains one child, it can be directly connected to a node of the parent level. // This is the root // Build the parent level iterators