| 90 | } |
| 91 | |
| 92 | Status ExchangeNode::Prepare(RuntimeState* state) { |
| 93 | RETURN_IF_ERROR(ExecNode::Prepare(state)); |
| 94 | convert_row_batch_timer_ = ADD_TIMER(runtime_profile(), "ConvertRowBatchTime"); |
| 95 | |
| 96 | #ifndef NDEBUG |
| 97 | if (FLAGS_stress_datastream_recvr_delay_ms > 0) { |
| 98 | SleepForMs(FLAGS_stress_datastream_recvr_delay_ms); |
| 99 | } |
| 100 | #endif |
| 101 | |
| 102 | RETURN_IF_ERROR(ExecEnv::GetInstance()->buffer_pool()->RegisterClient( |
| 103 | Substitute("Exchg Recvr (id=$0)", id_), nullptr, |
| 104 | ExecEnv::GetInstance()->buffer_reservation(), mem_tracker(), |
| 105 | numeric_limits<int64_t>::max(), runtime_profile(), &recvr_buffer_pool_client_, |
| 106 | MemLimit::HARD)); |
| 107 | |
| 108 | // TODO: figure out appropriate buffer size |
| 109 | DCHECK_GT(num_senders_, 0); |
| 110 | stream_recvr_ = ExecEnv::GetInstance()->stream_mgr()->CreateRecvr( |
| 111 | &input_row_desc_, *state, state->fragment_instance_id(), id_, num_senders_, |
| 112 | FLAGS_exchg_node_buffer_size_bytes, is_merging_, runtime_profile(), mem_tracker(), |
| 113 | &recvr_buffer_pool_client_); |
| 114 | |
| 115 | return Status::OK(); |
| 116 | } |
| 117 | |
| 118 | void ExchangePlanNode::Codegen(FragmentState* state) { |
| 119 | DCHECK(state->ShouldCodegen()); |
nothing calls this directly
no test coverage detected