processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ParallelCombiner.java
/**
   * Build a combining tree for the input iterators which combine input entries asynchronously.  Each node in the tree
   * is a combining task which iterates through child iterators, aggregates the inputs from those iterators, and returns
   * an iterator for the result of aggregation.
   * <p>
   * This method is called when data is spilled and thus streaming combine is preferred to avoid too many disk accesses.
   *
   * @return an iterator of the root grouper of the combining tree
   */
Build a combining tree for the input iterators which combine input entries asynchronously.  Each node in the tree
is a combining task which iterates through child iterators, aggregates the inputs from those iterators, and returns
an iterator for the result of aggregation.

This method is called when data is spilled and thus streaming combine is preferred to avoid too many disk accesses.

@return an iterator of the root grouper of the combining tree
Build a combining tree for the input iterators which combine input entries asynchronously . Each node in the tree is a combining task which iterates through child iterators aggregates the inputs from those iterators and returns an iterator for the result of aggregation . This method is called when data is spilled and thus streaming combine is preferred to avoid too many disk accesses . @return an iterator of the root grouper of the combining tree
public CloseableIterator<Entry<KeyType>> combine(
      List<? extends CloseableIterator<Entry<KeyType>>> sortedIterators,
      List<String> mergedDictionary
  )
  {
    final Closer closer = Closer.create();
    try {
      final ByteBuffer combineBuffer = combineBufferHolder.get();
      final int minimumRequiredBufferCapacity = StreamingMergeSortedGrouper.requiredBufferCapacity(
          combineKeySerdeFactory.factorizeWithDictionary(mergedDictionary),
          combiningFactories
      );
      final Pair<Integer, Integer> degreeAndNumBuffers = findLeafCombineDegreeAndNumBuffers(
          combineBuffer,
          minimumRequiredBufferCapacity,
          concurrencyHint,
          sortedIterators.size()
      );
      final int leafCombineDegree = degreeAndNumBuffers.lhs;
      final int numBuffers = degreeAndNumBuffers.rhs;
      final int sliceSize = combineBuffer.capacity() / numBuffers;
      final Supplier<ByteBuffer> bufferSupplier = createCombineBufferSupplier(combineBuffer, numBuffers, sliceSize);
      final Pair<List<CloseableIterator<Entry<KeyType>>>, List<Future>> combineIteratorAndFutures = buildCombineTree(
          sortedIterators,
          bufferSupplier,
          combiningFactories,
          leafCombineDegree,
          mergedDictionary
      );
      final CloseableIterator<Entry<KeyType>> combineIterator = Iterables.getOnlyElement(combineIteratorAndFutures.lhs);
      final List<Future> combineFutures = combineIteratorAndFutures.rhs;
      closer.register(() -> checkCombineFutures(combineFutures));
      return CloseableIterators.wrap(combineIterator, closer);
    }
    catch (Throwable t) {
      try {
        closer.close();
      }
      catch (Throwable t2) {
        t.addSuppressed(t2);
      }
      throw t;
    }
  }
public CloseableIterator < Entry < KeyType > > combine ( List < ? extends CloseableIterator < Entry < KeyType > > > sortedIterators , List < String > mergedDictionary ) { final Closer closer = Closer . create ( ) ; try { final ByteBuffer combineBuffer = combineBufferHolder . get ( ) ; final int minimumRequiredBufferCapacity = StreamingMergeSortedGrouper . requiredBufferCapacity ( combineKeySerdeFactory . factorizeWithDictionary ( mergedDictionary ) , combiningFactories ) ; final Pair < Integer , Integer > degreeAndNumBuffers = findLeafCombineDegreeAndNumBuffers ( combineBuffer , minimumRequiredBufferCapacity , concurrencyHint , sortedIterators . size ( ) ) ; final int leafCombineDegree = degreeAndNumBuffers . lhs ; final int numBuffers = degreeAndNumBuffers . rhs ; final int sliceSize = combineBuffer . capacity ( ) / numBuffers ; final Supplier < ByteBuffer > bufferSupplier = createCombineBufferSupplier ( combineBuffer , numBuffers , sliceSize ) ; final Pair < List < CloseableIterator < Entry < KeyType > > > , List < Future > > combineIteratorAndFutures = buildCombineTree ( sortedIterators , bufferSupplier , combiningFactories , leafCombineDegree , mergedDictionary ) ; final CloseableIterator < Entry < KeyType > > combineIterator = Iterables . getOnlyElement ( combineIteratorAndFutures . lhs ) ; final List < Future > combineFutures = combineIteratorAndFutures . rhs ; closer . register ( ( ) -> checkCombineFutures ( combineFutures ) ) ; return CloseableIterators . wrap ( combineIterator , closer ) ; } catch ( Throwable t ) { try { closer . close ( ) ; } catch ( Throwable t2 ) { t . addSuppressed ( t2 ) ; } throw t ; } }
Build a combining tree for the input iterators which combine input entries asynchronously.
Build a combining tree for the input iterators which combine input entries asynchronously .
// CombineBuffer is initialized when this method is called and closed after the result iterator is done // We want to maximize the parallelism while the size of buffer slice is greater than the minimum buffer size // required by StreamingMergeSortedGrouper. Here, we find the leafCombineDegree of the cominbing tree and the // required number of buffers maximizing the parallelism.