Method processSpillInput

bolt/exec/HashBuild.cpp:1473–1507

}

void HashBuild::processSpillInput() {
  checkRunning();
  {
    DeltaCpuWallTimer timer{[this](const CpuWallTiming& timing) {
      auto selfDelta = operatorCtx_->driver()->processLazyTiming(*this, timing);
      this->stats().wlock()->addInputTiming.add(selfDelta);
    }};

    if (spillConfig_->rowBasedSpillMode == common::RowBasedSpillMode::DISABLE) {
      while (spillInputReader_->nextBatch(input_)) {
        addInput(std::move(input_));
        if (!isRunning()) {
          return;
        }
      }
    } else { // read row based spill input
      std::vector<char*> rows;
      while (auto size = spillInputReader_->nextBatch(rows)) {
        if (size == 0) {
          break;
        }
        addSpilledRowInput(rows, size);
        if (!isRunning()) {
          return;
        }
      }
    }
  }
  recordSpillReadStats();
  DeltaCpuWallTimer timer{[this](const CpuWallTiming& timing) {
    this->stats().wlock()->finishTiming.add(timing);
  }};
  noMoreInputInternal();
  restoringRangePartition_ = false;
}

void HashBuild::addRuntimeStats() {
  // Report range sizes and number of distinct values for the join keys.