Method mergeNextWithAggregates

bolt/exec/GroupingSet.cpp:1734–1857

            container,
            result,
            initGroupCount);
        outputUniqueGroups(
            uniqueRows, uniqueRes, uniqueCount, estimateUniqueBytesPerRow);

        if (initGroupCount) {
          extractSpillResult(result, true);
        }
        if (uniqueCount) {
          result->append(uniqueRes->wrappedVector());
          stats_.aggOutputUniqueRows += uniqueCount;
        }
        return true;
      }
    }
  }
}

bool GroupingSet::mergeNextWithAggregates(
    int32_t maxOutputRows,
    int32_t maxOutputBytes,
    const RowVectorPtr& result) {
  // Merge all spill data, update aggregated intermediate result into groups,
  // and output the aggregated result. First, read the merged data and
  // temporarily store each row in a vector until the batch of a stream ends or
  // the number of rows reaches the limit. Then, copy all distinct keys to the
  // result and update the intermediate result to the groups until result size
  // reaches limit or all data processing is completed, finally extract result
  // in groups into result vector and return result.
  BOLT_CHECK_NOT_NULL(merge_);
  BOLT_CHECK(!isDistinct());

  // True if 'merge_' indicates that the next key is the same as the current
  // one.
  bool nextKeyIsEqual{false};

  // stores all rows need to be update to groups
  std::vector<const RowVector*> sources;
  std::vector<vector_size_t> sourceIndice;
  std::vector<char*> groupOfRows;

  // all distinct rows to extract keys
  std::vector<const RowVector*> distinctInputs;
  std::vector<vector_size_t> distinctIndices;
  std::vector<char*> groups;

  RowVectorPtr intermediate;
  uint64_t averageResultRowSize = 0;

  // gather all saved keys into result vector and all saved intermediate result
  // into intermediate and update into groups
  auto gatherAndUpdate = [&]() {
    NanosecondTimer aggTimer(&stats_.aggOutputUpdateTimeNs);
    if (!distinctInputs.empty()) {
      // copy keys to result
      auto oldResultSize = result->size();
      result->resize(oldResultSize + distinctInputs.size());
      for (auto i = 0; i < keyChannels_.size(); ++i) {
        gatherCopy(