Method getOutputWithSpill

bolt/exec/SortBuffer.cpp:598–685

}

void SortBuffer::getOutputWithSpill() {
  BOLT_DCHECK_EQ(sortedRows_.size(), 0);
  if (spillMerger_) {
    BOLT_CHECK_NOT_NULL(spillMerger_);

    int32_t outputRow = 0;
    int32_t outputSize = 0;
    bool isEndOfBatch = false;
    while (outputRow + outputSize < output_->size()) {
      SpillMergeStream* stream = spillMerger_->next();
      BOLT_CHECK_NOT_NULL(stream);

      spillSources_[outputSize] = &stream->current();
      spillSourceRows_[outputSize] = stream->currentIndex(&isEndOfBatch);
      ++outputSize;
      if (FOLLY_UNLIKELY(isEndOfBatch)) {
        // The stream is at end of input batch. Need to copy out the rows before
        // fetching next batch in 'pop'.
        gatherCopy(
            output_.get(),
            outputRow,
            outputSize,
            spillSources_,
            spillSourceRows_,
            columnMap_);
        outputRow += outputSize;
        outputSize = 0;
      }
      // Advance the stream.
      stream->pop();
    }
    BOLT_CHECK_EQ(outputRow + outputSize, output_->size());

    if (FOLLY_LIKELY(outputSize != 0)) {
      gatherCopy(
          output_.get(),
          outputRow,
          outputSize,
          spillSources_,
          spillSourceRows_,
          columnMap_);
    }

    numOutputRows_ += output_->size();
  } else {
    BOLT_CHECK_NOT_NULL(rowBasedSpillMerger_);

    int32_t outputRow = 0;
    bool isEndOfBatch = false;
    std::vector<char*> rows;
    while (outputRow + rows.size() < output_->size()) {
      RowBasedSpillMergeStream* stream = rowBasedSpillMerger_->next();
      BOLT_CHECK_NOT_NULL(stream);

      const auto& currentBatch = stream->current();
      auto index = stream->currentIndex(&isEndOfBatch);
      rows.push_back(currentBatch[index]);
      if (FOLLY_UNLIKELY(isEndOfBatch)) {