Method getOutputWithSpill

bolt/exec/GroupingSet.cpp:1179–1243

  BOLT_CHECK(!hasSpilled());

  if (table_ == nullptr) {
    return;
  }

  auto* spillConf = const_cast<common::SpillConfig*>(spillConfig_);
  operatorCtx_->adjustSpillCompressionKind(spillConf);
  spillConf->needSetNextEqual = bypassProbeHT_;
  auto* rows = table_->rows();
  BOLT_CHECK(pool_.trackUsage());
  spiller_ = std::make_unique<Spiller>(
      Spiller::Type::kAggregateOutput, rows, makeSpillType(), spillConfig_);
  spiller_->setSpillConfig(spillConfig_);
  spiller_->spill(rowIterator);
  adjustBypassHashTable(spillConf);
  table_->clear();
}

bool GroupingSet::getOutputWithSpill(
    int32_t maxOutputRows,
    int32_t maxOutputBytes,
    const RowVectorPtr& result) {
  if (merge_ == nullptr && rowBasedSpillMerge_ == nullptr) {
    LOG(INFO) << operatorCtx_->toString() << " prepare merge, files number: "
              << spiller_->state().numFinishedFiles(0)
              << " memory: " << pool_.currentBytes()
              << ", reserved: " << pool_.reservedBytes();
    BOLT_CHECK_NULL(mergeRows_);
    BOLT_CHECK(mergeArgs_.empty());

    if (!isDistinct()) {
      mergeArgs_.resize(1);
      std::vector<TypePtr> keyTypes;
      for (auto& hasher : table_->hashers()) {
        keyTypes.push_back(hasher->type());
      }

      mergeRows_ = std::make_unique<RowContainer>(
          keyTypes,
          !ignoreNullKeys_,
          accumulators(false),
          std::vector<TypePtr>(),
          false,
          false,
          true,
          false,
          false /*useListRowIndex*/,
          &pool_,
          table_->rows()->stringAllocatorShared());

      initializeAggregates(aggregates_, *mergeRows_, false);
    }

    BOLT_CHECK_EQ(table_->rows()->numRows(), 0);

    auto spillPartition = spiller_->finishSpill();
    if (spillPartition.numFiles() == 0) {
      BOLT_CHECK(spiller_->type() == Spiller::Type::kAggregateOutput);
      BOLT_CHECK(spiller_->filledZeroRows());