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 2019/03/19 03:28:16 UTC
[impala] 03/03: IMPALA-8097: mt_dop for all queries via hidden flag
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
commit 9090fc239b62c0d698c6c256c20d86b69f8cc64f
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Thu Jan 3 15:24:33 2019 -0800
IMPALA-8097: mt_dop for all queries via hidden flag
--unlock_mt_dop=true unlocks mt_dop for all queries
including joins and inserts.
This disables the parallel plans with separate join builds
when running standalone, because these are not executable
until IMPALA-4224 is implemented. Inserts work without
modification - they were disabled because of lack of
testing and the possibility for generating many small
files with unpartitioned inserts - see IMPALA-8125.
Testing:
Add custom cluster test that exercise joins, runtime filters
and inserts as a sanity check for the flag.
Ran exhaustive build.
Manually ran TPC-H and TPC-DS tests against a minicluster
with mt_dop = 4.
Change-Id: I72f0b02a005e8bf22fd17b8fb5aabf8c0d9b6b15
Reviewed-on: http://gerrit.cloudera.org:8080/12257
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/common/global-flags.cc | 3 +
be/src/exec/blocking-join-node.cc | 8 ++-
be/src/exec/partitioned-hash-join-builder.cc | 3 +-
be/src/runtime/coordinator-backend-state.cc | 1 -
be/src/runtime/coordinator.cc | 5 --
be/src/util/backend-gflag-util.cc | 2 +
common/thrift/BackendGflags.thrift | 2 +
.../java/org/apache/impala/planner/Planner.java | 10 ++-
.../apache/impala/planner/SingleNodePlanner.java | 13 ++--
.../org/apache/impala/service/BackendConfig.java | 4 ++
.../queries/PlannerTest/mt-dop-validation.test | 80 +++++++++++++++++++++-
.../queries/QueryTest/joins_mt_dop.test | 27 ++++++++
tests/custom_cluster/test_mt_dop.py | 62 +++++++++++++++++
13 files changed, 200 insertions(+), 20 deletions(-)
diff --git a/be/src/common/global-flags.cc b/be/src/common/global-flags.cc
index 470a111..fdf28b2 100644
--- a/be/src/common/global-flags.cc
+++ b/be/src/common/global-flags.cc
@@ -270,6 +270,9 @@ DEFINE_bool_hidden(enable_parquet_page_index_writing_debug_only, false, "If true
"We plan to remove this flag once Impala is able to read the page index and has "
"better test coverage around it.");
+DEFINE_bool_hidden(unlock_mt_dop, false,
+ "(Experimental) If true, allow specifying mt_dop for all queries.");
+
// ++========================++
// || Startup flag graveyard ||
// ++========================++
diff --git a/be/src/exec/blocking-join-node.cc b/be/src/exec/blocking-join-node.cc
index e9281c6..ba1a758 100644
--- a/be/src/exec/blocking-join-node.cc
+++ b/be/src/exec/blocking-join-node.cc
@@ -188,20 +188,22 @@ Status BlockingJoinNode::Open(RuntimeState* state) {
Status BlockingJoinNode::ProcessBuildInputAndOpenProbe(
RuntimeState* state, DataSink* build_sink) {
- // If this node is not inside a subplan and can get a thread token, initiate the
+ // If this node is not inside a subplan, we are running with mt_dop=0 (i.e. no
+ // fragment-level multithreading) and can get a thread token, initiate the
// construction of the build-side table in a separate thread, so that the left child
// can do any initialisation in parallel. Otherwise, do this in the main thread.
// Inside a subplan we expect Open() to be called a number of times proportional to the
// input data of the SubplanNode, so we prefer doing processing the build input in the
// main thread, assuming that thread creation is expensive relative to a single subplan
- // iteration. TODO-MT: disable async build thread when mt_dop >= 1.
+ // iteration.
//
// In this block, we also compute the 'overlap' time for the left and right child. This
// is the time (i.e. clock reads) when the right child stops overlapping with the left
// child. For the single threaded case, the left and right child never overlap. For the
// build side in a different thread, the overlap stops when the left child Open()
// returns.
- if (!IsInSubplan() && state->resource_pool()->TryAcquireThreadToken()) {
+ if (!IsInSubplan() && state->query_options().mt_dop == 0
+ && state->resource_pool()->TryAcquireThreadToken()) {
Status build_side_status;
runtime_profile()->AppendExecOption("Join Build-Side Prepared Asynchronously");
string thread_name = Substitute("join-build-thread (finst:$0, plan-node-id:$1)",
diff --git a/be/src/exec/partitioned-hash-join-builder.cc b/be/src/exec/partitioned-hash-join-builder.cc
index b21a033..67b6843 100644
--- a/be/src/exec/partitioned-hash-join-builder.cc
+++ b/be/src/exec/partitioned-hash-join-builder.cc
@@ -95,7 +95,7 @@ Status PhjBuilder::InitExprsAndFilters(RuntimeState* state,
for (const TRuntimeFilterDesc& filter_desc : filter_descs) {
DCHECK(state->query_options().runtime_filter_mode == TRuntimeFilterMode::GLOBAL ||
- filter_desc.is_broadcast_join);
+ filter_desc.is_broadcast_join || state->query_options().num_nodes == 1);
DCHECK(!state->query_options().disable_row_runtime_filtering ||
filter_desc.applied_on_partition_columns);
ScalarExpr* filter_expr;
@@ -105,6 +105,7 @@ Status PhjBuilder::InitExprsAndFilters(RuntimeState* state,
// TODO: Move to Prepare().
filter_ctxs_.emplace_back();
+ // TODO: IMPALA-4400 - implement local aggregation of runtime filters.
filter_ctxs_.back().filter = state->filter_bank()->RegisterFilter(filter_desc, true);
}
return Status::OK();
diff --git a/be/src/runtime/coordinator-backend-state.cc b/be/src/runtime/coordinator-backend-state.cc
index 8db379f..c0dfd8e 100644
--- a/be/src/runtime/coordinator-backend-state.cc
+++ b/be/src/runtime/coordinator-backend-state.cc
@@ -137,7 +137,6 @@ void Coordinator::BackendState::SetRpcParams(const DebugOptions& debug_options,
// Remove filters that weren't selected during filter routing table construction.
// TODO: do this more efficiently, we're looping over the entire plan for each
// instance separately
- DCHECK_EQ(rpc_params->query_ctx.client_request.query_options.mt_dop, 0);
int instance_idx = GetInstanceIdx(params.instance_id);
for (TPlanNode& plan_node: rpc_params->fragment_ctxs.back().fragment.plan.nodes) {
if (!plan_node.__isset.hash_join_node) continue;
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index 998017d..d04d6bc 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -110,10 +110,6 @@ Status Coordinator::Exec() {
const string& str = Substitute("Query $0", PrintId(query_id()));
progress_.Init(str, schedule_.num_scan_ranges());
- // runtime filters not yet supported for mt execution
- bool is_mt_execution = request.query_ctx.client_request.query_options.mt_dop > 0;
- if (is_mt_execution) filter_mode_ = TRuntimeFilterMode::OFF;
-
query_state_ = ExecEnv::GetInstance()->query_exec_mgr()->CreateQueryState(
query_ctx(), schedule_.per_backend_mem_limit());
filter_mem_tracker_ = query_state_->obj_pool()->Add(new MemTracker(
@@ -286,7 +282,6 @@ void Coordinator::ExecSummary::Init(const QuerySchedule& schedule) {
}
void Coordinator::InitFilterRoutingTable() {
- DCHECK(schedule_.request().query_ctx.client_request.query_options.mt_dop == 0);
DCHECK_NE(filter_mode_, TRuntimeFilterMode::OFF)
<< "InitFilterRoutingTable() called although runtime filters are disabled";
DCHECK(!filter_routing_table_complete_)
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index 07e2b9a..12d69eb 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -73,6 +73,7 @@ DECLARE_int32(kudu_mutation_buffer_size);
DECLARE_int32(kudu_error_buffer_size);
DECLARE_int32(hms_event_polling_interval_s);
DECLARE_string(authorization_factory_class);
+DECLARE_bool(unlock_mt_dop);
namespace impala {
@@ -145,6 +146,7 @@ Status GetThriftBackendGflags(JNIEnv* jni_env, jbyteArray* cfg_bytes) {
cfg.__set_hms_event_polling_interval_s(FLAGS_hms_event_polling_interval_s);
cfg.__set_impala_build_version(::GetDaemonBuildVersion());
cfg.__set_authorization_factory_class(FLAGS_authorization_factory_class);
+ cfg.__set_unlock_mt_dop(FLAGS_unlock_mt_dop);
RETURN_IF_ERROR(SerializeThriftMsg(jni_env, &cfg, cfg_bytes));
return Status::OK();
}
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index f252326..2d1b3a5 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -119,4 +119,6 @@ struct TBackendGflags {
46: required string impala_build_version
47: required string authorization_factory_class
+
+ 48: required bool unlock_mt_dop
}
diff --git a/fe/src/main/java/org/apache/impala/planner/Planner.java b/fe/src/main/java/org/apache/impala/planner/Planner.java
index 46371e6..d4cc37b 100644
--- a/fe/src/main/java/org/apache/impala/planner/Planner.java
+++ b/fe/src/main/java/org/apache/impala/planner/Planner.java
@@ -231,8 +231,14 @@ public class Planner {
Preconditions.checkState(ctx_.getQueryOptions().mt_dop > 0);
List<PlanFragment> distrPlan = createPlan();
Preconditions.checkNotNull(distrPlan);
- ParallelPlanner planner = new ParallelPlanner(ctx_);
- List<PlanFragment> parallelPlans = planner.createPlans(distrPlan.get(0));
+ List<PlanFragment> parallelPlans;
+ // TODO: IMPALA-4224: Parallel plans are not executable
+ if (RuntimeEnv.INSTANCE.isTestEnv()) {
+ ParallelPlanner planner = new ParallelPlanner(ctx_);
+ parallelPlans = planner.createPlans(distrPlan.get(0));
+ } else {
+ parallelPlans = Collections.singletonList(distrPlan.get(0));
+ }
// Only use one scanner thread per scan-node instance since intra-node
// parallelism is achieved via multiple fragment instances.
ctx_.getQueryOptions().setNum_scanner_threads(1);
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 6bd4e65..7f8c785 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -67,6 +67,7 @@ import org.apache.impala.common.InternalException;
import org.apache.impala.common.NotImplementedException;
import org.apache.impala.common.Pair;
import org.apache.impala.common.RuntimeEnv;
+import org.apache.impala.service.BackendConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -155,22 +156,22 @@ public class SingleNodePlanner {
* Checks that the given single-node plan is executable:
* - It may not contain right or full outer joins with no equi-join conjuncts that
* are not inside the right child of a SubplanNode.
- * - MT_DOP > 0 is not supported for plans with base table joins or table sinks.
+ * - MT_DOP > 0 is not supported by default for plans with base table joins or table
+ * sinks: we only allow MT_DOP > 0 with such plans if --unlock_mt_dop=true is
+ * specified.
* Throws a NotImplementedException if plan validation fails.
*/
public void validatePlan(PlanNode planNode) throws NotImplementedException {
if (ctx_.getQueryOptions().isSetMt_dop() && ctx_.getQueryOptions().mt_dop > 0
&& !RuntimeEnv.INSTANCE.isTestEnv()
+ && !BackendConfig.INSTANCE.isMtDopUnlocked()
&& (planNode instanceof JoinNode || ctx_.hasTableSink())) {
throw new NotImplementedException(
"MT_DOP not supported for plans with base table joins or table sinks.");
}
- // As long as MT_DOP is unset or 0 any join can run in a single-node plan.
- if (ctx_.isSingleNodeExec() &&
- (!ctx_.getQueryOptions().isSetMt_dop() || ctx_.getQueryOptions().mt_dop == 0)) {
- return;
- }
+ // Any join can run in a single-node plan.
+ if (ctx_.isSingleNodeExec()) return;
if (planNode instanceof NestedLoopJoinNode) {
JoinNode joinNode = (JoinNode) planNode;
diff --git a/fe/src/main/java/org/apache/impala/service/BackendConfig.java b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
index b6a3ab6..5d4d056 100644
--- a/fe/src/main/java/org/apache/impala/service/BackendConfig.java
+++ b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
@@ -138,6 +138,10 @@ public class BackendConfig {
return backendCfg_.getAuthorization_factory_class();
}
+ public boolean isMtDopUnlocked() {
+ return backendCfg_.unlock_mt_dop;
+ }
+
// Inits the auth_to_local configuration in the static KerberosName class.
private static void initAuthToLocal() {
// If auth_to_local is enabled, we read the configuration hadoop.security.auth_to_local
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/mt-dop-validation.test b/testdata/workloads/functional-planner/queries/PlannerTest/mt-dop-validation.test
index c25e99a..9b7098d 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/mt-dop-validation.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/mt-dop-validation.test
@@ -3,7 +3,43 @@ select count(*) from
functional_parquet.alltypestiny a,
functional_parquet.alltypestiny b
---- PLAN
-NotImplementedException: MT_DOP not supported for plans with base table joins or table sinks.
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=2.04GB mem-reservation=16.00KB thread-reservation=1
+PLAN-ROOT SINK
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+03:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=2 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT), 00(OPEN)
+|
+02:NESTED LOOP JOIN [CROSS JOIN]
+| mem-estimate=2.00GB mem-reservation=0B thread-reservation=0
+| tuple-ids=0,1 row-size=0B cardinality=unavailable
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [functional_parquet.alltypestiny b]
+| partitions=4/4 files=4 size=9.92KB
+| stored statistics:
+| table: rows=unavailable size=unavailable
+| partitions: 0/4 rows=unavailable
+| columns: all
+| extrapolated-rows=disabled max-scan-range-rows=unavailable
+| mem-estimate=16.00MB mem-reservation=8.00KB thread-reservation=0
+| tuple-ids=1 row-size=0B cardinality=unavailable
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [functional_parquet.alltypestiny a]
+ partitions=4/4 files=4 size=9.92KB
+ stored statistics:
+ table: rows=unavailable size=unavailable
+ partitions: 0/4 rows=unavailable
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=unavailable
+ mem-estimate=16.00MB mem-reservation=8.00KB thread-reservation=0
+ tuple-ids=0 row-size=0B cardinality=unavailable
+ in pipelines: 00(GETNEXT)
---- PARALLELPLANS
NotImplementedException: MT_DOP not supported for plans with base table joins or table sinks.
====
@@ -13,7 +49,47 @@ functional_parquet.alltypestiny a,
functional_parquet.alltypestiny b
where a.id = b.id
---- PLAN
-NotImplementedException: MT_DOP not supported for plans with base table joins or table sinks.
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=2.03GB mem-reservation=35.01MB thread-reservation=1 runtime-filters-memory=1.00MB
+PLAN-ROOT SINK
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+|
+03:AGGREGATE [FINALIZE]
+| output: count(*)
+| mem-estimate=10.00MB mem-reservation=0B spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=2 row-size=8B cardinality=1
+| in pipelines: 03(GETNEXT), 00(OPEN)
+|
+02:HASH JOIN [INNER JOIN]
+| hash predicates: a.id = b.id
+| fk/pk conjuncts: assumed fk/pk
+| runtime filters: RF000[bloom] <- b.id
+| mem-estimate=2.00GB mem-reservation=34.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=0,1 row-size=8B cardinality=unavailable
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [functional_parquet.alltypestiny b]
+| partitions=4/4 files=4 size=9.92KB
+| stored statistics:
+| table: rows=unavailable size=unavailable
+| partitions: 0/4 rows=unavailable
+| columns: unavailable
+| extrapolated-rows=disabled max-scan-range-rows=unavailable
+| mem-estimate=16.00MB mem-reservation=8.00KB thread-reservation=0
+| tuple-ids=1 row-size=4B cardinality=unavailable
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [functional_parquet.alltypestiny a]
+ partitions=4/4 files=4 size=9.92KB
+ runtime filters: RF000[bloom] -> a.id
+ stored statistics:
+ table: rows=unavailable size=unavailable
+ partitions: 0/4 rows=unavailable
+ columns: unavailable
+ extrapolated-rows=disabled max-scan-range-rows=unavailable
+ mem-estimate=16.00MB mem-reservation=8.00KB thread-reservation=0
+ tuple-ids=0 row-size=4B cardinality=unavailable
+ in pipelines: 00(GETNEXT)
---- PARALLELPLANS
NotImplementedException: MT_DOP not supported for plans with base table joins or table sinks.
====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/joins_mt_dop.test b/testdata/workloads/functional-query/queries/QueryTest/joins_mt_dop.test
new file mode 100644
index 0000000..ec8243a
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/joins_mt_dop.test
@@ -0,0 +1,27 @@
+====
+---- QUERY
+# Test that the right amount of memory is reserved for this plan - the planner assumes
+# that the join build does not run in parallel with the left side of the join.
+set num_nodes=1;
+set runtime_filter_wait_time_ms=$RUNTIME_FILTER_WAIT_TIME_MS;
+set runtime_filter_mode=LOCAL;
+select STRAIGHT_JOIN count(*) from alltypes p join [BROADCAST] alltypestiny b
+on p.month = b.int_col and b.month = 1 and b.string_col = "1"
+---- RESULTS
+620
+---- RUNTIME_PROFILE
+row_regex: .*Files rejected: 22 \(22\).*
+====
+---- QUERY
+# Test that the single-node plan is executable and produces the correct result for a
+# nested loop join mode that can't be executed in a distributed fashion.
+set num_nodes=1;
+select straight_join count(*)
+from functional.alltypestiny a right outer join functional.alltypes b
+ on a.id != b.id or a.int_col < b.int_col
+right outer join functional.alltypesagg c
+ on a.smallint_col >= c.smallint_col
+where a.id < 10 and c.bigint_col = 10
+---- RESULTS
+291990
+====
diff --git a/tests/custom_cluster/test_mt_dop.py b/tests/custom_cluster/test_mt_dop.py
new file mode 100644
index 0000000..b0eb01b
--- /dev/null
+++ b/tests/custom_cluster/test_mt_dop.py
@@ -0,0 +1,62 @@
+# 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 copy import deepcopy
+
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.environ import build_flavor_timeout
+
+WAIT_TIME_MS = build_flavor_timeout(60000, slow_build_timeout=100000)
+
+
+class TestMtDopFlags(CustomClusterTestSuite):
+ @classmethod
+ def get_workload(cls):
+ return 'functional-query'
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestMtDopFlags, cls).add_test_dimensions()
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(impalad_args="--unlock_mt_dop=true")
+ def test_mt_dop_all(self, vector, unique_database):
+ """Test joins, inserts and runtime filters with mt_dop > 0"""
+ vector = deepcopy(vector)
+ vector.get_value('exec_option')['mt_dop'] = 4
+ self.run_test_case('QueryTest/joins', vector, use_db="functional_parquet")
+ self.run_test_case('QueryTest/insert', vector)
+
+ # Runtime filter tests assume 3 fragments, which we can get with one instance per
+ # node.
+ vector.get_value('exec_option')['mt_dop'] = 1
+ self.run_test_case('QueryTest/runtime_filters', vector,
+ test_file_vars={'$RUNTIME_FILTER_WAIT_TIME_MS': str(WAIT_TIME_MS)})
+
+ # Allow test to override num_nodes.
+ del vector.get_value('exec_option')['num_nodes']
+ self.run_test_case('QueryTest/joins_mt_dop', vector,
+ test_file_vars={'$RUNTIME_FILTER_WAIT_TIME_MS': str(WAIT_TIME_MS)})
+
+ @CustomClusterTestSuite.with_args(impalad_args="--unlock_mt_dop=true", cluster_size=1)
+ def test_mt_dop_runtime_filters_one_node(self, vector, unique_database):
+ """Runtime filter tests, which assume 3 fragment instances, can also be run on a single
+ node cluster to test multiple filter sources/destinations per backend."""
+ vector.get_value('exec_option')['mt_dop'] = 3
+ self.run_test_case('QueryTest/runtime_filters', vector,
+ test_file_vars={'$RUNTIME_FILTER_WAIT_TIME_MS': str(WAIT_TIME_MS)})