You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by cs...@apache.org on 2020/05/05 11:16:07 UTC
[impala] branch master updated (cf8e41c -> 56c20ac)
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from cf8e41c IMPALA-9365: Disable acid-negative.test on non-HDFS filesystems
new 748e41a IMPALA-9380: async query unregistration
new 56c20ac IMPALA-9699: Skip '-1' values when aggregating num_null statistics
The 2 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/common/atomic.h | 4 +
be/src/exec/incr-stats-util-test.cc | 34 ++++++-
be/src/exec/incr-stats-util.cc | 108 ++++++++--------------
be/src/exec/incr-stats-util.h | 52 +++++++++++
be/src/service/client-request-state-map.cc | 12 +--
be/src/service/client-request-state-map.h | 9 +-
be/src/service/client-request-state.cc | 17 +++-
be/src/service/client-request-state.h | 19 +++-
be/src/service/impala-beeswax-server.cc | 19 ++--
be/src/service/impala-hs2-server.cc | 33 ++++---
be/src/service/impala-http-handler.cc | 5 +
be/src/service/impala-server.cc | 84 +++++++++++------
be/src/service/impala-server.h | 65 ++++++++-----
be/src/util/runtime-profile.cc | 27 +++---
be/src/util/runtime-profile.h | 24 +++--
common/thrift/generate_error_codes.py | 4 +
tests/common/impala_service.py | 9 +-
tests/comparison/leopard/report.py | 33 +++----
tests/custom_cluster/test_admission_controller.py | 2 +-
tests/hs2/hs2_test_suite.py | 20 ++--
tests/hs2/test_hs2.py | 103 ++++++++++++++++++++-
tests/hs2/test_json_endpoints.py | 17 +++-
tests/shell/test_shell_commandline.py | 2 +-
tests/stress/concurrent_select.py | 2 +-
tests/util/cancel_util.py | 8 +-
25 files changed, 486 insertions(+), 226 deletions(-)
[impala] 01/02: IMPALA-9380: async query unregistration
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 748e41ab41a839be38d0865ff0e1bd3e9de8d157
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Sun Apr 26 22:59:45 2020 -0700
IMPALA-9380: async query unregistration
This change improves query latency by doing much
of the heavyweight work of unregistering a query
asynchronously, instead of synchronously on the
RPC thread. The biggest win is to move the
profile serialization off the RPC thread.
Unregistration processing is done by a thread pool
with 4 threads by default. This is configurable by
--unregistration_thread_pool_size and
--unregistration_thread_pool_queue_depth.
This fixes a pre-existing bug where a
query was temporarily neither in the in-flight
queries nor the completed queries. It would be
much easier to hit this with async unregistration
because there is less synchronisation on the client
side. Now the query is briefly in both maps, but
this is handled as follows:
* All places that look up both the maps will check
the in-flight map first, and return a reference to
the ClientRequestState, i.e. ignoring the entry in
the query log.
* The /queries page does not return completed queries
if they were found in the in-flight queries map, so
avoids duplicate results.
The thread safety story changes slightly.
Before this change, only one thread could
remove the query from the map and close it,
with only one thread "winning" the race to remove
the ClientRequestState from the map. Since we leave
the query in the map while being finalized, we
instead use an atomic in ClientRequestState to ensure
that only one thread does the finalization.
Some misc cleanup was done as a result of these changes:
* Fix a pre-existing TSAN race in RuntimeProfile that
was revealed by the new concurrent unregister test.
* Consolidate the various unknown query handle errors into
an error code so that we consistently return the same
string.
* "Unregister query" should include flushing audit events.
Testing:
* Add a test that unregisters a query concurrent with other
operations.
* Ran exhaustive tests
Perf:
Ran TPC-H 30 with mt_dop=4. No regressions and some improvements:
+----------+-----------------------+---------+------------+------------+----------------+
| Workload | File Format | Avg (s) | Delta(Avg) | GeoMean(s) | Delta(GeoMean) |
+----------+-----------------------+---------+------------+------------+----------------+
| TPCH(30) | parquet / none / none | 5.38 | -2.67% | 4.02 | -2.01% |
+----------+-----------------------+---------+------------+------------+----------------+
+----------+----------+-----------------------+--------+-------------+------------+-----------+----------------+-------+----------------+---------+--------+
| Workload | Query | File Format | Avg(s) | Base Avg(s) | Delta(Avg) | StdDev(%) | Base StdDev(%) | Iters | Median Diff(%) | MW Zval | Tval |
+----------+----------+-----------------------+--------+-------------+------------+-----------+----------------+-------+----------------+---------+--------+
| TPCH(30) | TPCH-Q1 | parquet / none / none | 5.36 | 5.17 | +3.61% | 1.82% | 1.17% | 5 | +3.73% | 1.73 | 3.65 |
| TPCH(30) | TPCH-Q6 | parquet / none / none | 1.77 | 1.74 | +1.48% | 2.00% | 2.50% | 5 | +2.89% | 0.87 | 1.03 |
| TPCH(30) | TPCH-Q12 | parquet / none / none | 3.02 | 3.00 | +0.79% | 2.18% | 2.21% | 5 | +1.55% | 0.00 | 0.57 |
| TPCH(30) | TPCH-Q16 | parquet / none / none | 1.65 | 1.64 | +0.81% | 1.35% | 0.03% | 5 | +0.07% | 1.15 | 1.34 |
| TPCH(30) | TPCH-Q2 | parquet / none / none | 1.21 | 1.21 | -0.07% | 2.11% | 2.14% | 5 | -0.04% | -0.29 | -0.05 |
| TPCH(30) | TPCH-Q4 | parquet / none / none | 2.50 | 2.52 | -0.49% | 2.43% | 3.34% | 5 | -0.09% | -0.29 | -0.27 |
| TPCH(30) | TPCH-Q20 | parquet / none / none | 2.86 | 2.90 | -1.28% | 2.30% | 1.24% | 5 | -0.02% | -0.58 | -1.11 |
| TPCH(30) | TPCH-Q3 | parquet / none / none | 4.35 | 4.40 | -1.15% | 1.76% | 1.78% | 5 | -1.12% | -0.87 | -1.03 |
| TPCH(30) | TPCH-Q19 | parquet / none / none | 4.10 | 4.17 | -1.80% | 1.05% | 1.31% | 5 | -1.25% | -1.73 | -2.40 |
| TPCH(30) | TPCH-Q14 | parquet / none / none | 3.20 | 3.25 | -1.52% | 0.79% | 2.56% | 5 | -1.56% | -0.58 | -1.26 |
| TPCH(30) | TPCH-Q18 | parquet / none / none | 10.81 | 11.07 | -2.34% | 5.00% | 7.01% | 5 | -1.40% | -0.58 | -0.61 |
| TPCH(30) | TPCH-Q7 | parquet / none / none | 11.19 | 11.56 | -3.18% | 3.47% | 6.02% | 5 | -0.90% | -0.87 | -1.03 |
| TPCH(30) | TPCH-Q21 | parquet / none / none | 19.91 | 20.32 | -2.02% | 0.66% | 0.47% | 5 | -2.18% | -2.31 | -5.64 |
| TPCH(30) | TPCH-Q17 | parquet / none / none | 5.63 | 5.77 | -2.40% | 1.71% | 2.01% | 5 | -1.84% | -1.73 | -2.05 |
| TPCH(30) | TPCH-Q5 | parquet / none / none | 3.91 | 4.03 | -2.74% | 1.08% | 1.86% | 5 | -2.45% | -1.44 | -2.88 |
| TPCH(30) | TPCH-Q8 | parquet / none / none | 4.55 | 4.71 | -3.48% | 1.90% | 3.53% | 5 | -2.35% | -1.44 | -1.96 |
| TPCH(30) | TPCH-Q22 | parquet / none / none | 1.93 | 2.01 | -3.96% | 0.05% | 4.05% | 5 | -2.59% | -2.31 | -2.19 |
| TPCH(30) | TPCH-Q10 | parquet / none / none | 4.52 | 4.73 | -4.26% | 1.26% | 2.43% | 5 | -3.40% | -2.02 | -3.51 |
| TPCH(30) | TPCH-Q11 | parquet / none / none | 1.02 | 1.05 | -3.58% | 3.94% | 2.36% | 5 | -4.56% | -1.44 | -1.79 |
| TPCH(30) | TPCH-Q13 | parquet / none / none | 9.52 | 10.04 | I -5.24% | 2.14% | 0.56% | 5 | I -4.67% | -2.31 | -5.57 |
| TPCH(30) | TPCH-Q15 | parquet / none / none | 3.49 | 3.68 | I -5.08% | 0.07% | 0.56% | 5 | I -5.66% | -2.31 | -20.08 |
| TPCH(30) | TPCH-Q9 | parquet / none / none | 11.92 | 12.71 | I -6.19% | 0.57% | 3.15% | 5 | I -4.99% | -2.31 | -4.33 |
+----------+----------+-----------------------+--------+-------------+------------+-----------+----------------+-------+----------------+---------+--------+
Change-Id: I80027b1baeb4ab453938c0f6357b120f4035ba08
Reviewed-on: http://gerrit.cloudera.org:8080/15821
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/common/atomic.h | 4 +
be/src/service/client-request-state-map.cc | 12 +--
be/src/service/client-request-state-map.h | 9 +-
be/src/service/client-request-state.cc | 17 +++-
be/src/service/client-request-state.h | 19 +++-
be/src/service/impala-beeswax-server.cc | 19 ++--
be/src/service/impala-hs2-server.cc | 33 ++++---
be/src/service/impala-http-handler.cc | 5 ++
be/src/service/impala-server.cc | 84 ++++++++++++------
be/src/service/impala-server.h | 65 +++++++++-----
be/src/util/runtime-profile.cc | 27 +++---
be/src/util/runtime-profile.h | 24 ++---
common/thrift/generate_error_codes.py | 4 +
tests/common/impala_service.py | 9 +-
tests/comparison/leopard/report.py | 33 +++----
tests/custom_cluster/test_admission_controller.py | 2 +-
tests/hs2/hs2_test_suite.py | 20 +++--
tests/hs2/test_hs2.py | 103 +++++++++++++++++++++-
tests/hs2/test_json_endpoints.py | 17 +++-
tests/shell/test_shell_commandline.py | 2 +-
tests/stress/concurrent_select.py | 2 +-
tests/util/cancel_util.py | 8 +-
22 files changed, 362 insertions(+), 156 deletions(-)
diff --git a/be/src/common/atomic.h b/be/src/common/atomic.h
index 5aa5f86..0871148 100644
--- a/be/src/common/atomic.h
+++ b/be/src/common/atomic.h
@@ -175,6 +175,10 @@ class AtomicBool {
/// Atomic store with "release" memory-ordering semantic.
ALWAYS_INLINE void Store(bool val) { boolean_.Store(val); }
+ ALWAYS_INLINE bool CompareAndSwap(bool old_val, bool new_val) {
+ return boolean_.CompareAndSwap(old_val, new_val);
+ }
+
private:
internal::AtomicInt<int32_t> boolean_;
};
diff --git a/be/src/service/client-request-state-map.cc b/be/src/service/client-request-state-map.cc
index aaf2d98..40e27a7 100644
--- a/be/src/service/client-request-state-map.cc
+++ b/be/src/service/client-request-state-map.cc
@@ -42,18 +42,14 @@ Status ClientRequestStateMap::AddClientRequestState(
return Status::OK();
}
-Status ClientRequestStateMap::DeleteClientRequestState(
- const TUniqueId& query_id, shared_ptr<ClientRequestState>* request_state) {
+Status ClientRequestStateMap::DeleteClientRequestState(const TUniqueId& query_id) {
ScopedShardedMapRef<shared_ptr<ClientRequestState>> map_ref(query_id, this);
DCHECK(map_ref.get() != nullptr);
auto entry = map_ref->find(query_id);
if (entry == map_ref->end()) {
- string error_msg =
- strings::Substitute("Invalid or unknown query handle $0", PrintId(query_id));
- VLOG(1) << error_msg;
- return Status::Expected(error_msg);
- } else if (request_state != nullptr) {
- *request_state = entry->second;
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ VLOG(1) << err.GetDetail();
+ return err;
}
map_ref->erase(entry);
return Status::OK();
diff --git a/be/src/service/client-request-state-map.h b/be/src/service/client-request-state-map.h
index e131b6c..2699202 100644
--- a/be/src/service/client-request-state-map.h
+++ b/be/src/service/client-request-state-map.h
@@ -38,11 +38,8 @@ class ClientRequestStateMap
Status AddClientRequestState(
const TUniqueId& query_id, std::shared_ptr<ClientRequestState> request_state);
- /// Deletes the specified (query_id, request_state) pair from the map and sets the given
- /// request_state pointer to the ClientRequestState associated with the given query_id.
- /// If request_state == nullptr, it is not set. Returns an error Status if the query_id
- /// cannot be found in the map.
- Status DeleteClientRequestState(const TUniqueId& query_id,
- std::shared_ptr<ClientRequestState>* request_state = nullptr);
+ /// Deletes the specified (query_id, request_state) pair from the map. Returns an error
+ /// Status if the query_id cannot be found in the map.
+ Status DeleteClientRequestState(const TUniqueId& query_id);
};
}
diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index 27179a0..cb630f0 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -159,6 +159,7 @@ ClientRequestState::ClientRequestState(
ClientRequestState::~ClientRequestState() {
DCHECK(wait_thread_.get() == NULL) << "BlockOnWait() needs to be called!";
+ DCHECK(started_finalize_.Load()) << "Finalize() must have been called";
}
Status ClientRequestState::SetResultCache(QueryResultSet* cache,
@@ -724,7 +725,14 @@ Status ClientRequestState::ExecShutdownRequest() {
return Status::OK();
}
-void ClientRequestState::Done() {
+Status ClientRequestState::Finalize(bool check_inflight, const Status* cause) {
+ if (!started_finalize_.CompareAndSwap(false, true)) {
+ // Return error as-if the query was already unregistered, so that it appears to the
+ // client as-if unregistration already happened. We don't need a distinct
+ // client-visible error for this case.
+ return Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id()));
+ }
+ RETURN_IF_ERROR(Cancel(check_inflight, cause));
MarkActive();
// Make sure we join on wait_thread_ before we finish (and especially before this object
// is destroyed).
@@ -756,7 +764,6 @@ void ClientRequestState::Done() {
{
unique_lock<mutex> l(lock_);
- query_events_->MarkEvent("Unregister query");
// Update result set cache metrics, and update mem limit accounting before tearing
// down the coordinator.
ClearResultCache();
@@ -771,6 +778,10 @@ void ClientRequestState::Done() {
LogQueryEvents();
}
DCHECK(wait_thread_.get() == nullptr);
+
+ // Update the timeline here so that all of the above work is captured in the timeline.
+ query_events_->MarkEvent("Unregister query");
+ return Status::OK();
}
Status ClientRequestState::Exec(const TMetadataOpRequest& exec_request) {
@@ -1483,7 +1494,7 @@ void ClientRequestState::ClearTransactionState() {
void ClientRequestState::LogQueryEvents() {
// Wait until the results are available. This guarantees the completion of non QUERY
- // statemens like DDL/DML etc. Query events are logged if the query reaches a FINISHED
+ // statements like DDL/DML etc. Query events are logged if the query reaches a FINISHED
// state. For certain query types, events are logged regardless of the query state.
Status status;
{
diff --git a/be/src/service/client-request-state.h b/be/src/service/client-request-state.h
index 83badb4..dd6d112 100644
--- a/be/src/service/client-request-state.h
+++ b/be/src/service/client-request-state.h
@@ -171,9 +171,13 @@ class ClientRequestState {
/// in-flight (for cleaning up after an error on the query issuing path).
Status Cancel(bool check_inflight, const Status* cause) WARN_UNUSED_RESULT;
- /// This is called when the query is done (finished, cancelled, or failed).
+ /// This is called when the query is done (finished, cancelled, or failed). This runs
+ /// synchronously within the last client RPC and does any work that is required before
+ /// the query is finished from the client's point of view, including cancelling the
+ /// query with 'cause'. Returns an error if 'check_inflight' is true and the query is
+ /// not yet in-flight, or if another thread has already started finalizing the query.
/// Takes lock_: callers must not hold lock() before calling.
- void Done();
+ Status Finalize(bool check_inflight, const Status* cause);
/// Sets the API-specific (Beeswax, HS2) result cache and its size bound.
/// The given cache is owned by this client request state, even if an error is returned.
@@ -309,6 +313,9 @@ class ClientRequestState {
/// Returns the FETCH_ROWS_TIMEOUT_MS value for this query (converted to microseconds).
int64_t fetch_rows_timeout_us() const { return fetch_rows_timeout_us_; }
+ /// True if Finalize() was called.
+ bool started_finalize() const { return started_finalize_.Load(); }
+
protected:
/// Updates the end_time_us_ of this query if it isn't set. The end time is determined
/// when this function is called for the first time, calling it multiple times does not
@@ -515,10 +522,14 @@ protected:
/// Start/end time of the query, in Unix microseconds.
int64_t start_time_us_;
/// end_time_us_ is initialized to 0, which is used to indicate that the query is not
- /// yet done. It is assinged the final value in ClientRequestState::Done() or when the
- /// coordinator relases its admission control resources.
+ /// yet done. It is assigned the final value in ClientRequestState::Finalize() or when
+ /// the coordinator releases its admission control resources.
AtomicInt64 end_time_us_{0};
+ /// True if a thread has called Finalize(). Threads calling Finalize()
+ /// do a compare-and-swap on this so that only one thread can proceed.
+ AtomicBool started_finalize_{false};
+
/// Timeout, in microseconds, when waiting for rows to become available. Derived from
/// the query option FETCH_ROWS_TIMEOUT_MS.
const int64_t fetch_rows_timeout_us_;
diff --git a/be/src/service/impala-beeswax-server.cc b/be/src/service/impala-beeswax-server.cc
index 827e58f..3efa216 100644
--- a/be/src/service/impala-beeswax-server.cc
+++ b/be/src/service/impala-beeswax-server.cc
@@ -178,7 +178,8 @@ void ImpalaServer::fetch(Results& query_results, const QueryHandle& query_handle
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state == nullptr)) {
- string err_msg = Substitute("Invalid query handle: $0", PrintId(query_id));
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ string err_msg = err.GetDetail();
VLOG(1) << err_msg;
RaiseBeeswaxException(err_msg, SQLSTATE_GENERAL_ERROR);
}
@@ -209,8 +210,8 @@ void ImpalaServer::get_results_metadata(ResultsMetadata& results_metadata,
VLOG_QUERY << "get_results_metadata(): query_id=" << PrintId(query_id);
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state.get() == nullptr)) {
- RaiseBeeswaxException(Substitute("Invalid query handle: $0", PrintId(query_id)),
- SQLSTATE_GENERAL_ERROR);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ RaiseBeeswaxException(err.GetDetail(), SQLSTATE_GENERAL_ERROR);
}
// Validate that query can be accessed by user.
RAISE_IF_ERROR(CheckClientRequestSession(session.get(), request_state->effective_user(),
@@ -276,8 +277,8 @@ beeswax::QueryState::type ImpalaServer::get_state(const QueryHandle& handle) {
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state == nullptr)) {
VLOG_QUERY << "ImpalaServer::get_state invalid handle";
- RaiseBeeswaxException(Substitute("Invalid query handle: $0", PrintId(query_id)),
- SQLSTATE_GENERAL_ERROR);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ RaiseBeeswaxException(err.GetDetail(), SQLSTATE_GENERAL_ERROR);
}
// Validate that query can be accessed by user.
RAISE_IF_ERROR(CheckClientRequestSession(session.get(), request_state->effective_user(),
@@ -379,7 +380,7 @@ void ImpalaServer::Cancel(impala::TStatus& tstatus,
// proceed without validating the session/query relation so that workflows don't
// get broken. In future we could check that the users match OR that the user has
// admin priviliges on the server.
- RAISE_IF_ERROR(CancelInternal(query_id, true), SQLSTATE_GENERAL_ERROR);
+ RAISE_IF_ERROR(CancelInternal(query_id), SQLSTATE_GENERAL_ERROR);
tstatus.status_code = TErrorCode::OK;
}
@@ -600,9 +601,9 @@ Status ImpalaServer::CloseInsertInternal(SessionState* session, const TUniqueId&
TDmlResult* dml_result) {
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state == nullptr)) {
- string err_msg = Substitute("Invalid query handle: $0", PrintId(query_id));
- VLOG(1) << err_msg;
- return Status::Expected(err_msg);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ VLOG(1) << err.GetDetail();
+ return err;
}
RETURN_IF_ERROR(
diff --git a/be/src/service/impala-hs2-server.cc b/be/src/service/impala-hs2-server.cc
index 6702f30..ccbea11 100644
--- a/be/src/service/impala-hs2-server.cc
+++ b/be/src/service/impala-hs2-server.cc
@@ -710,9 +710,9 @@ void ImpalaServer::GetOperationStatus(TGetOperationStatusResp& return_val,
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state.get() == nullptr)) {
- // No handle was found
- HS2_RETURN_ERROR(return_val,
- Substitute("Invalid query handle: $0", PrintId(query_id)), SQLSTATE_GENERAL_ERROR);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ string err_msg = err.GetDetail();
+ HS2_RETURN_ERROR(return_val, err_msg, SQLSTATE_GENERAL_ERROR);
}
ScopedSessionState session_handle(this);
@@ -748,16 +748,16 @@ void ImpalaServer::CancelOperation(TCancelOperationResp& return_val,
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state.get() == nullptr)) {
- // No handle was found
- HS2_RETURN_ERROR(return_val,
- Substitute("Invalid query handle: $0", PrintId(query_id)), SQLSTATE_GENERAL_ERROR);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ string err_msg = err.GetDetail();
+ HS2_RETURN_ERROR(return_val, err_msg, SQLSTATE_GENERAL_ERROR);
}
ScopedSessionState session_handle(this);
const TUniqueId session_id = request_state->session_id();
HS2_RETURN_IF_ERROR(return_val,
session_handle.WithSession(session_id, SecretArg::Operation(op_secret, query_id)),
SQLSTATE_GENERAL_ERROR);
- HS2_RETURN_IF_ERROR(return_val, CancelInternal(query_id, true), SQLSTATE_GENERAL_ERROR);
+ HS2_RETURN_IF_ERROR(return_val, CancelInternal(query_id), SQLSTATE_GENERAL_ERROR);
return_val.status.__set_statusCode(thrift::TStatusCode::SUCCESS_STATUS);
}
@@ -781,9 +781,8 @@ void ImpalaServer::CloseImpalaOperation(TCloseImpalaOperationResp& return_val,
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state.get() == nullptr)) {
- // No handle was found
- HS2_RETURN_ERROR(return_val,
- Substitute("Invalid query handle: $0", PrintId(query_id)), SQLSTATE_GENERAL_ERROR);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ HS2_RETURN_ERROR(return_val, err.GetDetail(), SQLSTATE_GENERAL_ERROR);
}
ScopedSessionState session_handle(this);
const TUniqueId session_id = request_state->session_id();
@@ -816,9 +815,8 @@ void ImpalaServer::GetResultSetMetadata(TGetResultSetMetadataResp& return_val,
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state.get() == nullptr)) {
VLOG_QUERY << "GetResultSetMetadata(): invalid query handle";
- // No handle was found
- HS2_RETURN_ERROR(return_val,
- Substitute("Invalid query handle: $0", PrintId(query_id)), SQLSTATE_GENERAL_ERROR);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ HS2_RETURN_ERROR(return_val, err.GetDetail(), SQLSTATE_GENERAL_ERROR);
}
ScopedSessionState session_handle(this);
const TUniqueId session_id = request_state->session_id();
@@ -869,7 +867,8 @@ void ImpalaServer::FetchResults(TFetchResultsResp& return_val,
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state == nullptr)) {
- string err_msg = Substitute("Invalid query handle: $0", PrintId(query_id));
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ string err_msg = err.GetDetail();
VLOG(1) << err_msg;
HS2_RETURN_ERROR(return_val, err_msg, SQLSTATE_GENERAL_ERROR);
}
@@ -915,9 +914,9 @@ void ImpalaServer::GetLog(TGetLogResp& return_val, const TGetLogReq& request) {
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (UNLIKELY(request_state.get() == nullptr)) {
- // No handle was found
- HS2_RETURN_ERROR(return_val,
- Substitute("Invalid query handle: $0", PrintId(query_id)), SQLSTATE_GENERAL_ERROR);
+ Status err = Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ string err_msg = err.GetDetail();
+ HS2_RETURN_ERROR(return_val, err_msg, SQLSTATE_GENERAL_ERROR);
}
// GetLog doesn't have an associated session handle, so we presume that this request
diff --git a/be/src/service/impala-http-handler.cc b/be/src/service/impala-http-handler.cc
index 3128708..cff8a69 100644
--- a/be/src/service/impala-http-handler.cc
+++ b/be/src/service/impala-http-handler.cc
@@ -21,6 +21,7 @@
#include <mutex>
#include <sstream>
#include <boost/lexical_cast.hpp>
+#include <boost/unordered_set.hpp>
#include <rapidjson/prettywriter.h>
#include <rapidjson/stringbuffer.h>
@@ -458,6 +459,7 @@ void ImpalaHttpHandler::QueryStateHandler(const Webserver::WebRequest& req,
sorted_query_records.insert(ImpalaServer::QueryStateRecord(*request_state));
});
+ unordered_set<TUniqueId> in_flight_query_ids;
Value in_flight_queries(kArrayType);
int64_t num_waiting_queries = 0;
for (const ImpalaServer::QueryStateRecord& record: sorted_query_records) {
@@ -467,6 +469,7 @@ void ImpalaHttpHandler::QueryStateHandler(const Webserver::WebRequest& req,
if (record_json["waiting"].GetBool()) ++num_waiting_queries;
in_flight_queries.PushBack(record_json, document->GetAllocator());
+ in_flight_query_ids.insert(record.id);
}
document->AddMember("in_flight_queries", in_flight_queries, document->GetAllocator());
document->AddMember("num_in_flight_queries",
@@ -488,6 +491,8 @@ void ImpalaHttpHandler::QueryStateHandler(const Webserver::WebRequest& req,
lock_guard<mutex> l(server_->query_log_lock_);
for (const unique_ptr<ImpalaServer::QueryStateRecord>& log_entry :
server_->query_log_) {
+ // Don't show duplicated entries between in-flight and completed queries.
+ if (in_flight_query_ids.find(log_entry->id) != in_flight_query_ids.end()) continue;
Value record_json(kObjectType);
QueryStateToJson(*log_entry, &record_json, document);
completed_queries.PushBack(record_json, document->GetAllocator());
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 6db86d8..4b420db 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -185,6 +185,16 @@ DEFINE_int32(max_profile_log_files, 10, "Maximum number of profile log files to
DEFINE_int32(cancellation_thread_pool_size, 5,
"(Advanced) Size of the thread-pool processing cancellations due to node failure");
+DEFINE_int32(unregistration_thread_pool_size, 4,
+ "(Advanced) Size of the thread-pool for unregistering queries, including "
+ "finalizing runtime profiles");
+// Limit the number of queries that can be queued for unregistration to avoid holding
+// too many queries in memory unnecessary. The default is set fairly low so that if
+// queries are finishing faster than they can be unregistered, there will be backpressure
+// on query execution before too much memory fills up with queries pending unregistration.
+DEFINE_int32(unregistration_thread_pool_queue_depth, 16,
+ "(Advanced) Max number of queries that can be queued for unregistration.");
+
DEFINE_string(ssl_server_certificate, "", "The full path to the SSL certificate file used"
" to authenticate Impala to clients. If set, both Beeswax and HiveServer2 ports will "
"only accept SSL connections");
@@ -356,7 +366,7 @@ const int64_t EXPIRATION_CHECK_INTERVAL_MS = 1000L;
// Template to return error messages for client requests that could not be found, belonged
// to the wrong session, or had a mismatched secret. We need to use this particular string
// in some places because the shell has a regex for it.
-// TODO: Make consistent with "Invalid query handle: $0" template used elsewhere.
+// TODO: Make consistent "Invalid or unknown query handle: $0" template used elsewhere.
// TODO: this should be turned into a proper error code and used throughout ImpalaServer.
static const char* LEGACY_INVALID_QUERY_HANDLE_TEMPLATE = "Query id $0 not found.";
@@ -458,6 +468,14 @@ ImpalaServer::ImpalaServer(ExecEnv* exec_env)
bind<void>(&ImpalaServer::CancelFromThreadPool, this, _1, _2)));
ABORT_IF_ERROR(cancellation_thread_pool_->Init());
+ unreg_thread_pool_.reset(new ThreadPool<std::shared_ptr<ClientRequestState>>(
+ "impala-server", "unregistration-worker",
+ FLAGS_unregistration_thread_pool_size, FLAGS_unregistration_thread_pool_queue_depth,
+ [this] (uint32_t thread_id, shared_ptr<ClientRequestState> crs) {
+ FinishUnregisterQuery(move(crs));
+ }));
+ ABORT_IF_ERROR(unreg_thread_pool_->Init());
+
// Initialize a session expiry thread which blocks indefinitely until the first session
// with non-zero timeout value is opened. Note that a session which doesn't specify any
// idle session timeout value will use the default value FLAGS_idle_session_timeout.
@@ -623,7 +641,8 @@ Status ImpalaServer::GetRuntimeProfileOutput(const TUniqueId& query_id,
DCHECK(output != nullptr);
// Search for the query id in the active query map
{
- shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
+ shared_ptr<ClientRequestState> request_state =
+ GetClientRequestState(query_id, /*return_unregistered=*/ true);
if (request_state.get() != nullptr) {
// For queries in INITIALIZED state, the profile information isn't populated yet.
if (request_state->exec_state() == ClientRequestState::ExecState::INITIALIZED) {
@@ -689,7 +708,8 @@ Status ImpalaServer::GetExecSummary(const TUniqueId& query_id, const string& use
TExecSummary* result) {
// Search for the query id in the active query map.
{
- shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
+ shared_ptr<ClientRequestState> request_state =
+ GetClientRequestState(query_id, /*return_unregistered=*/ true);
if (request_state != nullptr) {
lock_guard<mutex> l(*request_state->lock());
RETURN_IF_ERROR(CheckProfileAccess(user, request_state->effective_user(),
@@ -1130,25 +1150,35 @@ void ImpalaServer::UpdateExecSummary(
Status ImpalaServer::UnregisterQuery(const TUniqueId& query_id, bool check_inflight,
const Status* cause) {
VLOG_QUERY << "UnregisterQuery(): query_id=" << PrintId(query_id);
- // Cancel the query.
- RETURN_IF_ERROR(CancelInternal(query_id, check_inflight, cause));
-
- // Delete it from the client_request_state_map_ and from the http_handler_.
- DCHECK_EQ(this, ExecEnv::GetInstance()->impala_server());
- shared_ptr<ClientRequestState> request_state;
- RETURN_IF_ERROR(
- client_request_state_map_.DeleteClientRequestState(query_id, &request_state));
-
- // Close and delete the ClientRequestState.
- RETURN_IF_ERROR(CloseClientRequestState(request_state));
+ shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
+ if (request_state == nullptr) {
+ return Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
+ }
+ // We want to do some of the query unregistration work synchronously. Finalize
+ // only succeeds for the first thread to call it to avoid multiple threads
+ // unregistering.
+ RETURN_IF_ERROR(request_state->Finalize(check_inflight, cause));
+ // Do the rest of the unregistration work in the background so that the client does
+ // not need to wait for profile serialization, etc.
+ unreg_thread_pool_->Offer(move(request_state));
return Status::OK();
}
-Status ImpalaServer::CloseClientRequestState(
- const std::shared_ptr<ClientRequestState>& request_state) {
- request_state->Done();
+void ImpalaServer::FinishUnregisterQuery(shared_ptr<ClientRequestState>&& request_state) {
+ DCHECK_EQ(this, ExecEnv::GetInstance()->impala_server());
+ // Do all the finalization before removing the ClientRequestState from the map so that
+ // concurrent operations, e.g. GetRuntimeProfile() can find the query.
+ CloseClientRequestState(request_state);
+ // Make the ClientRequestState inaccessible. There is a time window where the query is
+ // both in 'client_request_state_map_' and 'query_locations_'.
+ Status status =
+ client_request_state_map_.DeleteClientRequestState(request_state->query_id());
+ DCHECK(status.ok()) << "CRS can only be deleted once: " << status.GetDetail();
+}
+void ImpalaServer::CloseClientRequestState(
+ const std::shared_ptr<ClientRequestState>& request_state) {
int64_t duration_us = request_state->end_time_us() - request_state->start_time_us();
int64_t duration_ms = duration_us / MICROS_PER_MILLI;
@@ -1188,7 +1218,6 @@ Status ImpalaServer::CloseClientRequestState(
}
ArchiveQuery(request_state.get());
ImpaladMetrics::NUM_QUERIES_REGISTERED->Increment(-1L);
- return Status::OK();
}
Status ImpalaServer::UpdateCatalogMetrics() {
@@ -1228,14 +1257,13 @@ Status ImpalaServer::UpdateCatalogMetrics() {
}
-Status ImpalaServer::CancelInternal(const TUniqueId& query_id, bool check_inflight,
- const Status* cause) {
+Status ImpalaServer::CancelInternal(const TUniqueId& query_id) {
VLOG_QUERY << "Cancel(): query_id=" << PrintId(query_id);
shared_ptr<ClientRequestState> request_state = GetClientRequestState(query_id);
if (request_state == nullptr) {
- return Status::Expected("Invalid or unknown query handle");
+ return Status::Expected(TErrorCode::INVALID_QUERY_HANDLE, PrintId(query_id));
}
- RETURN_IF_ERROR(request_state->Cancel(check_inflight, cause));
+ RETURN_IF_ERROR(request_state->Cancel(/*check_inflight=*/ true, /*cause=*/ nullptr));
return Status::OK();
}
@@ -2513,18 +2541,22 @@ void ImpalaServer::Join() {
}
shared_ptr<ClientRequestState> ImpalaServer::GetClientRequestState(
- const TUniqueId& query_id) {
+ const TUniqueId& query_id, bool return_unregistered) {
DCHECK_EQ(this, ExecEnv::GetInstance()->impala_server());
ScopedShardedMapRef<std::shared_ptr<ClientRequestState>> map_ref(query_id,
&client_request_state_map_);
DCHECK(map_ref.get() != nullptr);
auto entry = map_ref->find(query_id);
- if (entry == map_ref->end()) {
+ if (entry == map_ref->end()) return shared_ptr<ClientRequestState>();
+ // This started_finalize() check can race with unregistration. It cannot prevent
+ // unregistration starting immediately after the value is loaded. This check, however,
+ // is sufficient to ensure that after a client operation has unregistered the request,
+ // subsequent operations won't spuriously find the request.
+ if (!return_unregistered && entry->second->started_finalize()) {
return shared_ptr<ClientRequestState>();
- } else {
- return entry->second;
}
+ return entry->second;
}
Status ImpalaServer::CheckClientRequestSession(
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index 64d127e..1396259 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -509,7 +509,8 @@ class ImpalaServer : public ImpalaServiceIf,
/// Per-session state. This object is reference counted using shared_ptrs. There
/// is one ref count in the SessionStateMap for as long as the session is active.
- /// All queries running from this session also have a reference.
+ /// All queries running from this session also have a reference, because query
+ /// unregistration may complete asynchronously after the session is unregistered.
struct SessionState {
/// The default hs2_version must be V1 so that child queries (which use HS2, but may
/// run as children of Beeswax sessions) get results back in the expected format -
@@ -644,9 +645,12 @@ class ImpalaServer : public ImpalaServiceIf,
static const char* SQLSTATE_GENERAL_ERROR;
static const char* SQLSTATE_OPTIONAL_FEATURE_NOT_IMPLEMENTED;
- /// Return exec state for given query_id, or NULL if not found.
+ /// Return exec state for given query_id, or NULL if not found. If
+ /// 'return_unregistered' is true, queries that have started unregistration
+ /// may be returned. Otherwise queries that have started unregistration will
+ /// not be returned.
std::shared_ptr<ClientRequestState> GetClientRequestState(
- const TUniqueId& query_id);
+ const TUniqueId& query_id, bool return_unregistered=false);
/// Used in situations where the client provides a session ID and a query ID and the
/// caller needs to validate that the query can be accessed from the session. The two
@@ -697,29 +701,30 @@ class ImpalaServer : public ImpalaServiceIf,
Status SetQueryInflight(std::shared_ptr<SessionState> session_state,
const std::shared_ptr<ClientRequestState>& exec_state) WARN_UNUSED_RESULT;
- /// Unregister the query by cancelling it, removing exec_state from
- /// client_request_state_map_, and removing the query id from session state's
- /// in-flight query list. If check_inflight is true, then return an error if the query
+ /// Starts the process of unregistering the query. The query is cancelled on the
+ /// current thread, then asynchronously the query's entry is removed from
+ /// client_request_state_map_ and the session state's in-flight query list.
+ /// If check_inflight is true, then return an error if the query
/// is not yet in-flight. Otherwise, proceed even if the query isn't yet in-flight (for
/// cleaning up after an error on the query issuing path).
Status UnregisterQuery(const TUniqueId& query_id, bool check_inflight,
const Status* cause = NULL) WARN_UNUSED_RESULT;
- /// Performs any final cleanup necessary before the given ClientRequestState goes out
- /// of scope and is deleted. Marks the given ClientRequestState as done, removes the
- /// query from the inflight queries list, updates query_locations_, and archives the
- /// query. Used when unregistering the query.
- Status CloseClientRequestState(
- const std::shared_ptr<ClientRequestState>& request_state);
-
- /// Initiates query cancellation reporting the given cause as the query status.
- /// Assumes deliberate cancellation by the user if the cause is NULL. Returns an
- /// error if query_id is not found. If check_inflight is true, then return an error
- /// if the query is not yet in-flight. Otherwise, returns OK. Queries still need to
- /// be unregistered, after cancellation. Caller should not hold any locks when
- /// calling this function.
- Status CancelInternal(const TUniqueId& query_id, bool check_inflight,
- const Status* cause = NULL) WARN_UNUSED_RESULT;
+ /// Unregisters the provided query, does all required finalization and removes it from
+ /// 'client_request_state_' map.
+ void FinishUnregisterQuery(std::shared_ptr<ClientRequestState>&& request_state);
+
+ /// Performs finalization of 'request_state' before the request state is removed from
+ /// the server and deleted. Runs asynchronously after the request is reported done
+ /// to the client. Removes the query from the inflight queries list, updates
+ /// query_locations_, and archives the query.
+ void CloseClientRequestState(const std::shared_ptr<ClientRequestState>& request_state);
+
+ /// Initiates query cancellation triggered by the user (i.e. deliberate cancellation).
+ /// Returns an error if query_id is not found or if the query is not yet in flight.
+ /// Otherwise, returns OK. Queries still need to be unregistered after cancellation.
+ /// Caller should not hold any locks when calling this function.
+ Status CancelInternal(const TUniqueId& query_id);
/// Close the session and release all resource used by this session.
/// Caller should not hold any locks when calling this function.
@@ -1038,7 +1043,9 @@ class ImpalaServer : public ImpalaServiceIf,
/// Guards query_log_ and query_log_index_
std::mutex query_log_lock_;
- /// FIFO list of query records, which are written after the query finishes executing
+ /// FIFO list of query records, which are written after the query finishes executing.
+ /// Queries may briefly have entries in 'query_log_' and 'client_request_state_map_'
+ /// while the query is being unregistered.
typedef std::list<std::unique_ptr<QueryStateRecord>> QueryLog;
QueryLog query_log_;
@@ -1076,6 +1083,10 @@ class ImpalaServer : public ImpalaServiceIf,
/// avoid blocking the statestore callback.
boost::scoped_ptr<ThreadPool<CancellationWork>> cancellation_thread_pool_;
+ /// Thread pool to unregister queries asynchronously from RPCs. FinishUnregisterQuery()
+ /// is called on all ClientRequestStates added to this pool.
+ boost::scoped_ptr<ThreadPool<std::shared_ptr<ClientRequestState>>> unreg_thread_pool_;
+
/// Thread that runs SessionMaintenance. It will wake up periodically to check for
/// sessions which are idle for more their timeout values.
std::unique_ptr<Thread> session_maintenance_thread_;
@@ -1096,6 +1107,16 @@ class ImpalaServer : public ImpalaServiceIf,
/// A ClientRequestStateMap maps query ids to ClientRequestStates. The
/// ClientRequestStates are owned by the ImpalaServer and ClientRequestStateMap
/// references them using shared_ptr to allow asynchronous deletion.
+ ///
+ /// ClientRequestStates are unregistered from the server as follows:
+ /// 1. UnregisterQuery() is called, which calls ClientRequestState::Finalize() to cancel
+ /// query execution and start the unregistration process. At this point the query is
+ /// considered unregistered from the client's point of view.
+ /// 2. The ClientRequestState is enqueued in 'unreg_thread_pool_' to complete
+ /// unregistration asynchronously.
+ /// 3. Additional cleanup work is done by CloseClientRequestState(), and an entry
+ /// is added to 'query_log_' for this query.
+ /// 4. The ClientRequestState is removed from this map.
ClientRequestStateMap client_request_state_map_;
/// Default query options in the form of TQueryOptions and beeswax::ConfigVariable
diff --git a/be/src/util/runtime-profile.cc b/be/src/util/runtime-profile.cc
index c8605a1..a78bfef 100644
--- a/be/src/util/runtime-profile.cc
+++ b/be/src/util/runtime-profile.cc
@@ -140,9 +140,7 @@ RuntimeProfile::RuntimeProfile(
name_(name),
is_averaged_profile_(is_averaged_profile),
counter_total_time_(TUnit::TIME_NS),
- inactive_timer_(TUnit::TIME_NS),
- local_time_percent_(0),
- local_time_ns_(0) {
+ inactive_timer_(TUnit::TIME_NS) {
Counter* total_time_counter;
Counter* inactive_timer;
if (!is_averaged_profile) {
@@ -503,31 +501,33 @@ void RuntimeProfile::ComputeTimeInProfile(int64_t total) {
// IMPALA-5200: Take the max, because the parent includes all of the time from the
// children, whether or not its total time counter has been updated recently enough
// to see this.
- total_time_ns_ = max(children_total_time, total_time_counter()->value());
+ total_time_ns_.Store(max(children_total_time, total_time_counter()->value()));
// If a local time counter exists, use its value as local time. Otherwise, derive the
// local time from total time and the child time.
+ int64_t local_time_ns = 0;
bool has_local_time_counter = false;
{
lock_guard<SpinLock> l(counter_map_lock_);
CounterMap::const_iterator itr = counter_map_.find(LOCAL_TIME_COUNTER_NAME);
if (itr != counter_map_.end()) {
- local_time_ns_ = itr->second->value();
+ local_time_ns = itr->second->value();
has_local_time_counter = true;
}
}
if (!has_local_time_counter) {
- local_time_ns_ = total_time_ns_ - children_total_time;
+ local_time_ns = total_time_ns_.Load() - children_total_time;
if (!is_averaged_profile_) {
- local_time_ns_ -= inactive_timer()->value();
+ local_time_ns -= inactive_timer()->value();
}
}
// Counters have some margin, set to 0 if it was negative.
- local_time_ns_ = ::max<int64_t>(0, local_time_ns_);
- local_time_percent_ =
- static_cast<double>(local_time_ns_) / total_time_ns_;
- local_time_percent_ = ::min(1.0, local_time_percent_) * 100;
+ local_time_ns = ::max<int64_t>(0, local_time_ns);
+ local_time_ns_.Store(local_time_ns);
+ double local_time_frac =
+ min(1.0, static_cast<double>(local_time_ns) / total_time_ns_.Load());
+ local_time_frac_.Store(*reinterpret_cast<int64_t*>(&local_time_frac));
}
void RuntimeProfile::AddChild(RuntimeProfile* child, bool indent, RuntimeProfile* loc) {
@@ -962,13 +962,14 @@ void RuntimeProfile::PrettyPrint(ostream* s, const string& prefix) const {
stream.flags(ios::fixed);
stream << prefix << name_ << ":";
if (total_time->second->value() != 0) {
+ int64_t local_time_frac_int = local_time_frac_.Load();
stream << "(Total: "
<< PrettyPrinter::Print(total_time->second->value(),
total_time->second->unit())
<< ", non-child: "
- << PrettyPrinter::Print(local_time_ns_, TUnit::TIME_NS)
+ << PrettyPrinter::Print(local_time_ns_.Load(), TUnit::TIME_NS)
<< ", % non-child: "
- << setprecision(2) << local_time_percent_
+ << setprecision(2) << *reinterpret_cast<double*>(&local_time_frac_int) * 100
<< "%)";
}
stream << endl;
diff --git a/be/src/util/runtime-profile.h b/be/src/util/runtime-profile.h
index f2943d5..b55ac54 100644
--- a/be/src/util/runtime-profile.h
+++ b/be/src/util/runtime-profile.h
@@ -306,8 +306,8 @@ class RuntimeProfile { // NOLINT: This struct is not packed, but there are not s
/// Returns the counter for the total elapsed time.
Counter* total_time_counter() { return counter_map_[TOTAL_TIME_COUNTER_NAME]; }
Counter* inactive_timer() { return counter_map_[INACTIVE_TIME_COUNTER_NAME]; }
- int64_t local_time() { return local_time_ns_; }
- int64_t total_time() { return total_time_ns_; }
+ int64_t local_time() { return local_time_ns_.Load(); }
+ int64_t total_time() { return total_time_ns_.Load(); }
/// Prints the contents of the profile in a name: value format.
/// Does not hold locks when it makes any function calls.
@@ -450,7 +450,7 @@ class RuntimeProfile { // NOLINT: This struct is not packed, but there are not s
/// Recursively compute the fraction of the 'total_time' spent in this profile and
/// its children.
- /// This function updates local_time_percent_ for each profile.
+ /// This function updates local_time_frac_ for each profile.
void ComputeTimeInProfile();
/// Set ExecSummary
@@ -546,22 +546,26 @@ class RuntimeProfile { // NOLINT: This struct is not packed, but there are not s
Counter counter_total_time_;
/// Total time spent waiting (on non-children) that should not be counted when
- /// computing local_time_percent_. This is updated for example in the exchange
+ /// computing local_time_frac_. This is updated for example in the exchange
/// node when waiting on the sender from another fragment.
Counter inactive_timer_;
/// Time spent in just in this profile (i.e. not the children) as a fraction
- /// of the total time in the entire profile tree.
- double local_time_percent_;
+ /// of the total time in the entire profile tree. This is a double's bit pattern
+ /// stored in an integer. Computed in ComputeTimeInProfile().
+ /// Atomic so that it can be read concurrently with the value being calculated.
+ AtomicInt64 local_time_frac_{0};
/// Time spent in this node (not including the children). Computed in
- /// ComputeTimeInProfile()
- int64_t local_time_ns_;
+ /// ComputeTimeInProfile(). Atomic b/c it can be read concurrently with
+ /// ComputeTimeInProfile() executing.
+ AtomicInt64 local_time_ns_{0};
/// Total time spent in this node. Computed in ComputeTimeInProfile() and is
/// the maximum of the total time spent in children and the value of
- /// counter_total_time_.
- int64_t total_time_ns_;
+ /// counter_total_time_. Atomic b/c it can be read concurrently with
+ /// ComputeTimeInProfile() executing.
+ AtomicInt64 total_time_ns_{0};
/// The Exec Summary
TExecSummary t_exec_summary_;
diff --git a/common/thrift/generate_error_codes.py b/common/thrift/generate_error_codes.py
index e840b7f..594e390 100755
--- a/common/thrift/generate_error_codes.py
+++ b/common/thrift/generate_error_codes.py
@@ -459,6 +459,10 @@ error_codes = (
"Unable to perform Null-Aware Anti-Join. Could not get enough reservation to fit "
"all rows with NULLs from the build side in memory. Memory required for $0 rows "
"was $1. $2/$3 of the join's reservation was available for the rows."),
+
+ # Note: impala_shell uses a regex to search for this specific error message, so
+ # changing it may break older shell version.
+ ("INVALID_QUERY_HANDLE", 150, "Invalid or unknown query handle: $0."),
)
import sys
diff --git a/tests/common/impala_service.py b/tests/common/impala_service.py
index ee5a6e5..1911331 100644
--- a/tests/common/impala_service.py
+++ b/tests/common/impala_service.py
@@ -212,16 +212,19 @@ class ImpaladService(BaseImpalaService):
groups[backend['executor_groups']].append(backend['krpc_address'])
return groups
+ def get_queries_json(self):
+ """Return the full JSON from the /queries page."""
+ return json.loads(self.read_debug_webpage('queries?json', timeout=30, interval=1))
+
def get_query_locations(self):
# Returns a dictionary of the format <host_address, num_of_queries_running_there>
- result = json.loads(self.read_debug_webpage('queries?json', timeout=30, interval=1))
+ result = self.get_queries_json()
if result['query_locations'] is not None:
return dict([(loc["location"], loc["count"]) for loc in result['query_locations']])
return None
def get_in_flight_queries(self, timeout=30, interval=1):
- result = json.loads(self.read_debug_webpage('queries?json', timeout, interval))
- return result['in_flight_queries']
+ return self.get_queries_json()['in_flight_queries']
def _get_pool_counter(self, pool_name, counter_name, timeout=30, interval=1):
"""Returns the value of the field 'counter_name' in pool 'pool_name' or 0 if the pool
diff --git a/tests/comparison/leopard/report.py b/tests/comparison/leopard/report.py
index 2fd4d77..d1bbf5c 100644
--- a/tests/comparison/leopard/report.py
+++ b/tests/comparison/leopard/report.py
@@ -49,22 +49,23 @@ class Report(object):
def classify_error(self, error):
d = {
- ur'LINE \d+:': 'Postgres_error',
- ur'Permission denied': 'permission_denied',
- ur'^AnalysisException': 'AnalysisException',
- ur'^Column \d+ in row \d+ does not match': 'mismatch',
- ur'^Could not connect': 'could_not_connect',
- ur'^IllegalStateException': 'IllegalStateException',
- ur'^Invalid query handle: ': 'invalid_query_handle',
- ur'^Known issue:': 'known_issue',
- ur'^Operation is in ERROR_STATE': 'error_state',
- ur'^Query timed out after \d+ seconds': 'timeout',
- ur'^Row counts do not match': 'row_counts',
- ur'^Too much data': 'too_much_data',
- ur'^Unknown expr node type: \d+': 'unkown_node',
- ur'^Year is out of valid range': 'year_range',
- ur'^[A-Za-z]+ out of range': 'out_of_range',
- ur'^division by zero': 'division_by_zero'}
+ ur'LINE \d+:': 'Postgres_error',
+ ur'Permission denied': 'permission_denied',
+ ur'^AnalysisException': 'AnalysisException',
+ ur'^Column \d+ in row \d+ does not match': 'mismatch',
+ ur'^Could not connect': 'could_not_connect',
+ ur'^IllegalStateException': 'IllegalStateException',
+ ur'^Invalid query handle: ': 'invalid_query_handle',
+ ur'^Invalid or unknown query handle: ': 'invalid_query_handle',
+ ur'^Known issue:': 'known_issue',
+ ur'^Operation is in ERROR_STATE': 'error_state',
+ ur'^Query timed out after \d+ seconds': 'timeout',
+ ur'^Row counts do not match': 'row_counts',
+ ur'^Too much data': 'too_much_data',
+ ur'^Unknown expr node type: \d+': 'unkown_node',
+ ur'^Year is out of valid range': 'year_range',
+ ur'^[A-Za-z]+ out of range': 'out_of_range',
+ ur'^division by zero': 'division_by_zero'}
for r in d:
if re.search(r, error):
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index 6f9b7c4..5cc3063 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -778,7 +778,7 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
client.close_query(handle)
client.close_query(queued_query_handle)
queued_profile = client.get_runtime_profile(queued_query_handle)
- assert "Admission result: Cancelled (queued)" in queued_profile
+ assert "Admission result: Cancelled (queued)" in queued_profile, queued_profile
self.assert_impalad_log_contains('INFO', "Dequeued cancelled query=")
client.clear_configuration()
diff --git a/tests/hs2/hs2_test_suite.py b/tests/hs2/hs2_test_suite.py
index ef3875a..01bd574 100644
--- a/tests/hs2/hs2_test_suite.py
+++ b/tests/hs2/hs2_test_suite.py
@@ -113,18 +113,24 @@ class HS2TestSuite(ImpalaTestSuite):
'doubleVal', 'binaryVal']
def setup(self):
- host, port = IMPALAD_HS2_HOST_PORT.split(":")
- self.socket = TSocket(host, port)
- self.transport = TBufferedTransport(self.socket)
- self.transport.open()
- self.protocol = TBinaryProtocol.TBinaryProtocol(self.transport)
- self.hs2_client = ImpalaHiveServer2Service.Client(self.protocol)
+ self.socket, self.hs2_client = self._open_hs2_connection()
def teardown(self):
if self.socket:
self.socket.close()
@staticmethod
+ def _open_hs2_connection():
+ """Opens a HS2 connection, returning the socket and the thrift client."""
+ host, port = IMPALAD_HS2_HOST_PORT.split(":")
+ socket = TSocket(host, port)
+ transport = TBufferedTransport(socket)
+ transport.open()
+ protocol = TBinaryProtocol.TBinaryProtocol(transport)
+ hs2_client = ImpalaHiveServer2Service.Client(protocol)
+ return socket, hs2_client
+
+ @staticmethod
def check_response(response,
expected_status_code = TCLIService.TStatusCode.SUCCESS_STATUS,
expected_error_prefix = None):
@@ -150,7 +156,7 @@ class HS2TestSuite(ImpalaTestSuite):
expected_err = "Query id"
else:
# We should standardise on this error message.
- expected_err = "Invalid query handle:"
+ expected_err = "Invalid or unknown query handle:"
HS2TestSuite.check_response(response, TCLIService.TStatusCode.ERROR_STATUS,
expected_err)
diff --git a/tests/hs2/test_hs2.py b/tests/hs2/test_hs2.py
index c0c4237..c80decd 100644
--- a/tests/hs2/test_hs2.py
+++ b/tests/hs2/test_hs2.py
@@ -22,7 +22,10 @@ from contextlib import contextmanager
import json
import logging
import pytest
+import random
+import threading
import time
+import uuid
from urllib2 import urlopen
@@ -314,7 +317,8 @@ class TestHS2(HS2TestSuite):
get_operation_status_resp = \
self.get_operation_status(execute_statement_resp.operationHandle)
- # GetOperationState should return 'Invalid query handle' if the query has been closed.
+ # GetOperationState should return 'Invalid or unknown query handle' if the query has
+ # been closed.
TestHS2.check_response(get_operation_status_resp, \
TCLIService.TStatusCode.ERROR_STATUS)
@@ -368,7 +372,7 @@ class TestHS2(HS2TestSuite):
TestHS2.check_response(get_operation_status_resp, \
TCLIService.TStatusCode.ERROR_STATUS)
- err_msg = "Invalid query handle: efcdab8967452301:3031323334353637"
+ err_msg = "Invalid or unknown query handle: efcdab8967452301:3031323334353637"
assert err_msg in get_operation_status_resp.status.errorMessage
get_result_set_metadata_req = TCLIService.TGetResultSetMetadataReq()
@@ -378,7 +382,7 @@ class TestHS2(HS2TestSuite):
TestHS2.check_response(get_result_set_metadata_resp, \
TCLIService.TStatusCode.ERROR_STATUS)
- err_msg = "Invalid query handle: efcdab8967452301:3031323334353637"
+ err_msg = "Invalid or unknown query handle: efcdab8967452301:3031323334353637"
assert err_msg in get_result_set_metadata_resp.status.errorMessage
@pytest.mark.execute_serially
@@ -789,6 +793,99 @@ class TestHS2(HS2TestSuite):
"JSON content is invalid. Content: {0}".format(get_profile_resp.profile)
break
+ @needs_session()
+ def test_concurrent_unregister(self):
+ """Test that concurrently unregistering a query from multiple clients is safe and
+ that the profile can be fetched during the process."""
+ # Attach a UUID to the query text to make it easy to identify in web UI.
+ query_uuid = str(uuid.uuid4())
+ statement = "/*{0}*/ SELECT COUNT(2) FROM functional.alltypes".format(query_uuid)
+ execute_statement_resp = self.execute_statement(statement)
+ op_handle = execute_statement_resp.operationHandle
+
+ fetch_results_req = TCLIService.TFetchResultsReq()
+ fetch_results_req.operationHandle = op_handle
+ fetch_results_req.maxRows = 100
+ fetch_results_resp = self.hs2_client.FetchResults(fetch_results_req)
+ TestHS2.check_response(fetch_results_resp)
+
+ # Create a profile fetch thread and multiple unregister threads.
+ NUM_UNREGISTER_THREADS = 10
+ threads = []
+ profile_fetch_exception = [None]
+ unreg_exceptions = [None] * NUM_UNREGISTER_THREADS
+ sockets = []
+ try:
+ # Start a profile fetch thread first that will fetch the profile
+ # as the query is unregistered.
+ threads.append(threading.Thread(target=self._fetch_profile_loop,
+ args=(profile_fetch_exception, query_uuid, op_handle)))
+
+ # Start threads that will race to unregister the query.
+ for i in xrange(NUM_UNREGISTER_THREADS):
+ socket, client = self._open_hs2_connection()
+ sockets.append(socket)
+ threads.append(threading.Thread(target=self._unregister_query,
+ args=(i, unreg_exceptions, client, op_handle)))
+ for thread in threads:
+ thread.start()
+ for thread in threads:
+ thread.join()
+ finally:
+ for socket in sockets:
+ socket.close()
+
+ if profile_fetch_exception[0] is not None:
+ raise profile_fetch_exception[0]
+
+ # Validate the exceptions and ensure only one thread successfully unregistered
+ # the query.
+ num_successful = 0
+ for exception in unreg_exceptions:
+ if exception is None:
+ num_successful += 1
+ elif "Invalid or unknown query handle" not in str(exception):
+ raise exception
+ assert num_successful == 1, "Only one client should have been able to unregister"
+
+ def _fetch_profile_loop(self, exception_array, query_uuid, op_handle):
+ try:
+ # This thread will keep fetching the profile to make sure that the
+ # ClientRequestState object can be continually accessed during unregistration.
+ get_profile_req = ImpalaHiveServer2Service.TGetRuntimeProfileReq()
+ get_profile_req.operationHandle = op_handle
+ get_profile_req.sessionHandle = self.session_handle
+
+ def find_query(array):
+ """Find the query for this test in a JSON array returned from web UI."""
+ return [q for q in array if query_uuid in q['stmt']]
+ # Loop until the query has been unregistered and moved out of in-flight
+ # queries.
+ registered = True
+ while registered:
+ get_profile_resp = self.hs2_client.GetRuntimeProfile(get_profile_req)
+ TestHS2.check_response(get_profile_resp)
+ if "Unregister query" in get_profile_resp.profile:
+ json = self.impalad_test_service.get_queries_json()
+ inflight_query = find_query(json['in_flight_queries'])
+ completed_query = find_query(json['completed_queries'])
+ # Query should only be in one list.
+ assert len(inflight_query) + len(completed_query) == 1
+ if completed_query:
+ registered = False
+ except BaseException as e:
+ exception_array[0] = e
+
+ def _unregister_query(self, thread_num, exceptions, client, op_handle):
+ # Add some delay/jitter so that unregisters come in at different times.
+ time.sleep(0.01 + 0.005 * random.random())
+ try:
+ close_operation_req = TCLIService.TCloseOperationReq()
+ close_operation_req.operationHandle = op_handle
+ TestHS2.check_response(client.CloseOperation(close_operation_req))
+ except BaseException as e:
+ exceptions[thread_num] = e
+
@needs_session(conf_overlay={"use:database": "functional"})
def test_change_default_database(self):
statement = "SELECT 1 FROM alltypes LIMIT 1"
diff --git a/tests/hs2/test_json_endpoints.py b/tests/hs2/test_json_endpoints.py
index 102ccca..d17628d 100644
--- a/tests/hs2/test_json_endpoints.py
+++ b/tests/hs2/test_json_endpoints.py
@@ -19,6 +19,7 @@
import json
import pytest
+from time import time
from urllib2 import urlopen
@@ -88,12 +89,24 @@ class TestJsonEndpoints(HS2TestSuite):
assert not query["executing"]
assert query["waiting"]
- # Close the query and check that in_flight_queries is empty.
+ # Close the query and check that in_flight_queries becomes empty when the query
+ # gets unregistered.
close_operation_req = TCLIService.TCloseOperationReq()
close_operation_req.operationHandle = select_statement_resp.operationHandle
close_operation_resp = self.hs2_client.CloseOperation(close_operation_req)
TestJsonEndpoints.check_response(close_operation_resp)
- queries_json = self._get_json_queries(http_addr)
+
+ def no_inflight_queries():
+ queries_json = self._get_json_queries(http_addr)
+ return len(queries_json["in_flight_queries"]) == 0
+
+ self.assert_eventually(60, 0.1, no_inflight_queries)
+ start_time = time()
+ TIMEOUT_S = 60
+ while time() - start_time < TIMEOUT_S:
+ queries_json = self._get_json_queries(http_addr)
+ if len(queries_json["in_flight_queries"]) == 0:
+ break
assert len(queries_json["in_flight_queries"]) == 0
assert queries_json["num_in_flight_queries"] == 0
assert queries_json["num_executing_queries"] == 0
diff --git a/tests/shell/test_shell_commandline.py b/tests/shell/test_shell_commandline.py
index 42d8560..9962676 100644
--- a/tests/shell/test_shell_commandline.py
+++ b/tests/shell/test_shell_commandline.py
@@ -434,7 +434,7 @@ class TestImpalaShell(ImpalaTestSuite):
os.kill(p.pid(), signal.SIGINT)
result = p.get_result()
assert "Cancelling Query" in result.stderr
- assert "Invalid query handle" not in result.stderr
+ assert "Invalid or unknown query handle" not in result.stderr
def test_get_log_once(self, vector, empty_table):
"""Test that get_log() is always called exactly once."""
diff --git a/tests/stress/concurrent_select.py b/tests/stress/concurrent_select.py
index 50fed55..bb2d968 100755
--- a/tests/stress/concurrent_select.py
+++ b/tests/stress/concurrent_select.py
@@ -607,7 +607,7 @@ class StressRunner(object):
# example due to hitting a mem limit), just before the cancellation request, the
# server may have already unregistered the query as part of the fetch failure.
# In that case the server gives an error response saying the handle is invalid.
- if "Invalid query handle" in error_msg and report.timed_out:
+ if "Invalid or unknown query handle" in error_msg and report.timed_out:
self._num_successive_errors.value = 0
continue
# Occasionally the network connection will fail, and depending on when the
diff --git a/tests/util/cancel_util.py b/tests/util/cancel_util.py
index 5c08d6a..39e4700 100644
--- a/tests/util/cancel_util.py
+++ b/tests/util/cancel_util.py
@@ -61,12 +61,12 @@ def cancel_query_and_validate_state(client, query, exec_option, table_format,
assert close_error is None
elif close_error is None:
# If the close rpc succeeded, then the fetch rpc should have either succeeded,
- # failed with 'Cancelled' or failed with 'Invalid query handle' (if the close
- # rpc occured before the fetch rpc).
+ # failed with 'Cancelled' or failed with 'Invalid or unknown query handle'
+ # (if the close rpc occured before the fetch rpc).
if thread.fetch_results_error is not None:
assert 'Cancelled' in str(thread.fetch_results_error) or \
- ('Invalid query handle' in str(thread.fetch_results_error)
- and not join_before_close)
+ ('Invalid or unknown query handle' in str(thread.fetch_results_error)
+ and not join_before_close), str(thread.fetch_results_error)
else:
# If the close rpc encountered an exception, then it must be due to fetch
# noticing the cancellation and doing the auto-close.
[impala] 02/02: IMPALA-9699: Skip '-1' values when aggregating
num_null statistics
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 56c20ac4639cab3d6e5e0ae77ce395934a9294f6
Author: Tamas Mate <tm...@cloudera.com>
AuthorDate: Tue Apr 28 22:03:06 2020 +0200
IMPALA-9699: Skip '-1' values when aggregating num_null statistics
This change partially reverts IMPALA-8566 to make Impala backward
compatible with the old incremental partition stats. IMPALA-7659 added
collecting the number of null value statistics and IMPALA-8566 changed
the initial value of the incremental partition statistics from '-1' to
'0', because with '-1' the estimates were off by 1.
The old statistics in a new release can make the table metadata
inaccessible when the column stats are recomputed from the incremental
partition stats, which can be triggered by a partition level
'COMPUTE INCREMENTAL STATS'. In this case the old '-1' values can be
aggregated to a '<-1' 'num_nulls' value that later can fail a
Precondition check during table load.
The new behavior ensures that if any incremental partition stat has a
value of '-1' for 'num_nulls', the aggregated stats will be '-1',
regardless of whether or not other partitions have valid values for
'num_nulls'. This will prevent the planner from utilizing incomplete
statistics and the users will be notified about the missing statistics
with the general warning in the profile:
'The following tables are missing relevant table and/or column
statistics.'
Testing:
- Added unit test to verify the accepted values and aggregation result
Change-Id: I3fdf48a6c88378145078e068e12ade48c460f956
Reviewed-on: http://gerrit.cloudera.org:8080/15835
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/incr-stats-util-test.cc | 34 +++++++++++-
be/src/exec/incr-stats-util.cc | 108 +++++++++++++-----------------------
be/src/exec/incr-stats-util.h | 52 +++++++++++++++++
3 files changed, 124 insertions(+), 70 deletions(-)
diff --git a/be/src/exec/incr-stats-util-test.cc b/be/src/exec/incr-stats-util-test.cc
index 0410f1e..9f3227a 100644
--- a/be/src/exec/incr-stats-util-test.cc
+++ b/be/src/exec/incr-stats-util-test.cc
@@ -22,8 +22,7 @@
#include "testutil/gtest-util.h"
#include "exprs/aggregate-functions.h"
-
-#include "common/names.h"
+#include "exec/incr-stats-util.h"
using namespace impala;
@@ -72,3 +71,34 @@ TEST(IncrStatsUtilTest, TestEncode) {
ASSERT_EQ(DecodeNdv(encoded, is_encoded), test);
}
+/**
+ * This test checks the acceptable 'new_num_null' values by the PerColumnStats.Update
+ * method. In earlier releases the number of null values were not counted and the
+ * 'num_nulls' was set to '-1' to indicate missing statistics. To avoid misleading
+ * behavior between partition statistics created by different releases the column stats
+ * should be set to '-1' when a '-1' value exists in the partition stats.
+ */
+TEST(IncrStatsUtilTest, TestNumNullAggregation) {
+ PerColumnStats* stat = new PerColumnStats();
+
+ stat->Update(string(AggregateFunctions::HLL_LEN, 0), 0, 0, 0, 1);
+ ASSERT_EQ(1, stat->ToTColumnStats().num_nulls);
+
+ stat->Update(string(AggregateFunctions::HLL_LEN, 0), 0, 0, 0, 0);
+ ASSERT_EQ(1, stat->ToTColumnStats().num_nulls);
+
+ stat->Update(string(AggregateFunctions::HLL_LEN, 0), 0, 0, 0, 2);
+ ASSERT_EQ(3, stat->ToTColumnStats().num_nulls);
+
+ stat->Update(string(AggregateFunctions::HLL_LEN, 0), 0, 0, 0, -1);
+ ASSERT_EQ(-1, stat->ToTColumnStats().num_nulls);
+
+ stat->Update(string(AggregateFunctions::HLL_LEN, 0), 0, 0, 0, 0);
+ ASSERT_EQ(-1, stat->ToTColumnStats().num_nulls);
+
+ stat->Update(string(AggregateFunctions::HLL_LEN, 0), 0, 0, 0, 3);
+ ASSERT_EQ(-1, stat->ToTColumnStats().num_nulls);
+
+ stat->Update(string(AggregateFunctions::HLL_LEN, 0), 0, 0, 0, -1);
+ ASSERT_EQ(-1, stat->ToTColumnStats().num_nulls);
+}
diff --git a/be/src/exec/incr-stats-util.cc b/be/src/exec/incr-stats-util.cc
index 6cc9f2f..54b4868 100644
--- a/be/src/exec/incr-stats-util.cc
+++ b/be/src/exec/incr-stats-util.cc
@@ -112,79 +112,51 @@ string DecodeNdv(const string& ndv, bool is_encoded) {
return decoded_ndv;
}
-// A container for statistics for a single column that are aggregated partition by
-// partition during the incremental computation of column stats. The aggregations are
-// updated during Update(), and the final statistics are computed by Finalize().
-struct PerColumnStats {
- // Should have length AggregateFunctions::HLL_PRECISION. Intermediate buckets for the
- // HLL calculation.
- string intermediate_ndv;
-
- // The total number of nulls counted, or -1 for no sample.
- int64_t num_nulls;
-
- // The maximum width of the column, in bytes.
- int32_t max_width;
-
- // The total number of rows
- int64_t num_rows;
-
- // The sum of avg_width * num_rows for each partition, so that avg_width can be
- // correctly computed during Finalize()
- double total_width;
-
- // Populated after Finalize(), the result of the HLL computation
- int64_t ndv_estimate;
-
- // The average column width, in bytes (but may have non-integer value)
- double avg_width;
-
- PerColumnStats()
- : intermediate_ndv(AggregateFunctions::HLL_LEN, 0), num_nulls(0),
- max_width(0), num_rows(0), avg_width(0) { }
-
- // Updates all aggregate statistics with a new set of measurements.
- void Update(const string& ndv, int64_t num_new_rows, double new_avg_width,
- int32_t max_new_width, int64_t num_new_nulls) {
- DCHECK_EQ(intermediate_ndv.size(), ndv.size()) << "Incompatible intermediate NDVs";
- DCHECK_GE(num_new_rows, 0);
- DCHECK_GE(max_new_width, 0);
- DCHECK_GE(new_avg_width, 0);
- DCHECK_GE(num_new_nulls, 0);
- for (int j = 0; j < ndv.size(); ++j) {
- intermediate_ndv[j] = ::max(intermediate_ndv[j], ndv[j]);
+void PerColumnStats::Update(const string& ndv, int64_t num_new_rows, double new_avg_width,
+ int32_t max_new_width, int64_t num_new_nulls) {
+ DCHECK_EQ(intermediate_ndv.size(), ndv.size())
+ << "Incompatible intermediate NDVs";
+ DCHECK_GE(num_new_rows, 0);
+ DCHECK_GE(max_new_width, 0);
+ DCHECK_GE(new_avg_width, 0);
+ DCHECK_GE(num_new_nulls, -1); // '-1' needed to be backward compatible
+ for (int j = 0; j < ndv.size(); ++j) {
+ intermediate_ndv[j] = ::max(intermediate_ndv[j], ndv[j]);
+ }
+ // Earlier the 'num_nulls' were initialized and persisted with '-1', this condition
+ // ensures metadata backward compatibility between releases
+ if (num_nulls >= 0) {
+ if (num_new_nulls >= 0) {
+ num_nulls += num_new_nulls;
+ } else {
+ num_nulls = -1;
}
- num_nulls += num_new_nulls;
- max_width = ::max(max_width, max_new_width);
- avg_width += (new_avg_width * num_new_rows);
- num_rows += num_new_rows;
}
+ max_width = ::max(max_width, max_new_width);
+ avg_width += (new_avg_width * num_new_rows);
+ num_rows += num_new_rows;
+}
- // Performs any stats computations that are not distributive, that is they may not be
- // computed in part during Update(). After this method returns, ndv_estimate and
- // avg_width contain valid values.
- void Finalize() {
- ndv_estimate = AggregateFunctions::HllFinalEstimate(
- reinterpret_cast<const uint8_t*>(intermediate_ndv.data()));
- avg_width = num_rows == 0 ? 0 : avg_width / num_rows;
- }
+void PerColumnStats::Finalize() {
+ ndv_estimate = AggregateFunctions::HllFinalEstimate(
+ reinterpret_cast<const uint8_t*>(intermediate_ndv.data()));
+ avg_width = num_rows == 0 ? 0 : avg_width / num_rows;
+}
- TColumnStats ToTColumnStats() const {
- TColumnStats col_stats;
- col_stats.__set_num_distinct_values(ndv_estimate);
- col_stats.__set_num_nulls(num_nulls);
- col_stats.__set_max_size(max_width);
- col_stats.__set_avg_size(avg_width);
- return col_stats;
- }
+TColumnStats PerColumnStats::ToTColumnStats() const {
+ TColumnStats col_stats;
+ col_stats.__set_num_distinct_values(ndv_estimate);
+ col_stats.__set_num_nulls(num_nulls);
+ col_stats.__set_max_size(max_width);
+ col_stats.__set_avg_size(avg_width);
+ return col_stats;
+}
- // Returns a string with debug information for this
- string DebugString() const {
- return Substitute(
- "ndv: $0, num_nulls: $1, max_width: $2, avg_width: $3, num_rows: $4",
- ndv_estimate, num_nulls, max_width, avg_width, num_rows);
- }
-};
+string PerColumnStats::DebugString() const {
+ return Substitute(
+ "ndv: $0, num_nulls: $1, max_width: $2, avg_width: $3, num_rows: $4",
+ ndv_estimate, num_nulls, max_width, avg_width, num_rows);
+}
namespace impala {
diff --git a/be/src/exec/incr-stats-util.h b/be/src/exec/incr-stats-util.h
index 0fea9b1..8e0480a 100644
--- a/be/src/exec/incr-stats-util.h
+++ b/be/src/exec/incr-stats-util.h
@@ -20,6 +20,58 @@
#include "gen-cpp/CatalogObjects_types.h"
#include <vector>
+#include "exprs/aggregate-functions.h"
+#include "common/names.h"
+
+using namespace impala;
+
+/// A container for statistics for a single column that are aggregated partition by
+/// partition during the incremental computation of column stats. The aggregations are
+/// updated during Update(), and the final statistics are computed by Finalize().
+struct PerColumnStats {
+ // Should have length AggregateFunctions::HLL_PRECISION. Intermediate buckets for the
+ // HLL calculation.
+ string intermediate_ndv;
+
+ // The total number of nulls counted, or -1 for no sample.
+ int64_t num_nulls;
+
+ // The maximum width of the column, in bytes.
+ int32_t max_width;
+
+ // The total number of rows
+ int64_t num_rows;
+
+ // The sum of avg_width * num_rows for each partition, so that avg_width can be
+ // correctly computed during Finalize()
+ // TODO: IMPALA-9722: consolidate unused total_width and the way avg_width is computed
+ double total_width;
+
+ // Populated after Finalize(), the result of the HLL computation
+ int64_t ndv_estimate;
+
+ // The average column width, in bytes (but may have non-integer value)
+ double avg_width;
+
+ PerColumnStats()
+ : intermediate_ndv(AggregateFunctions::HLL_LEN, 0), num_nulls(0),
+ max_width(0), num_rows(0), avg_width(0) { }
+
+ /// Updates all aggregate statistics with a new set of measurements.
+ void Update(const string& ndv, int64_t num_new_rows, double new_avg_width,
+ int32_t max_new_width, int64_t num_new_nulls);
+
+ /// Performs any stats computations that are not distributive, that is they may not be
+ /// computed in part during Update(). After this method returns, ndv_estimate and
+ /// avg_width contain valid values.
+ void Finalize();
+
+ /// Creates a TColumnStats object from PerColumnStats
+ TColumnStats ToTColumnStats() const;
+
+ /// Returns a string with debug information for this
+ string DebugString() const;
+};
namespace impala {