| 849 | } |
| 850 | |
| 851 | Status KrpcDataStreamSender::Prepare( |
| 852 | RuntimeState* state, MemTracker* parent_mem_tracker) { |
| 853 | RETURN_IF_ERROR(DataSink::Prepare(state, parent_mem_tracker)); |
| 854 | state_ = state; |
| 855 | SCOPED_TIMER(profile_->total_time_counter()); |
| 856 | RETURN_IF_ERROR(ScalarExprEvaluator::Create(partition_exprs_, state, state->obj_pool(), |
| 857 | expr_perm_pool_.get(), expr_results_pool_.get(), &partition_expr_evals_)); |
| 858 | serialize_batch_timer_ = ADD_TIMER(profile(), "SerializeBatchTime"); |
| 859 | transmit_data_timer_ = ADD_TIMER(profile(), "TransmitDataTime"); |
| 860 | rpc_retry_counter_ = ADD_COUNTER(profile(), "RpcRetry", TUnit::UNIT); |
| 861 | rpc_failure_counter_ = ADD_COUNTER(profile(), "RpcFailure", TUnit::UNIT); |
| 862 | rpc_success_counter_ = ADD_COUNTER(profile(), "RpcSuccess", TUnit::UNIT); |
| 863 | bytes_sent_counter_ = ADD_COUNTER(profile(), "TotalBytesSent", TUnit::BYTES); |
| 864 | state->AddBytesSentCounter(bytes_sent_counter_); |
| 865 | bytes_sent_time_series_counter_ = |
| 866 | ADD_TIME_SERIES_COUNTER(profile(), "BytesSent", bytes_sent_counter_); |
| 867 | network_throughput_counter_ = |
| 868 | ADD_SUMMARY_STATS_COUNTER(profile(), "NetworkThroughput", TUnit::BYTES_PER_SECOND); |
| 869 | network_time_stats_ = |
| 870 | ADD_SUMMARY_STATS_COUNTER(profile(), "RpcNetworkTime", TUnit::TIME_NS); |
| 871 | recvr_time_stats_ = |
| 872 | ADD_SUMMARY_STATS_COUNTER(profile(), "RpcRecvrTime", TUnit::TIME_NS); |
| 873 | eos_sent_counter_ = ADD_COUNTER(profile(), "EosSent", TUnit::UNIT); |
| 874 | uncompressed_bytes_counter_ = |
| 875 | ADD_COUNTER(profile(), "UncompressedRowBatchSize", TUnit::BYTES); |
| 876 | total_sent_rows_counter_ = ADD_COUNTER(profile(), "RowsSent", TUnit::UNIT); |
| 877 | |
| 878 | outbound_rb_mem_tracker_.reset( |
| 879 | new MemTracker(-1, "RowBatchSerialization", mem_tracker_.get())); |
| 880 | char_mem_tracker_allocator_.reset( |
| 881 | new CharMemTrackerAllocator(outbound_rb_mem_tracker_)); |
| 882 | |
| 883 | string process_address = |
| 884 | NetworkAddressPBToString(ExecEnv::GetInstance()->krpc_address()); |
| 885 | |
| 886 | serialization_batch_.reset(new OutboundRowBatch(*char_mem_tracker_allocator_)); |
| 887 | if (partition_type_ == TPartitionType::UNPARTITIONED) { |
| 888 | in_flight_batch_.reset(new OutboundRowBatch(*char_mem_tracker_allocator_)); |
| 889 | } |
| 890 | |
| 891 | compression_scratch_.reset(new TrackedString(*char_mem_tracker_allocator_)); |
| 892 | |
| 893 | for (int i = 0; i < channels_.size(); ++i) { |
| 894 | RETURN_IF_ERROR(channels_[i]->Init(state, char_mem_tracker_allocator_)); |
| 895 | } |
| 896 | for (PartitionRowCollector& collector: partition_row_collectors_) { |
| 897 | collector.collector_batch_.reset(new OutboundRowBatch(*char_mem_tracker_allocator_)); |
| 898 | collector.row_batch_capacity_ = collector.channel_->RowBatchCapacity(); |
| 899 | } |
| 900 | for (auto& [ch, ice_ch] : channel_to_ice_channel_) { |
| 901 | ice_ch->Prepare(mem_tracker_.get()); |
| 902 | } |
| 903 | return Status::OK(); |
| 904 | } |
| 905 | |
| 906 | Status KrpcDataStreamSender::Open(RuntimeState* state) { |
| 907 | SCOPED_TIMER(profile_->total_time_counter()); |
nothing calls this directly
no test coverage detected