Method GetNext

be/src/exec/exchange-node.cc:186–242

}

Status ExchangeNode::GetNext(RuntimeState* state, RowBatch* output_batch, bool* eos) {
  SCOPED_TIMER(runtime_profile_->total_time_counter());
  ScopedGetNextEventAdder ea(this, eos);
  RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT, state));
  if (ReachedLimit()) {
    ReleaseRecvrResources(output_batch);
    *eos = true;
    return Status::OK();
  } else {
    *eos = false;
  }

  if (is_merging_) return GetNextMerging(state, output_batch, eos);

  while (true) {
    {
      SCOPED_TIMER(convert_row_batch_timer_);
      RETURN_IF_CANCELLED(state);
      RETURN_IF_ERROR(QueryMaintenance(state));
      if (input_batch_ != nullptr) {
        // copy rows until we hit the limit/capacity or until we exhaust input_batch_
        int available_in_input = input_batch_->num_rows() - next_row_idx_;
        int free_in_output = output_batch->capacity() - output_batch->num_rows();
        int rows_to_copy = std::min(available_in_input, free_in_output);
        if (limit_ != -1) {
          rows_to_copy =
              static_cast<int>(std::min<int64_t>(rows_to_copy, limit_ - rows_returned()));
        }
        if (rows_to_copy > 0) {
          int dst_offset = output_batch->AddRows(rows_to_copy);
          output_batch->CopyRows(input_batch_, rows_to_copy, next_row_idx_, dst_offset);
          next_row_idx_ += rows_to_copy;
          output_batch->CommitRows(rows_to_copy);
          IncrementNumRowsReturned(rows_to_copy);
          COUNTER_SET(rows_returned_counter_, rows_returned());
        }
      }

      if (ReachedLimit()) {
        ReleaseRecvrResources(output_batch);
        *eos = true;
        return Status::OK();
      }
      if (output_batch->AtCapacity()) return Status::OK();
    }

    // we need more rows
    stream_recvr_->TransferAllResources(output_batch);
    RETURN_IF_ERROR(FillInputRowBatch(state));
    *eos = (input_batch_ == nullptr);
    // No need to call CancelStream() on the receiver here as all incoming row batches
    // have been consumed so we should have replied to all senders already.
    if (*eos) return Status::OK();
    next_row_idx_ = 0;
    DCHECK(input_batch_->row_desc()->LayoutIsPrefixOf(*output_batch->row_desc()));
  }
}