| 1178 | } |
| 1179 | |
| 1180 | void HdfsScanNodeBase::StopAndFinalizeCounters() { |
| 1181 | if (!counters_running_) return; |
| 1182 | counters_running_ = false; |
| 1183 | |
| 1184 | runtime_profile_->StopPeriodicCounters(); |
| 1185 | |
| 1186 | // Output hdfs read thread concurrency into info string |
| 1187 | stringstream ss; |
| 1188 | for (int i = 0; i < hdfs_read_thread_concurrency_bucket_->size(); ++i) { |
| 1189 | ss << i << ":" << setprecision(4) |
| 1190 | << (*hdfs_read_thread_concurrency_bucket_)[i]->double_value() << "% "; |
| 1191 | } |
| 1192 | runtime_profile_->AddInfoString("Hdfs Read Thread Concurrency Bucket", ss.str()); |
| 1193 | |
| 1194 | // Convert disk access bitmap to num of disk accessed |
| 1195 | uint64_t num_disk_bitmap = disks_accessed_bitmap_.value(); |
| 1196 | int64_t num_disk_accessed = BitUtil::Popcount(num_disk_bitmap); |
| 1197 | if (num_disks_accessed_counter_ != NULL) { |
| 1198 | num_disks_accessed_counter_->Set(num_disk_accessed); |
| 1199 | } |
| 1200 | |
| 1201 | // output completed file types and counts to info string |
| 1202 | if (!file_type_counts_.empty()) { |
| 1203 | stringstream ss; |
| 1204 | { |
| 1205 | for (FileTypeCountsMap::const_iterator it = file_type_counts_.begin(); |
| 1206 | it != file_type_counts_.end(); ++it) { |
| 1207 | |
| 1208 | THdfsFileFormat::type file_format = std::get<0>(it->first); |
| 1209 | bool skipped = std::get<1>(it->first); |
| 1210 | HdfsCompressionTypesSet compressions_set = std::get<2>(it->first); |
| 1211 | int file_cnt = it->second; |
| 1212 | |
| 1213 | if (skipped) { |
| 1214 | if (file_format == THdfsFileFormat::PARQUET) { |
| 1215 | // If a scan range stored as parquet is skipped, its compression type |
| 1216 | // cannot be figured out without reading the data. |
| 1217 | ss << file_format << "/" << "Unknown" << "(Skipped):" |
| 1218 | << file_cnt << " "; |
| 1219 | } else { |
| 1220 | ss << file_format << "/" |
| 1221 | << compressions_set.GetFirstType() << "(Skipped):" |
| 1222 | << file_cnt << " "; |
| 1223 | } |
| 1224 | } else if (compressions_set.Size() == 1) { |
| 1225 | ss << file_format << "/" |
| 1226 | << compressions_set.GetFirstType() << ":" << file_cnt |
| 1227 | << " "; |
| 1228 | } else { |
| 1229 | ss << file_format << "/" << "("; |
| 1230 | bool first = true; |
| 1231 | for (auto& elem : _THdfsCompression_VALUES_TO_NAMES) { |
| 1232 | THdfsCompression::type type = static_cast<THdfsCompression::type>( |
| 1233 | elem.first); |
| 1234 | if (!compressions_set.HasType(type)) continue; |
| 1235 | if (!first) ss << ","; |
| 1236 | ss << type; |
| 1237 | first = false; |
nothing calls this directly
no test coverage detected