Method initialize

bolt/exec/HashAggregation.cpp:103–230

}

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

  BOLT_CHECK(pool()->trackUsage());

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

  auto hashers =
      createVectorHashers(inputType, aggregationNode_->groupingKeys());
  auto numHashers = hashers.size();

  std::vector<column_index_t> preGroupedChannels;
  preGroupedChannels.reserve(aggregationNode_->preGroupedKeys().size());
  for (const auto& key : aggregationNode_->preGroupedKeys()) {
    auto channel = exprToChannel(key.get(), inputType);
    preGroupedChannels.push_back(channel);
  }

  std::shared_ptr<core::ExpressionEvaluator> expressionEvaluator;
  std::vector<AggregateInfo> aggregateInfos = toAggregateInfo(
      *aggregationNode_, *operatorCtx_, numHashers, expressionEvaluator);
  for (int i = 0; i < aggregateInfos.size(); i++) {
    if (aggregationNode_->aggregates()[i].call->name().find("vid_split") !=
        std::string::npos) {
      containsVidSplit_ = true;
      adaptiveAdjustment_ = false;
      break;
    }
  }

  bool canUseRowBasedSpill = true;
  bool hasDistinctAgg = false;
  size_t accumulatorSize = 0;
  int alignment = 1;
  // Check that aggregate result type match the output type.
  for (auto i = 0; i < aggregateInfos.size(); i++) {
    accumulatorSize += aggregateInfos[i].function->accumulatorFixedWidthSize();
    const auto& aggResultType = aggregateInfos[i].function->resultType();
    const auto& expectedType = outputType_->childAt(numHashers + i);
    BOLT_CHECK(
        aggResultType->kindEquals(expectedType),
        "Unexpected result type for an aggregation: {}, expected {}, step {}",
        aggResultType->toString(),
        expectedType->toString(),
        core::AggregationNode::stepName(aggregationNode_->step()));
    // only can used row based spill for fix accumulate and no sorting keys
    canUseRowBasedSpill &=
        (aggregateInfos[i].function->isFixedSize() ||
         aggregateInfos[i].function->supportAccumulatorSerde()) &&
        aggregateInfos[i].sortingKeys.empty();
    alignment = RowContainer::combineAlignments(
        aggregateInfos[i].function->accumulatorAlignmentSize(), alignment);
    hasDistinctAgg |= aggregateInfos[i].distinct;
  }
  const auto& queryConf = operatorCtx_->driverCtx()->queryConfig();
  if (canUseRowBasedSpill && spillConfig_) {
    auto rowbasedSpillMode = queryConf.rowBasedSpillMode();
    LOG(INFO) << "Operator " << name()