Method noMoreInput

bolt/exec/SortBuffer.cpp:212–295

}

void SortBuffer::noMoreInput() {
  bolt::common::testutil::TestValue::adjust(
      "bytedance::bolt::exec::SortBuffer::noMoreInput", this);
  BOLT_CHECK(!noMoreInput_);
  noMoreInput_ = true;

  // No data.
  if (numInputRows_ == 0) {
    return;
  }
  if (hybridSortEnabled_ && hybridData_ != nullptr && !scatteredMode_) {
    hybridData_->coalesceBatches();
  }

  if (spiller_ == nullptr) {
    BOLT_CHECK_EQ(numInputRows_, data_->numRows());
    updateEstimatedOutputRowSize();
    // Sort the pointers to the rows in RowContainer (data_) instead of sorting
    // the rows.
    // TODO: Reuse 'RowContainer::rowPointers_'.
    sortedRows_.resize(numInputRows_);
    RowContainerIterator iter;
    data_->listRows(&iter, numInputRows_, sortedRows_.data());

    MicrosecondTimer timer(&sortInSortTimeUs_);

#ifdef ENABLE_BOLT_JIT
    if (cmp_ == nullptr && operatorCtx_ &&
        operatorCtx_->driverCtx()->queryConfig().enableJitRowCmpRow()) {
      if (data_->JITable(data_->keyTypes())) {
        auto [jitMod, rowRowCmpfn] = data_->codegenCompare(
            data_->keyTypes(),
            sortCompareFlags_,
            bytedance::bolt::jit::CmpType::SORT_LESS,
            true);
        jitModule_ = std::move(jitMod);
        cmp_ = (RowRowCompare)jitModule_->getFuncPtr(rowRowCmpfn);
      }
    }
    if (cmp_) {
      sorter_.sort(sortedRows_.begin(), sortedRows_.end(), cmp_);
    } else {
#endif

#ifdef ENABLE_META_SORT
      MetaRowsSorterWraper<BufferRows>::MetaCodegenSort(
          sortedRows_,
          data_.get(),
          sorter_,
          data_->keyIndices(),
          sortCompareFlags_);
#else
    sorter_.sort(
        sortedRows_.begin(),
        sortedRows_.end(),
        [this](const char* leftRow, const char* rightRow) {
          for (vector_size_t index = 0; index < sortCompareFlags_.size();
               ++index) {