Repository: impala
Updated Branches:
  refs/heads/master 24b4ed0b2 -> ff3ddb51a


IMPALA-6482: add EXEC_TIME_LIMIT_S option

This is similar to the QUERY_TIMEOUT_S option and shares most of the
implementation. The difference is that the timeout doesn't reset at
any point.

The time limit is measured from the start of query execution,
after the query is admitted, so planning, scheduling and
time spent in admission control is not counted towards the time limit.

Also fix validation of the related QUERY_TIMEOUT_S option, which
previously could ignore invalid input.

Testing:
Added tests for various permutations:
* With and without query_timeout_s set
* With and without result fetching keeping the query active

Change-Id: Id81772ee223ffb64746e241027a5a734a811e1b8
Reviewed-on: http://gerrit.cloudera.org:8080/9227
Reviewed-by: Tim Armstrong <tarmstr...@cloudera.com>
Tested-by: Impala Public Jenkins


Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/ff3ddb51
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/ff3ddb51
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/ff3ddb51

Branch: refs/heads/master
Commit: ff3ddb51a49dcd38da2e0d1a09a1cc9405649f40
Parents: 24b4ed0
Author: Tim Armstrong <tarmstr...@cloudera.com>
Authored: Tue Feb 6 00:15:22 2018 -0800
Committer: Impala Public Jenkins <impala-public-jenk...@gerrit.cloudera.org>
Committed: Fri Feb 23 04:22:28 2018 +0000

----------------------------------------------------------------------
 be/src/service/client-request-state.cc        |  16 +-
 be/src/service/client-request-state.h         |  45 ++++--
 be/src/service/impala-server.cc               | 110 +++++++++-----
 be/src/service/impala-server.h                |  33 ++++-
 be/src/service/query-options-test.cc          |   4 +-
 be/src/service/query-options.cc               |  27 +++-
 be/src/service/query-options.h                |   3 +-
 common/thrift/ImpalaInternalService.thrift    |   7 +-
 common/thrift/ImpalaService.thrift            |   7 +-
 tests/custom_cluster/test_query_expiration.py | 163 +++++++++++++++++----
 10 files changed, 304 insertions(+), 111 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/be/src/service/client-request-state.cc
----------------------------------------------------------------------
diff --git a/be/src/service/client-request-state.cc 
b/be/src/service/client-request-state.cc
index 64ab950..282f736 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -68,30 +68,16 @@ ClientRequestState::ClientRequestState(
     ImpalaServer* server, shared_ptr<ImpalaServer::SessionState> session)
   : query_ctx_(query_ctx),
     last_active_time_ms_(numeric_limits<int64_t>::max()),
-    ref_count_(0L),
     child_query_executor_(new ChildQueryExecutor),
     exec_env_(exec_env),
-    is_block_on_wait_joining_(false),
     session_(session),
-    schedule_(NULL),
-    coord_(NULL),
-    result_cache_max_size_(-1),
     // Profile is assigned name w/ id after planning
     profile_(RuntimeProfile::Create(&profile_pool_, "Query")),
     server_profile_(RuntimeProfile::Create(&profile_pool_, "ImpalaServer")),
     summary_profile_(RuntimeProfile::Create(&profile_pool_, "Summary")),
-    is_cancelled_(false),
-    eos_(false),
-    query_state_(beeswax::QueryState::CREATED),
-    user_has_profile_access_(true),
-    current_batch_(NULL),
-    current_batch_row_(0),
-    num_rows_fetched_(0),
-    fetched_rows_(false),
     frontend_(frontend),
     parent_server_(server),
