Method createOutput

bolt/exec/StreamingAggregation.cpp:161–202

}

RowVectorPtr StreamingAggregation::createOutput(size_t numGroups) {
  NanosecondTimer timer(&stats_.aggExtractGroupsTimeNs);
  auto output = BaseVector::create<RowVector>(outputType_, numGroups, pool());

  for (auto i = 0; i < groupingKeys_.size(); ++i) {
    rows_->extractColumn(groups_.data(), numGroups, i, output->childAt(i));
  }

  auto numKeys = groupingKeys_.size();
  for (auto i = 0; i < aggregates_.size(); ++i) {
    const auto& aggregate = aggregates_.at(i);
    if (!aggregate.sortingKeys.empty()) {
      continue;
    }

    if (aggregate.distinct) {
      continue;
    }

    const auto& function = aggregate.function;
    auto& result = output->childAt(numKeys + i);
    if (isPartialOutput(step_)) {
      function->extractAccumulators(groups_.data(), numGroups, &result);
    } else {
      function->extractValues(groups_.data(), numGroups, &result);
    }
  }

  if (sortedAggregations_) {
    sortedAggregations_->extractValues(
        folly::Range<char**>(groups_.data(), numGroups), output);
  }

  for (const auto& aggregation : distinctAggregations_) {
    if (aggregation != nullptr) {
      aggregation->extractValues(
          folly::Range<char**>(groups_.data(), numGroups), output);
    }
  }

  return output;
}

void StreamingAggregation::assignGroups() {
  auto numInput = input_->size();