Method Sender

be/src/runtime/data-stream-test.cc:656–705

  }

  void Sender(int sender_num, int channel_buffer_size,
      TPartitionType::type partition_type, SenderInfo* info, bool reset_hash_seed,
      int num_batches) {
    RuntimeState state(TQueryCtx(), exec_env_.get(), desc_tbl_);
    VLOG_QUERY << "create sender " << sender_num;
    const TDataSink sink = GetSink(partition_type);
    TPlanFragment fragment;
    fragment.output_sink = sink;
    PlanFragmentCtxPB fragment_ctx;
    *fragment_ctx.mutable_destinations() = dest_;
    FragmentState fragment_state(state.query_state(), fragment, fragment_ctx);
    DataSinkConfig* data_sink = nullptr;
    EXPECT_OK(DataSinkConfig::CreateConfig(sink, row_desc_, &fragment_state, &data_sink));

    // We create an object of the base class DataSink and cast to the appropriate sender
    // according to the 'is_thrift' option.
    scoped_ptr<DataSink> sender;

    KrpcDataStreamSenderConfig& config =
        *(static_cast<KrpcDataStreamSenderConfig*>(data_sink));

    // Reset the hash seed with a new query id. Useful for testing hash exchanges are
    // random.
    if (reset_hash_seed) {
      config.exchange_hash_seed_ =
          GetExchangeHashSeed(UuidToQueryId(random_generator()()));
    }

    sender.reset(new KrpcDataStreamSender(-1, sender_num, config,
        data_sink->tsink_->stream_sink, dest_, channel_buffer_size, &state));
    EXPECT_OK(sender->Prepare(&state, &tracker_));
    EXPECT_OK(sender->Open(&state));
    scoped_ptr<RowBatch> batch(CreateRowBatch());
    int next_val = 0;
    for (int i = 0; i < num_batches; ++i) {
      GetNextBatch(batch.get(), &next_val);
      VLOG_QUERY << "sender " << sender_num << ": #rows=" << batch->num_rows();
      info->status = sender->Send(&state, batch.get());
      if (!info->status.ok()) break;
    }
    VLOG_QUERY << "closing sender" << sender_num;
    info->status.MergeStatus(sender->FlushFinal(&state));
    sender->Close(&state);
    info->num_bytes_sent = static_cast<KrpcDataStreamSender*>(
        sender.get())->GetNumDataBytesSent();
    data_sink->Close();

    batch->Reset();
    state.ReleaseResources();
  }

  void TestStream(TPartitionType::type stream_type, int num_senders, int num_receivers,
      int buffer_size, bool is_merging) {