You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by as...@apache.org on 2020/06/30 22:24:07 UTC
[impala] branch master updated (2a48f7d -> b142d03)
This is an automated email from the ASF dual-hosted git repository.
asherman pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from 2a48f7d IMPALA-9890 (Part 1): Add more TPCDS queries to Impala's test suite
new 36ced94 IMPALA-9907: Fix NullPointerException in ParallelFileMetadataLoader's load() method
new 8aeb282 IMPALA-9540 Test that Impala Shell no longer sends duplicate "Host" headers in http mode.
new b142d03 IMPALA-9692 (part 3): Model QuerySchedule as a protobuf
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/runtime/CMakeLists.txt | 1 +
.../runtime/coordinator-backend-resource-state.cc | 5 +-
be/src/runtime/coordinator-backend-state-test.cc | 65 ++---
be/src/runtime/coordinator-backend-state.cc | 137 +++++----
be/src/runtime/coordinator-backend-state.h | 45 +--
be/src/runtime/coordinator.cc | 134 ++++-----
be/src/runtime/coordinator.h | 19 +-
be/src/runtime/fragment-instance-state.cc | 1 -
be/src/runtime/query-driver.cc | 2 +-
be/src/runtime/query-exec-params.cc | 71 +++++
be/src/runtime/query-exec-params.h | 78 ++++++
be/src/scheduling/CMakeLists.txt | 4 +
be/src/scheduling/admission-controller-test.cc | 76 +++--
be/src/scheduling/admission-controller.cc | 161 ++++++-----
be/src/scheduling/admission-controller.h | 56 +++-
be/src/scheduling/cluster-membership-mgr-test.cc | 20 +-
be/src/scheduling/cluster-membership-mgr.cc | 12 +-
be/src/scheduling/cluster-membership-mgr.h | 4 +-
be/src/scheduling/cluster-membership-test-util.cc | 2 +
be/src/scheduling/query-schedule.cc | 217 +++++++--------
be/src/scheduling/query-schedule.h | 306 +++++++++------------
be/src/scheduling/scheduler-test-util.cc | 20 +-
be/src/scheduling/scheduler-test-util.h | 2 +-
be/src/scheduling/scheduler.cc | 191 +++++++------
be/src/scheduling/scheduler.h | 8 +-
be/src/service/client-request-state.cc | 17 +-
be/src/service/client-request-state.h | 10 +-
be/src/service/impala-http-handler.cc | 2 +-
be/src/service/impala-server.cc | 22 +-
be/src/service/impala-server.h | 7 +-
be/src/util/debug-util.h | 4 +
be/src/util/uid-util.h | 8 +
common/protobuf/CMakeLists.txt | 3 +-
common/protobuf/admission_control_service.proto | 154 +++++++++++
.../impala/catalog/ParallelFileMetadataLoader.java | 11 +-
tests/shell/test_shell_interactive.py | 76 ++---
tests/shell/util.py | 13 +-
37 files changed, 1146 insertions(+), 818 deletions(-)
create mode 100644 be/src/runtime/query-exec-params.cc
create mode 100644 be/src/runtime/query-exec-params.h
create mode 100644 common/protobuf/admission_control_service.proto
[impala] 02/03: IMPALA-9540 Test that Impala Shell no longer sends
duplicate "Host" headers in http mode.
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asherman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 8aeb28287f40987a755d5352a51cef24cf1f6c74
Author: Andrew Sherman <as...@cloudera.com>
AuthorDate: Tue Mar 24 18:10:36 2020 -0700
IMPALA-9540 Test that Impala Shell no longer sends duplicate "Host" headers in http mode.
Many http servers will not accept an http request that has multiple
copies of the "Host" header. A recent toolchain change patches
Thrift so that will not send the extraneous header (in THttpClient).
This change tests that the duplicate headers are not sent,
TESTING:
Ran all end-to-end tests.
Rewrote an existing Shell test to check that only one "Host" header
is sent.
Change-Id: I82996015d0205923e854dac8bb88604778684c46
Reviewed-on: http://gerrit.cloudera.org:8080/15752
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/shell/test_shell_interactive.py | 76 +++++++++++++++++++----------------
tests/shell/util.py | 13 +++++-
2 files changed, 54 insertions(+), 35 deletions(-)
diff --git a/tests/shell/test_shell_interactive.py b/tests/shell/test_shell_interactive.py
index 84e7036..245fb9b 100755
--- a/tests/shell/test_shell_interactive.py
+++ b/tests/shell/test_shell_interactive.py
@@ -29,7 +29,6 @@ import socket
import sys
import threading
from time import sleep
-from contextlib import closing
# This import is the actual ImpalaShell class from impala_shell.py.
# We rename it to ImpalaShellClass here because we later import another
@@ -42,6 +41,7 @@ from tests.common.impala_service import ImpaladService
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.skip import SkipIfLocal
from tests.common.test_dimensions import create_client_protocol_dimension
+from tests.shell.util import get_unused_port
from util import (assert_var_substitution, ImpalaShell, get_impalad_port, get_shell_cmd,
get_open_sessions_metric, IMPALA_SHELL_EXECUTABLE, spawn_shell)
import SimpleHTTPServer
@@ -76,18 +76,41 @@ def tmp_history_file(request):
return tmp.name
-class UnavailableRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
- """An HTTP server that always returns 503"""
- def do_POST(self):
- self.send_response(code=httplib.SERVICE_UNAVAILABLE, message="Service Unavailable")
+@pytest.yield_fixture
+def http_503_server():
+ class RequestHandler503(SimpleHTTPServer.SimpleHTTPRequestHandler):
+ """A custom http handler that checks for duplicate 'Host' headers from the most
+ recent http request, and always returns a 503 http code"""
+ def do_POST(self):
+ # The unfortunately named self.headers here is an instance of mimetools.Message that
+ # contains the request headers.
+ request_headers = self.headers.headers
-def get_unused_port():
- """ Find an unused port http://stackoverflow.com/questions/1365265 """
- with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
- s.bind(('', 0))
- s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- return s.getsockname()[1]
+ # Ensure that only one 'Host' header is contained in the request before responding.
+ host_hdr_count = sum([header.startswith('Host:') for header in request_headers])
+ assert host_hdr_count == 1, "duplicate 'Host:' headers in %s" % request_headers
+
+ # Respond with 503.
+ self.send_response(code=httplib.SERVICE_UNAVAILABLE, message="Service Unavailable")
+
+ class TestHTTPServer503(object):
+ def __init__(self):
+ self.HOST = "localhost"
+ self.PORT = get_unused_port()
+ self.httpd = SocketServer.TCPServer((self.HOST, self.PORT), RequestHandler503)
+
+ self.http_server_thread = threading.Thread(target=self.httpd.serve_forever)
+ self.http_server_thread.start()
+
+ server = TestHTTPServer503()
+ yield server
+
+ # Cleanup after test.
+ if server.httpd is not None:
+ server.httpd.shutdown()
+ if server.http_server_thread is not None:
+ server.http_server_thread.join()
class TestImpalaShellInteractive(ImpalaTestSuite):
@@ -990,33 +1013,18 @@ class TestImpalaShellInteractive(ImpalaTestSuite):
result = p.get_result()
assert "Fetched 0 row" in result.stderr
- def test_http_codes(self, vector):
- """Check that the shell prints a good message when using hs2-http protocol
- and the http server returns a 503 error."""
+ def test_http_interactions(self, vector, http_503_server):
+ """Test interactions with the http server when using hs2-http protocol.
+ Check that the shell prints a good message when the server returns a 503 error."""
protocol = vector.get_value("protocol")
if protocol != 'hs2-http':
pytest.skip()
- # Start an http server that always returns 503.
- HOST = "localhost"
- PORT = get_unused_port()
- httpd = None
- http_server_thread = None
- try:
- httpd = SocketServer.TCPServer((HOST, PORT), UnavailableRequestHandler)
- http_server_thread = threading.Thread(target=httpd.serve_forever)
- http_server_thread.start()
-
- # Check that we get a message about the 503 error when we try to connect.
- shell_args = ["--protocol={0}".format(protocol), "-i{0}:{1}".format(HOST, PORT)]
- shell_proc = spawn_shell([IMPALA_SHELL_EXECUTABLE] + shell_args)
- shell_proc.expect("HTTP code 503", timeout=10)
- finally:
- # Clean up.
- if httpd is not None:
- httpd.shutdown()
- if http_server_thread is not None:
- http_server_thread.join()
+ # Check that we get a message about the 503 error when we try to connect.
+ shell_args = ["--protocol={0}".format(protocol),
+ "-i{0}:{1}".format(http_503_server.HOST, http_503_server.PORT)]
+ shell_proc = spawn_shell([IMPALA_SHELL_EXECUTABLE] + shell_args)
+ shell_proc.expect("HTTP code 503", timeout=10)
def run_impala_shell_interactive(vector, input_lines, shell_args=None,
diff --git a/tests/shell/util.py b/tests/shell/util.py
index e48e8fa..2f35324 100755
--- a/tests/shell/util.py
+++ b/tests/shell/util.py
@@ -20,6 +20,9 @@
import logging
import os
+import socket
+from contextlib import closing
+
import pexpect
import pytest
import re
@@ -55,7 +58,7 @@ if IMPALA_SHELL_EXECUTABLE is None:
IMPALA_HOME, "shell/build", "impala-shell-" + IMPALA_LOCAL_BUILD_VERSION,
"impala-shell")
-
+
def build_shell_env(env=None):
""" Construct the environment for the shell to run in based on 'env', or the current
process's environment if env is None."""
@@ -311,3 +314,11 @@ class ImpalaShell(object):
if args is not None: cmd += args
return Popen(cmd, shell=False, stdout=PIPE, stdin=PIPE, stderr=PIPE,
env=build_shell_env(env))
+
+
+def get_unused_port():
+ """ Find an unused port http://stackoverflow.com/questions/1365265 """
+ with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+ s.bind(('', 0))
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ return s.getsockname()[1]
[impala] 03/03: IMPALA-9692 (part 3): Model QuerySchedule as a
protobuf
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asherman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit b142d0324cb155fd8dbda704ed569583eeed8af5
Author: Thomas Tauber-Marshall <tm...@cloudera.com>
AuthorDate: Mon Feb 24 13:48:16 2020 -0800
IMPALA-9692 (part 3): Model QuerySchedule as a protobuf
In order to support the new admission control service, we need to be
able to return the results of an admission attempt, i.e. the query
schedule, to the coordinator.
To enable this, this patch moves all parts of the QuerySchedule class
and related classes that are required by the coordinator into a new
message QuerySchedulePB. The main admission control interface,
SubmitForAdmission(), now returns a QuerySchedulePB.
Some notable changes:
- Previously, QuerySchedule was used by Coordinator as a way to pass
around references to parts of the TExecRequest to places like
Coordinator::ExecSummary and Coordinator::BackendState. This has
been replaced with the ExecParams class, which is a container for
references to the TExecRequest and QuerySchedulePB along with
convenience functions for accessing them.
- Similarly, FragmentExecParams, which is part of QuerySchedule,
contains references to the associated TPlanFragment, owned by the
TExecRequest, which were used by the Coordinator when iterating over
the schedule to initiate the query. Since FragmentExecParamsPB can't
contain these references, they were replaced by a map between
fragment idx and TPlanFragment in ExecParams.
- In order to keep payloads reasonable for the eventual RPC interface,
AdmissionController::ReleaseQuery() and ReleaseQueryBackend() now
take a query id as a parameter instead of a QuerySchedule. To
facilitate this, AdmissionController now maintains a map from query
ids of running queries to the resources that were allocated for them
so that it can look the resources up when releasing them. This map
will be necessary when implementing the admission control service to
facilitate proper accounting of resouces in cases like coordinator
failures.
- As scheduling is currently organized, we first construct the
FragmentExecParams with the FInstanceExecParams as their children,
then we construct the BackendExecParams which get references to
their FInstanceExecParams. Since we can't send references like these
through an rpc, we now instead Swap() the FInstanceExecParamsPB
into the BackendExecParamsPB.
Testing:
- Updated related tests.
- Passed a full run of existing tests.
Change-Id: I1db64e72f84604b1d8ac24e0bdd4ad6bedd6bcd9
Reviewed-on: http://gerrit.cloudera.org:8080/15961
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/CMakeLists.txt | 1 +
.../runtime/coordinator-backend-resource-state.cc | 5 +-
be/src/runtime/coordinator-backend-state-test.cc | 65 ++---
be/src/runtime/coordinator-backend-state.cc | 137 +++++----
be/src/runtime/coordinator-backend-state.h | 45 +--
be/src/runtime/coordinator.cc | 134 ++++-----
be/src/runtime/coordinator.h | 19 +-
be/src/runtime/fragment-instance-state.cc | 1 -
be/src/runtime/query-driver.cc | 2 +-
be/src/runtime/query-exec-params.cc | 71 +++++
be/src/runtime/query-exec-params.h | 78 ++++++
be/src/scheduling/CMakeLists.txt | 4 +
be/src/scheduling/admission-controller-test.cc | 76 +++--
be/src/scheduling/admission-controller.cc | 161 ++++++-----
be/src/scheduling/admission-controller.h | 56 +++-
be/src/scheduling/cluster-membership-mgr-test.cc | 20 +-
be/src/scheduling/cluster-membership-mgr.cc | 12 +-
be/src/scheduling/cluster-membership-mgr.h | 4 +-
be/src/scheduling/cluster-membership-test-util.cc | 2 +
be/src/scheduling/query-schedule.cc | 217 +++++++--------
be/src/scheduling/query-schedule.h | 306 +++++++++------------
be/src/scheduling/scheduler-test-util.cc | 20 +-
be/src/scheduling/scheduler-test-util.h | 2 +-
be/src/scheduling/scheduler.cc | 191 +++++++------
be/src/scheduling/scheduler.h | 8 +-
be/src/service/client-request-state.cc | 17 +-
be/src/service/client-request-state.h | 10 +-
be/src/service/impala-http-handler.cc | 2 +-
be/src/service/impala-server.cc | 22 +-
be/src/service/impala-server.h | 7 +-
be/src/util/debug-util.h | 4 +
be/src/util/uid-util.h | 8 +
common/protobuf/CMakeLists.txt | 3 +-
common/protobuf/admission_control_service.proto | 154 +++++++++++
34 files changed, 1085 insertions(+), 779 deletions(-)
diff --git a/be/src/runtime/CMakeLists.txt b/be/src/runtime/CMakeLists.txt
index 6195ba6..cff6b3f 100644
--- a/be/src/runtime/CMakeLists.txt
+++ b/be/src/runtime/CMakeLists.txt
@@ -60,6 +60,7 @@ add_library(Runtime
mem-pool.cc
query-driver.cc
query-exec-mgr.cc
+ query-exec-params.cc
query-state.cc
test-env.cc
types.cc
diff --git a/be/src/runtime/coordinator-backend-resource-state.cc b/be/src/runtime/coordinator-backend-resource-state.cc
index b721199..987fb9d 100644
--- a/be/src/runtime/coordinator-backend-resource-state.cc
+++ b/be/src/runtime/coordinator-backend-resource-state.cc
@@ -35,11 +35,10 @@ DECLARE_uint32(batched_release_decay_factor);
DECLARE_uint64(release_backend_states_delay_ms);
Coordinator::BackendResourceState::BackendResourceState(
- const vector<BackendState*>& backend_states, const QuerySchedule& schedule)
+ const vector<BackendState*>& backend_states)
: num_in_use_(backend_states.size()),
backend_states_(backend_states),
num_backends_(backend_states.size()),
- schedule_(schedule),
release_backend_states_delay_ns_(FLAGS_release_backend_states_delay_ms * 1000000),
batched_release_decay_value_(FLAGS_batched_release_decay_factor) {
DCHECK_GT(batched_release_decay_value_, 0)
@@ -123,7 +122,7 @@ void Coordinator::BackendResourceState::BackendsReleased(
backend_resource_states_.at(backend_state) = ResourceState::RELEASED;
// If the Backend running the Coordinator has completed and been released, set
// released_coordinator_ to true.
- if (backend_state->exec_params()->is_coord_backend) {
+ if (backend_state->exec_params().is_coord_backend()) {
released_coordinator_ = true;
}
}
diff --git a/be/src/runtime/coordinator-backend-state-test.cc b/be/src/runtime/coordinator-backend-state-test.cc
index 806c626..a731772 100644
--- a/be/src/runtime/coordinator-backend-state-test.cc
+++ b/be/src/runtime/coordinator-backend-state-test.cc
@@ -36,41 +36,26 @@ class CoordinatorBackendStateTest : public testing::Test {
// Pool for objects to be destroyed during test teardown.
ObjectPool pool_;
- RuntimeProfile* profile_ = nullptr;
-
- virtual void SetUp() { profile_ = RuntimeProfile::Create(&pool_, "pool1"); }
-
- /// Utility function to create a dummy QuerySchedule.
- void MakeQuerySchedule(QuerySchedule** query_schedule) {
- TUniqueId* query_id = pool_.Add(new TUniqueId());
- TQueryExecRequest* request = pool_.Add(new TQueryExecRequest());
- TQueryOptions* query_options = pool_.Add(new TQueryOptions());
-
- *query_schedule =
- pool_.Add(new QuerySchedule(*query_id, *request, *query_options, profile_));
- }
-
/// Utility function to create a specified number of dummy BackendStates and add them
/// to the supplied vector. 'coordinator_backend' points to the Backend running the
/// Coordinator. The 'coordinator_backend' is added to the supplied vector as well.
- void MakeBackendStates(int num_states, QuerySchedule* query_schedule,
- Coordinator::BackendState** coordinator_backend,
+ void MakeBackendStates(int num_states, Coordinator::BackendState** coordinator_backend,
std::vector<Coordinator::BackendState*>* backend_states) {
- TQueryCtx* query_ctx = pool_.Add(new TQueryCtx());
-
- PerBackendExecParams* per_backend_exec_params = pool_.Add(new PerBackendExecParams());
- query_schedule->set_per_backend_exec_params(*per_backend_exec_params);
+ UniqueIdPB* query_id = pool_.Add(new UniqueIdPB());
+ TExecRequest* request = pool_.Add(new TExecRequest());
+ QuerySchedulePB* query_schedule = pool_.Add(new QuerySchedulePB());
+ QueryExecParams* exec_params =
+ pool_.Add(new QueryExecParams(*query_id, *request, *query_schedule));
for (int i = 0; i < num_states; ++i) {
- TNetworkAddress addr = MakeNetworkAddress(strings::Substitute("host-$0", i), 25000);
- BackendExecParams* backend_exec_params = pool_.Add(new BackendExecParams());
- per_backend_exec_params->emplace(addr, *backend_exec_params);
+ BackendExecParamsPB* backend_exec_params =
+ query_schedule->add_backend_exec_params();
Coordinator::BackendState* backend_state = pool_.Add(new Coordinator::BackendState(
- *query_schedule, *query_ctx, 0, TRuntimeFilterMode::OFF, *backend_exec_params));
+ *exec_params, i, TRuntimeFilterMode::OFF, *backend_exec_params));
backend_states->push_back(backend_state);
// Mark the first BackendState as the Coordinator Backend.
if (i == 0) {
- backend_exec_params->is_coord_backend = true;
+ backend_exec_params->set_is_coord_backend(true);
*coordinator_backend = backend_state;
}
}
@@ -86,14 +71,11 @@ class CoordinatorBackendStateTest : public testing::Test {
TEST_F(CoordinatorBackendStateTest, StateMachine) {
// Initialize a BackendResourceState with three BackendStates.
int num_backends = 3;
- QuerySchedule* query_schedule = nullptr;
- MakeQuerySchedule(&query_schedule);
-
std::vector<Coordinator::BackendState*> backend_states;
Coordinator::BackendState* coordinator_backend = nullptr;
- MakeBackendStates(num_backends, query_schedule, &coordinator_backend, &backend_states);
+ MakeBackendStates(num_backends, &coordinator_backend, &backend_states);
Coordinator::BackendResourceState* backend_resource_state =
- pool_.Add(new Coordinator::BackendResourceState(backend_states, *query_schedule));
+ pool_.Add(new Coordinator::BackendResourceState(backend_states));
// Assert that all BackendStates are initially in the IN_USE state.
ASSERT_EQ(backend_resource_state->backend_resource_states_.size(), num_backends);
@@ -125,19 +107,16 @@ TEST_F(CoordinatorBackendStateTest, StateMachine) {
TEST_F(CoordinatorBackendStateTest, CoordinatorOnly) {
// Initialize a BackendResourceState with eight BackendStates.
int num_backends = 8;
- QuerySchedule* query_schedule = nullptr;
- MakeQuerySchedule(&query_schedule);
-
std::vector<Coordinator::BackendState*> backend_states;
Coordinator::BackendState* coordinator_backend = nullptr;
- MakeBackendStates(num_backends, query_schedule, &coordinator_backend, &backend_states);
+ MakeBackendStates(num_backends, &coordinator_backend, &backend_states);
Coordinator::BackendResourceState* backend_resource_state =
- pool_.Add(new Coordinator::BackendResourceState(backend_states, *query_schedule));
+ pool_.Add(new Coordinator::BackendResourceState(backend_states));
// Create a vector of non-Coordinator Backends.
std::vector<Coordinator::BackendState*> non_coord_backend_states;
for (auto backend_state : backend_states) {
- if (!backend_state->exec_params()->is_coord_backend) {
+ if (!backend_state->exec_params().is_coord_backend()) {
non_coord_backend_states.push_back(backend_state);
}
}
@@ -175,14 +154,11 @@ TEST_F(CoordinatorBackendStateTest, CoordinatorOnly) {
TEST_F(CoordinatorBackendStateTest, TimedRelease) {
// Initialize a BackendResourceState with eight BackendStates.
int num_backends = 8;
- QuerySchedule* query_schedule = nullptr;
- MakeQuerySchedule(&query_schedule);
-
std::vector<Coordinator::BackendState*> backend_states;
Coordinator::BackendState* coordinator_backend = nullptr;
- MakeBackendStates(num_backends, query_schedule, &coordinator_backend, &backend_states);
+ MakeBackendStates(num_backends, &coordinator_backend, &backend_states);
Coordinator::BackendResourceState* backend_resource_state =
- pool_.Add(new Coordinator::BackendResourceState(backend_states, *query_schedule));
+ pool_.Add(new Coordinator::BackendResourceState(backend_states));
// Sleep until the 'Timed Release' timeout is hit.
SleepForMs(timeout_release_ms_);
@@ -226,14 +202,11 @@ TEST_F(CoordinatorBackendStateTest, TimedRelease) {
TEST_F(CoordinatorBackendStateTest, BatchedRelease) {
// Initialize a BackendResouceState with 128 BackendStates.
int num_backends = 128;
- QuerySchedule* query_schedule = nullptr;
- MakeQuerySchedule(&query_schedule);
-
std::vector<Coordinator::BackendState*> backend_states;
Coordinator::BackendState* coordinator_backend = nullptr;
- MakeBackendStates(num_backends, query_schedule, &coordinator_backend, &backend_states);
+ MakeBackendStates(num_backends, &coordinator_backend, &backend_states);
Coordinator::BackendResourceState* backend_resource_state =
- pool_.Add(new Coordinator::BackendResourceState(backend_states, *query_schedule));
+ pool_.Add(new Coordinator::BackendResourceState(backend_states));
// Sleep until the 'Timed Release' timeout is hit.
SleepForMs(timeout_release_ms_);
diff --git a/be/src/runtime/coordinator-backend-state.cc b/be/src/runtime/coordinator-backend-state.cc
index db24acb..4bb199c 100644
--- a/be/src/runtime/coordinator-backend-state.cc
+++ b/be/src/runtime/coordinator-backend-state.cc
@@ -64,18 +64,17 @@ namespace impala {
const char* Coordinator::BackendState::InstanceStats::LAST_REPORT_TIME_DESC =
"Last report received time";
-Coordinator::BackendState::BackendState(const QuerySchedule& schedule,
- const TQueryCtx& query_ctx, int state_idx, TRuntimeFilterMode::type filter_mode,
- const BackendExecParams& exec_params)
- : schedule_(schedule),
+Coordinator::BackendState::BackendState(const QueryExecParams& exec_params, int state_idx,
+ TRuntimeFilterMode::type filter_mode, const BackendExecParamsPB& backend_exec_params)
+ : exec_params_(exec_params),
state_idx_(state_idx),
filter_mode_(filter_mode),
- backend_exec_params_(&exec_params),
- host_(backend_exec_params_->be_desc.address()),
- krpc_host_(backend_exec_params_->be_desc.krpc_address()),
- query_ctx_(query_ctx),
- query_id_(schedule.query_id()),
- num_remaining_instances_(backend_exec_params_->instance_params.size()) {}
+ backend_exec_params_(backend_exec_params),
+ host_(backend_exec_params_.address()),
+ krpc_host_(backend_exec_params_.krpc_address()),
+ query_ctx_(exec_params_.query_exec_request().query_ctx),
+ query_id_(exec_params_.query_id()),
+ num_remaining_instances_(backend_exec_params_.instance_params().size()) {}
void Coordinator::BackendState::Init(const vector<FragmentStats*>& fragment_stats,
RuntimeProfile* host_profile_parent, ObjectPool* obj_pool) {
@@ -83,16 +82,14 @@ void Coordinator::BackendState::Init(const vector<FragmentStats*>& fragment_stat
host_profile_parent->AddChild(host_profile_);
RuntimeProfile::Counter* admission_slots =
ADD_COUNTER(host_profile_, "AdmissionSlots", TUnit::UNIT);
- admission_slots->Set(backend_exec_params_->slots_to_use);
+ admission_slots->Set(backend_exec_params_.slots_to_use());
// populate instance_stats_map_ and install instance
// profiles as child profiles in fragment_stats' profile
int prev_fragment_idx = -1;
- for (const FInstanceExecParams* instance_params:
- backend_exec_params_->instance_params) {
- DCHECK_EQ(
- FromNetworkAddressPB(host_), instance_params->host); // all hosts must be the same
- int fragment_idx = instance_params->fragment().idx;
+ for (const FInstanceExecParamsPB& instance_params :
+ backend_exec_params_.instance_params()) {
+ int fragment_idx = instance_params.fragment_idx();
DCHECK_LT(fragment_idx, fragment_stats.size());
if (prev_fragment_idx != -1 && fragment_idx != prev_fragment_idx) {
// all instances of a fragment are contiguous
@@ -101,10 +98,11 @@ void Coordinator::BackendState::Init(const vector<FragmentStats*>& fragment_stat
}
fragments_.insert(fragment_idx);
+ const TPlanFragment* fragment = exec_params_.GetFragments()[fragment_idx];
+ InstanceStats* instance_stats = obj_pool->Add(new InstanceStats(
+ instance_params, fragment, host_, fragment_stats[fragment_idx], obj_pool));
instance_stats_map_.emplace(
- GetInstanceIdx(instance_params->instance_id),
- obj_pool->Add(
- new InstanceStats(*instance_params, fragment_stats[fragment_idx], obj_pool)));
+ GetInstanceIdx(instance_params.instance_id()), instance_stats);
}
}
@@ -112,59 +110,59 @@ void Coordinator::BackendState::SetRpcParams(const DebugOptions& debug_options,
const FilterRoutingTable& filter_routing_table, ExecQueryFInstancesRequestPB* request,
TExecPlanFragmentInfo* fragment_info) {
request->set_coord_state_idx(state_idx_);
- request->set_min_mem_reservation_bytes(backend_exec_params_->min_mem_reservation_bytes);
+ request->set_min_mem_reservation_bytes(
+ backend_exec_params_.min_mem_reservation_bytes());
request->set_initial_mem_reservation_total_claims(
- backend_exec_params_->initial_mem_reservation_total_claims);
- request->set_per_backend_mem_limit(schedule_.per_backend_mem_limit());
+ backend_exec_params_.initial_mem_reservation_total_claims());
+ request->set_per_backend_mem_limit(
+ exec_params_.query_schedule().per_backend_mem_limit());
// set fragment_ctxs and fragment_instance_ctxs
fragment_info->__isset.fragments = true;
fragment_info->__isset.fragment_instance_ctxs = true;
fragment_info->fragment_instance_ctxs.resize(
- backend_exec_params_->instance_params.size());
- for (int i = 0; i < backend_exec_params_->instance_params.size(); ++i) {
+ backend_exec_params_.instance_params().size());
+ for (int i = 0; i < backend_exec_params_.instance_params().size(); ++i) {
TPlanFragmentInstanceCtx& instance_ctx = fragment_info->fragment_instance_ctxs[i];
PlanFragmentInstanceCtxPB* instance_ctx_pb = request->add_fragment_instance_ctxs();
- const FInstanceExecParams& params = *backend_exec_params_->instance_params[i];
- int fragment_idx = params.fragment_exec_params.fragment.idx;
+ const FInstanceExecParamsPB& params = backend_exec_params_.instance_params(i);
+ int fragment_idx = params.fragment_idx();
+ DCHECK_LT(fragment_idx, exec_params_.query_schedule().fragment_exec_params().size());
+ const FragmentExecParamsPB& fragment_exec_params =
+ exec_params_.query_schedule().fragment_exec_params(fragment_idx);
// add a TPlanFragment, if we don't already have it
if (fragment_info->fragments.empty()
|| fragment_info->fragments.back().idx != fragment_idx) {
- fragment_info->fragments.push_back(params.fragment_exec_params.fragment);
+ const TPlanFragment* fragment = exec_params_.GetFragments()[fragment_idx];
+ fragment_info->fragments.push_back(*fragment);
PlanFragmentCtxPB* fragment_ctx = request->add_fragment_ctxs();
- fragment_ctx->set_fragment_idx(params.fragment_exec_params.fragment.idx);
- *fragment_ctx->mutable_destinations() = {
- params.fragment_exec_params.destinations.begin(),
- params.fragment_exec_params.destinations.end()};
+ fragment_ctx->set_fragment_idx(fragment_idx);
+ *fragment_ctx->mutable_destinations() = fragment_exec_params.destinations();
}
instance_ctx.fragment_idx = fragment_idx;
instance_ctx_pb->set_fragment_idx(fragment_idx);
- instance_ctx.fragment_instance_id = params.instance_id;
- instance_ctx.per_fragment_instance_idx = params.per_fragment_instance_idx;
- for (const auto& entry : params.per_node_scan_ranges) {
- ScanRangesPB& scan_ranges =
- (*instance_ctx_pb->mutable_per_node_scan_ranges())[entry.first];
- *scan_ranges.mutable_scan_ranges() = {entry.second.begin(), entry.second.end()};
+ 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();
+ for (const auto& entry : fragment_exec_params.per_exch_num_senders()) {
+ instance_ctx.per_exch_num_senders[entry.first] = entry.second;
}
- instance_ctx.__set_per_exch_num_senders(
- params.fragment_exec_params.per_exch_num_senders);
- instance_ctx.__set_sender_id(params.sender_id);
- *instance_ctx_pb->mutable_join_build_inputs() = {
- params.join_build_inputs.begin(), params.join_build_inputs.end()};
- if (params.num_join_build_outputs != -1) {
- instance_ctx.__set_num_join_build_outputs(params.num_join_build_outputs);
+ instance_ctx.__set_sender_id(params.sender_id());
+ *instance_ctx_pb->mutable_join_build_inputs() = params.join_build_inputs();
+ if (params.num_join_build_outputs() != -1) {
+ instance_ctx.__set_num_join_build_outputs(params.num_join_build_outputs());
}
if (debug_options.enabled()
&& (debug_options.instance_idx() == -1
- || debug_options.instance_idx() == GetInstanceIdx(params.instance_id))) {
+ || debug_options.instance_idx() == GetInstanceIdx(params.instance_id()))) {
instance_ctx.__set_debug_options(debug_options.ToThrift());
}
if (filter_mode_ == TRuntimeFilterMode::OFF) continue;
- int instance_idx = GetInstanceIdx(params.instance_id);
+ int instance_idx = GetInstanceIdx(params.instance_id());
auto& produced_map = filter_routing_table.finstance_filters_produced;
auto produced_it = produced_map.find(instance_idx);
if (produced_it == produced_map.end()) continue;
@@ -239,7 +237,7 @@ void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
// Do not issue an ExecQueryFInstances RPC if there are no fragment instances
// scheduled to run on this backend.
if (IsEmptyBackend()) {
- DCHECK(backend_exec_params_->is_coord_backend);
+ DCHECK(backend_exec_params_.is_coord_backend());
exec_done_ = true;
exec_status_barrier->Notify(Status::OK());
goto done;
@@ -266,7 +264,7 @@ void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options,
uint8_t* serialized_buf = nullptr;
uint32_t serialized_len = 0;
Status serialize_status =
- DebugAction(schedule_.query_options(), "EXEC_SERIALIZE_FRAGMENT_INFO");
+ DebugAction(exec_params_.query_options(), "EXEC_SERIALIZE_FRAGMENT_INFO");
if (LIKELY(serialize_status.ok())) {
serialize_status =
serializer.SerializeToBuffer(&fragment_info, &serialized_len, &serialized_buf);
@@ -429,12 +427,13 @@ bool Coordinator::BackendState::ApplyExecStatusReport(
DCHECK_EQ(instance_stats_map_.count(instance_idx), 1);
InstanceStats* instance_stats = instance_stats_map_[instance_idx];
int64_t last_report_seq_no = instance_stats->last_report_seq_no_;
- DCHECK(instance_stats->exec_params_.instance_id ==
- ProtoToQueryId(instance_exec_status.fragment_instance_id()));
+ DCHECK_EQ(instance_stats->exec_params_.instance_id(),
+ instance_exec_status.fragment_instance_id());
// Ignore duplicate or out-of-order messages.
if (report_seq_no <= last_report_seq_no) {
- VLOG_QUERY << Substitute("Ignoring stale update for query instance $0 with "
- "seq no $1", PrintId(instance_stats->exec_params_.instance_id), report_seq_no);
+ VLOG_QUERY << "Ignoring stale update for query instance "
+ << instance_stats->exec_params_.instance_id() << " with seq no "
+ << report_seq_no;
continue;
}
@@ -516,7 +515,7 @@ void Coordinator::BackendState::UpdateExecStats(
DCHECK(exec_done_) << "May only be called after WaitOnExecRpc() completes.";
for (const auto& entry: instance_stats_map_) {
const InstanceStats& instance_stats = *entry.second;
- int fragment_idx = instance_stats.exec_params_.fragment().idx;
+ int fragment_idx = instance_stats.exec_params_.fragment_idx();
DCHECK_LT(fragment_idx, fragment_stats.size());
FragmentStats* f = fragment_stats[fragment_idx];
int64_t completion_time = instance_stats.stopwatch_.ElapsedTime();
@@ -601,7 +600,7 @@ Coordinator::BackendState::CancelResult Coordinator::BackendState::Cancel(
}
CancelQueryFInstancesRequestPB request;
- TUniqueIdToUniqueIdPB(query_id_, request.mutable_query_id());
+ *request.mutable_query_id() = query_id_;
CancelQueryFInstancesResponsePB response;
const int num_retries = 3;
@@ -633,7 +632,7 @@ Coordinator::BackendState::CancelResult Coordinator::BackendState::Cancel(
void Coordinator::BackendState::PublishFilter(FilterState* state,
MemTracker* mem_tracker, const PublishFilterParamsPB& rpc_params,
RpcController& controller, PublishFilterResultPB& res) {
- DCHECK_EQ(ProtoToQueryId(rpc_params.dst_query_id()), query_id_);
+ DCHECK_EQ(rpc_params.dst_query_id(), query_id_);
// If the backend is already done, it's not waiting for this filter, so we skip
// sending it in this case.
if (IsDone()) return;
@@ -690,19 +689,18 @@ void Coordinator::BackendState::PublishFilterCompleteCb(
}
Coordinator::BackendState::InstanceStats::InstanceStats(
- const FInstanceExecParams& exec_params, FragmentStats* fragment_stats,
- ObjectPool* obj_pool)
- : exec_params_(exec_params),
- profile_(nullptr) {
+ const FInstanceExecParamsPB& exec_params, const TPlanFragment* fragment,
+ const NetworkAddressPB& address, FragmentStats* fragment_stats, ObjectPool* obj_pool)
+ : exec_params_(exec_params), fragment_(fragment), profile_(nullptr) {
const string& profile_name = Substitute("Instance $0 (host=$1)",
- PrintId(exec_params.instance_id), TNetworkAddressToString(exec_params.host));
+ PrintId(exec_params.instance_id()), NetworkAddressPBToString(address));
profile_ = RuntimeProfile::Create(obj_pool, profile_name);
profile_->AddInfoString(LAST_REPORT_TIME_DESC, ToStringFromUnixMillis(UnixMillis()));
fragment_stats->root_profile()->AddChild(profile_);
// add total split size to fragment_stats->bytes_assigned()
- for (const PerNodeScanRanges::value_type& entry: exec_params_.per_node_scan_ranges) {
- for (const ScanRangeParamsPB& scan_range_params : entry.second) {
+ for (const auto& entry : exec_params_.per_node_scan_ranges()) {
+ for (const ScanRangeParamsPB& scan_range_params : entry.second.scan_ranges()) {
if (!scan_range_params.scan_range().has_hdfs_file_split()) continue;
total_split_size_ += scan_range_params.scan_range().hdfs_file_split().length();
}
@@ -737,11 +735,11 @@ void Coordinator::BackendState::InstanceStats::Update(
exec_summary->data_sink_id_to_idx_map[exec_summary_entry.data_sink_id()];
}
TPlanNodeExecSummary& node_exec_summary = thrift_exec_summary.nodes[exec_summary_idx];
- DCHECK_EQ(node_exec_summary.fragment_idx, exec_params_.fragment().idx);
- int per_fragment_instance_idx = exec_params_.per_fragment_instance_idx;
+ DCHECK_EQ(node_exec_summary.fragment_idx, exec_params_.fragment_idx());
+ int per_fragment_instance_idx = exec_params_.per_fragment_instance_idx();
DCHECK_LT(per_fragment_instance_idx, node_exec_summary.exec_stats.size())
- << " instance_id=" << PrintId(exec_params_.instance_id)
- << " fragment_idx=" << exec_params_.fragment().idx;
+ << " instance_id=" << PrintId(exec_params_.instance_id())
+ << " fragment_idx=" << exec_params_.fragment_idx();
TExecStats& instance_stats = node_exec_summary.exec_stats[per_fragment_instance_idx];
if (exec_summary_entry.has_rows_returned()) {
@@ -760,8 +758,8 @@ void Coordinator::BackendState::InstanceStats::Update(
}
void Coordinator::BackendState::InstanceStats::ToJson(Value* value, Document* document) {
- Value instance_id_val(PrintId(exec_params_.instance_id).c_str(),
- document->GetAllocator());
+ Value instance_id_val(
+ PrintId(exec_params_.instance_id()).c_str(), document->GetAllocator());
value->AddMember("instance_id", instance_id_val, document->GetAllocator());
// We send 'done' explicitly so we don't have to infer it by comparison with a string
@@ -772,8 +770,7 @@ void Coordinator::BackendState::InstanceStats::ToJson(Value* value, Document* do
document->GetAllocator());
value->AddMember("current_state", state_val, document->GetAllocator());
- Value fragment_name_val(exec_params_.fragment().display_name.c_str(),
- document->GetAllocator());
+ Value fragment_name_val(fragment_->display_name.c_str(), document->GetAllocator());
value->AddMember("fragment_name", fragment_name_val, document->GetAllocator());
value->AddMember("first_status_update_received", last_report_time_ms_ > 0,
diff --git a/be/src/runtime/coordinator-backend-state.h b/be/src/runtime/coordinator-backend-state.h
index d92ca21..9de5715 100644
--- a/be/src/runtime/coordinator-backend-state.h
+++ b/be/src/runtime/coordinator-backend-state.h
@@ -29,10 +29,10 @@
#include <boost/accumulators/statistics/stats.hpp>
#include <boost/accumulators/statistics/variance.hpp>
+#include "gen-cpp/admission_control_service.pb.h"
#include "gen-cpp/control_service.proxy.h"
#include "kudu/rpc/rpc_controller.h"
#include "runtime/coordinator.h"
-#include "scheduling/query-schedule.h"
#include "util/error-util-internal.h"
#include "util/progress-updater.h"
#include "util/runtime-profile.h"
@@ -51,7 +51,6 @@ class ReportExecStatusRequestPB;
class TUniqueId;
class TQueryCtx;
class ExecSummary;
-struct FInstanceExecParams;
/// This class manages all aspects of the execution of all fragment instances of a
/// single query on a particular backend. For the coordinator backend its possible to have
@@ -68,16 +67,17 @@ struct FInstanceExecParams;
/// subject to the restrictions above on when methods may be called.
class Coordinator::BackendState {
public:
- BackendState(const QuerySchedule& schedule, const TQueryCtx& query_ctx, int state_idx,
- TRuntimeFilterMode::type filter_mode, const BackendExecParams& exec_params);
+ BackendState(const QueryExecParams& exec_params, int state_idx,
+ TRuntimeFilterMode::type filter_mode,
+ const BackendExecParamsPB& backend_exec_params);
/// The following are initialized in the constructor and always valid to call.
int state_idx() const { return state_idx_; }
- const BackendExecParams* exec_params() const { return backend_exec_params_; }
+ const BackendExecParamsPB& exec_params() const { return backend_exec_params_; }
const NetworkAddressPB& impalad_address() const { return host_; }
const NetworkAddressPB& krpc_impalad_address() const { return krpc_host_; }
/// Returns true if there are no fragment instances scheduled on this backend.
- bool IsEmptyBackend() const { return backend_exec_params_->instance_params.empty(); }
+ bool IsEmptyBackend() const { return backend_exec_params_.instance_params().empty(); }
/// Return true if execution at this backend is done.
bool IsDone();
@@ -223,7 +223,8 @@ class Coordinator::BackendState {
/// Not thread-safe.
class InstanceStats {
public:
- InstanceStats(const FInstanceExecParams& exec_params, FragmentStats* fragment_stats,
+ InstanceStats(const FInstanceExecParamsPB& exec_params, const TPlanFragment* fragment,
+ const NetworkAddressPB& address, FragmentStats* fragment_stats,
ObjectPool* obj_pool);
/// Updates 'this' with exec_status and the fragment intance's thrift profile. Also
@@ -233,7 +234,7 @@ class Coordinator::BackendState {
const TRuntimeProfileTree& thrift_profile, ExecSummary* exec_summary);
int per_fragment_instance_idx() const {
- return exec_params_.per_fragment_instance_idx;
+ return exec_params_.per_fragment_instance_idx();
}
/// Serializes instance stats to JSON by adding members to 'value', including its
@@ -244,8 +245,12 @@ class Coordinator::BackendState {
private:
friend class BackendState;
- /// query lifetime
- const FInstanceExecParams& exec_params_;
+ /// The exec params for this fragment instance. Owned by the QuerySchedulePB in
+ /// ClientRequestState.
+ const FInstanceExecParamsPB& exec_params_;
+
+ /// The corresponding fragment. Owned by the TExecRequest in QueryDriver.
+ const TPlanFragment* fragment_;
/// Unix time in milliseconds of the last status report update for this fragment
/// instance. Set in Update(). Uses UnixMillis() instead of MonotonicMillis() as
@@ -283,14 +288,15 @@ class Coordinator::BackendState {
FInstanceExecStatePB current_state_ = FInstanceExecStatePB::WAITING_FOR_EXEC;
};
- /// QuerySchedule associated with the Coordinator that owns this BackendState.
- const QuerySchedule& schedule_;
+ /// ExecParams associated with the Coordinator that owns this BackendState.
+ const QueryExecParams& exec_params_;
const int state_idx_; /// index of 'this' in Coordinator::backend_states_
const TRuntimeFilterMode::type filter_mode_;
- /// Backend exec params, owned by the QuerySchedule and has query lifetime.
- const BackendExecParams* backend_exec_params_;
+ /// Backend exec params, owned by the QuerySchedulePB in ClientRequestState and has
+ /// query lifetime.
+ const BackendExecParamsPB& backend_exec_params_;
/// map from instance idx to InstanceStats, the latter live in the obj_pool parameter
/// of Init()
@@ -312,8 +318,9 @@ class Coordinator::BackendState {
/// The query context of the Coordinator that owns this BackendState.
const TQueryCtx& query_ctx_;
- /// The query id of the Coordinator that owns this BackendState.
- const TUniqueId& query_id_;
+ /// The query id of the Coordinator that owns this BackendState. Owned by
+ /// 'exec_params_'.
+ const UniqueIdPB& query_id_;
/// Used to issue the ExecQueryFInstances() rpc.
kudu::rpc::RpcController exec_rpc_controller_;
@@ -523,8 +530,7 @@ class Coordinator::BackendResourceState {
public:
/// Create the BackendResourceState with the given vector of BackendStates. All
/// BackendStates are initially in the IN_USE state.
- BackendResourceState(
- const std::vector<BackendState*>& backend_states, const QuerySchedule& schedule);
+ BackendResourceState(const std::vector<BackendState*>& backend_states);
~BackendResourceState();
/// Mark a BackendState as finished and transition it to the PENDING state. Applies
@@ -587,9 +593,6 @@ class Coordinator::BackendResourceState {
// True if the BackendResourceState is closed, false otherwise.
bool closed_ = false;
- /// QuerySchedule associated with the Coordinator that owns this BackendResourceState.
- const QuerySchedule& schedule_;
-
/// Configured value of FLAGS_release_backend_states_delay_ms in nanoseconds.
const int64_t release_backend_states_delay_ns_;
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index 4b502e7..3551849 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -32,6 +32,7 @@
#include "exec/data-sink.h"
#include "exec/plan-root-sink.h"
#include "gen-cpp/ImpalaInternalService_constants.h"
+#include "gen-cpp/admission_control_service.pb.h"
#include "kudu/rpc/rpc_context.h"
#include "kudu/rpc/rpc_sidecar.h"
#include "runtime/coordinator-backend-state.h"
@@ -40,11 +41,10 @@
#include "runtime/exec-env.h"
#include "runtime/fragment-instance-state.h"
#include "runtime/hdfs-fs-cache.h"
-#include "runtime/query-exec-mgr.h"
#include "runtime/query-driver.h"
+#include "runtime/query-exec-mgr.h"
#include "runtime/query-state.h"
#include "scheduling/admission-controller.h"
-#include "scheduling/query-schedule.h"
#include "scheduling/scheduler.h"
#include "service/client-request-state.h"
#include "util/bloom-filter.h"
@@ -114,16 +114,16 @@ PROFILE_DEFINE_TIMER(FinalizationTimer, STABLE_LOW,
// Maximum number of fragment instances that can publish each broadcast filter.
static const int MAX_BROADCAST_FILTER_PRODUCERS = 3;
-Coordinator::Coordinator(ClientRequestState* parent, const QuerySchedule& schedule,
- RuntimeProfile::EventSequence* events)
+Coordinator::Coordinator(ClientRequestState* parent, const TExecRequest& exec_request,
+ const QuerySchedulePB& query_schedule, RuntimeProfile::EventSequence* events)
: parent_query_driver_(parent->parent_driver()),
parent_request_state_(parent),
- schedule_(schedule),
- filter_mode_(schedule.query_options().runtime_filter_mode),
+ exec_params_(exec_request, query_schedule),
+ filter_mode_(exec_params_.query_options().runtime_filter_mode),
obj_pool_(new ObjectPool()),
query_events_(events),
- exec_rpcs_status_barrier_(schedule_.per_backend_exec_params().size()),
- backend_released_barrier_(schedule_.per_backend_exec_params().size()),
+ exec_rpcs_status_barrier_(query_schedule.backend_exec_params().size()),
+ backend_released_barrier_(query_schedule.backend_exec_params().size()),
filter_routing_table_(new FilterRoutingTable) {}
Coordinator::~Coordinator() {
@@ -137,7 +137,7 @@ Coordinator::~Coordinator() {
}
Status Coordinator::Exec() {
- const TQueryExecRequest& request = schedule_.request();
+ const TQueryExecRequest& request = exec_params_.query_exec_request();
DCHECK(request.plan_exec_info.size() > 0);
VLOG_QUERY << "Exec() query_id=" << PrintId(query_id())
@@ -156,10 +156,10 @@ Status Coordinator::Exec() {
// initialize progress updater
const string& str = Substitute("Query $0", PrintId(query_id()));
- progress_.Init(str, schedule_.num_scan_ranges());
+ progress_.Init(str, exec_params_.query_schedule().num_scan_ranges());
query_state_ = ExecEnv::GetInstance()->query_exec_mgr()->CreateQueryState(
- query_ctx(), schedule_.coord_backend_mem_limit());
+ query_ctx(), exec_params_.query_schedule().coord_backend_mem_limit());
filter_mem_tracker_ = query_state_->obj_pool()->Add(new MemTracker(
-1, "Runtime Filter (Coordinator)", query_state_->query_mem_tracker(), false));
@@ -167,7 +167,7 @@ Status Coordinator::Exec() {
// create BackendStates and per-instance state, including profiles, and install
// the latter in the FragmentStats' root profile
InitBackendStates();
- exec_summary_.Init(schedule_);
+ exec_summary_.Init(exec_params_);
if (filter_mode_ != TRuntimeFilterMode::OFF) {
// Populate the runtime filter routing table. This should happen before starting the
@@ -184,7 +184,7 @@ Status Coordinator::Exec() {
RETURN_IF_ERROR(FinishBackendStartup());
// set coord_instance_ and coord_sink_
- if (schedule_.GetCoordFragment() != nullptr) {
+ if (exec_params_.GetCoordFragment() != nullptr) {
// this blocks until all fragment instances have finished their Prepare phase
Status query_status = query_state_->GetFInstanceState(query_id(), &coord_instance_);
if (!query_status.ok()) return UpdateExecState(query_status, nullptr, FLAGS_hostname);
@@ -199,20 +199,18 @@ Status Coordinator::Exec() {
}
void Coordinator::InitFragmentStats() {
- vector<const TPlanFragment*> fragments;
- schedule_.GetTPlanFragments(&fragments);
- const TPlanFragment* coord_fragment = schedule_.GetCoordFragment();
+ const TPlanFragment* coord_fragment = exec_params_.GetCoordFragment();
int64_t total_num_finstances = 0;
- for (const TPlanFragment* fragment: fragments) {
+ DCHECK_GT(exec_params_.num_fragments(), 0);
+ for (const TPlanFragment* fragment : exec_params_.GetFragments()) {
string root_profile_name =
- Substitute(
- fragment == coord_fragment ? "Coordinator Fragment $0" : "Fragment $0",
- fragment->display_name);
- string avg_profile_name =
- Substitute("Averaged Fragment $0", fragment->display_name);
- int num_instances =
- schedule_.GetFragmentExecParams(fragment->idx).instance_exec_params.size();
+ Substitute(fragment == coord_fragment ? "Coordinator Fragment $0" : "Fragment $0",
+ fragment->display_name);
+ string avg_profile_name = Substitute("Averaged Fragment $0", fragment->display_name);
+ int num_instances = exec_params_.query_schedule()
+ .fragment_exec_params(fragment->idx)
+ .instances_size();
total_num_finstances += num_instances;
// TODO: special-case the coordinator fragment?
FragmentStats* fragment_stats = obj_pool()->Add(
@@ -223,13 +221,13 @@ void Coordinator::InitFragmentStats() {
query_profile_->AddChild(fragment_stats->root_profile());
}
COUNTER_SET(PROFILE_NumFragments.Instantiate(query_profile_),
- static_cast<int64_t>(fragments.size()));
+ static_cast<int64_t>(exec_params_.num_fragments()));
COUNTER_SET(PROFILE_NumFragmentInstances.Instantiate(query_profile_),
total_num_finstances);
}
void Coordinator::InitBackendStates() {
- int num_backends = schedule_.per_backend_exec_params().size();
+ int num_backends = exec_params_.query_schedule().backend_exec_params().size();
DCHECK_GT(num_backends, 0);
lock_guard<SpinLock> l(backend_states_init_lock_);
@@ -239,9 +237,10 @@ void Coordinator::InitBackendStates() {
// create BackendStates
int backend_idx = 0;
- for (const auto& entry : schedule_.per_backend_exec_params()) {
- BackendState* backend_state = obj_pool()->Add(new BackendState(
- schedule_, query_ctx(), backend_idx, filter_mode_, entry.second));
+ for (const BackendExecParamsPB& backend_exec_params :
+ exec_params_.query_schedule().backend_exec_params()) {
+ BackendState* backend_state = obj_pool()->Add(
+ new BackendState(exec_params_, backend_idx, filter_mode_, backend_exec_params));
backend_state->Init(fragment_stats_, host_profiles_, obj_pool());
backend_states_[backend_idx++] = backend_state;
// was_inserted is true if the pair was successfully inserted into the map, false
@@ -254,13 +253,12 @@ void Coordinator::InitBackendStates() {
<< " associated with multiple BackendStates";
}
}
- backend_resource_state_ =
- obj_pool()->Add(new BackendResourceState(backend_states_, schedule_));
+ backend_resource_state_ = obj_pool()->Add(new BackendResourceState(backend_states_));
num_completed_backends_ = PROFILE_NumCompletedBackends.Instantiate(query_profile_);
}
-void Coordinator::ExecSummary::Init(const QuerySchedule& schedule) {
- const TQueryExecRequest& request = schedule.request();
+void Coordinator::ExecSummary::Init(const QueryExecParams& exec_params) {
+ const TQueryExecRequest& request = exec_params.query_exec_request();
// init exec_summary_.{nodes, exch_to_sender_map}
thrift_exec_summary.__isset.nodes = true;
DCHECK(thrift_exec_summary.nodes.empty());
@@ -274,14 +272,10 @@ void Coordinator::ExecSummary::Init(const QuerySchedule& schedule) {
const TPlan& plan = fragment.plan;
const TDataSink& output_sink = fragment.output_sink;
// Count the number of hosts and instances.
- const vector<FInstanceExecParams>& instance_params =
- schedule.GetFragmentExecParams(fragment.idx).instance_exec_params;
- unordered_set<TNetworkAddress> host_set;
- for (const FInstanceExecParams& instance: instance_params) {
- host_set.insert(instance.host);
- }
- int num_hosts = host_set.size();
- int num_instances = instance_params.size();
+ const FragmentExecParamsPB& fragment_exec_param =
+ exec_params.query_schedule().fragment_exec_params(fragment.idx);
+ int num_hosts = fragment_exec_param.num_hosts();
+ int num_instances = fragment_exec_param.instances_size();
// Add the data sink at the root of the fragment.
data_sink_id_to_idx_map[fragment.idx] = thrift_exec_summary.nodes.size();
@@ -351,16 +345,18 @@ void Coordinator::InitFilterRoutingTable() {
<< "InitFilterRoutingTable() called after table marked as complete";
lock_guard<shared_mutex> lock(filter_routing_table_->lock); // Exclusive lock.
- for (const FragmentExecParams& fragment_params: schedule_.fragment_exec_params()) {
- int num_instances = fragment_params.instance_exec_params.size();
+ for (const FragmentExecParamsPB& fragment_params :
+ exec_params_.query_schedule().fragment_exec_params()) {
+ int num_instances = fragment_params.instances_size();
DCHECK_GT(num_instances, 0);
- int num_backends = fragment_params.GetNumBackends();
+ int num_backends = fragment_params.num_hosts();
DCHECK_GT(num_backends, 0);
+ const TPlanFragment* fragment =
+ exec_params_.GetFragments()[fragment_params.fragment_idx()];
// Hash join build sinks can produce filters in mt_dop > 0 plans.
- if (fragment_params.fragment.output_sink.__isset.join_build_sink) {
- const TJoinBuildSink& join_sink =
- fragment_params.fragment.output_sink.join_build_sink;
+ if (fragment->output_sink.__isset.join_build_sink) {
+ const TJoinBuildSink& join_sink = fragment->output_sink.join_build_sink;
for (const TRuntimeFilterDesc& filter: join_sink.runtime_filters) {
// The join node ID is used to identify the join that produces the filter, even
// though the builder is separate from the actual node.
@@ -369,7 +365,7 @@ void Coordinator::InitFilterRoutingTable() {
fragment_params, num_instances, num_backends, filter, filter.src_node_id);
}
}
- for (const TPlanNode& plan_node: fragment_params.fragment.plan.nodes) {
+ for (const TPlanNode& plan_node : fragment->plan.nodes) {
if (!plan_node.__isset.runtime_filters) continue;
for (const TRuntimeFilterDesc& filter: plan_node.runtime_filters) {
DCHECK(filter_mode_ == TRuntimeFilterMode::GLOBAL || filter.has_local_targets);
@@ -385,7 +381,7 @@ void Coordinator::InitFilterRoutingTable() {
DCHECK(it != filter.planid_to_target_ndx.end());
const TRuntimeFilterTargetDesc& t_target = filter.targets[it->second];
DCHECK(filter_mode_ == TRuntimeFilterMode::GLOBAL || t_target.is_local_target);
- f->targets()->emplace_back(t_target, fragment_params.fragment.idx);
+ f->targets()->emplace_back(t_target, fragment->idx);
} else {
DCHECK(false) << "Unexpected plan node with runtime filters: "
<< ThriftDebugString(plan_node);
@@ -401,7 +397,7 @@ void Coordinator::InitFilterRoutingTable() {
filter_routing_table_->is_complete = true;
}
-void Coordinator::AddFilterSource(const FragmentExecParams& src_fragment_params,
+void Coordinator::AddFilterSource(const FragmentExecParamsPB& src_fragment_params,
int num_instances, int num_backends, const TRuntimeFilterDesc& filter,
int join_node_id) {
FilterState* f = filter_routing_table_->GetOrCreateFilterState(filter);
@@ -425,7 +421,10 @@ void Coordinator::AddFilterSource(const FragmentExecParams& src_fragment_params,
// local producers.
// This work was deferred from the IMPALA-4400 change because it provides only
// incremental performance benefits.
- vector<int> src_idxs = src_fragment_params.GetInstanceIdxs();
+ vector<int> src_idxs;
+ for (const UniqueIdPB& instance_id : src_fragment_params.instances()) {
+ src_idxs.push_back(GetInstanceIdx(instance_id));
+ }
// If this is a broadcast join with only non-local targets, build and publish it
// on MAX_BROADCAST_FILTER_PRODUCERS instances. If this is not a broadcast join
@@ -458,7 +457,7 @@ Status Coordinator::StartBackendExec() {
int num_backends = backend_states_.size();
backend_exec_complete_barrier_.reset(new CountingBarrier(num_backends));
- DebugOptions debug_options(schedule_.query_options());
+ DebugOptions debug_options(exec_params_.query_options());
VLOG_QUERY << "starting execution on " << num_backends << " backends for query_id="
<< PrintId(query_id());
@@ -481,7 +480,7 @@ Status Coordinator::StartBackendExec() {
// One of the backends has already indicated an error with Exec().
break;
}
- DebugActionNoFail(schedule_.query_options(), "COORD_BEFORE_EXEC_RPC");
+ DebugActionNoFail(exec_params_.query_options(), "COORD_BEFORE_EXEC_RPC");
// Safe for ExecAsync() to read 'filter_routing_table_' because it is complete
// at this point and won't be destroyed while this function is executing,
// because it won't be torn down until WaitOnExecRpcs() has returned.
@@ -503,8 +502,8 @@ Status Coordinator::StartBackendExec() {
failed_backend_states.push_back(backend_state);
LOG(INFO) << "Blacklisting " << backend_state->impalad_address()
<< " because an Exec() rpc to it failed.";
- const BackendDescriptorPB& be_desc = backend_state->exec_params()->be_desc;
- ExecEnv::GetInstance()->cluster_membership_mgr()->BlacklistExecutor(be_desc,
+ const UniqueIdPB& backend_id = backend_state->exec_params().backend_id();
+ ExecEnv::GetInstance()->cluster_membership_mgr()->BlacklistExecutor(backend_id,
FromKuduStatus(backend_state->exec_rpc_status(), "Exec() rpc failed"));
}
}
@@ -521,7 +520,7 @@ Status Coordinator::StartBackendExec() {
<< PrintId(query_id());
query_events_->MarkEvent(
Substitute("All $0 execution backends ($1 fragment instances) started",
- num_backends, schedule_.GetNumFragmentInstances()));
+ num_backends, exec_params_.GetNumFragmentInstances()));
return Status::OK();
}
@@ -1022,11 +1021,11 @@ Status Coordinator::UpdateBlacklistWithAuxErrorInfo(
}
// The execution parameters of the destination node for the failed RPC.
- const BackendExecParams* dest_node_exec_params =
+ const BackendExecParamsPB& dest_node_exec_params =
dest_node_and_be_state->second->exec_params();
// The Coordinator for the query should never be blacklisted.
- if (dest_node_exec_params->is_coord_backend) {
+ if (dest_node_exec_params.is_coord_backend()) {
VLOG_QUERY << "Query failed due to a failed RPC to the Coordinator";
continue;
}
@@ -1056,7 +1055,7 @@ Status Coordinator::UpdateBlacklistWithAuxErrorInfo(
retryable_status.MergeStatus(status);
ExecEnv::GetInstance()->cluster_membership_mgr()->BlacklistExecutor(
- dest_node_exec_params->be_desc, retryable_status);
+ dest_node_exec_params.backend_id(), retryable_status);
// Only blacklist one node per report.
return retryable_status;
@@ -1226,8 +1225,8 @@ void Coordinator::ReleaseQueryAdmissionControlResources() {
AdmissionController* admission_controller =
ExecEnv::GetInstance()->admission_controller();
DCHECK(admission_controller != nullptr);
- admission_controller->ReleaseQuery(
- schedule_, ComputeQueryResourceUtilization().peak_per_host_mem_consumption);
+ admission_controller->ReleaseQuery(exec_params_.query_id(),
+ ComputeQueryResourceUtilization().peak_per_host_mem_consumption);
query_events_->MarkEvent("Released admission control resources");
}
@@ -1236,11 +1235,11 @@ void Coordinator::ReleaseBackendAdmissionControlResources(
AdmissionController* admission_controller =
ExecEnv::GetInstance()->admission_controller();
DCHECK(admission_controller != nullptr);
- vector<TNetworkAddress> host_addrs;
+ vector<NetworkAddressPB> host_addrs;
for (auto backend_state : backend_states) {
- host_addrs.push_back(FromNetworkAddressPB(backend_state->impalad_address()));
+ host_addrs.push_back(backend_state->impalad_address());
}
- admission_controller->ReleaseQueryBackends(schedule_, host_addrs);
+ admission_controller->ReleaseQueryBackends(exec_params_.query_id(), host_addrs);
}
Coordinator::ResourceUtilization Coordinator::ComputeQueryResourceUtilization() {
@@ -1514,7 +1513,7 @@ void Coordinator::FInstanceStatsToJson(Document* doc) {
}
const TQueryCtx& Coordinator::query_ctx() const {
- return schedule_.request().query_ctx;
+ return exec_params_.query_exec_request().query_ctx;
}
const TUniqueId& Coordinator::query_id() const {
@@ -1522,8 +1521,9 @@ const TUniqueId& Coordinator::query_id() const {
}
const TFinalizeParams* Coordinator::finalize_params() const {
- return schedule_.request().__isset.finalize_params
- ? &schedule_.request().finalize_params : nullptr;
+ return exec_params_.query_exec_request().__isset.finalize_params ?
+ &exec_params_.query_exec_request().finalize_params :
+ nullptr;
}
bool Coordinator::IsExecuting() {
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index 9a488d6..277d619 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -31,6 +31,7 @@
#include "gen-cpp/Types_types.h"
#include "gen-cpp/data_stream_service.pb.h"
#include "runtime/dml-exec-state.h"
+#include "runtime/query-exec-params.h"
#include "util/counting-barrier.h"
#include "util/progress-updater.h"
#include "util/promise.h"
@@ -47,13 +48,13 @@ namespace impala {
class AuxErrorInfoPB;
class ClientRequestState;
+class FragmentExecParamsPB;
class FragmentInstanceState;
class MemTracker;
class ObjectPool;
class PlanRootSink;
class QueryDriver;
class QueryResultSet;
-class QuerySchedule;
class QueryState;
class ReportExecStatusRequestPB;
class RuntimeProfile;
@@ -61,7 +62,6 @@ class RuntimeState;
class TPlanExecRequest;
class TRuntimeProfileTree;
class TUpdateCatalogRequest;
-struct FragmentExecParams;
/// Query coordinator: handles execution of fragment instances on remote nodes, given a
/// TQueryExecRequest. As part of that, it handles all interactions with the executing
@@ -115,8 +115,8 @@ struct FragmentExecParams;
/// and unnest them
class Coordinator { // NOLINT: The member variables could be re-ordered to save space
public:
- Coordinator(ClientRequestState* parent, const QuerySchedule& schedule,
- RuntimeProfile::EventSequence* events);
+ Coordinator(ClientRequestState* parent, const TExecRequest& exec_request,
+ const QuerySchedulePB& query_schedule, RuntimeProfile::EventSequence* events);
~Coordinator();
/// Initiate asynchronous execution of a query with the given schedule. When it
@@ -263,8 +263,9 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// the constructor. It always outlives the coordinator.
ClientRequestState* parent_request_state_;
- /// owned by the ClientRequestState that owns this coordinator
- const QuerySchedule& schedule_;
+ /// Contains references to Thrift and Protobuf structs constituting the execution
+ /// parameters. These structs are owned by the QueryDriver and ClientRequestState.
+ QueryExecParams exec_params_;
/// Copied from TQueryExecRequest, governs when finalization occurs. Set in Exec().
TStmtType::type stmt_type_;
@@ -276,7 +277,7 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// A map from the NetworkAddressPB of a backend to the BackendState running on the
/// NetworkAddressPB. All values are non-nullptr and owned by obj_pool(). The address
/// is the kRPC address (Coordinator::BackendState::krpc_impalad_address) of the
- /// Backend. This map is distinct from QuerySchedule::per_backend_exec_params(),
+ /// Backend. This map is distinct from QuerySchedulePB::per_backend_exec_params(),
/// which uses the Thrift address as the key rather than the kRPC address.
boost::unordered_map<NetworkAddressPB, BackendState*> addr_to_backend_state_;
@@ -364,7 +365,7 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// A mapping of fragment data sink ids to index into thrift_exec_summary.nodes
boost::unordered_map<TPlanNodeId, int> data_sink_id_to_idx_map;
- void Init(const QuerySchedule& query_schedule);
+ void Init(const QueryExecParams& exec_params);
};
// Initialized by Exec().
@@ -548,7 +549,7 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// fragment containing the join node (if build is integrated) or build sink (if the
/// build is separate). 'num_instances' and 'num_backends' are the number of instances
/// and backends that the fragment runs on.
- void AddFilterSource(const FragmentExecParams& src_fragment_params,int num_instances,
+ void AddFilterSource(const FragmentExecParamsPB& src_fragment_params, int num_instances,
int num_backends, const TRuntimeFilterDesc& filter, int join_node_id);
/// Helper for HandleExecStateTransition(). Releases all resources associated with
diff --git a/be/src/runtime/fragment-instance-state.cc b/be/src/runtime/fragment-instance-state.cc
index 3322003..3cfc416 100644
--- a/be/src/runtime/fragment-instance-state.cc
+++ b/be/src/runtime/fragment-instance-state.cc
@@ -47,7 +47,6 @@
#include "runtime/runtime-filter-bank.h"
#include "runtime/runtime-state.h"
#include "runtime/thread-resource-mgr.h"
-#include "scheduling/query-schedule.h"
#include "util/container-util.h"
#include "util/debug-util.h"
#include "util/periodic-counter-updater.h"
diff --git a/be/src/runtime/query-driver.cc b/be/src/runtime/query-driver.cc
index 1cde910..a1de253 100644
--- a/be/src/runtime/query-driver.cc
+++ b/be/src/runtime/query-driver.cc
@@ -91,7 +91,7 @@ void QueryDriver::TryQueryRetry(
const TUniqueId& query_id = client_request_state->query_id();
DCHECK(client_request_state->schedule() != nullptr);
- if (client_request_state->schedule()->query_options().retry_failed_queries) {
+ if (exec_request_->query_options.retry_failed_queries) {
lock_guard<mutex> l(*client_request_state->lock());
// Queries can only be retried if no rows for the query have been fetched
diff --git a/be/src/runtime/query-exec-params.cc b/be/src/runtime/query-exec-params.cc
new file mode 100644
index 0000000..d453fcc
--- /dev/null
+++ b/be/src/runtime/query-exec-params.cc
@@ -0,0 +1,71 @@
+// 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.
+
+#include "runtime/query-exec-params.h"
+
+#include "common/logging.h"
+#include "gen-cpp/Frontend_types.h"
+#include "gen-cpp/ImpalaInternalService_types.h"
+#include "gen-cpp/admission_control_service.pb.h"
+#include "util/uid-util.h"
+
+namespace impala {
+
+QueryExecParams::QueryExecParams(
+ const TExecRequest& exec_request, const QuerySchedulePB& query_schedule)
+ : exec_request_(exec_request), query_schedule_(query_schedule) {
+ TUniqueIdToUniqueIdPB(exec_request_.query_exec_request.query_ctx.query_id, &query_id_);
+
+ int num_fragments = 0;
+ for (const TPlanExecInfo& plan_exec_info : query_exec_request().plan_exec_info) {
+ num_fragments += plan_exec_info.fragments.size();
+ }
+
+ fragments_.resize(num_fragments);
+ // Populate 'fragments_' with references into 'exec_request_'.
+ for (const TPlanExecInfo& plan_exec_info : query_exec_request().plan_exec_info) {
+ for (const TPlanFragment& fragment : plan_exec_info.fragments) {
+ fragments_[fragment.idx] = &fragment;
+ }
+ }
+ DCHECK(!fragments_.empty());
+}
+
+QueryExecParams::QueryExecParams(const UniqueIdPB& query_id,
+ const TExecRequest& exec_request, const QuerySchedulePB& query_schedule)
+ : query_id_(query_id), exec_request_(exec_request), query_schedule_(query_schedule) {}
+
+const TPlanFragment* QueryExecParams::GetCoordFragment() const {
+ // Only have coordinator fragment for statements that return rows.
+ if (query_exec_request().stmt_type != TStmtType::QUERY) return nullptr;
+ DCHECK_GE(query_exec_request().plan_exec_info.size(), 1);
+ DCHECK_GE(query_exec_request().plan_exec_info[0].fragments.size(), 1);
+ const TPlanFragment* fragment = &query_exec_request().plan_exec_info[0].fragments[0];
+ DCHECK_EQ(fragment->partition.type, TPartitionType::UNPARTITIONED);
+ return fragment;
+}
+
+int QueryExecParams::GetNumFragmentInstances() const {
+ int total = 0;
+ for (const BackendExecParamsPB& backend_exec_params :
+ query_schedule_.backend_exec_params()) {
+ total += backend_exec_params.instance_params().size();
+ }
+ return total;
+}
+
+} // namespace impala
diff --git a/be/src/runtime/query-exec-params.h b/be/src/runtime/query-exec-params.h
new file mode 100644
index 0000000..f8bd0b7
--- /dev/null
+++ b/be/src/runtime/query-exec-params.h
@@ -0,0 +1,78 @@
+// 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.
+
+#pragma once
+
+#include <unordered_map>
+
+#include "common/logging.h"
+#include "gen-cpp/Frontend_types.h"
+#include "gen-cpp/common.pb.h"
+
+namespace impala {
+
+class QuerySchedulePB;
+class TQueryExecRequest;
+class TQueryOptions;
+class TPlanFragment;
+
+/// This class is a wrapper around references to the parameters that the Coordinator needs
+/// for query execution - the TExecRequest which is owned by the QueryDriver and the
+/// QuerySchedulePB which is owned by the ClientRequestState - along with some convenience
+/// functions for accessing them.
+class QueryExecParams {
+ public:
+ QueryExecParams(
+ const TExecRequest& exec_request, const QuerySchedulePB& query_schedule);
+
+ const QuerySchedulePB& query_schedule() const { return query_schedule_; }
+
+ const UniqueIdPB& query_id() const { return query_id_; }
+ const TQueryExecRequest& query_exec_request() const {
+ return exec_request_.query_exec_request;
+ }
+ const TQueryOptions& query_options() const { return exec_request_.query_options; }
+
+ int32_t num_fragments() const { return fragments_.size(); }
+
+ const std::vector<const TPlanFragment*>& GetFragments() const { return fragments_; }
+
+ /// Return the coordinator fragment, or nullptr if there isn't one.
+ const TPlanFragment* GetCoordFragment() const;
+
+ /// Return the total number of instances across all fragments.
+ int GetNumFragmentInstances() const;
+
+ private:
+ friend class CoordinatorBackendStateTest;
+
+ UniqueIdPB query_id_;
+
+ /// Owned by the QueryDriver and has query lifetime.
+ const TExecRequest& exec_request_;
+
+ /// List of pointers to fragments in 'exec_request_', indexed by fragment idx.
+ std::vector<const TPlanFragment*> fragments_;
+
+ /// Owned by the ClientRwquestState and has query lifetime.
+ const QuerySchedulePB& query_schedule_;
+
+ /// Constructor used for testing only. Does not initialize 'fragments_'.
+ QueryExecParams(const UniqueIdPB& query_id, const TExecRequest& exec_request,
+ const QuerySchedulePB& query_schedule);
+};
+} // namespace impala
diff --git a/be/src/scheduling/CMakeLists.txt b/be/src/scheduling/CMakeLists.txt
index 0b5aa74..729be67 100644
--- a/be/src/scheduling/CMakeLists.txt
+++ b/be/src/scheduling/CMakeLists.txt
@@ -22,8 +22,12 @@ set(LIBRARY_OUTPUT_PATH "${BUILD_OUTPUT_ROOT_DIRECTORY}/scheduling")
# where to put generated binaries
set(EXECUTABLE_OUTPUT_PATH "${BUILD_OUTPUT_ROOT_DIRECTORY}/scheduling")
+# Mark the protobuf file as generated
+set_source_files_properties(${ADMISSION_CONTROL_SERVICE_PROTO_SRCS} PROPERTIES GENERATED TRUE)
+
# TODO: Move other scheduling-related classes here
add_library(Scheduling STATIC
+ ${ADMISSION_CONTROL_SERVICE_PROTO_SRCS}
admission-controller.cc
executor-blacklist.cc
cluster-membership-mgr.cc
diff --git a/be/src/scheduling/admission-controller-test.cc b/be/src/scheduling/admission-controller-test.cc
index 4308de0..672ae7d 100644
--- a/be/src/scheduling/admission-controller-test.cc
+++ b/be/src/scheduling/admission-controller-test.cc
@@ -94,7 +94,7 @@ class AdmissionControllerTest : public testing::Test {
request->__set_stmt_type(TStmtType::QUERY);
RuntimeProfile* profile = RuntimeProfile::Create(&pool_, "pool1");
- TUniqueId* query_id = pool_.Add(new TUniqueId()); // always 0,0
+ UniqueIdPB* query_id = pool_.Add(new UniqueIdPB()); // always 0,0
TQueryOptions* query_options = pool_.Add(new TQueryOptions());
query_options->__set_mem_limit(mem_limit);
QuerySchedule* query_schedule = pool_.Add(new QuerySchedule(
@@ -122,30 +122,29 @@ class AdmissionControllerTest : public testing::Test {
bool is_dedicated_coord, int64_t min_mem_reservation_bytes = 0,
int64_t admit_mem_limit = 200L * MEGABYTE, int slots_to_use = 1,
int slots_available = 8) {
- PerBackendExecParams* per_backend_exec_params = pool_.Add(new PerBackendExecParams());
+ query_schedule.ClearBackendExecParams();
for (int i = 0; i < count; ++i) {
const string host_name = Substitute("host$0", i);
- TNetworkAddress host_addr = MakeNetworkAddress(host_name, 25000);
- BackendExecParams* backend_exec_params = pool_.Add(new BackendExecParams());
- backend_exec_params->min_mem_reservation_bytes = min_mem_reservation_bytes;
- backend_exec_params->slots_to_use = slots_to_use;
- backend_exec_params->be_desc.set_admit_mem_limit(admit_mem_limit);
- backend_exec_params->be_desc.set_admission_slots(slots_available);
- backend_exec_params->be_desc.set_is_executor(true);
- *backend_exec_params->be_desc.mutable_address() = FromTNetworkAddress(host_addr);
+ NetworkAddressPB host_addr = MakeNetworkAddressPB(host_name, 25000);
+ BackendExecParams& backend_exec_params =
+ query_schedule.GetOrCreateBackendExecParams(host_addr);
+ backend_exec_params.pb->set_min_mem_reservation_bytes(min_mem_reservation_bytes);
+ backend_exec_params.pb->set_slots_to_use(slots_to_use);
+ backend_exec_params.be_desc.set_admit_mem_limit(admit_mem_limit);
+ backend_exec_params.be_desc.set_admission_slots(slots_available);
+ backend_exec_params.be_desc.set_is_executor(true);
+ *backend_exec_params.be_desc.mutable_address() = host_addr;
if (i == 0) {
// Add first element as the coordinator.
- backend_exec_params->is_coord_backend = true;
- backend_exec_params->be_desc.set_is_executor(!is_dedicated_coord);
+ backend_exec_params.pb->set_is_coord_backend(true);
+ backend_exec_params.be_desc.set_is_executor(!is_dedicated_coord);
}
- per_backend_exec_params->emplace(host_addr, *backend_exec_params);
}
- query_schedule.set_per_backend_exec_params(*per_backend_exec_params);
}
/// Extract the host network addresses from 'schedule'.
- vector<TNetworkAddress> GetHostAddrs(const QuerySchedule& schedule) {
- vector<TNetworkAddress> host_addrs;
+ vector<NetworkAddressPB> GetHostAddrs(const QuerySchedule& schedule) {
+ vector<NetworkAddressPB> host_addrs;
for (auto& backend_state : schedule.per_backend_exec_params()) {
host_addrs.push_back(backend_state.first);
}
@@ -154,9 +153,9 @@ class AdmissionControllerTest : public testing::Test {
/// Set the slots in use for all the hosts in 'host_addrs'.
void SetSlotsInUse(AdmissionController* admission_controller,
- const vector<TNetworkAddress>& host_addrs, int slots_in_use) {
- for (TNetworkAddress host_addr : host_addrs) {
- string host = TNetworkAddressToString(host_addr);
+ const vector<NetworkAddressPB>& host_addrs, int slots_in_use) {
+ for (NetworkAddressPB host_addr : host_addrs) {
+ string host = NetworkAddressPBToString(host_addr);
admission_controller->host_stats_[host].slots_in_use = slots_in_use;
}
}
@@ -446,7 +445,7 @@ TEST_F(AdmissionControllerTest, CanAdmitRequestSlots) {
SetHostsInQuerySchedule(
*schedule, 2, false, MEGABYTE, 200L * MEGABYTE, slots_per_query, slots_per_host);
}
- vector<TNetworkAddress> host_addrs = GetHostAddrs(*default_group_schedule);
+ vector<NetworkAddressPB> host_addrs = GetHostAddrs(*default_group_schedule);
string not_admitted_reason;
// Simulate that there are just enough slots free for the query on all hosts.
@@ -796,27 +795,26 @@ TEST_F(AdmissionControllerTest, DedicatedCoordAdmissionChecks) {
const int64_t COORD_MEM_ESTIMATE = 150 * MEGABYTE;
QuerySchedule* query_schedule = MakeQuerySchedule(
"default", 0, pool_config, 2, PER_EXEC_MEM_ESTIMATE, COORD_MEM_ESTIMATE, true);
- PerBackendExecParams* per_backend_exec_params = pool_.Add(new PerBackendExecParams());
+ query_schedule->ClearBackendExecParams();
// Add coordinator backend.
- BackendExecParams* coord_exec_params = pool_.Add(new BackendExecParams());
- coord_exec_params->is_coord_backend = true;
- coord_exec_params->thread_reservation = 1;
- coord_exec_params->be_desc.set_admit_mem_limit(512 * MEGABYTE);
- coord_exec_params->be_desc.set_is_executor(false);
const string coord_host_name = Substitute("host$0", 1);
- TNetworkAddress coord_addr = MakeNetworkAddress(coord_host_name, 25000);
- const string coord_host = TNetworkAddressToString(coord_addr);
- per_backend_exec_params->emplace(coord_addr, *coord_exec_params);
+ NetworkAddressPB coord_addr = MakeNetworkAddressPB(coord_host_name, 25000);
+ const string coord_host = NetworkAddressPBToString(coord_addr);
+ BackendExecParams& coord_exec_params =
+ query_schedule->GetOrCreateBackendExecParams(coord_addr);
+ coord_exec_params.pb->set_is_coord_backend(true);
+ coord_exec_params.pb->set_thread_reservation(1);
+ coord_exec_params.be_desc.set_admit_mem_limit(512 * MEGABYTE);
+ coord_exec_params.be_desc.set_is_executor(false);
// Add executor backend.
- BackendExecParams* backend_exec_params = pool_.Add(new BackendExecParams());
- backend_exec_params->thread_reservation = 1;
- backend_exec_params->be_desc.set_admit_mem_limit(GIGABYTE);
- backend_exec_params->be_desc.set_is_executor(true);
const string exec_host_name = Substitute("host$0", 2);
- TNetworkAddress exec_addr = MakeNetworkAddress(exec_host_name, 25000);
- const string exec_host = TNetworkAddressToString(exec_addr);
- per_backend_exec_params->emplace(exec_addr, *backend_exec_params);
- query_schedule->set_per_backend_exec_params(*per_backend_exec_params);
+ NetworkAddressPB exec_addr = MakeNetworkAddressPB(exec_host_name, 25000);
+ const string exec_host = NetworkAddressPBToString(exec_addr);
+ BackendExecParams& backend_exec_params =
+ query_schedule->GetOrCreateBackendExecParams(exec_addr);
+ backend_exec_params.pb->set_thread_reservation(1);
+ backend_exec_params.be_desc.set_admit_mem_limit(GIGABYTE);
+ backend_exec_params.be_desc.set_is_executor(true);
string not_admitted_reason;
// Test 1: coord's admit_mem_limit < executor's admit_mem_limit. Query should not
// be rejected because query fits on both executor and coordinator. It should be
@@ -855,9 +853,7 @@ TEST_F(AdmissionControllerTest, DedicatedCoordAdmissionChecks) {
// Test 2: coord's admit_mem_limit < executor's admit_mem_limit. Query rejected because
// coord's admit_mem_limit is less than mem_to_admit on the coord.
// Re-using previous QuerySchedule object.
- coord_exec_params->be_desc.set_admit_mem_limit(100 * MEGABYTE);
- (*per_backend_exec_params)[coord_addr] = *coord_exec_params;
- query_schedule->set_per_backend_exec_params(*per_backend_exec_params);
+ coord_exec_params.be_desc.set_admit_mem_limit(100 * MEGABYTE);
ASSERT_TRUE(admission_controller->RejectForSchedule(
*query_schedule, pool_config, ¬_admitted_reason));
EXPECT_STR_CONTAINS(not_admitted_reason,
diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc
index a5c1402..715ad13 100644
--- a/be/src/scheduling/admission-controller.cc
+++ b/be/src/scheduling/admission-controller.cc
@@ -38,6 +38,7 @@
#include "util/runtime-profile-counters.h"
#include "util/thread.h"
#include "util/time.h"
+#include "util/uid-util.h"
#include "common/names.h"
@@ -400,44 +401,42 @@ void AdmissionController::PoolStats::Dequeue(bool timed_out) {
}
}
-void AdmissionController::UpdateStatsOnReleaseForBackends(
- const QuerySchedule& schedule, const vector<TNetworkAddress>& host_addrs) {
+void AdmissionController::UpdateStatsOnReleaseForBackends(const UniqueIdPB& query_id,
+ const RunningQuery& running_query, const vector<NetworkAddressPB>& host_addrs) {
int64_t total_mem_to_release = 0;
for (auto host_addr : host_addrs) {
- auto backend_exec_params = schedule.per_backend_exec_params().find(host_addr);
- if (backend_exec_params == schedule.per_backend_exec_params().end()) {
+ auto backend_allocation = running_query.per_backend_resources.find(host_addr);
+ if (backend_allocation == running_query.per_backend_resources.end()) {
string err_msg =
strings::Substitute("Error: Cannot find exec params of host $0 for query $1.",
- PrintThrift(host_addr), PrintId(schedule.query_id()));
+ NetworkAddressPBToString(host_addr), PrintId(query_id));
DCHECK(false) << err_msg;
LOG(ERROR) << err_msg;
continue;
}
- int64_t mem_to_release = GetMemToAdmit(schedule, backend_exec_params->second);
- UpdateHostStats(
- host_addr, -mem_to_release, -1, -backend_exec_params->second.slots_to_use);
- total_mem_to_release += mem_to_release;
+ UpdateHostStats(host_addr, -backend_allocation->second.mem_to_admit, -1,
+ -backend_allocation->second.slots_to_use);
+ total_mem_to_release += backend_allocation->second.mem_to_admit;
}
- PoolStats* pool_stats = GetPoolStats(schedule);
+ PoolStats* pool_stats = GetPoolStats(running_query.request_pool);
pool_stats->ReleaseMem(total_mem_to_release);
- pools_for_updates_.insert(schedule.request_pool());
+ pools_for_updates_.insert(running_query.request_pool);
}
void AdmissionController::UpdateStatsOnAdmission(const QuerySchedule& schedule) {
for (const auto& entry : schedule.per_backend_exec_params()) {
- const TNetworkAddress& host_addr = entry.first;
+ const NetworkAddressPB& host_addr = entry.first;
int64_t mem_to_admit = GetMemToAdmit(schedule, entry.second);
- UpdateHostStats(host_addr, mem_to_admit, 1, entry.second.slots_to_use);
+ UpdateHostStats(host_addr, mem_to_admit, 1, entry.second.pb->slots_to_use());
}
PoolStats* pool_stats = GetPoolStats(schedule);
pool_stats->AdmitQueryAndMemory(schedule);
pools_for_updates_.insert(schedule.request_pool());
}
-void AdmissionController::UpdateHostStats(
- const TNetworkAddress& host_addr, int64_t mem_to_admit, int num_queries_to_admit,
- int num_slots_to_admit) {
- const string host = TNetworkAddressToString(host_addr);
+void AdmissionController::UpdateHostStats(const NetworkAddressPB& host_addr,
+ int64_t mem_to_admit, int num_queries_to_admit, int num_slots_to_admit) {
+ const string host = NetworkAddressPBToString(host_addr);
VLOG_ROW << "Update admitted mem reserved for host=" << host
<< " prev=" << PrintBytes(host_stats_[host].mem_admitted)
<< " new=" << PrintBytes(host_stats_[host].mem_admitted + mem_to_admit);
@@ -515,8 +514,8 @@ bool AdmissionController::HasAvailableMemResources(const QuerySchedule& schedule
// Case 2:
for (const auto& entry : schedule.per_backend_exec_params()) {
- const TNetworkAddress& host = entry.first;
- const string host_id = TNetworkAddressToString(host);
+ const NetworkAddressPB& host = entry.first;
+ const string host_id = NetworkAddressPBToString(host);
int64_t admit_mem_limit = entry.second.be_desc.admit_mem_limit();
const HostStats& host_stats = host_stats_[host_id];
int64_t mem_reserved = host_stats.mem_reserved;
@@ -550,17 +549,17 @@ bool AdmissionController::HasAvailableMemResources(const QuerySchedule& schedule
bool AdmissionController::HasAvailableSlots(const QuerySchedule& schedule,
const TPoolConfig& pool_cfg, string* unavailable_reason) {
for (const auto& entry : schedule.per_backend_exec_params()) {
- const TNetworkAddress& host = entry.first;
- const string host_id = TNetworkAddressToString(host);
+ const NetworkAddressPB& host = entry.first;
+ const string host_id = NetworkAddressPBToString(host);
int64_t admission_slots = entry.second.be_desc.admission_slots();
int64_t slots_in_use = host_stats_[host_id].slots_in_use;
VLOG_ROW << "Checking available slot on host=" << host_id
<< " slots_in_use=" << slots_in_use
- << " needs=" << slots_in_use + entry.second.slots_to_use
+ << " needs=" << slots_in_use + entry.second.pb->slots_to_use()
<< " executor admission_slots=" << admission_slots;
- if (slots_in_use + entry.second.slots_to_use > admission_slots) {
+ if (slots_in_use + entry.second.pb->slots_to_use() > admission_slots) {
*unavailable_reason = Substitute(HOST_SLOT_NOT_AVAILABLE, host_id,
- entry.second.slots_to_use, slots_in_use, admission_slots);
+ entry.second.pb->slots_to_use(), slots_in_use, admission_slots);
return false;
}
}
@@ -654,32 +653,34 @@ bool AdmissionController::RejectForSchedule(const QuerySchedule& schedule,
// Compute the max (over all backends), the cluster totals (across all backends) for
// min_mem_reservation_bytes, thread_reservation, the min admit_mem_limit
// (over all executors) and the admit_mem_limit of the coordinator.
- pair<const TNetworkAddress*, int64_t> largest_min_mem_reservation(nullptr, -1);
+ pair<const NetworkAddressPB*, int64_t> largest_min_mem_reservation(nullptr, -1);
int64_t cluster_min_mem_reservation_bytes = 0;
- pair<const TNetworkAddress*, int64_t> max_thread_reservation(nullptr, 0);
- pair<const TNetworkAddress*, int64_t> min_executor_admit_mem_limit(
+ pair<const NetworkAddressPB*, int64_t> max_thread_reservation(nullptr, 0);
+ pair<const NetworkAddressPB*, int64_t> min_executor_admit_mem_limit(
nullptr, std::numeric_limits<int64_t>::max());
- pair<const TNetworkAddress*, int64_t> coord_admit_mem_limit(
+ pair<const NetworkAddressPB*, int64_t> coord_admit_mem_limit(
nullptr, std::numeric_limits<int64_t>::max());
int64_t cluster_thread_reservation = 0;
for (const auto& e : schedule.per_backend_exec_params()) {
const BackendExecParams& bp = e.second;
// TODO(IMPALA-8757): Extend slot based admission to default executor group
- if (!default_group && bp.slots_to_use > bp.be_desc.admission_slots()) {
- *rejection_reason = Substitute(REASON_NOT_ENOUGH_SLOTS_ON_BACKEND, bp.slots_to_use,
- NetworkAddressPBToString(bp.be_desc.address()), bp.be_desc.admission_slots());
+ if (!default_group && bp.pb->slots_to_use() > bp.be_desc.admission_slots()) {
+ *rejection_reason = Substitute(REASON_NOT_ENOUGH_SLOTS_ON_BACKEND,
+ bp.pb->slots_to_use(), NetworkAddressPBToString(bp.be_desc.address()),
+ bp.be_desc.admission_slots());
return true;
}
- cluster_min_mem_reservation_bytes += bp.min_mem_reservation_bytes;
- if (bp.min_mem_reservation_bytes > largest_min_mem_reservation.second) {
- largest_min_mem_reservation = make_pair(&e.first, bp.min_mem_reservation_bytes);
+ cluster_min_mem_reservation_bytes += bp.pb->min_mem_reservation_bytes();
+ if (bp.pb->min_mem_reservation_bytes() > largest_min_mem_reservation.second) {
+ largest_min_mem_reservation =
+ make_pair(&e.first, bp.pb->min_mem_reservation_bytes());
}
- cluster_thread_reservation += bp.thread_reservation;
- if (bp.thread_reservation > max_thread_reservation.second) {
- max_thread_reservation = make_pair(&e.first, bp.thread_reservation);
+ cluster_thread_reservation += bp.pb->thread_reservation();
+ if (bp.pb->thread_reservation() > max_thread_reservation.second) {
+ max_thread_reservation = make_pair(&e.first, bp.pb->thread_reservation());
}
- if (bp.is_coord_backend) {
+ if (bp.pb->is_coord_backend()) {
coord_admit_mem_limit.first = &e.first;
coord_admit_mem_limit.second = bp.be_desc.admit_mem_limit();
} else if (bp.be_desc.admit_mem_limit() < min_executor_admit_mem_limit.second) {
@@ -693,7 +694,7 @@ bool AdmissionController::RejectForSchedule(const QuerySchedule& schedule,
if (query_opts.__isset.buffer_pool_limit && query_opts.buffer_pool_limit > 0) {
if (largest_min_mem_reservation.second > query_opts.buffer_pool_limit) {
*rejection_reason = Substitute(REASON_BUFFER_LIMIT_TOO_LOW_FOR_RESERVATION,
- TNetworkAddressToString(*largest_min_mem_reservation.first),
+ NetworkAddressPBToString(*largest_min_mem_reservation.first),
PrintBytes(largest_min_mem_reservation.second));
return true;
}
@@ -707,7 +708,7 @@ bool AdmissionController::RejectForSchedule(const QuerySchedule& schedule,
&& query_opts.thread_reservation_limit > 0
&& max_thread_reservation.second > query_opts.thread_reservation_limit) {
*rejection_reason = Substitute(REASON_THREAD_RESERVATION_LIMIT_EXCEEDED,
- TNetworkAddressToString(*max_thread_reservation.first),
+ NetworkAddressPBToString(*max_thread_reservation.first),
max_thread_reservation.second, query_opts.thread_reservation_limit);
return true;
}
@@ -749,7 +750,7 @@ bool AdmissionController::RejectForSchedule(const QuerySchedule& schedule,
*rejection_reason =
Substitute(REASON_REQ_OVER_NODE_MEM, PrintBytes(executor_mem_to_admit),
PrintBytes(min_executor_admit_mem_limit.second),
- TNetworkAddressToString(*min_executor_admit_mem_limit.first));
+ NetworkAddressPBToString(*min_executor_admit_mem_limit.first));
return true;
}
int64_t coord_mem_to_admit = schedule.coord_backend_mem_to_admit();
@@ -759,7 +760,7 @@ bool AdmissionController::RejectForSchedule(const QuerySchedule& schedule,
if (coord_mem_to_admit > coord_admit_mem_limit.second) {
*rejection_reason = Substitute(REASON_REQ_OVER_NODE_MEM,
PrintBytes(coord_mem_to_admit), PrintBytes(coord_admit_mem_limit.second),
- TNetworkAddressToString(*coord_admit_mem_limit.first));
+ NetworkAddressPBToString(*coord_admit_mem_limit.first));
return true;
}
}
@@ -778,7 +779,7 @@ void AdmissionController::PoolStats::UpdateConfigMetrics(const TPoolConfig& pool
Status AdmissionController::SubmitForAdmission(const AdmissionRequest& request,
Promise<AdmissionOutcome, PromiseMode::MULTIPLE_PRODUCER>* admit_outcome,
- std::unique_ptr<QuerySchedule>* schedule_result) {
+ unique_ptr<QuerySchedulePB>* schedule_result) {
DCHECK(schedule_result->get() == nullptr);
ClusterMembershipMgr::SnapshotPtr membership_snapshot =
@@ -834,10 +835,10 @@ Status AdmissionController::SubmitForAdmission(const AdmissionRequest& request,
}
if (queue_node.admitted_schedule.get() != nullptr) {
+ DCHECK(queue_node.admitted_schedule->query_schedule_pb().get() != nullptr);
const string& group_name = queue_node.admitted_schedule->executor_group();
VLOG(3) << "Can admit to group " << group_name << " (or cancelled)";
DCHECK_EQ(stats->local_stats().num_queued, 0);
- *schedule_result = std::move(queue_node.admitted_schedule);
AdmissionOutcome outcome = admit_outcome->Set(AdmissionOutcome::ADMITTED);
if (outcome != AdmissionOutcome::ADMITTED) {
DCHECK_ENUM_EQ(outcome, AdmissionOutcome::CANCELLED);
@@ -846,9 +847,10 @@ Status AdmissionController::SubmitForAdmission(const AdmissionRequest& request,
return Status::CANCELLED;
}
VLOG_QUERY << "Admitting query id=" << PrintId(request.query_id);
- AdmitQuery(schedule_result->get(), false);
+ AdmitQuery(queue_node.admitted_schedule.get(), false);
stats->UpdateWaitTime(0);
VLOG_RPC << "Final: " << stats->DebugString();
+ *schedule_result = move(queue_node.admitted_schedule->query_schedule_pb());
return Status::OK();
}
@@ -925,7 +927,7 @@ Status AdmissionController::SubmitForAdmission(const AdmissionRequest& request,
// not change them here.
DCHECK_ENUM_EQ(outcome, AdmissionOutcome::ADMITTED);
DCHECK(queue_node.admitted_schedule.get() != nullptr);
- *schedule_result = std::move(queue_node.admitted_schedule);
+ *schedule_result = move(queue_node.admitted_schedule->query_schedule_pb());
DCHECK(!queue->Contains(&queue_node));
VLOG_QUERY << "Admitted queued query id=" << PrintId(request.query_id);
VLOG_RPC << "Final: " << pool_stats->DebugString();
@@ -934,38 +936,53 @@ Status AdmissionController::SubmitForAdmission(const AdmissionRequest& request,
}
void AdmissionController::ReleaseQuery(
- const QuerySchedule& schedule, int64_t peak_mem_consumption) {
- const string& pool_name = schedule.request_pool();
+ const UniqueIdPB& query_id, int64_t peak_mem_consumption) {
{
lock_guard<mutex> lock(admission_ctrl_lock_);
- DCHECK_EQ(num_released_backends_.at(schedule.query_id()), 0);
- num_released_backends_.erase(num_released_backends_.find(schedule.query_id()));
- PoolStats* stats = GetPoolStats(schedule);
+ auto it = running_queries_.find(query_id);
+ if (it == running_queries_.end()) {
+ LOG(DFATAL) << "Unable to find resources to release for query "
+ << PrintId(query_id);
+ return;
+ }
+
+ const RunningQuery& running_query = it->second;
+ DCHECK_EQ(num_released_backends_.at(query_id), 0) << PrintId(query_id);
+ num_released_backends_.erase(num_released_backends_.find(query_id));
+ PoolStats* stats = GetPoolStats(running_query.request_pool);
stats->ReleaseQuery(peak_mem_consumption);
// No need to update the Host Stats as they should have been updated in
// ReleaseQueryBackends.
- pools_for_updates_.insert(pool_name);
- UpdateExecGroupMetric(schedule.executor_group(), -1);
- VLOG_RPC << "Released query id=" << PrintId(schedule.query_id()) << " "
- << stats->DebugString();
+ pools_for_updates_.insert(running_query.request_pool);
+ UpdateExecGroupMetric(running_query.executor_group, -1);
+ VLOG_RPC << "Released query id=" << PrintId(query_id) << " " << stats->DebugString();
pending_dequeue_ = true;
+ running_queries_.erase(it);
}
dequeue_cv_.NotifyOne();
}
void AdmissionController::ReleaseQueryBackends(
- const QuerySchedule& schedule, const vector<TNetworkAddress>& host_addrs) {
+ const UniqueIdPB& query_id, const vector<NetworkAddressPB>& host_addrs) {
{
lock_guard<mutex> lock(admission_ctrl_lock_);
- UpdateStatsOnReleaseForBackends(schedule, host_addrs);
+ auto it = running_queries_.find(query_id);
+ if (it == running_queries_.end()) {
+ LOG(DFATAL) << "Unable to find resources to release for query backends "
+ << PrintId(query_id);
+ return;
+ }
+
+ const RunningQuery& running_query = it->second;
+ UpdateStatsOnReleaseForBackends(query_id, running_query, host_addrs);
// Update num_released_backends_.
- auto released_backends = num_released_backends_.find(schedule.query_id());
+ auto released_backends = num_released_backends_.find(query_id);
if (released_backends != num_released_backends_.end()) {
released_backends->second -= host_addrs.size();
} else {
- string err_msg = Substitute("Unable to find num released backends for query $0",
- PrintId(schedule.query_id()));
+ string err_msg = Substitute(
+ "Unable to find num released backends for query $0", PrintId(query_id));
DCHECK(false) << err_msg;
LOG(ERROR) << err_msg;
}
@@ -973,9 +990,9 @@ void AdmissionController::ReleaseQueryBackends(
if (VLOG_IS_ON(2)) {
stringstream ss;
ss << "Released query backend(s) ";
- for (auto host_addr : host_addrs) ss << PrintThrift(host_addr) << " ";
- ss << "for query id=" << PrintId(schedule.query_id()) << " "
- << GetPoolStats(schedule)->DebugString();
+ for (auto host_addr : host_addrs) ss << host_addr << " ";
+ ss << "for query id=" << PrintId(query_id) << " "
+ << GetPoolStats(running_query.request_pool)->DebugString();
VLOG(2) << ss.str();
}
pending_dequeue_ = true;
@@ -1390,7 +1407,7 @@ void AdmissionController::DequeueLoop() {
}
DCHECK(is_cancelled || queue_node->admitted_schedule != nullptr);
- const TUniqueId& query_id = queue_node->admission_request.query_id;
+ const UniqueIdPB& query_id = queue_node->admission_request.query_id;
if (!is_cancelled) {
VLOG_QUERY << "Admitting from queue: query=" << PrintId(query_id);
AdmissionOutcome outcome =
@@ -1521,6 +1538,17 @@ void AdmissionController::AdmitQuery(QuerySchedule* schedule, bool was_queued) {
num_released_backends_.find(schedule->query_id()) == num_released_backends_.end());
num_released_backends_[schedule->query_id()] =
schedule->per_backend_exec_params().size();
+
+ // Store info about the admitted resources so that we can release them.
+ DCHECK(running_queries_.find(schedule->query_id()) == running_queries_.end());
+ RunningQuery& running_query = running_queries_[schedule->query_id()];
+ running_query.request_pool = schedule->request_pool();
+ running_query.executor_group = schedule->executor_group();
+ for (const auto& entry : schedule->per_backend_exec_params()) {
+ BackendAllocation& allocation = running_query.per_backend_resources[entry.first];
+ allocation.slots_to_use = entry.second.pb->slots_to_use();
+ allocation.mem_to_admit = GetMemToAdmit(*schedule, entry.second);
+ }
}
string AdmissionController::GetStalenessDetail(const string& prefix,
@@ -1835,8 +1863,9 @@ bool AdmissionController::PoolLimitsRunningQueriesCount(const TPoolConfig& pool_
int64_t AdmissionController::GetMemToAdmit(
const QuerySchedule& schedule, const BackendExecParams& backend_exec_params) {
- return backend_exec_params.is_coord_backend ? schedule.coord_backend_mem_to_admit() :
- schedule.per_backend_mem_to_admit();
+ return backend_exec_params.pb->is_coord_backend() ?
+ schedule.coord_backend_mem_to_admit() :
+ schedule.per_backend_mem_to_admit();
}
void AdmissionController::UpdateExecGroupMetricMap(
diff --git a/be/src/scheduling/admission-controller.h b/be/src/scheduling/admission-controller.h
index 9be49bf..1f91ca6 100644
--- a/be/src/scheduling/admission-controller.h
+++ b/be/src/scheduling/admission-controller.h
@@ -319,7 +319,7 @@ class AdmissionController {
/// admit it. None of the members are owned by the instances of this class (usually they
/// are owned by the ClientRequestState).
struct AdmissionRequest {
- const TUniqueId& query_id;
+ const UniqueIdPB& query_id;
const TQueryExecRequest& request;
const TQueryOptions& query_options;
RuntimeProfile* summary_profile;
@@ -338,23 +338,23 @@ class AdmissionController {
/// cancelled to ensure that the pool statistics are updated.
Status SubmitForAdmission(const AdmissionRequest& request,
Promise<AdmissionOutcome, PromiseMode::MULTIPLE_PRODUCER>* admit_outcome,
- std::unique_ptr<QuerySchedule>* schedule_result);
+ std::unique_ptr<QuerySchedulePB>* schedule_result);
/// Updates the pool statistics when a query completes (either successfully,
/// is cancelled or failed). This should be called for all requests that have
- /// been submitted via AdmitQuery(). 'schedule' is the QuerySchedule of the completed
- /// query and 'peak_mem_consumption' is the peak memory consumption of the query.
+ /// been submitted via AdmitQuery(). 'query_id' is the completed query and
+ /// 'peak_mem_consumption' is the peak memory consumption of the query.
/// This does not block.
- void ReleaseQuery(const QuerySchedule& schedule, int64_t peak_mem_consumption);
+ void ReleaseQuery(const UniqueIdPB& query_id, int64_t peak_mem_consumption);
/// Updates the pool statistics when a Backend running a query completes (either
/// successfully, is cancelled or failed). This should be called for all Backends part
/// of a query for all queries that have been submitted via AdmitQuery().
- /// 'schedule' is the QuerySchedule of the associated query and the vector of
- /// TNetworkAddresses identify the completed Backends.
+ /// 'query_id' is the associated query and the vector of NetworkAddressPBs identify the
+ /// completed Backends.
/// This does not block.
void ReleaseQueryBackends(
- const QuerySchedule& schedule, const vector<TNetworkAddress>& host_addr);
+ const UniqueIdPB& query_id, const vector<NetworkAddressPB>& host_addr);
/// Registers the request queue topic with the statestore, starts up the dequeue thread
/// and registers a callback with the cluster membership manager to receive updates for
@@ -716,6 +716,35 @@ class AdmissionController {
typedef boost::unordered_map<std::string, RequestQueue> RequestQueueMap;
RequestQueueMap request_queue_map_;
+ /// Container for info about the resources allocated to a query on a single backend.
+ struct BackendAllocation {
+ /// Number of admission control slots this query is using.
+ int32_t slots_to_use;
+
+ /// Amount of memory allocated to this query. This will be equal to
+ /// QuerySchedule::coord_backend_mem_to_admit() if this is the coordinator backend, or
+ /// QuerySchedule::per_backend_mem_to_admit() otherwise.
+ int64_t mem_to_admit;
+ };
+
+ /// Container for info about the resources allocated to a currently running query.
+ struct RunningQuery {
+ /// The request pool this query was scheduled on.
+ std::string request_pool;
+
+ /// The executor group this query was scheduled on.
+ std::string executor_group;
+
+ /// Map from backend addresses to the resouces this query was allocated on them.
+ std::unordered_map<NetworkAddressPB, BackendAllocation> per_backend_resources;
+ };
+
+ /// Map from query id of currently running queries to information about the resources
+ /// that were allocated to them. Used to properly account for resources when releasing
+ /// queries.
+ /// Protected by admission_ctrl_lock_.
+ std::unordered_map<UniqueIdPB, RunningQuery> running_queries_;
+
/// Map of pool names to the pool configs returned by request_pool_service_. Stored so
/// that the dequeue thread does not need to access the configs via the request pool
/// service again (which involves a JNI call and error checking).
@@ -736,7 +765,7 @@ class AdmissionController {
/// Tracks the number of released Backends for each active query. Used purely for
/// internal state validation. Used to ensure that all Backends are released before
/// the query is released.
- typedef boost::unordered_map<TUniqueId, int> NumReleasedBackends;
+ typedef boost::unordered_map<UniqueIdPB, int> NumReleasedBackends;
NumReleasedBackends num_released_backends_;
/// Resolves the resource pool name in 'query_ctx.request_pool' and stores the resulting
@@ -840,14 +869,13 @@ class AdmissionController {
/// 'schedule' which have been release/completed. The list of completed backends is
/// specified in 'host_addrs'. Also updates the stats related to the admitted memory of
/// its associated resource pool.
- void UpdateStatsOnReleaseForBackends(
- const QuerySchedule& schedule, const std::vector<TNetworkAddress>& host_addrs);
+ void UpdateStatsOnReleaseForBackends(const UniqueIdPB& query_id,
+ const RunningQuery& running_query, const std::vector<NetworkAddressPB>& host_addrs);
/// Updates the memory admitted and the num of queries running on the specified host by
/// adding the specified mem, num_queries and slots to the host stats.
- void UpdateHostStats(
- const TNetworkAddress& host_addr, int64_t mem_to_admit, int num_queries_to_admit,
- int num_slots_to_admit);
+ void UpdateHostStats(const NetworkAddressPB& host_addr, int64_t mem_to_admit,
+ int num_queries_to_admit, int num_slots_to_admit);
/// Rejection happens in several stages
/// 1) Based on static pool configuration
diff --git a/be/src/scheduling/cluster-membership-mgr-test.cc b/be/src/scheduling/cluster-membership-mgr-test.cc
index 268ad26..a3c62fb 100644
--- a/be/src/scheduling/cluster-membership-mgr-test.cc
+++ b/be/src/scheduling/cluster-membership-mgr-test.cc
@@ -69,7 +69,7 @@ class ClusterMembershipMgrTest : public testing::Test {
/// A struct to hold information related to a simulated backend during the test.
struct Backend {
- string backend_id;
+ UniqueIdPB backend_id;
std::unique_ptr<MetricGroup> metric_group;
std::unique_ptr<ClusterMembershipMgr> cmm;
std::shared_ptr<BackendDescriptorPB> desc;
@@ -167,7 +167,7 @@ class ClusterMembershipMgrTest : public testing::Test {
backends_.push_back(make_unique<Backend>());
auto& be = backends_.back();
be->desc = make_shared<BackendDescriptorPB>(MakeBackendDescriptor(idx));
- be->backend_id = be->desc->address().hostname();
+ be->backend_id = be->desc->backend_id();
offline_.push_back(be.get());
return be.get();
}
@@ -179,7 +179,7 @@ class ClusterMembershipMgrTest : public testing::Test {
ASSERT_TRUE(IsInVector(be, offline_));
be->metric_group = make_unique<MetricGroup>("test");
be->cmm = make_unique<ClusterMembershipMgr>(
- be->backend_id, nullptr, be->metric_group.get());
+ PrintId(be->backend_id), nullptr, be->metric_group.get());
RemoveFromVector(be, &offline_);
starting_.push_back(be);
}
@@ -237,7 +237,7 @@ class ClusterMembershipMgrTest : public testing::Test {
// Create topic item before erasing the backend.
TTopicItem deletion_item;
- deletion_item.key = be->backend_id;
+ deletion_item.key = PrintId(be->backend_id);
deletion_item.deleted = true;
// Delete the backend
@@ -376,17 +376,17 @@ TEST_F(ClusterMembershipMgrTest, ExecutorBlacklist) {
}
// Tell a BE to blacklist itself, should have no effect.
- backends_[0]->cmm->BlacklistExecutor(*backends_[0]->desc, Status("error"));
+ backends_[0]->cmm->BlacklistExecutor(backends_[0]->desc->backend_id(), Status("error"));
EXPECT_EQ(NUM_BACKENDS, backends_[0]->cmm->GetSnapshot()->current_backends.size());
EXPECT_EQ(NUM_BACKENDS, GetDefaultGroupSize(*backends_[0]->cmm));
// Tell a BE to blacklist another BE, should remove it from executor_groups but not
// current_backends.
- backends_[0]->cmm->BlacklistExecutor(*backends_[1]->desc, Status("error"));
+ backends_[0]->cmm->BlacklistExecutor(backends_[1]->desc->backend_id(), Status("error"));
EXPECT_EQ(NUM_BACKENDS, backends_[0]->cmm->GetSnapshot()->current_backends.size());
EXPECT_EQ(NUM_BACKENDS - 1, GetDefaultGroupSize(*backends_[0]->cmm));
// Blacklist a BE that is already blacklisted. Should have no effect.
- backends_[0]->cmm->BlacklistExecutor(*backends_[1]->desc, Status("error"));
+ backends_[0]->cmm->BlacklistExecutor(backends_[1]->desc->backend_id(), Status("error"));
EXPECT_EQ(NUM_BACKENDS, backends_[0]->cmm->GetSnapshot()->current_backends.size());
EXPECT_EQ(NUM_BACKENDS - 1, GetDefaultGroupSize(*backends_[0]->cmm));
@@ -397,7 +397,7 @@ TEST_F(ClusterMembershipMgrTest, ExecutorBlacklist) {
EXPECT_EQ(NUM_BACKENDS, GetDefaultGroupSize(*backends_[0]->cmm));
// Blacklist the BE and sleep again.
- backends_[0]->cmm->BlacklistExecutor(*backends_[1]->desc, Status("error"));
+ backends_[0]->cmm->BlacklistExecutor(backends_[1]->desc->backend_id(), Status("error"));
EXPECT_EQ(NUM_BACKENDS, backends_[0]->cmm->GetSnapshot()->current_backends.size());
EXPECT_EQ(NUM_BACKENDS - 1, GetDefaultGroupSize(*backends_[0]->cmm));
usleep(BLACKLIST_TIMEOUT_SLEEP_US);
@@ -407,12 +407,12 @@ TEST_F(ClusterMembershipMgrTest, ExecutorBlacklist) {
EXPECT_EQ(NUM_BACKENDS, backends_[0]->cmm->GetSnapshot()->current_backends.size());
EXPECT_EQ(NUM_BACKENDS - 1, GetDefaultGroupSize(*backends_[0]->cmm));
// Try blacklisting the quiesced BE, should have no effect.
- backends_[0]->cmm->BlacklistExecutor(*backends_[1]->desc, Status("error"));
+ backends_[0]->cmm->BlacklistExecutor(backends_[1]->desc->backend_id(), Status("error"));
EXPECT_EQ(NUM_BACKENDS, backends_[0]->cmm->GetSnapshot()->current_backends.size());
EXPECT_EQ(NUM_BACKENDS - 1, GetDefaultGroupSize(*backends_[0]->cmm));
// Blacklist another BE and sleep.
- backends_[0]->cmm->BlacklistExecutor(*backends_[2]->desc, Status("error"));
+ backends_[0]->cmm->BlacklistExecutor(backends_[2]->desc->backend_id(), Status("error"));
EXPECT_EQ(NUM_BACKENDS, backends_[0]->cmm->GetSnapshot()->current_backends.size());
EXPECT_EQ(NUM_BACKENDS - 2, GetDefaultGroupSize(*backends_[0]->cmm));
usleep(BLACKLIST_TIMEOUT_SLEEP_US);
diff --git a/be/src/scheduling/cluster-membership-mgr.cc b/be/src/scheduling/cluster-membership-mgr.cc
index de8caeb..1016fcb 100644
--- a/be/src/scheduling/cluster-membership-mgr.cc
+++ b/be/src/scheduling/cluster-membership-mgr.cc
@@ -344,10 +344,20 @@ void ClusterMembershipMgr::UpdateMembership(
}
void ClusterMembershipMgr::BlacklistExecutor(
- const BackendDescriptorPB& be_desc, const Status& cause) {
+ const UniqueIdPB& backend_id, const Status& cause) {
DCHECK(!cause.ok());
if (!ExecutorBlacklist::BlacklistingEnabled()) return;
lock_guard<mutex> l(update_membership_lock_);
+
+ // Look up the descriptor in the current snapshot.
+ auto it = current_membership_->current_backends.find(PrintId(backend_id));
+ if (it == current_membership_->current_backends.end()) {
+ LOG(INFO) << "Did not blacklist " << PrintId(backend_id)
+ << " because it was already removed from the cluster membership.";
+ return;
+ }
+ const BackendDescriptorPB& be_desc = it->second;
+
// Don't blacklist the local executor. Some queries may have root fragments that must be
// scheduled on the coordinator and will always fail if its blacklisted.
if (be_desc.ip_address() == current_membership_->local_be_desc->ip_address()
diff --git a/be/src/scheduling/cluster-membership-mgr.h b/be/src/scheduling/cluster-membership-mgr.h
index e2ce538..5898537 100644
--- a/be/src/scheduling/cluster-membership-mgr.h
+++ b/be/src/scheduling/cluster-membership-mgr.h
@@ -73,7 +73,7 @@ class ClusterMembershipMgr {
/// immutable copy of a backend descriptor.
typedef std::shared_ptr<const BackendDescriptorPB> BeDescSharedPtr;
- /// Maps statestore subscriber IDs to backend descriptors.
+ /// Maps backend IDs to backend descriptors.
typedef std::unordered_map<std::string, BackendDescriptorPB> BackendIdMap;
/// Maps executor group names to executor groups. For now, only a default group exists
@@ -162,7 +162,7 @@ class ClusterMembershipMgr {
/// Adds the given backend to the local blacklist. Updates 'current_membership_' to
/// remove the backend from 'executor_groups' so that it will not be scheduled on.
/// 'cause' is an error status representing the reason the node was blacklisted.
- void BlacklistExecutor(const BackendDescriptorPB& be_desc, const Status& cause);
+ void BlacklistExecutor(const UniqueIdPB& backend_id, const Status& cause);
/// Returns a pointer to the static empty group reserved for scheduling coord only
/// queries.
diff --git a/be/src/scheduling/cluster-membership-test-util.cc b/be/src/scheduling/cluster-membership-test-util.cc
index 17d84c3..ba9912f 100644
--- a/be/src/scheduling/cluster-membership-test-util.cc
+++ b/be/src/scheduling/cluster-membership-test-util.cc
@@ -20,6 +20,7 @@
#include "common/names.h"
#include "scheduling/executor-group.h"
#include "service/impala-server.h"
+#include "util/uid-util.h"
static const int BACKEND_PORT = 1000;
static const int KRPC_PORT = 2000;
@@ -47,6 +48,7 @@ string HostIdxToIpAddr(int host_idx) {
BackendDescriptorPB MakeBackendDescriptor(
int idx, const ExecutorGroupDescPB& group_desc, int port_offset) {
BackendDescriptorPB be_desc;
+ UUIDToUniqueIdPB(boost::uuids::random_generator()(), be_desc.mutable_backend_id());
be_desc.mutable_address()->set_hostname(HostIdxToHostname(idx));
be_desc.mutable_address()->set_port(BACKEND_PORT + port_offset);
be_desc.set_ip_address(HostIdxToIpAddr(idx));
diff --git a/be/src/scheduling/query-schedule.cc b/be/src/scheduling/query-schedule.cc
index 11bcb0c..6c3d92e 100644
--- a/be/src/scheduling/query-schedule.cc
+++ b/be/src/scheduling/query-schedule.cc
@@ -17,24 +17,14 @@
#include "scheduling/query-schedule.h"
-#include <sstream>
-#include <boost/algorithm/string/join.hpp>
-#include <boost/uuid/uuid.hpp>
-#include <boost/uuid/uuid_generators.hpp>
-
#include "runtime/bufferpool/reservation-util.h"
+#include "scheduling/scheduler.h"
#include "util/mem-info.h"
-#include "util/network-util.h"
-#include "util/parse-util.h"
#include "util/test-info.h"
#include "util/uid-util.h"
#include "common/names.h"
-using boost::uuids::random_generator;
-using boost::uuids::uuid;
-using namespace impala;
-
DEFINE_bool_hidden(use_dedicated_coordinator_estimates, true,
"Hidden option to fall back to legacy memory estimation logic for dedicated"
" coordinators wherein the same per backend estimate is used for both coordinators "
@@ -42,46 +32,68 @@ DEFINE_bool_hidden(use_dedicated_coordinator_estimates, true,
namespace impala {
-QuerySchedule::QuerySchedule(const TUniqueId& query_id, const TQueryExecRequest& request,
+FInstanceExecParams::FInstanceExecParams(const UniqueIdPB& instance_id,
+ const NetworkAddressPB& host, const NetworkAddressPB& krpc_host,
+ int per_fragment_instance_idx, const FragmentExecParams& fragment_exec_params)
+ : host(host), krpc_host(krpc_host) {
+ *pb.mutable_instance_id() = instance_id;
+ pb.set_per_fragment_instance_idx(per_fragment_instance_idx);
+ pb.set_fragment_idx(fragment_exec_params.fragment.idx);
+}
+
+void FInstanceExecParams::AddScanRanges(
+ int scan_idx, const vector<ScanRangeParamsPB>& scan_ranges) {
+ ScanRangesPB& scan_ranges_pb = (*pb.mutable_per_node_scan_ranges())[scan_idx];
+ *scan_ranges_pb.mutable_scan_ranges() = {scan_ranges.begin(), scan_ranges.end()};
+}
+
+FragmentExecParams::FragmentExecParams(
+ const TPlanFragment& fragment, FragmentExecParamsPB* pb)
+ : is_coord_fragment(false), fragment(fragment), pb(pb) {
+ pb->set_fragment_idx(fragment.idx);
+}
+
+QuerySchedule::QuerySchedule(const UniqueIdPB& query_id, const TQueryExecRequest& request,
const TQueryOptions& query_options, RuntimeProfile* summary_profile,
RuntimeProfile::EventSequence* query_events)
: query_id_(query_id),
request_(request),
query_options_(query_options),
+ query_schedule_pb_(new QuerySchedulePB()),
summary_profile_(summary_profile),
query_events_(query_events),
- num_scan_ranges_(0),
next_instance_id_(query_id) {
Init();
}
-QuerySchedule::QuerySchedule(const TUniqueId& query_id, const TQueryExecRequest& request,
+QuerySchedule::QuerySchedule(const UniqueIdPB& query_id, const TQueryExecRequest& request,
const TQueryOptions& query_options, RuntimeProfile* summary_profile)
: query_id_(query_id),
request_(request),
query_options_(query_options),
+ query_schedule_pb_(new QuerySchedulePB()),
summary_profile_(summary_profile),
- num_scan_ranges_(0),
next_instance_id_(query_id) {
// Init() is not called, this constructor is for white box testing only.
DCHECK(TestInfo::is_test());
}
void QuerySchedule::Init() {
+ *query_schedule_pb_->mutable_query_id() = query_id_;
// extract TPlanFragments and order by fragment idx
- vector<const TPlanFragment*> fragments;
for (const TPlanExecInfo& plan_exec_info: request_.plan_exec_info) {
for (const TPlanFragment& fragment: plan_exec_info.fragments) {
- fragments.push_back(&fragment);
+ fragments_.emplace(fragment.idx, fragment);
}
}
- sort(fragments.begin(), fragments.end(),
- [](const TPlanFragment* a, const TPlanFragment* b) { return a->idx < b->idx; });
// this must only be called once
DCHECK_EQ(fragment_exec_params_.size(), 0);
- for (const TPlanFragment* fragment: fragments) {
- fragment_exec_params_.emplace_back(*fragment);
+ for (int i = 0; i < fragments_.size(); ++i) {
+ auto it = fragments_.find(i);
+ DCHECK(it != fragments_.end());
+ fragment_exec_params_.emplace_back(
+ it->second, query_schedule_pb_->add_fragment_exec_params());
}
// mark coordinator fragment
@@ -143,24 +155,24 @@ void QuerySchedule::Validate() const {
// we assigned the correct number of scan ranges per (host, node id):
// assemble a map from host -> (map from node id -> #scan ranges)
- unordered_map<TNetworkAddress, map<TPlanNodeId, int>> count_map;
- for (const FragmentExecParams& fp: fragment_exec_params_) {
- for (const FInstanceExecParams& ip: fp.instance_exec_params) {
- auto host_it = count_map.find(ip.host);
+ unordered_map<NetworkAddressPB, map<TPlanNodeId, int>> count_map;
+ for (const auto& entry : per_backend_exec_params_) {
+ for (const FInstanceExecParamsPB& ip : entry.second.pb->instance_params()) {
+ auto host_it = count_map.find(entry.second.pb->address());
if (host_it == count_map.end()) {
- count_map.insert(make_pair(ip.host, map<TPlanNodeId, int>()));
- host_it = count_map.find(ip.host);
+ count_map.insert(make_pair(entry.second.pb->address(), map<TPlanNodeId, int>()));
+ host_it = count_map.find(entry.second.pb->address());
}
map<TPlanNodeId, int>& node_map = host_it->second;
- for (const PerNodeScanRanges::value_type& instance_entry: ip.per_node_scan_ranges) {
+ for (const auto& instance_entry : ip.per_node_scan_ranges()) {
TPlanNodeId node_id = instance_entry.first;
auto count_entry = node_map.find(node_id);
if (count_entry == node_map.end()) {
node_map.insert(make_pair(node_id, 0));
count_entry = node_map.find(node_id);
}
- count_entry->second += instance_entry.second.size();
+ count_entry->second += instance_entry.second.scan_ranges_size();
}
}
}
@@ -168,14 +180,14 @@ void QuerySchedule::Validate() const {
for (const FragmentExecParams& fp: fragment_exec_params_) {
for (const FragmentScanRangeAssignment::value_type& assignment_entry:
fp.scan_range_assignment) {
- const TNetworkAddress& host = assignment_entry.first;
+ const NetworkAddressPB& host = assignment_entry.first;
DCHECK_GT(count_map.count(host), 0);
map<TPlanNodeId, int>& node_map = count_map.find(host)->second;
for (const PerNodeScanRanges::value_type& node_assignment:
assignment_entry.second) {
TPlanNodeId node_id = node_assignment.first;
DCHECK_GT(node_map.count(node_id), 0)
- << host.hostname << " " << host.port << " node_id=" << node_id;
+ << host.hostname() << " " << host.port() << " node_id=" << node_id;
DCHECK_EQ(node_map[node_id], node_assignment.second.size());
}
}
@@ -186,10 +198,20 @@ void QuerySchedule::Validate() const {
DCHECK_GT(fp.instance_exec_params.size(), 0) << fp.fragment;
}
+ // Check that all backends have instances, except possibly the coordaintor backend.
for (const auto& elem: per_backend_exec_params_) {
const BackendExecParams& bp = elem.second;
- DCHECK(!bp.instance_params.empty() || bp.is_coord_backend);
+ DCHECK(!bp.pb->instance_params().empty() || bp.pb->is_coord_backend());
+ }
+}
+BackendExecParams& QuerySchedule::GetOrCreateBackendExecParams(
+ const NetworkAddressPB& address) {
+ auto it = per_backend_exec_params_.find(address);
+ if (it == per_backend_exec_params_.end()) {
+ BackendExecParamsPB* pb = query_schedule_pb_->add_backend_exec_params();
+ it = per_backend_exec_params_.emplace(address, BackendExecParams(pb)).first;
}
+ return it->second;
}
int64_t QuerySchedule::GetPerExecutorMemoryEstimate() const {
@@ -202,83 +224,31 @@ int64_t QuerySchedule::GetDedicatedCoordMemoryEstimate() const {
return request_.dedicated_coord_mem_estimate;
}
-TUniqueId QuerySchedule::GetNextInstanceId() {
- TUniqueId result = next_instance_id_;
- ++next_instance_id_.lo;
- return result;
-}
-
-const TPlanFragment& FInstanceExecParams::fragment() const {
- return fragment_exec_params.fragment;
-}
-
-const TPlanFragment* QuerySchedule::GetCoordFragment() const {
- // Only have coordinator fragment for statements that return rows.
- if (request_.stmt_type != TStmtType::QUERY) return nullptr;
- const TPlanFragment* fragment = &request_.plan_exec_info[0].fragments[0];
- DCHECK_EQ(fragment->partition.type, TPartitionType::UNPARTITIONED);
- return fragment;
+void QuerySchedule::IncNumScanRanges(int64_t delta) {
+ query_schedule_pb_->set_num_scan_ranges(query_schedule_pb_->num_scan_ranges() + delta);
}
-void QuerySchedule::GetTPlanFragments(vector<const TPlanFragment*>* fragments) const {
- fragments->clear();
- for (const TPlanExecInfo& plan_info: request_.plan_exec_info) {
- for (const TPlanFragment& fragment: plan_info.fragments) {
- fragments->push_back(&fragment);
- }
- }
-}
-
-const FInstanceExecParams& QuerySchedule::GetCoordInstanceExecParams() const {
- DCHECK(RequiresCoordinatorFragment());
- const TPlanFragment& coord_fragment = request_.plan_exec_info[0].fragments[0];
- const FragmentExecParams& fragment_params = fragment_exec_params_[coord_fragment.idx];
- DCHECK_EQ(fragment_params.instance_exec_params.size(), 1);
- return fragment_params.instance_exec_params[0];
-}
-
-vector<int> FragmentExecParams::GetInstanceIdxs() const {
- vector<int> result;
- for (const FInstanceExecParams& instance_params: instance_exec_params) {
- result.push_back(GetInstanceIdx(instance_params.instance_id));
- }
- return result;
-}
-
-int FragmentExecParams::GetNumBackends() const {
- int result = 0;
- const TNetworkAddress* prev_krpc_host = nullptr;
- for (const FInstanceExecParams& instance_params: instance_exec_params) {
- if (prev_krpc_host == nullptr || *prev_krpc_host != instance_params.krpc_host) {
- ++result;
- prev_krpc_host = &instance_params.krpc_host;
- }
- }
+UniqueIdPB QuerySchedule::GetNextInstanceId() {
+ UniqueIdPB result = next_instance_id_;
+ next_instance_id_.set_lo(next_instance_id_.lo() + 1);
return result;
}
-int QuerySchedule::GetNumFragmentInstances() const {
- int total = 0;
- for (const FragmentExecParams& p: fragment_exec_params_) {
- total += p.instance_exec_params.size();
- }
- return total;
-}
-
int64_t QuerySchedule::GetClusterMemoryToAdmit() const {
// There will always be an entry for the coordinator in per_backend_exec_params_.
- return per_backend_mem_to_admit() * (per_backend_exec_params_.size() - 1)
+ return query_schedule_pb_->per_backend_mem_to_admit()
+ * (per_backend_exec_params_.size() - 1)
+ coord_backend_mem_to_admit();
}
bool QuerySchedule::UseDedicatedCoordEstimates() const {
- for (auto& itr : per_backend_exec_params_) {
- if (!itr.second.is_coord_backend) continue;
+ for (const auto& itr : per_backend_exec_params_) {
+ if (!itr.second.pb->is_coord_backend()) continue;
auto& coord = itr.second;
bool is_dedicated_coord = !coord.be_desc.is_executor();
bool only_coord_fragment_scheduled =
- RequiresCoordinatorFragment() && coord.instance_params.size() == 1;
- bool no_fragment_scheduled = coord.instance_params.size() == 0;
+ RequiresCoordinatorFragment() && coord.pb->instance_params().size() == 1;
+ bool no_fragment_scheduled = coord.pb->instance_params().size() == 0;
return FLAGS_use_dedicated_coordinator_estimates && is_dedicated_coord
&& (only_coord_fragment_scheduled || no_fragment_scheduled);
}
@@ -295,75 +265,80 @@ void QuerySchedule::UpdateMemoryRequirements(const TPoolConfig& pool_cfg) {
pool_cfg.min_query_mem_limit == 0 && pool_cfg.max_query_mem_limit == 0;
bool use_dedicated_coord_estimates = UseDedicatedCoordEstimates();
- per_backend_mem_to_admit_ = 0;
- coord_backend_mem_to_admit_ = 0;
+ int64_t per_backend_mem_to_admit = 0;
+ int64_t coord_backend_mem_to_admit = 0;
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;
+ per_backend_mem_to_admit = query_options().mem_limit;
+ coord_backend_mem_to_admit = query_options().mem_limit;
is_mem_limit_set = true;
}
if (!is_mem_limit_set) {
- per_backend_mem_to_admit_ = GetPerExecutorMemoryEstimate();
- coord_backend_mem_to_admit_ = use_dedicated_coord_estimates ?
+ per_backend_mem_to_admit = GetPerExecutorMemoryEstimate();
+ coord_backend_mem_to_admit = use_dedicated_coord_estimates ?
GetDedicatedCoordMemoryEstimate() :
GetPerExecutorMemoryEstimate();
if (!mimic_old_behaviour) {
int64_t min_mem_limit_required =
ReservationUtil::GetMinMemLimitFromReservation(largest_min_reservation());
- per_backend_mem_to_admit_ = max(per_backend_mem_to_admit_, min_mem_limit_required);
+ per_backend_mem_to_admit = max(per_backend_mem_to_admit, min_mem_limit_required);
int64_t min_coord_mem_limit_required =
ReservationUtil::GetMinMemLimitFromReservation(coord_min_reservation());
- coord_backend_mem_to_admit_ =
- max(coord_backend_mem_to_admit_, min_coord_mem_limit_required);
+ coord_backend_mem_to_admit =
+ max(coord_backend_mem_to_admit, min_coord_mem_limit_required);
}
}
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);
+ per_backend_mem_to_admit =
+ max(per_backend_mem_to_admit, pool_cfg.min_query_mem_limit);
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.
- coord_backend_mem_to_admit_ =
- max(coord_backend_mem_to_admit_, pool_cfg.min_query_mem_limit);
+ coord_backend_mem_to_admit =
+ max(coord_backend_mem_to_admit, pool_cfg.min_query_mem_limit);
}
}
if (pool_cfg.max_query_mem_limit > 0) {
- per_backend_mem_to_admit_ =
- min(per_backend_mem_to_admit_, pool_cfg.max_query_mem_limit);
- coord_backend_mem_to_admit_ =
- min(coord_backend_mem_to_admit_, pool_cfg.max_query_mem_limit);
+ per_backend_mem_to_admit =
+ min(per_backend_mem_to_admit, pool_cfg.max_query_mem_limit);
+ coord_backend_mem_to_admit =
+ min(coord_backend_mem_to_admit, pool_cfg.max_query_mem_limit);
}
}
// Cap the memory estimate at the amount of physical memory available. The user's
// provided value or the estimate from planning can each be unreasonable.
- per_backend_mem_to_admit_ = min(per_backend_mem_to_admit_, MemInfo::physical_mem());
- coord_backend_mem_to_admit_ = min(coord_backend_mem_to_admit_, MemInfo::physical_mem());
+ per_backend_mem_to_admit = min(per_backend_mem_to_admit, MemInfo::physical_mem());
+ coord_backend_mem_to_admit = min(coord_backend_mem_to_admit, MemInfo::physical_mem());
// If the query is only scheduled to run on the coordinator.
if (per_backend_exec_params_.size() == 1 && RequiresCoordinatorFragment()) {
- per_backend_mem_to_admit_ = 0;
+ per_backend_mem_to_admit = 0;
}
+ int64_t per_backend_mem_limit;
if (mimic_old_behaviour && !is_mem_limit_set) {
- per_backend_mem_limit_ = -1;
- coord_backend_mem_limit_ = -1;
+ per_backend_mem_limit = -1;
+ query_schedule_pb_->set_coord_backend_mem_limit(-1);
} else {
- per_backend_mem_limit_ = per_backend_mem_to_admit_;
- coord_backend_mem_limit_ = coord_backend_mem_to_admit_;
+ per_backend_mem_limit = per_backend_mem_to_admit;
+ query_schedule_pb_->set_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_;
+ per_backend_mem_to_admit = query_options().mem_limit_executors;
+ per_backend_mem_limit = per_backend_mem_to_admit;
}
+
+ query_schedule_pb_->set_coord_backend_mem_to_admit(coord_backend_mem_to_admit);
+ query_schedule_pb_->set_per_backend_mem_limit(per_backend_mem_limit);
+ query_schedule_pb_->set_per_backend_mem_to_admit(per_backend_mem_to_admit);
}
void QuerySchedule::set_executor_group(string executor_group) {
diff --git a/be/src/scheduling/query-schedule.h b/be/src/scheduling/query-schedule.h
index 36527c7..642c3a2 100644
--- a/be/src/scheduling/query-schedule.h
+++ b/be/src/scheduling/query-schedule.h
@@ -15,18 +15,18 @@
// specific language governing permissions and limitations
// under the License.
-#ifndef SCHEDULING_QUERY_SCHEDULE_H
-#define SCHEDULING_QUERY_SCHEDULE_H
+#pragma once
-#include <vector>
+#include <map>
#include <string>
#include <unordered_map>
+#include <vector>
#include "common/global-types.h"
-#include "gen-cpp/Frontend_types.h"
-#include "gen-cpp/Types_types.h" // for TNetworkAddress
-#include "gen-cpp/control_service.pb.h"
+#include "gen-cpp/Types_types.h"
+#include "gen-cpp/admission_control_service.pb.h"
#include "gen-cpp/statestore_service.pb.h"
+#include "scheduling/executor-group.h"
#include "util/container-util.h"
#include "util/runtime-profile.h"
@@ -40,160 +40,133 @@ typedef std::map<TPlanNodeId, std::vector<ScanRangeParamsPB>> PerNodeScanRanges;
/// map from an impalad host address to the per-node assigned scan ranges;
/// records scan range assignment for a single fragment
-typedef std::unordered_map<TNetworkAddress, PerNodeScanRanges>
+typedef std::unordered_map<NetworkAddressPB, PerNodeScanRanges>
FragmentScanRangeAssignment;
/// Execution parameters for a single backend. This gets created for every backend that
-/// participates in query execution, which includes, every backend that has fragments
-/// scheduled on it and the coordinator backend. Computed by Scheduler::Schedule(), set
-/// via QuerySchedule::set_per_backend_exec_params(). Used as an input to
-/// AdmissionController and a BackendState.
+/// participates in query execution, which includes every backend that has fragments
+/// scheduled on it and the coordinator backend.
+///
+/// Created by QuerySchedule::GetOrCreateBackendExecParams() and initialized in
+/// Scheduler::ComputeBackendExecParams(). Used as an input to the
+/// AdmissionController and Coordinator::BackendState.
struct BackendExecParams {
BackendDescriptorPB be_desc;
- /// The fragment instance params assigned to this backend. All instances of a
- /// particular fragment are contiguous in this vector. Query lifetime;
- /// FInstanceExecParams are owned by QuerySchedule::fragment_exec_params_.
- /// This can be empty only for the coordinator backend, that is, if 'is_coord_backend'
- /// is true.
- std::vector<const FInstanceExecParams*> instance_params;
-
- // The minimum query-wide buffer reservation size (in bytes) required for this backend.
- // This is the peak minimum reservation that may be required by the
- // concurrently-executing operators at any point in query execution. It may be less
- // than the initial reservation total claims (below) if execution of some operators
- // never overlaps, which allows reuse of reservations.
- int64_t min_mem_reservation_bytes = 0;
-
- // Total of the initial buffer reservations that we expect to be claimed on this
- // backend for all fragment instances in instance_params. I.e. the sum over all
- // operators in all fragment instances that execute on this backend. This is used for
- // an optimization in InitialReservation. Measured in bytes.
- int64_t initial_mem_reservation_total_claims = 0;
-
- // Total thread reservation for fragment instances scheduled on this backend. This is
- // the peak number of required threads that may be required by the
- // concurrently-executing fragment instances at any point in query execution.
- int64_t thread_reservation = 0;
-
- // Number of slots that this query should count for in admission control.
- // This is calculated as the maximum # of instances of any fragment on this backend.
- // I.e. 1 if mt_dop is not used and at most the mt_dop value if mt_dop is specified
- // (but less if the query is not actually running with mt_dop instances on this node).
- int slots_to_use = 0;
-
- // Indicates whether this backend is the coordinator.
- bool is_coord_backend = false;
+ /// Pointer to the corresponding protobuf struct containing any parameters for this
+ /// backend that will need to be sent back to the coordinator. Owned by
+ /// QuerySchedule::query_schedule_pb_.
+ BackendExecParamsPB* pb;
+
+ explicit BackendExecParams(BackendExecParamsPB* pb) : pb(pb) {}
};
-/// Map from an impalad host address to the list of assigned fragment instance params.
-typedef std::unordered_map<TNetworkAddress, BackendExecParams> PerBackendExecParams;
+/// Map from an impalad backend address to the exec params for that backend.
+typedef std::unordered_map<NetworkAddressPB, BackendExecParams> PerBackendExecParams;
-/// Execution parameters for a single fragment instance; used to assemble the
-/// TPlanFragmentInstanceCtx
+/// Execution parameters for a single fragment instance. Contains both intermediate
+/// info needed by the scheduler and info that will be sent back to the coordinator.
+///
+/// FInstanceExecParams are created as children of FragmentExecParams (in
+/// 'instance_exec_params') and then the calculated execution parameters, 'pb', are
+/// transferred to the corresponding BackendExecParams in
+/// Scheduler::ComputeBackendExecParams().
struct FInstanceExecParams {
- TUniqueId instance_id;
- TNetworkAddress host; // Thrift address of execution backend.
- TNetworkAddress krpc_host; // Krpc address of execution backend.
- PerNodeScanRanges per_node_scan_ranges;
-
- /// 0-based ordinal of this particular instance within its fragment (not: query-wide)
- int per_fragment_instance_idx;
-
- /// In its role as a data sender, a fragment instance is assigned a "sender id" to
- /// uniquely identify it to a receiver. -1 = invalid.
- int sender_id = -1;
-
- // List of input join build finstances for joins in this finstance.
- std::vector<JoinBuildInputPB> join_build_inputs;
-
- // If this is a join build fragment, the number of fragment instances that consume the
- // join build. -1 = invalid.
- int num_join_build_outputs = -1;
-
- /// The parent FragmentExecParams
- const FragmentExecParams& fragment_exec_params;
- const TPlanFragment& fragment() const;
-
- FInstanceExecParams(const TUniqueId& instance_id, const TNetworkAddress& host,
- const TNetworkAddress& krpc_host, int per_fragment_instance_idx,
- const FragmentExecParams& fragment_exec_params)
- : instance_id(instance_id),
- host(host),
- krpc_host(krpc_host),
- per_fragment_instance_idx(per_fragment_instance_idx),
- fragment_exec_params(fragment_exec_params) {}
-};
+ /// Thrift address of execution backend.
+ NetworkAddressPB host;
-/// Execution parameters shared between fragment instances
-struct FragmentExecParams {
- /// output destinations of this fragment
- std::vector<PlanFragmentDestinationPB> destinations;
+ /// Krpc address of execution backend.
+ NetworkAddressPB krpc_host;
+
+ /// Contains any info that needs to be sent back to the coordinator. Computed during
+ /// Scheduler::ComputeFragmentExecParams() then transferred to the corresponding
+ /// BackendExecParamsPB in Scheduler::ComputeBackendExecParams(), after which it is no
+ /// longer valid to access.
+ FInstanceExecParamsPB pb;
- /// map from node id to the number of senders (node id expected to be for an
- /// ExchangeNode)
- std::map<PlanNodeId, int> per_exch_num_senders;
+ FInstanceExecParams(const UniqueIdPB& instance_id, const NetworkAddressPB& host,
+ const NetworkAddressPB& krpc_host, int per_fragment_instance_idx,
+ const FragmentExecParams& fragment_exec_params);
- // only needed as intermediate state during exec parameter computation;
- // for scheduling, refer to FInstanceExecParams.per_node_scan_ranges
+ /// Adds the ranges in 'scan_ranges' to the scan at 'scan_idx' in 'pb'.
+ void AddScanRanges(int scan_idx, const std::vector<ScanRangeParamsPB>& scan_ranges);
+};
+
+/// Execution parameters shared between fragment instances. This struct is a container for
+/// any intermediate data needed for scheduling that will not be sent back to the
+/// coordinator as part of the QuerySchedulePB along with a pointer to the corresponding
+/// FragmentExecParamsPB in the QuerySchedulePB.
+struct FragmentExecParams {
+ /// Only needed as intermediate state during exec parameter computation.
+ /// For scheduling, refer to FInstanceExecParams.per_node_scan_ranges
FragmentScanRangeAssignment scan_range_assignment;
bool is_coord_fragment;
const TPlanFragment& fragment;
- // Fragments that are inputs to an ExchangeNode of this fragment.
+ /// Fragments that are inputs to an ExchangeNode of this fragment.
std::vector<FragmentIdx> exchange_input_fragments;
- // Instances of this fragment. Instances on a backend are clustered together - i.e. all
- // instances for a given backend will be consecutive entries in the vector.
+ /// Instances of this fragment. Instances on a backend are clustered together - i.e. all
+ /// instances for a given backend will be consecutive entries in the vector. These have
+ /// their protobuf params Swap()-ed to the BackendExecParamsPB during
+ /// Scheduler::ComputeBackendExecParams() and are no longer valid after that.
std::vector<FInstanceExecParams> instance_exec_params;
- FragmentExecParams(const TPlanFragment& fragment)
- : is_coord_fragment(false), fragment(fragment) {}
-
- // extract instance indices from instance_exec_params.instance_id
- std::vector<int> GetInstanceIdxs() const;
+ /// Pointer to the corresponding FragmentExecParamsPB in the parent QuerySchedule's
+ /// 'query_schedule_pb_'
+ FragmentExecParamsPB* pb;
- // Calculate distinct number of backends finstances are scheduled on.
- int GetNumBackends() const;
+ FragmentExecParams(const TPlanFragment& fragment, FragmentExecParamsPB* pb);
};
-/// A QuerySchedule contains all necessary information for a query coordinator to
-/// generate fragment execution requests and start query execution. If resource management
-/// is enabled, then a schedule also contains the resource reservation request
-/// and the granted resource reservation.
+/// QuerySchedule is a container class for scheduling data used by Scheduler and
+/// AdmissionController, which perform the scheduling logic itself, and it is only
+/// intednded to be accessed by them. The information needed for the coordinator to begin
+/// execution is stored in 'query_schedule_pb_', which is returned from the
+/// AdmissionController on successful admission. Everything else is intermediate data
+/// needed to calculate the schedule but is discarded after a scheduling decision is made.
///
-/// QuerySchedule is a container class for scheduling data, but it doesn't contain
-/// scheduling logic itself.
-/// The general usage pattern is that part of its state gets set from the static
-/// TQueryExecRequest during initialization, then the actual schedule gets set by the
-/// scheduler, then finally it is passed to the admission controller that keeps updating
-/// the memory requirements by calling UpdateMemoryRequirements() every time it tries to
-/// admit the query but only sets the final values once the query gets admitted
-/// successfully. Note: Due to this usage pattern the memory requirement values should not
-/// be accessed by other clients of this class while the query is in admission control
-/// phase.
+/// The general usage pattern is:
+/// - FragmentExecParams are created for each fragment in the plan. They are given
+/// pointers to corresponding FragmentExecParamsPBs created in the QuerySchedulePB.
+/// - FInstanceExecParams are created as children of the FragmentExecParams for each
+/// finstance and assigned to hosts. The FInstanceExecParams each have a corresponding
+/// FInstanceExecParamsPB that they initially own.
+/// - The scheduler computes the BackendExecParams for each backend that was assigned a
+/// fragment instance (and the coordinator backend). They are given pointers to
+/// corresponding BackendExecParamsPBs created in the QuerySchedulePB and the
+/// FInstanceExecParamsPB are Swap()-ed into them.
+/// - The QuerySchedule is passed to the admission controller, which keeps updating the
+/// memory requirements by calling UpdateMemoryRequirements() every time it tries to
+/// admit the query and sets the final values once the query gets admitted successfully.
+/// - On successful admission, the QuerySchedulePB is returned to the coordinator and
+/// everything else is discarded.
class QuerySchedule {
public:
- QuerySchedule(const TUniqueId& query_id, const TQueryExecRequest& request,
+ QuerySchedule(const UniqueIdPB& query_id, const TQueryExecRequest& request,
const TQueryOptions& query_options, RuntimeProfile* summary_profile,
RuntimeProfile::EventSequence* query_events);
/// For testing only: build a QuerySchedule object but do not run Init().
- QuerySchedule(const TUniqueId& query_id, const TQueryExecRequest& request,
+ QuerySchedule(const UniqueIdPB& query_id, const TQueryExecRequest& request,
const TQueryOptions& query_options, RuntimeProfile* summary_profile);
/// Verifies that the schedule is well-formed (and DCHECKs if it isn't):
/// - all fragments have a FragmentExecParams
/// - all scan ranges are assigned
/// - all BackendExecParams have instances assigned except for coordinator.
+ /// Expected to be called after the BackendExecParams have been computed, i.e. the
+ /// FInstanceExecParamsPB will have already been swapped.
void Validate() const;
- const TUniqueId& query_id() const { return query_id_; }
+ const UniqueIdPB& query_id() const { return query_id_; }
const TQueryExecRequest& request() const { return request_; }
const TQueryOptions& query_options() const { return query_options_; }
- // Valid after Schedule() succeeds.
+ std::unique_ptr<QuerySchedulePB>& query_schedule_pb() { return query_schedule_pb_; }
+
+ /// Valid after Schedule() succeeds.
const std::string& request_pool() const { return request().query_ctx.request_pool; }
/// Returns the estimated memory (bytes) per-node from planning.
@@ -205,31 +178,20 @@ class QuerySchedule {
int64_t GetDedicatedCoordMemoryEstimate() const;
/// Helper methods used by scheduler to populate this QuerySchedule.
- void IncNumScanRanges(int64_t delta) { num_scan_ranges_ += delta; }
-
- /// Return the coordinator fragment, or nullptr if there isn't one.
- const TPlanFragment* GetCoordFragment() const;
-
- /// Return all fragments belonging to exec request in 'fragments'.
- void GetTPlanFragments(std::vector<const TPlanFragment*>* fragments) const;
-
- int64_t num_scan_ranges() const { return num_scan_ranges_; }
+ void IncNumScanRanges(int64_t delta);
/// Map node ids to the id of their containing fragment.
FragmentIdx GetFragmentIdx(PlanNodeId id) const {
return plan_node_to_fragment_idx_[id];
}
- /// Return the total number of instances across all fragments.
- int GetNumFragmentInstances() const;
-
/// Returns next instance id. Instance ids are consecutive numbers generated from
/// the query id.
/// If the query contains a coordinator fragment instance, the generated instance
/// ids start at 1 and the caller is responsible for assigning the correct id
/// to the coordinator instance. If the query does not contain a coordinator instance,
/// the generated instance ids start at 0.
- TUniqueId GetNextInstanceId();
+ UniqueIdPB GetNextInstanceId();
const TPlanFragment& GetContainingFragment(PlanNodeId node_id) const {
FragmentIdx fragment_idx = GetFragmentIdx(node_id);
@@ -245,7 +207,19 @@ class QuerySchedule {
const PerBackendExecParams& per_backend_exec_params() const {
return per_backend_exec_params_;
}
- const std::vector<FragmentExecParams>& fragment_exec_params() const {
+ PerBackendExecParams& per_backend_exec_params() { return per_backend_exec_params_; }
+
+ /// Returns a reference to the BackendExecParams corresponding to 'address', creating
+ /// it if it doesn't already exist.
+ BackendExecParams& GetOrCreateBackendExecParams(const NetworkAddressPB& address);
+
+ /// Removes any BackendExecParams that have been created. Only used in testing.
+ void ClearBackendExecParams() {
+ per_backend_exec_params_.clear();
+ query_schedule_pb_->clear_backend_exec_params();
+ }
+
+ std::vector<FragmentExecParams>& fragment_exec_params() {
return fragment_exec_params_;
}
const FragmentExecParams& GetFragmentExecParams(FragmentIdx idx) const {
@@ -255,7 +229,6 @@ class QuerySchedule {
return &fragment_exec_params_[idx];
}
- const FInstanceExecParams& GetCoordInstanceExecParams() const;
RuntimeProfile* summary_profile() { return summary_profile_; }
RuntimeProfile::EventSequence* query_events() { return query_events_; }
@@ -265,25 +238,25 @@ class QuerySchedule {
int64_t coord_min_reservation() const { return coord_min_reservation_; }
/// Must call UpdateMemoryRequirements() at least once before calling this.
- int64_t per_backend_mem_limit() const { return per_backend_mem_limit_; }
+ int64_t per_backend_mem_limit() const {
+ return query_schedule_pb_->per_backend_mem_limit();
+ }
/// Must call UpdateMemoryRequirements() at least once before calling this.
int64_t per_backend_mem_to_admit() const {
- DCHECK_GE(per_backend_mem_to_admit_, 0);
- return per_backend_mem_to_admit_;
+ DCHECK_GE(query_schedule_pb_->per_backend_mem_to_admit(), 0);
+ return query_schedule_pb_->per_backend_mem_to_admit();
}
/// Must call UpdateMemoryRequirements() at least once before calling this.
- int64_t coord_backend_mem_limit() const { return coord_backend_mem_limit_; }
+ int64_t coord_backend_mem_limit() const {
+ return query_schedule_pb_->coord_backend_mem_limit();
+ }
/// Must call UpdateMemoryRequirements() at least once before calling this.
int64_t coord_backend_mem_to_admit() const {
- DCHECK_GT(coord_backend_mem_to_admit_, 0);
- return coord_backend_mem_to_admit_;
- }
-
- void set_per_backend_exec_params(const PerBackendExecParams& params) {
- per_backend_exec_params_ = params;
+ DCHECK_GT(query_schedule_pb_->coord_backend_mem_to_admit(), 0);
+ return query_schedule_pb_->coord_backend_mem_to_admit();
}
void set_largest_min_reservation(const int64_t largest_min_reservation) {
@@ -320,12 +293,16 @@ class QuerySchedule {
private:
/// These references are valid for the lifetime of this query schedule because they
/// are all owned by the enclosing QueryExecState.
- const TUniqueId& query_id_;
+ const UniqueIdPB& query_id_;
const TQueryExecRequest& request_;
/// The query options from the TClientRequest
const TQueryOptions& query_options_;
+ /// Contains the results of scheduling that will be sent back to the coordinator.
+ /// Ownership is transferred to the coordinator after scheduling has completed.
+ std::unique_ptr<QuerySchedulePB> query_schedule_pb_;
+
/// TODO: move these into QueryState
RuntimeProfile* summary_profile_;
RuntimeProfile::EventSequence* query_events_;
@@ -336,46 +313,21 @@ class QuerySchedule {
/// Maps from plan node id to its index in plan.nodes. Filled in c'tor.
std::vector<int32_t> plan_node_to_plan_node_idx_;
- // populated in Init() and Scheduler::Schedule()
- // (Scheduler::ComputeFInstanceExecParams()), indexed by fragment idx
- // (TPlanFragment.idx)
+ /// Populated in Init(), then calculated in Scheduler::ComputeFragmentExecParams().
+ /// Indexed by fragment idx (TPlanFragment.idx).
std::vector<FragmentExecParams> fragment_exec_params_;
- // Map of host address to list of assigned FInstanceExecParams*, which
- // reference fragment_exec_params_. Computed in Scheduler::Schedule().
+ /// Map from backend address to pointers to the corresponding BackendExecParamsPB in
+ /// 'query_schedule_pb_'. Created in GetOrCreateBackendExecParams().
PerBackendExecParams per_backend_exec_params_;
- /// Total number of scan ranges of this query.
- int64_t num_scan_ranges_;
-
/// Used to generate consecutive fragment instance ids.
- TUniqueId next_instance_id_;
+ UniqueIdPB next_instance_id_;
/// The largest min memory reservation across all executors. Set in
/// Scheduler::Schedule().
int64_t largest_min_reservation_ = 0;
- /// The memory limit per executor that will be imposed on the query.
- /// Set by the admission controller with a value that is only valid if it was admitted
- /// successfully. -1 means no limit.
- int64_t per_backend_mem_limit_ = -1;
-
- /// The per executor memory used for admission accounting.
- /// Set by the admission controller with a value that is only valid if it was admitted
- /// successfully. Can be zero if the query is only scheduled to run on the coordinator.
- int64_t per_backend_mem_to_admit_ = -1;
-
- /// The memory limit for the coordinator that will be imposed on the query. Used only if
- /// the query has a coordinator fragment.
- /// Set by the admission controller with a value that is only valid if it was admitted
- /// successfully. -1 means no limit.
- int64_t coord_backend_mem_limit_ = -1;
-
- /// The coordinator memory used for admission accounting.
- /// Set by the admission controller with a value that is only valid if it was admitted
- /// successfully.
- int64_t coord_backend_mem_to_admit_ = -1;
-
/// The coordinator's backend memory reservation. Set in Scheduler::Schedule().
int64_t coord_min_reservation_ = 0;
@@ -383,7 +335,10 @@ class QuerySchedule {
/// Scheduler and only valid after scheduling completes successfully.
std::string executor_group_;
- /// Populate fragment_exec_params_ from request_.plan_exec_info.
+ /// Map from fragment idx to references into the 'request_'.
+ std::unordered_map<int32_t, const TPlanFragment&> fragments_;
+
+ /// Populate fragments_ and fragment_exec_params_ from request_.plan_exec_info.
/// Sets is_coord_fragment and exchange_input_fragments.
/// Also populates plan_node_to_fragment_idx_ and plan_node_to_plan_node_idx_.
void Init();
@@ -393,6 +348,5 @@ class QuerySchedule {
return request_.stmt_type == TStmtType::QUERY;
}
};
-}
-#endif
+} // namespace impala
diff --git a/be/src/scheduling/scheduler-test-util.cc b/be/src/scheduling/scheduler-test-util.cc
index f8a9e50..2974081 100644
--- a/be/src/scheduling/scheduler-test-util.cc
+++ b/be/src/scheduling/scheduler-test-util.cc
@@ -503,8 +503,9 @@ int64_t Result::MinNumAssignedBytesPerHost() const {
int Result::NumDistinctBackends() const {
unordered_set<IpAddr> backends;
- AssignmentCallback cb = [&backends](
- const AssignmentInfo& assignment) { backends.insert(assignment.addr.hostname); };
+ AssignmentCallback cb = [&backends](const AssignmentInfo& assignment) {
+ backends.insert(assignment.addr.hostname());
+ };
ProcessAssignments(cb);
return backends.size();
}
@@ -526,8 +527,9 @@ Result::AssignmentFilter Result::Any() const {
Result::AssignmentFilter Result::IsHost(int host_idx) const {
TNetworkAddress expected_addr;
plan_.cluster().GetBackendAddress(host_idx, &expected_addr);
- return [expected_addr](
- const AssignmentInfo& assignment) { return assignment.addr == expected_addr; };
+ return [expected_addr](const AssignmentInfo& assignment) {
+ return assignment.addr == FromTNetworkAddress(expected_addr);
+ };
}
Result::AssignmentFilter Result::IsCached(AssignmentFilter filter) const {
@@ -551,7 +553,7 @@ Result::AssignmentFilter Result::IsRemote(AssignmentFilter filter) const {
void Result::ProcessAssignments(const AssignmentCallback& cb) const {
for (const FragmentScanRangeAssignment& assignment : assignments_) {
for (const auto& assignment_elem : assignment) {
- const TNetworkAddress& addr = assignment_elem.first;
+ const NetworkAddressPB& addr = assignment_elem.first;
const PerNodeScanRanges& per_node_ranges = assignment_elem.second;
for (const auto& per_node_ranges_elem : per_node_ranges) {
const vector<ScanRangeParamsPB> scan_range_params_vector =
@@ -592,8 +594,8 @@ int64_t Result::CountAssignedBytesIf(const AssignmentFilter& filter) const {
void Result::CountAssignmentsPerBackend(
NumAssignmentsPerBackend* num_assignments_per_backend) const {
AssignmentCallback cb = [&num_assignments_per_backend](
- const AssignmentInfo& assignment) {
- ++(*num_assignments_per_backend)[assignment.addr.hostname];
+ const AssignmentInfo& assignment) {
+ ++(*num_assignments_per_backend)[assignment.addr.hostname()];
};
ProcessAssignments(cb);
}
@@ -601,8 +603,8 @@ void Result::CountAssignmentsPerBackend(
void Result::CountAssignedBytesPerBackend(
NumAssignedBytesPerBackend* num_assignments_per_backend) const {
AssignmentCallback cb = [&num_assignments_per_backend](
- const AssignmentInfo& assignment) {
- (*num_assignments_per_backend)[assignment.addr.hostname] +=
+ const AssignmentInfo& assignment) {
+ (*num_assignments_per_backend)[assignment.addr.hostname()] +=
assignment.hdfs_file_split.length();
};
ProcessAssignments(cb);
diff --git a/be/src/scheduling/scheduler-test-util.h b/be/src/scheduling/scheduler-test-util.h
index 0e998a0..72f906c 100644
--- a/be/src/scheduling/scheduler-test-util.h
+++ b/be/src/scheduling/scheduler-test-util.h
@@ -364,7 +364,7 @@ class Result {
/// Parameter type for callbacks, which are used to filter scheduling results.
struct AssignmentInfo {
- const TNetworkAddress& addr;
+ const NetworkAddressPB& addr;
const HdfsFileSplitPB& hdfs_file_split;
bool is_cached;
bool is_remote;
diff --git a/be/src/scheduling/scheduler.cc b/be/src/scheduling/scheduler.cc
index b93b16c..f0c1d6f 100644
--- a/be/src/scheduling/scheduler.cc
+++ b/be/src/scheduling/scheduler.cc
@@ -23,7 +23,6 @@
#include <random>
#include <unordered_map>
#include <vector>
-#include <boost/algorithm/string/join.hpp>
#include <boost/unordered_set.hpp>
#include <gutil/strings/substitute.h>
@@ -50,7 +49,6 @@
#include "common/names.h"
-using boost::algorithm::join;
using std::pop_heap;
using std::push_heap;
using namespace apache::thrift;
@@ -89,25 +87,24 @@ Scheduler::Scheduler(MetricGroup* metrics, RequestPoolService* request_pool_serv
}
const BackendDescriptorPB& Scheduler::LookUpBackendDesc(
- const ExecutorConfig& executor_config, const TNetworkAddress& host) {
- const BackendDescriptorPB* desc =
- executor_config.group.LookUpBackendDesc(FromTNetworkAddress(host));
+ const ExecutorConfig& executor_config, const NetworkAddressPB& host) {
+ const BackendDescriptorPB* desc = executor_config.group.LookUpBackendDesc(host);
if (desc == nullptr) {
// Local host may not be in executor_config's executor group if it's a dedicated
// coordinator, or if it is configured to be in a different executor group.
const BackendDescriptorPB& local_be_desc = executor_config.local_be_desc;
- DCHECK(host == FromNetworkAddressPB(local_be_desc.address()));
+ DCHECK(host == local_be_desc.address());
desc = &local_be_desc;
}
return *desc;
}
-TNetworkAddress Scheduler::LookUpKrpcHost(
- const ExecutorConfig& executor_config, const TNetworkAddress& backend_host) {
+NetworkAddressPB Scheduler::LookUpKrpcHost(
+ const ExecutorConfig& executor_config, const NetworkAddressPB& backend_host) {
const BackendDescriptorPB& backend_descriptor =
LookUpBackendDesc(executor_config, backend_host);
DCHECK(backend_descriptor.has_krpc_address());
- TNetworkAddress krpc_host = FromNetworkAddressPB(backend_descriptor.krpc_address());
+ const NetworkAddressPB& krpc_host = backend_descriptor.krpc_address();
DCHECK(IsResolvedAddress(krpc_host));
return krpc_host;
}
@@ -227,17 +224,16 @@ void Scheduler::ComputeFragmentExecParams(
FragmentExecParams* src_params = schedule->GetFragmentExecParams(src_fragment.idx);
// populate src_params->destinations
- src_params->destinations.resize(dest_params->instance_exec_params.size());
for (int i = 0; i < dest_params->instance_exec_params.size(); ++i) {
- PlanFragmentDestinationPB& dest = src_params->destinations[i];
- TUniqueIdToUniqueIdPB(dest_params->instance_exec_params[i].instance_id,
- dest.mutable_fragment_instance_id());
- const TNetworkAddress& host = dest_params->instance_exec_params[i].host;
- *dest.mutable_thrift_backend() = FromTNetworkAddress(host);
+ PlanFragmentDestinationPB* dest = src_params->pb->add_destinations();
+ *dest->mutable_fragment_instance_id() =
+ dest_params->instance_exec_params[i].pb.instance_id();
+ const NetworkAddressPB& host = dest_params->instance_exec_params[i].host;
+ *dest->mutable_thrift_backend() = host;
const BackendDescriptorPB& desc = LookUpBackendDesc(executor_config, host);
DCHECK(desc.has_krpc_address());
DCHECK(IsResolvedAddress(desc.krpc_address()));
- *dest.mutable_krpc_backend() = desc.krpc_address();
+ *dest->mutable_krpc_backend() = desc.krpc_address();
}
// enumerate senders consecutively;
@@ -248,12 +244,13 @@ void Scheduler::ComputeFragmentExecParams(
|| sink.output_partition.type == TPartitionType::RANDOM
|| sink.output_partition.type == TPartitionType::KUDU);
PlanNodeId exch_id = sink.dest_node_id;
- int sender_id_base = dest_params->per_exch_num_senders[exch_id];
- dest_params->per_exch_num_senders[exch_id] +=
- src_params->instance_exec_params.size();
+ google::protobuf::Map<int32_t, int32_t>* per_exch_num_senders =
+ dest_params->pb->mutable_per_exch_num_senders();
+ int sender_id_base = (*per_exch_num_senders)[exch_id];
+ (*per_exch_num_senders)[exch_id] += src_params->instance_exec_params.size();
for (int i = 0; i < src_params->instance_exec_params.size(); ++i) {
FInstanceExecParams& src_instance_params = src_params->instance_exec_params[i];
- src_instance_params.sender_id = sender_id_base + i;
+ src_instance_params.pb.set_sender_id(sender_id_base + i);
}
}
}
@@ -286,19 +283,21 @@ void Scheduler::ComputeFragmentExecParams(const ExecutorConfig& executor_config,
const NetworkAddressPB& krpc_coord = local_be_desc.krpc_address();
DCHECK(IsResolvedAddress(krpc_coord));
// make sure that the coordinator instance ends up with instance idx 0
- TUniqueId instance_id = fragment_params->is_coord_fragment
- ? schedule->query_id()
- : schedule->GetNextInstanceId();
- fragment_params->instance_exec_params.emplace_back(instance_id,
- FromNetworkAddressPB(coord), FromNetworkAddressPB(krpc_coord), 0,
- *fragment_params);
+ UniqueIdPB instance_id = fragment_params->is_coord_fragment ?
+ schedule->query_id() :
+ schedule->GetNextInstanceId();
+ fragment_params->instance_exec_params.emplace_back(
+ instance_id, coord, krpc_coord, 0, *fragment_params);
FInstanceExecParams& instance_params = fragment_params->instance_exec_params.back();
+ *fragment_params->pb->add_instances() = instance_id;
// That instance gets all of the scan ranges, if there are any.
if (!fragment_params->scan_range_assignment.empty()) {
DCHECK_EQ(fragment_params->scan_range_assignment.size(), 1);
auto first_entry = fragment_params->scan_range_assignment.begin();
- instance_params.per_node_scan_ranges = first_entry->second;
+ for (const PerNodeScanRanges::value_type& entry : first_entry->second) {
+ instance_params.AddScanRanges(entry.first, entry.second);
+ }
}
} else if (ContainsNode(fragment.plan, TPlanNodeType::UNION_NODE)
|| ContainsScanNode(fragment.plan)) {
@@ -348,7 +347,7 @@ void Scheduler::CreateCollocatedAndScanInstances(const ExecutorConfig& executor_
// Build a map of hosts to the num instances this fragment should have, before we take
// into account scan ranges. If this fragment has input fragments, we always run with
// at least the same num instances as the input fragment.
- std::unordered_map<TNetworkAddress, int> instances_per_host;
+ std::unordered_map<NetworkAddressPB, int> instances_per_host;
// Add hosts of input fragments, counting the number of instances of the fragment.
// Only do this if there's a union - otherwise only consider the parallelism of
@@ -356,7 +355,7 @@ void Scheduler::CreateCollocatedAndScanInstances(const ExecutorConfig& executor_
// the parallelism of the scan.
if (has_union) {
for (FragmentIdx idx : fragment_params->exchange_input_fragments) {
- std::unordered_map<TNetworkAddress, int> input_fragment_instances_per_host;
+ std::unordered_map<NetworkAddressPB, int> input_fragment_instances_per_host;
const FragmentExecParams& input_params = *schedule->GetFragmentExecParams(idx);
for (const FInstanceExecParams& instance_params :
input_params.instance_exec_params) {
@@ -379,10 +378,10 @@ void Scheduler::CreateCollocatedAndScanInstances(const ExecutorConfig& executor_
vector<TPlanNodeId> scan_node_ids = FindScanNodes(fragment.plan);
DCHECK(has_union || scan_node_ids.size() == 1) << "This method may need revisiting "
<< "for plans with no union and multiple scans per fragment";
- vector<TNetworkAddress> scan_hosts;
+ vector<NetworkAddressPB> scan_hosts;
GetScanHosts(
executor_config.local_be_desc, scan_node_ids, *fragment_params, &scan_hosts);
- for (const TNetworkAddress& host_addr : scan_hosts) {
+ for (const NetworkAddressPB& host_addr : scan_hosts) {
// Ensure that the num instances is at least as many as input fragments. We don't
// want to increment if there were already some instances from the input fragment,
// since that could result in too high a num_instances.
@@ -397,8 +396,8 @@ void Scheduler::CreateCollocatedAndScanInstances(const ExecutorConfig& executor_
// Track the index of the next instance to be created for this fragment.
int per_fragment_instance_idx = 0;
for (const auto& entry : instances_per_host) {
- const TNetworkAddress& host = entry.first;
- TNetworkAddress krpc_host = LookUpKrpcHost(executor_config, host);
+ const NetworkAddressPB& host = entry.first;
+ NetworkAddressPB krpc_host = LookUpKrpcHost(executor_config, host);
FragmentScanRangeAssignment& sra = fragment_params->scan_range_assignment;
auto assignment_it = sra.find(host);
// One entry in outer vector per scan node in 'scan_node_ids'.
@@ -431,8 +430,10 @@ void Scheduler::CreateCollocatedAndScanInstances(const ExecutorConfig& executor_
// because we may have other input fragments.
int host_finstance_start_idx = fragment_params->instance_exec_params.size();
for (int i = 0; i < num_instances; ++i) {
- fragment_params->instance_exec_params.emplace_back(schedule->GetNextInstanceId(),
- host, krpc_host, per_fragment_instance_idx++, *fragment_params);
+ UniqueIdPB instance_id = schedule->GetNextInstanceId();
+ fragment_params->instance_exec_params.emplace_back(
+ instance_id, host, krpc_host, per_fragment_instance_idx++, *fragment_params);
+ *fragment_params->pb->add_instances() = instance_id;
}
// Allocate scan ranges to the finstances if needed. Currently we simply allocate
// them to fragment instances in order. This may be suboptimal in cases where the
@@ -443,8 +444,8 @@ void Scheduler::CreateCollocatedAndScanInstances(const ExecutorConfig& executor_
for (int inst_idx = 0; inst_idx < per_instance_ranges.size(); ++inst_idx) {
FInstanceExecParams& instance_params =
fragment_params->instance_exec_params[host_finstance_start_idx + inst_idx];
- instance_params.per_node_scan_ranges[scan_node_ids[scan_idx]] =
- move(per_instance_ranges[inst_idx]);
+ instance_params.AddScanRanges(
+ scan_node_ids[scan_idx], per_instance_ranges[inst_idx]);
}
}
}
@@ -476,9 +477,11 @@ void Scheduler::CreateInputCollocatedInstances(
int per_fragment_instance_idx = 0;
for (const FInstanceExecParams& input_instance_params :
input_fragment_params.instance_exec_params) {
- fragment_params->instance_exec_params.emplace_back(schedule->GetNextInstanceId(),
+ UniqueIdPB instance_id = schedule->GetNextInstanceId();
+ fragment_params->instance_exec_params.emplace_back(instance_id,
input_instance_params.host, input_instance_params.krpc_host,
per_fragment_instance_idx++, *fragment_params);
+ *fragment_params->pb->add_instances() = instance_id;
}
}
@@ -502,20 +505,21 @@ void Scheduler::CreateCollocatedJoinBuildInstances(
// instance was on the same host (instances for a backend are clustered together).
if (!share_build || instance_exec_params->empty()
|| instance_exec_params->back().krpc_host != parent_exec_params.krpc_host) {
- TUniqueId instance_id = schedule->GetNextInstanceId();
+ UniqueIdPB instance_id = schedule->GetNextInstanceId();
instance_exec_params->emplace_back(instance_id, parent_exec_params.host,
parent_exec_params.krpc_host, per_fragment_instance_idx++, *fragment_params);
- instance_exec_params->back().num_join_build_outputs = 0;
+ instance_exec_params->back().pb.set_num_join_build_outputs(0);
+ *fragment_params->pb->add_instances() = instance_id;
}
JoinBuildInputPB build_input;
build_input.set_join_node_id(sink.dest_node_id);
- TUniqueIdToUniqueIdPB(instance_exec_params->back().instance_id,
- build_input.mutable_input_finstance_id());
- parent_exec_params.join_build_inputs.emplace_back(build_input);
+ FInstanceExecParamsPB& pb = instance_exec_params->back().pb;
+ *build_input.mutable_input_finstance_id() = pb.instance_id();
+ *parent_exec_params.pb.add_join_build_inputs() = build_input;
VLOG(3) << "Linked join build for node id=" << sink.dest_node_id
- << " build finstance=" << PrintId(instance_exec_params->back().instance_id)
- << " dst finstance=" << PrintId(parent_exec_params.instance_id);
- ++instance_exec_params->back().num_join_build_outputs;
+ << " build finstance=" << PrintId(pb.instance_id())
+ << " dst finstance=" << PrintId(parent_exec_params.pb.instance_id());
+ pb.set_num_join_build_outputs(pb.num_join_build_outputs() + 1);
}
}
@@ -717,7 +721,7 @@ std::vector<TPlanNodeId> Scheduler::FindScanNodes(const TPlan& plan) {
void Scheduler::GetScanHosts(const BackendDescriptorPB& local_be_desc,
const vector<TPlanNodeId>& scan_ids, const FragmentExecParams& params,
- vector<TNetworkAddress>* scan_hosts) {
+ vector<NetworkAddressPB>* scan_hosts) {
for (const TPlanNodeId& scan_id : scan_ids) {
// Get the list of impalad host from scan_range_assignment_
for (const FragmentScanRangeAssignment::value_type& scan_range_assignment :
@@ -733,7 +737,7 @@ void Scheduler::GetScanHosts(const BackendDescriptorPB& local_be_desc,
// TODO: we'll need to revisit this strategy once we can partition joins
// (in which case this fragment might be executing a right outer join
// with a large build table)
- scan_hosts->push_back(FromNetworkAddressPB(local_be_desc.address()));
+ scan_hosts->push_back(local_be_desc.address());
}
}
}
@@ -762,87 +766,96 @@ bool Scheduler::IsCoordinatorOnlyQuery(const TQueryExecRequest& exec_request) {
void Scheduler::ComputeBackendExecParams(
const ExecutorConfig& executor_config, QuerySchedule* schedule) {
- PerBackendExecParams per_backend_params;
- for (const FragmentExecParams& f : schedule->fragment_exec_params()) {
- const TNetworkAddress* prev_host = nullptr;
- for (const FInstanceExecParams& i : f.instance_exec_params) {
- BackendExecParams& be_params = per_backend_params[i.host];
- be_params.instance_params.push_back(&i);
+ for (FragmentExecParams& f : schedule->fragment_exec_params()) {
+ const NetworkAddressPB* prev_host = nullptr;
+ int num_hosts = 0;
+ for (FInstanceExecParams& i : f.instance_exec_params) {
+ BackendExecParams& be_params = schedule->GetOrCreateBackendExecParams(i.host);
+ be_params.pb->add_instance_params()->Swap(&i.pb);
// Different fragments do not synchronize their Open() and Close(), so the backend
// does not provide strong guarantees about whether one fragment instance releases
// resources before another acquires them. Conservatively assume that all fragment
// instances on this backend can consume their peak resources at the same time,
// i.e. that this backend's peak resources is the sum of the per-fragment-instance
// peak resources for the instances executing on this backend.
- be_params.min_mem_reservation_bytes +=
- f.fragment.instance_min_mem_reservation_bytes;
- be_params.initial_mem_reservation_total_claims +=
- f.fragment.instance_initial_mem_reservation_total_claims;
- be_params.thread_reservation += f.fragment.thread_reservation;
+ be_params.pb->set_min_mem_reservation_bytes(
+ be_params.pb->min_mem_reservation_bytes()
+ + f.fragment.instance_min_mem_reservation_bytes);
+ be_params.pb->set_initial_mem_reservation_total_claims(
+ be_params.pb->initial_mem_reservation_total_claims()
+ + f.fragment.instance_initial_mem_reservation_total_claims);
+ be_params.pb->set_thread_reservation(
+ be_params.pb->thread_reservation() + f.fragment.thread_reservation);
// Some memory is shared between fragments on a host. Only add it for the first
// instance of this fragment on the host.
if (prev_host == nullptr || *prev_host != i.host) {
- be_params.min_mem_reservation_bytes +=
- f.fragment.backend_min_mem_reservation_bytes;
- be_params.initial_mem_reservation_total_claims +=
- f.fragment.backend_initial_mem_reservation_total_claims;
+ be_params.pb->set_min_mem_reservation_bytes(
+ be_params.pb->min_mem_reservation_bytes()
+ + f.fragment.backend_min_mem_reservation_bytes);
+ be_params.pb->set_initial_mem_reservation_total_claims(
+ be_params.pb->initial_mem_reservation_total_claims()
+ + f.fragment.backend_initial_mem_reservation_total_claims);
prev_host = &i.host;
+ ++num_hosts;
}
}
+ f.pb->set_num_hosts(num_hosts);
}
// Compute 'slots_to_use' for each backend based on the max # of instances of
// any fragment on that backend.
- for (auto& backend : per_backend_params) {
+ for (auto& backend : schedule->per_backend_exec_params()) {
int be_max_instances = 0;
// Instances for a fragment are clustered together because of how the vector is
// constructed above. So we can compute the max # of instances of any fragment
// with a single pass over the vector.
- const FragmentExecParams* curr_fragment = nullptr;
+ int curr_fragment_idx = -1;
int curr_instance_count = 0; // Number of instances of the current fragment seen.
- for (auto& finstance : backend.second.instance_params) {
- if (curr_fragment == nullptr ||
- curr_fragment != &finstance->fragment_exec_params) {
- curr_fragment = &finstance->fragment_exec_params;
+ for (auto& finstance : backend.second.pb->instance_params()) {
+ if (curr_fragment_idx == -1 || curr_fragment_idx != finstance.fragment_idx()) {
+ curr_fragment_idx = finstance.fragment_idx();
curr_instance_count = 0;
}
++curr_instance_count;
be_max_instances = max(be_max_instances, curr_instance_count);
}
- backend.second.slots_to_use = be_max_instances;
+ backend.second.pb->set_slots_to_use(be_max_instances);
}
// This also ensures an entry always exists for the coordinator backend.
int64_t coord_min_reservation = 0;
const NetworkAddressPB& coord_addr = executor_config.local_be_desc.address();
- BackendExecParams& coord_be_params =
- per_backend_params[FromNetworkAddressPB(coord_addr)];
- coord_be_params.is_coord_backend = true;
- coord_min_reservation = coord_be_params.min_mem_reservation_bytes;
+ BackendExecParams& coord_be_params = schedule->GetOrCreateBackendExecParams(coord_addr);
+ coord_be_params.pb->set_is_coord_backend(true);
+ coord_min_reservation = coord_be_params.pb->min_mem_reservation_bytes();
int64_t largest_min_reservation = 0;
- for (auto& backend : per_backend_params) {
- const TNetworkAddress& host = backend.first;
+ for (auto& backend : schedule->per_backend_exec_params()) {
+ const NetworkAddressPB& host = backend.first;
const BackendDescriptorPB be_desc = LookUpBackendDesc(executor_config, host);
backend.second.be_desc = be_desc;
- if (!backend.second.is_coord_backend) {
+ *backend.second.pb->mutable_backend_id() = be_desc.backend_id();
+ *backend.second.pb->mutable_address() = be_desc.address();
+ *backend.second.pb->mutable_krpc_address() = be_desc.krpc_address();
+ if (!backend.second.pb->is_coord_backend()) {
largest_min_reservation =
- max(largest_min_reservation, backend.second.min_mem_reservation_bytes);
+ max(largest_min_reservation, backend.second.pb->min_mem_reservation_bytes());
}
}
- schedule->set_per_backend_exec_params(per_backend_params);
schedule->set_largest_min_reservation(largest_min_reservation);
schedule->set_coord_min_reservation(coord_min_reservation);
stringstream min_mem_reservation_ss;
stringstream num_fragment_instances_ss;
- for (const auto& e: per_backend_params) {
- min_mem_reservation_ss << TNetworkAddressToString(e.first) << "("
- << PrettyPrinter::Print(e.second.min_mem_reservation_bytes, TUnit::BYTES)
- << ") ";
- num_fragment_instances_ss << TNetworkAddressToString(e.first) << "("
- << PrettyPrinter::Print(e.second.instance_params.size(), TUnit::UNIT)
- << ") ";
+ for (const auto& e : schedule->per_backend_exec_params()) {
+ min_mem_reservation_ss << NetworkAddressPBToString(e.first) << "("
+ << PrettyPrinter::Print(
+ e.second.pb->min_mem_reservation_bytes(), TUnit::BYTES)
+ << ") ";
+ num_fragment_instances_ss << NetworkAddressPBToString(e.first) << "("
+ << PrettyPrinter::Print(
+ e.second.pb->instance_params_size(), TUnit::UNIT)
+ << ") ";
}
schedule->summary_profile()->AddInfoString("Per Host Min Memory Reservation",
min_mem_reservation_ss.str());
@@ -1098,8 +1111,8 @@ void Scheduler::AssignmentCtx::RecordScanRangeAssignment(
if (!remote_read) total_local_assignments_->Increment(1);
}
- PerNodeScanRanges* scan_ranges = FindOrInsert(
- assignment, FromNetworkAddressPB(executor.address()), PerNodeScanRanges());
+ PerNodeScanRanges* scan_ranges =
+ FindOrInsert(assignment, executor.address(), PerNodeScanRanges());
vector<ScanRangeParamsPB>* scan_range_params_list =
FindOrInsert(scan_ranges, node_id, vector<ScanRangeParamsPB>());
// Add scan range.
@@ -1127,7 +1140,7 @@ void Scheduler::AssignmentCtx::PrintAssignment(
<< PrettyPrinter::Print(assignment_byte_counters_.cached_bytes, TUnit::BYTES);
for (const FragmentScanRangeAssignment::value_type& entry : assignment) {
- VLOG_FILE << "ScanRangeAssignment: server=" << ThriftDebugString(entry.first);
+ VLOG_FILE << "ScanRangeAssignment: server=" << entry.first.DebugString();
for (const PerNodeScanRanges::value_type& per_node_scan_ranges : entry.second) {
stringstream str;
for (const ScanRangeParamsPB& params : per_node_scan_ranges.second) {
diff --git a/be/src/scheduling/scheduler.h b/be/src/scheduling/scheduler.h
index 0844203..8126de2 100644
--- a/be/src/scheduling/scheduler.h
+++ b/be/src/scheduling/scheduler.h
@@ -266,12 +266,12 @@ class Scheduler {
/// backend or the local host itself. The returned descriptor should not be retained
/// beyond the lifetime of 'executor_config'.
const BackendDescriptorPB& LookUpBackendDesc(
- const ExecutorConfig& executor_config, const TNetworkAddress& host);
+ const ExecutorConfig& executor_config, const NetworkAddressPB& host);
/// Returns the KRPC host in 'executor_config' based on the thrift backend address
/// 'backend_host'. Will DCHECK if the KRPC address is not valid.
- TNetworkAddress LookUpKrpcHost(
- const ExecutorConfig& executor_config, const TNetworkAddress& backend_host);
+ NetworkAddressPB LookUpKrpcHost(
+ const ExecutorConfig& executor_config, const NetworkAddressPB& backend_host);
/// Determine the pool for a user and query options via request_pool_service_.
Status GetRequestPool(const std::string& user, const TQueryOptions& query_options,
@@ -411,7 +411,7 @@ class Scheduler {
/// 'scan_hosts'.
void GetScanHosts(const BackendDescriptorPB& local_be_desc,
const std::vector<TPlanNodeId>& scan_ids, const FragmentExecParams& params,
- std::vector<TNetworkAddress>* scan_hosts);
+ std::vector<NetworkAddressPB>* scan_hosts);
/// Return true if 'plan' contains a node of the given type.
bool ContainsNode(const TPlan& plan, TPlanNodeType::type type);
diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index c919123..c9b68d4 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -36,9 +36,9 @@
#include "runtime/coordinator.h"
#include "runtime/exec-env.h"
#include "runtime/mem-tracker.h"
+#include "runtime/query-driver.h"
#include "runtime/row-batch.h"
#include "runtime/runtime-state.h"
-#include "runtime/query-driver.h"
#include "scheduling/admission-controller.h"
#include "scheduling/scheduler.h"
#include "service/frontend.h"
@@ -55,6 +55,7 @@
#include "util/redactor.h"
#include "util/runtime-profile-counters.h"
#include "util/time.h"
+#include "util/uid-util.h"
#include "gen-cpp/CatalogService.h"
#include "gen-cpp/CatalogService_types.h"
@@ -528,9 +529,11 @@ void ClientRequestState::FinishExecQueryOrDmlRequest() {
DCHECK(ExecEnv::GetInstance()->admission_controller() != nullptr);
DCHECK(exec_request_->__isset.query_exec_request);
+ UniqueIdPB query_id_pb;
+ TUniqueIdToUniqueIdPB(query_id(), &query_id_pb);
Status admit_status =
ExecEnv::GetInstance()->admission_controller()->SubmitForAdmission(
- {query_id(), exec_request_->query_exec_request, exec_request_->query_options,
+ {query_id_pb, exec_request_->query_exec_request, exec_request_->query_options,
summary_profile_, query_events_},
&admit_outcome_, &schedule_);
{
@@ -538,21 +541,19 @@ void ClientRequestState::FinishExecQueryOrDmlRequest() {
if (!UpdateQueryStatus(admit_status).ok()) return;
}
DCHECK(schedule_.get() != nullptr);
- DCHECK_EQ(schedule_->query_id(), query_id());
// Note that we don't need to check for cancellation between admission and query
// startup. The query was not cancelled right before being admitted and the window here
// is small enough to not require special handling. Instead we start the query and then
// cancel it through the check below if necessary.
- DebugActionNoFail(schedule_->query_options(), "CRS_BEFORE_COORD_STARTS");
+ DebugActionNoFail(exec_request_->query_options, "CRS_BEFORE_COORD_STARTS");
// Register the query with the server to support cancellation. This happens after
// admission because now the set of executors is fixed and an executor failure will
// cause a query failure.
- parent_server_->RegisterQueryLocations(
- schedule_->per_backend_exec_params(), query_id());
- coord_.reset(new Coordinator(this, *schedule_, query_events_));
+ parent_server_->RegisterQueryLocations(schedule_->backend_exec_params(), query_id());
+ coord_.reset(new Coordinator(this, *exec_request_, *schedule_.get(), query_events_));
Status exec_status = coord_->Exec();
- DebugActionNoFail(schedule_->query_options(), "CRS_AFTER_COORD_STARTS");
+ DebugActionNoFail(exec_request_->query_options, "CRS_AFTER_COORD_STARTS");
// Make coordinator profile visible, even upon failure.
if (coord_->query_profile() != nullptr) profile_->AddChild(coord_->query_profile());
diff --git a/be/src/service/client-request-state.h b/be/src/service/client-request-state.h
index a69ed1e..424cce2 100644
--- a/be/src/service/client-request-state.h
+++ b/be/src/service/client-request-state.h
@@ -21,7 +21,6 @@
#include "common/object-pool.h"
#include "common/status.h"
#include "exec/catalog-op-executor.h"
-#include "scheduling/query-schedule.h"
#include "service/child-query.h"
#include "service/impala-server.h"
#include "service/query-result-set.h"
@@ -46,6 +45,7 @@ class Thread;
class TRuntimeProfileTree;
class TupleRow;
enum class AdmissionOutcome;
+class QuerySchedulePB;
/// Execution state of the client-facing side of a query. This captures everything
/// necessary to convert row batches received by the coordinator into results
@@ -233,7 +233,7 @@ class ClientRequestState {
const TUniqueId& session_id() const { return query_ctx_.session.session_id; }
const std::string& default_db() const { return query_ctx_.session.database; }
bool eos() const { return eos_.Load(); }
- const QuerySchedule* schedule() const { return schedule_.get(); }
+ const QuerySchedulePB* schedule() const { return schedule_.get(); }
/// Returns the Coordinator for 'QUERY' and 'DML' requests once Coordinator::Exec()
/// completes successfully. Otherwise returns null.
@@ -454,8 +454,8 @@ class ClientRequestState {
/// Session that this query is from
std::shared_ptr<ImpalaServer::SessionState> session_;
- /// Resource assignment determined by scheduler. Owned by obj_pool_.
- std::unique_ptr<QuerySchedule> schedule_;
+ /// Resource assignment determined by scheduler.
+ std::unique_ptr<QuerySchedulePB> schedule_;
/// Thread for asynchronously running the admission control code-path and starting
/// execution in the following cases:
@@ -633,7 +633,7 @@ class ClientRequestState {
Status ExecAsyncQueryOrDmlRequest(const TQueryExecRequest& query_exec_request)
WARN_UNUSED_RESULT;
- /// Submits the QuerySchedule to the admission controller and on successful admission,
+ /// Submits the exec request to the admission controller and on successful admission,
/// starts up the coordinator execution, makes it accessible by setting
/// 'coord_exec_called_' to true and advances operation_state_ to RUNNING. Handles
/// async cancellation of queries and cleans up state if needed.
diff --git a/be/src/service/impala-http-handler.cc b/be/src/service/impala-http-handler.cc
index 5144345..1f2338b 100644
--- a/be/src/service/impala-http-handler.cc
+++ b/be/src/service/impala-http-handler.cc
@@ -1076,7 +1076,7 @@ void ImpalaHttpHandler::AdmissionStateHandler(
request_state->schedule()->coord_backend_mem_limit(),
request_state->schedule()->coord_backend_mem_to_admit(),
static_cast<unsigned long>(
- request_state->schedule()->per_backend_exec_params().size())});
+ request_state->schedule()->backend_exec_params().size())});
});
// Add the running queries to the resource_pools json.
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index f34bc41..3ab48cf 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -116,6 +116,7 @@ using boost::get_system_time;
using boost::system_time;
using boost::uuids::random_generator;
using boost::uuids::uuid;
+using google::protobuf::RepeatedPtrField;
using kudu::GetRandomSeed32;
using kudu::rpc::RpcContext;
using namespace apache::hive::service::cli::thrift;
@@ -1245,16 +1246,16 @@ void ImpalaServer::CloseClientRequestState(const QueryHandle& query_handle) {
}
if (query_handle->schedule() != nullptr) {
- const PerBackendExecParams& per_backend_params =
- query_handle->schedule()->per_backend_exec_params();
- if (!per_backend_params.empty()) {
+ const RepeatedPtrField<BackendExecParamsPB>& backend_exec_params =
+ query_handle->schedule()->backend_exec_params();
+ if (!backend_exec_params.empty()) {
lock_guard<mutex> l(query_locations_lock_);
- for (const auto& entry : per_backend_params) {
+ for (const BackendExecParamsPB& param : backend_exec_params) {
// Query may have been removed already by cancellation path. In particular, if
// node to fail was last sender to an exchange, the coordinator will realise and
// fail the query at the same time the failure detection path does the same
// thing. They will harmlessly race to remove the query from this map.
- auto it = query_locations_.find(entry.second.be_desc.backend_id());
+ auto it = query_locations_.find(param.backend_id());
if (it != query_locations_.end()) {
it->second.query_ids.erase(query_handle->query_id());
}
@@ -1893,16 +1894,17 @@ Status ImpalaServer::ProcessCatalogUpdateResult(
}
void ImpalaServer::RegisterQueryLocations(
- const PerBackendExecParams& per_backend_params, const TUniqueId& query_id) {
+ const RepeatedPtrField<BackendExecParamsPB>& backend_params,
+ const TUniqueId& query_id) {
VLOG_QUERY << "Registering query locations";
- if (!per_backend_params.empty()) {
+ if (!backend_params.empty()) {
lock_guard<mutex> l(query_locations_lock_);
- for (const auto& entry : per_backend_params) {
- const BackendIdPB& backend_id = entry.second.be_desc.backend_id();
+ for (const BackendExecParamsPB& param : backend_params) {
+ const BackendIdPB& backend_id = param.backend_id();
auto it = query_locations_.find(backend_id);
if (it == query_locations_.end()) {
query_locations_.emplace(
- backend_id, QueryLocationInfo(entry.second.be_desc.address(), query_id));
+ backend_id, QueryLocationInfo(param.address(), query_id));
} else {
it->second.query_ids.insert(query_id);
}
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index 1945534..f0236fb 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -37,7 +37,6 @@
#include "kudu/util/random.h"
#include "rpc/thrift-server.h"
#include "runtime/types.h"
-#include "scheduling/query-schedule.h"
#include "service/query-driver-map.h"
#include "service/query-options.h"
#include "statestore/statestore-subscriber.h"
@@ -57,6 +56,8 @@ class RpcContext;
namespace impala {
using kudu::ThreadSafeRandom;
+class BackendDescriptorPB;
+class BackendExecParamsPB;
class ExecEnv;
class DataSink;
class CancellationWork;
@@ -73,7 +74,6 @@ class TGetExecSummaryReq;
class ClientRequestState;
class QueryDriver;
struct QueryHandle;
-class QuerySchedule;
class SimpleLogger;
class UpdateFilterParamsPB;
class UpdateFilterResultPB;
@@ -465,7 +465,8 @@ class ImpalaServer : public ImpalaServiceIf,
/// to run there (query_locations_). After calling this function, the server will cancel
/// a query with an error if one of its backends fail.
void RegisterQueryLocations(
- const PerBackendExecParams& per_backend_params, const TUniqueId& query_id);
+ const google::protobuf::RepeatedPtrField<BackendExecParamsPB>& per_backend_params,
+ const TUniqueId& query_id);
/// Takes a set of backend ids of active backends and cancels all the queries running on
/// failed ones (that is, ids not in the active set).
diff --git a/be/src/util/debug-util.h b/be/src/util/debug-util.h
index a9e89c1..98c61e0 100644
--- a/be/src/util/debug-util.h
+++ b/be/src/util/debug-util.h
@@ -87,6 +87,10 @@ std::string PrintBatch(RowBatch* batch);
std::string PrintId(const TUniqueId& id, const std::string& separator = ":");
std::string PrintId(const UniqueIdPB& id, const std::string& separator = ":");
+inline ostream& operator<<(ostream& os, const UniqueIdPB& id) {
+ return os << PrintId(id);
+}
+
/// Returns the fully qualified path, e.g. "database.table.array_col.item.field"
std::string PrintPath(const TableDescriptor& tbl_desc, const SchemaPath& path);
/// Same as PrintPath(), but truncates the path after the given 'end_path_idx'.
diff --git a/be/src/util/uid-util.h b/be/src/util/uid-util.h
index 89b7c83..04c1ad7 100644
--- a/be/src/util/uid-util.h
+++ b/be/src/util/uid-util.h
@@ -102,6 +102,14 @@ inline TUniqueId CreateInstanceId(
return result;
}
+inline UniqueIdPB CreateInstanceId(const UniqueIdPB& query_id, int32_t instance_idx) {
+ DCHECK_EQ(GetInstanceIdx(query_id), 0); // well-formed query id
+ DCHECK_GE(instance_idx, 0);
+ UniqueIdPB result = query_id;
+ result.set_lo(result.lo() + instance_idx);
+ return result;
+}
+
template <typename F, typename T>
inline T CastTUniqueId(const F& from) {
T to;
diff --git a/common/protobuf/CMakeLists.txt b/common/protobuf/CMakeLists.txt
index 9357e8a..85b8ab9 100644
--- a/common/protobuf/CMakeLists.txt
+++ b/common/protobuf/CMakeLists.txt
@@ -38,7 +38,8 @@ foreach(pb_src common row_batch planner)
set(${_PROTO_SRCS} ${${_PROTO_SRCS}} PARENT_SCOPE)
endforeach()
-foreach(pb_src data_stream_service control_service rpc_test statestore_service)
+foreach(pb_src data_stream_service control_service rpc_test statestore_service
+ admission_control_service)
string(TOUPPER ${pb_src} _PB_SRC_UPPER)
set(_PROTO_SRCS ${_PB_SRC_UPPER}_PROTO_SRCS)
set(_PROTO_HDRS ${_PB_SRC_UPPER}_PROTO_HDRS)
diff --git a/common/protobuf/admission_control_service.proto b/common/protobuf/admission_control_service.proto
new file mode 100644
index 0000000..51d88f2
--- /dev/null
+++ b/common/protobuf/admission_control_service.proto
@@ -0,0 +1,154 @@
+// 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.
+//
+
+syntax = "proto2";
+
+package impala;
+
+import "common.proto";
+import "control_service.proto";
+import "statestore_service.proto";
+
+// Execution parameters for a single fragment instance. Used to assemble the
+// TPlanFragmentInstanceCtx/PlanFragmentInstanceCtxPB.
+message FInstanceExecParamsPB {
+ // The fragment instance id.
+ optional UniqueIdPB instance_id = 1;
+
+ // Ordinal number of the corresponding fragment in the query, i.e. TPlanFragment.idx.
+ optional int32 fragment_idx = 2;
+
+ // Map from plan node id to a list of scan ranges.
+ map<int32, ScanRangesPB> per_node_scan_ranges = 5;
+
+ // 0-based ordinal number of this particular instance. This is within its fragment, not
+ // query-wide, so eg. there will be one instance '0' for each fragment.
+ optional int32 per_fragment_instance_idx = 6;
+
+ // In its role as a data sender, a fragment instance is assigned a "sender id" to
+ // uniquely identify it to a receiver. -1 = invalid.
+ optional int32 sender_id = 7 [default = -1];
+
+ // List of input join build finstances for joins in this finstance.
+ repeated JoinBuildInputPB join_build_inputs = 8;
+
+ // If this is a join build fragment, the number of fragment instances that consume the
+ // join build. -1 = invalid.
+ optional int32 num_join_build_outputs = 9 [default = -1];
+}
+
+// Execution parameters for a single backend. Used to construct the
+// Coordinator::BackendStates.
+message BackendExecParamsPB {
+ // The id of this backend.
+ optional UniqueIdPB backend_id = 1;
+
+ // The hostname + port of the Thrift ImpalaInternalService on this backend.
+ optional NetworkAddressPB address = 8;
+
+ // The IP address + port of the KRPC ControlService on this backend.
+ optional NetworkAddressPB krpc_address = 9;
+
+ // The fragment instance params assigned to this backend. All instances of a
+ // particular fragment are contiguous in this list. This can be empty only for the
+ // coordinator backend, that is, if 'is_coord_backend' is true.
+ repeated FInstanceExecParamsPB instance_params = 2;
+
+ // The minimum query-wide buffer reservation size (in bytes) required for this backend.
+ // This is the peak minimum reservation that may be required by the
+ // concurrently-executing operators at any point in query execution. It may be less
+ // than the initial reservation total claims (below) if execution of some operators
+ // never overlaps, which allows reuse of reservations.
+ optional int64 min_mem_reservation_bytes = 3;
+
+ // Total of the initial buffer reservations that we expect to be claimed on this
+ // backend for all fragment instances in instance_params. I.e. the sum over all
+ // operators in all fragment instances that execute on this backend. This is used for
+ // an optimization in InitialReservation. Measured in bytes.
+ optional int64 initial_mem_reservation_total_claims = 4;
+
+ // Total thread reservation for fragment instances scheduled on this backend. This is
+ // the peak number of required threads that may be required by the
+ // concurrently-executing fragment instances at any point in query execution.
+ optional int64 thread_reservation = 5;
+
+ // Number of slots that this query should count for in admission control.
+ // This is calculated as the maximum # of instances of any fragment on this backend.
+ // I.e. 1 if mt_dop is not used and at most the mt_dop value if mt_dop is specified
+ // (but less if the query is not actually running with mt_dop instances on this node).
+ optional int32 slots_to_use = 6;
+
+ // Indicates whether this backend is the coordinator.
+ optional bool is_coord_backend = 7;
+}
+
+// Execution parameters shared between fragment instances
+message FragmentExecParamsPB {
+ // Ordinal number of the corresponding fragment in the query, i.e. TPlanFragment.idx.
+ optional int32 fragment_idx = 1;
+
+ // Output destinations of this fragment.
+ repeated PlanFragmentDestinationPB destinations = 2;
+
+ // map from node id to the number of senders (node id expected to be for an
+ // ExchangeNode)
+ map<int32, int32> per_exch_num_senders = 3;
+
+ // List of fragment instance ids for all instances of this fragment.
+ repeated UniqueIdPB instances = 4;
+
+ // Total number of backends this fragment is scheduled on. Note that this represents
+ // the number of individual impalads, not the number of physical hosts.
+ optional int32 num_hosts = 5;
+}
+
+// Contains the output from scheduling and admission control that is used by the
+// coordinator to start query execution.
+message QuerySchedulePB {
+ optional UniqueIdPB query_id = 1;
+
+ // The per-fragment execution parameters for this schedule.
+ repeated FragmentExecParamsPB fragment_exec_params = 2;
+
+ // The per-backend execution parameters for this schedule.
+ repeated BackendExecParamsPB backend_exec_params = 3;
+
+ // Total number of scan ranges of this query.
+ optional int64 num_scan_ranges = 4;
+
+ // The memory limit per executor that will be imposed on the query.
+ // Set by the admission controller with a value that is only valid if it was admitted
+ // successfully. -1 means no limit.
+ optional int64 per_backend_mem_limit = 5;
+
+ // The per executor memory used for admission accounting.
+ // Set by the admission controller with a value that is only valid if it was admitted
+ // successfully. Can be zero if the query is only scheduled to run on the coordinator.
+ optional int64 per_backend_mem_to_admit = 6;
+
+ // The memory limit for the coordinator that will be imposed on the query. Used only if
+ // the query has a coordinator fragment.
+ // Set by the admission controller with a value that is only valid if it was admitted
+ // successfully. -1 means no limit.
+ optional int64 coord_backend_mem_limit = 7;
+
+ // The coordinator memory used for admission accounting.
+ // Set by the admission controller with a value that is only valid if it was admitted
+ // successfully.
+ optional int64 coord_backend_mem_to_admit = 8;
+}
[impala] 01/03: IMPALA-9907: Fix NullPointerException in
ParallelFileMetadataLoader's load() method
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asherman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 36ced942d84e7237de28a0ebfae785139dbd3836
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Mon Jun 29 17:31:34 2020 +0200
IMPALA-9907: Fix NullPointerException in ParallelFileMetadataLoader's load() method
ParallelFileMetadataLoader.loaders_ became a Map, but at one place we
still treat it as a List:
loaders_.get(i).getPartDir()
'i' is an integer, while the loaders_'s key type is Path.
Unfortunately the code compiled which means we got a
NullPointerException for this statement. I changed the logic to avoid
the NPE.
Testing:
* Tested manually on an ACID table with deleted rows
* I didn't add an automated test because I'm about to extend the ACID
functionalities, hence getting rid of these exceptions
* The other case when we can hit this NPE is when we get an
IOException in the file metadata loader's load() method
Change-Id: I4322a18a1f07e4106789591b80d25105c565a2b6
Reviewed-on: http://gerrit.cloudera.org:8080/16121
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/catalog/ParallelFileMetadataLoader.java | 11 +++++++----
1 file changed, 7 insertions(+), 4 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
index 5381f2e..fa51cbf 100644
--- a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
@@ -35,6 +35,7 @@ import org.apache.impala.catalog.FeFsTable.Utils;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import org.apache.impala.common.FileSystemUtil;
+import org.apache.impala.common.Pair;
import org.apache.impala.compat.MetastoreShim;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TValidWriteIdList;
@@ -158,19 +159,21 @@ public class ParallelFileMetadataLoader {
int failedLoadTasks = 0;
ExecutorService pool = createPool();
try (ThreadNameAnnotator tna = new ThreadNameAnnotator(logPrefix_)) {
- List<Future<Void>> futures = new ArrayList<>(loaders_.size());
+ List<Pair<FileMetadataLoader, Future<Void>>> futures =
+ new ArrayList<>(loaders_.size());
for (FileMetadataLoader loader : loaders_.values()) {
- futures.add(pool.submit(() -> { loader.load(); return null; }));
+ futures.add(new Pair<FileMetadataLoader, Future<Void>>(
+ loader, pool.submit(() -> { loader.load(); return null; })));
}
// Wait for the loaders to finish.
for (int i = 0; i < futures.size(); i++) {
try {
- futures.get(i).get();
+ futures.get(i).second.get();
} catch (ExecutionException | InterruptedException e) {
if (++failedLoadTasks <= MAX_PATH_METADATA_LOADING_ERRORS_TO_LOG) {
LOG.error(logPrefix_ + " encountered an error loading data for path " +
- loaders_.get(i).getPartDir(), e);
+ futures.get(i).first.getPartDir(), e);
}
}
}