| 145 | |
| 146 | template <typename OID_T, typename VID_T, typename VD_T, typename ED_T> |
| 147 | void LoadFragment(vineyard::Client& client, grape::CommSpec& comm_spec) { |
| 148 | auto pair = splitAndGet(comm_spec, FLAGS_raw_data_ids); |
| 149 | auto cur_raw_data_id = pair.first; |
| 150 | auto graphx_pid = pair.second; |
| 151 | std::vector<int> pid2Fid; |
| 152 | { |
| 153 | std::vector<int> fid2Pid; |
| 154 | vineyard::GlobalAllGatherv(graphx_pid, fid2Pid, comm_spec); |
| 155 | pid2Fid.resize(fid2Pid.size()); |
| 156 | for (size_t i = 0; i < fid2Pid.size(); ++i) { |
| 157 | pid2Fid[fid2Pid[i]] = i; |
| 158 | } |
| 159 | } |
| 160 | |
| 161 | LOG(INFO) << "Worker [" << comm_spec.worker_id() |
| 162 | << "] got raw data id: " << cur_raw_data_id |
| 163 | << ", graphx pid: " << graphx_pid; |
| 164 | // Load Fragment. |
| 165 | gs::GraphXPartitioner<OID_T> partitioner; |
| 166 | partitioner.Init(pid2Fid); |
| 167 | gs::GraphXLoader<OID_T, VID_T, VD_T, ED_T> loader(cur_raw_data_id, client, |
| 168 | comm_spec, partitioner); |
| 169 | auto arrow_frag_id = boost::leaf::try_handle_all( |
| 170 | [&loader]() { return loader.LoadFragment(); }, |
| 171 | [](const vineyard::GSError& e) { |
| 172 | LOG(FATAL) << e.error_msg; |
| 173 | return 0; |
| 174 | }, |
| 175 | [](const boost::leaf::error_info& unmatched) { |
| 176 | LOG(FATAL) << "Unmatched error " << unmatched; |
| 177 | return 0; |
| 178 | }); |
| 179 | LOG(INFO) << "Got arrow fragment id: " << arrow_frag_id; |
| 180 | std::shared_ptr<vineyard::ArrowFragment<OID_T, VID_T>> arrowFragment = |
| 181 | std::dynamic_pointer_cast<vineyard::ArrowFragment<OID_T, VID_T>>( |
| 182 | client.GetObject(arrow_frag_id)); |
| 183 | // project |
| 184 | using ProjectedFragmentType = |
| 185 | typename gs::ArrowProjectedFragment<OID_T, VID_T, VD_T, ED_T>; |
| 186 | auto v_prop_num = arrowFragment->vertex_property_num(0); |
| 187 | auto e_prop_num = arrowFragment->edge_property_num(0); |
| 188 | LOG(INFO) << "vprop num " << v_prop_num << ", e prop num: " << e_prop_num; |
| 189 | std::shared_ptr<ProjectedFragmentType> projectedFragment = |
| 190 | ProjectedFragmentType::Project(arrowFragment, 0, FLAGS_v_prop_id, 0, |
| 191 | FLAGS_e_prop_id); |
| 192 | LOG(INFO) << gs::LOAD_FRAGMENT_RES_PREFIX << ":" << getHostName() << ":" |
| 193 | << graphx_pid << ":" << projectedFragment->id(); |
| 194 | } |
| 195 | |
| 196 | template <typename FRAG_T, typename APP_TYPE> |
| 197 | void Query(grape::CommSpec& comm_spec, std::shared_ptr<FRAG_T> fragment, |
no test coverage detected