Method getOutput

bolt/exec/StreamingAggregation.cpp:298–356

}

RowVectorPtr StreamingAggregation::getOutput() {
  NanosecondTimer timer(&stats_.aggOutputTimeNs);
  if (!input_) {
    if (noMoreInput_ && numGroups_ > 0) {
      auto output = createOutput(numGroups_);
      numGroups_ = 0;
      return output;
    }
    return nullptr;
  }

  // hold on accept new input if groups_ is too large to avoid large memory
  // usage and too many copy
  if (numGroups_ < groupNumberThreshold_) {
    auto numInput = input_->size();
    inputRows_.resize(numInput);
    inputRows_.setAll();

    masks_->addInput(input_, inputRows_);

    auto numPrevGroups = numGroups_;
    {
      NanosecondTimer aggTimer(&stats_.aggOutputUpdateTimeNs);
      assignGroups();
      initializeNewGroups(numPrevGroups);
    }
    evaluateAggregates();

    prevInput_ = input_;
    input_ = nullptr;
  }

  RowVectorPtr output;
  uint32_t outputSize = 0;
  if (numGroups_ > outputBatchSize_) {
    outputSize = outputBatchSize_;
  } else if (
      numGroups_ > 1 &&
      rows_->estimateRowSize().value_or(0) * rows_->numRows() > outputBytes_) {
    outputSize = numGroups_ - 1;
  }

  if (outputSize > 0) {
    output = createOutput(outputSize);

    // Rotate the entries in the groups_ vector to move the remaining groups to
    // the beginning and place reusable groups at the end.
    std::vector<char*> copy(groups_.size());
    std::copy(groups_.begin() + outputSize, groups_.end(), copy.begin());
    std::copy(
        groups_.begin(),
        groups_.begin() + outputSize,
        copy.begin() + groups_.size() - outputSize);
    groups_ = std::move(copy);
    numGroups_ -= outputSize;
  }

  return output;