query-exec-state.h

Go to the documentation of this file.

// Copyright 2013 Cloudera Inc.
//
// Licensed 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.

#ifndef IMPALA_SERVICE_QUERY_EXEC_STATE_H
#define IMPALA_SERVICE_QUERY_EXEC_STATE_H

#include "common/status.h"
#include "exec/catalog-op-executor.h"
#include "util/runtime-profile.h"
#include "runtime/timestamp-value.h"
#include "service/child-query.h"
#include "statestore/query-schedule.h"
#include "gen-cpp/Frontend_types.h"
#include "service/impala-server.h"
#include "gen-cpp/Frontend_types.h"

#include <boost/thread.hpp>
#include <boost/unordered_set.hpp>
#include <vector>

namespace impala {

class ExecEnv;
class Coordinator;
class RuntimeState;
class RowBatch;
class Expr;
class TupleRow;
class Frontend;
class QueryExecStateCleaner;

class ImpalaServer::QueryExecState {
 public:
  QueryExecState(const TQueryCtx& query_ctx, ExecEnv* exec_env, Frontend* frontend,
      ImpalaServer* server, boost::shared_ptr<ImpalaServer::SessionState> session);

  ~QueryExecState();

  Status Exec(TExecRequest* exec_request);

  Status Exec(const TMetadataOpRequest& exec_request);

  void Wait();

  void WaitAsync();

  void BlockOnWait();

  Status FetchRows(const int32_t max_rows, QueryResultSet* fetched_rows);

  Status RestartFetch();

  void UpdateQueryState(beeswax::QueryState::type query_state);

  Status UpdateQueryStatus(const Status& status);

  void Cancel(const Status* cause = NULL);

  void Done();

  Status SetResultCache(QueryResultSet* cache, int64_t max_size);

  ImpalaServer::SessionState* session() const { return session_.get(); }
  const std::string& effective_user() const {
    return do_as_user().empty() ? connected_user() : do_as_user();
  }
  const std::string& connected_user() const { return query_ctx_.session.connected_user; }
  const std::string& do_as_user() const { return session_->do_as_user; }
  TSessionType::type session_type() const { return query_ctx_.session.session_type; }
  const TUniqueId& session_id() const { return query_ctx_.session.session_id; }
  const std::string& default_db() const { return query_ctx_.session.database; }
  bool eos() const { return eos_; }
  Coordinator* coord() const { return coord_.get(); }
  QuerySchedule* schedule() { return schedule_.get(); }
  int num_rows_fetched() const { return num_rows_fetched_; }
  void set_fetched_rows() { fetched_rows_ = true; }
  bool fetched_rows() const { return fetched_rows_; }
  bool returns_result_set() { return !result_metadata_.columns.empty(); }
  const TResultSetMetadata* result_metadata() { return &result_metadata_; }
  const TUniqueId& query_id() const { return query_ctx_.query_id; }
  const TExecRequest& exec_request() const { return exec_request_; }
  TStmtType::type stmt_type() const { return exec_request_.stmt_type; }
  TCatalogOpType::type catalog_op_type() const {
    return exec_request_.catalog_op_request.op_type;
  }
  TDdlType::type ddl_type() const {
    return exec_request_.catalog_op_request.ddl_params.ddl_type;
  }
  boost::mutex* lock() { return &lock_; }
  boost::mutex* fetch_rows_lock() { return &fetch_rows_lock_; }
  const beeswax::QueryState::type query_state() const { return query_state_; }
  void set_query_state(beeswax::QueryState::type state) { query_state_ = state; }
  const Status& query_status() const { return query_status_; }
  void set_result_metadata(const TResultSetMetadata& md) { result_metadata_ = md; }
  const RuntimeProfile& profile() const { return profile_; }
  const RuntimeProfile& summary_profile() const { return summary_profile_; }
  const TimestampValue& start_time() const { return start_time_; }
  const TimestampValue& end_time() const { return end_time_; }
  const std::string& sql_stmt() const { return query_ctx_.request.stmt; }
  const TQueryOptions& query_options() const { return query_ctx_.request.query_options; }
  TUniqueId parent_query_id() const { return query_ctx_.parent_query_id; }

  const std::vector<std::string>& GetAnalysisWarnings() const {
    return exec_request_.analysis_warnings;
  }

  inline int64_t last_active() const {
    boost::lock_guard<boost::mutex> l(expiration_data_lock_);
    return last_active_time_;
  }

  inline bool is_active() const {
    boost::lock_guard<boost::mutex> l(expiration_data_lock_);
    return ref_count_ > 0;
  }

  RuntimeProfile::EventSequence* query_events() const { return query_events_; }
  RuntimeProfile* summary_profile() { return &summary_profile_; }

 private:
  const TQueryCtx query_ctx_;

  boost::mutex fetch_rows_lock_;

  mutable boost::mutex expiration_data_lock_;
  int64_t last_active_time_;

  uint32_t ref_count_;

  boost::scoped_ptr<Thread> wait_thread_;

  boost::mutex lock_;  // protects all following fields
  ExecEnv* exec_env_;

  boost::shared_ptr<SessionState> session_;

  boost::scoped_ptr<QuerySchedule> schedule_;

  boost::scoped_ptr<Coordinator> coord_;

  boost::scoped_ptr<CatalogOpExecutor> catalog_op_executor_;

  boost::scoped_ptr<std::vector<TResultRow> > request_result_set_;

  boost::scoped_ptr<QueryResultSet> result_cache_;

  int64_t result_cache_max_size_;

  boost::scoped_ptr<RuntimeState> local_runtime_state_;
  ObjectPool profile_pool_;

  //
  RuntimeProfile profile_;
  RuntimeProfile server_profile_;
  RuntimeProfile summary_profile_;
  RuntimeProfile::Counter* row_materialization_timer_;

  RuntimeProfile::Counter* client_wait_timer_;
  MonotonicStopWatch client_wait_sw_;

  RuntimeProfile::EventSequence* query_events_;
  std::vector<ExprContext*> output_expr_ctxs_;
  bool eos_;  // if true, there are no more rows to return
  beeswax::QueryState::type query_state_;
  Status query_status_;
  TExecRequest exec_request_;

  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

  bool fetched_rows_;

  Frontend* frontend_;

  ImpalaServer* parent_server_;

  TimestampValue start_time_, end_time_;

  std::vector<ChildQuery> child_queries_;

  Status child_queries_status_;
  boost::scoped_ptr<Thread> child_queries_thread_;

  Status ExecLocalCatalogOp(const TCatalogOpRequest& catalog_op);

  void MarkInactive();

  void MarkActive();

  Status ExecQueryOrDmlRequest(const TQueryExecRequest& query_exec_request);

  Status ExecDdlRequest();

  Status ExecLoadDataRequest();

  Status WaitInternal();

  Status FetchRowsInternal(const int32_t max_rows, QueryResultSet* fetched_rows);

  Status FetchNextBatch();

  Status GetRowValue(TupleRow* row, std::vector<void*>* result, std::vector<int>* scales);

  Status UpdateCatalog();

  void SetResultSet(const std::vector<std::string>& results);
  void SetResultSet(const std::vector<std::string>& col1,
      const std::vector<std::string>& col2);
  void SetResultSet(const std::vector<std::string>& col1,
      const std::vector<std::string>& col2, const std::vector<std::string>& col3,
      const std::vector<std::string>& col4);

  void SetCreateTableAsSelectResultSet();

  Status UpdateTableAndColumnStats();

  void ExecChildQueriesAsync();

  void ExecChildQueries();

  Status WaitForChildQueries();

  void ClearResultCache();
};

}
#endif