Method getOutput

bolt/exec/HashAggregation.cpp:544–650

}

RowVectorPtr HashAggregation::getOutput() {
  if (BOLT_TEST_VALUE_ENABLED()) {
    bool injectSegfault = false;
    BOLT_TEST_ADJUST(
        "bytedance::bolt::exec::HashAggregation::getOutput", &injectSegfault);
    if (injectSegfault) {
      triggerSegfault();
    }
  }
  if (finished_) {
    input_ = nullptr;
    return nullptr;
  }
  if (abandonedPartialAggregation_) {
    if (noMoreInput_) {
      finished_ = true;
    }
    if (!input_) {
      return nullptr;
    }
    prepareOutput(input_->size(), false);
    groupingSet_->toIntermediate(input_, output_);
    numOutputRows_ += input_->size();
    input_ = nullptr;
    return output_;
  }

  // Produce results if one of the following is true:
  // - received no-more-input message;
  // - partial aggregation reached memory limit;
  // - distinct aggregation has new keys;
  // - running in partial streaming mode and have some output ready.
  if (!noMoreInput_ && !partialFull_ && !newDistincts_ &&
      !groupingSet_->hasOutput()) {
    input_ = nullptr;
    return nullptr;
  }

  if (isDistinct_) {
    auto currentOutput = getDistinctOutput();
    // Accumulate the generated output using append()
    if (currentOutput) {
      if (!accumulatedOutput_) {
        if (currentOutput->size() >= minOutputRows_) {
          return currentOutput;
        }
        accumulatedOutput_ =
            RowVector::createEmpty(currentOutput->type(), operatorCtx_->pool());
      }
      accumulatedOutput_->append(currentOutput.get());
    }

    // Handle accumulated output when no more input or memory pressure
    if (accumulatedOutput_ &&
        (finished_ || partialFull_ ||
         accumulatedOutput_->size() >= minOutputRows_ ||
         abandonedPartialAggregation_)) {
      auto result = std::move(accumulatedOutput_);