You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2026/05/12 03:51:10 UTC
(impala) 01/02: IMPALA-14847: Add diagnostics for large ExecQueryFInstances RPCs
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 28a4d3861dba19dfcc0f9c5f655b5ff520dd56e7
Author: Joe McDonnell <jo...@cloudera.com>
AuthorDate: Tue Apr 28 14:15:00 2026 -0700
IMPALA-14847: Add diagnostics for large ExecQueryFInstances RPCs
For some complex queries, the ExecQueryFInstances RPCs can
be very large, sometimes hundreds of MBs. This can cause
high utilization of a coordinator's NIC and slow other
queries. This adds a few different diagnostics to help
track down the queries that have large ExecQueryFInstances
RPCs:
1. It prints a warning when a query has ExecQueryFInstances RPCs
that exceed a threshold (set by --large_execquery_rpc_threshold_bytes,
which defaults to 25MB). The warning includes a query level
summary of the number of RPCs that exceeded the threshold and
the average size of the ExecQueryFInstances RPCs for this query.
It also includes information about the size of the shared TQueryCtx
piece of the message.
It also prints information about the top three RPCs to backends
for this query. This contains detail about the sizes of
backend-specific pieces of the message as well as the number
of fragments and scan ranges.
Here are some outputs for different scenarios (with a 512KB threshold):
TPC-DS Q67 mt_dop=4:
ExecQueryFInstances RPCs for 3 out of 3 exceed the warning threshold
of 512.00 KB. Average RPC size: 534.64 KB Shared TQueryCtx: 393.91 KB
(390.82 KB from descriptor table)
Additional information about the 3 largest RPCs:
Destination: joemcdonnell-22743:27000 Total size: 536.98 KB
Backend-specific size: 143.08 KB TExecPlanFragmentInfo: 77.81 KB
ExecQueryFInstancesRequestPB: 65.27 KB num fragments: 10
num fragment instances: 19 num scan ranges: 611
...
Insert with large VALUES clause (about 166KB of strings inserted)
W20260501 11:44:23.116255 3332252 coordinator-backend-state.cc:367]
ExecQueryFInstances RPCs for 1 out of 1 exceed the warning threshold
of 512.00 KB. Average RPC size: 632.54 KB Shared TQueryCtx: 372.08 KB
(633.00 B from descriptor table)
Additional information about the 1 largest RPCs:
Destination: joemcdonnell-22743:27000 Total size: 632.54 KB
Backend-specific size: 260.46 KB TExecPlanFragmentInfo: 260.43 KB
ExecQueryFInstancesRequestPB: 29.00 B num fragments: 1
num fragment instances: 1 num scan ranges: 0
Query with lots of expressions from query_test/test_exprs.py TestExprLimit
ExecQueryFInstances RPCs for 3 out of 3 exceed the warning threshold
of 512.00 KB. Average RPC size: 13.79 MB Shared TQueryCtx: 3.71 MB
(7.84 KB from descriptor table)
Additional information about the 3 largest RPCs:
Destination: joemcdonnell-22743:27000 Total size: 33.94 MB
Backend-specific size: 30.22 MB TExecPlanFragmentInfo: 30.22 MB
ExecQueryFInstancesRequestPB: 965.00 B num fragments: 2
num fragment instances: 2 num scan ranges: 8
...
2. It adds a ExecQueryRPCSizes summary stats counter to the
profile. This provides output like:
ExecQueryRPCSizes: (Avg: 534.65 KB (547486) ; Min: 533.48 KB (546281) ;
Max: 537.00 KB (549890) ; Sum: 1.57 MB (1642460) ; Number of samples: 3)
3. This also includes the RPC size in the JSON used for
the WebUI's /query_backends page.
Change-Id: I10e1fd50bf2fec2b8969585f7fcec2ca2487e181
Reviewed-on: http://gerrit.cloudera.org:8080/24248
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Reviewed-by: Michael Smith <mi...@cloudera.com>
Tested-by: Joe McDonnell <jo...@cloudera.com>
---
be/src/rpc/sidecar-util.h | 6 ++-
be/src/runtime/coordinator-backend-state.cc | 65 +++++++++++++++++++++++---
be/src/runtime/coordinator-backend-state.h | 16 +++++--
be/src/runtime/coordinator.cc | 62 +++++++++++++++++++++++-
be/src/runtime/coordinator.h | 55 ++++++++++++++++++++++
be/src/scheduling/admission-control-service.cc | 4 +-
tests/webserver/test_web_pages.py | 2 +-
7 files changed, 196 insertions(+), 14 deletions(-)
diff --git a/be/src/rpc/sidecar-util.h b/be/src/rpc/sidecar-util.h
index 35f022afe..d43caca49 100644
--- a/be/src/rpc/sidecar-util.h
+++ b/be/src/rpc/sidecar-util.h
@@ -98,9 +98,10 @@ Status GetSidecar(int sidecar_idx, RPC* rpc, T* thrift_obj) {
// where it would be tricky to manually ensure that the sidecar's memory is released after
// the RPC has completed. 'rpc' may be either an RpcContext or an RpcController.
//
-// If successful, the sidecar idx is returned in 'sidecar_idx'.
+// If successful, the sidecar idx is returned in 'sidecar_idx' and 'length' is set to
+// the serialized size.
template <typename RPC, typename T>
-Status SetFaststringSidecar(const T& obj, RPC* rpc, int* sidecar_idx) {
+Status SetFaststringSidecar(const T& obj, RPC* rpc, int* sidecar_idx, int64_t* length) {
ThriftSerializer serializer(/* compact */ true);
uint8_t* serialized_buf = nullptr;
uint32_t serialized_len = 0;
@@ -114,6 +115,7 @@ Status SetFaststringSidecar(const T& obj, RPC* rpc, int* sidecar_idx) {
kudu::rpc::RpcSidecar::FromFaststring(std::move(sidecar_str));
RETURN_IF_ERROR(FromKuduStatus(
rpc->AddOutboundSidecar(move(rpc_sidecar), sidecar_idx), "Failed to add sidecar"));
+ *length = serialized_len;
return Status::OK();
}
diff --git a/be/src/runtime/coordinator-backend-state.cc b/be/src/runtime/coordinator-backend-state.cc
index 4d0388649..0f63ccfcf 100644
--- a/be/src/runtime/coordinator-backend-state.cc
+++ b/be/src/runtime/coordinator-backend-state.cc
@@ -61,6 +61,10 @@ DECLARE_bool(gen_experimental_profile);
DECLARE_int32(backend_client_rpc_timeout_ms);
DECLARE_int64(rpc_max_message_size);
+DEFINE_int64(large_execquery_rpc_threshold_bytes, 25 * 1024 * 1024,
+ "(Advanced) Threshold for considering an ExecQueryFInstances RPC to be unusually "
+ "large. Large RPCs trigger additional logging and other diagnostics.");
+
namespace impala {
PROFILE_DEFINE_COUNTER(BytesAssigned, STABLE_HIGH, TUnit::BYTES,
"Total number of bytes of filesystem scan ranges assigned to this fragment "
@@ -127,7 +131,8 @@ void Coordinator::BackendState::Init(const vector<FragmentStats*>& fragment_stat
void Coordinator::BackendState::SetRpcParams(const DebugOptions& debug_options,
const FilterRoutingTable& filter_routing_table, ExecQueryFInstancesRequestPB* request,
- TExecPlanFragmentInfo* fragment_info) {
+ TExecPlanFragmentInfo* fragment_info, int64_t* total_scan_ranges) {
+ *total_scan_ranges = 0;
request->set_coord_state_idx(state_idx_);
request->set_min_mem_reservation_bytes(
backend_exec_params_.min_mem_reservation_bytes());
@@ -166,6 +171,10 @@ void Coordinator::BackendState::SetRpcParams(const DebugOptions& debug_options,
UniqueIdPBToTUniqueId(params.instance_id(), &instance_ctx.fragment_instance_id);
instance_ctx.per_fragment_instance_idx = params.per_fragment_instance_idx();
*instance_ctx_pb->mutable_per_node_scan_ranges() = params.per_node_scan_ranges();
+ // Sum the scan ranges over all the fragments
+ for (const auto& pair : params.per_node_scan_ranges()) {
+ *total_scan_ranges += pair.second.scan_ranges().size();
+ }
for (const auto& entry : fragment_exec_params.per_exch_num_senders()) {
instance_ctx.per_exch_num_senders[entry.first] = entry.second;
}
@@ -281,7 +290,8 @@ done:
void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
const FilterRoutingTable& filter_routing_table,
const kudu::Slice& serialized_query_ctx,
- TypedCountingBarrier<Status>* exec_status_barrier) {
+ TypedCountingBarrier<Status>* exec_status_barrier,
+ Coordinator::ExecQueryRpcStats* execquery_rpc_stats) {
{
lock_guard<mutex> l(lock_);
DCHECK(!exec_done_);
@@ -305,7 +315,9 @@ void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
ExecQueryFInstancesRequestPB request;
TExecPlanFragmentInfo fragment_info;
- SetRpcParams(debug_options, filter_routing_table, &request, &fragment_info);
+ int64_t total_scan_ranges;
+ SetRpcParams(debug_options, filter_routing_table, &request, &fragment_info,
+ &total_scan_ranges);
exec_rpc_controller_.set_timeout(
MonoDelta::FromMilliseconds(FLAGS_backend_client_rpc_timeout_ms));
@@ -319,9 +331,11 @@ void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
}
int sidecar_idx;
+ int64_t fragment_info_length = 0;
// TODO: eliminate the extra copy here by using a Slice
Status sidecar_status =
- SetFaststringSidecar(fragment_info, &exec_rpc_controller_, &sidecar_idx);
+ SetFaststringSidecar(fragment_info, &exec_rpc_controller_, &sidecar_idx,
+ &fragment_info_length);
if (UNLIKELY(!sidecar_status.ok())) {
SetExecError(sidecar_status, exec_status_barrier);
goto done;
@@ -344,9 +358,47 @@ void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
CopyFilepathToHostsMappingToRequest(&request);
+ // Three components to the size of the RPC:
+ // 1. serialized_query_ctx.size(): Serialized shared TQueryCtx sent for every message
+ // to every executor.
+ // 2. fragment_info_length: Serialized TExecPlanFragmentInfo for only this executor
+ // 3. request.ByteSizeLong(): Other fields on ExecQueryFInstancesRequestPB (including
+ // the scan ranges)
+ // This may be off by a few bytes due to KRPC headers, but it covers the
+ // Impala-specific content.
+ rpc_size_ = request.ByteSizeLong() + serialized_query_ctx.size() +
+ fragment_info_length;
+ if (rpc_size_ >= FLAGS_large_execquery_rpc_threshold_bytes) {
+ // If this exceeds the threshold, we report a warning through the ExecQueryRpcStats
+ // structure. That triggers logging of the shared TQueryCtx part of the
+ // message. It logs more detailed information about the top few messages.
+ // To help diagnosing the backend specific pieces, we need to break down
+ // the size of the different pieces of the message. The number of fragments and
+ // scan ranges can be important contributors to the size, so include that
+ // information as well.
+
+ stringstream warn;
+ warn << "Destination: " << NetworkAddressPBToString(impalad_address())
+ << " Total size: " << PrettyPrinter::Print(rpc_size_, TUnit::BYTES)
+ << " Backend-specific size: "
+ << PrettyPrinter::Print(rpc_size_ - serialized_query_ctx.size(), TUnit::BYTES)
+ << " TExecPlanFragmentInfo: "
+ << PrettyPrinter::Print(fragment_info_length, TUnit::BYTES)
+ << " ExecQueryFInstancesRequestPB: "
+ << PrettyPrinter::Print(request.ByteSizeLong(), TUnit::BYTES)
+ << " num fragments: " << fragment_info.fragments.size()
+ << " num fragment instances: "
+ << fragment_info.fragment_instance_ctxs.size()
+ << " num scan ranges: " << total_scan_ranges;
+ // stringstream::str() returns a copy, so warn.str() is an rvalue reference and
+ // doesn't need std::move().
+ execquery_rpc_stats->ReportRpcSizeWithWarning(rpc_size_, warn.str());
+ } else {
+ execquery_rpc_stats->ReportRpcSize(rpc_size_);
+ }
VLOG_FILE << "making rpc: ExecQueryFInstances"
- << " host=" << impalad_address() << " query_id=" << PrintId(query_id_);
-
+ << " host=" << impalad_address() << " query_id=" << PrintId(query_id_)
+ << " size=" << PrettyPrinter::Print(rpc_size_, TUnit::BYTES);
proxy->ExecQueryFInstancesAsync(request, &exec_response_, &exec_rpc_controller_,
std::bind(&Coordinator::BackendState::ExecCompleteCb, this, exec_status_barrier,
MonotonicMillis()));
@@ -1023,6 +1075,7 @@ void Coordinator::BackendState::ToJson(Value* value, Document* document) {
value->AddMember("host", val, document->GetAllocator());
value->AddMember("rpc_latency", rpc_latency(), document->GetAllocator());
+ value->AddMember("rpc_size", rpc_size(), document->GetAllocator());
value->AddMember("time_since_last_heard_from", MonotonicMillis() - last_report_time_ms_,
document->GetAllocator());
diff --git a/be/src/runtime/coordinator-backend-state.h b/be/src/runtime/coordinator-backend-state.h
index a8755954b..aae96a6df 100644
--- a/be/src/runtime/coordinator-backend-state.h
+++ b/be/src/runtime/coordinator-backend-state.h
@@ -98,10 +98,14 @@ class Coordinator::BackendState {
/// No RPC is issued if there are no fragment instances scheduled on this backend.
/// The 'debug_options' are applied to the appropriate TPlanFragmentInstanceCtxs, based
/// on their node_id/instance_idx.
+ /// If it issues an RPC, 'execquery_rpc_stats' is updated with the size of the
+ /// ExecQueryFInstances RPC. If the RPC is above the warning threshold, it will include
+ /// additional information to help diagnose the large RPCs.
void ExecAsync(const DebugOptions& debug_options,
const FilterRoutingTable& filter_routing_table,
const kudu::Slice& serialized_query_ctx,
- TypedCountingBarrier<Status>* exec_status_barrier);
+ TypedCountingBarrier<Status>* exec_status_barrier,
+ Coordinator::ExecQueryRpcStats* execquery_rpc_stats);
/// Waits until the ExecQueryFInstances() rpc has completed. May be called multiple
/// times and from different threads.
@@ -204,6 +208,7 @@ class Coordinator::BackendState {
bool HasFragmentIdx(const std::unordered_set<int>& fragment_idxs) const;
int64_t rpc_latency() const { return rpc_latency_; }
+ int64_t rpc_size() const { return rpc_size_; }
kudu::Status exec_rpc_status() const { return exec_rpc_status_; }
int64_t last_report_time_ms() {
@@ -377,6 +382,9 @@ class Coordinator::BackendState {
/// Time, in ms, that it took to execute the ExecQueryFInstances() rpc.
int64_t rpc_latency_ = 0;
+ /// Size, in bytes, for the ExecQueryFInstances rpc
+ int64_t rpc_size_ = 0;
+
/////////////////////////////////////////
/// BEGIN: Members that are protected by 'lock_'.
@@ -441,10 +449,12 @@ class Coordinator::BackendState {
/////////////////////////////////////////
/// Fill in 'request' and 'fragment_info' based on state. Uses 'filter_routing_table' to
- /// remove filters that weren't selected during its construction.
+ /// remove filters that weren't selected during its construction. Sums the scan ranges
+ /// across all fragments and puts it in 'total_scan_ranges'.
void SetRpcParams(const DebugOptions& debug_options,
const FilterRoutingTable& filter_routing_table,
- ExecQueryFInstancesRequestPB* request, TExecPlanFragmentInfo* fragment_info);
+ ExecQueryFInstancesRequestPB* request, TExecPlanFragmentInfo* fragment_info,
+ int64_t* total_scan_ranges);
/// Expects that 'status' is an error. Sets 'status_' to a formatted version of its
/// message and notifies 'exec_status_barrier' with it. Caller must hold 'lock_'.
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index 8a9576358..a8f6c3d0a 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -84,6 +84,7 @@ using boost::filesystem::path;
DECLARE_bool(gen_experimental_profile);
DECLARE_string(hostname);
+DECLARE_int64(large_execquery_rpc_threshold_bytes);
using namespace impala;
@@ -122,6 +123,8 @@ PROFILE_DEFINE_COUNTER(NumCompletedBackends, STABLE_HIGH, TUnit::UNIT,"The numbe
"Does not count the number of CANCELLED Backends.");
PROFILE_DEFINE_TIMER(FinalizationTimer, STABLE_LOW,
"Total time spent in finalization (typically 0 except for INSERT into hdfs tables).");
+PROFILE_DEFINE_SUMMARY_STATS_COUNTER(ExecQueryRPCSizes, STABLE_LOW, TUnit::BYTES,
+ "Summary statistics about the size of RPCs sent to start query execution.");
const string Coordinator::PROFILE_EVENT_LABEL_FIRST_ROW_FETCHED = "First row fetched";
@@ -162,6 +165,8 @@ Status Coordinator::Exec() {
obj_pool(), "Execution Profile " + PrintId(query_id()), false);
finalization_timer_ = PROFILE_FinalizationTimer.Instantiate(query_profile_);
filter_updates_received_ = PROFILE_FiltersReceived.Instantiate(query_profile_);
+ execquery_rpc_stats_ = make_unique<ExecQueryRpcStats>(
+ PROFILE_ExecQueryRPCSizes.Instantiate(query_profile_));
host_profiles_ = RuntimeProfile::Create(obj_pool(), "Per Node Profiles", false);
query_profile_->AddChild(host_profiles_);
@@ -546,6 +551,12 @@ Status Coordinator::StartBackendExec() {
return UpdateExecState(serialize_status, nullptr, FLAGS_hostname);
}
kudu::Slice query_ctx_slice(serialized_buf, serialized_len);
+ // Incorporate the size of the shared TQueryCtx into the ExecQueryRpcStats. This also
+ // includes information about the size of the descriptor table, because that can
+ // be a major component of the TQueryCtx's size.
+ int64_t descriptor_table_size = query_ctx().desc_tbl_serialized.thrift_desc_tbl.size();
+ execquery_rpc_stats_->SetSharedTQueryCtxSize(query_ctx_slice.size(),
+ descriptor_table_size);
for (BackendState* backend_state: backend_states_) {
if (exec_rpcs_status_barrier_.pending() <= 0) {
@@ -558,8 +569,11 @@ Status Coordinator::StartBackendExec() {
// because it won't be torn down until WaitOnExecRpcs() has returned.
DCHECK(filter_mode_ == TRuntimeFilterMode::OFF || filter_routing_table_->is_complete);
backend_state->ExecAsync(debug_options, *filter_routing_table_, query_ctx_slice,
- &exec_rpcs_status_barrier_);
+ &exec_rpcs_status_barrier_, execquery_rpc_stats_.get());
}
+ // The ExecQueryRpcStats accumulated any warnings from sending the exec RPCs. Print the
+ // summary now.
+ execquery_rpc_stats_->PrintWarnings();
Status exec_rpc_status = exec_rpcs_status_barrier_.Wait();
if (!exec_rpc_status.ok()) {
// One of the backends failed to startup, so we cancel the other ones.
@@ -1898,3 +1912,49 @@ bool Coordinator::IsExecuting() {
ExecState current_state = exec_state_.Load();
return current_state == ExecState::EXECUTING;
}
+
+void Coordinator::ExecQueryRpcStats::SetSharedTQueryCtxSize(
+ int64_t tqueryctx_size, int64_t descriptor_table_size) {
+ tqueryctx_size_ = tqueryctx_size;
+ descriptor_table_size_ = descriptor_table_size;
+}
+
+void Coordinator::ExecQueryRpcStats::ReportRpcSize(int64_t size) {
+ execquery_rpc_sizes_->UpdateCounter(size);
+}
+
+void Coordinator::ExecQueryRpcStats::ReportRpcSizeWithWarning(int64_t size,
+ string&& warning_message) {
+ ReportRpcSize(size);
+ rpcs_above_warning_threshold_++;
+ if (largest_rpc_warnings_.size() >= MAX_EXEC_RPC_WARNINGS) {
+ // Compare to the smallest element and replace if bigger
+ auto smallest = largest_rpc_warnings_.begin();
+ if (smallest->first > size) return;
+ largest_rpc_warnings_.erase(smallest);
+ }
+ largest_rpc_warnings_.emplace(size, warning_message);
+}
+
+void Coordinator::ExecQueryRpcStats::PrintWarnings() {
+ if (rpcs_above_warning_threshold_ == 0) return;
+ LOG(WARNING) << "ExecQueryFInstances RPCs for " << rpcs_above_warning_threshold_
+ << " out of " << execquery_rpc_sizes_->TotalNumValues()
+ << " exceed the warning threshold of "
+ << PrettyPrinter::Print(FLAGS_large_execquery_rpc_threshold_bytes, TUnit::BYTES)
+ << ". Average RPC size: "
+ << PrettyPrinter::Print(execquery_rpc_sizes_->value(), TUnit::BYTES)
+ << " Shared TQueryCtx: " << PrettyPrinter::Print(tqueryctx_size_, TUnit::BYTES)
+ << " (" << PrettyPrinter::Print(descriptor_table_size_, TUnit::BYTES)
+ << " from descriptor table)";
+ LOG(WARNING) << "Additional information about the " << largest_rpc_warnings_.size()
+ << " largest RPCs:";
+ // Print the warnings from largest to smallest using a reverse iterator
+ DCHECK_LE(largest_rpc_warnings_.size(), MAX_EXEC_RPC_WARNINGS);
+ for (auto it = largest_rpc_warnings_.crbegin();
+ it != largest_rpc_warnings_.crend(); ++it) {
+ LOG(WARNING) << it->second;
+ }
+ // We no longer need the warnings structure, so free the memory.
+ largest_rpc_warnings_.clear();
+}
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index 40c7e3870..b8ea6533e 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -425,6 +425,61 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// True if all Exec() rpcs have completed.
AtomicBool exec_rpcs_complete_{false};
+ class ExecQueryRpcStats {
+ public:
+ ExecQueryRpcStats(RuntimeProfile::SummaryStatsCounter* exec_rpc_sizes)
+ : execquery_rpc_sizes_(exec_rpc_sizes) {}
+
+ /// The shared TQueryCtx can be a major component of the size of the
+ /// ExecQueryFInstances RPC. This needs to be called before printing the warnings
+ /// so that the shared size information is available.
+ void SetSharedTQueryCtxSize(int64_t tqueryctx_size, int64_t descriptor_table_size);
+
+ /// Report the size of the ExecQueryFInstances RPC. This incorporates it into the
+ /// summary statistics.
+ void ReportRpcSize(int64_t size);
+
+ /// Report the size of an ExecQueryFInstances RPC that exceeded the warning
+ /// threshold. This will move the warning into the map of RPC warnings.
+ void ReportRpcSizeWithWarning(int64_t size, std::string&& warning_message);
+
+ /// If an RPC exceeded the warning threshold, print information to help
+ /// diagnose the issue. Specifically, this prints a summary for the whole query
+ /// with information about the average size of the RPCs and the size of the
+ /// shared components (the TQueryCtx) of the message. Then it prints more detailed
+ /// information about the largest few RPCs with more information about the backend
+ /// specific content of the message. This is not idempotent. It is intended to be
+ /// called once as the last operation on this structure, so it clears the warnings
+ /// data structure to save memory.
+ void PrintWarnings();
+
+ private:
+ /// Summary statistics about the size of ExecQueryFInstances RPCs. Updated as
+ /// the RPCs are sent.
+ RuntimeProfile::SummaryStatsCounter* execquery_rpc_sizes_ = nullptr;
+
+ /// Number of ExecQueryFInstances RPCs above the threshold
+ int64_t rpcs_above_warning_threshold_ = 0;
+
+ /// Size of the shared TQueryCtx piece of the RPC
+ int64_t tqueryctx_size_ = 0;
+
+ /// Size of the descriptor table, which can be a large portion of the TQueryCtx
+ int64_t descriptor_table_size_ = 0;
+
+ /// Structure to keep a warning string for the top three RPCs. This uses an
+ /// ordered map for simplicity. We want to be able to quickly evict the smallest
+ /// element, but we also want to be able to iterate from largest to smallest.
+ /// This is very small, so there is no benefit to using a heap based structure
+ /// like a std::priority_queue.
+ static constexpr int64_t MAX_EXEC_RPC_WARNINGS = 3;
+ std::multimap<int64_t, std::string> largest_rpc_warnings_;
+ };
+
+ /// Aggregated information about the ExecQueryFInstances RPCs for diagnosing
+ /// large RPCs.
+ std::unique_ptr<ExecQueryRpcStats> execquery_rpc_stats_;
+
/// Barrier that is released when all backends have indicated execution completion,
/// or when all backends are cancelled due to an execution error or client requested
/// cancellation. Initialized in StartBackendExec().
diff --git a/be/src/scheduling/admission-control-service.cc b/be/src/scheduling/admission-control-service.cc
index 350b00f0d..47bd1ceb1 100644
--- a/be/src/scheduling/admission-control-service.cc
+++ b/be/src/scheduling/admission-control-service.cc
@@ -236,7 +236,9 @@ void AdmissionControlService::GetQueryStatus(const GetQueryStatusRequestPB* req,
TRuntimeProfileTree tree;
admission_state->summary_profile->ToThrift(&tree);
int sidecar_idx;
- Status sidecar_status = SetFaststringSidecar(tree, rpc_context, &sidecar_idx);
+ int64_t sidecar_length = 0;
+ Status sidecar_status = SetFaststringSidecar(tree, rpc_context, &sidecar_idx,
+ &sidecar_length);
if (!sidecar_status.ok()) {
// We don't need to fail the query just because we can't return the profile, so
// just log the error.
diff --git a/tests/webserver/test_web_pages.py b/tests/webserver/test_web_pages.py
index 8e78d6c10..505a4c461 100644
--- a/tests/webserver/test_web_pages.py
+++ b/tests/webserver/test_web_pages.py
@@ -769,7 +769,7 @@ class TestWebPage(ImpalaTestSuite):
backend_state_properties = ['cpu_user_s', 'rpc_latency', 'num_remaining_instances',
'num_instances', 'peak_per_host_mem_consumption',
'time_since_last_heard_from', 'status', 'host',
- 'cpu_sys_s', 'done', 'bytes_read']
+ 'cpu_sys_s', 'done', 'bytes_read', 'rpc_size']
for query in [sleep_query, ctas_sleep_query]:
(_, response_json) = self.run_query_and_get_debug_page(