impala-server.h

Go to the documentation of this file.

// Copyright 2012 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_IMPALA_SERVER_H
#define IMPALA_SERVICE_IMPALA_SERVER_H

#include <boost/thread/mutex.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/unordered_map.hpp>
#include <boost/unordered_set.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>

#include "gen-cpp/ImpalaService.h"
#include "gen-cpp/ImpalaHiveServer2Service.h"
#include "gen-cpp/ImpalaInternalService.h"
#include "gen-cpp/Frontend_types.h"
#include "rpc/thrift-server.h"
#include "common/status.h"
#include "service/frontend.h"
#include "util/metrics.h"
#include "util/runtime-profile.h"
#include "util/simple-logger.h"
#include "util/thread-pool.h"
#include "util/time.h"
#include "util/uid-util.h"
#include "runtime/coordinator.h"
#include "runtime/runtime-state.h"
#include "runtime/timestamp-value.h"
#include "runtime/types.h"
#include "rapidjson/rapidjson.h"

namespace impala {

class ExecEnv;
class DataSink;
class CancellationWork;
class Coordinator;
class RowDescriptor;
class TCatalogUpdate;
class TPlanExecRequest;
class TPlanExecParams;
class TInsertResult;
class TReportExecStatusArgs;
class TReportExecStatusResult;
class TTransmitDataArgs;
class TTransmitDataResult;
class TNetworkAddress;
class TClientRequest;
class TExecRequest;
class TSessionState;
class TQueryOptions;
class TGetExecSummaryResp;
class TGetExecSummaryReq;

//
class ImpalaServer : public ImpalaServiceIf, public ImpalaHiveServer2ServiceIf,
                     public ThriftServer::ConnectionHandlerIf {
 public:
  ImpalaServer(ExecEnv* exec_env);
  ~ImpalaServer();

  virtual void query(beeswax::QueryHandle& query_handle, const beeswax::Query& query);
  virtual void executeAndWait(beeswax::QueryHandle& query_handle,
      const beeswax::Query& query, const beeswax::LogContextId& client_ctx);
  virtual void explain(beeswax::QueryExplanation& query_explanation,
      const beeswax::Query& query);
  virtual void fetch(beeswax::Results& query_results,
      const beeswax::QueryHandle& query_handle, const bool start_over,
      const int32_t fetch_size);
  virtual void get_results_metadata(beeswax::ResultsMetadata& results_metadata,
      const beeswax::QueryHandle& handle);
  virtual void close(const beeswax::QueryHandle& handle);
  virtual beeswax::QueryState::type get_state(const beeswax::QueryHandle& handle);
  virtual void echo(std::string& echo_string, const std::string& input_string);
  virtual void clean(const beeswax::LogContextId& log_context);
  virtual void get_log(std::string& log, const beeswax::LogContextId& context);

  virtual void get_default_configuration(
      std::vector<beeswax::ConfigVariable>& configurations, const bool include_hadoop);

  virtual void dump_config(std::string& config);

  virtual void Cancel(impala::TStatus& status, const beeswax::QueryHandle& query_id);
  virtual void CloseInsert(impala::TInsertResult& insert_result,
      const beeswax::QueryHandle& query_handle);

  virtual void PingImpalaService(TPingImpalaServiceResp& return_val);

  virtual void GetRuntimeProfile(std::string& profile_output,
      const beeswax::QueryHandle& query_id);

  virtual void GetExecSummary(impala::TExecSummary& result,
      const beeswax::QueryHandle& query_id);

  virtual void ResetCatalog(impala::TStatus& status);

  virtual void ResetTable(impala::TStatus& status, const TResetTableReq& request);

  virtual void OpenSession(
      apache::hive::service::cli::thrift::TOpenSessionResp& return_val,
      const apache::hive::service::cli::thrift::TOpenSessionReq& request);
  virtual void CloseSession(
      apache::hive::service::cli::thrift::TCloseSessionResp& return_val,
      const apache::hive::service::cli::thrift::TCloseSessionReq& request);
  virtual void GetInfo(
      apache::hive::service::cli::thrift::TGetInfoResp& return_val,
      const apache::hive::service::cli::thrift::TGetInfoReq& request);
  virtual void ExecuteStatement(
      apache::hive::service::cli::thrift::TExecuteStatementResp& return_val,
      const apache::hive::service::cli::thrift::TExecuteStatementReq& request);
  virtual void GetTypeInfo(
      apache::hive::service::cli::thrift::TGetTypeInfoResp& return_val,
      const apache::hive::service::cli::thrift::TGetTypeInfoReq& request);
  virtual void GetCatalogs(
      apache::hive::service::cli::thrift::TGetCatalogsResp& return_val,
      const apache::hive::service::cli::thrift::TGetCatalogsReq& request);
  virtual void GetSchemas(
      apache::hive::service::cli::thrift::TGetSchemasResp& return_val,
      const apache::hive::service::cli::thrift::TGetSchemasReq& request);
  virtual void GetTables(
      apache::hive::service::cli::thrift::TGetTablesResp& return_val,
      const apache::hive::service::cli::thrift::TGetTablesReq& request);
  virtual void GetTableTypes(
      apache::hive::service::cli::thrift::TGetTableTypesResp& return_val,
      const apache::hive::service::cli::thrift::TGetTableTypesReq& request);
  virtual void GetColumns(
      apache::hive::service::cli::thrift::TGetColumnsResp& return_val,
      const apache::hive::service::cli::thrift::TGetColumnsReq& request);
  virtual void GetFunctions(
      apache::hive::service::cli::thrift::TGetFunctionsResp& return_val,
      const apache::hive::service::cli::thrift::TGetFunctionsReq& request);
  virtual void GetOperationStatus(
      apache::hive::service::cli::thrift::TGetOperationStatusResp& return_val,
      const apache::hive::service::cli::thrift::TGetOperationStatusReq& request);
  virtual void CancelOperation(
      apache::hive::service::cli::thrift::TCancelOperationResp& return_val,
      const apache::hive::service::cli::thrift::TCancelOperationReq& request);
  virtual void CloseOperation(
      apache::hive::service::cli::thrift::TCloseOperationResp& return_val,
      const apache::hive::service::cli::thrift::TCloseOperationReq& request);
  virtual void GetResultSetMetadata(
      apache::hive::service::cli::thrift::TGetResultSetMetadataResp& return_val,
      const apache::hive::service::cli::thrift::TGetResultSetMetadataReq& request);
  virtual void FetchResults(
      apache::hive::service::cli::thrift::TFetchResultsResp& return_val,
      const apache::hive::service::cli::thrift::TFetchResultsReq& request);
  virtual void GetLog(apache::hive::service::cli::thrift::TGetLogResp& return_val,
      const apache::hive::service::cli::thrift::TGetLogReq& request);
  virtual void GetExecSummary(TGetExecSummaryResp& return_val,
      const TGetExecSummaryReq& request);
  virtual void GetRuntimeProfile(TGetRuntimeProfileResp& return_val,
      const TGetRuntimeProfileReq& request);
  virtual void GetDelegationToken(
      apache::hive::service::cli::thrift::TGetDelegationTokenResp& return_val,
      const apache::hive::service::cli::thrift::TGetDelegationTokenReq& req);
  virtual void CancelDelegationToken(
      apache::hive::service::cli::thrift::TCancelDelegationTokenResp& return_val,
      const apache::hive::service::cli::thrift::TCancelDelegationTokenReq& req);
  virtual void RenewDelegationToken(
      apache::hive::service::cli::thrift::TRenewDelegationTokenResp& return_val,
      const apache::hive::service::cli::thrift::TRenewDelegationTokenReq& req);

  void ReportExecStatus(TReportExecStatusResult& return_val,
      const TReportExecStatusParams& params);
  void TransmitData(TTransmitDataResult& return_val,
      const TTransmitDataParams& params);

  static void PrepareQueryContext(TQueryCtx* query_ctx);


  virtual void ConnectionStart(const ThriftServer::ConnectionContext& session_context);

  virtual void ConnectionEnd(const ThriftServer::ConnectionContext& session_context);

  void MembershipCallback(const StatestoreSubscriber::TopicDeltaMap&
      incoming_topic_deltas, std::vector<TTopicDelta>* subscriber_topic_updates);

  void CatalogUpdateCallback(const StatestoreSubscriber::TopicDeltaMap& topic_deltas,
      std::vector<TTopicDelta>* topic_updates);

  bool IsOffline() {
    boost::lock_guard<boost::mutex> l(is_offline_lock_);
    return is_offline_;
  }

  bool IsLineageLoggingEnabled();

 private:
  friend class ChildQuery;

  class QueryResultSet {
   public:
    QueryResultSet() {}
    virtual ~QueryResultSet() {}

    virtual Status AddOneRow(
        const std::vector<void*>& row, const std::vector<int>& scales) = 0;

    virtual Status AddOneRow(const TResultRow& row) = 0;

    virtual int AddRows(const QueryResultSet* other, int start_idx, int num_rows) = 0;

    int64_t ByteSize() { return ByteSize(0, size()); }

    virtual int64_t ByteSize(int start_idx, int num_rows) = 0;

    virtual size_t size() = 0;
  };

  class AsciiQueryResultSet;
  class HS2RowOrientedResultSet;
  class HS2ColumnarResultSet;

  struct SessionState;

  class QueryExecState;

  static const char* SQLSTATE_SYNTAX_ERROR_OR_ACCESS_VIOLATION;
  static const char* SQLSTATE_GENERAL_ERROR;
  static const char* SQLSTATE_OPTIONAL_FEATURE_NOT_IMPLEMENTED;

  static const int ASCII_PRECISION;

  QueryResultSet* CreateHS2ResultSet(
      apache::hive::service::cli::thrift::TProtocolVersion::type version,
      const TResultSetMetadata& metadata,
      apache::hive::service::cli::thrift::TRowSet* rowset = NULL);

  boost::shared_ptr<QueryExecState> GetQueryExecState(
      const TUniqueId& query_id, bool lock);

  bool GetSessionIdForQuery(const TUniqueId& query_id, TUniqueId* session_id);

  Status UpdateCatalogMetrics();

  Status Execute(TQueryCtx* query_ctx,
                 boost::shared_ptr<SessionState> session_state,
                 boost::shared_ptr<QueryExecState>* exec_state);

  Status ExecuteInternal(const TQueryCtx& query_ctx,
                         boost::shared_ptr<SessionState> session_state,
                         bool* registered_exec_state,
                         boost::shared_ptr<QueryExecState>* exec_state);

  Status RegisterQuery(boost::shared_ptr<SessionState> session_state,
      const boost::shared_ptr<QueryExecState>& exec_state);

  Status SetQueryInflight(boost::shared_ptr<SessionState> session_state,
      const boost::shared_ptr<QueryExecState>& exec_state);

  Status UnregisterQuery(const TUniqueId& query_id, bool check_inflight,
      const Status* cause = NULL);

  Status CancelInternal(const TUniqueId& query_id, bool check_inflight,
      const Status* cause = NULL);

  Status CloseSessionInternal(const TUniqueId& session_id, bool ignore_if_absent);

  Status GetRuntimeProfileStr(const TUniqueId& query_id, bool base64_encoded,
      std::stringstream* output);

  Status GetExecSummary(const TUniqueId& query_id, TExecSummary* result);

  void HadoopVarzUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  //
  void QueryStateUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  void QueryProfileUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  void QuerySummaryCallback(bool include_plan_json, bool include_summary,
      const Webserver::ArgumentMap& args, rapidjson::Document* document);

  void CancelQueryUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  void QueryProfileEncodedUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  void InflightQueryIdsUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  void SessionsUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  void CatalogUrlCallback(const Webserver::ArgumentMap& args, rapidjson::Document* output);

  void CatalogObjectsUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* output);

