runtime-state.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_RUNTIME_RUNTIME_STATE_H
#define IMPALA_RUNTIME_RUNTIME_STATE_H

#include "common/object-pool.h"

#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include <vector>
#include <string>
#include <sstream>

#include "statestore/query-resource-mgr.h"
#include "runtime/exec-env.h"
#include "runtime/descriptors.h"  // for PlanNodeId
#include "runtime/disk-io-mgr.h"  // for DiskIoMgr::RequestContext
#include "runtime/mem-tracker.h"
#include "runtime/thread-resource-mgr.h"
#include "gen-cpp/PlanNodes_types.h"
#include "gen-cpp/Types_types.h"  // for TUniqueId
#include "gen-cpp/ImpalaInternalService_types.h"  // for TQueryOptions
#include "util/runtime-profile.h"

namespace impala {

class Bitmap;
class BufferedBlockMgr;
class DescriptorTbl;
class ObjectPool;
class Status;
class ExecEnv;
class Expr;
class LlvmCodeGen;
class TimestampValue;
class DataStreamRecvr;

typedef std::map<std::string, TInsertPartitionStatus> PartitionStatusMap;

typedef std::map<std::string, TInsertStats> PartitionInsertStats;

typedef std::map<std::string, std::string> FileMoveMap;

class RuntimeState {
 public:
  RuntimeState(const TPlanFragmentInstanceCtx& fragment_instance_ctx,
      const std::string& cgroup, ExecEnv* exec_env);

  RuntimeState(const TQueryCtx& query_ctx);

  ~RuntimeState();

  void InitMemTrackers(const TUniqueId& query_id, const std::string* request_pool,
      int64_t query_bytes_limit, int64_t query_rm_reservation_limit_bytes = -1);

  Status CreateBlockMgr();

  ObjectPool* obj_pool() const { return obj_pool_.get(); }
  const DescriptorTbl& desc_tbl() const { return *desc_tbl_; }
  void set_desc_tbl(DescriptorTbl* desc_tbl) { desc_tbl_ = desc_tbl; }
  const TQueryOptions& query_options() const {
    return query_ctx().request.query_options;
  }
  int batch_size() const { return query_ctx().request.query_options.batch_size; }
  bool abort_on_error() const {
    return query_ctx().request.query_options.abort_on_error;
  }
  bool abort_on_default_limit_exceeded() const {
    return query_ctx().request.query_options.abort_on_default_limit_exceeded;
  }
  int max_errors() const { return query_options().max_errors; }
  const TQueryCtx& query_ctx() const { return fragment_instance_ctx_.query_ctx; }
  const TPlanFragmentInstanceCtx& fragment_ctx() const { return fragment_instance_ctx_; }
  const std::string& effective_user() const {
    if (query_ctx().session.__isset.delegated_user &&
        !query_ctx().session.delegated_user.empty()) {
      return do_as_user();
    }
    return connected_user();
  }
  const std::string& do_as_user() const { return query_ctx().session.delegated_user; }
  const std::string& connected_user() const {
    return query_ctx().session.connected_user;
  }
  const TimestampValue* now() const { return now_.get(); }
  void set_now(const TimestampValue* now);
  const ErrorLogMap& error_log() const { return error_log_; }
  const std::vector<std::pair<std::string, int> >& file_errors() const {
    return file_errors_;
  }
  const TUniqueId& query_id() const { return query_ctx().query_id; }
  const TUniqueId& fragment_instance_id() const {
    return fragment_instance_ctx_.fragment_instance_id;
  }
  const std::string& cgroup() const { return cgroup_; }
  ExecEnv* exec_env() { return exec_env_; }
  DataStreamMgr* stream_mgr() { return exec_env_->stream_mgr(); }
  HBaseTableFactory* htable_factory() { return exec_env_->htable_factory(); }
  ImpalaInternalServiceClientCache* impalad_client_cache() {
    return exec_env_->impalad_client_cache();
  }
  CatalogServiceClientCache* catalogd_client_cache() {
    return exec_env_->catalogd_client_cache();
  }
  DiskIoMgr* io_mgr() { return exec_env_->disk_io_mgr(); }
  MemTracker* instance_mem_tracker() { return instance_mem_tracker_.get(); }
  MemTracker* query_mem_tracker() { return query_mem_tracker_.get(); }
  ThreadResourceMgr::ResourcePool* resource_pool() { return resource_pool_; }

  FileMoveMap* hdfs_files_to_move() { return &hdfs_files_to_move_; }
  std::vector<DiskIoMgr::RequestContext*>* reader_contexts() { return &reader_contexts_; }

  void set_fragment_root_id(PlanNodeId id) {
    DCHECK_EQ(root_node_id_, -1) << "Should not set this twice.";
    root_node_id_ = id;
  }

