| 104 | } |
| 105 | |
| 106 | void DmlExecState::Update(const DmlExecStatusPB& dml_exec_status) { |
| 107 | lock_guard<mutex> l(lock_); |
| 108 | const PerPartitionStatusPBMap& new_partition_status_map = |
| 109 | dml_exec_status.per_partition_status(); |
| 110 | for (const PerPartitionStatusPBMap::value_type& part : new_partition_status_map) { |
| 111 | DmlPartitionStatusPB* status = &(per_partition_status_[part.first]); |
| 112 | status->set_num_modified_rows( |
| 113 | status->num_modified_rows() + part.second.num_modified_rows()); |
| 114 | status->set_num_deleted_rows( |
| 115 | status->num_deleted_rows() + part.second.num_deleted_rows()); |
| 116 | status->set_kudu_latest_observed_ts(max<uint64_t>( |
| 117 | part.second.kudu_latest_observed_ts(), status->kudu_latest_observed_ts())); |
| 118 | status->set_id(part.second.id()); |
| 119 | status->set_partition_base_dir(part.second.partition_base_dir()); |
| 120 | if (part.second.has_stats()) { |
| 121 | MergeDmlStats(part.second.stats(), status->mutable_stats()); |
| 122 | } |
| 123 | if (!part.second.staging_dir_to_clean_up().empty()) { |
| 124 | // Empty destination means a directory to delete in files_to_move_. |
| 125 | files_to_move_[part.second.staging_dir_to_clean_up()] = ""; |
| 126 | } |
| 127 | for (int i = 0; i < part.second.created_files_size(); ++i) { |
| 128 | const DmlFileStatusPb& file = part.second.created_files(i); |
| 129 | *status->add_created_files() = file; |
| 130 | if (!file.has_staging_path()) continue; |
| 131 | DCHECK(!file.staging_path().empty()); |
| 132 | files_to_move_[file.staging_path()] = file.final_path(); |
| 133 | } |
| 134 | for (int i = 0; i < part.second.created_delete_files_size(); ++i) { |
| 135 | const DmlFileStatusPb& file = part.second.created_delete_files(i); |
| 136 | *status->add_created_delete_files() = file; |
| 137 | if (!file.has_staging_path()) continue; |
| 138 | DCHECK(!file.staging_path().empty()); |
| 139 | files_to_move_[file.staging_path()] = file.final_path(); |
| 140 | } |
| 141 | } |
| 142 | data_files_referenced_by_position_deletes_.insert( |
| 143 | data_files_referenced_by_position_deletes_.end(), |
| 144 | dml_exec_status.data_files_referenced_by_position_deletes().begin(), |
| 145 | dml_exec_status.data_files_referenced_by_position_deletes().end()); |
| 146 | } |
| 147 | |
| 148 | uint64_t DmlExecState::GetKuduLatestObservedTimestamp() { |
| 149 | lock_guard<mutex> l(lock_); |
no test coverage detected