Method addInput

bolt/exec/HashAggregation.cpp:267–353

}

void HashAggregation::addInput(RowVectorPtr input) {
  if (!pushdownChecked_) {
    mayPushdown_ = operatorCtx_->driver()->mayPushdownAggregation(this);
    pushdownChecked_ = true;
  }
  if (abandonedPartialAggregation_) {
    input_ = input;
    numInputRows_ += input->size();
    addRuntimeStat("abandonedVectors", RuntimeCounter(1));
    addRuntimeStat("abandonedRows", RuntimeCounter(input_->size()));
    return;
  }
  if (!isPartialStep_ && acceptCompositeVectorInput_) {
    CompositeRowVectorPtr compositeInput =
        std::dynamic_pointer_cast<CompositeRowVector>(input);
    if (compositeInput) {
      if (convertedInput_) {
        VectorPtr converted = std::move(convertedInput_);
        BaseVector::prepareForReuse(converted, input->size());
        convertedInput_ = std::static_pointer_cast<RowVector>(converted);
      } else {
        convertedInput_ =
            BaseVector::create<RowVector>(input->type(), input->size(), pool());
      }
      groupingSet_->convertCompositeInput(
          aggregatesForExtractColumns_, compositeInput, convertedInput_);
      input = convertedInput_;
    }
  }
  // Reset tracking between batches to prevent memory buildup
  groupingSet_->resetDistinctNewGroups();
  if (input->size() > maxInputBatchCount_) {
    if (isLazyNotLoaded(*input)) {
      input->loadedVector();
    }
    for (int64_t i = 0; i < input->size(); i += maxInputBatchCount_) {
      auto length = std::min(input->size() - i, maxInputBatchCount_);
      auto inputSlice =
          std::dynamic_pointer_cast<RowVector>(input->slice(i, length));
      groupingSet_->addInput(inputSlice, mayPushdown_);

      if (isDistinct_) {
        // Only adjust indices for distinct case needing cross-batch tracking
        groupingSet_->putDistinctNewGroupsIndices(i);
      }
    }
  } else {
    groupingSet_->addInput(input, mayPushdown_);
    if (isDistinct_) {
      // Only adjust indices for distinct case needing cross-batch tracking
      groupingSet_->putDistinctNewGroupsIndices(0);
    }
  }
  numInputRows_ += input->size();

  updateRuntimeStats();

  // NOTE: we should not trigger partial output flush in case of global