| 133 | } |
| 134 | |
| 135 | Status ExchangeNode::Open(RuntimeState* state) { |
| 136 | SCOPED_TIMER(runtime_profile_->total_time_counter()); |
| 137 | ScopedOpenEventAdder ea(this); |
| 138 | RETURN_IF_ERROR(ExecNode::Open(state)); |
| 139 | RETURN_IF_CANCELLED(state); |
| 140 | if (is_merging_) { |
| 141 | const ExchangePlanNode& pnode = static_cast<const ExchangePlanNode&>(plan_node_); |
| 142 | // CreateMerger() will populate its merging heap with batches from the stream_recvr_, |
| 143 | // so it is not necessary to call FillInputRowBatch(). |
| 144 | RETURN_IF_ERROR( |
| 145 | less_than_->Open(pool_, state, expr_perm_pool(), expr_results_pool())); |
| 146 | RETURN_IF_ERROR(stream_recvr_->CreateMerger(*less_than_.get(), |
| 147 | pnode.codegend_heapify_helper_fn_)); |
| 148 | } else { |
| 149 | RETURN_IF_ERROR(FillInputRowBatch(state)); |
| 150 | } |
| 151 | return Status::OK(); |
| 152 | } |
| 153 | |
| 154 | Status ExchangeNode::Reset(RuntimeState* state, RowBatch* row_batch) { |
| 155 | DCHECK(false) << "NYI"; |
nothing calls this directly
no test coverage detected