Method addSpillInput

bolt/exec/HashProbe.cpp:528–567

}

void HashProbe::addSpillInput() {
  checkRunning();

  if (input_ != nullptr || noMoreSpillInput_) {
    return;
  }
  if (FOLLY_UNLIKELY(!spillInputReader_->nextBatch(input_))) {
    recordSpillReadStats();
    if (probeRangePartition_ && needLastProbeSideOutput()) {
      spillMatchFlagReader_.reset();
      if (matchFlagSpiller_) {
        auto&& spilledFlags = matchFlagSpiller_->finishSpill();
        // for next range partition
        spillMatchFlagReader_ = spilledFlags.createUnorderedReader(pool());
        matchFlagSpiller_.reset();
      }
    }
    DeltaCpuWallTimer timer{[this](const CpuWallTiming& timing) {
      this->stats().wlock()->finishTiming.add(timing);
    }};
    noMoreInputInternal();
    return;
  }

  if (probeRangePartition_ && needLastProbeSideOutput()) {
    if (spillMatchFlagReader_) {
      spillMatchFlagReader_->nextBatch(accumulatedMatchFlag_);
      // probe flags's bits should equals input rows's size
      BOLT_CHECK(accumulatedMatchFlag_->size() == input_->size());
    } else {
      // probe first range partiion
      prepareMatchFlag(input_->size(), accumulatedMatchFlag_);
    }
  }
  DeltaCpuWallTimer timer{[this](const CpuWallTiming& timing) {
    auto selfDelta = operatorCtx_->driver()->processLazyTiming(*this, timing);
    this->stats().wlock()->addInputTiming.add(selfDelta);
  }};
  addInput(std::move(input_));
}

void HashProbe::spillInput(RowVectorPtr& input) {
  BOLT_CHECK(needSpillInput());