-    start_time_us_(UnixMicros()),
-    end_time_us_(0LL) {
+    start_time_us_(UnixMicros()) {
 #ifndef NDEBUG
   profile_->AddInfoString("DEBUG MODE WARNING", "Query profile created while 
running a "
       "DEBUG build of Impala. Use RELEASE builds to measure query 
performance.");

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/be/src/service/client-request-state.h
----------------------------------------------------------------------
diff --git a/be/src/service/client-request-state.h 
b/be/src/service/client-request-state.h
index d93512e..0341ec5 100644
--- a/be/src/service/client-request-state.h
+++ b/be/src/service/client-request-state.h
@@ -211,6 +211,16 @@ class ClientRequestState {
     return ref_count_ > 0;
   }
 
+  bool is_expired() const {
+    boost::lock_guard<boost::mutex> l(expiration_data_lock_);
+    return is_expired_;
+  }
+
+  void set_expired() {
+    boost::lock_guard<boost::mutex> l(expiration_data_lock_);
+    is_expired_ = true;
+  }
+
   RuntimeProfile::EventSequence* query_events() const { return query_events_; }
   RuntimeProfile* summary_profile() { return summary_profile_; }
 
@@ -224,8 +234,8 @@ class ClientRequestState {
   /// See "Locking" in the class comment for lock acquisition order.
   boost::mutex fetch_rows_lock_;
 
-  /// Protects last_active_time_ms_ and ref_count_. Only held during short 
function calls
-  /// - no other locks should be acquired while holding this lock.
+  /// Protects last_active_time_ms_, ref_count_ and is_expired_. Only held 
during short
+  /// function calls - no other locks should be acquired while holding this 
lock.
   mutable boost::mutex expiration_data_lock_;
 
   /// Stores the last time that the query was actively doing work, in Unix 
milliseconds.
@@ -234,7 +244,10 @@ class ClientRequestState {
   /// ref_count_ > 0 if Impala is currently performing work on this query's 
behalf. Every
   /// time a client instructs Impala to do work on behalf of this query, the 
ref count is
   /// increased, and decreased once that work is completed.
-  uint32_t ref_count_;
+  uint32_t ref_count_ = 0;
+
+  /// True if the query expired by timing out.
+  bool is_expired_ = false;
 
   /// Executor for any child queries (e.g. compute stats subqueries). Always 
non-NULL.
   const boost::scoped_ptr<ChildQueryExecutor> child_query_executor_;
@@ -258,7 +271,7 @@ class ClientRequestState {
   ConditionVariable block_on_wait_cv_;
 
   /// Used in conjunction with block_on_wait_cv_ to make BlockOnWait() 
thread-safe.
-  bool is_block_on_wait_joining_;
+  bool is_block_on_wait_joining_ = false;
 
   /// Session that this query is from
   std::shared_ptr<ImpalaServer::SessionState> session_;
@@ -284,7 +297,7 @@ class ClientRequestState {
   boost::scoped_ptr<QueryResultSet> result_cache_;
 
   /// Max size of the result_cache_ in number of rows. A value <= 0 means no 
caching.
-  int64_t result_cache_max_size_;
+  int64_t result_cache_max_size_ = -1;
 
   ObjectPool profile_pool_;
 
@@ -320,25 +333,25 @@ class ClientRequestState {
 
   RuntimeProfile::EventSequence* query_events_;
 
-  bool is_cancelled_; // if true, Cancel() was called.
-  bool eos_;  // if true, there are no more rows to return
+  bool is_cancelled_ = false; // if true, Cancel() was called.
+  bool eos_ = false;  // if true, there are no more rows to return
   /// We enforce the invariant that query_status_ is not OK iff query_state_ 
is EXCEPTION,
   /// given that lock_ is held. query_state_ should only be updated using
   /// UpdateQueryState(), to ensure that the query profile is also updated.
-  beeswax::QueryState::type query_state_;
+  beeswax::QueryState::type query_state_ = beeswax::QueryState::CREATED;
   Status query_status_;
   TExecRequest exec_request_;
   /// If true, effective_user() has access to the runtime profile and execution
   /// summary.
-  bool user_has_profile_access_;
+  bool user_has_profile_access_ = true;
   TResultSetMetadata result_metadata_; // metadata for select query
-  RowBatch* current_batch_; // the current row batch; only applicable if coord 
is set
-  int current_batch_row_; // number of rows fetched within the current batch
-  int num_rows_fetched_; // number of rows fetched by client for the entire 
query
+  RowBatch* current_batch_ = nullptr; // the current row batch; only 
applicable if coord is set
+  int current_batch_row_ = 0 ; // number of rows fetched within the current 
batch
+  int num_rows_fetched_ = 0; // number of rows fetched by client for the 
entire query
 
   /// True if a fetch was attempted by a client, regardless of whether a 
result set
   /// (or error) was returned to the client.
-  bool fetched_rows_;
+  bool fetched_rows_ = false;
 
   /// To get access to UpdateCatalog, LOAD, and DDL methods. Not owned.
   Frontend* frontend_;
@@ -348,10 +361,10 @@ class ClientRequestState {
   ImpalaServer* parent_server_;
 
   /// Start/end time of the query, in Unix microseconds.
-  /// end_time_us_ is initialized to 0 in the constructor, which is used to 
indicate
-  /// that the query is not yet done. It is assinged the final value in
+  /// end_time_us_ is initialized to 0, which is used to indicate that the 
query is not
+  /// yet done. It is assinged the final value in
   /// ClientRequestState::Done().
-  int64_t start_time_us_, end_time_us_;
+  int64_t start_time_us_, end_time_us_ = 0;
 
   /// Executes a local catalog operation (an operation that does not need to 
execute
   /// against the catalog service). Includes USE, SHOW, DESCRIBE, and EXPLAIN 
statements.

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/be/src/service/impala-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index c43c850..4452205 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -966,21 +966,31 @@ Status 
ImpalaServer::SetQueryInflight(shared_ptr<SessionState> session_state,
   // Add query to the set that will be unregistered if sesssion is closed.
   session_state->inflight_queries.insert(query_id);
   ++session_state->total_queries;
-  // Set query expiration.
-  int32_t timeout_s = request_state->query_options().query_timeout_s;
-  if (FLAGS_idle_query_timeout > 0 && timeout_s > 0) {
-    timeout_s = min(FLAGS_idle_query_timeout, timeout_s);
+
+  // If the query has a timeout or time limit, schedule checks.
+  int32_t idle_timeout_s = request_state->query_options().query_timeout_s;
+  if (FLAGS_idle_query_timeout > 0 && idle_timeout_s > 0) {
+    idle_timeout_s = min(FLAGS_idle_query_timeout, idle_timeout_s);
   } else {
     // Use a non-zero timeout, if one exists
-    timeout_s = max(FLAGS_idle_query_timeout, timeout_s);
+    idle_timeout_s = max(FLAGS_idle_query_timeout, idle_timeout_s);
   }
-  if (timeout_s > 0) {
+  int32_t exec_time_limit_s = request_state->query_options().exec_time_limit_s;
+  if (idle_timeout_s > 0 || exec_time_limit_s > 0) {
     lock_guard<mutex> l2(query_expiration_lock_);
-    VLOG_QUERY << "Query " << PrintId(query_id) << " has timeout of "
-               << PrettyPrinter::Print(timeout_s * 1000L * 1000L * 1000L,
-                     TUnit::TIME_NS);
-    queries_by_timestamp_.insert(
-        make_pair(UnixMillis() + (1000L * timeout_s), query_id));
+    int64_t now = UnixMillis();
+    if (idle_timeout_s > 0) {
+      VLOG_QUERY << "Query " << PrintId(query_id) << " has idle timeout of "
+                 << PrettyPrinter::Print(idle_timeout_s, TUnit::TIME_S);
+      queries_by_timestamp_.emplace(ExpirationEvent{
+          now + (1000L * idle_timeout_s), query_id, 
ExpirationKind::IDLE_TIMEOUT});
+    }
+    if (exec_time_limit_s > 0) {
+      VLOG_QUERY << "Query " << PrintId(query_id) << " has execution time 
limit of "
+                 << PrettyPrinter::Print(exec_time_limit_s, TUnit::TIME_S);
+      queries_by_timestamp_.emplace(ExpirationEvent{
+          now + (1000L * exec_time_limit_s), query_id, 
ExpirationKind::EXEC_TIME_LIMIT});
+    }
   }
   return Status::OK();
 }
@@ -1847,56 +1857,71 @@ void ImpalaServer::UnregisterSessionTimeout(int32_t 
session_timeout) {
       ExpirationQueue::iterator expiration_event = 
queries_by_timestamp_.begin();
       now = UnixMillis();
       while (expiration_event != queries_by_timestamp_.end()) {
-        // If the last-observed expiration time for this query is still in the 
future, we
-        // know that the true expiration time will be at least that far off. 
So we can
-        // break here and sleep.
-        if (expiration_event->first > now) break;
+        // 'queries_by_timestamp_' is stored in ascending order of deadline so 
we can
+        // break out of the loop and sleep as soon as we see a deadline in the 
future.
+        if (expiration_event->deadline > now) break;
         shared_ptr<ClientRequestState> query_state =
-            GetClientRequestState(expiration_event->second);
-        if (query_state.get() == nullptr) {
-          // Query was deleted some other way.
-          queries_by_timestamp_.erase(expiration_event++);
+            GetClientRequestState(expiration_event->query_id);
+        if (query_state == nullptr || query_state->is_expired()) {
+          // Query was deleted or expired already from a previous expiration 
event.
+          expiration_event = queries_by_timestamp_.erase(expiration_event);
           continue;
         }
-        // First, check the actual expiration time in case the query has 
updated it
-        // since the last time we looked.
-        int32_t timeout_s = query_state->query_options().query_timeout_s;
-        if (FLAGS_idle_query_timeout > 0 && timeout_s > 0) {
-          timeout_s = min(FLAGS_idle_query_timeout, timeout_s);
+
+        // If the query time limit expired, we must cancel the query.
+        if (expiration_event->kind == ExpirationKind::EXEC_TIME_LIMIT) {
+          int32_t exec_time_limit_s = 
query_state->query_options().exec_time_limit_s;
+          VLOG_QUERY << "Expiring query " << expiration_event->query_id
+                     << " due to execution time limit of " << 
exec_time_limit_s << "s.";
+          const string& err_msg = Substitute(
+              "Query $0 expired due to execution time limit of $1",
+              PrintId(expiration_event->query_id),
+              PrettyPrinter::Print(exec_time_limit_s, TUnit::TIME_S));
+          ExpireQuery(query_state.get(), Status(err_msg));
+          expiration_event = queries_by_timestamp_.erase(expiration_event);
+          continue;
+        }
+        DCHECK(expiration_event->kind == ExpirationKind::IDLE_TIMEOUT)
+            << static_cast<int>(expiration_event->kind);
+
+        // Now check to see if the idle timeout has expired. We must check the 
actual
+        // expiration time in case the query has updated 'last_active_ms' 
since the last
+        // time we looked.
+        int32_t idle_timeout_s = query_state->query_options().query_timeout_s;
+        if (FLAGS_idle_query_timeout > 0 && idle_timeout_s > 0) {
+          idle_timeout_s = min(FLAGS_idle_query_timeout, idle_timeout_s);
         } else {
           // Use a non-zero timeout, if one exists
-          timeout_s = max(FLAGS_idle_query_timeout, timeout_s);
+          idle_timeout_s = max(FLAGS_idle_query_timeout, idle_timeout_s);
         }
-        int64_t expiration = query_state->last_active_ms() + (timeout_s * 
1000L);
+        int64_t expiration = query_state->last_active_ms() + (idle_timeout_s * 
1000L);
         if (now < expiration) {
           // If the real expiration date is in the future we may need to 
re-insert the
           // query's expiration event at its correct location.
-          if (expiration == expiration_event->first) {
+          if (expiration == expiration_event->deadline) {
             // The query hasn't been updated since it was inserted, so we know 
(by the
             // fact that queries are inserted in-expiration-order initially) 
that it is
             // still the next query to expire. No need to re-insert it.
             break;
           } else {
             // Erase and re-insert with an updated expiration time.
-            TUniqueId query_id = expiration_event->second;
-            queries_by_timestamp_.erase(expiration_event++);
-            queries_by_timestamp_.insert(make_pair(expiration, query_id));
+            TUniqueId query_id = expiration_event->query_id;
+            expiration_event = queries_by_timestamp_.erase(expiration_event);
+            queries_by_timestamp_.emplace(ExpirationEvent{
+                expiration, query_id, ExpirationKind::IDLE_TIMEOUT});
           }
         } else if (!query_state->is_active()) {
           // Otherwise time to expire this query
           VLOG_QUERY
-              << "Expiring query due to client inactivity: " << 
expiration_event->second
-              << ", last activity was at: "
+              << "Expiring query due to client inactivity: "
+              << expiration_event->query_id << ", last activity was at: "
               << ToStringFromUnixMillis(query_state->last_active_ms());
           const string& err_msg = Substitute(
               "Query $0 expired due to client inactivity (timeout is $1)",
-              PrintId(expiration_event->second),
-              PrettyPrinter::Print(timeout_s * 1000000000L, TUnit::TIME_NS));
-
-          cancellation_thread_pool_->Offer(
-              CancellationWork(expiration_event->second, Status(err_msg), 
false));
-          queries_by_timestamp_.erase(expiration_event++);
-          ImpaladMetrics::NUM_QUERIES_EXPIRED->Increment(1L);
+              PrintId(expiration_event->query_id),
+              PrettyPrinter::Print(idle_timeout_s, TUnit::TIME_S));
+          ExpireQuery(query_state.get(), Status(err_msg));
+          expiration_event = queries_by_timestamp_.erase(expiration_event);
         } else {
           // Iterator is moved on in every other branch.
           ++expiration_event;
@@ -1911,6 +1936,13 @@ void ImpalaServer::UnregisterSessionTimeout(int32_t 
session_timeout) {
   }
 }
 
+void ImpalaServer::ExpireQuery(ClientRequestState* crs, const Status& status) {
+  DCHECK(!status.ok());
+  cancellation_thread_pool_->Offer(CancellationWork(crs->query_id(), status, 
false));
+  ImpaladMetrics::NUM_QUERIES_EXPIRED->Increment(1L);
+  crs->set_expired();
+}
+
 Status ImpalaServer::Start(int32_t thrift_be_port, int32_t beeswax_port,
    int32_t hs2_port) {
   exec_env_->SetImpalaServer(this);

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/be/src/service/impala-server.h
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index abe8694..23d4687 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -467,9 +467,10 @@ class ImpalaServer : public ImpalaServiceIf,
   };
 
  private:
+  struct ExpirationEvent;
   friend class ChildQuery;
   friend class ImpalaHttpHandler;
-  friend class SessionState;
+  friend struct SessionState;
 
   boost::scoped_ptr<ImpalaHttpHandler> http_handler_;
 
@@ -816,6 +817,9 @@ class ImpalaServer : public ImpalaServiceIf,
   /// FLAGS_idle_query_timeout seconds.
   [[noreturn]] void ExpireQueries();
 
+  /// Expire 'crs' and cancel it with status 'status'.
+  void ExpireQuery(ClientRequestState* crs, const Status& status);
+
   /// Guards query_log_ and query_log_index_
   boost::mutex query_log_lock_;
 
@@ -1026,16 +1030,31 @@ class ImpalaServer : public ImpalaServiceIf,
   /// acquisition order.
   boost::mutex query_expiration_lock_;
 
-  /// Describes a query expiration event (t, q) where t is the expiration 
deadline in
-  /// seconds, and q is the query ID.
-  typedef std::pair<int64_t, TUniqueId> ExpirationEvent;
+  enum class ExpirationKind {
+    // The query is cancelled if the query has been inactive this long. The 
event may
+    // cancel the query after checking the last active time.
+    IDLE_TIMEOUT,
+    // A hard time limit on query execution. The query is cancelled if this 
event occurs
+    // before the query finishes.
+    EXEC_TIME_LIMIT
+  };
+
+  // Describes a query expiration event where the query identified by 
'query_id' is
+  // checked for expiration when UnixMillis() exceeds 'deadline'.
+  struct ExpirationEvent {
+    int64_t deadline;
+    TUniqueId query_id;
+    ExpirationKind kind;
+  };
 
   /// Comparator that breaks ties when two queries have identical expiration 
deadlines.
   struct ExpirationEventComparator {
     bool operator()(const ExpirationEvent& t1, const ExpirationEvent& t2) {
-      if (t1.first < t2.first) return true;
-      if (t2.first < t1.first) return false;
-      return t1.second < t2.second;
+      if (t1.deadline < t2.deadline) return true;
+      if (t2.deadline < t1.deadline) return false;
+      if (t1.query_id < t2.query_id) return true;
+      if (t2.query_id < t1.query_id) return false;
+      return t1.kind < t2.kind;
     }
   };
 

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/be/src/service/query-options-test.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options-test.cc 
b/be/src/service/query-options-test.cc
index 06be431..3352bd1 100644
--- a/be/src/service/query-options-test.cc
+++ b/be/src/service/query-options-test.cc
@@ -225,7 +225,9 @@ TEST(QueryOptions, SetIntOptions) {
       {MAKE_OPTIONDEF(mt_dop),                         {0, 64}},
       {MAKE_OPTIONDEF(disable_codegen_rows_threshold), {0, I32_MAX}},
       {MAKE_OPTIONDEF(max_num_runtime_filters),        {0, I32_MAX}},
-      {MAKE_OPTIONDEF(batch_size),                     {0, 65536}}
+      {MAKE_OPTIONDEF(batch_size),                     {0, 65536}},
+      {MAKE_OPTIONDEF(query_timeout_s),                {0, I32_MAX}},
+      {MAKE_OPTIONDEF(exec_time_limit_s),              {0, I32_MAX}},
   };
   for (const auto& test_case : case_set) {
     const OptionDef<int32_t>& option_def = test_case.first;

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/be/src/service/query-options.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index 7c76b10..3c56f89 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -295,9 +295,18 @@ Status impala::SetQueryOption(const string& key, const 
string& value,
         query_options->__set_disable_outermost_topn(
             iequals(value, "true") || iequals(value, "1"));
         break;
-      case TImpalaQueryOptions::QUERY_TIMEOUT_S:
-        query_options->__set_query_timeout_s(atoi(value.c_str()));
+      case TImpalaQueryOptions::QUERY_TIMEOUT_S: {
+        StringParser::ParseResult result;
+        const int32_t timeout_s =
+            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), 
&result);
+        if (result != StringParser::PARSE_SUCCESS || timeout_s < 0) {
+          return Status(
+              Substitute("Invalid query timeout: '$0'. "
+                         "Only non-negative numbers are allowed.", value));
+        }
+        query_options->__set_query_timeout_s(timeout_s);
         break;
+      }
       case TImpalaQueryOptions::BUFFER_POOL_LIMIT: {
         int64_t mem;
         RETURN_IF_ERROR(ParseMemValue(value, "buffer pool limit", &mem));
@@ -596,7 +605,7 @@ Status impala::SetQueryOption(const string& key, const 
string& value,
         if (result != StringParser::PARSE_SUCCESS || requested_timeout < 0) {
           return Status(
               Substitute("Invalid idle session timeout: '$0'. "
-                         "Only positive numbers are allowed.", value));
+                         "Only non-negative numbers are allowed.", value));
         }
         query_options->__set_idle_session_timeout(requested_timeout);
         break;
@@ -611,6 +620,18 @@ Status impala::SetQueryOption(const string& key, const 
string& value,
         query_options->__set_compute_stats_min_sample_size(min_sample_size);
         break;
       }
+      case TImpalaQueryOptions::EXEC_TIME_LIMIT_S: {
+        StringParser::ParseResult result;
+        const int32_t time_limit =
+            StringParser::StringToInt<int32_t>(value.c_str(), value.length(), 
&result);
+        if (result != StringParser::PARSE_SUCCESS || time_limit < 0) {
+          return Status(
+              Substitute("Invalid query time limit: '$0'. "
+                         "Only non-negative numbers are allowed.", value));
+        }
+        query_options->__set_exec_time_limit_s(time_limit);
+        break;
+      }
       default:
         if (IsRemovedQueryOption(key)) {
           LOG(WARNING) << "Ignoring attempt to set removed query option '" << 
key << "'";

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/be/src/service/query-options.h
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index bb07552..2280cff 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -41,7 +41,7 @@ typedef std::unordered_map<string, 
beeswax::TQueryOptionLevel::type>
 // the DCHECK.
 #define QUERY_OPTS_TABLE\
   DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(),\
-      TImpalaQueryOptions::COMPUTE_STATS_MIN_SAMPLE_SIZE + 1);\
+      TImpalaQueryOptions::EXEC_TIME_LIMIT_S + 1);\
   REMOVED_QUERY_OPT_FN(abort_on_default_limit_exceeded, 
ABORT_ON_DEFAULT_LIMIT_EXCEEDED)\
   QUERY_OPT_FN(abort_on_error, ABORT_ON_ERROR, TQueryOptionLevel::REGULAR)\
   QUERY_OPT_FN(allow_unsupported_formats, ALLOW_UNSUPPORTED_FORMATS,\
@@ -128,6 +128,7 @@ typedef std::unordered_map<string, 
beeswax::TQueryOptionLevel::type>
   QUERY_OPT_FN(idle_session_timeout, IDLE_SESSION_TIMEOUT, 
TQueryOptionLevel::REGULAR)\
   QUERY_OPT_FN(compute_stats_min_sample_size, COMPUTE_STATS_MIN_SAMPLE_SIZE,\
       TQueryOptionLevel::ADVANCED)\
+  QUERY_OPT_FN(exec_time_limit_s, EXEC_TIME_LIMIT_S, 
TQueryOptionLevel::REGULAR)\
   ;
 
 /// Enforce practical limits on some query options to avoid undesired query 
state.

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/common/thrift/ImpalaInternalService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaInternalService.thrift 
b/common/thrift/ImpalaInternalService.thrift
index 613e4b0..dc37fc2 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -115,7 +115,7 @@ struct TQueryOptions {
   24: optional bool disable_outermost_topn = 0
 
   // Time, in s, before a query will be timed out if it is inactive. May not 
exceed
-  // --idle_query_timeout if that flag > 0.
+  // --idle_query_timeout if that flag > 0. If 0, falls back to 
--idle_query_timeout.
   26: optional i32 query_timeout_s = 0
 
   // test hook to cap max memory for spilling operators (to force them to 
spill).
@@ -267,6 +267,11 @@ struct TQueryOptions {
   // Minimum number of bytes that will be scanned in COMPUTE STATS TABLESAMPLE,
   // regardless of the user-supplied sampling percent. Default value: 1GB
   62: optional i64 compute_stats_min_sample_size = 1073741824;
+
+  // Time limit, in s, before a query will be timed out after it starts 
executing. Does
+  // not include time spent in planning, scheduling or admission control. A 
value of 0
+  // means no time limit.
+  63: optional i32 exec_time_limit_s = 0;
 }
 
 // Impala currently has two types of sessions: Beeswax and HiveServer2

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/common/thrift/ImpalaService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaService.thrift 
b/common/thrift/ImpalaService.thrift
index 1911fc6..356f5e5 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -143,7 +143,7 @@ enum TImpalaQueryOptions {
   RM_INITIAL_MEM, // Removed
 
   // Time, in s, before a query will be timed out if it is inactive. May not 
exceed
-  // --idle_query_timeout if that flag > 0.
+  // --idle_query_timeout if that flag > 0. If 0, falls back to 
--idle_query_timeout.
   QUERY_TIMEOUT_S,
 
   // Test hook for spill to disk operators
@@ -278,6 +278,11 @@ enum TImpalaQueryOptions {
   // Minimum number of bytes that will be scanned in COMPUTE STATS TABLESAMPLE,
   // regardless of the user-supplied sampling percent.
   COMPUTE_STATS_MIN_SAMPLE_SIZE,
+
+  // Time limit, in s, before a query will be timed out after it starts 
executing. Does
+  // not include time spent in planning, scheduling or admission control. A 
value of 0
+  // means no time limit.
+  EXEC_TIME_LIMIT_S,
 }
 
 // The summary of a DML statement.

http://git-wip-us.apache.org/repos/asf/impala/blob/ff3ddb51/tests/custom_cluster/test_query_expiration.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_query_expiration.py 
b/tests/custom_cluster/test_query_expiration.py
index df203b3..0e47ce2 100644
--- a/tests/custom_cluster/test_query_expiration.py
+++ b/tests/custom_cluster/test_query_expiration.py
@@ -18,6 +18,7 @@
 # Tests for query expiration.
 
 import pytest
+import re
 import threading
 from time import sleep, time
 
@@ -43,31 +44,69 @@ class TestQueryExpiration(CustomClusterTestSuite):
     impalad = self.cluster.get_first_impalad()
     client = impalad.service.create_beeswax_client()
     num_expired = 
impalad.service.get_metric_value('impala-server.num-queries-expired')
-    handle = client.execute_async("SELECT SLEEP(1000000)")
+    handles = []
+
+    # This query will time out with the default idle timeout (6s).
+    query1 = "SELECT SLEEP(1000000)"
+    default_timeout_expire_handle = client.execute_async(query1)
+    handles.append(default_timeout_expire_handle)
+
+    # This query will hit a lower time limit.
+    client.execute("SET EXEC_TIME_LIMIT_S=1")
+    time_limit_expire_handle = client.execute_async(query1);
+    handles.append(time_limit_expire_handle)
+
+    # This query will hit a lower idle timeout instead of the default timeout 
or time
+    # limit.
+    client.execute("SET EXEC_TIME_LIMIT_S=5")
     client.execute("SET QUERY_TIMEOUT_S=1")
-    handle2 = client.execute_async("SELECT SLEEP(2000000)")
+    short_timeout_expire_handle = client.execute_async("SELECT SLEEP(2000000)")
+    handles.append(short_timeout_expire_handle)
+    client.execute("SET EXEC_TIME_LIMIT_S=0")
 
     # Set a huge timeout, to check that the server bounds it by 
--idle_query_timeout
     client.execute("SET QUERY_TIMEOUT_S=1000")
-    handle3 = client.execute_async("SELECT SLEEP(3000000)")
-    self._check_num_executing(impalad, 3)
+    default_timeout_expire_handle2 = client.execute_async("SELECT 
SLEEP(3000000)")
+    handles.append(default_timeout_expire_handle2)
+    self._check_num_executing(impalad, len(handles))
 
     before = time()
     sleep(4)
 
-    # Query with timeout of 1 should have expired, other query should still be 
running.
-    assert num_expired + 1 == impalad.service.get_metric_value(
+    # Queries with timeout or time limit of 1 should have expired, other 
queries should
+    # still be running.
+    assert num_expired + 2 == impalad.service.get_metric_value(
       'impala-server.num-queries-expired')
+    assert (client.get_state(short_timeout_expire_handle) ==
+            client.QUERY_STATES['EXCEPTION'])
+    assert (client.get_state(time_limit_expire_handle) ==
+            client.QUERY_STATES['EXCEPTION'])
+    assert (client.get_state(default_timeout_expire_handle) ==
+            client.QUERY_STATES['FINISHED'])
+    assert (client.get_state(default_timeout_expire_handle2) ==
+            client.QUERY_STATES['FINISHED'])
+    self.__expect_expired(client, query1, short_timeout_expire_handle,
+        "Query [0-9a-f]+:[0-9a-f]+ expired due to client inactivity \(timeout 
is 1s000ms\)")
+    self.__expect_expired(client, query1, time_limit_expire_handle,
+        "Query [0-9a-f]+:[0-9a-f]+ expired due to execution time limit of 
1s000ms")
     self._check_num_executing(impalad, 2)
     self.assert_impalad_log_contains('INFO', "Expiring query due to client 
inactivity: "
         "[0-9a-f]+:[0-9a-f]+, last activity was at: \d\d\d\d-\d\d-\d\d 
\d\d:\d\d:\d\d")
+    self.assert_impalad_log_contains('INFO',
+        "Expiring query [0-9a-f]+:[0-9a-f]+ due to execution time limit of 1s")
+
+    # Wait until the remaining queries expire. The time limit query will have 
hit
+    # expirations but only one should be counted.
     impalad.service.wait_for_metric_value('impala-server.num-queries-expired',
-                                          num_expired + 3)
+                                          num_expired + len(handles))
+    assert (client.get_state(default_timeout_expire_handle) ==
+            client.QUERY_STATES['EXCEPTION'])
+    assert (client.get_state(default_timeout_expire_handle2) ==
+            client.QUERY_STATES['EXCEPTION'])
 
     # Check that we didn't wait too long to be expired (double the timeout is 
sufficiently
     # large to avoid most noise in measurement)
     assert time() - before < 12
-    assert client.get_state(handle) == client.QUERY_STATES['EXCEPTION']
 
     client.execute("SET QUERY_TIMEOUT_S=0")
     # Synchronous execution; calls fetch() and query should not time out.
@@ -77,37 +116,66 @@ class TestQueryExpiration(CustomClusterTestSuite):
 
     # Confirm that no extra expirations happened
     assert 
impalad.service.get_metric_value('impala-server.num-queries-expired') \
-        == num_expired + 3
+        == len(handles)
     self._check_num_executing(impalad, 0)
+    for handle in handles:
+      try:
+        client.close_query(handle)
+      except Exception, e:
+        # We fetched from some cancelled handles above, which unregistered the 
queries.
+        assert 'Invalid or unknown query handle' in str(e)
 
 
   @pytest.mark.execute_serially
   @CustomClusterTestSuite.with_args("--idle_query_timeout=0")
   def test_query_expiration_no_default(self, vector):
     """Confirm that single queries expire if no default is set, but a per-query
-    expiration is set"""
+    expiration or time limit is set"""
     impalad = self.cluster.get_any_impalad()
     client = impalad.service.create_beeswax_client()
     num_expired = 
impalad.service.get_metric_value('impala-server.num-queries-expired')
+    query = "SELECT SLEEP(1000000)"
     client.execute("SET QUERY_TIMEOUT_S=1")
-    handle = client.execute_async("SELECT SLEEP(1000000)")
+    timeout_handle = client.execute_async(query)
+    client.execute("SET QUERY_TIMEOUT_S=0")
+
+    client.execute("SET EXEC_TIME_LIMIT_S=1")
+    time_limit_handle = client.execute_async(query)
+    client.execute("SET EXEC_TIME_LIMIT_S=0")
 
     # Set a huge timeout, server should not expire the query while this test 
is running
     client.execute("SET QUERY_TIMEOUT_S=1000")
-    handle3 = client.execute_async("SELECT SLEEP(2000000)")
+    no_timeout_handle = client.execute_async(query)
 
     before = time()
     sleep(4)
 
     # Query with timeout of 1 should have expired, other query should still be 
running.
-    assert num_expired + 1 == impalad.service.get_metric_value(
+    assert num_expired + 2 == impalad.service.get_metric_value(
       'impala-server.num-queries-expired')
 
+    assert client.get_state(timeout_handle) == client.QUERY_STATES['EXCEPTION']
+    assert client.get_state(time_limit_handle) == 
client.QUERY_STATES['EXCEPTION']
+    assert client.get_state(no_timeout_handle) == 
client.QUERY_STATES['FINISHED']
+    self.__expect_expired(client, query, timeout_handle,
+        "Query [0-9a-f]+:[0-9a-f]+ expired due to client inactivity \(timeout 
is 1s000ms\)")
+    self.__expect_expired(client, query, time_limit_handle,
+        "Query [0-9a-f]+:[0-9a-f]+ expired due to execution time limit of 
1s000ms")
+
+  def __expect_expired(self, client, query, handle, exception_regex):
+    """Check that the query handle expired, with an error containing 
exception_regex"""
+    try:
+      client.fetch(query, handle)
+      assert False
+    except Exception, e:
+      assert re.search(exception_regex, str(e))
+
   @pytest.mark.execute_serially
   @CustomClusterTestSuite.with_args("--idle_query_timeout=1")
   def test_concurrent_query_expiration(self, vector):
     """Confirm that multiple concurrent queries are correctly expired if not 
fetched"""
     class ExpiringQueryThread(threading.Thread):
+      """Thread that runs a query and does not fetch so it will time out."""
       def __init__(self, client):
         super(ExpiringQueryThread, self).__init__()
         self.client = client
@@ -117,6 +185,7 @@ class TestQueryExpiration(CustomClusterTestSuite):
         self.handle = self.client.execute_async("SELECT SLEEP(3000000)")
 
     class NonExpiringQueryThread(threading.Thread):
+      """Thread that runs a query that does not hit the idle timeout."""
       def __init__(self, client):
         super(NonExpiringQueryThread, self).__init__()
         self.client = client
@@ -126,6 +195,37 @@ class TestQueryExpiration(CustomClusterTestSuite):
         result = self.client.execute("SELECT SLEEP(2500)")
         self.success = result.success
 
+    class TimeLimitThread(threading.Thread):
+      """Thread that runs a query that hits a time limit."""
+      def __init__(self, client):
+        super(TimeLimitThread, self).__init__()
+        self.client = client
+        self.success = False
+
+      def run(self):
+        # Query will not be idle but will hit time limit.
+        self.client.execute("SET EXEC_TIME_LIMIT_S=1")
+        try:
+          result = self.client.execute("SELECT SLEEP(2500)")
+          assert "Expected to hit time limit"
+        except Exception, e:
+          self.exception = e
+
+    class NonExpiringTimeLimitThread(threading.Thread):
+      """Thread that runs a query that finishes before time limit."""
+      def __init__(self, client):
+        super(NonExpiringTimeLimitThread, self).__init__()
+        self.client = client
+        self.success = False
+        self.data = None
+
+      def run(self):
+        # Query will complete before time limit.
+        self.client.execute("SET EXEC_TIME_LIMIT_S=10")
+        result = self.client.execute("SELECT count(*) FROM 
functional.alltypes")
+        self.success = result.success
+        self.data = result.data
+
     impalad = self.cluster.get_any_impalad()
     client = impalad.service.create_beeswax_client()
     num_expired = 
impalad.service.get_metric_value('impala-server.num-queries-expired')
@@ -134,18 +234,27 @@ class TestQueryExpiration(CustomClusterTestSuite):
          for _ in xrange(5)]
     expiring_threads = 
[ExpiringQueryThread(impalad.service.create_beeswax_client())
                         for _ in xrange(5)]
-    all_threads = zip(non_expiring_threads, expiring_threads)
-    for n, e in all_threads:
-      n.start()
-      e.start()
-
-    for n, e in all_threads:
-      n.join()
-      e.join()
-
+    time_limit_threads = 
[TimeLimitThread(impalad.service.create_beeswax_client())
+                        for _ in xrange(5)]
+    non_expiring_time_limit_threads = [
+        NonExpiringTimeLimitThread(impalad.service.create_beeswax_client())
+        for _ in xrange(5)]
+    all_threads = non_expiring_threads + expiring_threads + time_limit_threads 
+\
+        non_expiring_time_limit_threads
+    for t in all_threads:
+      t.start()
+    for t in all_threads:
+      t.join()
     impalad.service.wait_for_metric_value('impala-server.num-queries-expired',
-                                          num_expired + 5)
-
-    for n, e in all_threads:
-      assert n.success
-      assert client.get_state(e.handle) == client.QUERY_STATES['EXCEPTION']
+                                          num_expired + 10)
+    for t in non_expiring_threads:
+      assert t.success
+    for t in expiring_threads:
+      assert client.get_state(t.handle) == client.QUERY_STATES['EXCEPTION']
+    for t in time_limit_threads:
+      assert re.search(
+          "Query [0-9a-f]+:[0-9a-f]+ expired due to execution time limit of 
1s000ms",
+          str(t.exception))
+    for t in non_expiring_time_limit_threads:
+      assert t.success
+      assert t.data[0] == '7300' # Number of rows in alltypes.

Reply via email to