  void InitializeConfigVariables();

  void RegisterWebserverCallbacks(Webserver* webserver);

  Status InitProfileLogging();

  Status InitAuditEventLogging();

  Status InitLineageLogging();

  static Status InitLoggingDir(const std::string& log_dir);

  bool IsAuditEventLoggingEnabled();

  Status LogAuditRecord(const QueryExecState& exec_state, const TExecRequest& request);

  Status LogLineageRecord(const TExecRequest& request);

  void LogQueryEvents(const QueryExecState& exec_state);

  void LogFileFlushThread();

  void AuditEventLoggerFlushThread();

  void LineageLoggerFlushThread();

  void ArchiveQuery(const QueryExecState& query);

  Status AuthorizeProxyUser(const std::string& user, const std::string& do_as_user);

  struct QueryStateRecord {
    std::string profile_str;

    std::string encoded_profile_str;

    TUniqueId id;

    std::string effective_user;

    std::string default_db;

    std::string stmt;

    std::string plan;

    TStmtType::type stmt_type;

    bool has_coord;

    int64_t num_complete_fragments;

    int64_t total_fragments;

    int64_t num_rows_fetched;

    beeswax::QueryState::type query_state;

    TimestampValue start_time, end_time;

    TExecSummary exec_summary;

    Status query_status;

