Method initialize

bolt/exec/StreamingAggregation.cpp:63–105

}

void StreamingAggregation::initialize() {
  Operator::initialize();

  auto numKeys = aggregationNode_->groupingKeys().size();
  decodedKeys_.resize(numKeys);

  auto inputType = aggregationNode_->sources()[0]->outputType();

  std::vector<TypePtr> groupingKeyTypes;
  groupingKeyTypes.reserve(numKeys);

  groupingKeys_.reserve(numKeys);
  for (const auto& key : aggregationNode_->groupingKeys()) {
    auto channel = exprToChannel(key.get(), inputType);
    groupingKeys_.push_back(channel);
    groupingKeyTypes.push_back(inputType->childAt(channel));
  }

  std::shared_ptr<core::ExpressionEvaluator> expressionEvaluator;
  aggregates_ = toAggregateInfo(
      *aggregationNode_, *operatorCtx_, numKeys, expressionEvaluator, true);

  // Setup SortedAggregations.
  sortedAggregations_ =
      SortedAggregations::create(aggregates_, inputType, pool());

  distinctAggregations_.reserve(aggregates_.size());
  for (auto& aggregate : aggregates_) {
    if (aggregate.distinct) {
      distinctAggregations_.emplace_back(
          DistinctAggregations::create({&aggregate}, inputType, pool()));
    } else {
      distinctAggregations_.push_back(nullptr);
    }
  }

  masks_ = std::make_unique<AggregationMasks>(extractMaskChannels(aggregates_));
  rows_ = makeRowContainer(groupingKeyTypes);

  initializeAggregates(numKeys);

  aggregationNode_.reset();
}

void StreamingAggregation::close() {
  if (rows_ != nullptr) {