You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2020/11/27 19:44:18 UTC
[impala] branch master updated: IMPALA-9382: part 2/3: aggregate
profiles sent to coordinator
This is an automated email from the ASF dual-hosted git repository.
tarmstrong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new 9429bd7 IMPALA-9382: part 2/3: aggregate profiles sent to coordinator
9429bd7 is described below
commit 9429bd779de986d3e61858bef7e258bd73a2cacd
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Sun May 17 16:37:46 2020 -0700
IMPALA-9382: part 2/3: aggregate profiles sent to coordinator
This reworks the status reporting so that serialized
AggregatedRuntimeProfile objects are sent from executors
to coordinators. These profiles are substantially denser
and faster to process for higher mt_dop values. The aggregation
is also done in a single step, merging the aggregated thrift
profile from the executor directly into the final aggregated
profile, instead of converting it to an unaggregated profile
first.
The changes required were:
* A new Update() method for AggregatedRuntimeProfile that
updates the profile from a serialised AggregateRuntimeProfile
for a subset of the instances. The code is generalized from the
existing InitFromThrift() code path.
* Per-fragment reports included in the status report protobuf
when --gen_experimental_profile=true.
* Logic on the coordinator that either consumes serialized
AggregatedRuntimeProfile per fragment, when
--gen_experimental_profile=true, or consumes a serialized
RuntimeProfile per finstance otherwise.
This also adds support for event sequences and time series
in the aggregated profile, so the amount of information
in the aggregated profile is now on par with the basic profile.
We also finish off support for JSON profile. The JSON profile is
more stripped down because we do not need to round-trip profiles
via JSON and it is a much less dense profile representation.
Part 3 will clean up and improve the display of the profile.
Testing:
* Add sanity tests for aggregated runtime profile.
* Add unit tests to exercise aggregation of the various counter types
* Ran core tests.
Change-Id: Ic680cbfe94c939c2a8fad9d0943034ed058c6bca
Reviewed-on: http://gerrit.cloudera.org:8080/16057
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/coordinator-backend-state.cc | 62 +-
be/src/runtime/coordinator-backend-state.h | 30 +-
be/src/runtime/fragment-instance-state.cc | 14 +-
be/src/runtime/fragment-instance-state.h | 5 +-
be/src/runtime/fragment-state.cc | 5 +
be/src/runtime/fragment-state.h | 6 +
be/src/runtime/query-state.cc | 40 +-
be/src/runtime/query-state.h | 5 +
be/src/service/impala-server.cc | 2 +
be/src/util/runtime-profile-counters.h | 30 +-
be/src/util/runtime-profile-test.cc | 337 ++++++++-
be/src/util/runtime-profile.cc | 814 ++++++++++++++++-----
be/src/util/runtime-profile.h | 118 ++-
common/protobuf/control_service.proto | 23 +-
common/thrift/ImpalaInternalService.thrift | 4 +
common/thrift/RuntimeProfile.thrift | 38 +-
.../tpch/queries/runtime-profile-aggregated.test | 29 +
tests/custom_cluster/test_runtime_profile.py | 46 ++
18 files changed, 1348 insertions(+), 260 deletions(-)
diff --git a/be/src/runtime/coordinator-backend-state.cc b/be/src/runtime/coordinator-backend-state.cc
index 602bfca..c7beac4 100644
--- a/be/src/runtime/coordinator-backend-state.cc
+++ b/be/src/runtime/coordinator-backend-state.cc
@@ -384,6 +384,9 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
vector<AuxErrorInfoPB>* aux_error_info,
const vector<FragmentStats*>& fragment_stats) {
DCHECK(!IsEmptyBackend());
+ CHECK(FLAGS_gen_experimental_profile ||
+ backend_exec_status.fragment_exec_status().empty())
+ << "Received pre-aggregated profile but --gen_experimental_profile=false";
// Hold the exec_summary's lock to avoid exposing it half-way through
// the update loop below.
lock_guard<SpinLock> l1(exec_summary->lock);
@@ -399,21 +402,28 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
if (IsDoneLocked(lock)) return false;
// Use empty profile in case profile serialization/deserialization failed.
- // 'thrift_profiles' and 'instance_exec_status' vectors have one-to-one correspondance.
+ // Depending on the --gen_experimental_profile value, there is one profile tree per
+ // fragment (if true) or per fragment instance (if false).
vector<TRuntimeProfileTree> empty_profiles;
+ // We iterate over the profiles in order. The profile trees include the instance
+ // profiles in the same order as 'instance_exec_status', then the aggregated fragment
+ // profiles in the same order as 'fragment_exec_status'.
vector<TRuntimeProfileTree>::const_iterator profile_iter;
+ int expected_num_profiles = FLAGS_gen_experimental_profile ?
+ backend_exec_status.fragment_exec_status().size() :
+ backend_exec_status.instance_exec_status().size();
if (UNLIKELY(thrift_profiles.profile_trees.size() == 0)) {
- empty_profiles.resize(backend_exec_status.instance_exec_status().size());
+ empty_profiles.resize(expected_num_profiles);
profile_iter = empty_profiles.begin();
} else {
- DCHECK_EQ(thrift_profiles.profile_trees.size(),
- backend_exec_status.instance_exec_status().size());
+ DCHECK_EQ(expected_num_profiles, thrift_profiles.profile_trees.size());
profile_iter = thrift_profiles.profile_trees.begin();
}
+ int64_t report_time_ms = UnixMillis();
for (auto status_iter = backend_exec_status.instance_exec_status().begin();
status_iter != backend_exec_status.instance_exec_status().end();
- ++status_iter, ++profile_iter) {
+ ++status_iter) {
const FragmentInstanceExecStatusPB& instance_exec_status = *status_iter;
int64_t report_seq_no = instance_exec_status.report_seq_no();
int instance_idx = GetInstanceIdx(instance_exec_status.fragment_instance_id());
@@ -431,7 +441,9 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
}
DCHECK(!instance_stats->done_);
- instance_stats->Update(instance_exec_status, *profile_iter, exec_summary);
+ instance_stats->Update(instance_exec_status,
+ FLAGS_gen_experimental_profile ? nullptr : &*profile_iter, report_time_ms,
+ exec_summary);
// Update DML stats
if (instance_exec_status.has_dml_exec_status()) {
@@ -463,6 +475,7 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
instance_stats->done_ = true;
--num_remaining_instances_;
}
+ if (!FLAGS_gen_experimental_profile) ++profile_iter;
}
// Determine newly-completed scan ranges and update scan_range_progress.
int64_t scan_ranges_complete = backend_exec_status.scan_ranges_complete();
@@ -479,6 +492,25 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
backend_utilization_.exchange_bytes_sent = backend_exec_status.exchange_bytes_sent();
backend_utilization_.scan_bytes_sent = backend_exec_status.scan_bytes_sent();
+ // Only merge fragment stats if they are newer than that last report received.
+ if (backend_exec_status.backend_report_seq_no() <= last_backend_report_seq_no_) {
+ VLogForBackend(Substitute("Ignoring stale update with seq no $0",
+ backend_exec_status.backend_report_seq_no()));
+ } else {
+ for (const FragmentExecStatusPB& fragment_exec_status :
+ backend_exec_status.fragment_exec_status()) {
+ DCHECK(FLAGS_gen_experimental_profile)
+ << "Per-fragment agg profiles are experimental";
+ int fragment_idx = fragment_exec_status.fragment_idx();
+ FragmentStats* f = fragment_stats[fragment_idx];
+ const TRuntimeProfileTree& tmp_profile = *profile_iter;
+ f->agg_profile_->UpdateAggregatedFromInstances(
+ tmp_profile, fragment_exec_status.min_per_fragment_instance_idx());
+ ++profile_iter;
+ }
+ last_backend_report_seq_no_ = backend_exec_status.backend_report_seq_no();
+ }
+
// Update state that depends on the instance profile updates we just received.
// Skip this in the edge case where the exec RPC didn't complete.
if (exec_done_) UpdateExecStatsLocked(lock, fragment_stats, /*finalize=*/false);
@@ -714,15 +746,19 @@ Coordinator::BackendState::InstanceStats::InstanceStats(
void Coordinator::BackendState::InstanceStats::Update(
const FragmentInstanceExecStatusPB& exec_status,
- const TRuntimeProfileTree& thrift_profile, ExecSummary* exec_summary) {
- last_report_time_ms_ = UnixMillis();
+ const TRuntimeProfileTree* thrift_profile, int64_t report_time_ms,
+ ExecSummary* exec_summary) {
+ last_report_time_ms_ = report_time_ms;
DCHECK_GT(exec_status.report_seq_no(), last_report_seq_no_);
last_report_seq_no_ = exec_status.report_seq_no();
if (exec_status.done()) stopwatch_.Stop();
profile_->UpdateInfoString(LAST_REPORT_TIME_DESC,
ToStringFromUnixMillis(last_report_time_ms_));
- profile_->Update(thrift_profile);
- profile_->ComputeTimeInProfile();
+ if (!FLAGS_gen_experimental_profile) {
+ DCHECK(thrift_profile != nullptr);
+ profile_->Update(*thrift_profile);
+ profile_->ComputeTimeInProfile();
+ }
// update exec_summary
TExecSummary& thrift_exec_summary = exec_summary->thrift_exec_summary;
@@ -784,7 +820,11 @@ void Coordinator::BackendState::InstanceStats::UpdateExecStats(
completion_time_set_ = true;
}
if (!agg_profile_up_to_date_) {
- f->agg_profile_->Update(profile_, per_fragment_instance_idx());
+ // Merge this instance profile (which may or may not included the full instance
+ // profile from the executor) into the aggregated profile. We need to do this
+ // regardless of the --gen_experimental_profile setting so that coordinator counters
+ // end up in the final aggregated profile.
+ f->agg_profile_->UpdateAggregatedFromInstance(profile_, per_fragment_instance_idx());
agg_profile_up_to_date_ = true;
}
}
diff --git a/be/src/runtime/coordinator-backend-state.h b/be/src/runtime/coordinator-backend-state.h
index 9797b13..d9b5bf8 100644
--- a/be/src/runtime/coordinator-backend-state.h
+++ b/be/src/runtime/coordinator-backend-state.h
@@ -233,11 +233,14 @@ class Coordinator::BackendState {
const NetworkAddressPB& address, FragmentStats* fragment_stats,
ObjectPool* obj_pool);
- /// Updates 'this' with exec_status and the fragment intance's thrift profile. Also
- /// updates the fragment instance's TExecStats in exec_summary. Also updates the
- /// instance's avg profile. Caller must hold BackendState::lock_.
+ /// Updates 'this' with exec_status and, if --gen_experimental_profile=false, the
+ /// fragment instance's thrift profile. Also updates the fragment instance's
+ /// TExecStats in exec_summary. Also updates the instance's avg profile.
+ /// 'report_time_ms' should be the UnixMillis() value of when the report was received.
+ /// Caller must hold BackendState::lock_.
void Update(const FragmentInstanceExecStatusPB& exec_status,
- const TRuntimeProfileTree& thrift_profile, ExecSummary* exec_summary);
+ const TRuntimeProfileTree* thrift_profile, int64_t report_time_ms,
+ ExecSummary* exec_summary);
/// Update completion_times, rates, and agg_profile for this instance's
/// corresponding fragment in 'fragment_stats'. This may only be called
@@ -280,10 +283,16 @@ class Coordinator::BackendState {
/// persists for the duration of one status report update (5 seconds by default).
int64_t last_report_time_ms_ = 0;
- /// The sequence number of the last report.
+ /// The sequence number of the last report for this fragment instance.
int64_t last_report_seq_no_ = 0;
- /// owned by coordinator object pool provided in the c'tor, created in Update()
+ /// The profile tree for this instance. Owned by coordinator object pool provided in
+ /// the c'tor, created in Update().
+ /// When --gen_experimental_profile=true, this contains a copy of the full instance
+ /// profile from the executor, which is updated with each status report. It also
+ /// includes some additional counters added by the coordinator.
+ /// When --gen_experimental_profile=false, this only includes the counters added
+ /// by the coordinator.
RuntimeProfile* profile_ = nullptr;
/// True if 'agg_profile_' for the fragment is up-to-date with 'profile_'.
@@ -403,6 +412,9 @@ class Coordinator::BackendState {
/// Uses GenerateReportTimeout().
int64_t last_report_time_ms_ = 0;
+ /// The sequence number of the last report for this backend.
+ int64_t last_backend_report_seq_no_ = 0;
+
/// True if a CancelQueryFInstances RPC was already sent to this backend.
bool sent_cancel_rpc_ = false;
@@ -491,15 +503,15 @@ class Coordinator::FragmentStats {
int num_instances_;
/// Bytes assigned for instances of this fragment
- /// TODO: IMPALA-9382: can remove when we switch to the transposed profile.
+ /// TODO: IMPALA-9846: can remove when we switch to the transposed profile.
SummaryStats bytes_assigned_;
/// Completion times for instances of this fragment
- /// TODO: IMPALA-9382: can remove when we switch to the transposed profile.
+ /// TODO: IMPALA-9846: can remove when we switch to the transposed profile.
SummaryStats completion_times_;
/// Execution rates for instances of this fragment
- /// TODO: IMPALA-9382: can remove when we switch to the transposed profile.
+ /// TODO: IMPALA-9846: can remove when we switch to the transposed profile.
SummaryStats rates_;
};
diff --git a/be/src/runtime/fragment-instance-state.cc b/be/src/runtime/fragment-instance-state.cc
index 5b865f3..61cd7cc 100644
--- a/be/src/runtime/fragment-instance-state.cc
+++ b/be/src/runtime/fragment-instance-state.cc
@@ -259,7 +259,8 @@ Status FragmentInstanceState::Prepare() {
}
void FragmentInstanceState::GetStatusReport(FragmentInstanceExecStatusPB* instance_status,
- TRuntimeProfileTree* thrift_profile) {
+ TRuntimeProfileTree* unagg_profile, AggregatedRuntimeProfile* agg_profile) {
+ DCHECK_NE(unagg_profile == nullptr, agg_profile == nullptr);
DFAKE_SCOPED_LOCK(report_status_lock_);
DCHECK(!final_report_sent_);
// Update the counter for the peak per host mem usage.
@@ -279,7 +280,16 @@ void FragmentInstanceState::GetStatusReport(FragmentInstanceExecStatusPB* instan
instance_status->set_done(done);
instance_status->set_current_state(current_state());
DCHECK(profile() != nullptr);
- profile()->ToThrift(thrift_profile);
+ if (agg_profile != nullptr) {
+ // Figure out the index of this instance relative to other instances of this fragment
+ // on the backend.
+ int instance_idx = instance_ctx_.per_fragment_instance_idx -
+ fragment_state_->min_per_fragment_instance_idx();
+ agg_profile->UpdateAggregatedFromInstance(profile(), instance_idx);
+ } else {
+ DCHECK(unagg_profile != nullptr);
+ profile()->ToThrift(unagg_profile);
+ }
// Pull out and aggregate counters from the profile.
RuntimeProfile::Counter* user_time = profile()->GetCounter("TotalThreadsUserTime");
diff --git a/be/src/runtime/fragment-instance-state.h b/be/src/runtime/fragment-instance-state.h
index 35f94ba..a4cf413 100644
--- a/be/src/runtime/fragment-instance-state.h
+++ b/be/src/runtime/fragment-instance-state.h
@@ -102,9 +102,10 @@ class FragmentInstanceState {
/// Called periodically by query state thread to get the current status of this fragment
/// instance. The fragment instance's status is stored in 'instance_status' and its
- /// Thrift runtime profile is stored in 'thrift_profile'.
+ /// Thrift runtime profile is stored in either 'unagg_profile' or 'agg_profile',
+ /// depending on whether aggregated profiles are enabled.
void GetStatusReport(FragmentInstanceExecStatusPB* instance_status,
- TRuntimeProfileTree* thrift_profile);
+ TRuntimeProfileTree* unagg_profile, AggregatedRuntimeProfile* agg_profile);
/// After each call to GetStatusReport(), the query state thread should call one of the
/// following to indicate if the report rpc was successful. Note that in the case of
diff --git a/be/src/runtime/fragment-state.cc b/be/src/runtime/fragment-state.cc
index 3bf9ef5..60a2751 100644
--- a/be/src/runtime/fragment-state.cc
+++ b/be/src/runtime/fragment-state.cc
@@ -65,6 +65,11 @@ Status FragmentState::CreateFragmentStateMap(const TExecPlanFragmentInfo& fragme
fragment_map[fragment_state->fragment_idx()] = fragment_state;
// we expect fragment and instance contexts to follow the same order
DCHECK_EQ(fragment_state->fragment_idx(), instance_ctx.fragment_idx);
+ } else if (!fragment_state->instance_ctxs().empty()) {
+ // This invariant is needed for min_per_fragment_instance_idx() to be correct.
+ DCHECK_EQ(fragment_state->instance_ctxs().back()->per_fragment_instance_idx + 1,
+ instance_ctx.per_fragment_instance_idx)
+ << "Instance indexes must be sequential";
}
fragment_state->AddInstance(&instance_ctx, &instance_ctx_pb);
}
diff --git a/be/src/runtime/fragment-state.h b/be/src/runtime/fragment-state.h
index db45456..6bda279 100644
--- a/be/src/runtime/fragment-state.h
+++ b/be/src/runtime/fragment-state.h
@@ -68,6 +68,12 @@ class FragmentState {
const std::vector<const PlanFragmentInstanceCtxPB*>& instance_ctx_pbs() const {
return instance_ctx_pbs_;
}
+ /// Return the minimum per-fragment index of an instance on this backend.
+ int min_per_fragment_instance_idx() const {
+ // 'instance_ctxs_' is in ascending order, so can just return the first one.
+ DCHECK(!instance_ctxs_.empty());
+ return instance_ctxs_[0]->per_fragment_instance_idx;
+ }
const PlanNode* plan_tree() const { return plan_tree_; }
const DataSinkConfig* sink_config() const { return sink_config_; }
const TUniqueId& query_id() const { return query_state_->query_id(); }
diff --git a/be/src/runtime/query-state.cc b/be/src/runtime/query-state.cc
index c9768d3..aff7310 100644
--- a/be/src/runtime/query-state.cc
+++ b/be/src/runtime/query-state.cc
@@ -502,6 +502,7 @@ int64_t QueryState::AsyncCodegenThreadSysTime() const {
void QueryState::ConstructReport(bool instances_started,
ReportExecStatusRequestPB* report, TRuntimeProfileForest* profiles_forest) {
report->Clear();
+ report->set_backend_report_seq_no(++last_report_seq_no_);
TUniqueIdToUniqueIdPB(query_id(), report->mutable_query_id());
DCHECK(exec_rpc_params_.has_coord_state_idx());
report->set_coord_state_idx(exec_rpc_params_.coord_state_idx());
@@ -526,6 +527,12 @@ void QueryState::ConstructReport(bool instances_started,
host_profile_->ClearChunkedTimeSeriesCounters();
if (instances_started) {
+ // Map from fragment idx to the averaged profile. When aggregated profiles are
+ // enabled (IMPALA-9382), we populate this map with profiles that are aggregated
+ // from all the instances on this backend.
+ unordered_map<int, AggregatedRuntimeProfile*> agg_profiles;
+ ObjectPool agg_profile_pool;
+
// Stats that we aggregate across the instances.
int64_t cpu_user_ns = AsyncCodegenThreadUserTime();
int64_t cpu_sys_ns = AsyncCodegenThreadSysTime();
@@ -544,8 +551,21 @@ void QueryState::ConstructReport(bool instances_started,
// Update the status and profiles of this fragment instance.
FragmentInstanceExecStatusPB* instance_status =
report->add_instance_exec_status();
- profiles_forest->profile_trees.emplace_back();
- fis->GetStatusReport(instance_status, &profiles_forest->profile_trees.back());
+ if (query_ctx_.gen_aggregated_profile) {
+ int fragment_idx = fis->instance_ctx().fragment_idx;
+ AggregatedRuntimeProfile*& agg_profile = agg_profiles[fragment_idx];
+ if (agg_profile == nullptr) {
+ const auto it = fragment_state_map_.find(fragment_idx);
+ DCHECK(it != fragment_state_map_.end());
+ agg_profile = AggregatedRuntimeProfile::Create(&agg_profile_pool,
+ "tmp profile", it->second->instance_ctxs().size(), /*is_root=*/ true);
+ }
+ fis->GetStatusReport(instance_status, nullptr, agg_profile);
+ } else {
+ profiles_forest->profile_trees.emplace_back();
+ fis->GetStatusReport(
+ instance_status, &profiles_forest->profile_trees.back(), nullptr);
+ }
}
// Include these values for running and completed finstances in the status report.
@@ -561,6 +581,22 @@ void QueryState::ConstructReport(bool instances_started,
exchange_bytes_sent += fis->total_bytes_sent();
}
}
+
+ // Construct the per-fragment status reports, including runtime profiles.
+ for (const auto& entry : agg_profiles) {
+ int fragment_idx = entry.first;
+ const AggregatedRuntimeProfile* agg_profile = entry.second;
+ const auto it = fragment_state_map_.find(fragment_idx);
+ DCHECK(it != fragment_state_map_.end());
+
+ // Add the aggregated runtime profile and additional metadata to the report.
+ FragmentExecStatusPB* fragment_status = report->add_fragment_exec_status();
+ fragment_status->set_fragment_idx(fragment_idx);
+ fragment_status->set_min_per_fragment_instance_idx(
+ it->second->min_per_fragment_instance_idx());
+ profiles_forest->profile_trees.emplace_back();
+ agg_profile->ToThrift(&profiles_forest->profile_trees.back());
+ }
report->set_peak_mem_consumption(query_mem_tracker_->peak_consumption());
report->set_cpu_user_ns(cpu_user_ns);
report->set_cpu_sys_ns(cpu_sys_ns);
diff --git a/be/src/runtime/query-state.h b/be/src/runtime/query-state.h
index a72534a..49a1d1a 100644
--- a/be/src/runtime/query-state.h
+++ b/be/src/runtime/query-state.h
@@ -320,6 +320,11 @@ class QueryState {
/// Thread-safety: Only updated by the query state thread.
BackendExecState backend_exec_state_ = BackendExecState::PREPARING;
+ /// Monotonically increasing sequence number for status reports, incremented for each
+ /// report sent.
+ /// Thread-safety: Only updated by the query state thread.
+ int64_t last_report_seq_no_ = 0;
+
/// Protects 'overall_status_' and 'failed_finstance_id_'.
SpinLock status_lock_;
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index b19ebae..8b1ca2f 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -137,6 +137,7 @@ DECLARE_string(debug_actions);
DECLARE_bool(abort_on_config_error);
DECLARE_bool(disk_spill_encryption);
DECLARE_bool(enable_ldap_auth);
+DECLARE_bool(gen_experimental_profile);
DECLARE_bool(use_local_catalog);
DEFINE_int32(beeswax_port, 21000, "port on which Beeswax client requests are served."
@@ -1206,6 +1207,7 @@ void ImpalaServer::PrepareQueryContext(
query_ctx->__set_local_time_zone(local_tz_name);
query_ctx->__set_status_report_interval_ms(FLAGS_status_report_interval_ms);
query_ctx->__set_status_report_max_retry_s(FLAGS_status_report_max_retry_s);
+ query_ctx->__set_gen_aggregated_profile(FLAGS_gen_experimental_profile);
// Creating a random_generator every time is not free, but
// benchmarks show it to be slightly cheaper than contending for a
diff --git a/be/src/util/runtime-profile-counters.h b/be/src/util/runtime-profile-counters.h
index 7f05763..2f7dfe3 100644
--- a/be/src/util/runtime-profile-counters.h
+++ b/be/src/util/runtime-profile-counters.h
@@ -392,8 +392,7 @@ class RuntimeProfile::DerivedCounter : public RuntimeProfileBase::Counter {
/// average of the values in that set. The average is updated through calls
/// to UpdateCounter(), which may add a new counter or update an existing counter.
/// Set() and Add() should not be called.
-/// TODO: IMPALA-9382: rename counter. CounterVector? AggregatedCounter?
-/// TODO: IMPALA-9382: consider adding more descriptive stats, e.g. median.
+/// TODO: IMPALA-9846: rename counter. CounterVector? AggregatedCounter?
class RuntimeProfileBase::AveragedCounter : public RuntimeProfileBase::Counter {
public:
/// Construct an empty counter with no values added.
@@ -405,11 +404,19 @@ class RuntimeProfileBase::AveragedCounter : public RuntimeProfileBase::Counter {
/// Update the counter with a new value for the input instance at 'idx'.
/// No locks are obtained within this method because UpdateCounter() is called from
- /// Update(), which obtains locks on the entire counter map in a profile.
+ /// the AggregatedRuntimeProfile::Update*() functions, which obtain locks on the entire
+ /// counter map in a profile.
/// Note that it is not thread-safe to call this from two threads at the same time.
/// It is safe for it to be read at the same time as it is updated.
void UpdateCounter(Counter* new_counter, int idx);
+ /// Update the values of this counter, setting any provided values that were not
+ /// previously set. 'have_value' and 'value' are applied to values of this counter
+ /// starting at 'start_idx'.
+ /// The same thread safety rules as UpdateCounter() apply.
+ void Update(int start_idx, const std::vector<bool>& have_value,
+ const std::vector<int64_t>& values);
+
/// Answer the question whether there exists skew among all valid raw values
/// backing this average counter.
///
@@ -436,6 +443,8 @@ class RuntimeProfileBase::AveragedCounter : public RuntimeProfileBase::Counter {
void ToThrift(const std::string& name, TAggCounter* tcounter) const;
+ void ToJson(rapidjson::Document& document, rapidjson::Value* val) const override;
+
int64_t value() const override;
private:
@@ -471,6 +480,11 @@ class RuntimeProfileBase::AveragedCounter : public RuntimeProfileBase::Counter {
void PrettyPrintImpl(
const std::string& prefix, const std::string& name, std::ostream* s) const;
+ /// Implementation of ToJson parameterized by the type that 'values_' is
+ /// interpreted as - either int64_t or double, depending on the T parameter.
+ template <typename T>
+ void ToJsonImpl(rapidjson::Document& document, rapidjson::Value* val) const;
+
/// Compute all of the stats in Stats, interpreting the values in 'vals' as type T,
/// which must be double if 'unit_' is DOUBLE_VALUE or int64_t otherwise.
template <typename T>
@@ -719,6 +733,8 @@ class RuntimeProfile::EventSequence {
/// accessing the private interface. Methods are thread-safe where explicitly stated.
class RuntimeProfile::TimeSeriesCounter {
public:
+ virtual ~TimeSeriesCounter() {}
+
// Adds a sample. Thread-safe.
void AddSample(int ms_elapsed);
@@ -728,13 +744,13 @@ class RuntimeProfile::TimeSeriesCounter {
return GetSamplesLockedForSend(num_samples, period);
}
- virtual ~TimeSeriesCounter() {}
+ void ToThrift(TTimeSeriesCounter* counter);
+
+ TUnit::type unit() const { return unit_; }
private:
friend class RuntimeProfile;
- void ToThrift(TTimeSeriesCounter* counter);
-
/// Builds a new Value into 'value', using (if required) the allocator from
/// 'document'. Should set the following fields where appropriate:
/// {
@@ -773,8 +789,6 @@ class RuntimeProfile::TimeSeriesCounter {
SampleFunction fn = SampleFunction())
: name_(name), unit_(unit), sample_fn_(fn) {}
- TUnit::type unit() const { return unit_; }
-
std::string name_;
TUnit::type unit_;
SampleFunction sample_fn_;
diff --git a/be/src/util/runtime-profile-test.cc b/be/src/util/runtime-profile-test.cc
index db9e532..8ec3552 100644
--- a/be/src/util/runtime-profile-test.cc
+++ b/be/src/util/runtime-profile-test.cc
@@ -25,6 +25,7 @@
#include "common/object-pool.h"
#include "testutil/gtest-util.h"
#include "testutil/rand-util.h"
+#include "testutil/scoped-flag-setter.h"
#include "util/container-util.h"
#include "util/periodic-counter-updater.h"
#include "util/runtime-profile-counters.h"
@@ -32,6 +33,7 @@
#include "common/names.h"
+DECLARE_bool(gen_experimental_profile);
DECLARE_int32(status_report_interval_ms);
DECLARE_int32(periodic_counter_update_period_ms);
@@ -124,17 +126,17 @@ TEST(CountersTest, Basic) {
// Averaged
AggregatedRuntimeProfile* averaged_profile =
AggregatedRuntimeProfile::Create(&pool, "Merged", 2, true);
- averaged_profile->Update(from_thrift, 0);
+ averaged_profile->UpdateAggregatedFromInstance(from_thrift, 0);
counter_merged = averaged_profile->GetCounter("A");
EXPECT_EQ(counter_merged->value(), 1);
// Update again, there should be no change.
- averaged_profile->Update(from_thrift, 0);
+ averaged_profile->UpdateAggregatedFromInstance(from_thrift, 0);
EXPECT_EQ(counter_merged->value(), 1);
counter_a = profile_a2->AddCounter("A", TUnit::UNIT);
counter_a->Set(3);
- averaged_profile->Update(profile_a2, 1);
+ averaged_profile->UpdateAggregatedFromInstance(profile_a2, 1);
EXPECT_EQ(counter_merged->value(), 2);
// Update
@@ -212,8 +214,8 @@ TEST(CountersTest, MergeAndUpdate) {
profile1->ToThrift(&tprofile1);
AggregatedRuntimeProfile* averaged_profile =
AggregatedRuntimeProfile::Create(&pool, "merged", 2, true);
- averaged_profile->Update(profile1, 0);
- averaged_profile->Update(profile2, 1);
+ averaged_profile->UpdateAggregatedFromInstance(profile1, 0);
+ averaged_profile->UpdateAggregatedFromInstance(profile2, 1);
EXPECT_EQ(5, averaged_profile->num_counters());
ValidateCounter(averaged_profile, "Parent Shared", 2);
ValidateCounter(averaged_profile, "Parent 1 Only", 2);
@@ -292,7 +294,7 @@ TEST(CountersTest, MergeAndUpdateChildOrder) {
AggregatedRuntimeProfile* averaged_profile =
AggregatedRuntimeProfile::Create(&pool, "merged", 2, true);
RuntimeProfile* deserialized_profile = RuntimeProfile::Create(&pool, "Parent");
- averaged_profile->Update(profile1, 0);
+ averaged_profile->UpdateAggregatedFromInstance(profile1, 0);
deserialized_profile->Update(tprofile1_v1);
std::vector<RuntimeProfileBase*> tmp_children;
@@ -307,7 +309,7 @@ TEST(CountersTest, MergeAndUpdateChildOrder) {
RuntimeProfile* p1_child1 = RuntimeProfile::Create(&pool, "Child1");
profile1->PrependChild(p1_child1);
profile1->ToThrift(&tprofile1_v2);
- averaged_profile->Update(profile1, 0);
+ averaged_profile->UpdateAggregatedFromInstance(profile1, 0);
deserialized_profile->Update(tprofile1_v2);
averaged_profile->GetChildren(&tmp_children);
@@ -329,7 +331,7 @@ TEST(CountersTest, MergeAndUpdateChildOrder) {
EXPECT_EQ("Child2", tmp_children[0]->name());
EXPECT_EQ("Child1", tmp_children[1]->name());
profile1->ToThrift(&tprofile1_v3);
- averaged_profile->Update(profile1, 0);
+ averaged_profile->UpdateAggregatedFromInstance(profile1, 0);
deserialized_profile->Update(tprofile1_v2);
// The previous order of children that were already present is preserved.
@@ -536,6 +538,78 @@ TEST(CountersTest, SummaryStatsCounters) {
}
+// Helper for the AggregateSummaryStats that verifies the encoded event sequence
+// in the thrift representation when it was merged into the profile at instance offset
+// 'offset'.
+static void VerifyThriftSummaryStats(
+ const TRuntimeProfileNode& tnode, int offset, int total_instances) {
+ const int NUM_VALID_INSTANCES = 3;
+ DCHECK_LE(offset + NUM_VALID_INSTANCES, total_instances);
+ ASSERT_TRUE(tnode.__isset.aggregated);
+
+ const TAggregatedRuntimeProfileNode& agg_node = tnode.aggregated;
+ ASSERT_TRUE(agg_node.__isset.summary_stats_counters);
+
+ const TAggSummaryStatsCounter& tcounter = agg_node.summary_stats_counters[0];
+ EXPECT_EQ("test ss", tcounter.name);
+ EXPECT_EQ(TUnit::UNIT, tcounter.unit);
+
+ EXPECT_LE(offset + NUM_VALID_INSTANCES, tcounter.has_value.size());
+ EXPECT_LE(offset + NUM_VALID_INSTANCES, tcounter.sum.size());
+ EXPECT_LE(offset + NUM_VALID_INSTANCES, tcounter.total_num_values.size());
+ EXPECT_LE(offset + NUM_VALID_INSTANCES, tcounter.min_value.size());
+ EXPECT_LE(offset + NUM_VALID_INSTANCES, tcounter.max_value.size());
+
+ for (int i = 0; i < total_instances; ++i) {
+ if (i < offset || i >= offset + NUM_VALID_INSTANCES) {
+ EXPECT_FALSE(tcounter.has_value[i]);
+ continue;
+ }
+ EXPECT_TRUE(tcounter.has_value[i]);
+ int min_val = i - offset;
+ EXPECT_EQ(min_val * 2 + 1, tcounter.sum[i]);
+ EXPECT_EQ(2, tcounter.total_num_values[i]);
+ EXPECT_EQ(min_val, tcounter.min_value[i]);
+ EXPECT_EQ(min_val + 1, tcounter.max_value[i]);
+ }
+}
+
+// Test handling of event sequences in the aggregated profile.
+TEST(CountersTest, AggregateSummaryStats) {
+ auto cert = ScopedFlagSetter<bool>::Make(&FLAGS_gen_experimental_profile, true);
+ const int NUM_PROFILES = 3;
+ // Create a profile with event sequences with some shared event keys.
+ ObjectPool pool;
+ RuntimeProfile* profiles[NUM_PROFILES];
+ RuntimeProfile::SummaryStatsCounter* counters[NUM_PROFILES];
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ profiles[i] = RuntimeProfile::Create(&pool, "Profile");
+ counters[i] = profiles[i]->AddSummaryStatsCounter("test ss", TUnit::UNIT);
+ counters[i]->UpdateCounter(i);
+ counters[i]->UpdateCounter(i + 1);
+ }
+
+ AggregatedRuntimeProfile* averaged_profile =
+ AggregatedRuntimeProfile::Create(&pool, "Merged", 3, true);
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ averaged_profile->UpdateAggregatedFromInstance(profiles[i], i);
+ }
+
+ TRuntimeProfileTree ttree;
+ averaged_profile->ToThrift(&ttree);
+ VerifyThriftSummaryStats(ttree.nodes[0], 0, NUM_PROFILES);
+
+ // Test merging into another averaged profile at an offset
+ const int NUM_UNINIT_PROFILES = 2;
+ const int OFFSET = 1;
+ AggregatedRuntimeProfile* averaged_profile2 = AggregatedRuntimeProfile::Create(
+ &pool, "Merged 2", NUM_PROFILES + NUM_UNINIT_PROFILES, true);
+ averaged_profile2->UpdateAggregatedFromInstances(ttree, OFFSET);
+ TRuntimeProfileTree ttree2;
+ averaged_profile2->ToThrift(&ttree2);
+ VerifyThriftSummaryStats(ttree2.nodes[0], OFFSET, NUM_PROFILES + NUM_UNINIT_PROFILES);
+}
+
TEST(CountersTest, DerivedCounters) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
@@ -694,6 +768,77 @@ TEST(CountersTest, InfoStringTest) {
EXPECT_EQ(*update_dst_profile->GetInfoString("Foo"), "Bar");
}
+// Helper for the AggregateInfoStrings that verifies the encoded event sequence
+// in the thrift representation when it was merged into the profile at instance offset
+// 'offset'.
+static void VerifyThriftInfoStrings(
+ const TRuntimeProfileNode& tnode, int offset, int total_instances) {
+ ASSERT_TRUE(tnode.__isset.aggregated);
+ const int NUM_VALID_INSTANCES = 3;
+ DCHECK_LE(offset + NUM_VALID_INSTANCES, total_instances);
+
+ const TAggregatedRuntimeProfileNode& agg_node = tnode.aggregated;
+ ASSERT_TRUE(agg_node.__isset.info_strings);
+ const map<string, map<string, vector<int32_t>>>& info_strings = agg_node.info_strings;
+ auto it = info_strings.find("shared");
+ EXPECT_TRUE(it != info_strings.end());
+ EXPECT_EQ(1, it->second.size()) << "Only one distinct value for shared";
+
+ // Same value should be present in all instances, i.e. all indices should be present.
+ auto it2 = it->second.find("same value");
+ EXPECT_TRUE(it2 != it->second.end());
+ EXPECT_EQ(it2->second, vector<int32_t>({offset + 0, offset + 1, offset + 2}));
+
+ // Distinct value should have different value per instance.
+ it = info_strings.find("distinct");
+ EXPECT_TRUE(it != info_strings.end());
+ EXPECT_EQ(NUM_VALID_INSTANCES, it->second.size()) << "One distinct value per instance";
+ it2 = it->second.find("val0");
+ EXPECT_TRUE(it2 != it->second.end());
+ EXPECT_EQ(it2->second, vector<int32_t>({offset + 0}));
+ it2 = it->second.find("val1");
+ EXPECT_TRUE(it2 != it->second.end());
+ EXPECT_EQ(it2->second, vector<int32_t>({offset + 1}));
+ it2 = it->second.find("val2");
+ EXPECT_TRUE(it2 != it->second.end());
+ EXPECT_EQ(it2->second, vector<int32_t>({offset + 2}));
+}
+
+// Test handling of event sequences in the aggregated profile.
+TEST(CountersTest, AggregateInfoStrings) {
+ auto cert = ScopedFlagSetter<bool>::Make(&FLAGS_gen_experimental_profile, true);
+ const int NUM_PROFILES = 3;
+ // Create a profile with info strings that are shared across instances and then
+ // distinct across instances to test that they are deduplicated appropriately.
+ ObjectPool pool;
+ RuntimeProfile* profiles[NUM_PROFILES];
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ profiles[i] = RuntimeProfile::Create(&pool, "Profile");
+ profiles[i]->AddInfoString("shared", "same value");
+ profiles[i]->AddInfoString("distinct", Substitute("val$0", i));
+ }
+
+ AggregatedRuntimeProfile* averaged_profile =
+ AggregatedRuntimeProfile::Create(&pool, "Merged", 3, true);
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ averaged_profile->UpdateAggregatedFromInstance(profiles[i], i);
+ }
+
+ TRuntimeProfileTree ttree;
+ averaged_profile->ToThrift(&ttree);
+ VerifyThriftInfoStrings(ttree.nodes[0], 0, NUM_PROFILES);
+
+ // Test merging into another averaged profile at an offset
+ const int NUM_UNINIT_PROFILES = 2;
+ const int OFFSET = 1;
+ AggregatedRuntimeProfile* averaged_profile2 = AggregatedRuntimeProfile::Create(
+ &pool, "Merged 2", NUM_PROFILES + NUM_UNINIT_PROFILES, true);
+ averaged_profile2->UpdateAggregatedFromInstances(ttree, OFFSET);
+ TRuntimeProfileTree ttree2;
+ averaged_profile2->ToThrift(&ttree2);
+ VerifyThriftInfoStrings(ttree2.nodes[0], OFFSET, NUM_PROFILES + NUM_UNINIT_PROFILES);
+}
+
TEST(CountersTest, RateCounters) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
@@ -809,6 +954,108 @@ TEST(CountersTest, EventSequences) {
}
}
+static void CheckAscending(const vector<int64_t>& v) {
+ if (v.empty()) return;
+ int64_t prev = v[0];
+ for (int i = 1; i < v.size(); ++i) {
+ EXPECT_LE(prev, v[i]);
+ prev = v[i];
+ }
+}
+
+// Helper for the AggregateEventSequences that verifies the encoded event sequence
+// in the thrift representation when it was merged into the profile at instance offset
+// 'offset'.
+static void VerifyThriftEventSequences(
+ const TRuntimeProfileNode& tnode, int offset, int total_instances) {
+ ASSERT_TRUE(tnode.__isset.aggregated);
+ const int NUM_VALID_INSTANCES = 3;
+ DCHECK_LE(offset + NUM_VALID_INSTANCES, total_instances);
+
+ const TAggregatedRuntimeProfileNode& agg_node = tnode.aggregated;
+ ASSERT_TRUE(agg_node.__isset.event_sequences);
+ ASSERT_EQ(1, agg_node.event_sequences.size());
+ // Check that the dictionary encoding worked.
+ const TAggEventSequence& tseq = agg_node.event_sequences[0];
+ EXPECT_EQ("event sequence", tseq.name);
+ EXPECT_EQ("aaaa", tseq.label_dict[0]);
+ EXPECT_EQ("bbbb", tseq.label_dict[1]);
+ EXPECT_EQ("cccc", tseq.label_dict[2]);
+ EXPECT_EQ("dddd", tseq.label_dict[3]);
+
+ // Validate that the right number of instances are present and that the invalid
+ // instances do not have any data associated with them.
+ EXPECT_EQ(total_instances, tseq.label_idxs.size());
+ EXPECT_EQ(total_instances, tseq.timestamps.size());
+ for (int i = 0; i < total_instances; ++i) {
+ if (i < offset || i >= offset + NUM_VALID_INSTANCES) {
+ EXPECT_EQ(0, tseq.label_idxs[i].size());
+ EXPECT_EQ(0, tseq.timestamps[i].size());
+ }
+ }
+
+ // Validate the label/timestamp values for the valid instances.
+ EXPECT_EQ(3, tseq.label_idxs[offset + 0].size());
+ EXPECT_EQ(0, tseq.label_idxs[offset + 0][0]);
+ EXPECT_EQ(1, tseq.label_idxs[offset + 0][1]);
+ EXPECT_EQ(2, tseq.label_idxs[offset + 0][2]);
+ EXPECT_EQ(3, tseq.timestamps[offset + 0].size());
+ CheckAscending(tseq.timestamps[offset + 0]);
+
+ EXPECT_EQ(2, tseq.label_idxs[offset + 1].size());
+ EXPECT_EQ(0, tseq.label_idxs[offset + 1][0]);
+ EXPECT_EQ(2, tseq.label_idxs[offset + 1][1]);
+ EXPECT_EQ(2, tseq.timestamps[offset + 1].size());
+ CheckAscending(tseq.timestamps[offset + 1]);
+
+ EXPECT_EQ(3, tseq.label_idxs[offset + 2].size());
+ EXPECT_EQ(0, tseq.label_idxs[offset + 2][0]);
+ EXPECT_EQ(3, tseq.label_idxs[offset + 2][1]);
+ EXPECT_EQ(1, tseq.label_idxs[offset + 2][2]);
+ EXPECT_EQ(3, tseq.timestamps[offset + 2].size());
+ CheckAscending(tseq.timestamps[offset + 2]);
+}
+
+// Test handling of event sequences in the aggregated profile.
+TEST(CountersTest, AggregateEventSequences) {
+ auto cert = ScopedFlagSetter<bool>::Make(&FLAGS_gen_experimental_profile, true);
+ const int NUM_PROFILES = 3;
+ // Create a profile with event sequences with some shared event keys.
+ ObjectPool pool;
+ RuntimeProfile* profiles[NUM_PROFILES];
+ RuntimeProfile::EventSequence* seqs[NUM_PROFILES];
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ profiles[i] = RuntimeProfile::Create(&pool, "Profile");
+ seqs[i] = profiles[i]->AddEventSequence("event sequence");
+ seqs[i]->MarkEvent("aaaa");
+ }
+ seqs[0]->MarkEvent("bbbb");
+ seqs[0]->MarkEvent("cccc");
+ seqs[1]->MarkEvent("cccc");
+ seqs[2]->MarkEvent("dddd");
+ seqs[2]->MarkEvent("bbbb");
+
+ AggregatedRuntimeProfile* averaged_profile =
+ AggregatedRuntimeProfile::Create(&pool, "Merged", 3, true);
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ averaged_profile->UpdateAggregatedFromInstance(profiles[i], i);
+ }
+
+ TRuntimeProfileTree ttree;
+ averaged_profile->ToThrift(&ttree);
+ VerifyThriftEventSequences(ttree.nodes[0], 0, NUM_PROFILES);
+
+ // Test merging into another averaged profile at an offset
+ const int NUM_UNINIT_PROFILES = 2;
+ const int OFFSET = 1;
+ AggregatedRuntimeProfile* averaged_profile2 = AggregatedRuntimeProfile::Create(
+ &pool, "Merged 2", NUM_PROFILES + NUM_UNINIT_PROFILES, true);
+ averaged_profile2->UpdateAggregatedFromInstances(ttree, OFFSET);
+ TRuntimeProfileTree ttree2;
+ averaged_profile2->ToThrift(&ttree2);
+ VerifyThriftEventSequences(ttree2.nodes[0], OFFSET, NUM_PROFILES + NUM_UNINIT_PROFILES);
+}
+
TEST(CountersTest, UpdateEmptyEventSequence) {
// IMPALA-6824: This test makes sure that adding events to an empty event sequence does
// not crash.
@@ -1144,6 +1391,80 @@ TEST(TimeSeriesCounterTest, TestMaximumSize) {
ASSERT_EQ(samples[0], 10);
}
+// Helper for the AggregateTimeSeries that verifies the encoded event sequence
+// in the thrift representation when it was merged into the profile at instance offset
+// 'offset'.
+static void VerifyThriftTimeSeries(
+ const TRuntimeProfileNode& tnode, int offset, int total_instances) {
+ ASSERT_TRUE(tnode.__isset.aggregated);
+ const int NUM_VALID_INSTANCES = 3;
+ DCHECK_LE(offset + NUM_VALID_INSTANCES, total_instances);
+
+ const TAggregatedRuntimeProfileNode& agg_node = tnode.aggregated;
+ ASSERT_TRUE(agg_node.__isset.time_series_counters);
+
+ const TAggTimeSeriesCounter& tcounter = agg_node.time_series_counters[0];
+ EXPECT_EQ("TestCounter", tcounter.name);
+ const int test_period = FLAGS_periodic_counter_update_period_ms;
+ for (int i = 0; i < total_instances; ++i) {
+ if (i < offset || i >= offset + NUM_VALID_INSTANCES) {
+ EXPECT_EQ(0, tcounter.period_ms[i]);
+ EXPECT_EQ(0, tcounter.values[i].size());
+ EXPECT_EQ(0, tcounter.start_index[i]);
+ continue;
+ }
+ EXPECT_EQ(test_period, tcounter.period_ms[i]);
+ EXPECT_GE(tcounter.start_index[i], 0);
+ EXPECT_EQ(vector<int64_t>({0, 1, 2, 3, 4, 5, 6, 7, 8, 9}), tcounter.values[i]);
+ }
+}
+
+// Test handling of event sequences in the aggregated profile.
+TEST(TimeSeriesCounterTest, AggregateTimeSeries) {
+ auto cert = ScopedFlagSetter<bool>::Make(&FLAGS_gen_experimental_profile, true);
+ const int NUM_PROFILES = 3;
+ // Create a profile with event sequences with some shared event keys.
+ ObjectPool pool;
+ RuntimeProfile* profiles[NUM_PROFILES];
+ RuntimeProfile::TimeSeriesCounter* counters[NUM_PROFILES];
+ const int test_period = FLAGS_periodic_counter_update_period_ms;
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ profiles[i] = RuntimeProfile::Create(&pool, "Profile");
+ // Add a counter with a sample function that counts up, starting from 0.
+ int value = 0;
+ auto sample_fn = [&value]() { return value++; };
+ counters[i] =
+ profiles[i]->AddChunkedTimeSeriesCounter("TestCounter", TUnit::UNIT, sample_fn);
+
+ // Stop counter updates from interfering with the rest of the test.
+ StopAndClearCounter(profiles[i], counters[i]);
+
+ // Reset value after previous values have been retrieved.
+ value = 0;
+
+ for (int j = 0; j < 10; ++j) counters[i]->AddSample(test_period);
+ }
+
+ AggregatedRuntimeProfile* averaged_profile =
+ AggregatedRuntimeProfile::Create(&pool, "Merged", 3, true);
+ for (int i = 0; i < NUM_PROFILES; ++i) {
+ averaged_profile->UpdateAggregatedFromInstance(profiles[i], i);
+ }
+ TRuntimeProfileTree ttree;
+ averaged_profile->ToThrift(&ttree);
+ VerifyThriftTimeSeries(ttree.nodes[0], 0, NUM_PROFILES);
+
+ // Test merging into another averaged profile at an offset
+ const int NUM_UNINIT_PROFILES = 2;
+ const int OFFSET = 1;
+ AggregatedRuntimeProfile* averaged_profile2 = AggregatedRuntimeProfile::Create(
+ &pool, "Merged 2", NUM_PROFILES + NUM_UNINIT_PROFILES, true);
+ averaged_profile2->UpdateAggregatedFromInstances(ttree, OFFSET);
+ TRuntimeProfileTree ttree2;
+ averaged_profile2->ToThrift(&ttree2);
+ VerifyThriftTimeSeries(ttree2.nodes[0], OFFSET, NUM_PROFILES + NUM_UNINIT_PROFILES);
+}
+
/// Test parameter class that helps to test time series resampling during profile pretty
/// printing with a varying number of test samples.
struct TimeSeriesTestParam {
diff --git a/be/src/util/runtime-profile.cc b/be/src/util/runtime-profile.cc
index 32e2e75..831043e 100644
--- a/be/src/util/runtime-profile.cc
+++ b/be/src/util/runtime-profile.cc
@@ -50,9 +50,12 @@
DECLARE_int32(status_report_interval_ms);
DECLARE_int32(periodic_counter_update_period_ms);
+// This must be set on the coordinator to enable the alternative profile representation.
+// It should not be set on executors - the setting is sent to the executors by
+// TQueryCtx.gen_aggregated_profile.
DEFINE_bool_hidden(gen_experimental_profile, false,
- "(Experimental) generate a new aggregated runtime profile layout. Format is subject "
- "to change.");
+ "(Experimental) when set on coordinator, generate a new aggregated runtime profile "
+ "layout. Format is subject to change.");
using namespace rapidjson;
@@ -179,6 +182,46 @@ RuntimeProfileBase::RuntimeProfileBase(ObjectPool* pool, const string& name,
RuntimeProfileBase::~RuntimeProfileBase() {}
+template <typename CreateFn>
+RuntimeProfileBase* RuntimeProfileBase::AddOrCreateChild(const string& name,
+ ChildVector::iterator* insert_pos, CreateFn create_fn, bool indent) {
+ RuntimeProfileBase* child = nullptr;
+ ChildMap::iterator it = child_map_.find(name);
+ if (it != child_map_.end()) {
+ child = it->second;
+ DCHECK(child != nullptr);
+ // Search forward until the insert position is either at the end of the vector
+ // or after this child. This preserves the order if the relative order of
+ // children in all updates is consistent. If the order is inconsistent we
+ // may not find the child because it is somewhere before '*insert_pos' in the
+ // vector.
+ bool found_child = false;
+ while (*insert_pos != children_.end() && !found_child) {
+ found_child = (*insert_pos)->first == child;
+ ++*insert_pos;
+ }
+ } else {
+ child = create_fn();
+ child_map_[child->name_] = child;
+ *insert_pos = children_.insert(*insert_pos, make_pair(child, indent));
+ ++*insert_pos;
+ }
+ return child;
+}
+
+void RuntimeProfileBase::UpdateChildCountersLocked(const unique_lock<SpinLock>& lock,
+ const ChildCounterMap& src) {
+ DCHECK(lock.mutex() == &counter_map_lock_ && lock.owns_lock());
+ // Merge all new parent->child edges into the map. This assumes that all 'src' maps
+ // provided are consistent, i.e. one child does not have multiple parents.
+ ChildCounterMap::const_iterator child_counter_src_itr;
+ for (const auto& entry : src) {
+ set<string>* child_counters =
+ FindOrInsert(&child_counter_map_, entry.first, set<string>());
+ child_counters->insert(entry.second.begin(), entry.second.end());
+ }
+}
+
RuntimeProfile* RuntimeProfile::Create(ObjectPool* pool, const string& name) {
return pool->Add(new RuntimeProfile(pool, name));
}
@@ -237,15 +280,15 @@ RuntimeProfileBase* RuntimeProfileBase::CreateFromThriftHelper(
node.aggregated.num_instances, node.aggregated.__isset.input_profiles);
} else {
// If we're not using the transposed profile representation, just convert
- // the averaged profile to a regular profile (this is what this code
- // always did in the past).
+ // the averaged profile to a regular profile (this preserves previous behaviour
+ // of the code when --gen_experimental_profile=false).
profile = RuntimeProfile::Create(pool, node.name);
}
profile->metadata_ = node.node_metadata;
profile->InitFromThrift(node, pool);
profile->child_counter_map_ = node.child_counters_map;
- // TODO: IMPALA-9382: move to RuntimeProfile::InitFromThrift() once 'info_strings_' is
+ // TODO: IMPALA-9846: move to RuntimeProfile::InitFromThrift() once 'info_strings_' is
// moved.
profile->info_strings_ = node.info_strings;
profile->info_strings_display_order_ = node.info_strings_display_order;
@@ -295,7 +338,8 @@ void RuntimeProfile::InitFromThrift(const TRuntimeProfileNode& node, ObjectPool*
}
}
-void AggregatedRuntimeProfile::Update(RuntimeProfile* other, int idx) {
+void AggregatedRuntimeProfile::UpdateAggregatedFromInstance(
+ RuntimeProfile* other, int idx) {
{
lock_guard<SpinLock> l(input_profile_name_lock_);
DCHECK(!input_profile_names_.empty())
@@ -303,126 +347,376 @@ void AggregatedRuntimeProfile::Update(RuntimeProfile* other, int idx) {
input_profile_names_[idx] = other->name();
}
- UpdateRecursive(other, idx);
+ UpdateAggregatedFromInstanceRecursive(other, idx);
// Recursively compute times on the whole tree.
ComputeTimeInProfile();
}
-void AggregatedRuntimeProfile::UpdateRecursive(RuntimeProfile* other, int idx) {
+void AggregatedRuntimeProfile::UpdateAggregatedFromInstanceRecursive(
+ RuntimeProfile* other, int idx) {
DCHECK(other != NULL);
DCHECK_GE(idx, 0);
DCHECK_LT(idx, num_input_profiles_);
- // Merge this level
+ // Merge all counters and other items in the profile at this level.
+ UpdateCountersFromInstance(other, idx);
+ UpdateInfoStringsFromInstance(other, idx);
+ UpdateSummaryStatsFromInstance(other, idx);
+ UpdateEventSequencesFromInstance(other, idx);
+
{
- CounterMap::iterator dst_iter;
- CounterMap::const_iterator src_iter;
- lock_guard<SpinLock> l(counter_map_lock_);
- lock_guard<SpinLock> m(other->counter_map_lock_);
- for (src_iter = other->counter_map_.begin();
- src_iter != other->counter_map_.end(); ++src_iter) {
- dst_iter = counter_map_.find(src_iter->first);
- AveragedCounter* avg_counter;
-
- // Get the counter with the same name in dst_iter (this->counter_map_)
- // Create one if it doesn't exist.
- if (dst_iter == counter_map_.end()) {
- avg_counter = pool_->Add(
- new AveragedCounter(src_iter->second->unit(), num_input_profiles_));
- counter_map_[src_iter->first] = avg_counter;
+ lock_guard<SpinLock> l(children_lock_);
+ lock_guard<SpinLock> m(other->children_lock_);
+ // Recursively merge children with matching names.
+ // Track the current position in the vector so we preserve the order of children
+ // if children are added after the first Update() call (IMPALA-6694).
+ // E.g. if the first update sends [B, D] and the second update sends [A, B, C, D],
+ // then this code makes sure that children_ is [A, B, C, D] afterwards.
+ ChildVector::iterator insert_pos = children_.begin();
+ for (int i = 0; i < other->children_.size(); ++i) {
+ RuntimeProfile* other_child =
+ dynamic_cast<RuntimeProfile*>(other->children_[i].first);
+ DCHECK(other_child != nullptr)
+ << other->children_[i].first->name() << " must be a RuntimeProfile";
+ bool indent_other_child = other->children_[i].second;
+ AggregatedRuntimeProfile* child = dynamic_cast<AggregatedRuntimeProfile*>(
+ AddOrCreateChild(other_child->name(), &insert_pos, [this, other_child] () {
+ AggregatedRuntimeProfile* child2 = Create(
+ pool_, other_child->name(), num_input_profiles_, /*is_root=*/false);
+ child2->metadata_ = other_child->metadata();
+ return child2;
+ }, indent_other_child));
+ DCHECK(child != nullptr);
+ child->UpdateAggregatedFromInstanceRecursive(other_child, idx);
+ }
+ }
+}
+
+void AggregatedRuntimeProfile::UpdateCountersFromInstance(
+ RuntimeProfile* other, int idx) {
+ CounterMap::iterator dst_iter;
+ CounterMap::const_iterator src_iter;
+ unique_lock<SpinLock> l(counter_map_lock_);
+ unique_lock<SpinLock> m(other->counter_map_lock_);
+ for (src_iter = other->counter_map_.begin();
+ src_iter != other->counter_map_.end(); ++src_iter) {
+ dst_iter = counter_map_.find(src_iter->first);
+ AveragedCounter* avg_counter;
+
+ // Get the counter with the same name in dst_iter (this->counter_map_)
+ // Create one if it doesn't exist.
+ if (dst_iter == counter_map_.end()) {
+ avg_counter = pool_->Add(
+ new AveragedCounter(src_iter->second->unit(), num_input_profiles_));
+ counter_map_[src_iter->first] = avg_counter;
+ } else {
+ DCHECK(dst_iter->second->unit() == src_iter->second->unit());
+ avg_counter = static_cast<AveragedCounter*>(dst_iter->second);
+ }
+ avg_counter->UpdateCounter(src_iter->second, idx);
+ }
+
+ for (const auto& src_entry : other->time_series_counter_map_) {
+ RuntimeProfile::TimeSeriesCounter* src = src_entry.second;
+ TAggTimeSeriesCounter& dst = time_series_counter_map_[src_entry.first];
+ if (dst.values.empty()) {
+ dst.name = src_entry.first;
+ dst.unit = src->unit();
+ dst.period_ms.resize(num_input_profiles_);
+ dst.values.resize(num_input_profiles_);
+ dst.start_index.resize(num_input_profiles_);
+ } else {
+ DCHECK_EQ(dst.unit, src->unit());
+ }
+ // Use a temporary thrift counter so we can reuse the thrift conversion code.
+ TTimeSeriesCounter tcounter;
+ src->ToThrift(&tcounter);
+ dst.period_ms[idx] = tcounter.period_ms;
+ dst.values[idx] = tcounter.values;
+ dst.start_index[idx] = tcounter.start_index;
+ }
+
+ UpdateChildCountersLocked(l, other->child_counter_map_);
+}
+
+void AggregatedRuntimeProfile::UpdateInfoStringsFromInstance(
+ RuntimeProfile* other, int idx) {
+ lock_guard<SpinLock> l(agg_info_strings_lock_);
+ lock_guard<SpinLock> m(other->info_strings_lock_);
+ for (const auto& entry : other->info_strings_) {
+ vector<string>& values = agg_info_strings_[entry.first];
+ if (values.empty()) values.resize(num_input_profiles_);
+ if (values[idx] != entry.second) values[idx] = entry.second;
+ }
+}
+
+void AggregatedRuntimeProfile::UpdateSummaryStatsFromInstance(
+ RuntimeProfile* other, int idx) {
+ lock_guard<SpinLock> l(summary_stats_map_lock_);
+ lock_guard<SpinLock> m(other->summary_stats_map_lock_);
+ for (const RuntimeProfile::SummaryStatsCounterMap::value_type& src_entry :
+ other->summary_stats_map_) {
+ DCHECK_GT(num_input_profiles_, 0);
+ AggSummaryStatsCounterMap::mapped_type& agg_entry =
+ summary_stats_map_[src_entry.first];
+ vector<SummaryStatsCounter*>& agg_instance_counters = agg_entry.second;
+ if (agg_instance_counters.empty()) {
+ agg_instance_counters.resize(num_input_profiles_);
+ agg_entry.first = src_entry.second->unit();
+ } else {
+ DCHECK_EQ(agg_entry.first, src_entry.second->unit()) << "Unit must be consistent";
+ }
+
+ // Get the counter with the same name. Create one if it doesn't exist.
+ if (agg_instance_counters[idx] == nullptr) {
+ agg_instance_counters[idx] =
+ pool_->Add(new SummaryStatsCounter(src_entry.second->unit()));
+ }
+ // Overwrite the previous value with the new value.
+ agg_instance_counters[idx]->SetStats(*src_entry.second);
+ }
+}
+
+void AggregatedRuntimeProfile::UpdateEventSequencesFromInstance(
+ RuntimeProfile* other, int idx) {
+ lock_guard<SpinLock> l(event_sequence_lock_);
+ lock_guard<SpinLock> m(other->event_sequence_lock_);
+ for (const auto& src_entry : other->event_sequence_map_) {
+ DCHECK_GT(num_input_profiles_, 0);
+ AggEventSequence& seq = event_sequence_map_[src_entry.first];
+ if (seq.label_idxs.empty()) {
+ seq.label_idxs.resize(num_input_profiles_);
+ seq.timestamps.resize(num_input_profiles_);
+ } else {
+ DCHECK_EQ(num_input_profiles_, seq.label_idxs.size());
+ DCHECK_EQ(num_input_profiles_, seq.timestamps.size());
+ }
+
+ std::vector<int32_t>& label_idxs = seq.label_idxs[idx];
+ std::vector<int64_t>& timestamps = seq.timestamps[idx];
+ label_idxs.clear();
+ timestamps.clear();
+ RuntimeProfile::EventSequence::EventList events;
+ src_entry.second->GetEvents(&events);
+ for (const RuntimeProfile::EventSequence::Event& event : events) {
+ auto it = seq.labels.find(event.first);
+ int32_t label_idx;
+ if (it == seq.labels.end()) {
+ label_idx = seq.labels.size();
+ seq.labels.emplace(event.first, label_idx);
} else {
- DCHECK(dst_iter->second->unit() == src_iter->second->unit());
- avg_counter = static_cast<AveragedCounter*>(dst_iter->second);
+ label_idx = it->second;
}
- avg_counter->UpdateCounter(src_iter->second, idx);
+ label_idxs.push_back(label_idx);
+ timestamps.push_back(event.second);
}
+ }
+}
- ChildCounterMap::const_iterator child_counter_src_itr;
- for (child_counter_src_itr = other->child_counter_map_.begin();
- child_counter_src_itr != other->child_counter_map_.end();
- ++child_counter_src_itr) {
- set<string>* child_counters = FindOrInsert(&child_counter_map_,
- child_counter_src_itr->first, set<string>());
- child_counters->insert(child_counter_src_itr->second.begin(),
- child_counter_src_itr->second.end());
+// TODO: do we need any special handling of the computed timers? Like local/total time.
+void AggregatedRuntimeProfile::UpdateAggregatedFromInstances(
+ const TRuntimeProfileTree& src, int start_idx) {
+ DCHECK(FLAGS_gen_experimental_profile)
+ << "Aggregated profiles should only be used on the coordinator when enabled by flag";
+ if (UNLIKELY(src.nodes.size()) == 0) return;
+ const TRuntimeProfileNode& root = src.nodes[0];
+ DCHECK(root.__isset.aggregated);
+ const TAggregatedRuntimeProfileNode& agg_root = root.aggregated;
+ {
+ lock_guard<SpinLock> l(input_profile_name_lock_);
+ DCHECK(!input_profile_names_.empty())
+ << "Update() can only be called on root of averaged profile tree";
+ DCHECK_EQ(agg_root.input_profiles.size(), agg_root.num_instances);
+ for (int i = 0; i < agg_root.num_instances; ++i) {
+ input_profile_names_[start_idx + i] = agg_root.input_profiles[i];
}
}
- if (FLAGS_gen_experimental_profile) {
- lock_guard<SpinLock> l(agg_info_strings_lock_);
- lock_guard<SpinLock> m(other->info_strings_lock_);
- for (const auto& entry : other->info_strings_) {
- vector<string>& values = agg_info_strings_[entry.first];
- if (values.empty()) values.resize(num_input_profiles_);
- if (values[idx] != entry.second) values[idx] = entry.second;
+ int node_idx = 0;
+ UpdateAggregatedFromInstancesRecursive(src, &node_idx, start_idx);
+
+ // Recursively compute times on the whole tree.
+ ComputeTimeInProfile();
+}
+
+void AggregatedRuntimeProfile::UpdateAggregatedFromInstancesRecursive(
+ const TRuntimeProfileTree& src, int* node_idx, int start_idx) {
+ DCHECK_LT(*node_idx, src.nodes.size());
+ const TRuntimeProfileNode& node = src.nodes[*node_idx];
+ DCHECK(node.__isset.aggregated);
+
+ UpdateFromInstances(node, start_idx, pool_);
+
+ ++*node_idx;
+ {
+ lock_guard<SpinLock> l(children_lock_);
+ ChildVector::iterator insert_pos = children_.begin();
+ // Update children with matching names; create new ones if they don't match.
+ for (int i = 0; i < node.num_children; ++i) {
+ const TRuntimeProfileNode& tchild = src.nodes[*node_idx];
+ AggregatedRuntimeProfile* child = dynamic_cast<AggregatedRuntimeProfile*>(
+ AddOrCreateChild(tchild.name, &insert_pos, [this, tchild] () {
+ AggregatedRuntimeProfile* child2 =
+ Create(pool_, tchild.name, num_input_profiles_, /*is_root=*/false);
+ child2->metadata_ = tchild.node_metadata;
+ return child2;
+ }, tchild.indent));
+ DCHECK(child != nullptr);
+ child->UpdateAggregatedFromInstancesRecursive(src, node_idx, start_idx);
}
}
+}
- if (FLAGS_gen_experimental_profile) {
- // Merge summary stats.
- lock_guard<SpinLock> l(summary_stats_map_lock_);
- lock_guard<SpinLock> m(other->summary_stats_map_lock_);
- for (const RuntimeProfile::SummaryStatsCounterMap::value_type& src_entry :
- other->summary_stats_map_) {
- DCHECK_GT(num_input_profiles_, 0);
- AggSummaryStatsCounterMap::mapped_type& agg_entry =
- summary_stats_map_[src_entry.first];
- vector<SummaryStatsCounter*>& agg_instance_counters = agg_entry.second;
- if (agg_instance_counters.empty()) {
- agg_instance_counters.resize(num_input_profiles_);
- agg_entry.first = src_entry.second->unit();
- } else {
- DCHECK_EQ(agg_entry.first, src_entry.second->unit()) << "Unit must be consistent";
+void AggregatedRuntimeProfile::UpdateFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool) {
+ UpdateCountersFromInstances(node, start_idx, pool);
+ UpdateInfoStringsFromInstances(node, start_idx, pool);
+ UpdateSummaryStatsFromInstances(node, start_idx, pool);
+ UpdateEventSequencesFromInstances(node, start_idx, pool);
+}
+
+void AggregatedRuntimeProfile::UpdateCountersFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool) {
+ unique_lock<SpinLock> l(counter_map_lock_);
+ DCHECK(node.__isset.aggregated);
+ const TAggregatedRuntimeProfileNode& agg_node = node.aggregated;
+ for (const TAggCounter& tcounter : agg_node.counters) {
+ auto dst_iter = counter_map_.find(tcounter.name);
+ // Get the counter with the same name in dst_iter (this->counter_map_)
+ // Create one if it doesn't exist.
+ AveragedCounter* avg_counter;
+ if (dst_iter == counter_map_.end()) {
+ avg_counter = pool->Add(new AveragedCounter(tcounter.unit, num_input_profiles_));
+ counter_map_[tcounter.name] = avg_counter;
+ } else {
+ DCHECK(dst_iter->second->unit() == tcounter.unit);
+ avg_counter = static_cast<AveragedCounter*>(dst_iter->second);
+ }
+ avg_counter->Update(start_idx, tcounter.has_value, tcounter.values);
+ }
+
+ for (const TAggTimeSeriesCounter& tcounter : agg_node.time_series_counters) {
+ TAggTimeSeriesCounter& dst = time_series_counter_map_[tcounter.name];
+ if (dst.values.empty()) {
+ dst.name = tcounter.name;
+ dst.unit = tcounter.unit;
+ dst.period_ms.resize(num_input_profiles_);
+ dst.values.resize(num_input_profiles_);
+ dst.start_index.resize(num_input_profiles_);
+ } else {
+ DCHECK_EQ(dst.unit, tcounter.unit);
+ }
+ DCHECK_LE(start_idx + tcounter.values.size(), num_input_profiles_);
+ DCHECK_EQ(tcounter.values.size(), tcounter.period_ms.size());
+ DCHECK_EQ(tcounter.values.size(), tcounter.start_index.size()) << tcounter.name;
+ for (int i = 0; i < tcounter.values.size(); ++i) {
+ int idx = start_idx + i;
+ dst.period_ms[idx] = tcounter.period_ms[i];
+ dst.values[idx] = tcounter.values[i];
+ dst.start_index[idx] = tcounter.start_index[i];
+ }
+ }
+
+ UpdateChildCountersLocked(l, node.child_counters_map);
+}
+
+void AggregatedRuntimeProfile::UpdateInfoStringsFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool) {
+ lock_guard<SpinLock> l(agg_info_strings_lock_);
+ DCHECK(node.__isset.aggregated);
+ const TAggregatedRuntimeProfileNode& agg_node = node.aggregated;
+ for (const auto& entry : agg_node.info_strings) {
+ vector<string>& values = agg_info_strings_[entry.first];
+ if (values.empty()) values.resize(num_input_profiles_);
+ for (const auto& distinct_val : entry.second) {
+ const string& val = distinct_val.first;
+ for (int offset : distinct_val.second) {
+ if (values[start_idx + offset] != val) values[start_idx + offset] = val;
}
+ }
+ }
+}
+
+void AggregatedRuntimeProfile::UpdateSummaryStatsFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool) {
+ lock_guard<SpinLock> l(summary_stats_map_lock_);
+ DCHECK(node.__isset.aggregated);
+ const TAggregatedRuntimeProfileNode& agg_node = node.aggregated;
+ for (const TAggSummaryStatsCounter& tcounter : agg_node.summary_stats_counters) {
+ DCHECK_GT(num_input_profiles_, 0);
+ DCHECK_LE(start_idx + tcounter.has_value.size(), num_input_profiles_);
+ auto& entry = summary_stats_map_[tcounter.name];
+ vector<SummaryStatsCounter*>& instance_counters = entry.second;
+ if (instance_counters.empty()) {
+ instance_counters.resize(num_input_profiles_);
+ entry.first = tcounter.unit;
+ } else {
+ DCHECK_EQ(entry.first, tcounter.unit) << "Unit must be consistent";
+ }
+ TSummaryStatsCounter tmp_counter;
+ tmp_counter.name = tcounter.name;
+ tmp_counter.unit = tcounter.unit;
+ for (int i = 0; i < tcounter.has_value.size(); ++i) {
+ if (!tcounter.has_value[i]) continue;
+ int idx = start_idx + i;
// Get the counter with the same name. Create one if it doesn't exist.
- if (agg_instance_counters[idx] == nullptr) {
- agg_instance_counters[idx] =
- pool_->Add(new SummaryStatsCounter(src_entry.second->unit()));
+ if (instance_counters[idx] == nullptr) {
+ instance_counters[idx] = pool->Add(new SummaryStatsCounter(tcounter.unit));
}
// Overwrite the previous value with the new value.
- agg_instance_counters[idx]->SetStats(*src_entry.second);
+ tmp_counter.sum = tcounter.sum[i];
+ tmp_counter.total_num_values = tcounter.total_num_values[i];
+ tmp_counter.min_value = tcounter.min_value[i];
+ tmp_counter.max_value = tcounter.max_value[i];
+ instance_counters[idx]->SetStats(tmp_counter);
}
}
+}
- {
- lock_guard<SpinLock> l(children_lock_);
- lock_guard<SpinLock> m(other->children_lock_);
- // Recursively merge children with matching names.
- // Track the current position in the vector so we preserve the order of children
- // if children are added after the first Update() call (IMPALA-6694).
- // E.g. if the first update sends [B, D] and the second update sends [A, B, C, D],
- // then this code makes sure that children_ is [A, B, C, D] afterwards.
- ChildVector::iterator insert_pos = children_.begin();
- for (int i = 0; i < other->children_.size(); ++i) {
- RuntimeProfile* other_child =
- dynamic_cast<RuntimeProfile*>(other->children_[i].first);
- DCHECK(other_child != nullptr)
- << other->children_[i].first->name() << " must be a RuntimeProfile";
- ChildMap::iterator j = child_map_.find(other_child->name());
- AggregatedRuntimeProfile* child = NULL;
- if (j != child_map_.end()) {
- child = dynamic_cast<AggregatedRuntimeProfile*>(j->second);
- DCHECK(child != nullptr);
- // Search forward until the insert position is either at the end of the vector
- // or after this child. This preserves the order if the relative order of
- // children in all updates is consistent.
- bool found_child = false;
- while (insert_pos != children_.end() && !found_child) {
- found_child = insert_pos->first == child;
- ++insert_pos;
+void AggregatedRuntimeProfile::UpdateEventSequencesFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool) {
+ lock_guard<SpinLock> l(event_sequence_lock_);
+ DCHECK(node.__isset.aggregated);
+ const TAggregatedRuntimeProfileNode& agg_node = node.aggregated;
+ for (const TAggEventSequence& tseq : agg_node.event_sequences) {
+ DCHECK_GT(num_input_profiles_, 0);
+ AggEventSequence& seq = event_sequence_map_[tseq.name];
+ if (seq.label_idxs.empty()) {
+ seq.label_idxs.resize(num_input_profiles_);
+ seq.timestamps.resize(num_input_profiles_);
+ } else {
+ DCHECK_EQ(num_input_profiles_, seq.label_idxs.size());
+ DCHECK_EQ(num_input_profiles_, seq.timestamps.size());
+ }
+
+ DCHECK_LE(start_idx + tseq.timestamps.size(), num_input_profiles_);
+ DCHECK_EQ(tseq.timestamps.size(), tseq.label_idxs.size());
+ for (int i = 0; i < tseq.timestamps.size(); ++i) {
+ int idx = start_idx + i;
+ std::vector<int32_t>& label_idxs = seq.label_idxs[idx];
+ std::vector<int64_t>& timestamps = seq.timestamps[idx];
+ label_idxs.clear();
+ timestamps.clear();
+ DCHECK_EQ(tseq.timestamps[i].size(), tseq.label_idxs[i].size());
+ for (int j = 0; j < tseq.timestamps[i].size(); ++j) {
+ // Decode label using dictionary and encode using this profile's dictionary.
+ int32_t src_label_idx = tseq.label_idxs[i][j];
+ DCHECK_GE(src_label_idx, 0);
+ DCHECK_LT(src_label_idx, tseq.label_dict.size());
+ const string& label = tseq.label_dict[src_label_idx];
+ auto it = seq.labels.find(label);
+ int32_t label_idx;
+ if (it == seq.labels.end()) {
+ label_idx = seq.labels.size();
+ seq.labels.emplace(label, label_idx);
+ } else {
+ label_idx = it->second;
}
- } else {
- child =
- Create(pool_, other_child->name(), num_input_profiles_, /*is_root=*/false);
- child->metadata_ = other_child->metadata();
- bool indent_other_child = other->children_[i].second;
- child_map_[child->name_] = child;
- insert_pos = children_.insert(insert_pos, make_pair(child, indent_other_child));
- ++insert_pos;
+ label_idxs.push_back(label_idx);
+ timestamps.push_back(tseq.timestamps[i][j]);
}
- child->UpdateRecursive(other_child, idx);
}
}
}
@@ -544,26 +838,13 @@ void RuntimeProfile::Update(const vector<TRuntimeProfileNode>& nodes, int* idx)
// Update children with matching names; create new ones if they don't match.
for (int i = 0; i < node.num_children; ++i) {
const TRuntimeProfileNode& tchild = nodes[*idx];
- ChildMap::iterator j = child_map_.find(tchild.name);
- RuntimeProfile* child = NULL;
- if (j != child_map_.end()) {
- child = dynamic_cast<RuntimeProfile*>(j->second);
- DCHECK(child != nullptr) << j->second->name() << " must be a RuntimeProfile";
- // Search forward until the insert position is either at the end of the vector
- // or after this child. This preserves the order if the relative order of
- // children in all updates is consistent.
- bool found_child = false;
- while (insert_pos != children_.end() && !found_child) {
- found_child = insert_pos->first == child;
- ++insert_pos;
- }
- } else {
- child = Create(pool_, tchild.name);
- child->metadata_ = tchild.node_metadata;
- child_map_[tchild.name] = child;
- insert_pos = children_.insert(insert_pos, make_pair(child, tchild.indent));
- ++insert_pos;
- }
+ RuntimeProfile* child = dynamic_cast<RuntimeProfile*>(
+ AddOrCreateChild(tchild.name, &insert_pos, [this, tchild] () {
+ RuntimeProfile* child2 = Create(pool_, tchild.name);
+ child2->metadata_ = tchild.node_metadata;
+ return child2;
+ }, tchild.indent));
+ DCHECK(child != nullptr);
child->Update(nodes, idx);
}
}
@@ -947,14 +1228,6 @@ void RuntimeProfileBase::ToJsonHelper(Value* parent, Document* d) const {
parent->AddMember("num_children", children_.size(), allocator);
// 3. Metadata
- // Set the required metadata field to the plan node ID for compatibility with any tools
- // that rely on the plan node id being set there.
- // Legacy field. May contain the node ID for plan nodes.
- // Replaced by node_metadata, which contains richer metadata.
- parent->AddMember("metadata",
- metadata_.__isset.plan_node_id ? metadata_.plan_node_id : -1, allocator);
- // requires exactly one field of a union to be set so we only mark node_metadata
- // as set if that is the case.
if (metadata_.__isset.plan_node_id || metadata_.__isset.data_sink_id){
Value node_metadata_json(kObjectType);
if (metadata_.__isset.plan_node_id){
@@ -967,7 +1240,7 @@ void RuntimeProfileBase::ToJsonHelper(Value* parent, Document* d) const {
}
// 4. Info strings
- // TODO: IMPALA-9382: move to subclass once 'info_strings_' is also moved
+ // TODO: IMPALA-9846: move to subclass once 'info_strings_' is also moved
{
lock_guard<SpinLock> l(info_strings_lock_);
if (!info_strings_.empty()) {
@@ -1105,7 +1378,7 @@ void RuntimeProfileBase::PrettyPrint(ostream* s, const string& prefix) const {
}
stream << endl;
- // TODO: IMPALA-9382: move to RuntimeProfile::PrettyPrintInfoStrings() once we move
+ // TODO: IMPALA-9846: move to RuntimeProfile::PrettyPrintInfoStrings() once we move
// 'info_strings_'.
{
lock_guard<SpinLock> l(info_strings_lock_);
@@ -1135,6 +1408,32 @@ void RuntimeProfileBase::PrettyPrint(ostream* s, const string& prefix) const {
void RuntimeProfile::PrettyPrintInfoStrings(ostream* s, const string& prefix) const {}
+// Helper to pretty-print the data for a time series counter. 'num' is the number of
+// values in 'samples'.
+//
+// The format is as follow:
+// - <Label> (<period>): <val1>, <val2>, <etc>
+static void PrettyPrintTimeSeries(const string& label, const int64_t* samples, int num,
+ int period_ms, const string& prefix, TUnit::type unit, ostream* s) {
+ ostream& stream = *s;
+ if (num == 0) return;
+ // Clamp number of printed values at 64, the maximum number of values in the
+ // SamplingTimeSeriesCounter.
+ int step = 1 + (num - 1) / 64;
+ period_ms *= step;
+ stream << prefix << " - " << label << " ("
+ << PrettyPrinter::Print(period_ms * 1000000L, TUnit::TIME_NS) << "): ";
+ for (int i = 0; i < num; i += step) {
+ stream << PrettyPrinter::Print(samples[i], unit);
+ if (i + step < num) stream << ", ";
+ }
+ if (step > 1) {
+ stream << " (Showing " << ((num + 1) / step) << " of " << num << " values from "
+ "Thrift Profile)";
+ }
+ stream << endl;
+}
+
void RuntimeProfile::PrettyPrintSubclassCounters(ostream* s, const string& prefix) const {
ostream& stream = *s;
{
@@ -1171,31 +1470,14 @@ void RuntimeProfile::PrettyPrintSubclassCounters(ostream* s, const string& prefi
}
{
- // Print all time series counters as following:
- // - <Name> (<period>): <val1>, <val2>, <etc>
+ // Print time series counters.
lock_guard<SpinLock> l(counter_map_lock_);
for (const auto& v : time_series_counter_map_) {
const TimeSeriesCounter* counter = v.second;
lock_guard<SpinLock> l(counter->lock_);
int num, period;
const int64_t* samples = counter->GetSamplesLocked(&num, &period);
- if (num > 0) {
- // Clamp number of printed values at 64, the maximum number of values in the
- // SamplingTimeSeriesCounter.
- int step = 1 + (num - 1) / 64;
- period *= step;
- stream << prefix << " - " << v.first << " ("
- << PrettyPrinter::Print(period * 1000000L, TUnit::TIME_NS) << "): ";
- for (int i = 0; i < num; i += step) {
- stream << PrettyPrinter::Print(samples[i], counter->unit());
- if (i + step < num) stream << ", ";
- }
- if (step > 1) {
- stream << " (Showing " << ((num + 1) / step) << " of " << num << " values from "
- "Thrift Profile)";
- }
- stream << endl;
- }
+ PrettyPrintTimeSeries(v.first, samples, num, period, prefix, counter->unit(), s);
}
}
@@ -1308,13 +1590,17 @@ void RuntimeProfile::SetTExecSummary(const TExecSummary& summary) {
t_exec_summary_ = summary;
}
-void RuntimeProfile::ToThrift(TRuntimeProfileTree* tree) const {
+void RuntimeProfileBase::ToThrift(TRuntimeProfileTree* tree) const {
tree->nodes.clear();
ToThriftHelper(&tree->nodes);
- ExecSummaryToThrift(tree);
if (FLAGS_gen_experimental_profile) tree->__set_profile_version(2);
}
+void RuntimeProfile::ToThrift(TRuntimeProfileTree* tree) const {
+ RuntimeProfileBase::ToThrift(tree);
+ ExecSummaryToThrift(tree);
+}
+
void RuntimeProfileBase::ToThriftHelper(vector<TRuntimeProfileNode>* nodes) const {
// Use a two-pass approach where we first collect nodes with a pre-order traversal and
// then serialize them. This is done to allow reserving the full vector of
@@ -1359,7 +1645,7 @@ void RuntimeProfileBase::ToThriftHelper(TRuntimeProfileNode* out_node) const {
node.child_counters_map = child_counter_map_;
}
- // TODO: IMPALA-9382: move to RuntimeProfile::ToThriftSubclass() once 'info_strings_'
+ // TODO: IMPALA-9846: move to RuntimeProfile::ToThriftSubclass() once 'info_strings_'
// is moved.
{
lock_guard<SpinLock> l(info_strings_lock_);
@@ -1782,12 +2068,7 @@ RuntimeProfileBase::AveragedCounter::AveragedCounter(TUnit::type unit,
has_value_(make_unique<AtomicBool[]>(values.size())),
values_(make_unique<AtomicInt64[]>(values.size())) {
DCHECK_EQ(has_value.size(), values.size());
- for (int i = 0; i < values.size(); ++i) {
- if (has_value[i]) {
- has_value_[i].Store(true);
- values_[i].Store(values[i]);
- }
- }
+ Update(0, has_value, values);
}
void RuntimeProfileBase::AveragedCounter::UpdateCounter(Counter* new_counter, int idx) {
@@ -1805,6 +2086,20 @@ void RuntimeProfileBase::AveragedCounter::UpdateCounter(Counter* new_counter, in
has_value_[idx].Store(true);
}
+void RuntimeProfileBase::AveragedCounter::Update(int start_idx,
+ const vector<bool>& has_value, const vector<int64_t>& values) {
+ DCHECK_EQ(values.size(), has_value.size());
+ DCHECK_LE(has_value.size() + start_idx, num_values_);
+ for (int i = 0; i < values.size(); ++i) {
+ if (has_value[i]) {
+ values_[i + start_idx].Store(values[i]);
+ // Set has_value_ after the value is valid above so that readers won't
+ // see invalid values.
+ has_value_[i + start_idx].Store(true);
+ }
+ }
+}
+
int64_t RuntimeProfileBase::AveragedCounter::value() const {
return unit_ == TUnit::DOUBLE_VALUE ? ComputeMean<double>() : ComputeMean<int64_t>();
}
@@ -1842,7 +2137,8 @@ RuntimeProfileBase::AveragedCounter::GetStats() const {
if (!vals.empty()) {
sort(vals.begin(), vals.end());
result.num_vals = vals.size();
- result.mean = std::accumulate(vals.begin(), vals.end(), 0) / result.num_vals;
+ T sum = std::accumulate(vals.begin(), vals.end(), 0L);
+ result.mean = sum / result.num_vals;
result.min = vals[0];
result.max = vals.back();
int end_idx = vals.size() - 1;
@@ -1902,6 +2198,50 @@ void RuntimeProfileBase::AveragedCounter::PrettyPrintImpl(
}
}
+void RuntimeProfileBase::AveragedCounter::ToJson(Document& document, Value* val) const {
+ if (unit_ == TUnit::DOUBLE_VALUE) {
+ ToJsonImpl<double>(document, val);
+ } else {
+ ToJsonImpl<int64_t>(document, val);
+ }
+}
+
+template <typename T>
+void RuntimeProfileBase::AveragedCounter::ToJsonImpl(
+ Document& document, Value* val) const {
+ Value counter_json(kObjectType);
+ auto unit_itr = _TUnit_VALUES_TO_NAMES.find(unit_);
+ DCHECK(unit_itr != _TUnit_VALUES_TO_NAMES.end());
+ Value unit_json(unit_itr->second, document.GetAllocator());
+ counter_json.AddMember("unit", unit_json, document.GetAllocator());
+
+ Stats<T> stats = GetStats<T>();
+ if (!FLAGS_gen_experimental_profile || stats.num_vals == 1) {
+ // Generate the traditional single-value representation if we're using the older
+ // profile version, or if the detailed statistics would just be noise.
+ counter_json.AddMember("value", stats.mean, document.GetAllocator());
+ return;
+ }
+
+ // For counters with <> 1 values, show summary stats and the values.
+ counter_json.AddMember("count", stats.num_vals, document.GetAllocator());
+ if (stats.num_vals > 0) {
+ counter_json.AddMember("min", stats.min, document.GetAllocator());
+ counter_json.AddMember("max", stats.max, document.GetAllocator());
+ // Only include detailed statistics if they are meaningful
+ // (i.e. min and max are not the same).
+ if (stats.min != stats.max) {
+ counter_json.AddMember("mean", stats.mean, document.GetAllocator());
+ counter_json.AddMember("p50", stats.p50, document.GetAllocator());
+ counter_json.AddMember("p75", stats.p75, document.GetAllocator());
+ counter_json.AddMember("p90", stats.p90, document.GetAllocator());
+ counter_json.AddMember("p95", stats.p95, document.GetAllocator());
+ }
+ }
+ *val = counter_json;
+}
+
+
void RuntimeProfileBase::AveragedCounter::ToThrift(
const string& name, TAggCounter* tcounter) const {
tcounter->name = name;
@@ -2095,7 +2435,11 @@ void RuntimeProfileBase::Counter::PrettyPrint(
void RuntimeProfileBase::Counter::ToJson(Document& document, Value* val) const {
Value counter_json(kObjectType);
- counter_json.AddMember("value", value(), document.GetAllocator());
+ if (unit_ == TUnit::DOUBLE_VALUE) {
+ counter_json.AddMember("value", double_value(), document.GetAllocator());
+ } else {
+ counter_json.AddMember("value", value(), document.GetAllocator());
+ }
auto unit_itr = _TUnit_VALUES_TO_NAMES.find(unit_);
DCHECK(unit_itr != _TUnit_VALUES_TO_NAMES.end());
Value unit_json(unit_itr->second, document.GetAllocator());
@@ -2179,42 +2523,9 @@ void AggregatedRuntimeProfile::InitFromThrift(
if (node.aggregated.__isset.input_profiles) {
input_profile_names_ = node.aggregated.input_profiles;
}
- for (const TAggCounter& counter : node.aggregated.counters) {
- counter_map_[counter.name] =
- pool->Add(new AveragedCounter(counter.unit, counter.has_value, counter.values));
- }
- DCHECK(node.aggregated.__isset.info_strings);
- for (const auto& entry : node.aggregated.info_strings) {
- vector<string>& per_instance_vals = agg_info_strings_[entry.first];
- if (per_instance_vals.empty()) {
- per_instance_vals.resize(num_input_profiles_);
- }
- for (const auto& distinct_val_entry : entry.second) {
- for (int32_t idx : distinct_val_entry.second) {
- per_instance_vals[idx] = distinct_val_entry.first;
- }
- }
- }
- if (node.aggregated.__isset.summary_stats_counters) {
- for (const TAggSummaryStatsCounter& val : node.aggregated.summary_stats_counters) {
- DCHECK_GT(num_input_profiles_, 0);
- DCHECK_EQ(val.sum.size(), num_input_profiles_);
- DCHECK_EQ(val.total_num_values.size(), num_input_profiles_);
- DCHECK_EQ(val.min_value.size(), num_input_profiles_);
- DCHECK_EQ(val.max_value.size(), num_input_profiles_);
- auto& entry = summary_stats_map_[val.name];
- entry.first = val.unit;
- vector<SummaryStatsCounter*>& instance_counters = entry.second;
- instance_counters.resize(num_input_profiles_);
- for (int idx = 0; idx < num_input_profiles_; ++idx) {
- if (!val.has_value[idx]) continue;
- instance_counters[idx] =
- pool->Add(new SummaryStatsCounter(val.unit, val.total_num_values[idx],
- val.min_value[idx], val.max_value[idx], val.sum[idx]));
- }
- }
- }
+ // Most of the logic can be shared with the update code path.
+ UpdateFromInstances(node, 0, pool);
}
// Print a sorted vector of indices in compressed form with subsequent indices
@@ -2239,6 +2550,7 @@ static void PrettyPrintIndexRanges(ostream* s, const vector<int32_t>& indices) {
stream << start_idx;
if (start_idx < prev_idx) stream << "-" << prev_idx;
}
+
void AggregatedRuntimeProfile::PrettyPrintInfoStrings(
ostream* s, const string& prefix) const {
// Aggregated info strings are only shown in experimental profile.
@@ -2247,7 +2559,7 @@ void AggregatedRuntimeProfile::PrettyPrintInfoStrings(
{
lock_guard<SpinLock> l(input_profile_name_lock_);
if (!input_profile_names_.empty()) {
- // TODO: IMPALA-9382: improve pretty-printing here
+ // TODO: IMPALA-9846: improve pretty-printing here
stream << prefix
<< "Instances: " << boost::algorithm::join(input_profile_names_, ", ")
<< endl;
@@ -2273,6 +2585,53 @@ void AggregatedRuntimeProfile::PrettyPrintSubclassCounters(
// Hide aggregated state when we are not using the transposed profile
// format.
if (!FLAGS_gen_experimental_profile) return;
+ ostream& stream = *s;
+ {
+ lock_guard<SpinLock> l(counter_map_lock_);
+ for (const auto& v : time_series_counter_map_) {
+ const TAggTimeSeriesCounter& counter = v.second;
+ for (int idx = 0; idx < counter.values.size(); ++idx) {
+ const vector<int64_t>& values = counter.values[idx];
+ const string& label = Substitute("$0[$1]", counter.name, idx);
+ PrettyPrintTimeSeries(label, values.data(), values.size(),
+ counter.period_ms[idx], prefix, counter.unit, s);
+ }
+ }
+ }
+ {
+ // Print all the event timers as the following:
+ // <EventKey>[instance index] Timeline: 2s719ms
+ // - Event 1: 6.522us (6.522us)
+ // - Event 2: 2s288ms (2s288ms)
+ // - Event 3: 2s410ms (121.138ms)
+ // The times in parentheses are the time elapsed since the last event.
+ lock_guard<SpinLock> l(event_sequence_lock_);
+ for (const auto& entry : event_sequence_map_) {
+ const AggEventSequence& seq = entry.second;
+ vector<const string*> labels(seq.labels.size());
+ for (const auto& lab_entry : seq.labels) {
+ labels[lab_entry.second] = &lab_entry.first;
+ }
+ for (int idx = 0; idx < seq.label_idxs.size(); ++idx) {
+ DCHECK_EQ(seq.label_idxs[idx].size(), seq.timestamps[idx].size());
+ int64_t last = seq.timestamps[idx].empty() ? 0 : seq.timestamps[idx].back();
+ stream << prefix << " " << entry.first << "[" << idx << "]: "
+ << PrettyPrinter::Print(last, TUnit::TIME_NS)
+ << endl;
+ int64_t prev = 0L;
+ for (int j = 0; j < seq.label_idxs[idx].size(); ++j) {
+ const string& label = *labels[seq.label_idxs[idx][j]];
+ int64_t ts = seq.timestamps[idx][j];
+ stream << prefix << " - " << label << ": "
+ << PrettyPrinter::Print(ts, TUnit::TIME_NS) << " ("
+ << PrettyPrinter::Print(ts - prev, TUnit::TIME_NS) << ")"
+ << endl;
+ prev = ts;
+ }
+ }
+ }
+ }
+
{
lock_guard<SpinLock> l(summary_stats_map_lock_);
for (const auto& v : summary_stats_map_) {
@@ -2322,6 +2681,19 @@ void AggregatedRuntimeProfile::ToThriftSubclass(
avg_counter->ToThrift(entry.first, &out_node->aggregated.counters.back());
}
+ out_node->aggregated.__isset.time_series_counters = true;
+ {
+ lock_guard<SpinLock> l(counter_map_lock_);
+ out_node->aggregated.time_series_counters.reserve(time_series_counter_map_.size());
+ for (const auto& entry : time_series_counter_map_) {
+ out_node->aggregated.time_series_counters.push_back(entry.second);
+ DCHECK_EQ(out_node->aggregated.time_series_counters.back().values.size(),
+ out_node->aggregated.time_series_counters.back().period_ms.size());
+ DCHECK_EQ(out_node->aggregated.time_series_counters.back().values.size(),
+ out_node->aggregated.time_series_counters.back().start_index.size());
+ }
+ }
+
out_node->aggregated.__isset.info_strings = true;
{
lock_guard<SpinLock> l(agg_info_strings_lock_);
@@ -2331,6 +2703,32 @@ void AggregatedRuntimeProfile::ToThriftSubclass(
}
}
+ out_node->aggregated.__isset.event_sequences = true;
+ {
+ lock_guard<SpinLock> l(event_sequence_lock_);
+ for (const auto& entry : event_sequence_map_) {
+ const AggEventSequence& seq = entry.second;
+ out_node->aggregated.event_sequences.emplace_back();
+ TAggEventSequence& tseq = out_node->aggregated.event_sequences.back();
+ tseq.name = entry.first;
+
+ tseq.label_dict.resize(seq.labels.size());
+ for (const auto& lab_entry : seq.labels) {
+ tseq.label_dict[lab_entry.second] = lab_entry.first;
+ }
+ DCHECK_EQ(seq.label_idxs.size(), seq.timestamps.size());
+ tseq.label_idxs.resize(seq.label_idxs.size());
+ tseq.timestamps.resize(seq.timestamps.size());
+ for (int idx = 0; idx < seq.label_idxs.size(); ++idx) {
+ DCHECK_EQ(seq.label_idxs[idx].size(), seq.timestamps[idx].size());
+ for (int j = 0; j < seq.label_idxs[idx].size(); ++j) {
+ tseq.label_idxs[idx].push_back(seq.label_idxs[idx][j]);
+ tseq.timestamps[idx].push_back(seq.timestamps[idx][j]);
+ }
+ }
+ }
+ }
+
out_node->__isset.summary_stats_counters = true;
{
lock_guard<SpinLock> l(summary_stats_map_lock_);
@@ -2354,7 +2752,37 @@ void AggregatedRuntimeProfile::ToThriftSubclass(
}
void AggregatedRuntimeProfile::ToJsonSubclass(Value* parent, Document* d) const {
- // TODO: IMPALA-9382: add Json aggregated profile.
+ Document::AllocatorType& allocator = d->GetAllocator();
+ // We do not include all of the per-instance values in the JSON representation. We do
+ // not need to be able to round-trip RuntimeProfile to/from the JSON representation so
+ // we can afford to discard some less-valuable information.
+ //
+ // The following information is omitted deliberately:
+ // * Time series counters from time_series_counter_map_ are not included - they would
+ // simply be too large for queries with 100s or 1000s of instances.
+ // * Event sequences from event_sequence_map_ are not included for the same reason.
+ // * Per-instance summary stat values are omitted (although we include the aggregates).
+ // * Per-instance values for regular counters.
+
+ // SummaryStatsCounter
+ // Only include the aggregated summary stats, not the per-instance values to avoid
+ // blowing up profile size.
+ {
+ lock_guard<SpinLock> l(summary_stats_map_lock_);
+ if (!summary_stats_map_.empty()) {
+ Value summary_stats_counters_json(kArrayType);
+ for (const auto& v : summary_stats_map_) {
+ SummaryStatsCounter aggregated_stats(v.second.first);
+ AggregateSummaryStats(v.second.second, &aggregated_stats);
+ Value summary_stats_counter(kObjectType);
+ Value summary_name_json(v.first.c_str(), v.first.size(), allocator);
+ aggregated_stats.ToJson(*d, &summary_stats_counter);
+ summary_stats_counter.AddMember("counter_name", summary_name_json, allocator);
+ summary_stats_counters_json.PushBack(summary_stats_counter, allocator);
+ }
+ parent->AddMember("summary_stats_counters", summary_stats_counters_json, allocator);
+ }
+ }
}
map<string, vector<int32_t>> AggregatedRuntimeProfile::GroupDistinctInfoStrings(
diff --git a/be/src/util/runtime-profile.h b/be/src/util/runtime-profile.h
index 6c061b6..67774d5 100644
--- a/be/src/util/runtime-profile.h
+++ b/be/src/util/runtime-profile.h
@@ -18,11 +18,13 @@
#pragma once
#include <iosfwd>
+#include <mutex>
#include <boost/function.hpp>
#include <rapidjson/document.h>
#include "common/atomic.h"
#include "common/status.h"
+#include "testutil/gtest-util.h"
#include "util/spinlock.h"
#include "gen-cpp/RuntimeProfile_types.h"
@@ -167,10 +169,17 @@ class RuntimeProfileBase {
/// Adds a string to the runtime profile. If a value already exists for 'key',
/// the value will be updated.
- /// TODO: IMPALA-9382: this can be moved to RuntimeProfile once we remove callsites for
+ /// TODO: IMPALA-9846: this can be moved to RuntimeProfile once we remove callsites for
/// this function on AggregatedRuntimeProfile.
void AddInfoString(const std::string& key, const std::string& value);
+ /// Serializes an entire runtime profile tree to thrift. Can be overridden by subclasses
+ /// to add additional info. The thrift representation is the public-facing
+ /// representation if this is a RuntimeProfile that is the root of the whole query's
+ /// tree. Otherwise this generates a serialized profile for internal use.
+ /// Does not hold locks when it makes any function calls.
+ virtual void ToThrift(TRuntimeProfileTree* tree) const;
+
/// Returns name of this profile
const std::string& name() const { return name_; }
const TRuntimeProfileNodeMetadata& metadata() const { return metadata_; }
@@ -200,6 +209,8 @@ class RuntimeProfileBase {
void AddSkewInfo(RuntimeProfileBase* root, double threshold);
protected:
+ FRIEND_TEST(CountersTest, AggregateEventSequences);
+
/// Name of the counter maintaining the total time.
static const std::string TOTAL_TIME_COUNTER_NAME;
static const std::string LOCAL_TIME_COUNTER_NAME;
@@ -239,7 +250,7 @@ class RuntimeProfileBase {
/// acquiring 'counter_map_lock_'.
Counter* const inactive_timer_;
- /// TODO: IMPALA-9382: info strings can be moved to RuntimeProfile once we remove
+ /// TODO: IMPALA-9846: info strings can be moved to RuntimeProfile once we remove
/// callsites for this function on AggregatedRuntimeProfile.
typedef std::map<std::string, std::string> InfoStrings;
InfoStrings info_strings_;
@@ -289,6 +300,27 @@ class RuntimeProfileBase {
void AddChildLocked(
RuntimeProfileBase* child, bool indent, ChildVector::iterator insert_pos);
+ /// Helper for the various Update*() methods that finds an existing child with 'name'
+ /// or creates it by calling 'create_fn', which is a zero-argument function that returns
+ /// a pointer to a subclass of RuntimeProfileBase.
+ ///
+ /// If the child is created, it is inserted at 'insert_pos'. In either case 'insert_pos'
+ /// is advanced to after the returned child. When callers add children in sequence, this
+ /// is sufficient to preserve the order of children if children are added after the
+ /// first Update*() call (IMPALA-6694). E.g. if the first update sends [B, D]
+ /// and the second update sends [A, B, C, D], then this code makes sure that children_
+ /// is [A, B, C, D] afterwards.
+ ///
+ /// 'children_lock_' must be held by the caller.
+ template <typename CreateFn>
+ RuntimeProfileBase* AddOrCreateChild(const std::string& name,
+ ChildVector::iterator* insert_pos, CreateFn create_fn, bool indent);
+
+ /// Update 'child_counters_map_' to add the entries in 'src'.
+ /// Call must hold 'counter_map_lock_' via 'lock'.
+ void UpdateChildCountersLocked(const std::unique_lock<SpinLock>& lock,
+ const ChildCounterMap& src);
+
/// Clear all chunked time series counters in this profile and all children.
virtual void ClearChunkedTimeSeriesCounters();
@@ -356,7 +388,7 @@ class RuntimeProfileBase {
/// implements AddInfoString(), otherwise implements AppendInfoString().
/// Redaction rules are applied on the info string if 'redact' is true.
/// Trailing whitespace is removed.
- /// TODO: IMPALA-9382: this can be moved to RuntimeProfile once we remove callsites for
+ /// TODO: IMPALA-9846: this can be moved to RuntimeProfile once we remove callsites for
/// this function on AggregatedRuntimeProfile.
void AddInfoStringInternal(
const std::string& key, std::string value, bool append, bool redact = false);
@@ -497,7 +529,8 @@ class RuntimeProfile : public RuntimeProfileBase {
/// Creates and returns a new EventSequence (owned by the runtime
/// profile) - unless a timer with the same 'key' already exists, in
/// which case it is returned.
- /// TODO: EventSequences are not merged by Merge() or Update()
+ /// TODO: EventSequences are not merged by Update()
+ /// TODO: is this comment out of date?
EventSequence* AddEventSequence(const std::string& key);
EventSequence* AddEventSequence(const std::string& key, const TEventSequence& from);
EventSequence* GetEventSequence(const std::string& name) const;
@@ -603,11 +636,7 @@ class RuntimeProfile : public RuntimeProfileBase {
/// Clear all chunked time series counters in this profile and all children.
void ClearChunkedTimeSeriesCounters() override;
- /// Serializes an entire runtime profile tree to thrift.
- /// This is defined in RuntimeProfile instead of RuntimeProfileBase becase we require
- /// that a runtime profile root be a RuntimeProfile.
- /// Does not hold locks when it makes any function calls.
- void ToThrift(TRuntimeProfileTree* tree) const;
+ void ToThrift(TRuntimeProfileTree* tree) const override;
/// Store an entire runtime profile tree into JSON document 'd'.
void ToJson(rapidjson::Document* d) const;
@@ -749,20 +778,28 @@ class AggregatedRuntimeProfile : public RuntimeProfileBase {
/// from other profiles.
///
/// 'is_root' must be true if this is the root of the averaged profile tree which
- /// will have Update() called on it. The descendants of the root are created and
- /// updated via Update().
+ /// will have UpdateAggregated*() called on it. The descendants of the root are
+ /// created and updated by that call.
static AggregatedRuntimeProfile* Create(ObjectPool* pool, const std::string& name,
int num_input_profiles, bool is_root = true);
/// Updates the AveragedCounter counters in this profile with the counters from the
- /// 'src' profile. If a counter is present in 'src' but missing in this profile, a new
- /// AveragedCounter is created with the same name. Obtains locks on the counter maps
- /// and child counter maps in both this and 'src' profiles.
- /// TODO: IMPALA-9382: update method name and comment.
+ /// 'src' profile, which is the input instance that was assigned index 'idx'. If a
+ /// counter is present in 'src' but missing in this profile, a new AveragedCounter
+ /// is created with the same name. Obtains locks on the counter maps and child
+ /// counter maps in both this and 'src' profiles.
///
/// Note that 'src' must be all instances of RuntimeProfile - no
/// AggregatedRuntimeProfiles can be part of the input.
- void Update(RuntimeProfile* src, int idx);
+ void UpdateAggregatedFromInstance(RuntimeProfile* src, int idx);
+
+ /// Updates the AveragedCounter counters in this profile with the counters from the
+ /// 'src' profile, which must be a serialized AggregatedProfile. The instances in
+ /// 'src' must correspond to instances [start_idx, start_idx + # src instances) in
+ /// this profile. If a counter is present in 'src' but missing in this profile, a
+ /// new AveragedCounter is created with the same name. Obtains locks on the counter
+ /// maps and child counter maps in this profile.
+ void UpdateAggregatedFromInstances(const TRuntimeProfileTree& src, int start_idx);
protected:
virtual int GetNumInputProfiles() const override { return num_input_profiles_; }
@@ -814,6 +851,27 @@ class AggregatedRuntimeProfile : public RuntimeProfileBase {
/// Protects summary_stats_map_.
mutable SpinLock summary_stats_map_lock_;
+ struct AggEventSequence {
+ // Unique labels in the event sequence. Values are [0, labels.size() - 1]
+ std::map<std::string, int32_t> labels;
+
+ // One entry per instance. Each entry contains the label indices for that instance's
+ // event sequence.
+ std::vector<std::vector<int32_t>> label_idxs;
+
+ // One entry per instance. Each entry contains the timestamps for that instance's
+ // event sequence.
+ std::vector<std::vector<int64_t>> timestamps;
+ };
+
+ std::map<std::string, AggEventSequence> event_sequence_map_;
+
+ /// Protects event_sequence_map_ and event_sequence_labels_.
+ mutable SpinLock event_sequence_lock_;
+
+ /// Time series counters. Protected by 'counter_map_lock_'.
+ std::map<std::string, TAggTimeSeriesCounter> time_series_counter_map_;
+
AggregatedRuntimeProfile(
ObjectPool* pool, const std::string& name, int num_input_profiles, bool is_root);
@@ -822,8 +880,32 @@ class AggregatedRuntimeProfile : public RuntimeProfileBase {
std::map<std::string, std::vector<int32_t>> GroupDistinctInfoStrings(
const std::vector<std::string>& info_string_values) const;
- /// Helper for Update() that is invoked recursively on 'src'.
- void UpdateRecursive(RuntimeProfile* src, int idx);
+ /// Helper for UpdateAggregatedFromInstance() that are invoked recursively on 'src'.
+ void UpdateAggregatedFromInstanceRecursive(RuntimeProfile* src, int idx);
+
+ /// Helpers for UpdateAggregatedFromInstanceRecursive() that update particular parts
+ /// of this profile node from 'src'.
+ void UpdateCountersFromInstance(RuntimeProfile* src, int idx);
+ void UpdateInfoStringsFromInstance(RuntimeProfile* src, int idx);
+ void UpdateSummaryStatsFromInstance(RuntimeProfile* src, int idx);
+ void UpdateEventSequencesFromInstance(RuntimeProfile* src, int idx);
+
+ /// Helper for UpdateAggregatedFromInstances() that are invoked recursively on 'src'.
+ void UpdateAggregatedFromInstancesRecursive(
+ const TRuntimeProfileTree& src, int* node_idx, int start_idx);
+
+ /// Helpers for UpdateAggregatedFromInstancesRecursive() that update particular parts
+ /// of this profile node from 'src'. 'src' must have an aggregated profile set.
+ void UpdateFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool);
+ void UpdateCountersFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool);
+ void UpdateInfoStringsFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool);
+ void UpdateSummaryStatsFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool);
+ void UpdateEventSequencesFromInstances(
+ const TRuntimeProfileNode& node, int start_idx, ObjectPool* pool);
/// Aggregate summary stats into a single counter. Entries in
/// 'counter' may be NULL.
diff --git a/common/protobuf/control_service.proto b/common/protobuf/control_service.proto
index d8483ff..13e6008 100644
--- a/common/protobuf/control_service.proto
+++ b/common/protobuf/control_service.proto
@@ -194,6 +194,16 @@ message FragmentInstanceExecStatusPB {
repeated ExecSummaryDataPB exec_summary_data = 7;
}
+message FragmentExecStatusPB {
+ // Ordinal number of fragment in the query.
+ optional int32 fragment_idx = 1;
+
+ // The index of the first instance included in this state report. The serialized
+ // profile (in a corresponding sidecar) includes the number of instances and the
+ // profiles for those instances. The instance indexes for a backend are sequential.
+ optional int32 min_per_fragment_instance_idx = 2;
+}
+
message ReportExecStatusRequestPB {
// The query id which this report is for.
optional UniqueIdPB query_id = 1;
@@ -203,8 +213,8 @@ message ReportExecStatusRequestPB {
repeated FragmentInstanceExecStatusPB instance_exec_status = 3;
- // Sidecar index of the cumulative profiles of all fragment instances
- // in instance_exec_status.
+ // Sidecar index of the cumulative profiles in instance_exec_status and
+ // fragment_exec_status
optional int32 thrift_profiles_sidecar_idx = 4;
// Cumulative status for this backend.
@@ -237,6 +247,15 @@ message ReportExecStatusRequestPB {
// Total bytes sent by instances that contained a scan node.
optional int64 scan_bytes_sent = 13;
+ // Aggregated status for all instances of a fragment on a backend. Used when
+ // TQueryCtx.gen_aggregated_profile is true.
+ repeated FragmentExecStatusPB fragment_exec_status = 14;
+
+ // Sequence number prevents out-of-order or duplicated updates from being applied.
+ // Incremented for each backend status report. 'fragment_exec_status' should be
+ // disregarded if the sequence number of this report is less than or equal to
+ // the previous report received.
+ optional int64 backend_report_seq_no = 15;
}
message ReportExecStatusResponsePB {
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index 9f7aade..51ee39f 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -636,6 +636,10 @@ struct TQueryCtx {
// The initiating coordinator's backend_id.
27: optional Types.TUniqueId coord_backend_id
+
+ // True if the new runtime profile format added by IMPALA-9382 should be generated
+ // by this query.
+ 28: optional bool gen_aggregated_profile
}
// Descriptor that indicates that a runtime filter is produced by a plan node.
diff --git a/common/thrift/RuntimeProfile.thrift b/common/thrift/RuntimeProfile.thrift
index ea63b9f..02dd1e7 100644
--- a/common/thrift/RuntimeProfile.thrift
+++ b/common/thrift/RuntimeProfile.thrift
@@ -74,6 +74,19 @@ struct TEventSequence {
3: required list<string> labels
}
+// Aggregated version of TEventSequence.
+struct TAggEventSequence {
+ 1: required string name
+ // One entry per unique label.
+ 2: required list<string> label_dict
+
+ // The below lists have one entry per instance.
+
+ // The label of each event, represented as the index in label_dict.
+ 3: required list<list<i32>> label_idxs
+ 4: required list<list<i64>> timestamps
+}
+
// Struct to contain data sampled at even time intervals (e.g. ram usage every
// N seconds).
// values[0] represents the value when the counter stated (e.g. fragment started)
@@ -97,6 +110,20 @@ struct TTimeSeriesCounter {
5: optional i64 start_index
}
+// Aggregated version of TTimeSeriesCounter
+struct TAggTimeSeriesCounter {
+ 1: required string name
+ 2: required Metrics.TUnit unit
+
+ // The below lists have one entry per instance, with each list entry containing
+ // the equivalent values to the similarly-named field in TTimeSeriesCounter.
+ 3: required list<i32> period_ms
+
+ 4: required list<list<i64>> values
+
+ 5: required list<i64> start_index
+}
+
// Thrift version of RuntimeProfile::SummaryStatsCounter.
struct TSummaryStatsCounter {
1: required string name
@@ -172,11 +199,11 @@ struct TAggregatedRuntimeProfileNode {
// List of summary stats counters.
5: optional list<TAggSummaryStatsCounter> summary_stats_counters
- // TODO: IMPALA-9382: decide on representation. This is a placeholder
- // 6: optional list<list<TEventSequence>> event_sequences
+ // List of event sequences.
+ 6: optional list<TAggEventSequence> event_sequences
- // TODO: IMPALA-9382: decide on representation. This is a placeholder
- // 7: optional list<list<TTimeSeriesCounter>> time_series_counters
+ // List of time series counters.
+ 7: optional list<TAggTimeSeriesCounter> time_series_counters
}
@@ -253,6 +280,7 @@ struct TRuntimeProfileNode {
// Different from version 1, there is only a single aggregated profile tree for each
// fragment. All nodes in this tree use the aggregated profile representation. Otherwise
// the structure of the profile is the same.
+// TODO: IMPALA-9382 - update when non-experimental
struct TRuntimeProfileTree {
1: required list<TRuntimeProfileNode> nodes
2: optional ExecStats.TExecSummary exec_summary
@@ -263,7 +291,7 @@ struct TRuntimeProfileTree {
// information.
// Version 1: this field is unset
// Version 2: this field is set to 2
- // TODO: IMPALA-9382: document which versions of Impala generate which version.
+ // TODO: IMPALA-9846: document which versions of Impala generate which version.
3: optional i32 profile_version
}
diff --git a/testdata/workloads/tpch/queries/runtime-profile-aggregated.test b/testdata/workloads/tpch/queries/runtime-profile-aggregated.test
new file mode 100644
index 0000000..a3d4646
--- /dev/null
+++ b/testdata/workloads/tpch/queries/runtime-profile-aggregated.test
@@ -0,0 +1,29 @@
+====
+---- QUERY
+# Sanity check aggregated runtime profile. Pick a query that runs for long enough for
+# time series counters, etc to collect some data.
+select STRAIGHT_JOIN count(distinct l_partkey)
+from lineitem join [BROADCAST] supplier on l_suppkey = s_suppkey
+---- RESULTS
+200000
+---- RUNTIME_PROFILE
+# Top-level counter (not aggregated)
+row_regex: .*NumBackends: 4 .*
+# Fragment headers.
+row_regex: .*Coordinator Fragment F03.*
+row_regex: .*Fragment F00 \[4 instances\].*
+# Coordinator counter - single instance.
+row_regex: .*RowsSent: 1.*
+# Scan node counter - many instances.
+row_regex: .*BytesRead: mean=.* min=.* p50=.* max=.*
+# Info string from aggregated scan node.
+row_regex: .*Table Name\[0-3\]: tpch.*lineitem.*
+# Timeline from aggregated scan node
+row_regex: .*Node Lifecycle Event Timeline\[.*\]: .*
+# Summary stats counter from aggregated scan node
+row_regex: .*InitialRangeActualReservation: .*Avg:.*Min.*Max.*
+# Time series form aggregated scan node.
+row_regex: .*BytesReadSeries\[.*\] (.*): .*
+# Fragment-level codegen profile should be present.
+row_regex: .*CodegenTotalWallClockTime: .*
+====
diff --git a/tests/custom_cluster/test_runtime_profile.py b/tests/custom_cluster/test_runtime_profile.py
new file mode 100644
index 0000000..73d7b50
--- /dev/null
+++ b/tests/custom_cluster/test_runtime_profile.py
@@ -0,0 +1,46 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+
+
+class TestRuntimeProfile(CustomClusterTestSuite):
+ """Tests for non-default runtime profile behaviour."""
+
+ @classmethod
+ def get_workload(self):
+ return 'tpch'
+
+ PERIODIC_COUNTER_UPDATE_FLAG = '--periodic_counter_update_period_ms=50'
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args('--gen_experimental_profile=true ' +
+ PERIODIC_COUNTER_UPDATE_FLAG)
+ def test_runtime_profile_aggregated(self, vector):
+ """Sanity-check the new experimental profile format.
+ --periodic_counter_update_period_ms is decreased to get more samples for time series
+ counters."""
+ # Add an executor with the flag disabled. The aggregated profile should still be
+ # generated correctly
+ self._start_impala_cluster(options=['--impalad_args=' +
+ self.PERIODIC_COUNTER_UPDATE_FLAG],
+ cluster_size=1,
+ num_coordinators=0,
+ add_executors=True,
+ expected_num_executors=4)
+ self.run_test_case('runtime-profile-aggregated', vector)