    TEventSequence event_sequence;

    vector<TPlanFragment> fragments;

    QueryStateRecord(const QueryExecState& exec_state, bool copy_profile = false,
        const std::string& encoded_str = "");

    QueryStateRecord() { }

    bool operator() (const QueryStateRecord& lhs, const QueryStateRecord& rhs) const;
  };

  void QueryStateToJson(const ImpalaServer::QueryStateRecord& record,
      rapidjson::Value* value, rapidjson::Document* document);


  Status QueryToTQueryContext(const beeswax::Query& query, TQueryCtx* query_ctx);
  void TUniqueIdToQueryHandle(const TUniqueId& query_id, beeswax::QueryHandle* handle);
  void QueryHandleToTUniqueId(const beeswax::QueryHandle& handle, TUniqueId* query_id);

  void RaiseBeeswaxException(const std::string& msg, const char* sql_state);

  Status FetchInternal(const TUniqueId& query_id, bool start_over,
      int32_t fetch_size, beeswax::Results* query_results);

  Status CloseInsertInternal(const TUniqueId& query_id, TInsertResult* insert_result);


  void ExecuteMetadataOp(
      const apache::hive::service::cli::thrift::THandleIdentifier& session_handle,
      TMetadataOpRequest* request,
      apache::hive::service::cli::thrift::TOperationHandle* handle,
      apache::hive::service::cli::thrift::TStatus* status);

