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/09/27 04:24:05 UTC
[impala] 02/02: IMPALA-8928: Add MEM_LIMIT_EXECUTORS query option
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 104a454d3e88d96ec28837e1032de8b292daa84d
Author: Bikramjeet Vig <bi...@cloudera.com>
AuthorDate: Fri Sep 6 12:54:32 2019 -0700
IMPALA-8928: Add MEM_LIMIT_EXECUTORS query option
This developer only option provides a way to set the memory limit for
only the executors which would be useful for writing tests in the
future.
Testing:
- Added a simple sanity check
Change-Id: I20dfd4e8fc7ffd9130db9f942efa78965c724e18
Reviewed-on: http://gerrit.cloudera.org:8080/14294
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/scheduling/query-schedule.cc | 19 +++++++++++++------
be/src/service/query-options-test.cc | 1 +
be/src/service/query-options.cc | 8 ++++++++
be/src/service/query-options.h | 5 +++--
common/thrift/ImpalaInternalService.thrift | 3 +++
common/thrift/ImpalaService.thrift | 8 ++++++++
tests/custom_cluster/test_admission_controller.py | 19 ++++++++++++++++++-
7 files changed, 54 insertions(+), 9 deletions(-)
diff --git a/be/src/scheduling/query-schedule.cc b/be/src/scheduling/query-schedule.cc
index 86fe8fd..a86be2b 100644
--- a/be/src/scheduling/query-schedule.cc
+++ b/be/src/scheduling/query-schedule.cc
@@ -279,14 +279,14 @@ void QuerySchedule::UpdateMemoryRequirements(const TPoolConfig& pool_cfg) {
per_backend_mem_to_admit_ = 0;
coord_backend_mem_to_admit_ = 0;
- bool has_query_option = false;
+ bool is_mem_limit_set = false;
if (query_options().__isset.mem_limit && query_options().mem_limit > 0) {
per_backend_mem_to_admit_ = query_options().mem_limit;
coord_backend_mem_to_admit_ = query_options().mem_limit;
- has_query_option = true;
+ is_mem_limit_set = true;
}
- if (!has_query_option) {
+ if (!is_mem_limit_set) {
per_backend_mem_to_admit_ = GetPerExecutorMemoryEstimate();
coord_backend_mem_to_admit_ = use_dedicated_coord_estimates ?
GetDedicatedCoordMemoryEstimate() :
@@ -302,11 +302,11 @@ void QuerySchedule::UpdateMemoryRequirements(const TPoolConfig& pool_cfg) {
}
}
- if (!has_query_option || pool_cfg.clamp_mem_limit_query_option) {
+ if (!is_mem_limit_set || pool_cfg.clamp_mem_limit_query_option) {
if (pool_cfg.min_query_mem_limit > 0) {
per_backend_mem_to_admit_ =
max(per_backend_mem_to_admit_, pool_cfg.min_query_mem_limit);
- if (!use_dedicated_coord_estimates || has_query_option) {
+ if (!use_dedicated_coord_estimates || is_mem_limit_set) {
// The minimum mem limit option does not apply to dedicated coordinators -
// this would result in over-reserving of memory. Treat coordinator and
// executor mem limits the same if the query option was explicitly set.
@@ -332,13 +332,20 @@ void QuerySchedule::UpdateMemoryRequirements(const TPoolConfig& pool_cfg) {
per_backend_mem_to_admit_ = 0;
}
- if (mimic_old_behaviour && !has_query_option) {
+ if (mimic_old_behaviour && !is_mem_limit_set) {
per_backend_mem_limit_ = -1;
coord_backend_mem_limit_ = -1;
} else {
per_backend_mem_limit_ = per_backend_mem_to_admit_;
coord_backend_mem_limit_ = coord_backend_mem_to_admit_;
}
+
+ // Finally, enforce the MEM_LIMIT_EXECUTORS query option if MEM_LIMIT is not specified.
+ if (!is_mem_limit_set && query_options().__isset.mem_limit_executors
+ && query_options().mem_limit_executors > 0) {
+ per_backend_mem_to_admit_ = query_options().mem_limit_executors;
+ per_backend_mem_limit_ = per_backend_mem_to_admit_;
+ }
}
void QuerySchedule::set_executor_group(string executor_group) {
diff --git a/be/src/service/query-options-test.cc b/be/src/service/query-options-test.cc
index 30d7791..e587103 100644
--- a/be/src/service/query-options-test.cc
+++ b/be/src/service/query-options-test.cc
@@ -150,6 +150,7 @@ TEST(QueryOptions, SetByteOptions) {
{MAKE_OPTIONDEF(max_mem_estimate_for_admission), {-1, I64_MAX}},
{MAKE_OPTIONDEF(scan_bytes_limit), {-1, I64_MAX}},
{MAKE_OPTIONDEF(topn_bytes_limit), {-1, I64_MAX}},
+ {MAKE_OPTIONDEF(mem_limit_executors), {-1, I64_MAX}},
};
vector<pair<OptionDef<int32_t>, Range<int32_t>>> case_set_i32{
{MAKE_OPTIONDEF(runtime_filter_min_size),
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index cd7007b..5d1ab07 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -899,6 +899,14 @@ Status impala::SetQueryOption(const string& key, const string& value,
parquet_object_store_split_size);
break;
}
+ case TImpalaQueryOptions::MEM_LIMIT_EXECUTORS: {
+ // Parse the mem limit spec and validate it.
+ int64_t bytes_limit;
+ RETURN_IF_ERROR(
+ ParseMemValue(value, "query memory limit for executors", &bytes_limit));
+ query_options->__set_mem_limit_executors(bytes_limit);
+ break;
+ }
default:
if (IsRemovedQueryOption(key)) {
LOG(WARNING) << "Ignoring attempt to set removed query option '" << key << "'";
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index baabd6a..01112ad 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -47,7 +47,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
// time we add or remove a query option to/from the enum TImpalaQueryOptions.
#define QUERY_OPTS_TABLE\
DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(),\
- TImpalaQueryOptions::PARQUET_OBJECT_STORE_SPLIT_SIZE + 1);\
+ TImpalaQueryOptions::MEM_LIMIT_EXECUTORS + 1);\
REMOVED_QUERY_OPT_FN(abort_on_default_limit_exceeded, ABORT_ON_DEFAULT_LIMIT_EXCEEDED)\
QUERY_OPT_FN(abort_on_error, ABORT_ON_ERROR, TQueryOptionLevel::REGULAR)\
REMOVED_QUERY_OPT_FN(allow_unsupported_formats, ALLOW_UNSUPPORTED_FORMATS)\
@@ -191,7 +191,8 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
TQueryOptionLevel::ADVANCED)\
QUERY_OPT_FN(now_string, NOW_STRING, TQueryOptionLevel::DEVELOPMENT)\
QUERY_OPT_FN(parquet_object_store_split_size, PARQUET_OBJECT_STORE_SPLIT_SIZE,\
- TQueryOptionLevel::ADVANCED)
+ TQueryOptionLevel::ADVANCED)\
+ QUERY_OPT_FN(mem_limit_executors, MEM_LIMIT_EXECUTORS, TQueryOptionLevel::DEVELOPMENT)
;
/// Enforce practical limits on some query options to avoid undesired query state.
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index d52e2ca..48e89fc 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -400,6 +400,9 @@ struct TQueryOptions {
// See comment in ImpalaService.thrift
96: optional i64 parquet_object_store_split_size = 268435456;
+
+ // See comment in ImpalaService.thrift
+ 97: optional i64 mem_limit_executors = 0
}
// Impala currently has two types of sessions: Beeswax and HiveServer2
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 8ba1066..2006f97 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -475,6 +475,14 @@ enum TImpalaQueryOptions {
// Parquet files written by Impala (Impala writes Parquet files with a single row
// group per file). Must be >= 1 MB.
PARQUET_OBJECT_STORE_SPLIT_SIZE = 95
+
+ // For testing purposes only. A per executor approximate limit on the memory consumption
+ // of this query. Only applied if MEM_LIMIT is not specified.
+ // unspecified or a limit of 0 means no limit;
+ // Otherwise specified either as:
+ // a) an int (= number of bytes);
+ // b) a float followed by "M" (MB) or "G" (GB)
+ MEM_LIMIT_EXECUTORS = 96
}
// The summary of a DML statement.
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index 4698e77..27c3182 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -630,7 +630,7 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
"""Planner tests to add coverage for coordinator estimates when using dedicated
coordinators. Also includes coverage for verifying cluster memory admitted."""
vector_copy = copy(vector)
- exec_options = vector.get_value('exec_option')
+ exec_options = vector_copy.get_value('exec_option')
# Remove num_nodes from the options to allow test case runner to set it in one of
# the test cases.
del exec_options['num_nodes']
@@ -638,6 +638,23 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
self.run_test_case('QueryTest/dedicated-coord-mem-estimates', vector_copy,
unique_database)
+ @SkipIfNotHdfsMinicluster.tuned_for_minicluster
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(num_exclusive_coordinators=1, cluster_size=2)
+ def test_mem_limit_executors(self, vector, unique_database):
+ """Verify that the query option mem_limit_executors is only enforced on the
+ executors."""
+ expected_exec_mem_limit = "999999999"
+ ImpalaTestSuite.change_database(self.client, vector.get_value('table_format'))
+ self.client.set_configuration({"MEM_LIMIT_EXECUTORS": expected_exec_mem_limit})
+ handle = self.client.execute_async(QUERY.format(1))
+ self.client.wait_for_finished_timeout(handle, 1000)
+ expected_mem_limits = self.__get_mem_limits_admission_debug_page()
+ assert expected_mem_limits['executor'] > expected_mem_limits[
+ 'coordinator'], expected_mem_limits
+ assert expected_mem_limits['executor'] == float(
+ expected_exec_mem_limit), expected_mem_limits
+
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=impalad_admission_ctrl_flags(max_requests=2, max_queued=1,