Method GetNext

be/src/exec/partial-sort-node.cc:116–173

}

Status PartialSortNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
  SCOPED_TIMER(runtime_profile_->total_time_counter());
  ScopedGetNextEventAdder ea(this, eos);
  RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT, state));
  RETURN_IF_CANCELLED(state);
  RETURN_IF_ERROR(QueryMaintenance(state));

  DCHECK_EQ(row_batch->num_rows(), 0);
  if (!sorter_eos_) {
    // There were rows in the current run that didn't fit in the last output batch.
    RETURN_IF_ERROR(sorter_->GetNext(row_batch, &sorter_eos_));
    if (sorter_eos_) {
      sorter_->Reset();
      *eos = input_eos_;
    }
    IncrementNumRowsReturned(row_batch->num_rows());
    COUNTER_SET(rows_returned_counter_, rows_returned());
    return Status::OK();
  }

  if (input_eos_) {
    *eos = true;
    return Status::OK();
  }

  DCHECK(sorter_eos_);
  RETURN_IF_ERROR(sorter_->Open());
  do {
    if (input_batch_index_ == input_batch_->num_rows()) {
      input_batch_->Reset();
      input_batch_index_ = 0;
      MonotonicStopWatch timer;
      timer.Start();
      Status status = child(0)->GetNext(state, input_batch_.get(), &input_eos_);
      timer.Stop();
      RETURN_IF_ERROR(status);
      child_get_next_timer_->UpdateCounter(timer.ElapsedTime());
    }

    int num_processed;
    RETURN_IF_ERROR(
        sorter_->AddBatchNoSpill(input_batch_.get(), input_batch_index_, &num_processed));
    input_batch_index_ += num_processed;
    DCHECK(input_batch_index_ <= input_batch_->num_rows());
    RETURN_IF_ERROR(QueryMaintenance(state));
  } while (input_batch_index_ == input_batch_->num_rows() && !input_eos_);

  RETURN_IF_ERROR(sorter_->InputDone());
  RETURN_IF_ERROR(sorter_->GetNext(row_batch, &sorter_eos_));
  if (sorter_eos_) {
    sorter_->Reset();
    *eos = input_eos_;
  }

  IncrementNumRowsReturned(row_batch->num_rows());
  COUNTER_SET(rows_returned_counter_, rows_returned());
  return Status::OK();
}