  Status FetchInternal(const TUniqueId& query_id, int32_t fetch_size, bool fetch_first,
      apache::hive::service::cli::thrift::TFetchResultsResp* fetch_results);


  static Status THandleIdentifierToTUniqueId(
      const apache::hive::service::cli::thrift::THandleIdentifier& handle,
      TUniqueId* unique_id, TUniqueId* secret);
  static void TUniqueIdToTHandleIdentifier(
      const TUniqueId& unique_id, const TUniqueId& secret,
      apache::hive::service::cli::thrift::THandleIdentifier* handle);
  Status TExecuteStatementReqToTQueryContext(
      const apache::hive::service::cli::thrift::TExecuteStatementReq execute_request,
      TQueryCtx* query_ctx);

  void CancelFromThreadPool(uint32_t thread_id,
      const CancellationWork& cancellation_work);

  Status ProcessCatalogUpdateResult(const TCatalogUpdateResult& catalog_update_result,
      bool wait_for_all_subscribers);

  void ExpireSessions();

  void ExpireQueries();

  void DetectNmFailures();

  void SetOffline(bool offline);

  boost::mutex query_log_lock_;

  typedef std::list<QueryStateRecord> QueryLog;
  QueryLog query_log_;

  typedef boost::unordered_map<TUniqueId, QueryLog::iterator> QueryLogIndex;
  QueryLogIndex query_log_index_;

  boost::scoped_ptr<SimpleLogger> profile_logger_;

  boost::scoped_ptr<SimpleLogger> audit_event_logger_;

  boost::scoped_ptr<SimpleLogger> lineage_logger_;

  boost::scoped_ptr<Thread> profile_log_file_flush_thread_;

  boost::scoped_ptr<Thread> audit_event_logger_flush_thread_;

  boost::scoped_ptr<Thread> lineage_logger_flush_thread_;

  ExecEnv* exec_env_;  // not owned

  boost::scoped_ptr<ThreadPool<CancellationWork> > cancellation_thread_pool_;

  boost::scoped_ptr<Thread> session_timeout_thread_;

  typedef boost::unordered_map<TUniqueId, boost::shared_ptr<QueryExecState> >
      QueryExecStateMap;
  QueryExecStateMap query_exec_state_map_;
  boost::mutex query_exec_state_map_lock_;  // protects query_exec_state_map_

  TQueryOptions default_query_options_;
  std::vector<beeswax::ConfigVariable> default_configs_;

