You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2024/03/19 22:31:06 UTC
(impala) branch master updated: IMPALA-12426: Query History Table
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new 711a9f2ba IMPALA-12426: Query History Table
711a9f2ba is described below
commit 711a9f2bad84f92dc4af61d49ae115f0dc4239da
Author: jasonmfehr <jf...@cloudera.com>
AuthorDate: Mon Oct 2 13:28:41 2023 -0700
IMPALA-12426: Query History Table
Adds the ability for users to specify that Impala will create and
maintain an internal Iceberg table that contains data about all
completed queries. This table is automatically created at startup by
each coordinator if it does not exist. Then, most completed queries are
queued in memory and flushed to the query history table at a set
interval (either minutes or number of records). Set, use, and show
queries are not written to this table. This commit leverages the
InternalServer class to maintain the query history table.
Ctest unit tests have been added to assert the various pieces of code.
New custom cluster tests have been added to assert the query history
table is properly populated with completed queries.
Negative testing consists of attempting sql injection attacks and
syntactically incorrect queries.
Impala built-in string functions benchmarks have been updated to include
the new built-in functions.
Change-Id: I2d2da9d450fba4e789400cfa62927fc25d34f844
Reviewed-on: http://gerrit.cloudera.org:8080/20770
Reviewed-by: Riza Suminto <ri...@cloudera.com>
Reviewed-by: Michael Smith <mi...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/query-driver.cc | 8 +
be/src/runtime/query-driver.h | 9 +
be/src/service/CMakeLists.txt | 3 +
be/src/service/impala-http-handler.cc | 3 +
be/src/service/impala-server.cc | 32 +-
be/src/service/impala-server.h | 61 +-
be/src/service/internal-server-test.cc | 22 +-
be/src/service/internal-server.cc | 53 +-
be/src/service/internal-server.h | 57 +-
be/src/service/query-state-record.cc | 27 +-
be/src/service/query-state-record.h | 28 +
be/src/service/workload-management-fields.cc | 417 ++++++++
be/src/service/workload-management-flags.cc | 172 ++++
be/src/service/workload-management.cc | 418 ++++++++
be/src/service/workload-management.h | 97 ++
be/src/util/CMakeLists.txt | 3 +
be/src/util/backend-gflag-util.cc | 4 +
be/src/util/impalad-metrics.cc | 25 +
be/src/util/impalad-metrics.h | 24 +
be/src/util/sql-util-test.cc | 95 ++
be/src/util/sql-util.cc | 70 ++
be/src/util/sql-util.h | 48 +
common/thrift/BackendGflags.thrift | 4 +
common/thrift/metrics.json | 50 +
.../impala/catalog/CatalogServiceCatalog.java | 12 +
.../org/apache/impala/service/BackendConfig.java | 8 +
.../apache/impala/util/CatalogBlacklistUtils.java | 4 +
tests/authorization/test_provider.py | 4 +-
tests/beeswax/impala_beeswax.py | 25 +-
tests/common/custom_cluster_test_suite.py | 17 +-
tests/common/impala_connection.py | 22 +-
tests/custom_cluster/test_query_log.py | 1007 ++++++++++++++++++++
tests/util/assert_time.py | 57 ++
tests/util/memory.py | 74 ++
tests/util/retry.py | 87 ++
tests/util/workload_management.py | 730 ++++++++++++++
36 files changed, 3705 insertions(+), 72 deletions(-)
diff --git a/be/src/runtime/query-driver.cc b/be/src/runtime/query-driver.cc
index edd18c49d..18442370b 100644
--- a/be/src/runtime/query-driver.cc
+++ b/be/src/runtime/query-driver.cc
@@ -547,6 +547,14 @@ Status QueryDriver::Unregister(ImpalaServer::QueryDriverMap* query_driver_map) {
return Status::OK();
}
+void QueryDriver::IncludeInQueryLog(const bool include) noexcept{
+ include_in_query_log_ = include;
+}
+
+bool QueryDriver::IncludedInQueryLog() const noexcept {
+ return include_in_query_log_;
+}
+
void QueryDriver::CreateNewDriver(ImpalaServer* impala_server, QueryHandle* query_handle,
const TQueryCtx& query_ctx, shared_ptr<ImpalaServer::SessionState> session_state) {
query_handle->query_driver_ = std::make_shared<QueryDriver>(impala_server);
diff --git a/be/src/runtime/query-driver.h b/be/src/runtime/query-driver.h
index b9ee2b3de..4f2f352bc 100644
--- a/be/src/runtime/query-driver.h
+++ b/be/src/runtime/query-driver.h
@@ -190,6 +190,11 @@ class QueryDriver {
/// True if Finalize() was called while the query was inflight.
bool finalized() const { return finalized_.Load(); }
+ /// Functions to set/get whether or not the query managed by this class should be
+ /// recorded in the query log table.
+ void IncludeInQueryLog(const bool include) noexcept;
+ bool IncludedInQueryLog() const noexcept;
+
/// Creates a new QueryDriver instance using the given ImpalaServer. Creates the
/// ClientRequestState for the given 'query_ctx' and 'session_state'. Sets the given
/// QueryHandle's QueryDriver.
@@ -273,5 +278,9 @@ class QueryDriver {
/// True if a thread has called Finalize() and the query is inflight. Threads calling
/// Finalize() do a compare-and-swap on this so that only one thread can proceed.
AtomicBool finalized_{false};
+
+ /// True if this query should be recorded in the query log table.
+ /// Default: `true`
+ bool include_in_query_log_ = true;
};
}
diff --git a/be/src/service/CMakeLists.txt b/be/src/service/CMakeLists.txt
index 7941c2e73..a6b51395e 100644
--- a/be/src/service/CMakeLists.txt
+++ b/be/src/service/CMakeLists.txt
@@ -44,6 +44,9 @@ add_library(Service
query-options.cc
query-result-set.cc
query-state-record.cc
+ workload-management.cc
+ workload-management-fields.cc
+ workload-management-flags.cc
)
add_dependencies(Service gen-deps)
diff --git a/be/src/service/impala-http-handler.cc b/be/src/service/impala-http-handler.cc
index 390a218bb..27aba2618 100644
--- a/be/src/service/impala-http-handler.cc
+++ b/be/src/service/impala-http-handler.cc
@@ -65,6 +65,7 @@ using namespace rapidjson;
using namespace strings;
DECLARE_int32(query_log_size);
+DECLARE_int64(query_log_size_in_bytes);
DECLARE_int32(query_stmt_size);
DECLARE_bool(use_local_catalog);
DECLARE_string(admission_service_host);
@@ -675,6 +676,8 @@ void ImpalaHttpHandler::QueryStateHandler(const Webserver::WebRequest& req,
document->AddMember("completed_queries", completed_queries, document->GetAllocator());
document->AddMember("completed_log_size", FLAGS_query_log_size,
document->GetAllocator());
+ document->AddMember("completed_log_size_in_bytes", FLAGS_query_log_size_in_bytes,
+ document->GetAllocator());
Value query_locations(kArrayType);
{
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 3a307e1de..6e5c3d325 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -1146,7 +1146,6 @@ void ImpalaServer::ArchiveQuery(const QueryHandle& query_handle) {
}
}
- if (FLAGS_query_log_size == 0 || FLAGS_query_log_size_in_bytes == 0) return;
// 'fetch_rows_lock()' protects several fields in ClientReqestState that are read
// during QueryStateRecord creation. There should be no contention on this lock because
// the query has already been closed (e.g. no more results can be fetched).
@@ -1158,10 +1157,14 @@ void ImpalaServer::ArchiveQuery(const QueryHandle& query_handle) {
if (query_handle->GetCoordinator() != nullptr) {
query_handle->GetCoordinator()->GetTExecSummary(&record->exec_summary);
}
- int64_t record_size = EstimateSize(record.get());
- VLOG(3) << "QueryStateRecord of " << PrintId(query_handle->query_id()) << " is "
- << record_size << " bytes";
- {
+
+ EnqueueCompletedQuery(query_handle, record);
+
+ if (FLAGS_query_log_size != 0 && FLAGS_query_log_size_in_bytes != 0) {
+ int64_t record_size = EstimateSize(record.get());
+ VLOG(3) << "QueryStateRecord of " << PrintId(query_handle->query_id()) << " is "
+ << record_size << " bytes";
+
lock_guard<mutex> l(query_log_lock_);
// Add record to the beginning of the log, and to the lookup index.
ImpaladMetrics::QUERY_LOG_EST_TOTAL_BYTES->Increment(record_size);
@@ -1250,8 +1253,8 @@ void ImpalaServer::EnforceMaxMtDop(TQueryCtx* query_ctx, int64_t max_mt_dop) {
}
Status ImpalaServer::Execute(TQueryCtx* query_ctx, shared_ptr<SessionState> session_state,
- QueryHandle* query_handle,
- const TExecRequest* external_exec_request) {
+ QueryHandle* query_handle, const TExecRequest* external_exec_request,
+ const bool include_in_query_log) {
PrepareQueryContext(query_ctx);
ScopedThreadContext debug_ctx(GetThreadDebugInfo(), query_ctx->query_id);
ImpaladMetrics::IMPALA_SERVER_NUM_QUERIES->Increment(1L);
@@ -1264,6 +1267,9 @@ Status ImpalaServer::Execute(TQueryCtx* query_ctx, shared_ptr<SessionState> sess
bool registered_query = false;
Status status = ExecuteInternal(*query_ctx, external_exec_request, session_state,
®istered_query, query_handle);
+
+ query_handle->query_driver()->IncludeInQueryLog(include_in_query_log);
+
if (!status.ok() && registered_query) {
UnregisterQueryDiscardResult((*query_handle)->query_id(), false, &status);
}
@@ -3165,6 +3171,10 @@ Status ImpalaServer::Start(int32_t beeswax_port, int32_t hs2_port,
hs2_http_server_.reset(http_server);
hs2_http_server_->SetConnectionHandler(this);
}
+
+ internal_server_ = shared_from_this();
+
+ RETURN_IF_ERROR(InitWorkloadManagement());
}
LOG(INFO) << "Initialized coordinator/executor Impala server on "
<< TNetworkAddressToString(exec_env_->configured_backend_address());
@@ -3346,6 +3356,10 @@ Status ImpalaServer::StartShutdown(
break;
}
}
+
+ // Drain the completed queries queue to the query log table.
+ ShutdownWorkloadManagement();
+
LOG(INFO) << "Shutdown complete, going down.";
// Use _exit here instead since exit() does cleanup which interferes with the shutdown
// signal handler thread causing a data race.
@@ -3382,6 +3396,10 @@ void ImpalaServer::GetAllConnectionContexts(
if (external_fe_server_.get()) {
external_fe_server_->GetConnectionContextList(connection_contexts);
}
+ // Get the connection contexts of the internal server
+ if (internal_server_.get()) {
+ internal_server_->GetConnectionContextList(connection_contexts);
+ }
}
TUniqueId ImpalaServer::RandomUniqueID() {
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index 7e0d6b96d..332b4725a 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -18,6 +18,7 @@
#pragma once
#include <atomic>
+#include <memory>
#include <mutex>
#include <unordered_map>
#include <boost/random/random_device.hpp>
@@ -34,12 +35,14 @@
#include "gen-cpp/ImpalaHiveServer2Service.h"
#include "gen-cpp/ImpalaService.h"
#include "gen-cpp/control_service.pb.h"
+#include "gen-cpp/Query_types.h"
#include "kudu/util/random.h"
#include "rpc/thrift-server.h"
#include "runtime/types.h"
#include "service/internal-server.h"
#include "service/query-options.h"
#include "service/query-state-record.h"
+#include "service/workload-management.h"
#include "statestore/statestore-subscriber.h"
#include "util/condition-variable.h"
#include "util/container-util.h"
@@ -47,6 +50,7 @@
#include "util/sharded-query-map-util.h"
#include "util/simple-logger.h"
#include "util/thread-pool.h"
+#include "util/ticker.h"
#include "util/time.h"
namespace kudu {
@@ -406,15 +410,18 @@ class ImpalaServer : public ImpalaServiceIf,
const TQueryOptions& query_opts = TQueryOptions());
virtual bool CloseSession(const impala::TUniqueId& session_id);
virtual Status ExecuteIgnoreResults(const std::string& user_name,
- const std::string& sql, TUniqueId* query_id = nullptr);
+ const std::string& sql, const TQueryOptions& query_opts = TQueryOptions(),
+ const bool persist_in_db = true, TUniqueId* query_id = nullptr);
virtual Status ExecuteAndFetchAllText(const std::string& user_name,
const std::string& sql, query_results& results, results_columns* columns = nullptr,
TUniqueId* query_id = nullptr);
virtual Status SubmitAndWait(const std::string& user_name, const std::string& sql,
- TUniqueId& new_session_id, TUniqueId& new_query_id);
+ TUniqueId& new_session_id, TUniqueId& new_query_id,
+ const TQueryOptions& query_opts = TQueryOptions(),
+ const bool persist_in_db = true);
virtual Status WaitForResults(TUniqueId& query_id);
virtual Status SubmitQuery(const std::string& sql, const impala::TUniqueId& session_id,
- TUniqueId& new_query_id);
+ TUniqueId& new_query_id, const bool persist_in_db = true);
virtual Status FetchAllRows(const TUniqueId& query_id, query_results& results,
results_columns* columns = nullptr);
virtual void CloseQuery(const TUniqueId& query_id);
@@ -738,8 +745,8 @@ class ImpalaServer : public ImpalaServiceIf,
/// external_exec_request is a statement that was prepared by an external frontend using
/// Impala PlanNodes or null if the external frontend isn't being used.
Status Execute(TQueryCtx* query_ctx, std::shared_ptr<SessionState> session_state,
- QueryHandle* query_handle,
- const TExecRequest* external_exec_request) WARN_UNUSED_RESULT;
+ QueryHandle* query_handle, const TExecRequest* external_exec_request,
+ const bool include_in_query_log = true) WARN_UNUSED_RESULT;
/// Implements Execute() logic, but doesn't unregister query on error.
Status ExecuteInternal(const TQueryCtx& query_ctx,
@@ -942,6 +949,11 @@ class ImpalaServer : public ImpalaServiceIf,
/// which are returned as strings (see IMPALA-11041).
std::string ColumnTypeToBeeswaxTypeString(const TColumnType& type);
+ /// Places a completed query into the in-memory queue of completed queries.
+ /// (implemented in workload-management.cc)
+ void EnqueueCompletedQuery(const QueryHandle& query_handle,
+ const std::shared_ptr<QueryStateRecord> qs_rec);
+
/// Returns the active QueryHandle for this query id. The QueryHandle contains the
/// active ClientRequestState. Returns an error Status if the query id cannot be found.
/// If caller is an RPC thread, RPC context will be registered for time tracking
@@ -1106,6 +1118,18 @@ class ImpalaServer : public ImpalaServiceIf,
/// current query ids to the admissiond.
[[noreturn]] void AdmissionHeartbeatThread();
+ /// If workload management is enabled, starts workload management threads.
+ /// (implemented in workload-management.cc)
+ Status InitWorkloadManagement();
+
+ /// Blocks until running workload management threads are shut down.
+ /// (implemented in workload-management.cc)
+ void ShutdownWorkloadManagement();
+
+ /// Periodically writes out completed queries (if configured)
+ /// (implemented in workload-management.cc)
+ void CompletedQueriesThread();
+
/// Called from ExpireQueries() to check query resource limits for 'crs'. If the query
/// exceeded a resource limit, returns a non-OK status with information about what
/// limit was exceeded. Returns OK if the query will continue running and expiration
@@ -1583,6 +1607,7 @@ class ImpalaServer : public ImpalaServiceIf,
boost::scoped_ptr<ThriftServer> hs2_server_;
boost::scoped_ptr<ThriftServer> hs2_http_server_;
boost::scoped_ptr<ThriftServer> external_fe_server_;
+ std::shared_ptr<InternalServer> internal_server_;
/// Flag that records if backend and/or client services have been started. The flag is
/// set after all services required for the server have been started.
@@ -1600,5 +1625,31 @@ class ImpalaServer : public ImpalaServiceIf,
/// Stores the last version number for the admission heartbeat that was sent.
/// Incremented every time a new admission heartbeat is sent.
int64_t admission_heartbeat_version_ = 0;
+
+ /// Workload Management Related Declarations.
+ /// Coordinate periodic execution of the completed queries queue processing thread.
+ std::condition_variable completed_queries_cv_;
+
+ /// Coordinate shutdown of the completed queries queue processing thread.
+ std::condition_variable completed_queries_shutdown_cv_;
+
+ /// Tracks the state of the thread that drains the completed queries queue to the table.
+ /// The associated lock must be held before reading/modifying this variable.
+ impala::workload_management::ThreadState completed_queries_thread_state_ =
+ impala::workload_management::NOT_STARTED;
+ std::mutex completed_queries_threadstate_mu_;
+
+ /// Thread that runs CompletedQueriesThread().
+ std::unique_ptr<Thread> completed_queries_thread_;
+
+ /// Ticker that wakes up the completed_queried_thread at set intervals to process the
+ /// queued completed queries. Uses the completed_queries_lock_ to synchonize access to
+ /// the completed_queries_ list.
+ std::unique_ptr<TickerSecondsBool> completed_queries_ticker_;
+
+ /// Queue of completed queries and the lock to synchronize access to it.
+ std::list<impala::workload_management::CompletedQuery> completed_queries_;
+ std::mutex completed_queries_lock_;
+
};
}
diff --git a/be/src/service/internal-server-test.cc b/be/src/service/internal-server-test.cc
index 04f55fd81..c1cb1ab79 100644
--- a/be/src/service/internal-server-test.cc
+++ b/be/src/service/internal-server-test.cc
@@ -129,7 +129,8 @@ class DatabaseTest {
database_name_ = StrCat(name_prefix, "_", GetCurrentTimeMicros());
TUniqueId query_id;
EXPECT_OK(impala_server_->ExecuteIgnoreResults("impala", StrCat("create database ",
- database_name_), &query_id));
+ database_name_, " comment 'Temporary database created and managed by "
+ "internal-server-test'"), TQueryOptions(), false, &query_id));
assertQueryState(query_id, QUERY_STATE_SUCCESS);
if (create_table) {
@@ -137,7 +138,7 @@ class DatabaseTest {
EXPECT_OK(impala_server_->ExecuteIgnoreResults("impala", StrCat("create table ",
table_name_, "(id INT, name STRING, first_sold TIMESTAMP, "
"last_sold TIMESTAMP, price DECIMAL(30, 2)) partitioned by (category INT)"),
- &query_id));
+ TQueryOptions(), false, &query_id));
assertQueryState(query_id, QUERY_STATE_SUCCESS);
// Insert some products that have a last_sold time.
@@ -161,7 +162,8 @@ class DatabaseTest {
}
}
- EXPECT_OK(impala_server_->ExecuteIgnoreResults("impala", sql1, &query_id));
+ EXPECT_OK(impala_server_->ExecuteIgnoreResults("impala", sql1, TQueryOptions(),
+ false, &query_id));
assertQueryState(query_id, QUERY_STATE_SUCCESS);
// Insert some products that do not have a last_sold time.
@@ -181,7 +183,8 @@ class DatabaseTest {
}
}
- EXPECT_OK(impala_server_->ExecuteIgnoreResults("impala", sql2, &query_id));
+ EXPECT_OK(impala_server_->ExecuteIgnoreResults("impala", sql2, TQueryOptions(),
+ false, &query_id));
assertQueryState(query_id, QUERY_STATE_SUCCESS);
}
}
@@ -278,7 +281,7 @@ TEST(InternalServerTest, MultipleQueriesMultipleSessions) {
// Insert a record into the test table using a new session.
ASSERT_OK(fixture->ExecuteIgnoreResults("impala", StrCat("insert into ",
test_table_name, "(id,first_name,last_name) VALUES (1,'test','person1')"),
- &query_id));
+ TQueryOptions(), false, &query_id));
assertQueryState(query_id, QUERY_STATE_SUCCESS);
// Select a record from the test table using a new session.
@@ -412,7 +415,7 @@ TEST(InternalServerTest, MissingClosingQuote) {
const string expected_msg = "ParseException: Unmatched string literal";
res = fixture->ExecuteIgnoreResults("impala",StrCat( "select * from ",
- db_test.GetTableName(), " where name = 'foo"), &query_id);
+ db_test.GetTableName(), " where name = 'foo"), TQueryOptions(), false, &query_id);
EXPECT_EQ(TErrorCode::GENERAL, res.code());
EXPECT_EQ(expected_msg, res.msg().msg().substr(0, expected_msg.length()));
EXPECT_EQ(TUniqueId(), query_id);
@@ -426,7 +429,7 @@ TEST(InternalServerTest, SyntaxError) {
const string expected_msg = "ParseException: Syntax error in line 1";
res = fixture->ExecuteIgnoreResults("impala", StrCat("select * from ",
- db_test.GetTableName(), "; select"), &query_id);
+ db_test.GetTableName(), "; select"), TQueryOptions(), false, &query_id);
EXPECT_EQ(TErrorCode::GENERAL, res.code());
EXPECT_EQ(expected_msg, res.msg().msg().substr(0, expected_msg.length()));
EXPECT_EQ(TUniqueId(), query_id);
@@ -438,7 +441,8 @@ TEST(InternalServerTest, UnclosedComment) {
Status res;
const string expected_msg = "ParseException: Syntax error in line 1";
- res = fixture->ExecuteIgnoreResults("impala", "select 1 /*foo", &query_id);
+ res = fixture->ExecuteIgnoreResults("impala", "select 1 /*foo", TQueryOptions(), false,
+ &query_id);
EXPECT_EQ(TErrorCode::GENERAL, res.code());
EXPECT_EQ(expected_msg, res.msg().msg().substr(0, expected_msg.length()));
EXPECT_EQ(TUniqueId(), query_id);
@@ -455,7 +459,7 @@ TEST(InternalServerTest, TableNotExist) {
ASSERT_OK(fixture->ExecuteIgnoreResults("impala", StrCat("drop table ",
db_test.GetTableName(), " purge")));
res = fixture->ExecuteIgnoreResults("impala", StrCat("select * from ",
- db_test.GetTableName()), &query_id);
+ db_test.GetTableName()), TQueryOptions(), false, &query_id);
EXPECT_EQ(TErrorCode::GENERAL, res.code());
EXPECT_EQ(expected_msg, res.msg().msg().substr(0, expected_msg.length()));
EXPECT_EQ(TUniqueId(), query_id);
diff --git a/be/src/service/internal-server.cc b/be/src/service/internal-server.cc
index ed1db08ff..829b79225 100644
--- a/be/src/service/internal-server.cc
+++ b/be/src/service/internal-server.cc
@@ -102,21 +102,25 @@ bool ImpalaServer::CloseSession(const TUniqueId& session_id) {
} // ImpalaServer::CloseSession
Status ImpalaServer::ExecuteIgnoreResults(const string& user_name, const string& sql,
- TUniqueId* query_id) {
+ const TQueryOptions& query_opts, const bool persist_in_db, TUniqueId* query_id) {
TUniqueId session_id;
TUniqueId internal_query_id;
+ Status result;
- RETURN_IF_ERROR(SubmitAndWait(user_name, sql, session_id, internal_query_id));
+ result = SubmitAndWait(user_name, sql, session_id, internal_query_id, query_opts,
+ persist_in_db);
if (query_id != nullptr) {
*query_id = internal_query_id;
}
- CloseQuery(internal_query_id);
+ if (!UUIDEmpty(internal_query_id)) {
+ CloseQuery(internal_query_id);
+ }
CloseSession(session_id);
- return Status::OK();
+ return result;
} //ImpalaServer::ExecuteIgnoreResults
Status ImpalaServer::ExecuteAndFetchAllText(const std::string& user_name,
@@ -124,26 +128,33 @@ Status ImpalaServer::ExecuteAndFetchAllText(const std::string& user_name,
TUniqueId* query_id){
TUniqueId session_id;
TUniqueId internal_query_id;
+ Status result;
- RETURN_IF_ERROR(SubmitAndWait(user_name, sql, session_id, internal_query_id));
+ result = SubmitAndWait(user_name, sql, session_id, internal_query_id, TQueryOptions());
if (query_id != nullptr) {
*query_id = internal_query_id;
}
- RETURN_IF_ERROR(FetchAllRows(internal_query_id, results, columns));
+ if (result.ok()) {
+ result = FetchAllRows(internal_query_id, results, columns);
+ }
+
+ if (!UUIDEmpty(internal_query_id)) {
+ CloseQuery(internal_query_id);
+ }
- CloseQuery(internal_query_id);
CloseSession(session_id);
- return Status::OK();
+ return result;
} // ImpalaServer::ExecuteAndFetchAllText
Status ImpalaServer::SubmitAndWait(const string& user_name, const string& sql,
- TUniqueId& new_session_id, TUniqueId& new_query_id) {
+ TUniqueId& new_session_id, TUniqueId& new_query_id, const TQueryOptions& query_opts,
+ const bool persist_in_db) {
- RETURN_IF_ERROR(OpenSession(user_name, new_session_id));
- RETURN_IF_ERROR(SubmitQuery(sql, new_session_id, new_query_id));
+ RETURN_IF_ERROR(OpenSession(user_name, new_session_id, query_opts));
+ RETURN_IF_ERROR(SubmitQuery(sql, new_session_id, new_query_id, persist_in_db));
return WaitForResults(new_query_id);
} // ImpalaServer::SubmitAndWait
@@ -168,7 +179,7 @@ Status ImpalaServer::WaitForResults(TUniqueId& query_id) {
} // ImpalaServer::WaitForResults
Status ImpalaServer::SubmitQuery(const string& sql, const TUniqueId& session_id,
- TUniqueId& new_query_id) {
+ TUniqueId& new_query_id, const bool persist_in_db) {
// build a query context
TQueryCtx query_context;
@@ -196,15 +207,13 @@ Status ImpalaServer::SubmitQuery(const string& sql, const TUniqueId& session_id,
AddPoolConfiguration(&query_context, ~set_query_options_mask);
QueryHandle query_handle;
- RETURN_IF_ERROR(Execute(&query_context, session_state, &query_handle, nullptr));
+ RETURN_IF_ERROR(Execute(&query_context, session_state, &query_handle, nullptr,
+ persist_in_db));
new_query_id = query_handle->query_id();
- RETURN_IF_ERROR(SetQueryInflight(session_state, query_handle));
-
- return Status::OK();
+ return SetQueryInflight(session_state, query_handle);
} // ImpalaServer::SubmitQuery
-
Status ImpalaServer::FetchAllRows(const TUniqueId& query_id, query_results& results,
results_columns* columns) {
QueryResultSet* result_set;
@@ -251,11 +260,19 @@ Status ImpalaServer::FetchAllRows(const TUniqueId& query_id, query_results& resu
RETURN_IF_ERROR(query_handle->FetchRows(10, result_set, block_wait_time));
results->insert(results->cend(), row_set.cbegin(), row_set.cend());
}
+
return Status::OK();
} // ImpalaServer::FetchAllRows
void ImpalaServer::CloseQuery(const TUniqueId& query_id) {
- UnregisterQueryDiscardResult(query_id, false);
+ QueryHandle query_handle;
+ Status found = GetActiveQueryHandle(query_id, &query_handle);
+
+ if (!found.ok()) {
+ return;
+ }
+
+ UnregisterQueryDiscardResult(query_handle->query_id(), false);
} // ImpalaServer::CloseQuery
void ImpalaServer::GetConnectionContextList(
diff --git a/be/src/service/internal-server.h b/be/src/service/internal-server.h
index b6bbf9405..aedaeb058 100644
--- a/be/src/service/internal-server.h
+++ b/be/src/service/internal-server.h
@@ -21,9 +21,10 @@
#include <string>
#include <vector>
+#include "common/status.h"
#include "gen-cpp/Query_types.h"
#include "gen-cpp/Types_types.h"
-#include "common/status.h"
+#include "rpc/thrift-server.h"
namespace impala {
@@ -87,18 +88,25 @@ namespace impala {
/// never retrieved.
///
/// Parameters:
- /// `user_name` Specifies the username that will be reported as running this
- /// query.
- /// `sql` Text of the sql query/ddl/dml to run.
- /// `query_id` Optional output parameter, if specified, it will be overwritten
- /// with the id of the query that was executed. Since the query is
- /// closed by this function, the query id is informational only.
+ /// `user_name` Specifies the username that will be reported as running this
+ /// query.
+ /// `sql` Text of the sql query/ddl/dml to run.
+ /// `query_opts` Optional, contains query options that will apply to all
+ /// queries executed during this session.
+ /// `persist_in_db` Optional boolean indicating if the query data should be
+ /// written to the completed queries table after it is closed.
+ /// Defaults to `true`.
+ /// `query_id` Optional output parameter, if specified, it will be
+ /// overwritten with the id of the query that was executed. Since
+ /// the query is closed by this function, the query id is
+ /// informational only.
///
/// Return:
/// `impala::Status` Indicates the result of submitting the query and waiting for
/// it to return.
virtual Status ExecuteIgnoreResults(const std::string& user_name,
- const std::string& sql, TUniqueId* query_id = nullptr) = 0;
+ const std::string& sql, const TQueryOptions& query_opts = TQueryOptions(),
+ const bool persist_in_db = true, TUniqueId* query_id = nullptr) = 0;
/// Creates a new session under the specified user and submits a query under that
/// session. No authentication is performed. Blocks until result rows are available.
@@ -140,15 +148,22 @@ namespace impala {
/// `user_name` Specifies the username that will be reported as running this
/// query.
/// `sql` Text of the sql query/ddl/dml to run.
- /// `new_session_id` Output parameter that will be set to the id of the newly
- /// created session.
- /// `new_query_id` Output parameter that will be set to the id of the newly
- // started query.
+ /// `new_session_id` Output parameter that will be set to the id of the
+ /// newly created session.
+ /// `new_query_id` Output parameter that will be set to the id of the
+ /// newly started query.
+ /// `query_opts` Optional, contains query options that will apply to all
+ /// queries executed by this session opened by this function.
+ /// `persist_in_db` Optional boolean indicating if the query data should be
+ /// written to the completed queries table after it is closed.
+ /// Defaults to `true`.
///
/// Return:
/// `impala::Status` Indicates the result of submitting and waiting for the query.
- virtual Status SubmitAndWait(const std::string& user_name,
- const std::string& sql, TUniqueId& new_session_id, TUniqueId& new_query_id) = 0;
+ virtual Status SubmitAndWait(const std::string& user_name, const std::string& sql,
+ TUniqueId& new_session_id, TUniqueId& new_query_id,
+ const TQueryOptions& query_opts = TQueryOptions(),
+ const bool persist_in_db = true) = 0;
/// Waits until the given query has results available.
///
@@ -162,15 +177,19 @@ namespace impala {
/// sets the query as in-flight before returning.
///
/// Parameters:
- /// `sql` Text of the sql query/ddl/dml to run.
- /// `session_id` Id of the session that will run the query.
- /// `new_query_id` Output parameter that will be set to the id of the newly
- // started query.
+ /// `sql` Text of the sql query/ddl/dml to run.
+ /// `session_id` Id of the session that will run the query.
+ /// `new_query_id` Output parameter that will be set to the id of the newly
+ /// started query.
+ /// `persist_in_db` Optional boolean indicating if the query data should be
+ /// written to the completed queries table after it is closed.
+ /// Defaults to `true`.
///
/// Return:
/// `impala::Status` Indicates the result of submitting the query.
virtual Status SubmitQuery(const std::string& sql,
- const impala::TUniqueId& session_id, TUniqueId& new_query_id) = 0;
+ const impala::TUniqueId& session_id, TUniqueId& new_query_id,
+ const bool persist_in_db = true) = 0;
/// Retrieves all result rows for a given query. The query must have already been
/// submitted and one of the Wait methods called on the query to ensure results are
diff --git a/be/src/service/query-state-record.cc b/be/src/service/query-state-record.cc
index 7bea6ac3b..0b594df83 100644
--- a/be/src/service/query-state-record.cc
+++ b/be/src/service/query-state-record.cc
@@ -201,10 +201,12 @@ QueryStateExpanded::QueryStateExpanded(const ClientRequestState& exec_state,
.query_exec_request.planner_per_host_mem_estimate;
dedicated_coord_mem_estimate = exec_state.exec_request()
.query_exec_request.dedicated_coord_mem_estimate;
- base->num_rows_fetched = exec_state.num_rows_fetched_counter();
row_materialization_rate = exec_state.row_materialization_rate();
row_materialization_time = exec_state.row_materialization_timer();
+ // Update name_rows_fetched with the final count after query close.
+ base_state->num_rows_fetched = exec_state.num_rows_fetched_counter();
+
// Fields from the schedule.
if (exec_state.schedule() != nullptr) {
// Per-Host Metrics
@@ -343,6 +345,29 @@ QueryStateExpanded::QueryStateExpanded(const ClientRequestState& exec_state,
}
executor_groups = exec_group_str.str();
boost::algorithm::trim_if(executor_groups, boost::algorithm::is_any_of("\n"));
+
+ // Find important events in the events timeline and store them in their own map.
+ for (const auto& event : EventsTimeline()) {
+ if (boost::algorithm::iequals(event.first, "planning finished")) {
+ events.insert_or_assign(PLANNING_FINISHED, event.second);
+ } else if (boost::algorithm::iequals(event.first, "submit for admission")) {
+ events.insert_or_assign(SUBMIT_FOR_ADMISSION, event.second);
+ } else if (boost::algorithm::iequals(event.first, "completed admission")) {
+ events.insert_or_assign(COMPLETED_ADMISSION, event.second);
+ } else if (boost::algorithm::istarts_with(event.first, "all ")
+ && boost::algorithm::icontains(event.first, " execution backends ")
+ && boost::algorithm::iends_with(event.first, " started")) {
+ events.insert_or_assign(ALL_BACKENDS_STARTED, event.second);
+ } else if (boost::algorithm::iequals(event.first, "rows available")) {
+ events.insert_or_assign(ROWS_AVAILABLE, event.second);
+ } else if (boost::algorithm::iequals(event.first, "first row fetched")) {
+ events.insert_or_assign(FIRST_ROW_FETCHED, event.second);
+ } else if (boost::algorithm::iequals(event.first, "last row fetched")) {
+ events.insert_or_assign(LAST_ROW_FETCHED, event.second);
+ } else if (boost::algorithm::iequals(event.first, "unregister query")) {
+ events.insert_or_assign(UNREGISTER_QUERY, event.second);
+ }
+ }
} // QueryStateExpanded constructor
bool QueryStateExpanded::events_timeline_empty() const {
diff --git a/be/src/service/query-state-record.h b/be/src/service/query-state-record.h
index e30714ed7..fec77f22c 100644
--- a/be/src/service/query-state-record.h
+++ b/be/src/service/query-state-record.h
@@ -19,6 +19,7 @@
#include <cstdint>
#include <map>
+#include <unordered_map>
#include <vector>
#include "gen-cpp/ExecStats_types.h"
@@ -229,6 +230,20 @@ private:
size_t cur_;
}; // class EventsTimelineIterator
+// Enum of all query events that are relevant to workload management.
+// Note: if adding to this enum, also add to the initialization of the events member of
+// the QueryStateExpanded struct.
+enum QueryEvent {
+ PLANNING_FINISHED,
+ SUBMIT_FOR_ADMISSION,
+ COMPLETED_ADMISSION,
+ ALL_BACKENDS_STARTED,
+ ROWS_AVAILABLE,
+ FIRST_ROW_FETCHED,
+ LAST_ROW_FETCHED,
+ UNREGISTER_QUERY
+};
+
/// Expanded snapshot of the query including its state along with other fields relevant
/// to workload management. Not mutated after construction.
struct QueryStateExpanded {
@@ -306,6 +321,19 @@ struct QueryStateExpanded {
/// Executor Groups
std::string executor_groups;
+ /// Events
+ /// Guaranteed to have one element for every member of the QueryEvent enum.
+ std::unordered_map<QueryEvent, std::int64_t> events = {
+ {PLANNING_FINISHED, 0},
+ {SUBMIT_FOR_ADMISSION, 0},
+ {COMPLETED_ADMISSION, 0},
+ {ALL_BACKENDS_STARTED, 0},
+ {ROWS_AVAILABLE, 0},
+ {FIRST_ROW_FETCHED, 0},
+ {LAST_ROW_FETCHED, 0},
+ {UNREGISTER_QUERY, 0}
+ };
+
/// Events Timeline Empty
bool events_timeline_empty() const;
diff --git a/be/src/service/workload-management-fields.cc b/be/src/service/workload-management-fields.cc
new file mode 100644
index 000000000..5f52d2bcf
--- /dev/null
+++ b/be/src/service/workload-management-fields.cc
@@ -0,0 +1,417 @@
+// 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.
+
+// This file contains the definition of the FIELD_DEFINITIONS list from the associated
+// header file. Each field definition consists of the database column name for the field,
+// the sql type of the database column, and a function that extracts the actual value from
+// a `QueryStateExpanded` instance and writes it to the stream that is collecting all the
+// values for the insert dml.
+
+#include "service/workload-management.h"
+
+#include <algorithm>
+#include <string>
+#include <utility>
+
+#include <boost/algorithm/string.hpp>
+#include <gflags/gflags_declare.h>
+#include <glog/logging.h>
+
+#include "common/compiler-util.h"
+#include "gen-cpp/Types_types.h"
+#include "runtime/exec-env.h"
+#include "service/query-options.h"
+#include "service/query-state-record.h"
+#include "util/debug-util.h"
+#include "util/network-util.h"
+#include "util/sql-util.h"
+
+DECLARE_int32(query_log_max_sql_length);
+DECLARE_int32(query_log_max_plan_length);
+
+namespace impala {
+
+namespace workload_management {
+
+/// Helper type for event timeline timestamp functions.
+using _event_compare_pred = std::function<bool(const std::string& comp)>;
+
+/// Helper function to write the timestamp for a single event from the events timeline
+/// into the completed queries sql statement stream.
+///
+/// Parameters:
+/// `ctx` The field parse context object.
+/// `target_event` The element from the QueryEvent enum that represents the event being
+/// inserted into the sql statement. The corresponding event value will
+/// be retrieved from the map of query events.
+static void _write_event(FieldParserContext& ctx, QueryEvent target_event) {
+ const auto& event = ctx.record->events.find(target_event);
+ DCHECK(event != ctx.record->events.end());
+ ctx.sql << event->second;
+}
+
+const std::list<FieldDefinition> FIELD_DEFINITIONS = {
+ // Cluster Id
+ // Required
+ FieldDefinition("cluster_id", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.cluster_id << "'";
+ }),
+
+ // Query Id
+ FieldDefinition("query_id", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << PrintId(ctx.record->base_state->id) << "'";
+ }),
+
+ // Session Id
+ FieldDefinition("session_id", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << PrintId(ctx.record->session_id) << "'";
+ }),
+
+ // Session Type
+ FieldDefinition("session_type", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->session_type << "'";
+ }),
+
+ // Hiveserver2 Protocol Version
+ FieldDefinition("hiveserver2_protocol_version", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'";
+ if (ctx.record->session_type == TSessionType::HIVESERVER2) {
+ ctx.sql << ctx.record->hiveserver2_protocol_version;
+ }
+ ctx.sql << "'";
+ }),
+
+ // Effective User
+ FieldDefinition("db_user", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->base_state->effective_user << "'";
+ }),
+
+ // DB User
+ FieldDefinition("db_user_connection", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->db_user_connection << "'";
+ }),
+
+ // Default DB
+ FieldDefinition("db_name", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->base_state->default_db << "'";
+ }),
+
+ // Impala Coordinator
+ FieldDefinition("impala_coordinator", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" <<TNetworkAddressToString(
+ ExecEnv::GetInstance()->configured_backend_address()) << "'";
+ }),
+
+ // Query Status
+ FieldDefinition("query_status", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'";
+ if (ctx.record->base_state->query_status.ok()) {
+ ctx.sql << "OK";
+ } else {
+ ctx.sql << EscapeSql(ctx.record->base_state->query_status.msg().msg());
+ }
+ ctx.sql << "'";
+ }),
+
+ // Query State
+ FieldDefinition("query_state", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->base_state->query_state << "'";
+ }),
+
+ // Impala Query End State
+ FieldDefinition("impala_query_end_state",
+ TPrimitiveType::STRING, [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->impala_query_end_state << "'";
+ }),
+
+ // Query Type
+ FieldDefinition("query_type", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->base_state->stmt_type << "'";
+ }),
+
+ // Client Network Address
+ FieldDefinition("network_address", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << TNetworkAddressToString(ctx.record->client_address) << "'";
+ }),
+
+ // Query Start Time in UTC
+ // Required
+ FieldDefinition("start_time_utc", TPrimitiveType::TIMESTAMP,
+ [](FieldParserContext& ctx){
+ ctx.sql << "UNIX_MICROS_TO_UTC_TIMESTAMP(" <<
+ ctx.record->base_state->start_time_us << ")";
+ }),
+
+ // Query Duration
+ FieldDefinition("total_time_ns", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << (ctx.record->base_state->end_time_us -
+ ctx.record->base_state->start_time_us) * 1000;
+ }),
+
+ // Query Options set by Configuration
+ FieldDefinition("query_opts_config", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ const std::string opts_str = DebugQueryOptions(ctx.record->query_options);
+ ctx.sql << "'" << EscapeSql(opts_str) << "'";
+ }),
+
+ // Resource Pool
+ FieldDefinition("resource_pool", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << EscapeSql(ctx.record->base_state->resource_pool) << "'";
+ }),
+
+ // Per-host Memory Estimate
+ FieldDefinition("per_host_mem_estimate", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->per_host_mem_estimate;
+ }),
+
+ // Dedicated Coordinator Memory Estimate
+ FieldDefinition("dedicated_coord_mem_estimate", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->dedicated_coord_mem_estimate;
+ }),
+
+ // Per-Host Fragment Instances
+ FieldDefinition("per_host_fragment_instances", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'";
+
+ if (!ctx.record->per_host_state.empty()) {
+ for (const auto& iter : ctx.record->per_host_state) {
+ ctx.sql << TNetworkAddressToString(iter.first) << "=" <<
+ iter.second.fragment_instance_count << ',';
+ }
+ ctx.sql.move_back();
+ }
+
+ ctx.sql << "'";
+ }),
+
+ // Backends Count
+ FieldDefinition("backends_count", TPrimitiveType::INT,
+ [](FieldParserContext& ctx){
+ if (ctx.record->per_host_state.empty()) {
+ ctx.sql << 0;
+ } else {
+ ctx.sql << ctx.record->per_host_state.size();
+ }
+ }),
+
+ // Admission Result
+ FieldDefinition("admission_result", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->admission_result << "'";
+ }),
+
+ // Cluster Memory Admitted
+ FieldDefinition("cluster_memory_admitted", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->base_state->cluster_mem_est;
+ }),
+
+ // Executor Group
+ FieldDefinition("executor_group", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << ctx.record->executor_group << "'";
+ }),
+
+ // Executor Groups
+ FieldDefinition("executor_groups", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << EscapeSql(ctx.record->executor_groups) << "'";
+ }),
+
+ // Exec Summary (also known as the operator summary)
+ FieldDefinition("exec_summary", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" << EscapeSql(ctx.record->exec_summary) << "'";
+ }),
+
+ // Number of rows fetched
+ FieldDefinition("num_rows_fetched", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->base_state->num_rows_fetched;
+ }),
+
+ // Row Materialization Rate
+ FieldDefinition("row_materialization_rows_per_sec", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->row_materialization_rate;
+ }),
+
+ // Row Materialization Time
+ FieldDefinition("row_materialization_time_ns", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->row_materialization_time;
+ }),
+
+ // Compressed Bytes Spilled to Disk
+ FieldDefinition("compressed_bytes_spilled", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->compressed_bytes_spilled;
+ }),
+
+ // Events Timeline Planning Finished
+ FieldDefinition("event_planning_finished", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, PLANNING_FINISHED);
+ }),
+
+ // Events Timeline Submit for Admission
+ FieldDefinition("event_submit_for_admission", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, SUBMIT_FOR_ADMISSION);
+ }),
+
+ // Events Timeline Completed Admission
+ FieldDefinition("event_completed_admission", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, COMPLETED_ADMISSION);
+ }),
+
+ // Events Timeline All Execution Backends Started
+ FieldDefinition("event_all_backends_started", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, ALL_BACKENDS_STARTED);
+ }),
+
+ // Events Timeline Rows Available
+ FieldDefinition("event_rows_available", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, ROWS_AVAILABLE);
+ }),
+
+ // Events Timeline First Row Fetched
+ FieldDefinition("event_first_row_fetched", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, FIRST_ROW_FETCHED);
+ }),
+
+ // Events Timeline Last Row Fetched
+ FieldDefinition("event_last_row_fetched", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, LAST_ROW_FETCHED);
+ }),
+
+ // Events Timeline Unregister Query
+ FieldDefinition("event_unregister_query", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ _write_event(ctx, UNREGISTER_QUERY);
+ }),
+
+ // Read IO Wait Time Total
+ FieldDefinition("read_io_wait_total_ns", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->read_io_wait_time_total;
+ }),
+
+ // Read IO Wait Time Mean
+ FieldDefinition("read_io_wait_mean_ns", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->read_io_wait_time_mean;
+ }),
+
+ // Bytes Read from the Data Cache Total
+ FieldDefinition("bytes_read_cache_total", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->bytes_read_cache_total;
+ }),
+
+ // Bytes Read Total
+ FieldDefinition("bytes_read_total", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ ctx.sql << ctx.record->bytes_read_total;
+ }),
+
+ // Per-Node Peak Memory Usage Min
+ FieldDefinition("pernode_peak_mem_min", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ auto min_elem = min_element(ctx.record->per_host_state.cbegin(),
+ ctx.record->per_host_state.cend(), PerHostPeakMemoryComparator);
+
+ if (LIKELY(min_elem != ctx.record->per_host_state.cend())) {
+ ctx.sql << min_elem->second.peak_memory_usage;
+ } else {
+ ctx.sql << 0;
+ }
+ }),
+
+ // Per-Node Peak Memory Usage Max
+ FieldDefinition("pernode_peak_mem_max", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ auto max_elem = max_element(ctx.record->per_host_state.cbegin(),
+ ctx.record->per_host_state.cend(), PerHostPeakMemoryComparator);
+
+ if (UNLIKELY(max_elem == ctx.record->per_host_state.cend())) {
+ ctx.sql << 0;
+ } else {
+ ctx.sql << max_elem->second.peak_memory_usage;
+ }
+ }),
+
+ // Per-Node Peak Memory Usage Mean
+ FieldDefinition("pernode_peak_mem_mean", TPrimitiveType::BIGINT,
+ [](FieldParserContext& ctx){
+ int64_t calc_mean = 0;
+
+ if (LIKELY(!ctx.record->per_host_state.empty())) {
+ for (const auto& host : ctx.record->per_host_state) {
+ calc_mean += host.second.peak_memory_usage;
+ }
+
+ calc_mean = calc_mean / ctx.record->per_host_state.size();
+ }
+
+ ctx.sql << calc_mean;
+ }),
+
+ // SQL Statement
+ FieldDefinition("sql", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'" <<
+ EscapeSql(ctx.record->redacted_sql, FLAGS_query_log_max_sql_length) << "'";
+ }),
+
+ // Query Plan
+ FieldDefinition("plan", TPrimitiveType::STRING,
+ [](FieldParserContext& ctx){
+ ctx.sql << "'"
+ << EscapeSql(ctx.record->base_state->plan, FLAGS_query_log_max_plan_length)
+ << "'";
+ }),
+
+ }; // FIELDS_PARSERS constant list
+
+} //namespace workload_management
+
+} // namespace impala
diff --git a/be/src/service/workload-management-flags.cc b/be/src/service/workload-management-flags.cc
new file mode 100644
index 000000000..7bd6249ba
--- /dev/null
+++ b/be/src/service/workload-management-flags.cc
@@ -0,0 +1,172 @@
+// 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 <regex>
+#include <string>
+
+#include <gflags/gflags.h>
+#include <glog/logging.h>
+
+using namespace std;
+
+static regex alphanum_underscore_dash("^[A-Za-z0-9\\-_]+$");
+
+DEFINE_bool(enable_workload_mgmt, false,
+ "Specifies if Impala will automatically write completed queries in the query log "
+ "table. If this value is set to true and then later removed, the query log table "
+ "will remain intact and accessible.");
+
+DEFINE_string_hidden(query_log_table_name, "impala_query_log", "Specifies the name of "
+ "the query log table where completed queries will be stored.");
+
+DEFINE_validator(query_log_table_name, [](const char* name, const string& val) {
+ if (regex_match(val, alphanum_underscore_dash)) return true;
+
+ LOG(ERROR) << "Invalid value for --" << name << ": must only contain alphanumeric "
+ "characters, underscores, or dashes and must not be empty";
+ return false;
+});
+
+DEFINE_string_hidden(query_log_table_location, "", "Specifies the location of the query "
+ "log table where completed queries will be stored.");
+
+DEFINE_validator(query_log_table_location, [](const char* name, const string& val) {
+ if (val.find_first_of('\'') != string::npos) {
+ LOG(ERROR) << "Invalid value for --" << name << ": must not contain single quotes.";
+ return false;
+ }
+
+ if (val.find_first_of('"') != string::npos) {
+ LOG(ERROR) << "Invalid value for --" << name << ": must not contain double quotes.";
+ return false;
+ }
+
+ if (val.find_first_of('\n') != string::npos) {
+ LOG(ERROR) << "Invalid value for --" << name << ": must not contain newlines.";
+ return false;
+ }
+
+ return true;
+});
+
+DEFINE_int32(query_log_write_interval_s, 300, "Number of seconds to wait between "
+ "batches of inserts to the query log table. The countdown to the next write starts "
+ "immediately when a write begins, but a new write will not start until the prior "
+ "write has completed. Min value is 1. Max value is 14400.");
+
+DEFINE_validator(query_log_write_interval_s, [](const char* name, int32_t val) {
+ if (val > 0 && val <= 14400) return true;
+ LOG(ERROR) << "Invalid value for --" << name << ": must be greater than 0 and less "
+ "than or equal to 14,400.";
+ return false;
+});
+
+DEFINE_int32_hidden(query_log_write_timeout_s, 0, "Specifies the query timeout in "
+ "seconds for inserts to the query log table. A value less than 1 indicates to use "
+ "the same value as the query_log_write_interval_s flag.");
+
+DEFINE_int32(query_log_max_queued, 100000, "Maximum number of records that can be queued "
+ "before they are written to the impala query log table. This flag operates "
+ "independently of the 'query_log_write_interval_s' flag. If the number of queued "
+ "records reaches this value, the records will be written to the query log table no "
+ "matter how much time has passed since the last write. The countdown to the next "
+ "write (based on the time period defined in the 'query_log_write_interval_s' flag) "
+ "is not restarted.");
+
+DEFINE_validator(query_log_max_queued, [](const char* name, int32_t val) {
+ if (val >= 0) return true;
+ LOG(ERROR) << "Invalid value for --" << name << ": must be greater than or equal to 0";
+ return false;
+});
+
+DEFINE_string_hidden(workload_mgmt_user, "impala", "Specifies the user that will be used "
+ "to create, update, and insert records into the query log table.");
+
+DEFINE_validator(workload_mgmt_user, [](const char* name, const string& val) {
+ if (FLAGS_enable_workload_mgmt && !regex_match(val, alphanum_underscore_dash)) {
+ LOG(ERROR) << "Invalid value for --" << name << ": must be a valid user when "
+ "workload management is enabled and must only contain alphanumeric characters, "
+ "underscores, or dashes";
+ return false;
+ }
+
+ return true;
+});
+
+DEFINE_int32(query_log_max_sql_length, 16777216, "Maximum length of a sql statement that "
+ "will be recorded in the completed queries table. If a sql statement with a length "
+ "longer than this value is executed, the sql inserted into the completed queries "
+ "table will be trimmed to this length. Any characters that need escaping will have "
+ "their backslash character counted towards this limit.");
+
+DEFINE_validator(query_log_max_sql_length, [](const char* name, int32_t val) {
+ if (val >= 0) return true;
+ LOG(ERROR) << "Invalid value for --" << name << ": must be greater than or equal to 0";
+ return false;
+});
+
+DEFINE_int32(query_log_max_plan_length, 16777216, "Maximum length of the sql plan that "
+ "will be recorded in the completed queries table. If a plan has a length longer than "
+ "this value, the plan inserted into the completed queries table will be trimmed to "
+ "this length. Any characters that need escaping will have their backslash character "
+ "counted towards this limit.");
+
+DEFINE_validator(query_log_max_plan_length, [](const char* name, int32_t val) {
+ if (val >= 0) return true;
+ LOG(ERROR) << "Invalid value for --" << name << ": must be greater than or equal to 0";
+ return false;
+});
+
+DEFINE_int32_hidden(query_log_shutdown_timeout_s, 30, "Number of seconds to wait for "
+ "the queue of completed queries to be drained to the query log table before timing "
+ "out and continuing the shutdown process. The completed queries drain process runs "
+ "after the shutdown process completes, thus the max shutdown time is extended by the "
+ "value specified in this flag.");
+
+DEFINE_validator(query_log_shutdown_timeout_s, [](const char* name, int32_t val) {
+ if (val >= 0) return true;
+ LOG(ERROR) << "Invalid value for --" << name << ": must be a positive value";
+ return false;
+});
+
+DEFINE_string(cluster_id, "", "Specifies an identifier string that uniquely represents "
+ "this cluster. This identifier is included in the query log table if enabled.");
+
+DEFINE_validator(cluster_id, [](const char* name, const string& val) {
+ if (val.length() == 0 || regex_match(val, alphanum_underscore_dash)) return true;
+
+ LOG(ERROR) << "Invalid value for --" << name << ": must only contain alphanumeric "
+ "characters, underscores, or dashes and must not be empty";
+ return false;
+});
+
+DEFINE_int32_hidden(query_log_max_insert_attempts, 3, "Maximum number of times to "
+ "attempt to insert a record into the completed queries table before abandining it.");
+
+DEFINE_validator(query_log_max_insert_attempts, [](const char* name, int32_t val) {
+ if (val > 0) return true;
+ LOG(ERROR) << "Invalid value for --" << name << ": must be greater than 1";
+ return false;
+});
+
+DEFINE_string(query_log_request_pool, "", "Specifies a pool or queue used by the queries "
+ "that insert into the query log table. Empty value causes no pool to be set.");
+
+DEFINE_string_hidden(query_log_table_props, "", "Comma separated list of additional "
+ "Iceberg table properties in the format 'key'='value' to apply when creating the "
+ "query log table. Only applies when the table is being created. After table "
+ "creation, this property does nothing");
diff --git a/be/src/service/workload-management.cc b/be/src/service/workload-management.cc
new file mode 100644
index 000000000..8545eb3e8
--- /dev/null
+++ b/be/src/service/workload-management.cc
@@ -0,0 +1,418 @@
+// 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 "service/workload-management.h"
+
+#include <chrono>
+#include <limits>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <utility>
+
+#include <gflags/gflags_declare.h>
+#include <glog/logging.h>
+#include <gutil/strings/strcat.h>
+
+#include "common/compiler-util.h"
+#include "common/logging.h"
+#include "common/status.h"
+#include "gen-cpp/Frontend_types.h"
+#include "gen-cpp/Query_types.h"
+#include "gen-cpp/Types_types.h"
+#include "runtime/query-driver.h"
+#include "service/client-request-state.h"
+#include "service/impala-server.h"
+#include "service/internal-server.h"
+#include "service/query-state-record.h"
+#include "util/debug-util.h"
+#include "util/impalad-metrics.h"
+#include "util/metrics.h"
+#include "util/string-util.h"
+#include "util/thread.h"
+#include "util/ticker.h"
+
+using namespace impala;
+using namespace impala::workload_management;
+using namespace std;
+
+DECLARE_bool(enable_workload_mgmt);
+DECLARE_string(query_log_table_name);
+DECLARE_string(query_log_table_location);
+DECLARE_int32(query_log_write_interval_s);
+DECLARE_int32(query_log_write_timeout_s);
+DECLARE_int32(query_log_max_queued);
+DECLARE_string(workload_mgmt_user);
+DECLARE_int32(query_log_max_sql_length);
+DECLARE_int32(query_log_max_plan_length);
+DECLARE_int32(query_log_shutdown_timeout_s);
+DECLARE_string(cluster_id);
+DECLARE_int32(query_log_max_insert_attempts);
+DECLARE_string(query_log_request_pool);
+DECLARE_string(query_log_table_props);
+
+namespace impala {
+
+/// Name of the database where all workload management tables will be stored.
+static const string DB = "sys";
+
+/// Default query options that will be provided on all queries that insert rows into the
+/// completed queries table. See the initialization code in the
+/// ImpalaServer::CompletedQueriesThread function for details on which options are set.
+static TQueryOptions insert_query_opts;
+
+/// Non-values portion of the sql DML to insert records into the completed queries table.
+/// Generates the first portion of the DML that inserts records into the completed queries
+/// table. This portion of the statement is constant and thus is only generated once.
+static string _insert_dml;
+
+/// Determine if the maximum number of queued completed queries has been exceeded.
+///
+/// Return:
+/// `true` There is a max limit on the number of queued completed queries and that
+/// limit has been exceeded.
+/// `false` Either there is no max number of queued completed queries or there is a
+/// limit that has not been exceeded.
+static inline bool MaxRecordsExceeded(size_t record_count) noexcept {
+ return FLAGS_query_log_max_queued > 0 && record_count > FLAGS_query_log_max_queued;
+} // function MaxRecordsExceeded
+
+/// Sets up the completed queries database and table by generating and executing the
+/// necessary DML statements.
+static const Status SetupDbTable(InternalServer* server, const string& table_name) {
+ insert_query_opts.__set_sync_ddl(true);
+
+ RETURN_IF_ERROR(server->ExecuteIgnoreResults(FLAGS_workload_mgmt_user,
+ StrCat("create database if not exists ", DB, " comment "
+ "'System database for Impala introspection'"), insert_query_opts, false));
+
+ StringStreamPop create_table_sql;
+ create_table_sql << "CREATE TABLE IF NOT EXISTS " << table_name << "(";
+
+ for (const auto& field : FIELD_DEFINITIONS) {
+ create_table_sql << field.db_column_name << " " << field.db_column_type << ",";
+ }
+ create_table_sql.move_back();
+
+ create_table_sql << ") PARTITIONED BY SPEC(identity(cluster_id), HOUR(start_time_utc)) "
+ << "STORED AS iceberg ";
+
+ if (!FLAGS_query_log_table_location.empty()) {
+ create_table_sql << "LOCATION '" << FLAGS_query_log_table_location << "' ";
+ }
+
+ create_table_sql << "TBLPROPERTIES ('schema_version'='1.0.0','format-version'='2'";
+
+ if (!FLAGS_query_log_table_props.empty()) {
+ create_table_sql << "," << FLAGS_query_log_table_props;
+ }
+
+ create_table_sql << ")";
+
+ RETURN_IF_ERROR(server->ExecuteIgnoreResults(FLAGS_workload_mgmt_user,
+ create_table_sql.str(), insert_query_opts, false));
+
+ insert_query_opts.__set_sync_ddl(false);
+
+ LOG(INFO) << "Completed query log initialization. storage_type=\""
+ << FLAGS_enable_workload_mgmt << "\" write_interval=\"" <<
+ FLAGS_query_log_write_interval_s << "s\"";
+
+ return Status::OK();
+} // function SetupDbTable
+
+/// Iterates through the list of field in `FIELDS_PARSERS` executing each parser for the
+/// given `QueryStateExpanded` object. This function builds the `FieldParserContext`
+/// object that is passed to each parser.
+///
+/// Parameters:
+/// `rec` - `QueryStateExpanded` object, an insert sql statement will be generated to
+/// insert a row into the completed queries table representing the query in
+/// this object.
+///
+/// Return:
+/// `string` - Contains the insert sql statement.
+static const string QueryStateToSql(const QueryStateExpanded* rec) noexcept {
+ DCHECK(rec != nullptr);
+ StringStreamPop sql;
+ FieldParserContext ctx(rec, FLAGS_cluster_id, sql);
+
+ sql << "(";
+
+ for (const auto& field : FIELD_DEFINITIONS) {
+ field.parser(ctx);
+ sql << ",";
+ }
+
+ sql.move_back();
+ sql << ")";
+
+ return sql.str();
+} // function QueryStateToSql
+
+Status ImpalaServer::InitWorkloadManagement() {
+ if (FLAGS_enable_workload_mgmt) {
+ return Thread::Create("impala-server", "completed-queries",
+ bind<void>(&ImpalaServer::CompletedQueriesThread, this),
+ &completed_queries_thread_);
+ }
+
+ return Status::OK();
+} // ImpalaServer::InitWorkloadManagement
+
+void ImpalaServer::ShutdownWorkloadManagement() {
+ unique_lock<mutex> l(completed_queries_threadstate_mu_);
+ // If the completed queries thread is not yet running, then we don't need to give it a
+ // chance to flush the in-memory queue to the completed queries table.
+ if (completed_queries_thread_state_ == RUNNING) {
+ completed_queries_thread_state_ = SHUTTING_DOWN;
+ completed_queries_cv_.notify_all();
+ completed_queries_shutdown_cv_.wait_for(l,
+ chrono::seconds(FLAGS_query_log_shutdown_timeout_s),
+ [this]{ return completed_queries_thread_state_ == SHUTDOWN; });
+ }
+} // ImpalaServer::ShutdownWorkloadManagement
+
+void ImpalaServer::EnqueueCompletedQuery(const QueryHandle& query_handle,
+ const shared_ptr<QueryStateRecord> qs_rec) {
+
+ // Do not enqueue queries that are not written to the table or if workload management is
+ // not enabled.
+ if (query_handle->stmt_type() == TStmtType::SET
+ || !query_handle.query_driver()->IncludedInQueryLog()
+ || !FLAGS_enable_workload_mgmt){
+ return; // Note: early return
+ }
+
+ // Do not enqueue use and show ddl queries. This check is separate because combining it
+ // with the previous check resulted in very confusing code that had duplication.
+ if (query_handle->stmt_type() == TStmtType::DDL) {
+ switch (query_handle->catalog_op_type()) {
+ case TCatalogOpType::SHOW_TABLES:
+ case TCatalogOpType::SHOW_DBS:
+ case TCatalogOpType::SHOW_STATS:
+ case TCatalogOpType::USE:
+ case TCatalogOpType::SHOW_FUNCTIONS:
+ case TCatalogOpType::SHOW_CREATE_TABLE:
+ case TCatalogOpType::SHOW_DATA_SRCS:
+ case TCatalogOpType::SHOW_ROLES:
+ case TCatalogOpType::SHOW_GRANT_PRINCIPAL:
+ case TCatalogOpType::SHOW_FILES:
+ case TCatalogOpType::SHOW_CREATE_FUNCTION:
+ case TCatalogOpType::SHOW_VIEWS:
+ case TCatalogOpType::DESCRIBE_TABLE:
+ case TCatalogOpType::DESCRIBE_DB:
+ case TCatalogOpType::DESCRIBE_HISTORY:
+ return; // Note: early return
+ case TCatalogOpType::RESET_METADATA:
+ case TCatalogOpType::DDL:
+ break;
+ default:
+ LOG(FATAL) << "unknown ddl type: " << to_string(query_handle->catalog_op_type());
+ }
+ }
+
+ shared_ptr<QueryStateExpanded> exp_rec = make_shared<QueryStateExpanded>(*query_handle,
+ move(qs_rec));
+
+ {
+ lock_guard<mutex> l(completed_queries_lock_);
+ completed_queries_.emplace_back(CompletedQuery(move(exp_rec)));
+ ImpaladMetrics::COMPLETED_QUERIES_QUEUED->Increment(1L);
+
+ if (MaxRecordsExceeded(completed_queries_.size())) {
+ completed_queries_cv_.notify_all();
+ }
+ }
+} // ImpalaServer::EnqueueCompletedQuery
+
+void ImpalaServer::CompletedQueriesThread() {
+ {
+ lock_guard<mutex> l(completed_queries_threadstate_mu_);
+ completed_queries_thread_state_ = INITIALIZING;
+ }
+
+ // Setup default query options.
+ insert_query_opts.__set_timezone("UTC");
+ insert_query_opts.__set_query_timeout_s((FLAGS_query_log_write_timeout_s < 1 ?
+ FLAGS_query_log_write_interval_s : FLAGS_query_log_write_timeout_s));
+ if (!FLAGS_query_log_request_pool.empty()) {
+ insert_query_opts.__set_request_pool(FLAGS_query_log_request_pool);
+ }
+
+ // Fully qualified table name based on startup flags.
+ const string table_name = StrCat(DB, ".", FLAGS_query_log_table_name);
+
+ // Non-values portion of the completed queries insert dml. Does not change across
+ // queries.
+ StringStreamPop fields;
+ fields << "INSERT INTO " << table_name << "(";
+ for (const auto& field : FIELD_DEFINITIONS) {
+ fields << field.db_column_name << ",";
+ }
+ fields.move_back();
+ fields << ") VALUES ";
+ _insert_dml = fields.str();
+
+ // The initialization code only works when run in a separate thread for reasons unknown.
+ ABORT_IF_ERROR(SetupDbTable(internal_server_.get(), table_name));
+
+ {
+ lock_guard<mutex> l(completed_queries_threadstate_mu_);
+ // This condition will evaluate to false only if a clean shutdown was initiated while
+ // the previous function was running.
+ if (LIKELY(completed_queries_thread_state_ == INITIALIZING)) {
+ completed_queries_thread_state_ = RUNNING;
+ } else {
+ return; // Note: early return
+ }
+
+ completed_queries_ticker_ = make_unique<TickerSecondsBool>(
+ FLAGS_query_log_write_interval_s, completed_queries_cv_, completed_queries_lock_);
+ ABORT_IF_ERROR(completed_queries_ticker_->Start("impala-server",
+ "completed-queries-ticker"));
+ }
+
+ while (true) {
+ // Exit this thread if a shutdown was initiated.
+ {
+ lock_guard<mutex> l(completed_queries_threadstate_mu_);
+
+ DCHECK(completed_queries_thread_state_ != SHUTDOWN);
+
+ if (UNLIKELY(completed_queries_thread_state_ == SHUTTING_DOWN)) {
+ completed_queries_thread_state_ = SHUTDOWN;
+ completed_queries_shutdown_cv_.notify_all();
+ return; // Note: early return
+ }
+ }
+
+ // Sleep this thread until it is time to process queued completed queries. During the
+ // wait, the completed_queries_lock_ is only locked while calling the lambda function
+ // predicate. After waking up, the completed_queries_lock_ will be locked.
+ unique_lock<mutex> l(completed_queries_lock_);
+ completed_queries_cv_.wait(l,
+ [this]{
+ lock_guard<mutex> l2(completed_queries_threadstate_mu_);
+ // To guard against spurious wakeups, this predicate ensures there are completed
+ // queries queued up before waking up the thread.
+ return (completed_queries_ticker_->WakeupGuard()()
+ && !completed_queries_.empty())
+ || MaxRecordsExceeded(completed_queries_.size())
+ || UNLIKELY(completed_queries_thread_state_ == SHUTTING_DOWN);
+ });
+ completed_queries_ticker_->ResetWakeupGuard();
+
+ // transfer all currently queued completed queries to another list for processing
+ // so that the completed queries queue is not blocked while creating and executing the
+ // DML to insert into the query log table
+ if (!completed_queries_.empty()) {
+ if (MaxRecordsExceeded(completed_queries_.size())) {
+ ImpaladMetrics::COMPLETED_QUERIES_MAX_RECORDS_WRITES->Increment(1L);
+ } else {
+ ImpaladMetrics::COMPLETED_QUERIES_SCHEDULED_WRITES->Increment(1L);
+ }
+
+ // Copy all completed queries to a temporary list so that inserts to the
+ // completed_queries list are not blocked while generating and running an insert
+ // SQL statement for the completed queries.
+ list<CompletedQuery> queries_to_insert;
+ queries_to_insert.splice(queries_to_insert.cend(), completed_queries_);
+ completed_queries_lock_.unlock();
+
+ string sql;
+ uint32_t max_row_size = 0;
+
+ for (auto iter = queries_to_insert.begin(); iter != queries_to_insert.end();
+ iter++) {
+ if (iter->insert_attempts_count >= FLAGS_query_log_max_insert_attempts) {
+ LOG(ERROR) << "could not write completed query table=\"" << table_name <<
+ "\" query_id=\"" << PrintId(iter->query->base_state->id) << "\"";
+ iter = queries_to_insert.erase(iter);
+ ImpaladMetrics::COMPLETED_QUERIES_QUEUED->Increment(-1);
+ continue;
+ }
+
+ // Increment the count of attempts to insert this query into the completed
+ // queries table.
+ iter->insert_attempts_count += 1;
+
+ const string row = QueryStateToSql(iter->query.get());
+ if (row.size() > max_row_size) {
+ max_row_size = row.size();
+ }
+
+ StrAppend(&sql, move(row), ",");
+ }
+
+ DCHECK(ImpaladMetrics::COMPLETED_QUERIES_QUEUED->GetValue() >=
+ queries_to_insert.size());
+
+ // In the case where queries_to_insert only contains records that have exceeded
+ // the max insert attempts, sql will be empty.
+ if (LIKELY(!sql.empty())) {
+ // Remove the last comma and determine the final sql statement length.
+ sql.pop_back();
+ const size_t final_sql_len = _insert_dml.size() + sql.size();
+
+ TUniqueId tmp_query_id;
+
+ // Build query options to ensure the query is not rejected.
+ TQueryOptions opts = insert_query_opts;
+
+ if (UNLIKELY(final_sql_len > numeric_limits<int32_t>::max())) {
+ LOG(ERROR) << "Completed queries table insert sql statement of length '" <<
+ final_sql_len << "' was longer than the maximum of '" <<
+ numeric_limits<int32_t>::max() << "', skipping";
+ continue; // NOTE: early loop continuation
+ }
+
+ opts.__set_max_statement_length_bytes(final_sql_len < 1024 ? 1024 :
+ final_sql_len);
+ opts.__set_max_row_size(max_row_size);
+
+ // Execute the insert dml.
+ const Status ret_status = internal_server_->ExecuteIgnoreResults(
+ FLAGS_workload_mgmt_user, StrCat(_insert_dml, sql), opts, false,
+ &tmp_query_id);
+
+ if (ret_status.ok()) {
+ LOG(INFO) << "wrote completed queries table=\"" << table_name << "\" "
+ "record_count=\"" << queries_to_insert.size() << "\"";
+ ImpaladMetrics::COMPLETED_QUERIES_QUEUED->Increment(
+ queries_to_insert.size() * -1);
+ DCHECK(ImpaladMetrics::COMPLETED_QUERIES_QUEUED->GetValue() >= 0);
+ ImpaladMetrics::COMPLETED_QUERIES_WRITTEN->Increment(
+ queries_to_insert.size());
+ } else {
+ LOG(WARNING) << "failed to write completed queries table=\"" << table_name <<
+ "\" record_count=\"" << queries_to_insert.size() << "\"";
+ LOG(WARNING) << ret_status.GetDetail();
+ ImpaladMetrics::COMPLETED_QUERIES_FAIL->Increment(queries_to_insert.size());
+ completed_queries_lock_.lock();
+ completed_queries_.splice(
+ completed_queries_.cend(), queries_to_insert);
+ completed_queries_lock_.unlock();
+ }
+ }
+ }
+ }
+} // ImpalaServer::CompletedQueriesThread
+
+} // namespace impala
diff --git a/be/src/service/workload-management.h b/be/src/service/workload-management.h
new file mode 100644
index 000000000..8305e1e9b
--- /dev/null
+++ b/be/src/service/workload-management.h
@@ -0,0 +1,97 @@
+// 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 <list>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include <gflags/gflags.h>
+
+#include "gen-cpp/Types_types.h"
+#include "service/query-state-record.h"
+#include "util/string-util.h"
+
+namespace impala {
+
+namespace workload_management {
+
+/// Struct defining the context for generating the sql DML that inserts records into the
+/// completed queries table.
+struct FieldParserContext {
+ const QueryStateExpanded* record;
+ const std::string cluster_id;
+ StringStreamPop& sql;
+
+ FieldParserContext(const QueryStateExpanded* rec, const std::string& cluster_id,
+ StringStreamPop& s) : record(rec), cluster_id(cluster_id), sql(s) {}
+}; // struct FieldParserContext
+
+/// Type of a function that retrieves one piece of information from the context and writes
+/// it to the SQL statement that inserts rows into the completed queries table.
+using FieldParser = void (*)(FieldParserContext&);
+
+/// Contains all necessary information for the definition and parsing of a single field
+/// in workload management.
+struct FieldDefinition {
+ const std::string db_column_name;
+ const TPrimitiveType::type db_column_type;
+ const FieldParser parser;
+
+ FieldDefinition(const std::string db_col_name, const TPrimitiveType::type db_col_type,
+ const FieldParser fp) : db_column_name(std::move(db_col_name)),
+ db_column_type(std::move(db_col_type)), parser(std::move(fp)) {}
+}; // struct FieldDefinition
+
+/// This list is the main data structure for workload management. Each list entry
+/// contains the name of a column in the completed queries table, the type of that column,
+/// and an implementation of a FieldParser for generating the value of that column from a
+/// `QueryStateExpanded` object.
+extern const std::list<FieldDefinition> FIELD_DEFINITIONS;
+
+/// Track the state of the thread that processes the completed queries queue. Access to
+/// the ThreadState variable must only happen after taking a lock on the associated mutex.
+/// Can be used to track the lifecycle of a thread.
+enum ThreadState {
+ NOT_STARTED,
+ INITIALIZING,
+ RUNNING,
+ SHUTTING_DOWN,
+ SHUTDOWN
+};
+
+// Represents one query that has completed.
+struct CompletedQuery {
+ // Contains information about the completed query.
+ const std::shared_ptr<QueryStateExpanded> query;
+
+ // Count of the number of times the completed query has attempted to be inserted into
+ // the completed queries table. he count is tracked so that the number of attempts can
+ // be limited and failing inserts do not retry indefinitely.
+ uint8_t insert_attempts_count;
+
+ CompletedQuery(const std::shared_ptr<QueryStateExpanded> query) :
+ query(std::move(query)) {
+ insert_attempts_count = 0;
+ }
+};
+
+} //namespace workload_management
+
+} // namespace impala
diff --git a/be/src/util/CMakeLists.txt b/be/src/util/CMakeLists.txt
index ab913f027..bc77b9291 100644
--- a/be/src/util/CMakeLists.txt
+++ b/be/src/util/CMakeLists.txt
@@ -98,6 +98,7 @@ set(UTIL_SRCS
runtime-profile.cc
sharded-query-map-util.cc
simple-logger.cc
+ sql-util.cc
string-parser.cc
string-util.cc
symbols-util.cc
@@ -207,6 +208,7 @@ add_library(UtilTests STATIC
rle-test.cc
runtime-profile-test.cc
simple-logger-test.cc
+ sql-util-test.cc
string-parser-test.cc
string-util-test.cc
summary-util-test.cc
@@ -270,6 +272,7 @@ ADD_UNIFIED_BE_LSAN_TEST(redactor-unconfigured-test "RedactorUnconfigTest.*")
ADD_UNIFIED_BE_LSAN_TEST(rle-test "BitArray.*:RleTest.*")
ADD_UNIFIED_BE_LSAN_TEST(runtime-profile-test "CountersTest.*:TimerCounterTest.*:TimeSeriesCounterTest.*:VariousNumbers/TimeSeriesCounterResampleTest.*:ToThrift.*:ToJson.*")
ADD_UNIFIED_BE_LSAN_TEST(simple-logger-test "SimpleLoggerTest.*")
+ADD_UNIFIED_BE_LSAN_TEST(sql-util-test "SqlUtilTest.*")
ADD_UNIFIED_BE_LSAN_TEST(string-parser-test "StringToInt.*:StringToIntWithBase.*:StringToFloat.*:StringToBool.*:StringToDate.*")
ADD_UNIFIED_BE_LSAN_TEST(string-util-test "TruncateDownTest.*:TruncateUpTest.*:CommaSeparatedContainsTest.*:FindUtf8PosForwardTest.*:FindUtf8PosBackwardTest.*:RandomFindUtf8PosTest.*:StringStreamPopTest.*")
ADD_UNIFIED_BE_LSAN_TEST(summary-util-test "PrintTableTest.*")
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index 74675921f..cfc028b1d 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -121,6 +121,8 @@ DECLARE_bool(invalidate_metadata_on_event_processing_failure);
DECLARE_bool(invalidate_global_metadata_on_event_processing_failure);
DECLARE_string(inject_process_event_failure_event_types);
DECLARE_double(inject_process_event_failure_ratio);
+DECLARE_bool(enable_workload_mgmt);
+DECLARE_string(query_log_table_name);
// HS2 SAML2.0 configuration
// Defined here because TAG_FLAG caused issues in global-flags.cc
@@ -466,6 +468,8 @@ Status PopulateThriftBackendGflags(TBackendGflags& cfg) {
cfg.__set_inject_process_event_failure_event_types(
FLAGS_inject_process_event_failure_event_types);
cfg.__set_inject_process_event_failure_ratio(FLAGS_inject_process_event_failure_ratio);
+ cfg.__set_enable_workload_mgmt(FLAGS_enable_workload_mgmt);
+ cfg.__set_query_log_table_name(FLAGS_query_log_table_name);
return Status::OK();
}
diff --git a/be/src/util/impalad-metrics.cc b/be/src/util/impalad-metrics.cc
index 2ec4814bd..2079755a4 100644
--- a/be/src/util/impalad-metrics.cc
+++ b/be/src/util/impalad-metrics.cc
@@ -166,6 +166,16 @@ const char* ImpaladMetricKeys::HEDGED_READ_OPS =
"impala-server.hedged-read-ops";
const char* ImpaladMetricKeys::HEDGED_READ_OPS_WIN =
"impala-server.hedged-read-ops-win";
+const char* ImpaladMetricKeys::COMPLETED_QUERIES_QUEUED =
+ "impala-server.completed-queries.queued";
+const char* ImpaladMetricKeys::COMPLETED_QUERIES_WRITTEN =
+ "impala-server.completed-queries.written";
+const char* ImpaladMetricKeys::COMPLETED_QUERIES_FAIL =
+ "impala-server.completed-queries.failure";
+const char* ImpaladMetricKeys::COMPLETED_QUERIES_SCHEDULED_WRITES =
+ "impala-server.completed-queries.scheduled-writes";
+const char* ImpaladMetricKeys::COMPLETED_QUERIES_MAX_RECORDS_WRITES =
+ "impala-server.completed-queries.max-records-writes";
const char* ImpaladMetricKeys::DEBUG_ACTION_NUM_FAIL = "impala.debug_action.fail";
const char* ImpaladMetricKeys::QUERY_LOG_EST_TOTAL_BYTES =
"impala-server.query-log-est-total-bytes";
@@ -216,6 +226,10 @@ IntCounter* ImpaladMetrics::CATALOG_CACHE_MISS_COUNT = nullptr;
IntCounter* ImpaladMetrics::CATALOG_CACHE_REQUEST_COUNT = nullptr;
IntCounter* ImpaladMetrics::CATALOG_CACHE_TOTAL_LOAD_TIME = nullptr;
IntCounter* ImpaladMetrics::DEBUG_ACTION_NUM_FAIL = nullptr;
+IntCounter* ImpaladMetrics::COMPLETED_QUERIES_WRITTEN = nullptr;
+IntCounter* ImpaladMetrics::COMPLETED_QUERIES_FAIL = nullptr;
+IntCounter* ImpaladMetrics::COMPLETED_QUERIES_SCHEDULED_WRITES = nullptr;
+IntCounter* ImpaladMetrics::COMPLETED_QUERIES_MAX_RECORDS_WRITES = nullptr;
// Gauges
IntGauge* ImpaladMetrics::CATALOG_NUM_DBS = nullptr;
@@ -246,6 +260,7 @@ DoubleGauge* ImpaladMetrics::CATALOG_CACHE_LOAD_EXCEPTION_RATE = nullptr;
DoubleGauge* ImpaladMetrics::CATALOG_CACHE_MISS_RATE = nullptr;
DoubleGauge* ImpaladMetrics::CATALOG_CACHE_ENTRY_MEDIAN_SIZE = nullptr;
DoubleGauge* ImpaladMetrics::CATALOG_CACHE_ENTRY_99TH_SIZE = nullptr;
+IntGauge* ImpaladMetrics::COMPLETED_QUERIES_QUEUED = nullptr;
// Properties
BooleanProperty* ImpaladMetrics::CATALOG_READY = nullptr;
@@ -358,6 +373,16 @@ void ImpaladMetrics::CreateMetrics(MetricGroup* m) {
NUM_FILES_OPEN_FOR_INSERT = m->AddGauge(
ImpaladMetricKeys::NUM_FILES_OPEN_FOR_INSERT, 0);
+ // Initialize completed queries metrics.
+ COMPLETED_QUERIES_QUEUED = m->AddGauge(ImpaladMetricKeys::COMPLETED_QUERIES_QUEUED, 0);
+ COMPLETED_QUERIES_WRITTEN = m->AddCounter(
+ ImpaladMetricKeys::COMPLETED_QUERIES_WRITTEN, 0);
+ COMPLETED_QUERIES_FAIL = m->AddCounter(ImpaladMetricKeys::COMPLETED_QUERIES_FAIL, 0);
+ COMPLETED_QUERIES_SCHEDULED_WRITES = m->AddCounter(
+ ImpaladMetricKeys::COMPLETED_QUERIES_SCHEDULED_WRITES, 0);
+ COMPLETED_QUERIES_MAX_RECORDS_WRITES = m->AddCounter(
+ ImpaladMetricKeys::COMPLETED_QUERIES_MAX_RECORDS_WRITES, 0);
+
// Initialize IO mgr metrics
IO_MGR_METRICS = m->GetOrCreateChildGroup("io-mgr");
IO_MGR_NUM_OPEN_FILES = IO_MGR_METRICS->AddGauge(
diff --git a/be/src/util/impalad-metrics.h b/be/src/util/impalad-metrics.h
index 7170d752d..c30ff0e53 100644
--- a/be/src/util/impalad-metrics.h
+++ b/be/src/util/impalad-metrics.h
@@ -265,6 +265,25 @@ class ImpaladMetricKeys {
// Associated metric is modified in ImpalaServer::ArchiveQuery and must hold
// ImpalaServer::query_log_lock_ on modification.
static const char* QUERY_LOG_EST_TOTAL_BYTES;
+
+ /// The number of completed queries queued up and waiting to be written to the query
+ /// log table.
+ static const char* COMPLETED_QUERIES_QUEUED;
+
+ /// The number of completed queries successfully written to the query log table.
+ static const char* COMPLETED_QUERIES_WRITTEN;
+
+ /// The number of completed queries that failed to be written to the query log
+ /// table.
+ static const char* COMPLETED_QUERIES_FAIL;
+
+ /// Number of writes to the query log table that happened at the regularly scheduled
+ /// time.
+ static const char* COMPLETED_QUERIES_SCHEDULED_WRITES;
+
+ /// Number of writes to the query log table that happened because the max queued
+ /// completed queries records was reached.
+ static const char* COMPLETED_QUERIES_MAX_RECORDS_WRITES;
};
/// Global impalad-wide metrics. This is useful for objects that want to update metrics
@@ -319,6 +338,10 @@ class ImpaladMetrics {
static IntCounter* CATALOG_CACHE_REQUEST_COUNT;
static IntCounter* CATALOG_CACHE_TOTAL_LOAD_TIME;
static IntCounter* DEBUG_ACTION_NUM_FAIL;
+ static IntCounter* COMPLETED_QUERIES_WRITTEN;
+ static IntCounter* COMPLETED_QUERIES_FAIL;
+ static IntCounter* COMPLETED_QUERIES_SCHEDULED_WRITES;
+ static IntCounter* COMPLETED_QUERIES_MAX_RECORDS_WRITES;
// Gauges
static IntGauge* CATALOG_NUM_DBS;
@@ -349,6 +372,7 @@ class ImpaladMetrics {
static IntGauge* RESULTSET_CACHE_TOTAL_NUM_ROWS;
static IntGauge* RESULTSET_CACHE_TOTAL_BYTES;
static IntGauge* QUERY_LOG_EST_TOTAL_BYTES;
+ static IntGauge* COMPLETED_QUERIES_QUEUED;
// Properties
static BooleanProperty* CATALOG_READY;
diff --git a/be/src/util/sql-util-test.cc b/be/src/util/sql-util-test.cc
new file mode 100644
index 000000000..3193e0642
--- /dev/null
+++ b/be/src/util/sql-util-test.cc
@@ -0,0 +1,95 @@
+// 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 <string>
+
+#include "testutil/gtest-util.h"
+#include "util/sql-util.h"
+
+namespace impala {
+
+TEST(SqlUtilTest, EscapeSqlEmptyLenEqual) {
+ const string str = "";
+ EXPECT_EQ(str, EscapeSql(str, str.length()));
+}
+
+TEST(SqlUtilTest, EscapeSqlEmptyLenTooBig) {
+ const string str = "";
+ EXPECT_EQ(str, EscapeSql(str, str.length() + 1));
+}
+
+TEST(SqlUtilTest, EscapeSqlEmptyLenMissing) {
+ const string str = "";
+ EXPECT_EQ(str, EscapeSql(str));
+}
+
+TEST(SqlUtilTest, EscapeSqlNotEmptyLenZero) {
+ const string str = "select * from my_table where field1=1";
+ EXPECT_EQ("", EscapeSql(str, 0));
+}
+
+TEST(SqlUtilTest, EscapeSqlNoEscapedCharsLenEqual) {
+ const string str = "select * from my_table where field1=1";
+ EXPECT_EQ(str, EscapeSql(str, str.length()));
+}
+
+TEST(SqlUtilTest, EscapeSqlNoEscapedCharsLenTooBig) {
+ const string str = "select * from my_table where field1=1";
+ EXPECT_EQ(str, EscapeSql(str, str.length() + 1));
+}
+
+TEST(SqlUtilTest, EscapeSqlNoEscapedCharsLenSmaller) {
+ const string str = "select * from my_table where field1=1";
+ EXPECT_EQ("select", EscapeSql(str, 6));
+}
+
+TEST(SqlUtilTest, EscapeSqlNoEscapedCharsLenMissing) {
+ const string str = "select * from my_table where field1=1";
+ EXPECT_EQ(str, EscapeSql(str));
+}
+
+TEST(SqlUtilTest, EscapeSqlEscapedCharsLenEqual) {
+ const string str = ";select * from \"my_table\" where field1='\\val1'\n";
+ EXPECT_EQ("\\;select * from \"my_table\" where field1=\\'\\\\val1\\'\\n",
+ EscapeSql(str, str.length() + 5));
+}
+
+TEST(SqlUtilTest, EscapeSqlEscapedCharsLenTooBig) {
+ const string str = ";select * from \"my_table\" where field1='\\val1'\n";
+ EXPECT_EQ("\\;select * from \"my_table\" where field1=\\'\\\\val1\\'\\n",
+ EscapeSql(str, str.length() + 100));
+}
+
+TEST(SqlUtilTest, EscapeSqlEscapedCharsLenMissing) {
+ const string str = ";select * from \"my_table\" where field1='\\val1'\n";
+ EXPECT_EQ("\\;select * from \"my_table\" where field1=\\'\\\\val1\\'\\n",
+ EscapeSql(str));
+}
+
+TEST(SqlUtilTest, EscapeSqlEscapedCharsLenSmallerLastCharEscaped) {
+ const string str = ";select 'fld1', 'fld;2'\n from \"my_table\" where fld1='\\val1'\n";
+ EXPECT_EQ("\\;select \\'fld1\\', \\'fld\\;",
+ EscapeSql(str, 26));
+}
+
+TEST(SqlUtilTest, EscapeSqlEscapedCharsLenSmallerLastCharNotEscaped) {
+ const string str = ";select 'fld1', 'fld;2'\n from \"my_table\" where fld1='\\val1'\n";
+ EXPECT_EQ("\\;select \\'fld1",
+ EscapeSql(str, 15));
+}
+
+} // namespace impala
diff --git a/be/src/util/sql-util.cc b/be/src/util/sql-util.cc
new file mode 100644
index 000000000..e5e6f4f1c
--- /dev/null
+++ b/be/src/util/sql-util.cc
@@ -0,0 +1,70 @@
+// 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 "util/sql-util.h"
+
+#include<functional>
+#include <string>
+#include <sstream>
+#include <unordered_map>
+
+#include "common/compiler-util.h"
+
+using namespace std;
+
+namespace impala {
+
+const unordered_map<char, char> CHARS_TO_ESCAPE = {
+ {'\\', '\\'},
+ {'\'', '\''},
+ {'\n', 'n'},
+ {';', ';'}
+};
+
+const string EscapeSql(const string& sql, const int64_t& max_escaped_len) noexcept {
+ stringstream ret;
+ function<bool()> pred;
+
+ if (max_escaped_len == 0) {
+ return ""; // <-- Note: early return
+ }
+
+ for (const auto& c : sql) {
+ switch(c) {
+ case '\n':
+ ret << '\\';
+ ret << 'n';
+ break;
+ case '\\':
+ case '\'':
+ case ';':
+ ret << '\\';
+ default:
+ ret << c;
+ break;
+ }
+
+ if (UNLIKELY(max_escaped_len > 0
+ && static_cast<size_t>(ret.tellp() >= max_escaped_len))) {
+ break;
+ }
+ }
+
+ return ret.str();
+} // function EscapeSql
+
+} // namespace impala
diff --git a/be/src/util/sql-util.h b/be/src/util/sql-util.h
new file mode 100644
index 000000000..9464f21f8
--- /dev/null
+++ b/be/src/util/sql-util.h
@@ -0,0 +1,48 @@
+// 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 <string>
+#include <unordered_map>
+
+namespace impala {
+
+/// Map of characters that will be escaped if found in a sql string and the character to
+/// use as the replacement.
+extern const std::unordered_map<char, char> CHARS_TO_ESCAPE;
+
+constexpr char ESCAPE_CHAR = '\\';
+
+/// Performs escaping on a sql string by adding a backslash before all instances of the
+/// characters \, ', \n, and ; in a sql string. Double quotes are not escaped.
+///
+/// Parameters:
+/// sql - String containing a sql statement to escape.
+/// max_escaped_len - Optional, if specified, the string escaping will stop after this
+/// many characters. Note that this parameter limits the size of the
+/// returned escaped string instead of limiting the number of
+/// characters from the sql string that are processed. The returned
+/// string will have a length equal to this value except if the last
+/// character is escaped, then the returned string will be have a
+/// length one longer than this value.
+///
+/// Returns the escaped sql string.
+const std::string EscapeSql(const std::string& sql,
+ const std::int64_t& max_escaped_len = -1) noexcept;
+
+} // namespace impala
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index 09f62d1b7..650223fbf 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -288,4 +288,8 @@ struct TBackendGflags {
128: required string inject_process_event_failure_event_types
129: required double inject_process_event_failure_ratio
+
+ 130: required bool enable_workload_mgmt
+
+ 131: required string query_log_table_name
}
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index 0180b6ff9..7cfc5d19b 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -3743,5 +3743,55 @@
"units": "NONE",
"kind" : "GAUGE",
"key" : "catalogd.hms.cache.status.api.requests.15min"
+ },
+ {
+ "description": "The number of completed queries queued up and waiting to be written to the query log table.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Queued Completed Queries",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "impala-server.completed-queries.queued"
+ },
+ {
+ "description": "The number of completed queries successfully written to the query log table.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Completed Queries Successfully Written",
+ "units": "NONE",
+ "kind": "COUNTER",
+ "key": "impala-server.completed-queries.written"
+ },
+ {
+ "description": "The number of completed queries that failed to be written to the query log table. If the same query fails to be written multiple times, each attempt will increase this counter by 1.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Completed Queries Failed to Write",
+ "units": "NONE",
+ "kind": "COUNTER",
+ "key": "impala-server.completed-queries.failure"
+ },
+ {
+ "description": "Count of the times that completed queries were written to the query log table at the regularly scheduled time.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Completed Queries Write Attempts",
+ "units": "NONE",
+ "kind": "COUNTER",
+ "key": "impala-server.completed-queries.scheduled-writes"
+ },
+ {
+ "description": "Count of the times that completed queries were written to the query log table because the max records queued count was reached.",
+ "contexts": [
+ "IMPALAD"
+ ],
+ "label": "Max Records Hit",
+ "units": "NONE",
+ "kind": "COUNTER",
+ "key": "impala-server.completed-queries.max-records-writes"
}
]
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index 89026127f..62f95cd0d 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -28,6 +28,7 @@ import static org.apache.impala.thrift.TCatalogObjectType.TABLE;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -341,6 +342,8 @@ public class CatalogServiceCatalog extends Catalog {
private int numTables_ = 0;
private int numFunctions_ = 0;
+ private final List<String> impalaSysTables;
+
/**
* Initialize the CatalogServiceCatalog using a given MetastoreClientPool impl.
*
@@ -365,6 +368,7 @@ public class CatalogServiceCatalog extends Catalog {
.getBackendCfg().topic_update_tbl_max_wait_time_ms;
Preconditions.checkState(topicUpdateTblLockMaxWaitTimeMs_ >= 0,
"topic_update_tbl_max_wait_time_ms must be positive");
+ impalaSysTables = Arrays.asList(BackendConfig.INSTANCE.queryLogTableName());
tableLoadingMgr_ = new TableLoadingMgr(this, numLoadingThreads);
loadInBackground_ = loadInBackground;
try {
@@ -423,6 +427,10 @@ public class CatalogServiceCatalog extends Catalog {
*/
public boolean isBlacklistedDb(String dbName) {
Preconditions.checkNotNull(dbName);
+ if (BackendConfig.INSTANCE.enableWorkloadMgmt() && dbName.equalsIgnoreCase("sys")) {
+ // Override 'sys' for Impala system tables.
+ return false;
+ }
return blacklistedDbs_.contains(dbName.toLowerCase());
}
@@ -431,6 +439,10 @@ public class CatalogServiceCatalog extends Catalog {
*/
public boolean isBlacklistedTable(TableName table) {
Preconditions.checkNotNull(table);
+ if (table.getDb().equalsIgnoreCase("sys") && blacklistedDbs_.contains("sys")) {
+ // If we've overridden the database blacklist, only allow Impala system tables.
+ return !impalaSysTables.contains(table.getTbl());
+ }
return blacklistedTables_.contains(table);
}
diff --git a/fe/src/main/java/org/apache/impala/service/BackendConfig.java b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
index bbd684294..dae343c8f 100644
--- a/fe/src/main/java/org/apache/impala/service/BackendConfig.java
+++ b/fe/src/main/java/org/apache/impala/service/BackendConfig.java
@@ -479,4 +479,12 @@ public class BackendConfig {
public double getProcessEventFailureRatio() {
return backendCfg_.inject_process_event_failure_ratio;
}
+
+ public boolean enableWorkloadMgmt() {
+ return backendCfg_.enable_workload_mgmt;
+ }
+
+ public String queryLogTableName() {
+ return backendCfg_.query_log_table_name;
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/util/CatalogBlacklistUtils.java b/fe/src/main/java/org/apache/impala/util/CatalogBlacklistUtils.java
index aedd9118f..113dd7788 100644
--- a/fe/src/main/java/org/apache/impala/util/CatalogBlacklistUtils.java
+++ b/fe/src/main/java/org/apache/impala/util/CatalogBlacklistUtils.java
@@ -94,6 +94,10 @@ public class CatalogBlacklistUtils {
}
public static void verifyDbName(String dbName) throws AnalysisException {
+ if (BackendConfig.INSTANCE.enableWorkloadMgmt() && dbName.equalsIgnoreCase("sys")) {
+ // Override 'sys' for Impala system tables.
+ return;
+ }
if (BLACKLISTED_DBS.contains(dbName)) {
throw new AnalysisException("Invalid db name: " + dbName
+ ". It has been blacklisted using --blacklisted_dbs");
diff --git a/tests/authorization/test_provider.py b/tests/authorization/test_provider.py
index f7e71ac61..4d07e7082 100644
--- a/tests/authorization/test_provider.py
+++ b/tests/authorization/test_provider.py
@@ -45,12 +45,12 @@ class TestAuthorizationProvider(CustomClusterTestSuite):
@CustomClusterTestSuite.with_args(
expect_cores=True,
impala_log_dir=LOG_DIR,
- impalad_args="--minidump_path={0}"
+ impalad_args="--minidump_path={0} "
"--server-name=server1 "
"--ranger_service_type=hive "
"--ranger_app_id=impala "
"--authorization_provider={1}".format(MINIDUMP_PATH, BAD_FLAG),
- catalogd_args="--minidump_path={0}"
+ catalogd_args="--minidump_path={0} "
"--server-name=server1 "
"--ranger_service_type=hive "
"--ranger_app_id=impala "
diff --git a/tests/beeswax/impala_beeswax.py b/tests/beeswax/impala_beeswax.py
index dbe2e81fc..183c54fdf 100644
--- a/tests/beeswax/impala_beeswax.py
+++ b/tests/beeswax/impala_beeswax.py
@@ -182,7 +182,7 @@ class ImpalaBeeswaxClient(object):
service='impala', transport_type=trans_type, user=self.user,
password=self.password, use_ssl=self.use_ssl)
- def execute(self, query_string, user=None):
+ def execute(self, query_string, user=None, fetch_profile_after_close=False):
"""Re-directs the query to its appropriate handler, returns ImpalaBeeswaxResult"""
# Take care of leading/trailing whitespaces.
query_string = query_string.strip()
@@ -193,13 +193,20 @@ class ImpalaBeeswaxClient(object):
# DML queries are finished by this point.
time_taken = time.time() - start
- # fetch_results() will close the query after which there is no guarantee that
- # profile and log will be available so fetch them first.
- runtime_profile = self.get_runtime_profile(handle)
+ if not fetch_profile_after_close:
+ # fetch_results() will close the query after which there is no guarantee that
+ # profile and log will be available so fetch them first.
+ runtime_profile = self.get_runtime_profile(handle)
+
exec_summary = self.get_exec_summary_and_parse(handle)
log = self.get_log(handle.log_context)
result = self.fetch_results(query_string, handle)
+
+ if fetch_profile_after_close:
+ # Fetch the profile again after the query has closed and the profile is complete.
+ runtime_profile = self.get_runtime_profile(handle)
+
result.time_taken, result.start_time, result.runtime_profile, result.log = \
time_taken, start_time, runtime_profile, log
result.exec_summary = exec_summary
@@ -211,8 +218,16 @@ class ImpalaBeeswaxClient(object):
result.start_time = start_time
result.exec_summary = self.get_exec_summary_and_parse(handle)
result.log = self.get_log(handle.log_context)
- result.runtime_profile = self.get_runtime_profile(handle)
+
+ if not fetch_profile_after_close:
+ result.runtime_profile = self.get_runtime_profile(handle)
+
self.close_query(handle)
+
+ if fetch_profile_after_close:
+ # Fetch the profile again after the query has closed and the profile is complete.
+ result.runtime_profile = self.get_runtime_profile(handle)
+
return result
def get_exec_summary(self, handle):
diff --git a/tests/common/custom_cluster_test_suite.py b/tests/common/custom_cluster_test_suite.py
index 104be6c09..9a8626391 100644
--- a/tests/common/custom_cluster_test_suite.py
+++ b/tests/common/custom_cluster_test_suite.py
@@ -26,6 +26,7 @@ import pipes
import pytest
import subprocess
from impala_py_lib.helpers import find_all_files, is_core_dump
+from signal import SIGRTMIN
from subprocess import check_call
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.impala_cluster import ImpalaCluster
@@ -57,6 +58,11 @@ IMPALAD_TIMEOUT_S = 'impalad_timeout_s'
EXPECT_CORES = 'expect_cores'
# Additional arg to determine whether we should reset the Ranger policy repository.
RESET_RANGER = 'reset_ranger'
+# By default, Impalad processes are left running at the end of the test. The next test run
+# terminates the processes when calling the `bin/start-impala-cluster.py`` script. Setting
+# this to `True` causes Impalad processes to be sent the SIGRTMIN signal at the end of the
+# test which runs the Impalad shutdown steps instead of abruptly ending the process.
+IMPALAD_GRACEFUL_SHUTDOWN = 'impalad_graceful_shutdown'
# Run with fast topic updates by default to reduce time to first query running.
DEFAULT_STATESTORE_ARGS = '--statestore_update_frequency_ms=50 \
@@ -110,7 +116,8 @@ class CustomClusterTestSuite(ImpalaTestSuite):
start_args=None, default_query_options=None, jvm_args=None,
impala_log_dir=None, hive_conf_dir=None, cluster_size=None,
num_exclusive_coordinators=None, kudu_args=None, statestored_timeout_s=None,
- impalad_timeout_s=None, expect_cores=None, reset_ranger=False):
+ impalad_timeout_s=None, expect_cores=None, reset_ranger=False,
+ impalad_graceful_shutdown=False):
"""Records arguments to be passed to a cluster by adding them to the decorated
method's func_dict"""
def decorate(func):
@@ -143,6 +150,8 @@ class CustomClusterTestSuite(ImpalaTestSuite):
func.__dict__[EXPECT_CORES] = expect_cores
if reset_ranger is not False:
func.__dict__[RESET_RANGER] = True
+ if impalad_graceful_shutdown is not False:
+ func.__dict__[IMPALAD_GRACEFUL_SHUTDOWN] = True
return func
return decorate
@@ -209,6 +218,12 @@ class CustomClusterTestSuite(ImpalaTestSuite):
super(CustomClusterTestSuite, self).setup_class()
def teardown_method(self, method):
+ if method is not None and method.__dict__.get(IMPALAD_GRACEFUL_SHUTDOWN, False):
+ for impalad in self.cluster.impalads:
+ impalad.kill(SIGRTMIN)
+ for impalad in self.cluster.impalads:
+ impalad.wait_for_exit()
+
if HIVE_CONF_DIR in method.__dict__:
self._start_hive_service(None) # Restart Hive Service using default configs
diff --git a/tests/common/impala_connection.py b/tests/common/impala_connection.py
index f3832beda..3f8f2a9e0 100644
--- a/tests/common/impala_connection.py
+++ b/tests/common/impala_connection.py
@@ -208,10 +208,11 @@ class BeeswaxConnection(ImpalaConnection):
LOG.info("-- closing DML query for operation handle: %s" % operation_handle)
self.__beeswax_client.close_dml(operation_handle.get_handle())
- def execute(self, sql_stmt, user=None):
+ def execute(self, sql_stmt, user=None, fetch_profile_after_close=False):
LOG.info("-- executing against %s\n" % (self.__host_port))
log_sql_stmt(sql_stmt)
- return self.__beeswax_client.execute(sql_stmt, user=user)
+ return self.__beeswax_client.execute(sql_stmt, user=user,
+ fetch_profile_after_close=fetch_profile_after_close)
def execute_async(self, sql_stmt, user=None):
LOG.info("-- executing async: %s\n" % (self.__host_port))
@@ -344,7 +345,8 @@ class ImpylaHS2Connection(ImpalaConnection):
LOG.info("-- closing query for operation handle: {0}".format(operation_handle))
operation_handle.get_handle().close_operation()
- def execute(self, sql_stmt, user=None, profile_format=TRuntimeProfileFormat.STRING):
+ def execute(self, sql_stmt, user=None, profile_format=TRuntimeProfileFormat.STRING,
+ fetch_profile_after_close=False):
self.__cursor.execute(sql_stmt, configuration=self.__query_options)
handle = OperationHandle(self.__cursor, sql_stmt)
@@ -359,6 +361,14 @@ class ImpylaHS2Connection(ImpalaConnection):
self.close_query(handle)
except Exception:
pass
+ elif fetch_profile_after_close:
+ op_handle = handle.get_handle()._last_operation
+ self.close_query(handle)
+
+ # Match ImpalaBeeswaxResult by placing the full profile including end time and
+ # duration into the return object.
+ r.runtime_profile = op_handle.get_profile(profile_format)
+ return r
else:
self.close_query(handle)
return r
@@ -464,20 +474,22 @@ class ImpylaHS2Connection(ImpalaConnection):
profile = None
return ImpylaHS2ResultSet(success=True, result_tuples=result_tuples,
column_labels=column_labels, column_types=column_types,
- query=handle.sql_stmt(), log=log, profile=profile)
+ query=handle.sql_stmt(), log=log, profile=profile,
+ query_id=self.get_query_id(handle))
class ImpylaHS2ResultSet(object):
"""This emulates the interface of ImpalaBeeswaxResult so that it can be used in
place of it. TODO: when we deprecate/remove Beeswax, clean this up."""
def __init__(self, success, result_tuples, column_labels, column_types, query, log,
- profile):
+ profile, query_id):
self.success = success
self.column_labels = column_labels
self.column_types = column_types
self.query = query
self.log = log
self.profile = profile
+ self.query_id = query_id
self.__result_tuples = result_tuples
# self.data is the data in the ImpalaBeeswaxResult format: a list of rows with each
# row represented as a tab-separated string.
diff --git a/tests/custom_cluster/test_query_log.py b/tests/custom_cluster/test_query_log.py
new file mode 100644
index 000000000..48efbb78d
--- /dev/null
+++ b/tests/custom_cluster/test_query_log.py
@@ -0,0 +1,1007 @@
+# 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.
+
+from __future__ import absolute_import, division, print_function
+
+import os
+import pytest
+import string
+import tempfile
+
+from random import choice, randint
+from signal import SIGRTMIN
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.test_vector import ImpalaTestDimension
+from tests.util.retry import retry
+from tests.util.workload_management import assert_query, COMPRESSED_BYTES_SPILLED, \
+ BYTES_READ_CACHE_TOTAL
+from time import sleep, time
+
+
+class TestQueryLogTableBase(CustomClusterTestSuite):
+ """Base class for all query log tests. Sets up the tests to use the Beeswax and HS2
+ client protocols."""
+
+ WM_DB = "sys"
+ QUERY_TBL = "{0}.impala_query_log".format(WM_DB)
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestQueryLogTableBase, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension('protocol', 'beeswax', 'hs2'))
+
+
+class TestQueryLogTableBeeswax(TestQueryLogTableBase):
+ """Tests to assert the query log table is correctly populated when using the Beeswax
+ client protocol."""
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestQueryLogTableBeeswax, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_constraint(lambda v:
+ v.get_value('protocol') == 'beeswax')
+
+ CACHE_DIR = tempfile.mkdtemp(prefix="cache_dir")
+ MAX_SQL_PLAN_LEN = 2000
+ LOG_DIR_MAX_WRITES = tempfile.mkdtemp(prefix="max_writes")
+ FLUSH_MAX_RECORDS_CLUSTER_ID = "test_query_log_max_records_" + str(int(time()))
+ FLUSH_MAX_RECORDS_QUERY_COUNT = 2
+ OTHER_TBL = "completed_queries_table_{0}".format(int(time()))
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_max_select "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60 "
+ "--query_log_max_sql_length={0} "
+ "--query_log_max_plan_length={0}"
+ .format(MAX_SQL_PLAN_LEN),
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_lower_max_sql_plan(self, vector):
+ """Asserts that lower limits on the sql and plan columns in the completed queries
+ table are respected."""
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+ rand_long_str = "".join(choice(string.ascii_letters) for _ in
+ range(self.MAX_SQL_PLAN_LEN))
+
+ try:
+ handle = client.execute_async("select '{0}'".format(rand_long_str))
+ query_id = handle.get_handle().id
+ client.wait_for_finished_timeout(handle, 10)
+ client.close_query(handle)
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 1, 60)
+
+ # Force Impala to process the inserts to the completed queries table.
+ client.execute("refresh " + self.QUERY_TBL)
+
+ res = client.execute("select length(sql),plan from {0} where query_id='{1}'"
+ .format(self.QUERY_TBL, query_id))
+ assert res.success
+ assert len(res.data) == 1
+
+ data = res.data[0].split("\t")
+ assert len(data) == 2
+ assert int(data[0]) == self.MAX_SQL_PLAN_LEN - 1, "incorrect sql statement length"
+ assert len(data[1]) == self.MAX_SQL_PLAN_LEN - data[1].count("\n") - 1, \
+ "incorrect plan length"
+
+ finally:
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_max_select "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_over_max_sql_plan(self, vector):
+ """Asserts that very long queries have their corresponding plan and sql columns
+ shortened in the completed queries table."""
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+ rand_long_str = "".join(choice(string.ascii_letters) for _ in range(16778200))
+
+ try:
+ client.set_configuration_option("MAX_STATEMENT_LENGTH_BYTES", 16780000)
+ handle = client.execute_async("select '{0}'".format(rand_long_str))
+ query_id = handle.get_handle().id
+ client.wait_for_finished_timeout(handle, 10)
+ client.close_query(handle)
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 1, 60)
+
+ # Force Impala to process the inserts to the completed queries table.
+ client.execute("refresh " + self.QUERY_TBL)
+
+ client.set_configuration_option("MAX_ROW_SIZE", 35000000)
+ res = client.execute("select length(sql),plan from {0} where query_id='{1}'"
+ .format(self.QUERY_TBL, query_id))
+ assert res.success
+ assert len(res.data) == 1
+ data = res.data[0].split("\t")
+ assert len(data) == 2
+ assert data[0] == "16777215"
+
+ # Newline characters are not counted by Impala's length function.
+ assert len(data[1]) == 16777216 - data[1].count("\n") - 1
+
+ finally:
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_1 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60 "
+ "--query_log_size=0 "
+ "--query_log_size_in_bytes=0",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_no_query_log_select(self, vector):
+ """Asserts queries are written to the completed queries table when the query log is
+ turned off."""
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ # Run a select query.
+ random_val = randint(1, 1000000)
+ select_sql = "select {0}".format(random_val)
+ res = client.execute(select_sql)
+ assert res.success
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 1, 60)
+
+ # Force Impala to process the inserts to the completed queries table.
+ client.execute("refresh " + self.QUERY_TBL)
+
+ actual = client.execute("select sql from {0} where query_id='{1}'".format(
+ self.QUERY_TBL, res.query_id))
+ assert actual.success
+ assert len(actual.data) == 1
+ assert actual.data[0] == select_sql
+ finally:
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_2 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60 "
+ "--always_use_data_cache "
+ "--data_cache={0}:5GB".format(CACHE_DIR),
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True,
+ cluster_size=1)
+ def test_query_log_table_query_cache(self, vector):
+ """Asserts the values written to the query log table match the values from the
+ query profile. Specifically focuses on the data cache metrics."""
+ tbl_name = "default.test_query_log_cache" + str(int(time()))
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ # Create the test table.
+ create_tbl_sql = "create table {0} (id INT, product_name STRING) " \
+ "partitioned by (category INT)".format(tbl_name)
+ create_tbl_results = client.execute(create_tbl_sql)
+ assert create_tbl_results.success
+
+ # Insert some rows into the test table.
+ insert_sql = "insert into {0} (id,category,product_name) VALUES ".format(tbl_name)
+ for i in range(1, 11):
+ for j in range(1, 11):
+ if i * j > 1:
+ insert_sql += ","
+
+ random_product_name = "".join(choice(string.ascii_letters)
+ for _ in range(10))
+ insert_sql += "({0},{1},'{2}')".format((i * j), i, random_product_name)
+
+ insert_results = client.execute(insert_sql)
+ assert insert_results.success
+
+ # Select all rows from the test table. Run the query multiple times to ensure data
+ # is cached.
+ select_sql = "select * from {0}".format(tbl_name)
+ for i in range(3):
+ res = client.execute(select_sql)
+ assert res.success
+ sleep(1)
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 5, 60)
+
+ # Allow some time for the cache to be written to disk.
+ sleep(10)
+
+ # Run the same query again so results are read from the data cache.
+ res = client.execute(select_sql, fetch_profile_after_close=True)
+ assert res.success
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 6, 60)
+
+ data = assert_query(self.QUERY_TBL, client, "test_query_hist_2",
+ res.runtime_profile)
+
+ # Since the assert_query function only asserts that the bytes read from cache
+ # column is equal to the bytes read from cache in the profile, there is a potential
+ # for this test to not actually assert anything different than other tests. Thus, an
+ # additional assert is needed to ensure that there actually was data read from the
+ # cache.
+ assert data[BYTES_READ_CACHE_TOTAL] != "0", "bytes read from cache total was " \
+ "zero, test did not assert anything"
+ finally:
+ client.execute("drop table if exists {0}".format(tbl_name))
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=5 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60 "
+ "--log_dir={0}"
+ .format(LOG_DIR_MAX_WRITES),
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_max_attempts_exceeded(self, vector):
+ """Asserts that completed queries are only attempted 3 times to be inserted into the
+ completed queries table. This test deletes the completed queries table thus it must
+ not come last otherwise the table stays deleted. Subsequent tests will re-create
+ the table."""
+
+ print("USING LOG DIRECTORY: {0}".format(self.LOG_DIR_MAX_WRITES))
+
+ impalad = self.cluster.get_first_impalad()
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ res = client.execute("drop table {0} purge".format(self.QUERY_TBL))
+ assert res.success
+ impalad.service.wait_for_metric_value(
+ "impala-server.completed-queries.scheduled-writes", 3, 60)
+ impalad.service.wait_for_metric_value("impala-server.completed-queries.failure", 3,
+ 60)
+
+ query_count = 0
+
+ # Allow time for logs to be written to disk.
+ sleep(5)
+
+ with open(os.path.join(self.LOG_DIR_MAX_WRITES, "impalad.ERROR")) as file:
+ for line in file:
+ if line.find('could not write completed query table="{0}" query_id="{1}"'
+ .format(self.QUERY_TBL, res.query_id)) >= 0:
+ query_count += 1
+
+ assert query_count == 1
+
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.max-records-writes") == 0
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.queued") == 0
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.failure") == 3
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.scheduled-writes") == 4
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.written") == 0
+ finally:
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_max_queued={0} "
+ "--query_log_write_interval_s=9999 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60 "
+ "--cluster_id={1}"
+ .format(FLUSH_MAX_RECORDS_QUERY_COUNT,
+ FLUSH_MAX_RECORDS_CLUSTER_ID),
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_flush_max_records(self, vector):
+ """Asserts that queries that have completed are written to the query log table when
+ the maximum number of queued records it reached."""
+
+ impalad = self.cluster.get_first_impalad()
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ rand_str = "{0}-{1}".format(vector.get_value('protocol'), time())
+
+ test_sql = "select '{0}','{1}'".format(rand_str,
+ self.FLUSH_MAX_RECORDS_CLUSTER_ID)
+ test_sql_assert = "select '{0}', count(*) from {1} where sql='{2}'".format(
+ rand_str, self.QUERY_TBL, test_sql.replace("'", r"\'"))
+
+ try:
+ for _ in range(0, self.FLUSH_MAX_RECORDS_QUERY_COUNT):
+ res = client.execute(test_sql)
+ assert res.success
+
+ # Running this query results in the number of queued completed queries to exceed
+ # the max and thus all completed queries will be written to the query log table.
+ res = client.execute(test_sql_assert)
+ assert res.success
+ assert 1 == len(res.data)
+ assert "0" == res.data[0].split("\t")[1]
+
+ # Wait until the completed queries have all been written out because the max queued
+ # count was exceeded.
+ impalad.service.wait_for_metric_value(
+ "impala-server.completed-queries.max-records-writes", 1, 60)
+
+ # Force Impala to process the inserts to the completed queries table.
+ sleep(5)
+ client.execute("refresh " + self.QUERY_TBL)
+
+ # This query will remain queued due to the long write interval and max queued
+ # records limit not being reached.
+ res = client.execute(r"select count(*) from {0} where sql like 'select \'{1}\'%'"
+ .format(self.QUERY_TBL, rand_str))
+ assert res.success
+ assert 1 == len(res.data)
+ assert "3" == res.data[0]
+ impalad.service.wait_for_metric_value(
+ "impala-server.completed-queries.queued", 2, 60)
+ finally:
+ client.close()
+
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.max-records-writes") == 1
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.scheduled-writes") == 0
+ assert impalad.service.get_metric_value("impala-server.completed-queries.written") \
+ == self.FLUSH_MAX_RECORDS_QUERY_COUNT + 1
+ assert impalad.service.get_metric_value(
+ "impala-server.completed-queries.queued") == 2
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=30 "
+ "--blacklisted_dbs=information_schema "
+ "--query_log_table_name={0}"
+ .format(OTHER_TBL),
+ catalogd_args="--enable_workload_mgmt "
+ "--blacklisted_dbs=information_schema",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_different_table(self, vector):
+ """Asserts that queries that have completed but are not yet written to the query
+ log table are flushed to the table before the coordinator exits."""
+
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ res = client.execute("show tables in {0}".format(self.WM_DB))
+ assert res.success
+ assert len(res.data) > 0, "could not find any tables in database {0}" \
+ .format(self.DB)
+
+ tbl_found = False
+ for tbl in res.data:
+ if tbl.startswith(self.OTHER_TBL):
+ tbl_found = True
+ break
+ assert tbl_found, "could not find table '{0}' in database '{1}'" \
+ .format(self.OTHER_TBL, self.DB)
+ finally:
+ client.execute("drop table {0}.{1} purge".format(self.WM_DB, self.OTHER_TBL))
+ client.close()
+
+
+class TestQueryLogTableHS2(TestQueryLogTableBase):
+ """Tests to assert the query log table is correctly populated when using the HS2
+ client protocol."""
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestQueryLogTableHS2, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_constraint(lambda v:
+ v.get_value('protocol') == 'hs2')
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_mult "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ cluster_size=2,
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_query_multiple(self, vector):
+ """Asserts the values written to the query log table match the values from the
+ query profile for a query that reads from multiple tables."""
+ tbl_name = "default.test_query_log_" + str(int(time()))
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ # Create the first test table.
+ create_tbl_sql = "create table {0}_products (id INT, product_name STRING)" \
+ .format(tbl_name)
+ create_tbl_results = client.execute(create_tbl_sql)
+ assert create_tbl_results.success
+
+ # Insert some rows into the test products table.
+ insert_sql = "insert into {0}_products (id,product_name) VALUES ".format(tbl_name)
+ for i in range(1, 11):
+ for j in range(1, 11):
+ if i * j > 1:
+ insert_sql += ","
+
+ random_product_name = "".join(choice(string.ascii_letters) for _ in range(10))
+ insert_sql += "({0},'{1}')".format((i * j), random_product_name)
+
+ insert_results = client.execute(insert_sql)
+ assert insert_results.success
+
+ # Create the second test table.
+ create_tbl_sql = "create table {0}_customers (id INT, name STRING) " \
+ .format(tbl_name)
+ create_tbl_results = client.execute(create_tbl_sql)
+ assert create_tbl_results.success
+
+ # Insert rows into the test customers table.
+ insert_sql = "insert into {0}_customers (id,name) VALUES ".format(tbl_name)
+ for i in range(1, 11):
+ if i > 1:
+ insert_sql += ","
+ rand_cust_name = "".join(choice(string.ascii_letters) for _ in range(10))
+ insert_sql += "({0},'{1}')".format(i, rand_cust_name)
+
+ insert_results = client.execute(insert_sql)
+ assert insert_results.success
+
+ # Create the third test table.
+ create_tbl_sql = "create table {0}_sales (id INT, product_id INT, " \
+ "customer_id INT) ".format(tbl_name)
+ create_tbl_results = client.execute(create_tbl_sql)
+ assert create_tbl_results.success
+
+ # Insert rows into the test sales table.
+ insert_sql = "insert into {0}_sales (id, product_id, customer_id) VALUES " \
+ .format(tbl_name)
+ for i in range(1, 1001):
+ if i != 1:
+ insert_sql += ","
+ insert_sql += "({0},{1},{2})".format(i * j, randint(1, 100), randint(1, 10))
+
+ insert_results = client.execute(insert_sql)
+ assert insert_results.success
+
+ # Select all rows from the test table.
+ client.set_configuration_option("MAX_MEM_ESTIMATE_FOR_ADMISSION", "10MB")
+ res = client.execute("select s.id, p.product_name, c.name from {0}_sales s "
+ "inner join {0}_products p on s.product_id=p.id "
+ "inner join {0}_customers c on s.customer_id=c.id".format(tbl_name),
+ fetch_profile_after_close=True)
+ assert res.success
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 7, 60)
+
+ client2 = self.create_client_for_nth_impalad(1, vector.get_value('protocol'))
+ assert client2 is not None
+ assert_query(self.QUERY_TBL, client2, "test_query_hist_mult", res.runtime_profile,
+ max_mem_for_admission=10485760)
+ finally:
+ client.execute("drop table if exists {0}_sales".format(tbl_name))
+ client.execute("drop table if exists {0}_customers".format(tbl_name))
+ client.execute("drop table if exists {0}_products".format(tbl_name))
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_3 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_query_insert_select(self, vector):
+ """Asserts the values written to the query log table match the values from the
+ query profile for a query that insert selects."""
+ tbl_name = "default.test_query_log_insert_select" + str(int(time()))
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ # Create the source test table.
+ assert client.execute("create table {0}_source (id INT, product_name STRING) "
+ .format(tbl_name)).success, "could not create source table"
+
+ # Insert some rows into the test table.
+ insert_sql = "insert into {0}_source (id,product_name) VALUES " \
+ .format(tbl_name)
+ for i in range(1, 100):
+ if i > 1:
+ insert_sql += ","
+
+ random_product_name = "".join(choice(string.ascii_letters)
+ for _ in range(10))
+ insert_sql += "({0},'{1}')".format(i, random_product_name)
+
+ assert client.execute(insert_sql).success, "could not insert rows"
+
+ # Create the destination test table.
+ assert client.execute("create table {0}_dest (id INT, product_name STRING) "
+ .format(tbl_name)).success, "could not create destination table"
+
+ # Insert select from the source table to the destination table.
+ res = client.execute("insert into {0}_dest (id, product_name) select id, "
+ "product_name from {0}_source".format(tbl_name), fetch_profile_after_close=True)
+ assert res.success, "could not insert select"
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 4, 60)
+
+ client2 = self.create_client_for_nth_impalad(2, vector.get_value('protocol'))
+ assert client2 is not None
+ assert_query(self.QUERY_TBL, client2, "test_query_hist_3", res.runtime_profile)
+ finally:
+ client.execute("drop table if exists {0}_source".format(tbl_name))
+ client.execute("drop table if exists {0}_dest".format(tbl_name))
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=15 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_flush_interval(self, vector):
+ """Asserts that queries that have completed are written to the query log table
+ after the specified write interval elapses."""
+
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ query_count = 10
+
+ for i in range(query_count):
+ res = client.execute("select sleep(1000)")
+ assert res.success
+
+ # At least 10 seconds have already elapsed, wait up to 10 more seconds for the
+ # queries to be written to the completed queries table.
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", query_count, 10)
+ finally:
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=9999 "
+ "--shutdown_grace_period_s=30 "
+ "--shutdown_deadline_s=30",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=False)
+ def test_query_log_table_flush_on_shutdown(self, vector):
+ """Asserts that queries that have completed but are not yet written to the query
+ log table are flushed to the table before the coordinator exits."""
+
+ impalad = self.cluster.get_first_impalad()
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ # Execute sql statements to ensure all get written to the query log table.
+ sql1 = client.execute("select 1")
+ assert sql1.success
+
+ sql2 = client.execute("select 2")
+ assert sql2.success
+
+ sql3 = client.execute("select 3")
+ assert sql3.success
+
+ impalad.service.wait_for_metric_value("impala-server.completed-queries.queued", 3,
+ 60)
+
+ impalad.kill_and_wait_for_exit(SIGRTMIN)
+
+ client2 = self.create_client_for_nth_impalad(1, vector.get_value('protocol'))
+
+ def assert_func(last_iteration):
+ results = client2.execute("select query_id,sql from {0} where query_id in "
+ "('{1}','{2}','{3}')".format(self.QUERY_TBL,
+ sql1.query_id, sql2.query_id, sql3.query_id))
+
+ success = len(results.data) == 3
+ if last_iteration:
+ assert len(results.data) == 3
+
+ return success
+
+ assert retry(func=assert_func, max_attempts=5, sleep_time_s=5)
+ finally:
+ client.close()
+ client2.close()
+
+
+class TestQueryLogTableAll(TestQueryLogTableBase):
+ """Tests to assert the query log table is correctly populated when using all the
+ client protocols."""
+
+ SCRATCH_DIR = tempfile.mkdtemp(prefix="scratch_dir")
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_2 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_ddl(self, vector):
+ """Asserts the values written to the query log table match the values from the
+ query profile for a DDL query."""
+ tbl_name = "default.test_query_log_ddl_" + str(int(time()))
+ create_tbl_sql = "create table {0} (id INT, product_name STRING) " \
+ "partitioned by (category INT)".format(tbl_name)
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ res = client.execute(create_tbl_sql, fetch_profile_after_close=True)
+ assert res.success
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 1, 60)
+
+ client2 = self.create_client_for_nth_impalad(2, vector.get_value('protocol'))
+ assert client2 is not None
+ assert_query(self.QUERY_TBL, client2, "test_query_hist_2", res.runtime_profile)
+ finally:
+ client.execute("drop table if exists {0}".format(tbl_name))
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_3 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_dml(self, vector):
+ """Asserts the values written to the query log table match the values from the
+ query profile for a DML query."""
+ tbl_name = "default.test_query_log_dml_" + str(int(time()))
+ create_tbl_sql = "create table {0} (id INT, product_name STRING) " \
+ "partitioned by (category INT)".format(tbl_name)
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ # Create the test table.
+ create_tbl_sql = "create table {0} (id INT, product_name STRING) " \
+ "partitioned by (category INT)".format(tbl_name)
+ create_tbl_results = client.execute(create_tbl_sql)
+ assert create_tbl_results.success
+
+ insert_sql = "insert into {0} (id,category,product_name) values " \
+ "(0,1,'the product')".format(tbl_name)
+ res = client.execute(insert_sql, fetch_profile_after_close=True)
+ assert res.success
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 2, 60)
+
+ client2 = self.create_client_for_nth_impalad(2, vector.get_value('protocol'))
+ assert client2 is not None
+ assert_query(self.QUERY_TBL, client2, "test_query_hist_3", res.runtime_profile)
+ finally:
+ client.execute("drop table if exists {0}".format(tbl_name))
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_1 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60 "
+ "--scratch_dirs={0}:5G"
+ .format(SCRATCH_DIR),
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ @pytest.mark.parametrize("buffer_pool_limit", [(None), ("16.05MB")])
+ def test_query_log_table_query_select(self, vector, buffer_pool_limit):
+ """Asserts the values written to the query log table match the values from the
+ query profile. If the buffer_pool_limit parameter is not None, then this test
+ requires that the query spills to disk to assert that the spill metrics are correct
+ in the completed queries table."""
+ tbl_name = "default.test_query_log_" + str(int(time()))
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+ query_cnt = 0
+
+ try:
+ # Create the test table.
+ create_tbl_sql = "create table {0} (id INT, product_name STRING, create_dt STRING" \
+ ",descr STRING) partitioned by (category INT)".format(tbl_name)
+ print("CREATE TABLE SQL: {0}".format(create_tbl_sql))
+ create_tbl_results = client.execute(create_tbl_sql)
+ assert create_tbl_results.success
+ query_cnt += 1
+
+ # Insert some rows into the test table.
+ def __run_insert(values):
+ insert_results = client.execute("insert into {0} (id,category,product_name,"
+ "create_dt,descr) VALUES ({1})".format(tbl_name, values))
+ assert insert_results.success
+
+ # When buffer pool limit is not None, the test is forcing the query to spill. Thus,
+ # a large number of records is needed to force the spilling.
+ record_count_to_insert = 99
+ if buffer_pool_limit is not None:
+ record_count_to_insert = 24999
+
+ insert_vals = ""
+ for i in range(1, record_count_to_insert):
+ random_product_name = "".join(choice(string.ascii_letters) for _ in range(100))
+ random_dt = "{:}-{:0>2}-{:0>2}".format(randint(1982, 2022), randint(1, 12),
+ randint(1, 31))
+ random_desc = "".join(choice(string.ascii_letters) for _ in range(1000))
+ insert_vals += "({0},{1},'{2}','{3}','{4}'),".format(i, (i % 50),
+ random_product_name, random_dt, random_desc)
+
+ if i % 500 == 0:
+ __run_insert(insert_vals[:-1])
+ query_cnt += 1
+ insert_vals = ""
+
+ __run_insert(insert_vals[:-1])
+ query_cnt += 1
+
+ # Set up query configuration
+ client.set_configuration_option("MAX_MEM_ESTIMATE_FOR_ADMISSION", "10MB")
+ if buffer_pool_limit is not None:
+ client.set_configuration_option("BUFFER_POOL_LIMIT", buffer_pool_limit)
+ client.set_configuration_option("SPOOL_QUERY_RESULTS", "TRUE")
+
+ # Select all rows from the test table.
+ res = client.execute("select * from {0} order by create_dt".format(tbl_name),
+ fetch_profile_after_close=True)
+ assert res.success
+ query_cnt += 1
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", query_cnt, 60)
+
+ client2 = self.create_client_for_nth_impalad(2, vector.get_value('protocol'))
+ assert client2 is not None
+ data = assert_query(self.QUERY_TBL, client2, "test_query_hist_1",
+ res.runtime_profile, max_mem_for_admission=10485760)
+
+ if buffer_pool_limit is not None:
+ # Since the assert_query function only asserts that the compressed bytes spilled
+ # column is equal to the compressed bytes spilled in the profile, there is a
+ # potential for this test to not actually assert anything different than other
+ # tests. Thus, an additional assert is needed to ensure that there actually was
+ # data that was spilled.
+ assert data[COMPRESSED_BYTES_SPILLED] != "0", "compressed bytes spilled total " \
+ "was zero, test did not assert anything"
+ finally:
+ client.execute("drop table if exists {0}".format(tbl_name))
+ client.close()
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--cluster_id=test_query_hist_2 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60 ",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_invalid_query(self, vector):
+ """Asserts correct values are written to the completed queries table for a failed
+ query. The query profile is used as the source of expected values."""
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ # Assert an invalid query
+ unix_now = time()
+ try:
+ client.execute("{0}".format(unix_now))
+ except Exception as _:
+ pass
+
+ # Get the query id from the completed queries table since the call to execute errors
+ # instead of return the results object which contains the query id.
+ impalad = self.cluster.get_first_impalad()
+ impalad.service.wait_for_metric_value("impala-server.completed-queries.written", 1,
+ 60)
+
+ result = client.execute("select query_id from {0} where sql='{1}'"
+ .format(self.QUERY_TBL, unix_now),
+ fetch_profile_after_close=True)
+ assert result.success
+ assert len(result.data) == 1
+
+ assert_query(query_tbl=self.QUERY_TBL, client=client,
+ expected_cluster_id="test_query_hist_2", impalad=impalad, query_id=result.data[0])
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_ignored_sqls(self, vector):
+ """Asserts that expected queries are not written to the query log table."""
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ sqls = {}
+ sqls["use default"] = False
+ sqls["USE default"] = False
+ sqls["uSe default"] = False
+ sqls["--mycomment\nuse default"] = False
+ sqls["/*mycomment*/ use default"] = False
+
+ sqls["set all"] = False
+ sqls["SET all"] = False
+ sqls["SeT all"] = False
+ sqls["--mycomment\nset all"] = False
+ sqls["/*mycomment*/ set all"] = False
+
+ sqls["show tables"] = False
+ sqls["SHOW tables"] = False
+ sqls["ShoW tables"] = False
+ sqls["ShoW create table {0}".format(self.QUERY_TBL)] = False
+ sqls["show databases"] = False
+ sqls["SHOW databases"] = False
+ sqls["ShoW databases"] = False
+ sqls["show schemas"] = False
+ sqls["SHOW schemas"] = False
+ sqls["ShoW schemas"] = False
+ sqls["--mycomment\nshow tables"] = False
+ sqls["/*mycomment*/ show tables"] = False
+ sqls["/*mycomment*/ show tables"] = False
+ sqls["/*mycomment*/ show create table {0}".format(self.QUERY_TBL)] = False
+ sqls["/*mycomment*/ show files in {0}".format(self.QUERY_TBL)] = False
+ sqls["/*mycomment*/ show functions"] = False
+ sqls["/*mycomment*/ show data sources"] = False
+ sqls["/*mycomment*/ show views"] = False
+
+ sqls["describe database default"] = False
+ sqls["/*mycomment*/ describe database default"] = False
+ sqls["describe {0}".format(self.QUERY_TBL)] = False
+ sqls["/*mycomment*/ describe {0}".format(self.QUERY_TBL)] = False
+ sqls["describe history {0}".format(self.QUERY_TBL)] = False
+ sqls["/*mycomment*/ describe history {0}".format(self.QUERY_TBL)] = False
+ sqls["select 1"] = True
+
+ control_queries_count = 0
+ try:
+ for sql, experiment_control in sqls.items():
+ results = client.execute(sql)
+ assert results.success, "could not execute query '{0}'".format(sql)
+ sqls[sql] = results.query_id
+
+ # Ensure at least one sql statement was written to the completed queries table
+ # to avoid false negatives where the sql statements that are ignored are not
+ # written to the completed queries table because of another issue.
+ if experiment_control:
+ control_queries_count += 1
+ sql_results = None
+ for _ in range(6):
+ sql_results = client.execute("select * from {0} where query_id='{1}'".format(
+ self.QUERY_TBL, results.query_id))
+ control_queries_count += 1
+ if sql_results.success and len(sql_results.data) == 1:
+ break
+ else:
+ sleep(5)
+ assert sql_results.success
+ assert len(sql_results.data) == 1, "query not found in completed queries table"
+ sqls.pop(sql)
+
+ for sql, query_id in sqls.items():
+ log_results = client.execute("select * from {0} where query_id='{1}'"
+ .format(self.QUERY_TBL, query_id))
+ assert log_results.success
+ assert len(log_results.data) == 0, "found query in query log table: {0}" \
+ .format(sql)
+ finally:
+ client.close()
+
+ # Assert there was one query per sql item written to the query log table. The queries
+ # inserted into the completed queries table are the queries used to assert the ignored
+ # queries were not written to the table.
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", len(sqls) + control_queries_count, 60)
+ assert self.cluster.get_first_impalad().service.get_metric_value(
+ "impala-server.completed-queries.failure") == 0
+
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--shutdown_grace_period_s=10 "
+ "--shutdown_deadline_s=60",
+ catalogd_args="--enable_workload_mgmt",
+ impalad_graceful_shutdown=True)
+ def test_query_log_table_sql_injection(self, vector):
+ tbl_name = "default.test_query_log_sql_injection_" + str(int(time()))
+ client = self.create_impala_client(protocol=vector.get_value('protocol'))
+
+ try:
+ # Create the test table.
+ create_tbl_sql = "create table {0} (id INT, product_name STRING) " \
+ "partitioned by (category INT)".format(tbl_name)
+ create_tbl_results = client.execute(create_tbl_sql)
+ assert create_tbl_results.success
+
+ # Insert some rows into the test table.
+ insert_sql = "insert into {0} (id,category,product_name) VALUES ".format(tbl_name)
+ for i in range(1, 11):
+ for j in range(1, 11):
+ if i * j > 1:
+ insert_sql += ","
+
+ insert_sql += "({0},{1},'{2}')".format((i * j), i,
+ "product-{0}-{1}".format(i, j))
+
+ insert_results = client.execute(insert_sql)
+ assert insert_results.success
+
+ impalad = self.cluster.get_first_impalad()
+
+ # Try a sql injection attack with closing quotes.
+ sql1_str = "select * from {0} where product_name='product-2-3'".format(tbl_name)
+ self.__run_sql_inject(impalad, client, sql1_str, "closing quotes", 3)
+
+ # Try a sql inject attack with terminating quote and semicolon.
+ sql2_str = "select 1'); drop table {0}; select('" \
+ .format(self.QUERY_TBL)
+ self.__run_sql_inject(impalad, client, sql2_str, "terminating semicolon", 6)
+
+ # Attempt to cause an error using multiline comments.
+ sql3_str = "select 1' /* foo"
+ self.__run_sql_inject(impalad, client, sql3_str, "multiline comments", 9, False)
+
+ # Attempt to cause an error using single line comments.
+ sql4_str = "select 1' -- foo"
+ self.__run_sql_inject(impalad, client, sql4_str, "single line comments", 12, False)
+
+ finally:
+ client.execute("drop table if exists {0}".format(tbl_name))
+ client.close()
+
+ def __run_sql_inject(self, impalad, client, sql, test_case, expected_writes,
+ expect_success=True):
+ sql_result = None
+ try:
+ sql_result = client.execute(sql)
+ except Exception as e:
+ if expect_success:
+ raise e
+
+ if expect_success:
+ assert sql_result.success
+
+ impalad.service.wait_for_metric_value(
+ "impala-server.completed-queries.written", expected_writes, 60)
+
+ # Force Impala to process the inserts to the completed queries table.
+ client.execute("refresh " + self.QUERY_TBL)
+
+ if expect_success:
+ sql_verify = client.execute(
+ "select sql from {0} where query_id='{1}'"
+ .format(self.QUERY_TBL, sql_result.query_id))
+
+ assert sql_verify.success, test_case
+ assert len(sql_verify.data) == 1, "did not find query '{0}' in query log " \
+ "table for test case '{1}" \
+ .format(sql_result.query_id, test_case)
+ assert sql_verify.data[0] == sql, test_case
+ else:
+ esc_sql = sql.replace("'", "\\'")
+ sql_verify = client.execute("select sql from {0} where sql='{1}' "
+ "and start_time_utc > "
+ "date_sub(utc_timestamp(), interval 25 seconds);"
+ .format(self.QUERY_TBL, esc_sql))
+ assert sql_verify.success, test_case
+ assert len(sql_verify.data) == 1, "did not find query '{0}' in query log " \
+ "table for test case '{1}" \
+ .format(esc_sql, test_case)
diff --git a/tests/util/assert_time.py b/tests/util/assert_time.py
new file mode 100644
index 000000000..7c6bc3f72
--- /dev/null
+++ b/tests/util/assert_time.py
@@ -0,0 +1,57 @@
+# 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.
+
+def assert_time_str(expected_str, actual_time_ns, msg, tolerance=0.005):
+ """Asserts a pretty printed time string matches a specific number of nanoseconds."""
+
+ total_nanoseconds = convert_to_nanos(expected_str)
+ actual_time_ns = int(actual_time_ns)
+
+ expected_min = total_nanoseconds - (total_nanoseconds * tolerance)
+ expected_max = total_nanoseconds + (total_nanoseconds * tolerance)
+ assert expected_min <= actual_time_ns <= expected_max, \
+ "{0} -- expected: {1}, actual: {2}, calculated: {3}, tolerance: {4}" \
+ .format(msg, expected_str, actual_time_ns, total_nanoseconds, tolerance)
+
+
+def convert_to_nanos(time_str):
+ """Convert a pretty printed time string into integer nanoseconds."""
+ units = {'h': 3600 * 1e9, 'm': 60 * 1e9, 's': 1e9, 'ms': 1e6, 'us': 1e3, 'ns': 1}
+
+ total_nanoseconds = 0
+ current_number = ''
+ current_unit = ''
+
+ for char in time_str:
+ if char.isdigit() or char == '.':
+ if current_unit != '':
+ if current_unit in units:
+ total_nanoseconds += int(float(current_number) * units[current_unit])
+ current_number = ''
+ current_unit = ''
+ else:
+ raise ValueError("Invalid alphabetic unit '{0}' in time string"
+ .format(current_unit))
+ current_number += char
+ elif char.isalpha():
+ current_unit += char
+ else:
+ raise ValueError("Invalid character in time string")
+
+ total_nanoseconds += int(float(current_number) * units[current_unit])
+
+ return total_nanoseconds
diff --git a/tests/util/memory.py b/tests/util/memory.py
new file mode 100644
index 000000000..8b28cfe15
--- /dev/null
+++ b/tests/util/memory.py
@@ -0,0 +1,74 @@
+# 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.
+
+from __future__ import absolute_import, division, print_function
+
+from re import match
+
+
+def assert_byte_str(expected_str, actual_bytes, msg, unit_combined=False,
+ tolerance=0.005):
+ """Asserts a pretty printed memory string matches a specifiec number of bytes."""
+ calc = convert_to_bytes(expected_str, unit_combined)
+
+ # Allow a tolerance of +- half a percent to allow for conversion differences between
+ # Impala and the Python tests.
+ expected_min = calc - (calc * tolerance)
+ expected_max = calc + (calc * tolerance)
+
+ assert expected_min <= int(actual_bytes) <= expected_max, "{0} -- expected: {1}, " \
+ "actual: {2}, calculated: {3}, tolerance: {4}".format(msg, expected_str,
+ actual_bytes, calc, tolerance)
+
+
+def convert_to_bytes(mem_str, unit_combined=False):
+ """Converts a pretty printed memory string into bytes. Since pretty printing removes
+ precision, the result may not equal the original number of bytes. Returns an int.
+
+ By default, the format of mem_str is to have a space between the number and the
+ units, but setting the unit_combined causes this function to use the last two
+ characters of mem_str as the units and all other characters as the number."""
+ unit = ""
+ calc = 0
+ multiplier = 1024
+ unit_length = -2
+
+ # Memory can have a unit of 'GB', 'MB', 'KB', or 'B'. If this regular expression
+ # matches the provided memory string, then a unit of 'B' was used.
+ if match(r'\d+\.?\d*B$', mem_str):
+ unit_length = -1
+
+ if unit_combined:
+ unit = mem_str[unit_length:]
+ calc = float(mem_str[:unit_length])
+ else:
+ split_str = mem_str.split(' ')
+ unit = split_str[1]
+ calc = float(split_str[0])
+
+ if unit == 'B' or unit == '':
+ calc *= 1
+ elif unit == 'KB':
+ calc *= multiplier
+ elif unit == 'MB':
+ calc *= multiplier**2
+ elif unit == 'GB':
+ calc *= multiplier**3
+ else:
+ raise ValueError("Invalid unit for byte string: {}".format(mem_str))
+
+ return int(calc)
diff --git a/tests/util/retry.py b/tests/util/retry.py
new file mode 100644
index 000000000..fb91c9ad4
--- /dev/null
+++ b/tests/util/retry.py
@@ -0,0 +1,87 @@
+# 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.
+
+from __future__ import absolute_import, division, print_function
+
+from time import sleep
+
+
+def retry(func, max_attempts=3, sleep_time_s=1, backoff=2, raise_immediately=False):
+ """
+ Repeatedly executes a provided function until the function either returns True or the
+ maximum number of iterations is it.
+
+ Inputs:
+ func - A function to execute, this function must take a single boolean
+ parameter which indicates if the current attempt is the last
+ attempt, the purpose of this parameter is to enable assertions to
+ run during the lastattempt so that better error messages a
+ provided.
+ max_attempts - The maximum number of times to run the function.
+ sleep_time_s - The number of seconds to sleep before retrying.
+ backoff - Defines the amount the the sleep time will be multiplied by after
+ each attempt. Specify a value of `1` if the same amount of time
+ should be used as the sleep between each attempt.
+ raise_immediately - Controls the behavior of this function when an exception is raised
+ by the provided function to be executed. If an exception is raised
+ out of that function and this parameter is `True`, the exception
+ will be immediately raised up the stack. If this parameter is
+ `False`, then the exception is only raised if it is the last
+ attempt.
+
+ Return:
+ `True` - Indicates the provided function returned `True`.
+ `False` - Indicates the provided function returned `False` for each attempt.
+
+ Raises:
+ See the doc for the `raise_immediately` parameter.
+
+ To use this function, code a separate named function to return `True` if it succeeds or
+ `False` if it fails. If the named function returns True, this function will also return
+ `True`. If the named function returns `False`, this function will try again. If, at the
+ last iteration, the named function returns `False`, this function returns `False`.
+
+ If the named function raises an exception, the behavior of this function depends on the
+ input parameter `raise_immediately`. If set to `True`, the exception is
+ immediately raised from this function. If set to `False`, the exception is ignored,
+ unless the maximum number of attempts is hit at which point the exception will be
+ raised.
+ """
+ success = False
+ iterations = 0
+ e = None
+
+ while(iterations < max_attempts):
+ try:
+ success = func(iterations == max_attempts - 1)
+
+ if success:
+ return True
+ except Exception as func_e:
+ e = func_e
+ if raise_immediately:
+ raise e
+
+ sleep(sleep_time_s)
+ iterations += 1
+ sleep_time_s *= backoff
+
+ print("max retries '{0}' exceeded".format(max_attempts))
+ if e is not None:
+ raise e
+
+ return False
diff --git a/tests/util/workload_management.py b/tests/util/workload_management.py
new file mode 100644
index 000000000..0d6827baf
--- /dev/null
+++ b/tests/util/workload_management.py
@@ -0,0 +1,730 @@
+# 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.
+
+from __future__ import absolute_import, division, print_function
+
+import re
+import requests
+
+from datetime import datetime
+from tests.util.assert_time import assert_time_str, convert_to_nanos
+from tests.util.memory import assert_byte_str, convert_to_bytes
+
+DEDICATED_COORD_SAFETY_BUFFER_BYTES = 104857600
+EXPECTED_QUERY_COLS = 48
+
+
+CLUSTER_ID = "CLUSTER_ID"
+QUERY_ID = "QUERY_ID"
+SESSION_ID = "SESSION_ID"
+SESSION_TYPE = "SESSION_TYPE"
+HIVESERVER2_PROTOCOL_VERSION = "HIVESERVER2_PROTOCOL_VERSION"
+DB_USER = "DB_USER"
+DB_USER_CONNECTION = "DB_USER_CONNECTION"
+DB_NAME = "DB_NAME"
+IMPALA_COORDINATOR = "IMPALA_COORDINATOR"
+QUERY_STATUS = "QUERY_STATUS"
+QUERY_STATE = "QUERY_STATE"
+IMPALA_QUERY_END_STATE = "IMPALA_QUERY_END_STATE"
+QUERY_TYPE = "QUERY_TYPE"
+NETWORK_ADDRESS = "NETWORK_ADDRESS"
+START_TIME_UTC = "START_TIME_UTC"
+TOTAL_TIME_NS = "TOTAL_TIME_NS"
+QUERY_OPTS_CONFIG = "QUERY_OPTS_CONFIG"
+RESOURCE_POOL = "RESOURCE_POOL"
+PER_HOST_MEM_ESTIMATE = "PER_HOST_MEM_ESTIMATE"
+DEDICATED_COORD_MEM_ESTIMATE = "DEDICATED_COORD_MEM_ESTIMATE"
+PER_HOST_FRAGMENT_INSTANCES = "PER_HOST_FRAGMENT_INSTANCES"
+BACKENDS_COUNT = "BACKENDS_COUNT"
+ADMISSION_RESULT = "ADMISSION_RESULT"
+CLUSTER_MEMORY_ADMITTED = "CLUSTER_MEMORY_ADMITTED"
+EXECUTOR_GROUP = "EXECUTOR_GROUP"
+EXECUTOR_GROUPS = "EXECUTOR_GROUPS"
+EXEC_SUMMARY = "EXEC_SUMMARY"
+NUM_ROWS_FETCHED = "NUM_ROWS_FETCHED"
+ROW_MATERIALIZATION_ROWS_PER_SEC = "ROW_MATERIALIZATION_ROWS_PER_SEC"
+ROW_MATERIALIZATION_TIME_NS = "ROW_MATERIALIZATION_TIME_NS"
+COMPRESSED_BYTES_SPILLED = "COMPRESSED_BYTES_SPILLED"
+EVENT_PLANNING_FINISHED = "EVENT_PLANNING_FINISHED"
+EVENT_SUBMIT_FOR_ADMISSION = "EVENT_SUBMIT_FOR_ADMISSION"
+EVENT_COMPLETED_ADMISSION = "EVENT_COMPLETED_ADMISSION"
+EVENT_ALL_BACKENDS_STARTED = "EVENT_ALL_BACKENDS_STARTED"
+EVENT_ROWS_AVAILABLE = "EVENT_ROWS_AVAILABLE"
+EVENT_FIRST_ROW_FETCHED = "EVENT_FIRST_ROW_FETCHED"
+EVENT_LAST_ROW_FETCHED = "EVENT_LAST_ROW_FETCHED"
+EVENT_UNREGISTER_QUERY = "EVENT_UNREGISTER_QUERY"
+READ_IO_WAIT_TOTAL_NS = "READ_IO_WAIT_TOTAL_NS"
+READ_IO_WAIT_MEAN_NS = "READ_IO_WAIT_MEAN_NS"
+BYTES_READ_CACHE_TOTAL = "BYTES_READ_CACHE_TOTAL"
+BYTES_READ_TOTAL = "BYTES_READ_TOTAL"
+PERNODE_PEAK_MEM_MIN = "PERNODE_PEAK_MEM_MIN"
+PERNODE_PEAK_MEM_MAX = "PERNODE_PEAK_MEM_MAX"
+PERNODE_PEAK_MEM_MEAN = "PERNODE_PEAK_MEM_MEAN"
+SQL = "SQL"
+PLAN = "PLAN"
+
+
+def assert_query(query_tbl, client, expected_cluster_id, raw_profile=None, impalad=None,
+ query_id=None, max_mem_for_admission=None, max_row_size=None):
+ """Helper function to assert that the values in the completed query log table
+ match the values from the query profile."""
+
+ ret_data = {}
+
+ # If query_id was specified, read the profile from the Impala webserver.
+ if query_id is not None:
+ assert impalad is not None
+ assert raw_profile is None, "cannot specify both query_id and raw_profile"
+ resp = requests.get("http://{0}:{1}/query_profile_plain_text?query_id={2}"
+ .format(impalad.hostname, impalad.get_webserver_port(), query_id))
+ assert resp.status_code == 200, "Response code was: {0}".format(resp.status_code)
+ profile_text = resp.text
+ else:
+ profile_text = raw_profile
+ assert query_id is None, "cannot specify both raw_profile and query_id"
+ match = re.search(r'Query \(id=(.*?)\)', profile_text)
+ assert match is not None
+ query_id = match.group(1)
+
+ print("Query Id: {0}".format(query_id))
+ profile_lines = profile_text.split("\n")
+
+ # Force Impala to process the inserts to the completed queries table.
+ client.execute("refresh " + query_tbl)
+
+ # Assert the query was written correctly to the query log table.
+ if max_row_size is not None:
+ client.set_configuration_option("MAX_ROW_SIZE", max_row_size)
+ sql_results = client.execute("select * from {0} where query_id='{1}'".format(
+ query_tbl, query_id))
+ assert sql_results.success
+ assert len(sql_results.data) == 1, "did not find query in completed queries table"
+
+ # Assert the expected columns were included.
+ assert len(sql_results.data) == 1
+ assert len(sql_results.column_labels) == EXPECTED_QUERY_COLS
+ data = sql_results.data[0].split("\t")
+ assert len(data) == len(sql_results.column_labels)
+
+ # Cluster ID
+ index = 0
+ assert sql_results.column_labels[index] == CLUSTER_ID
+ ret_data[CLUSTER_ID] = data[index]
+ assert data[index] == expected_cluster_id, "cluster id incorrect"
+
+ # Query ID
+ index += 1
+ assert sql_results.column_labels[index] == QUERY_ID
+ ret_data[QUERY_ID] = data[index]
+ assert data[index] == query_id
+
+ # Session ID
+ index += 1
+ assert sql_results.column_labels[index] == SESSION_ID
+ ret_data[SESSION_ID] = data[index]
+ session_id = re.search(r'\n\s+Session ID:\s+(.*)\n', profile_text)
+ assert session_id is not None
+ assert data[index] == session_id.group(1), "session id incorrect"
+
+ # Session Type
+ index += 1
+ assert sql_results.column_labels[index] == SESSION_TYPE
+ ret_data[SESSION_TYPE] = data[index]
+ session_type = re.search(r'\n\s+Session Type:\s+(.*)\n', profile_text)
+ assert session_type is not None
+ assert data[index] == session_type.group(1), "session type incorrect"
+
+ # HS2 Protocol Version
+ index += 1
+ assert sql_results.column_labels[index] == HIVESERVER2_PROTOCOL_VERSION
+ ret_data[HIVESERVER2_PROTOCOL_VERSION] = data[index]
+ if session_type.group(1) == "HIVESERVER2":
+ hs2_ver = re.search(r'\n\s+HiveServer2 Protocol Version:\s+(.*)', profile_text)
+ assert hs2_ver is not None
+ assert data[index] == "HIVE_CLI_SERVICE_PROTOCOL_{0}".format(hs2_ver.group(1))
+ else:
+ assert data[index] == ""
+
+ # Database User
+ index += 1
+ assert sql_results.column_labels[index] == DB_USER
+ ret_data[DB_USER] = data[index]
+ user = re.search(r'\n\s+User:\s+(.*?)\n', profile_text)
+ assert user is not None
+ assert data[index] == user.group(1), "db user incorrect"
+
+ # Connected Database User
+ index += 1
+ assert sql_results.column_labels[index] == DB_USER_CONNECTION
+ ret_data[DB_USER_CONNECTION] = data[index]
+ db_user = re.search(r'\n\s+Connected User:\s+(.*?)\n', profile_text)
+ assert db_user is not None
+ assert data[index] == db_user.group(1), "db user connection incorrect"
+
+ # Database Name
+ index += 1
+ assert sql_results.column_labels[index] == DB_NAME
+ ret_data[DB_NAME] = data[index]
+ default_db = re.search(r'\n\s+Default Db:\s+(.*?)\n', profile_text)
+ assert default_db is not None
+ assert data[index] == default_db.group(1), "database name incorrect"
+
+ # Coordinator
+ index += 1
+ assert sql_results.column_labels[index] == IMPALA_COORDINATOR
+ ret_data[IMPALA_COORDINATOR] = data[index]
+ coordinator = re.search(r'\n\s+Coordinator:\s+(.*?)\n', profile_text)
+ assert coordinator is not None
+ assert data[index] == coordinator.group(1), "impala coordinator incorrect"
+
+ # Query Status (can be multiple lines if the query errored)
+ index += 1
+ assert sql_results.column_labels[index] == QUERY_STATUS
+ ret_data[QUERY_STATUS] = data[index]
+ query_status = re.search(r'\n\s+Query Status:\s+(.*?)\n\s+Impala Version', profile_text,
+ re.DOTALL)
+ assert query_status is not None
+ assert data[index] == query_status.group(1), "query status incorrect"
+
+ # Query State
+ index += 1
+ assert sql_results.column_labels[index] == QUERY_STATE
+ ret_data[QUERY_STATE] = data[index]
+ query_state = re.search(r'\n\s+Query State:\s+(.*?)\n', profile_text)
+ assert query_state is not None
+ query_state_value = query_state.group(1)
+ assert data[index] == query_state_value, "query state incorrect"
+
+ # Impala Query End State
+ index += 1
+ assert sql_results.column_labels[index] == IMPALA_QUERY_END_STATE
+ ret_data[IMPALA_QUERY_END_STATE] = data[index]
+ impala_query_state = re.search(r'\n\s+Impala Query State:\s+(.*?)\n', profile_text)
+ assert impala_query_state is not None
+ assert data[index] == impala_query_state.group(1), "impala query end state incorrect"
+
+ # Query Type
+ index += 1
+ assert sql_results.column_labels[index] == QUERY_TYPE
+ ret_data[QUERY_TYPE] = data[index]
+ if query_state_value == "EXCEPTION":
+ assert data[index] == "UNKNOWN", "query type incorrect"
+ else:
+ query_type = re.search(r'\n\s+Query Type:\s+(.*?)\n', profile_text)
+ assert query_type is not None
+ assert data[index] == query_type.group(1), "query type incorrect"
+ query_type = query_type.group(1)
+
+ # Client Network Address
+ index += 1
+ assert sql_results.column_labels[index] == NETWORK_ADDRESS
+ ret_data[NETWORK_ADDRESS] = data[index]
+ network_address = re.search(r'\n\s+Network Address:\s+(.*?)\n', profile_text)
+ assert network_address is not None
+ assert data[index] == network_address.group(1), "network address incorrect"
+
+ # offset from UTC
+ utc_now = datetime.utcnow().replace(microsecond=0, second=0)
+ local_now = datetime.now().replace(microsecond=0, second=0)
+ utc_offset = utc_now - local_now
+
+ # Start Time
+ index += 1
+ assert sql_results.column_labels[index] == START_TIME_UTC
+ ret_data[START_TIME_UTC] = data[index]
+ start_time = re.search(r'\n\s+Start Time:\s+(.*?)\n', profile_text)
+ assert start_time is not None
+ start_time_obj = datetime.strptime(start_time.group(1)[:-3], "%Y-%m-%d %H:%M:%S.%f")
+ start_time_obj_utc = start_time_obj + utc_offset
+ assert data[index][:-3] == start_time_obj_utc.strftime("%Y-%m-%d %H:%M:%S.%f"), \
+ "start time incorrect"
+
+ # End Time (not in table, but needed for duration calculation)
+ end_time = re.search(r'\n\s+End Time:\s+(.*?)\n', profile_text)
+ assert end_time is not None
+ end_time_obj = datetime.strptime(end_time.group(1)[:-3], "%Y-%m-%d %H:%M:%S.%f")
+
+ # Query Duration (allow values that are within 1 second)
+ index += 1
+ assert sql_results.column_labels[index] == TOTAL_TIME_NS
+ ret_data[TOTAL_TIME_NS] = data[index]
+ duration = end_time_obj - start_time_obj
+ min_allowed = int(duration.total_seconds() * 1000000000) - 1
+ max_allowed = min_allowed + 2
+ assert min_allowed <= int(data[index]) <= max_allowed, "total time incorrect"
+
+ # Query Options Set By Configuration
+ index += 1
+ assert sql_results.column_labels[index] == QUERY_OPTS_CONFIG
+ ret_data[QUERY_OPTS_CONFIG] = data[index]
+ if query_state_value == "EXCEPTION":
+ assert data[index] != "", "query options set by config incorrect"
+ else:
+ query_opts = re.search(r'\n\s+Query Options \(set by configuration\):\s+(.*?)\n',
+ profile_text)
+ assert query_opts is not None
+ assert data[index] == query_opts.group(1), "query opts set by config incorrect"
+
+ # Resource Pool
+ index += 1
+ assert sql_results.column_labels[index] == RESOURCE_POOL
+ ret_data[RESOURCE_POOL] = data[index]
+ if query_state_value == "EXCEPTION":
+ assert data[index] == "", "resource pool incorrect"
+ else:
+ if query_type != "DDL":
+ req_pool = re.search(r'\n\s+Request Pool:\s+(.*?)\n', profile_text)
+ assert req_pool is not None
+ assert data[index] == req_pool.group(1), "request pool incorrect"
+ else:
+ assert data[index] == "", "request pool not empty"
+
+ # Per-host Memory Estimate
+ index += 1
+ assert sql_results.column_labels[index] == PER_HOST_MEM_ESTIMATE
+ ret_data[PER_HOST_MEM_ESTIMATE] = data[index]
+ if query_state_value == "EXCEPTION":
+ assert data[index] == "0", "per-host memory estimate incorrect"
+ else:
+ if query_type != "DDL":
+ perhost_mem_est = re.search(r'\nPer-Host Resource Estimates:\s+Memory\=(.*?)\n',
+ profile_text)
+ assert perhost_mem_est is not None
+ assert_byte_str(expected_str=perhost_mem_est.group(1), actual_bytes=data[index],
+ msg="per-host memory estimate incorrect", unit_combined=True)
+ else:
+ assert data[index] == "0", "per-host memory estimate not 0"
+
+ # Dedicated Coordinator Memory Estimate
+ # This value is different because it is the minimum of the query option
+ # MAX_MEM_ESTIMATE_FOR_ADMISSION or a calculation that includes a 100mb buffer.
+ # Thus, callers must specify if the query being asserted had that option set.
+ index += 1
+ assert sql_results.column_labels[index] == DEDICATED_COORD_MEM_ESTIMATE
+ ret_data[DEDICATED_COORD_MEM_ESTIMATE] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert data[index] == "0", "dedicated coordinator memory estimate incorrect"
+ elif query_type == "DML":
+ assert data[index] == str(DEDICATED_COORD_SAFETY_BUFFER_BYTES), \
+ "dedicated coordinator memory estimate incorrect"
+ else:
+ if max_mem_for_admission is not None:
+ # The MAX_MEM_ESTIMATE_FOR_ADMISSION query option was specified, thus that should
+ # be the value that was written to the database.
+ assert str(max_mem_for_admission) == data[index], \
+ "dedicated coordinator memory estimate incorrect"
+ else:
+ root_mem = re.search(r'\n\nF\d+:PLAN FRAGMENT.*?mem-estimate=(\S+?) mem',
+ profile_text, re.DOTALL)
+ assert root_mem is not None, "dedicated coordinator memory estimate incorrect"
+ buffer = DEDICATED_COORD_SAFETY_BUFFER_BYTES
+ assert_byte_str(expected_str=root_mem.group(1),
+ actual_bytes=int(data[index]) - buffer,
+ msg="dedicated coordinator memory estimate incorrect", unit_combined=True)
+
+ # Per-Host Fragment Instances
+ index += 1
+ assert sql_results.column_labels[index] == PER_HOST_FRAGMENT_INSTANCES
+ ret_data[PER_HOST_FRAGMENT_INSTANCES] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert data[index] == "", "per-host fragment instances incorrect"
+ else:
+ perhost_frags = re.search(r'\n\s+Per Host Number of Fragment Instances:\s+(.*?)\n',
+ profile_text)
+ assert perhost_frags is not None
+ assert data[index] == ",".join(sorted(perhost_frags.group(1).replace("(", "=")
+ .replace(")", "").split(" "))), "per-host fragment instances incorrect"
+
+ # Backends Count
+ index += 1
+ assert sql_results.column_labels[index] == BACKENDS_COUNT
+ ret_data[BACKENDS_COUNT] = data[index]
+ num_bck = re.search(r'\n\s+\- NumBackends:\s+(\d+)', profile_text)
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert num_bck is None
+ assert data[index] == "0", "backends count incorrect"
+ else:
+ assert num_bck is not None
+ assert data[index] == num_bck.group(1), "backends count incorrect"
+
+ # Admission Result
+ index += 1
+ assert sql_results.column_labels[index] == ADMISSION_RESULT
+ ret_data[ADMISSION_RESULT] = data[index]
+ adm_result = re.search(r'\n\s+Admission result:\s+(.*?)\n', profile_text)
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert adm_result is None
+ assert data[index] == "", "admission result incorrect"
+ else:
+ assert adm_result is not None
+ assert data[index] == adm_result.group(1), "admission result incorrect"
+
+ # Cluster Memory Admitted
+ index += 1
+ assert sql_results.column_labels[index] == CLUSTER_MEMORY_ADMITTED
+ ret_data[CLUSTER_MEMORY_ADMITTED] = data[index]
+ clust_mem = re.search(r'\n\s+Cluster Memory Admitted:\s+(.*?)\n', profile_text)
+ if query_state_value == "EXCEPTION":
+ assert clust_mem is None
+ else:
+ if query_type != "DDL":
+ assert clust_mem is not None
+ assert_byte_str(expected_str=clust_mem.group(1), actual_bytes=data[index],
+ msg="cluster memory admitted incorrect")
+ else:
+ assert data[index] == "0", "cluster memory not zero"
+
+ # Executor Group
+ index += 1
+ assert sql_results.column_labels[index] == EXECUTOR_GROUP
+ ret_data[EXECUTOR_GROUP] = data[index]
+ exec_group = re.search(r'\n\s+Executor Group:\s+(.*?)\n', profile_text)
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert exec_group is None
+ assert data[index] == "", "executor group should not have been found"
+ else:
+ assert exec_group is not None
+ assert data[index] == exec_group.group(1), "executor group incorrect"
+
+ # Executor Groups
+ index += 1
+ assert sql_results.column_labels[index] == EXECUTOR_GROUPS
+ ret_data[EXECUTOR_GROUPS] = data[index]
+ exec_groups = re.search(r'\n\s+(Executor group \d+:.*?)\n\s+ImpalaServer', profile_text,
+ re.DOTALL)
+ if query_state_value == "EXCEPTION":
+ assert exec_groups is None, "executor groups should not have been found"
+ else:
+ assert exec_groups is not None
+ dedent_str = re.sub(r'^\s{6}', '', exec_groups.group(1), flags=re.MULTILINE)
+ assert data[index] == dedent_str, "executor groups incorrect"
+
+ # Exec Summary
+ index += 1
+ assert sql_results.column_labels[index] == EXEC_SUMMARY
+ ret_data[EXEC_SUMMARY] = data[index]
+ exec_sum = re.search(r'\n\s+ExecSummary:\s*\n(.*)\n\s+Errors', profile_text, re.DOTALL)
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert exec_sum is None
+ assert data[index] == ""
+ else:
+ assert exec_sum is not None
+ assert data[index] == exec_sum.group(1)
+
+ # Rows Fetched
+ index += 1
+ assert sql_results.column_labels[index] == NUM_ROWS_FETCHED
+ ret_data[NUM_ROWS_FETCHED] = data[index]
+ rows_fetched = re.search(r'\n\s+\-\s+NumRowsFetched:\s+\S+\s+\((\d+)\)', profile_text)
+ if query_state_value == "EXCEPTION":
+ assert rows_fetched is None
+ else:
+ assert rows_fetched is not None
+ assert data[index] == rows_fetched.group(1)
+
+ # Row Materialization Rate
+ index += 1
+ assert sql_results.column_labels[index] == ROW_MATERIALIZATION_ROWS_PER_SEC
+ ret_data[ROW_MATERIALIZATION_ROWS_PER_SEC] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL" or query_type == 'DML':
+ assert data[index] == "0", "row materialization rate incorrect"
+ else:
+ row_mat = re.search(r'\n\s+\-\s+RowMaterializationRate:\s+(\S+)\s+([MK])?',
+ profile_text)
+ assert row_mat is not None
+ tolerance = int(data[index]) * 0.005
+ expected_row_mat = 0
+ if row_mat.group(2) == "K":
+ expected_row_mat = int(float(row_mat.group(1)) * 1000)
+ elif row_mat.group(2) == "M":
+ expected_row_mat = int(float(row_mat.group(1)) * 1000000)
+ else:
+ expected_row_mat = int(float(row_mat.group(1)))
+ assert expected_row_mat - tolerance <= int(data[index]) \
+ <= expected_row_mat + tolerance, "row materialization rate incorrect"
+
+ # Row Materialization Time
+ index += 1
+ assert sql_results.column_labels[index] == ROW_MATERIALIZATION_TIME_NS
+ ret_data[ROW_MATERIALIZATION_TIME_NS] = data[index]
+ row_mat_tmr = re.search(r'\n\s+\-\s+RowMaterializationTimer:\s+(.*?)\n', profile_text)
+ if query_state_value == "EXCEPTION":
+ assert row_mat_tmr is None
+ elif query_type == "DDL" or query_type == 'DML':
+ assert row_mat_tmr is not None
+ assert row_mat_tmr.group(1) == "0.000ns", "row materialization timer incorrect"
+ else:
+ assert row_mat_tmr is not None
+ assert_time_str(row_mat_tmr.group(1), (int(data[index])),
+ "row materialization time incorrect")
+
+ # Compressed Bytes Spilled
+ index += 1
+ assert sql_results.column_labels[index] == COMPRESSED_BYTES_SPILLED
+ ret_data[COMPRESSED_BYTES_SPILLED] = data[index]
+ scratch_bytes_total = 0
+ for sbw in re.findall(r'\n\s+\-\s+ScratchBytesWritten:.*?\((\d+)\)', profile_text):
+ scratch_bytes_total += int(sbw)
+ assert int(data[index]) == scratch_bytes_total
+
+ # Parse out only the query timeline.
+ timeline = re.search(r'\n\s+Query Timeline:(.*?)\n\s+Frontend', profile_text, re.DOTALL)
+ assert timeline is not None, "query timeline not found"
+ timeline = timeline.group(1)
+
+ # Event Timeline Planning Finished
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_PLANNING_FINISHED
+ ret_data[EVENT_PLANNING_FINISHED] = data[index]
+ if query_state_value == "EXCEPTION":
+ assert data[index] == "0", "planning finished event incorrect"
+ else:
+ event = re.search(r'\n\s+\-\s+Planning finished:\s+(\S+)', timeline)
+ assert event is not None, "planning finished event missing"
+ assert_time_str(event.group(1), data[index], "planning finished event incorrect")
+
+ # Event Timeline Submit for Admission
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_SUBMIT_FOR_ADMISSION
+ ret_data[EVENT_SUBMIT_FOR_ADMISSION] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert data[index] == "0", "submit for admission event incorrect"
+ else:
+ event = re.search(r'\n\s+\-\s+Submit for admission:\s+(\S+)', timeline)
+ assert event is not None, "submit for admission event missing"
+ assert_time_str(event.group(1), data[index], "submit for admission event incorrect")
+
+ # Event Timeline Completed Admission
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_COMPLETED_ADMISSION
+ ret_data[EVENT_COMPLETED_ADMISSION] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert data[index] == "0", "completed admission event incorrect"
+ else:
+ event = re.search(r'\n\s+\-\s+Completed admission:\s+(\S+)', timeline)
+ assert event is not None, "completed admission event missing"
+ assert_time_str(event.group(1), data[index], "completed admission event incorrect")
+
+ # Event Timeline All Backends Started
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_ALL_BACKENDS_STARTED
+ ret_data[EVENT_ALL_BACKENDS_STARTED] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert data[index] == "0", "all backends started event incorrect"
+ else:
+ event = re.search(r'\n\s+\-\s+All \d+ execution backends \(\d+ fragment instances\)'
+ r' started:\s+(\S+)', timeline)
+ assert event is not None, "all backends started event missing"
+ assert_time_str(event.group(1), data[index], "all backends started event incorrect")
+
+ # Event Timeline Rows Available
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_ROWS_AVAILABLE
+ ret_data[EVENT_ROWS_AVAILABLE] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DML":
+ assert data[index] == "0", "rows available event incorrect"
+ else:
+ event = re.search(r'\n\s+\-\s+Rows available:\s+(\S+)', timeline)
+ assert event is not None, "rows available event missing"
+ assert_time_str(event.group(1), data[index], "rows available event incorrect")
+
+ # Event Timeline First Row Fetched
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_FIRST_ROW_FETCHED
+ ret_data[EVENT_FIRST_ROW_FETCHED] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL" or query_type == "DML":
+ assert data[index] == "0", "first row fetched event incorrect"
+ else:
+ event = re.search(r'\n\s+\-\s+First row fetched:\s+(\S+)', timeline)
+ assert event is not None, "first row fetched event missing"
+ assert_time_str(event.group(1), data[index], "first row fetched event incorrect")
+
+ # Event Timeline Last Row Fetched
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_LAST_ROW_FETCHED
+ ret_data[EVENT_LAST_ROW_FETCHED] = data[index]
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert data[index] == "0", "last row fetched event incorrect"
+ else:
+ event = re.search(r'\n\s+\-\s+Last row fetched:\s+(\S+)', timeline)
+ assert event is not None, "last row fetched event missing"
+ assert_time_str(event.group(1), data[index], "last row fetched event incorrect")
+
+ # Event Timeline Unregister Query
+ index += 1
+ assert sql_results.column_labels[index] == EVENT_UNREGISTER_QUERY
+ ret_data[EVENT_UNREGISTER_QUERY] = data[index]
+ event = re.search(r'\n\s+\-\s+Unregister query:\s+(\S+)', timeline)
+ assert event is not None, "unregister query event missing"
+ assert_time_str(event.group(1), data[index], "unregister query event incorrect")
+
+ # Read IO Wait Total
+ index += 1
+ assert sql_results.column_labels[index] == READ_IO_WAIT_TOTAL_NS
+ ret_data[READ_IO_WAIT_TOTAL_NS] = data[index]
+ total_read_wait = 0
+ if (query_state_value != "EXCEPTION" and query_type == "QUERY") or data[index] != "0":
+ re_wait_time = re.compile(r'^\s+\-\s+ScannerIoWaitTime:\s+(.*?)$')
+ read_waits = assert_scan_node_metrics(re_wait_time, profile_lines)
+ for r in read_waits:
+ total_read_wait += int(convert_to_nanos(r))
+
+ tolerance = total_read_wait * 0.001
+
+ assert total_read_wait - tolerance <= int(data[index]) <= \
+ total_read_wait + tolerance, "read io wait time total incorrect"
+ else:
+ assert data[index] == "0"
+
+ # Read IO Wait Average
+ index += 1
+ assert sql_results.column_labels[index] == READ_IO_WAIT_MEAN_NS
+ ret_data[READ_IO_WAIT_MEAN_NS] = data[index]
+ if (query_state_value != "EXCEPTION" and query_type == "QUERY"
+ and len(read_waits) != 0) or data[index] != "0":
+ avg_read_wait = int(total_read_wait / len(read_waits))
+ assert avg_read_wait - tolerance <= int(data[index]) <= avg_read_wait + tolerance, \
+ "read io wait time average incorrect"
+ else:
+ assert data[index] == "0"
+
+ # Total Bytes Read From Cache
+ index += 1
+ assert sql_results.column_labels[index] == BYTES_READ_CACHE_TOTAL
+ ret_data[BYTES_READ_CACHE_TOTAL] = data[index]
+ if (query_state_value != "EXCEPTION" and query_type == "QUERY") or data[index] != "0":
+ re_cache_read = re.compile(r'^\s+\-\s+DataCacheHitBytes:\s+.*?\((\d+)\)$')
+ read_from_cache = assert_scan_node_metrics(re_cache_read, profile_lines)
+
+ total_read = 0
+ for r in read_from_cache:
+ total_read += int(r)
+ assert total_read == int(data[index]), "bytes read from cache total incorrect"
+ else:
+ assert data[index] == "0"
+
+ # Total Bytes Read
+ index += 1
+ assert sql_results.column_labels[index] == BYTES_READ_TOTAL
+ ret_data[BYTES_READ_TOTAL] = data[index]
+ bytes_read = re.search(r'\n\s+\-\s+TotalBytesRead:\s+.*?\((\d+)\)\n', profile_text)
+ if query_state_value != "EXCEPTION" and query_type == "QUERY":
+ assert bytes_read is not None, "total bytes read missing"
+ if bytes_read is not None:
+ assert data[index] == bytes_read.group(1), "total bytes read incorrect"
+
+ # Calculate all peak memory usage stats by scraping the query profile.
+ peak_mem_cnt = 0
+ min_peak_mem = 0
+ max_peak_mem = 0
+ total_peak_mem = 0
+ for peak_mem in re.findall(r'\n\s+Per Node Peak Memory Usage:(.*?)\n', profile_text):
+ for node in re.findall(r'\s+.*?:\d+\((.*?)\)', peak_mem):
+ peak_mem_cnt += 1
+ conv = convert_to_bytes(node)
+ total_peak_mem += conv
+ if conv < min_peak_mem or min_peak_mem == 0:
+ min_peak_mem = conv
+ if conv > max_peak_mem:
+ max_peak_mem = conv
+ if query_state_value != "EXCEPTION" and query_type != "DDL":
+ assert peak_mem_cnt > 0, "did not find per node peak memory usage"
+
+ # Per Node Peak Memory Usage Min
+ index += 1
+ assert sql_results.column_labels[index] == PERNODE_PEAK_MEM_MIN
+ ret_data[PERNODE_PEAK_MEM_MIN] = data[index]
+ tolerance = int(min_peak_mem * 0.005)
+ assert min_peak_mem - tolerance <= int(data[index]) <= min_peak_mem + tolerance, \
+ "pernode peak memory minimum incorrect"
+
+ # Per Node Peak Memory Usage Max
+ index += 1
+ assert sql_results.column_labels[index] == PERNODE_PEAK_MEM_MAX
+ ret_data[PERNODE_PEAK_MEM_MAX] = data[index]
+ tolerance = int(max_peak_mem * 0.005)
+ assert max_peak_mem - tolerance <= int(data[index]) <= max_peak_mem + tolerance, \
+ "pernode peak memory maximum incorrect"
+
+ # Per Node Peak Memory Usage Mean
+ index += 1
+ assert sql_results.column_labels[index] == PERNODE_PEAK_MEM_MEAN
+ ret_data[PERNODE_PEAK_MEM_MEAN] = data[index]
+ mean_peak_mem = 0
+ if peak_mem_cnt > 0:
+ mean_peak_mem = int(total_peak_mem / peak_mem_cnt)
+ tolerance = int(max_peak_mem * 0.005)
+ assert mean_peak_mem - tolerance <= int(data[index]) <= mean_peak_mem + tolerance, \
+ "pernode peak memory mean incorrect"
+
+ # SQL statement
+ index += 1
+ assert sql_results.column_labels[index] == SQL
+ ret_data[SQL] = data[index]
+ sql_stmt = re.search(r'\n\s+Sql Statement:\s+(.*?)\n', profile_text)
+ assert sql_stmt is not None
+ assert data[index] == sql_stmt.group(1), "sql incorrect"
+
+ # Query Plan
+ index += 1
+ assert sql_results.column_labels[index] == PLAN
+ ret_data[PLAN] = data[index]
+ plan = re.search(r'\n\s+Plan:\s*\n(.*)\n\s+Estimated Per-Host Mem', profile_text,
+ re.DOTALL)
+ if query_state_value == "EXCEPTION" or query_type == "DDL":
+ assert plan is None
+ assert data[index] == ""
+ else:
+ assert plan is not None
+ assert data[index] == plan.group(1)
+
+ return ret_data
+# function assert_query
+
+
+def assert_scan_node_metrics(re_metric, profile_lines):
+ """Retrieves metrics reported under HDFS_SCAN_NODEs removing any metrics from
+ Averaged Fragments. The provided re_metric must be a compiled regular expression
+ with at least one capture group. Returns a list of the contents of the first
+ capture group in the re_metrics regular expression for all matching metrics."""
+ metrics = []
+
+ re_in_scan = re.compile(r'^\s+HDFS_SCAN_NODE')
+ re_avg_fgmt = re.compile(r'^(\s+)Averaged Fragment')
+ in_scan = False
+ in_avg_fgmt = 0
+ for line in profile_lines:
+ avg_fmt_res = re_avg_fgmt.search(line)
+ if avg_fmt_res is not None:
+ # Averaged Fragments sometimes have HDFS_SCAN_NODEs which must be skipped.
+ in_avg_fgmt = len(avg_fmt_res.group(1))
+ elif in_avg_fgmt > 0 and line[in_avg_fgmt + 1] != " ":
+ # Found a line at the same indentation as the previous Averaged Fragement, thus
+ # we successfully skipped over any HDFS_SCAN_NODEs if they existed.
+ in_avg_fgmt = 0
+ elif in_avg_fgmt == 0 and re_in_scan.match(line) is not None:
+ # Found a HDFS_SCAN_NODE that was not under an Averaged Fragment.
+ in_scan = True
+ elif in_scan:
+ # Search through the HDFS_SCAN_NODE for the metric.
+ res = re_metric.search(line)
+ if res is not None:
+ metrics.append(res.group(1))
+ in_scan = False
+
+ return metrics
+# function assert_scan_node_metrics