Method getDistinctOutput

bolt/exec/HashAggregation.cpp:652–720

}

RowVectorPtr HashAggregation::getDistinctOutput() {
  BOLT_CHECK(isDistinct_);
  BOLT_CHECK(!finished_);

  if (newDistincts_) {
    BOLT_CHECK_NOT_NULL(input_);

    const auto& distinctNewGroups =
        groupingSet_->hashLookup().distinctNewGroups;

    // Accumulate results from all saved batches
    const auto size = distinctNewGroups.size();
    BufferPtr indices = allocateIndices(size, operatorCtx_->pool());
    auto indicesPtr = indices->asMutable<vector_size_t>();

    // Copy all batches into indices buffer
    std::copy(distinctNewGroups.begin(), distinctNewGroups.end(), indicesPtr);

    newDistincts_ = false;
    auto output = fillOutput(size, indices);
    numOutputRows_ += size;

    recordRuntimeMetrics();

    // Drop reference to input_ to make it singly-referenced at the producer and
    // allow for memory reuse.
    input_ = nullptr;

    // Clear tracking after processing to prevent memory retention
    groupingSet_->resetDistinctNewGroups();
    resetPartialOutputIfNeed();
    return output;
  }
  BOLT_CHECK(!newDistincts_);

  auto outputGuard = folly::makeGuard([&]() { recordRuntimeMetrics(); });
  if (!groupingSet_->hasSpilled()) {
    if (noMoreInput_) {
      finished_ = true;
      if (auto numRows = groupingSet_->numDefaultGlobalGroupingSetRows()) {
        prepareOutput(numRows.value(), false);
        if (groupingSet_->getDefaultGlobalGroupingSetOutput(
                resultIterator_, output_)) {
          numOutputRows_ += output_->size();
          return output_;
        }
      }
    }
    return nullptr;
  }

  if (!noMoreInput_) {
    return nullptr;
  }

  const auto& queryConfig = operatorCtx_->driverCtx()->queryConfig();
  const auto maxOutputRows = outputBatchRows(estimatedOutputRowSize_);
  prepareOutput(maxOutputRows, false);