Method GetNextMerging

be/src/exec/exchange-node.cc:244–276

}

Status ExchangeNode::GetNextMerging(RuntimeState* state, RowBatch* output_batch,
    bool* eos) {
  DCHECK_EQ(output_batch->num_rows(), 0);
  RETURN_IF_CANCELLED(state);
  RETURN_IF_ERROR(QueryMaintenance(state));
  // Clear any expr result allocations made by the merger.
  expr_results_pool_->Clear();
  RETURN_IF_ERROR(stream_recvr_->GetNext(output_batch, eos));

  while (num_rows_skipped_ < offset_) {
    num_rows_skipped_ += output_batch->num_rows();
    // Throw away rows in the output batch until the offset is skipped.
    int64_t rows_to_keep = num_rows_skipped_ - offset_;
    if (rows_to_keep > 0) {
      output_batch->CopyRows(0, output_batch->num_rows() - rows_to_keep, rows_to_keep);
      output_batch->set_num_rows(rows_to_keep);
    } else {
      output_batch->set_num_rows(0);
    }
    if (rows_to_keep > 0 || *eos || output_batch->AtCapacity()) break;
    RETURN_IF_ERROR(stream_recvr_->GetNext(output_batch, eos));
  }

  CheckLimitAndTruncateRowBatchIfNeeded(output_batch, eos);

  // On eos, transfer all remaining resources from the input batches maintained
  // by the merger to the output batch. Also cancel the underlying receiver so
  // the senders' fragments can exit early.
  if (*eos) ReleaseRecvrResources(output_batch);

  COUNTER_SET(rows_returned_counter_, rows_returned());
  return Status::OK();
}

void ExchangeNode::DebugString(int indentation_level, stringstream* out) const {
  *out << string(indentation_level * 2, ' ');