| 1007 | } |
| 1008 | |
| 1009 | void ClientRequestState::ExecLoadIcebergDataRequestImpl(TLoadDataResp response) { |
| 1010 | TLoadDataReq load_data_req = exec_request().load_data_request; |
| 1011 | RuntimeProfile* child_profile = |
| 1012 | RuntimeProfile::Create(&profile_pool_, "Child Queries"); |
| 1013 | profile_->AddChild(child_profile); |
| 1014 | // Add child queries for computing table and column stats. |
| 1015 | vector<ChildQuery> child_queries; |
| 1016 | // Prepare CREATE |
| 1017 | RuntimeProfile* create_profile = |
| 1018 | RuntimeProfile::Create(&profile_pool_, "Create table query"); |
| 1019 | child_profile->AddChild(create_profile); |
| 1020 | child_queries.emplace_back(response.create_tmp_tbl_query, this, parent_server_, |
| 1021 | create_profile, &profile_pool_); |
| 1022 | // Prepare INSERT |
| 1023 | RuntimeProfile* insert_profile = |
| 1024 | RuntimeProfile::Create(&profile_pool_, "Insert query"); |
| 1025 | child_profile->AddChild(insert_profile); |
| 1026 | child_queries.emplace_back(load_data_req.insert_into_dst_tbl_query, this, |
| 1027 | parent_server_, insert_profile, &profile_pool_); |
| 1028 | // Prepare DROP |
| 1029 | RuntimeProfile* drop_profile = |
| 1030 | RuntimeProfile::Create(&profile_pool_, "Drop table query"); |
| 1031 | child_profile->AddChild(drop_profile); |
| 1032 | child_queries.emplace_back(load_data_req.drop_tmp_tbl_query, this, |
| 1033 | parent_server_, drop_profile, &profile_pool_); |
| 1034 | // Execute queries |
| 1035 | RETURN_VOID_IF_ERROR(child_query_executor_->ExecAsync(move(child_queries))); |
| 1036 | vector<ChildQuery*> completed_queries; |
| 1037 | Status query_status = child_query_executor_->WaitForAll(&completed_queries); |
| 1038 | if (query_status.ok()) { |
| 1039 | const char* path = response.create_location.c_str(); |
| 1040 | string delete_err = "Load was succesful, but failed to remove staging data under '" |
| 1041 | + response.create_location + "', HDFS error: "; |
| 1042 | hdfsFS hdfs_conn; |
| 1043 | Status hdfs_ret = HdfsFsCache::instance()->GetConnection(path, &hdfs_conn); |
| 1044 | if (!hdfs_ret.ok()) { |
| 1045 | lock_guard<mutex> l(lock_); |
| 1046 | RETURN_VOID_IF_ERROR(UpdateQueryStatus(Status(delete_err + hdfs_ret.GetDetail()))); |
| 1047 | } |
| 1048 | if (hdfsDelete(hdfs_conn, path, 1)) { |
| 1049 | lock_guard<mutex> l(lock_); |
| 1050 | RETURN_VOID_IF_ERROR(UpdateQueryStatus(Status(delete_err + strerror(errno)))); |
| 1051 | } |
| 1052 | } else { |
| 1053 | const char* dst_path = load_data_req.source_path.c_str(); |
| 1054 | hdfsFS hdfs_dst_conn; |
| 1055 | string revert_err = "Failed to load data and failed to revert data movement, " |
| 1056 | "please check source and staging directory under '" + response.create_location |
| 1057 | + "', Query error: " + query_status.GetDetail() + " HDFS error: "; |
| 1058 | Status hdfs_ret = HdfsFsCache::instance()->GetConnection(dst_path, &hdfs_dst_conn); |
| 1059 | if (!hdfs_ret.ok()) { |
| 1060 | lock_guard<mutex> l(lock_); |
| 1061 | RETURN_VOID_IF_ERROR(UpdateQueryStatus(Status(revert_err + hdfs_ret.GetDetail()))); |
| 1062 | } |
| 1063 | for (const string& src_path : response.loaded_files) { |
| 1064 | hdfsFS hdfs_src_conn; |
| 1065 | hdfs_ret = HdfsFsCache::instance()->GetConnection(dst_path, &hdfs_src_conn); |
| 1066 | if (!hdfs_ret.ok()) { |
nothing calls this directly
no test coverage detected