You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2026/05/12 01:08:20 UTC
(impala) 02/02: IMPALA-14843: Show cancelled nodes in ExecSummary
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 9582b8bff2a013309507905e21a1020a72bacee4
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Thu Mar 19 20:07:42 2026 +0800
IMPALA-14843: Show cancelled nodes in ExecSummary
In backend execution, a plan node instance could be closed before it
produces all its output rows, e.g., when a UnionNode reaches its output
limit, the children operands are all closed regardless what their states
(prepared, opened, executing, etc.) are. In ExecSummary shown in the
query profile or WebUI, such nodes usually have 0 as the output
cardinality without any indication, which is confusing. This patch adds
a "CANCELLED" marker in the Detail column to represent such nodes, i.e.,
nodes that haven't reached the "Last Batch Returned" state.
Implementation:
- Adds a last_batch_returned_ flag in ExecNode to indicate whether the
"Last Batch Returned" event has been added for this node.
- For ExecNodes under a SubplanNode, use the flag of the top-level
SubplanNode since nodes in Subplan are reset multiple times.
- Extends ExecSummaryDataPB and TExecStats to add a last_batch_returned
field to reflect such case. Note that ExecSummaryDataPB is used in
executors reporting execution status to the coordinator. TExecStats
is used in coordinator to track execution stats for plan nodes.
- Executor mirrors the flag to ExecSummaryDataPB for each node when
generating status report for coordinator. To lookup the ExecNode
instance based on the plan node id, a map is added in
FragmentInstanceState when the ExecNode tree is built.
- Coordinator appends the CANCELLED marker when returning TExecSummary
if the query finishes execution. This avoids showing the CANCELLED
for inflight queries.
Example:
For query
"with l as
(select * from tpch.lineitem UNION ALL select * from tpch.lineitem)
select STRAIGHT_JOIN count(*)
from (select * from tpch.lineitem a LIMIT 1) a
join (select * from l LIMIT 125000) b
on a.l_orderkey = -b.l_orderkey"
the ExecSummary is
Operator #Hosts #Inst Avg Time Max Time #Rows Est. #Rows Peak Mem Est. Peak Mem Detail
---------------------------------------------------------------------------------------------------------------------------------------
F01:ROOT 1 1 8.647us 8.647us 0 0
05:AGGREGATE 1 1 0.000ns 0.000ns 1 1 24.00 KB 16.00 KB FINALIZE
04:HASH JOIN 1 1 251.664us 251.664us 0 1 9.06 MB 4.75 MB INNER JOIN, BROADCAST
|--08:EXCHANGE 1 1 427.557us 427.557us 125.00K 125.00K 232.00 KB 337.52 KB UNPARTITIONED
| F05:EXCHANGE SENDER 1 1 3.007ms 3.007ms 54.05 KB 48.00 KB
| 07:EXCHANGE 1 1 499.833us 499.833us 125.00K 125.00K 5.12 MB 361.52 KB UNPARTITIONED
| F04:EXCHANGE SENDER 3 3 2.681ms 2.833ms 54.05 KB 48.00 KB
| 01:UNION 3 3 43.817us 48.670us 375.00K 125.00K 0 0
| |--03:SCAN HDFS 3 3 40.710us 49.726us 0 6.00M 0 264.00 MB tpch.lineitem, CANCELLED
| 02:SCAN HDFS 3 3 5.813ms 7.886ms 377.86K 6.00M 48.23 MB 264.00 MB tpch.lineitem, CANCELLED
06:EXCHANGE 1 1 10.288us 10.288us 1 1 16.00 KB 16.00 KB UNPARTITIONED
F00:EXCHANGE SENDER 3 3 32.298us 35.572us 31.00 B 48.00 KB
00:SCAN HDFS 3 3 76.692ms 85.439ms 3 1 48.07 MB 264.00 MB tpch.lineitem a
Note that "02:SCAN HDFS" and "03:SCAN HDFS" are cancelled due to
"01:UNION" reaches its limit.
Test:
- Added e2e tests in query_test/test_observability.py
Assisted-by: Claude Sonnet 4.5
Change-Id: Ibd1bf51112e7c8c5609bb9e9525d960efc524215
Reviewed-on: http://gerrit.cloudera.org:8080/24118
Reviewed-by: Michael Smith <mi...@cloudera.com>
Reviewed-by: Jason Fehr <jf...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/exec-node-util.h | 5 +-
be/src/exec/exec-node.h | 4 +
be/src/runtime/coordinator-backend-state.cc | 3 +
be/src/runtime/coordinator.cc | 32 ++++-
be/src/runtime/coordinator.h | 4 +
be/src/runtime/fragment-instance-state.cc | 31 ++++-
be/src/runtime/fragment-instance-state.h | 5 +
be/src/service/impala-http-handler.cc | 5 +-
be/src/util/summary-util.cc | 2 +-
be/src/util/summary-util.h | 3 +
common/protobuf/control_service.proto | 3 +
common/thrift/ExecStats.thrift | 3 +
tests/query_test/test_observability.py | 179 ++++++++++++++++++++++++++++
13 files changed, 271 insertions(+), 8 deletions(-)
diff --git a/be/src/exec/exec-node-util.h b/be/src/exec/exec-node-util.h
index b342307e1..0087a2ff9 100644
--- a/be/src/exec/exec-node-util.h
+++ b/be/src/exec/exec-node-util.h
@@ -65,7 +65,10 @@ class ScopedGetNextEventAdder {
}
// Note that we won't mark the last batch if the node is closed before eos,
// e.g. if an ancestor node hits a limit.
- if (*eos_) node_->events_->MarkEvent("Last Batch Returned");
+ if (*eos_) {
+ node_->events_->MarkEvent("Last Batch Returned");
+ node_->last_batch_returned_ = true;
+ }
}
private:
diff --git a/be/src/exec/exec-node.h b/be/src/exec/exec-node.h
index 550142d6f..d40ad8293 100644
--- a/be/src/exec/exec-node.h
+++ b/be/src/exec/exec-node.h
@@ -354,6 +354,7 @@ class ExecNode {
MemPool* expr_results_pool() { return expr_results_pool_.get(); }
const TBackendResourceProfile& resource_profile() { return resource_profile_; }
bool is_closed() const { return is_closed_; }
+ bool last_batch_returned() const { return last_batch_returned_; }
/// Returns true if this node is inside the right-hand side plan tree of a SubplanNode.
bool IsInSubplan() const { return plan_node_.IsInSubplan(); }
@@ -417,6 +418,9 @@ class ExecNode {
/// ScopedGetNextEventAdder can avoid adding a duplicate event.
bool first_getnext_added_ = false;
+ /// Used to track whether the "Last Batch Returned" event was added to 'events_'.
+ bool last_batch_returned_ = false;
+
/// Conjuncts and their evaluators in this node. 'conjuncts_' live in the
/// query-state's object pool while the evaluators live in this exec node's
/// object pool. Note: conjunct_evals_ are not created for Aggregation nodes.
diff --git a/be/src/runtime/coordinator-backend-state.cc b/be/src/runtime/coordinator-backend-state.cc
index 19e23a3bb..4d0388649 100644
--- a/be/src/runtime/coordinator-backend-state.cc
+++ b/be/src/runtime/coordinator-backend-state.cc
@@ -884,6 +884,9 @@ void Coordinator::BackendState::InstanceStats::Update(
}
DCHECK(exec_summary_entry.has_local_time_ns());
instance_stats.__set_latency_ns(exec_summary_entry.local_time_ns());
+ if (exec_summary_entry.has_last_batch_returned()) {
+ instance_stats.__set_last_batch_returned(exec_summary_entry.last_batch_returned());
+ }
node_exec_summary.__isset.exec_stats = true;
}
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index eaf508942..8a9576358 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -64,6 +64,7 @@
#include "util/in-list-filter.h"
#include "util/min-max-filter.h"
#include "util/pretty-printer.h"
+#include "util/summary-util.h"
#include "util/table-printer.h"
#include "util/uid-util.h"
@@ -309,7 +310,7 @@ void Coordinator::ExecSummary::Init(const QueryExecParams& exec_params) {
// Note that some clients like impala-shell depend on many of these fields being
// set, even if they are optional in the thrift.
TPlanNodeExecSummary& node_summary = thrift_exec_summary.nodes.back();
- node_summary.__set_node_id(-1);
+ node_summary.__set_node_id(SINK_NODE_ID);
node_summary.__set_fragment_idx(fragment.idx);
node_summary.__set_label(output_sink.label);
node_summary.__set_label_detail("");
@@ -1813,9 +1814,38 @@ string Coordinator::FilterState::DebugString() const {
return ss.str();
}
+void Coordinator::MarkCancelledNodes(TExecSummary* exec_summary) {
+ if (!exec_summary->__isset.nodes) return;
+
+ for (TPlanNodeExecSummary& node : exec_summary->nodes) {
+ if (!node.__isset.exec_stats || node.exec_stats.empty()) continue;
+ // Skip data sinks
+ if (node.node_id == SINK_NODE_ID) continue;
+
+ // Check if any instance didn't complete execution
+ bool has_cancelled_instance = false;
+ for (const TExecStats& stat : node.exec_stats) {
+ if (stat.__isset.last_batch_returned && !stat.last_batch_returned) {
+ has_cancelled_instance = true;
+ break;
+ }
+ }
+ // Append a "CANCELLED" marker which will be shown in the Detail column in the
+ // ExecSummary table.
+ if (has_cancelled_instance) {
+ if (node.label_detail.empty()) {
+ node.label_detail = "CANCELLED";
+ } else {
+ node.label_detail += ", CANCELLED";
+ }
+ }
+ }
+}
+
void Coordinator::GetTExecSummary(TExecSummary* exec_summary) {
lock_guard<SpinLock> l(exec_summary_.lock);
*exec_summary = exec_summary_.thrift_exec_summary;
+ if (!IsExecuting()) MarkCancelledNodes(exec_summary);
}
MemTracker* Coordinator::query_mem_tracker() const {
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index 3a6b129ba..40c7e3870 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -524,6 +524,10 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// Return the string representation of 'state'.
static const char* ExecStateToString(const ExecState state);
+ /// Appends "CANCELLED" to label_detail for any nodes where at least one fragment
+ /// instance didn't complete execution, e.g., due to parent node reaches limit.
+ static void MarkCancelledNodes(TExecSummary* exec_summary);
+
// For DCHECK_EQ, etc of ExecState values.
friend std::ostream& operator<<(std::ostream& o, const ExecState s) {
return o << ExecStateToString(s);
diff --git a/be/src/runtime/fragment-instance-state.cc b/be/src/runtime/fragment-instance-state.cc
index 99e91cbd5..66ddb2555 100644
--- a/be/src/runtime/fragment-instance-state.cc
+++ b/be/src/runtime/fragment-instance-state.cc
@@ -33,6 +33,7 @@
#include "exec/partitioned-hash-join-builder.h"
#include "exec/plan-root-sink.h"
#include "exec/scan-node.h"
+#include "exec/subplan-node.h"
#include "gen-cpp/ImpalaInternalService_types.h"
#include "kudu/rpc/rpc_context.h"
#include "runtime/client-cache.h"
@@ -208,6 +209,13 @@ Status FragmentInstanceState::Prepare() {
if (instance_ctx_.__isset.debug_options) {
ExecNode::SetDebugOptions(instance_ctx_.debug_options, exec_tree_);
}
+ std::function<void(ExecNode*)> build_node_map = [&](ExecNode* node) {
+ exec_node_map_[node->id()] = node;
+ for (int i = 0; i < node->num_children(); ++i) {
+ build_node_map(node->child(i));
+ }
+ };
+ build_node_map(exec_tree_);
// set #senders of exchange nodes before calling Prepare()
vector<ExecNode*> exch_nodes;
@@ -276,6 +284,25 @@ Status FragmentInstanceState::Prepare() {
return Status::OK();
}
+void FragmentInstanceState::SetLastBatchReturnedForPlanNode(
+ TPlanNodeId node_id, ExecSummaryDataPB* summary_data) {
+ auto it = exec_node_map_.find(node_id);
+ if (it == exec_node_map_.end()) {
+ DCHECK(false) << "Unknown plan node " << node_id << " in profile";
+ LOG(ERROR) << "Unknown plan node " << node_id << " in profile";
+ return;
+ }
+ ExecNode* exec_node = it->second;
+ while (exec_node->IsInSubplan()) {
+ // Nodes inside subplans don't track last_batch_returned_ (events_ is null).
+ // Walk up the containing_subplan_ chain to find the outermost SubplanNode
+ // that has events_ initialized (i.e. is not itself in a subplan).
+ exec_node = exec_node->get_containing_subplan();
+ DCHECK(exec_node != nullptr);
+ }
+ summary_data->set_last_batch_returned(exec_node->last_batch_returned());
+}
+
void FragmentInstanceState::GetStatusReport(FragmentInstanceExecStatusPB* instance_status,
TRuntimeProfileTree* unagg_profile, AggregatedRuntimeProfile* agg_profile,
const Status& overall_status) {
@@ -341,7 +368,9 @@ void FragmentInstanceState::GetStatusReport(FragmentInstanceExecStatusPB* instan
if (is_plan_node || is_data_sink) {
ExecSummaryDataPB* summary_data = instance_status->add_exec_summary_data();
if (is_plan_node) {
- summary_data->set_plan_node_id(node->metadata().plan_node_id);
+ TPlanNodeId node_id = node->metadata().plan_node_id;
+ summary_data->set_plan_node_id(node_id);
+ SetLastBatchReturnedForPlanNode(node_id, summary_data);
} else {
summary_data->set_data_sink_id(node->metadata().data_sink_id);
}
diff --git a/be/src/runtime/fragment-instance-state.h b/be/src/runtime/fragment-instance-state.h
index c3210af85..eda531dd7 100644
--- a/be/src/runtime/fragment-instance-state.h
+++ b/be/src/runtime/fragment-instance-state.h
@@ -181,6 +181,7 @@ class FragmentInstanceState {
/// in Prepare().
ExecNode* exec_tree_ = nullptr; // lives in obj_pool()
RuntimeState* runtime_state_ = nullptr; // lives in obj_pool()
+ std::unordered_map<TPlanNodeId, ExecNode*> exec_node_map_;
/// A 'fake mutex' to detect any race condition in accessing 'report_seq_no_' below.
/// There should be only one thread doing status report at the same time.
@@ -337,6 +338,10 @@ class FragmentInstanceState {
/// Print stats about scan ranges for each volumeId in params to info log.
void PrintVolumeIds();
+
+ /// Sets last_batch_returned on 'summary_data' for plan node 'node_id'.
+ void SetLastBatchReturnedForPlanNode(
+ TPlanNodeId node_id, ExecSummaryDataPB* summary_data);
};
}
diff --git a/be/src/service/impala-http-handler.cc b/be/src/service/impala-http-handler.cc
index 0f86585fc..fa7addaf2 100644
--- a/be/src/service/impala-http-handler.cc
+++ b/be/src/service/impala-http-handler.cc
@@ -1083,9 +1083,6 @@ void ImpalaHttpHandler::CatalogObjectsHandler(const Webserver::WebRequest& req,
namespace {
-// Summary is stored with -1 as id if it is for a data sink at the root of a fragment.
-constexpr int SINK_ID = -1;
-
void ExecStatsToJsonHelper(
const TPlanNodeExecSummary& summary, rapidjson::Document* document, Value* value) {
int64_t cardinality = 0;
@@ -1184,7 +1181,7 @@ void SinkToJsonHelper(const TDataSink& sink,
value->AddMember("label_detail", label_detail, document->GetAllocator());
map<TPlanNodeId, TPlanNodeExecSummary>::const_iterator summary_it =
- summaries.find(SINK_ID);
+ summaries.find(SINK_NODE_ID);
if (summary_it != summaries.end()) {
// Sometimes we don't have a summary for the sink, e.g.:
// - Query was rejected by admission control
diff --git a/be/src/util/summary-util.cc b/be/src/util/summary-util.cc
index cdb945b28..b3badbcad 100644
--- a/be/src/util/summary-util.cc
+++ b/be/src/util/summary-util.cc
@@ -77,7 +77,7 @@ void PrintExecSummary(const TExecSummary& exec_summary, int indent_level,
row.push_back(lexical_cast<string>(node.exec_stats.size())); // Num instances
row.push_back(PrettyPrinter::Print(avg_time, TUnit::TIME_NS));
row.push_back(PrettyPrinter::Print(max_stats.latency_ns, TUnit::TIME_NS));
- if (node.node_id == -1) {
+ if (node.node_id == SINK_NODE_ID) {
// Cardinality stats are not valid for sinks.
row.push_back("");
row.push_back("");
diff --git a/be/src/util/summary-util.h b/be/src/util/summary-util.h
index c9ecb323e..c602d5228 100644
--- a/be/src/util/summary-util.h
+++ b/be/src/util/summary-util.h
@@ -22,6 +22,9 @@
namespace impala {
+// Plan node ID for the fragment root data sink.
+constexpr int32_t SINK_NODE_ID = -1;
+
/// Print the exec summary as a formatted table.
std::string PrintExecSummary(const TExecSummary& exec_summary);
diff --git a/common/protobuf/control_service.proto b/common/protobuf/control_service.proto
index f838e5501..9cad3203f 100644
--- a/common/protobuf/control_service.proto
+++ b/common/protobuf/control_service.proto
@@ -175,6 +175,9 @@ message ExecSummaryDataPB {
// Local time in nanoseconds spent in this plan node.
optional int64 local_time_ns = 5;
+
+ // True if this plan node instance completed execution.
+ optional bool last_batch_returned = 6;
}
// RPC error metadata that can be associated with a AuxErrorInfoPB object. Created if a
diff --git a/common/thrift/ExecStats.thrift b/common/thrift/ExecStats.thrift
index 0a6073c86..48d0c5e13 100644
--- a/common/thrift/ExecStats.thrift
+++ b/common/thrift/ExecStats.thrift
@@ -55,6 +55,9 @@ struct TExecStats {
// Peak memory used (in bytes).
4: optional i64 memory_used
+
+ // True if this instance completed execution.
+ 5: optional bool last_batch_returned
}
// Summary for a single plan node or data sink. This includes labels for how to display
diff --git a/tests/query_test/test_observability.py b/tests/query_test/test_observability.py
index e8d5f88d1..98c3b3403 100644
--- a/tests/query_test/test_observability.py
+++ b/tests/query_test/test_observability.py
@@ -162,6 +162,185 @@ class TestObservability(ImpalaTestSuite):
assert result.exec_summary[0]['peak_mem'] >= 0
assert result.exec_summary[0]['est_peak_mem'] >= 0
+ def test_cancelled_nodes_in_exec_summary(self):
+ """Test nodes that don't complete are marked as cancelled"""
+ # UNION operands are cancelled when UNION reaches the limit.
+ query = """
+ with l as (select 1 from tpch.lineitem UNION ALL select 1 from tpch.customer)
+ select * from l LIMIT 10"""
+ result = self.client.execute(query, fetch_exec_summary=True)
+ # ExecSummary:
+ # Operator #Rows Est. #Rows Detail
+ # ------------------------------------------------------------------
+ # F03:ROOT
+ # 03:EXCHANGE 10 10 UNPARTITIONED
+ # F02:EXCHANGE SENDER
+ # 00:UNION 30 10
+ # |--02:SCAN HDFS 0 150.00K tpch.customer, CANCELLED
+ # 01:SCAN HDFS 3.07K 6.00M tpch.lineitem, CANCELLED
+ assert result.exec_summary[4]['operator'] == '02:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[4]['detail'] == 'tpch.customer, CANCELLED', \
+ result.runtime_profile
+ assert result.exec_summary[5]['operator'] == '01:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[5]['detail'] == 'tpch.lineitem, CANCELLED', \
+ result.runtime_profile
+
+ # A more complex case with completed and cancelled nodes.
+ query = """
+ with l as (select * from tpch.lineitem UNION ALL select * from tpch.lineitem)
+ select STRAIGHT_JOIN count(*) from
+ (select * from tpch.lineitem a LIMIT 1) a
+ join
+ (select * from l LIMIT 125000) b
+ on a.l_orderkey = -b.l_orderkey"""
+ result = self.client.execute(query, fetch_exec_summary=True)
+ # ExecSummary:
+ # Operator #Rows Est. #Rows Detail
+ # -----------------------------------------------------------------------
+ # F01:ROOT
+ # 05:AGGREGATE 1 1 FINALIZE
+ # 04:HASH JOIN 0 1 INNER JOIN, BROADCAST
+ # |--08:EXCHANGE 125.00K 125.00K UNPARTITIONED
+ # | F05:EXCHANGE SENDER
+ # | 07:EXCHANGE 125.00K 125.00K UNPARTITIONED
+ # | F04:EXCHANGE SENDER
+ # | 01:UNION 375.00K 125.00K
+ # | |--03:SCAN HDFS 0 6.00M tpch.lineitem, CANCELLED
+ # | 02:SCAN HDFS 377.86K 6.00M tpch.lineitem, CANCELLED
+ # 06:EXCHANGE 1 1 UNPARTITIONED
+ # F00:EXCHANGE SENDER
+ # 00:SCAN HDFS 3 1 tpch.lineitem a
+ assert result.exec_summary[8]['operator'] == '03:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[8]['detail'] == 'tpch.lineitem, CANCELLED', \
+ result.runtime_profile
+ assert result.exec_summary[9]['operator'] == '02:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[9]['detail'] == 'tpch.lineitem, CANCELLED', \
+ result.runtime_profile
+ assert result.exec_summary[12]['operator'] == '00:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[12]['detail'] == 'tpch.lineitem a', result.runtime_profile
+
+ # Test on other node types
+ query = """
+ with l as (
+ select row_number() over(order by l_orderkey) from tpch.lineitem
+ UNION ALL select 1 from tpch.customer)
+ select * from l LIMIT 10"""
+ result = self.client.execute(query, fetch_exec_summary=True)
+ # ExecSummary:
+ # Operator #Rows Est. #Rows Detail
+ # --------------------------------------------------------------------
+ # F04:ROOT
+ # 07:EXCHANGE 10 10 UNPARTITIONED
+ # F03:EXCHANGE SENDER
+ # 00:UNION 20 10
+ # |--04:SCAN HDFS 0 150.00K tpch.customer, CANCELLED
+ # 06:EXCHANGE 2.05K 6.00M RANDOM, CANCELLED
+ # F01:EXCHANGE SENDER
+ # 03:ANALYTIC 29.70K 6.00M CANCELLED
+ # 05:MERGING-EXCHANGE 30.72K 6.00M UNPARTITIONED, CANCELLED
+ # F00:EXCHANGE SENDER
+ # 02:SORT 661.05K 6.00M CANCELLED
+ # 01:SCAN HDFS 6.00M 6.00M tpch.lineitem
+ assert result.exec_summary[4]['operator'] == '04:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[4]['detail'] == 'tpch.customer, CANCELLED', \
+ result.runtime_profile
+ assert result.exec_summary[5]['operator'] == '06:EXCHANGE', result.runtime_profile
+ assert result.exec_summary[5]['detail'] == 'RANDOM, CANCELLED', result.runtime_profile
+ assert result.exec_summary[7]['operator'] == '03:ANALYTIC', result.runtime_profile
+ assert result.exec_summary[7]['detail'] == 'CANCELLED', result.runtime_profile
+ assert result.exec_summary[8]['operator'] == '05:MERGING-EXCHANGE', \
+ result.runtime_profile
+ assert result.exec_summary[8]['detail'] == 'UNPARTITIONED, CANCELLED', \
+ result.runtime_profile
+ assert result.exec_summary[10]['operator'] == '02:SORT', result.runtime_profile
+ assert result.exec_summary[10]['detail'] == 'CANCELLED', result.runtime_profile
+ assert result.exec_summary[11]['operator'] == '01:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[11]['detail'] == 'tpch.lineitem', result.runtime_profile
+
+ # Test limit on Subplan. As the SubplanNode reaches its limit, it finishes as expected
+ # and shouldn't be marked as CANCELLED.
+ query = """
+ select c.c_custkey, arr.o_orderkey
+ from tpch_nested_parquet.customer c, c.c_orders arr
+ limit 10"""
+ # Set num_nodes to 1 to ensure only one fragment instance is created. By default the
+ # query has 3 fragment instances. Some might be cancelled by the coordinator when it
+ # receives enough rows. Using one fragment instance ensures that when coordinator
+ # reaches the limit, the only SubplanNode has finished (also reached its limit).
+ self.client.set_configuration_option("num_nodes", 1)
+ result = self.client.execute(query, fetch_exec_summary=True)
+ self.client.clear_configuration()
+ # ExecSummary:
+ # Operator #Rows Est. #Rows Detail
+ # ------------------------------------------------------------------------------------
+ # F00:ROOT
+ # 01:SUBPLAN 10 10
+ # |--04:NESTED LOOP JOIN 7 10 CROSS JOIN
+ # | |--02:SINGULAR ROW SRC 0 1
+ # | 03:UNNEST 7 10 c.c_orders arr
+ # 00:SCAN HDFS 6 150.00K tpch_nested_parquet.customer c, CANCELLED
+ assert result.exec_summary[1]['operator'] == '01:SUBPLAN', result.runtime_profile
+ assert result.exec_summary[1]['detail'] == '', result.runtime_profile
+ assert result.exec_summary[2]['operator'] == '04:NESTED LOOP JOIN', \
+ result.runtime_profile
+ assert result.exec_summary[2]['detail'] == 'CROSS JOIN', result.runtime_profile
+ assert result.exec_summary[3]['operator'] == '02:SINGULAR ROW SRC', \
+ result.runtime_profile
+ assert result.exec_summary[3]['detail'] == '', result.runtime_profile
+ assert result.exec_summary[4]['operator'] == '03:UNNEST', result.runtime_profile
+ assert result.exec_summary[4]['detail'] == 'c.c_orders arr', result.runtime_profile
+ assert result.exec_summary[5]['operator'] == '00:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[5]['detail'] == \
+ 'tpch_nested_parquet.customer c, CANCELLED', result.runtime_profile
+
+ # Test limit on parent (UnionNode) of SubplanNode. All nodes in the SubplanNode should
+ # be marked as CANCELLED.
+ query = """
+ with l as (
+ select 1, 2
+ union all
+ select c.c_custkey, arr.o_orderkey
+ from tpch_nested_orc_def.customer c, c.c_orders arr
+ ) select * from l limit 10"""
+ result = self.client.execute(query, fetch_exec_summary=True)
+ # Operator #Rows Est. #Rows Detail
+ # ------------------------------------------------------------------------------------
+ # F02:ROOT
+ # 06:EXCHANGE 10 10 UNPARTITIONED
+ # F01:EXCHANGE SENDER
+ # 00:UNION 20 10 CANCELLED
+ # 02:SUBPLAN 1.02K 1.50M CANCELLED
+ # |--05:NESTED LOOP JOIN 21 10 CROSS JOIN, CANCELLED
+ # | |--03:SINGULAR ROW SRC 0 1 CANCELLED
+ # | 04:UNNEST 22 10 c.c_orders arr, CANCELLED
+ # 01:SCAN HDFS 1.03K 150.00K tpch_nested_orc_def.customer c, CANCELLED
+ assert result.exec_summary[4]['operator'] == '02:SUBPLAN', result.runtime_profile
+ assert result.exec_summary[4]['detail'] == 'CANCELLED', result.runtime_profile
+ assert result.exec_summary[5]['operator'] == '05:NESTED LOOP JOIN', \
+ result.runtime_profile
+ assert result.exec_summary[5]['detail'] == 'CROSS JOIN, CANCELLED', \
+ result.runtime_profile
+ assert result.exec_summary[6]['operator'] == '03:SINGULAR ROW SRC', \
+ result.runtime_profile
+ assert result.exec_summary[6]['detail'] == 'CANCELLED', result.runtime_profile
+ assert result.exec_summary[7]['operator'] == '04:UNNEST', result.runtime_profile
+ assert result.exec_summary[7]['detail'] == 'c.c_orders arr, CANCELLED', \
+ result.runtime_profile
+ assert result.exec_summary[8]['operator'] == '01:SCAN HDFS', result.runtime_profile
+ assert result.exec_summary[8]['detail'] == \
+ 'tpch_nested_orc_def.customer c, CANCELLED', result.runtime_profile
+
+ def test_cancelled_markers_in_running_query(self):
+ """Test that the CANCELLED markers are not added to the summary for running queries"""
+ query = "select count(*) from functional.alltypestiny where id = sleep(1000)"
+ handle = self.execute_query_async(query)
+ while not self.client.is_finished(handle):
+ exec_summary = self.client.get_exec_summary_table(handle)
+ for row in exec_summary:
+ assert 'CANCELLED' not in row['detail'], \
+ "CANCELLED marker found in running query: " + str(row)
+ sleep(0.5)
+
def test_query_options(self):
"""Test that the query profile shows expected non-default query options, both set
explicitly through client and those set by planner"""