| 74 | } |
| 75 | |
| 76 | void ScheduleState::Init() { |
| 77 | *query_schedule_pb_->mutable_query_id() = query_id_; |
| 78 | // extract TPlanFragments and order by fragment idx |
| 79 | for (const TPlanExecInfo& plan_exec_info: request_.plan_exec_info) { |
| 80 | for (const TPlanFragment& fragment: plan_exec_info.fragments) { |
| 81 | fragments_.emplace(fragment.idx, fragment); |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | // this must only be called once |
| 86 | DCHECK_EQ(fragment_schedule_states_.size(), 0); |
| 87 | for (int i = 0; i < fragments_.size(); ++i) { |
| 88 | auto it = fragments_.find(i); |
| 89 | DCHECK(it != fragments_.end()); |
| 90 | fragment_schedule_states_.emplace_back( |
| 91 | it->second, query_schedule_pb_->add_fragment_exec_params()); |
| 92 | } |
| 93 | |
| 94 | // mark root coordinator fragment |
| 95 | const TPlanFragment& root_fragment = request_.plan_exec_info[0].fragments[0]; |
| 96 | if (RequiresCoordinatorFragment()) { |
| 97 | fragment_schedule_states_[root_fragment.idx].is_root_coord_fragment = true; |
| 98 | // the coordinator instance gets index 0, generated instance ids start at 1 |
| 99 | next_instance_id_ = CreateInstanceId(next_instance_id_, 1); |
| 100 | } |
| 101 | |
| 102 | // find max node id |
| 103 | int max_node_id = 0; |
| 104 | for (const TPlanExecInfo& plan_exec_info: request_.plan_exec_info) { |
| 105 | for (const TPlanFragment& fragment: plan_exec_info.fragments) { |
| 106 | for (const TPlanNode& node: fragment.plan.nodes) { |
| 107 | max_node_id = max(node.node_id, max_node_id); |
| 108 | } |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | // populate plan_node_to_fragment_idx_ and plan_node_to_plan_node_idx_ |
| 113 | plan_node_to_fragment_idx_.resize(max_node_id + 1); |
| 114 | plan_node_to_plan_node_idx_.resize(max_node_id + 1); |
| 115 | for (const TPlanExecInfo& plan_exec_info: request_.plan_exec_info) { |
| 116 | for (const TPlanFragment& fragment: plan_exec_info.fragments) { |
| 117 | for (int i = 0; i < fragment.plan.nodes.size(); ++i) { |
| 118 | const TPlanNode& node = fragment.plan.nodes[i]; |
| 119 | plan_node_to_fragment_idx_[node.node_id] = fragment.idx; |
| 120 | plan_node_to_plan_node_idx_[node.node_id] = i; |
| 121 | } |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | // compute input fragments |
| 126 | for (const TPlanExecInfo& plan_exec_info: request_.plan_exec_info) { |
| 127 | // each fragment sends its output to the fragment containing the destination node |
| 128 | // of its output sink |
| 129 | for (const TPlanFragment& fragment: plan_exec_info.fragments) { |
| 130 | if (!fragment.output_sink.__isset.stream_sink) continue; |
| 131 | PlanNodeId dest_node_id = fragment.output_sink.stream_sink.dest_node_id; |
| 132 | FragmentIdx dest_idx = plan_node_to_fragment_idx_[dest_node_id]; |
| 133 | FragmentScheduleState& dest_state = fragment_schedule_states_[dest_idx]; |