  struct SessionState {
    SessionState() : closed(false), expired(false), hs2_version(
        apache::hive::service::cli::thrift::
        TProtocolVersion::HIVE_CLI_SERVICE_PROTOCOL_V1), ref_count(0) { }

    TSessionType::type session_type;

    TimestampValue start_time;

    std::string connected_user;

    std::string do_as_user;

    TNetworkAddress network_address;

    boost::mutex lock;

    bool closed;

    bool expired;

    std::string database;

    TQueryOptions default_query_options;

    apache::hive::service::cli::thrift::TProtocolVersion::type hs2_version;

    boost::unordered_set<TUniqueId> inflight_queries;

    int64_t last_accessed_ms;

    uint32_t ref_count;

    void ToThrift(const TUniqueId& session_id, TSessionState* session_state);
  };

  class ScopedSessionState {
   public:
    ScopedSessionState(ImpalaServer* impala) : impala_(impala) { }

    Status WithSession(const TUniqueId& session_id,
        boost::shared_ptr<SessionState>* session = NULL) {
      DCHECK(session_.get() == NULL);
      RETURN_IF_ERROR(impala_->GetSessionState(session_id, &session_, true));
      if (session != NULL) (*session) = session_;
      return Status::OK;
    }

    ~ScopedSessionState() {
      if (session_.get() != NULL) {
        impala_->MarkSessionInactive(session_);
      }
    }

   private:
    boost::shared_ptr<SessionState> session_;

    ImpalaServer* impala_;
  };

  friend class ScopedSessionState;

  boost::mutex session_state_map_lock_;

  typedef boost::unordered_map<TUniqueId, boost::shared_ptr<SessionState> >
    SessionStateMap;
  SessionStateMap session_state_map_;

  boost::mutex connection_to_sessions_map_lock_;

  typedef boost::unordered_map<TUniqueId, std::vector<TUniqueId> >
    ConnectionToSessionMap;
  ConnectionToSessionMap connection_to_sessions_map_;

  Status GetSessionState(const TUniqueId& session_id,
      boost::shared_ptr<SessionState>* session_state, bool mark_active = false);

  inline void MarkSessionInactive(boost::shared_ptr<SessionState> session) {
    boost::lock_guard<boost::mutex> l(session->lock);
    DCHECK_GT(session->ref_count, 0);
    --session->ref_count;
    session->last_accessed_ms = UnixMillis();
  }

  boost::mutex query_locations_lock_;

  typedef boost::unordered_map<TNetworkAddress, boost::unordered_set<TUniqueId> >
      QueryLocations;
  QueryLocations query_locations_;

  typedef boost::unordered_map<std::string, TNetworkAddress> BackendAddressMap;
  BackendAddressMap known_backends_;

  boost::uuids::random_generator uuid_generator_;

  boost::mutex uuid_lock_;

  boost::mutex catalog_version_lock_;

  boost::condition_variable catalog_version_update_cv_;

  struct CatalogUpdateVersionInfo {
    CatalogUpdateVersionInfo() :
      catalog_version(0L),
      catalog_topic_version(0L) {
    }

    int64_t catalog_version;
    TUniqueId catalog_service_id;
    int64_t catalog_topic_version;
  };

  CatalogUpdateVersionInfo catalog_update_info_;

  int64_t min_subscriber_catalog_topic_version_;

  typedef boost::unordered_map<std::string, boost::unordered_set<std::string> >
      ProxyUserMap;
  ProxyUserMap authorized_proxy_user_config_;

  boost::mutex query_expiration_lock_;

  typedef std::pair<int64_t, TUniqueId> ExpirationEvent;

  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;
    }
  };

  //
  //
  //
  typedef std::set<ExpirationEvent, ExpirationEventComparator> ExpirationQueue;
  ExpirationQueue queries_by_timestamp_;

  boost::scoped_ptr<Thread> query_expiration_thread_;

  boost::scoped_ptr<Thread> nm_failure_detection_thread_;

  boost::mutex is_offline_lock_;

  bool is_offline_;
};

Status CreateImpalaServer(ExecEnv* exec_env, int beeswax_port, int hs2_port,
    int be_port, ThriftServer** beeswax_server, ThriftServer** hs2_server,
    ThriftServer** be_server, ImpalaServer** impala_server);

}

#endif