  uint32_t fragment_hash_seed() const { return root_node_id_ + 1; }

  uint32_t slot_filter_bitmap_size() const { return 32213; }

  void AddBitmapFilter(SlotId slot, Bitmap* bitmap, bool* acquired_ownership);

  const Bitmap* GetBitmapFilter(SlotId slot) {
    if (slot_bitmap_filters_.find(slot) == slot_bitmap_filters_.end()) return NULL;
    return slot_bitmap_filters_[slot];
  }

  PartitionStatusMap* per_partition_status() { return &per_partition_status_; }

  RuntimeProfile* runtime_profile() { return &profile_; }

  bool codegen_enabled() const { return !query_options().disable_codegen; }

  bool codegen_created() const { return codegen_.get() != NULL; }

  Status GetCodegen(LlvmCodeGen** codegen, bool initialize = true);

  BufferedBlockMgr* block_mgr() {
    DCHECK(block_mgr_.get() != NULL);
    return block_mgr_.get();
  }

  Status query_status() {
    boost::lock_guard<SpinLock> l(query_status_lock_);
    return query_status_;
  };

  bool LogError(const ErrorMsg& msg);

  bool LogHasSpace() {
    boost::lock_guard<SpinLock> l(error_log_lock_);
    return error_log_.size() < query_options().max_errors;
  }

  void ReportFileErrors(const std::string& file_name, int num_errors);

  void ClearFileErrors() { file_errors_.clear(); }

  bool ErrorLogIsEmpty();

  std::string ErrorLog();

  void GetUnreportedErrors(ErrorLogMap* new_errors);

  std::string FileErrors();

  bool is_cancelled() const { return is_cancelled_; }
  void set_is_cancelled(bool v) { is_cancelled_ = v; }

  RuntimeProfile::Counter* total_cpu_timer() { return total_cpu_timer_; }
  RuntimeProfile::Counter* total_storage_wait_timer() {
    return total_storage_wait_timer_;
  }
  RuntimeProfile::Counter* total_network_send_timer() {
    return total_network_send_timer_;
  }
  RuntimeProfile::Counter* total_network_receive_timer() {
    return total_network_receive_timer_;
  }

  void set_query_status(const std::string& err_msg) {
    boost::lock_guard<SpinLock> l(query_status_lock_);
    if (!query_status_.ok()) return;
    query_status_ = Status(err_msg);
  }

  Status SetMemLimitExceeded(MemTracker* tracker = NULL,
      int64_t failed_allocation_size = 0);

  Status CheckQueryState();

  QueryResourceMgr* query_resource_mgr() const { return query_resource_mgr_; }
  void SetQueryResourceMgr(QueryResourceMgr* res_mgr) { query_resource_mgr_ = res_mgr; }

 private:
  Status Init(ExecEnv* exec_env);

  Status CreateCodegen();

  static const int DEFAULT_BATCH_SIZE = 1024;

  DescriptorTbl* desc_tbl_;
  boost::scoped_ptr<ObjectPool> obj_pool_;

  SpinLock error_log_lock_;

  ErrorLogMap error_log_;

  SpinLock file_errors_lock_;

  std::vector<std::pair<std::string, int> > file_errors_;

  TPlanFragmentInstanceCtx fragment_instance_ctx_;

  boost::scoped_ptr<TimestampValue> now_;

  std::string cgroup_;
  ExecEnv* exec_env_;
  boost::scoped_ptr<LlvmCodeGen> codegen_;

  ThreadResourceMgr::ResourcePool* resource_pool_;

  FileMoveMap hdfs_files_to_move_;

  PartitionStatusMap per_partition_status_;

  RuntimeProfile profile_;

  RuntimeProfile::Counter* total_cpu_timer_;

  RuntimeProfile::Counter* total_storage_wait_timer_;

  RuntimeProfile::Counter* total_network_send_timer_;

  RuntimeProfile::Counter* total_network_receive_timer_;

  boost::shared_ptr<MemTracker> query_mem_tracker_;

  boost::scoped_ptr<MemTracker> instance_mem_tracker_;

  bool is_cancelled_;

  SpinLock query_status_lock_;
  Status query_status_;

  QueryResourceMgr* query_resource_mgr_;

  std::vector<DiskIoMgr::RequestContext*> reader_contexts_;

  boost::shared_ptr<BufferedBlockMgr> block_mgr_;

  PlanNodeId root_node_id_;

  SpinLock bitmap_lock_;

  boost::unordered_map<SlotId, Bitmap*> slot_bitmap_filters_;

  RuntimeState(const RuntimeState&);
};

#define RETURN_IF_CANCELLED(state) \
  do { \
    if (UNLIKELY((state)->is_cancelled())) return Status::CANCELLED; \
  } while (false)

}

#endif