Method writeSpill

bolt/exec/Spiller.cpp:652–707

}

std::unique_ptr<Spiller::SpillStatus> Spiller::writeSpill(int32_t partition) {
  BOLT_CHECK(type_ != Type::kHashJoinProbe && type_ != Type::kLocalMergeInput);

  // 1. The flush threshold is controlled by writeBufferSize_ from configuration
  // 2. The materialized size is controlled by kMaxReadBufferSize
  constexpr int32_t kTargetBatchBytes = (1UL << 20) -
      AlignedBuffer::kPaddedSize; // 1M, Same as kMaxReadBufferSize
  constexpr int32_t kTargetBatchRows = 4096;

  RowVectorPtr spillVector;
  auto& run = spillRuns_[partition];
  try {
    ensureSorted(run);
    int64_t totalBytes = 0;
    size_t written = 0;
    if (spillMode_ == Mode::kRowVector) {
      while (written < run.rows.size()) {
        if (hybridSortEnabled_) {
          extractSpillVectorHybrid(
              run.rows,
              kTargetBatchRows,
              kTargetBatchBytes,
              spillVector,
              written);
        } else {
          extractSpillVector(
              run.rows,
              kTargetBatchRows,
              kTargetBatchBytes,
              spillVector,
              written);
        }
        totalBytes += state_.appendToPartition(partition, spillVector);
        spillVector->prepareForReuse();
        if (totalBytes > state_.targetFileSize()) {
          BOLT_CHECK(!needSort());
          state_.finishFile(partition);
          // reset to 0 after close file
          totalBytes = 0;
        }
      }
    } else {
      if (spillConfig_->needSetNextEqual) {
        spillConfig_->aggBypassHTEqualNum += setNextEqualForAgg(run);
      }
      written = run.rows.size();
      state_.appendToPartition(partition, run.rows, rowType_, rowInfo_.value());
    }
    return std::make_unique<SpillStatus>(partition, written, nullptr);
  } catch (const std::exception& e) {
    // The exception is passed to the caller thread which checks this in
    // advanceSpill().
    return std::make_unique<SpillStatus>(
        partition, 0, std::current_exception());
  }
}

void Spiller::runSpill(bool lastRun) {