Method SortInput

be/src/exec/sort-node.cc:223–247

}

Status SortNode::SortInput(RuntimeState* state) {
  RowBatch batch(child(0)->row_desc(), state->batch_size(), mem_tracker());
  bool eos;
  do {
    RETURN_IF_ERROR(child(0)->GetNext(state, &batch, &eos));

    MonotonicStopWatch timer;
    timer.Start();
    Status add_status = sorter_->AddBatch(&batch);
    timer.Stop();
    add_batch_timer_->UpdateCounter(timer.ElapsedTime());
    if (UNLIKELY(!add_status.ok())) return add_status;

    batch.Reset();
    RETURN_IF_CANCELLED(state);
    RETURN_IF_ERROR(QueryMaintenance(state));
  } while(!eos);

  // Unless we are inside a subplan expecting to call Open()/GetNext() on the child
  // again, the child can be closed at this point to release resources.
  if (!IsInSubplan()) child(0)->Close(state);

  RETURN_IF_ERROR(sorter_->InputDone());
  return Status::OK();
}

}