| 85 | const char* KrpcDataStreamSender::TOTAL_BYTES_SENT_COUNTER = "TotalBytesSent"; |
| 86 | |
| 87 | Status KrpcDataStreamSenderConfig::Init( |
| 88 | const TDataSink& tsink, const RowDescriptor* input_row_desc, FragmentState* state) { |
| 89 | RETURN_IF_ERROR(DataSinkConfig::Init(tsink, input_row_desc, state)); |
| 90 | DCHECK(tsink_->__isset.stream_sink); |
| 91 | partition_type_ = tsink_->stream_sink.output_partition.type; |
| 92 | if (partition_type_ == TPartitionType::HASH_PARTITIONED |
| 93 | || partition_type_ == TPartitionType::KUDU) { |
| 94 | RETURN_IF_ERROR( |
| 95 | ScalarExpr::Create(tsink_->stream_sink.output_partition.partition_exprs, |
| 96 | *input_row_desc_, state, &partition_exprs_)); |
| 97 | exchange_hash_seed_ = |
| 98 | KrpcDataStreamSender::EXCHANGE_HASH_SEED_CONST ^ state->query_id().hi; |
| 99 | } |
| 100 | num_channels_ = state->fragment_ctx().destinations().size(); |
| 101 | state->CheckAndAddCodegenDisabledMessage(codegen_status_msgs_); |
| 102 | return Status::OK(); |
| 103 | } |
| 104 | |
| 105 | DataSink* KrpcDataStreamSenderConfig::CreateSink(RuntimeState* state) const { |
| 106 | // We have one fragment per sink, so we can use the fragment index as the sink ID. |