coordinator.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_COORDINATOR_H
#define IMPALA_RUNTIME_COORDINATOR_H

#include <vector>
#include <string>
#include <boost/scoped_ptr.hpp>
#include <boost/accumulators/accumulators.hpp>
#include <boost/accumulators/statistics/stats.hpp>
#include <boost/accumulators/statistics/min.hpp>
#include <boost/accumulators/statistics/mean.hpp>
#include <boost/accumulators/statistics/median.hpp>
#include <boost/accumulators/statistics/max.hpp>
#include <boost/accumulators/statistics/variance.hpp>
#include <boost/unordered_map.hpp>
#include <boost/unordered_set.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>

#include "common/status.h"
#include "common/global-types.h"
#include "util/progress-updater.h"
#include "util/runtime-profile.h"
#include "runtime/runtime-state.h"
#include "statestore/simple-scheduler.h"
#include "gen-cpp/Types_types.h"
#include "gen-cpp/Frontend_types.h"

namespace impala {

class DataStreamMgr;
class DataSink;
class RowBatch;
class RowDescriptor;
class PlanFragmentExecutor;
class ObjectPool;
class RuntimeState;
class ImpalaInternalServiceClient;
class Expr;
class ExprContext;
class ExecEnv;
class TUpdateCatalogRequest;
class TQueryExecRequest;
class TReportExecStatusParams;
class TRowBatch;
class TPlanExecRequest;
class TRuntimeProfileTree;
class RuntimeProfile;
class TablePrinter;

//
//
class Coordinator {
 public:
  Coordinator(ExecEnv* exec_env, RuntimeProfile::EventSequence* events);
  ~Coordinator();

  Status Exec(QuerySchedule& schedule, std::vector<ExprContext*>* output_expr_ctxs);

  Status Wait();

  Status GetNext(RowBatch** batch, RuntimeState* state);

  void Cancel(const Status* cause = NULL);

  Status UpdateFragmentExecStatus(const TReportExecStatusParams& params);

  RuntimeState* runtime_state();
  const RowDescriptor& row_desc() const;

  MemTracker* query_mem_tracker();

  RuntimeProfile* query_profile() const { return query_profile_.get(); }

  const TUniqueId& query_id() const { return query_id_; }

  const PartitionStatusMap& per_partition_status() { return per_partition_status_; }

  bool PrepareCatalogUpdate(TUpdateCatalogRequest* catalog_update);

  std::string GetErrorLog();

  const ProgressUpdater& progress() { return progress_; }

  Status GetStatus();

  const TExecSummary& exec_summary() const {
    exec_summary_lock_.DCheckLocked();
    return exec_summary_;
  }

  SpinLock& GetExecSummaryLock() const { return exec_summary_lock_; }

 private:
  class BackendExecState;

  typedef boost::accumulators::accumulator_set<int64_t,
      boost::accumulators::features<
      boost::accumulators::tag::min,
      boost::accumulators::tag::max,
      boost::accumulators::tag::mean,
      boost::accumulators::tag::variance>
  > SummaryStats;

  ExecEnv* exec_env_;
  TUniqueId query_id_;

  TDescriptorTable desc_tbl_;
  TQueryCtx query_ctx_;

  TStmtType::type stmt_type_;

  typedef std::map<PlanNodeId, RuntimeProfile::Counter*> CounterMap;

  struct FragmentInstanceCounters {
    CounterMap throughput_counters;

    CounterMap scan_ranges_complete_counters;
  };

  std::vector<BackendExecState*> backend_exec_states_;

  bool needs_finalization_;

  TFinalizeParams finalize_params_;

  boost::mutex wait_lock_;

  bool has_called_wait_;  // if true, Wait() was called; protected by wait_lock_

  ProgressUpdater progress_;

  boost::mutex lock_;

  Status query_status_;

  bool returned_all_results_;

  boost::scoped_ptr<PlanFragmentExecutor> executor_;

  boost::shared_ptr<MemTracker> query_mem_tracker_;

  const RowDescriptor* row_desc_;

  typedef boost::unordered_map<TUniqueId, BackendExecState*> BackendExecStateMap;
  BackendExecStateMap backend_exec_state_map_;

  ObjectPool* obj_pool() { return obj_pool_.get(); }

  bool execution_completed_;

  int num_remote_fragements_complete_;

  boost::condition_variable backend_completion_cv_;

  int num_remaining_backends_;


  PartitionStatusMap per_partition_status_;

  FileMoveMap files_to_move_;

  boost::scoped_ptr<ObjectPool> obj_pool_;

  mutable SpinLock exec_summary_lock_;
  TExecSummary exec_summary_;

  boost::unordered_map<TPlanNodeId, int> plan_node_id_to_summary_map_;

  boost::scoped_ptr<RuntimeProfile> query_profile_;

  RuntimeProfile::EventSequence* query_events_;

  struct PerFragmentProfileData {
    RuntimeProfile* averaged_profile;

    int num_instances;

    RuntimeProfile* root_profile;

    SummaryStats bytes_assigned;

    SummaryStats completion_times;

    SummaryStats rates;
  };

  std::vector<PerFragmentProfileData> fragment_profiles_;

  FragmentInstanceCounters coordinator_counters_;

  boost::unordered_set<TNetworkAddress> unique_hosts_;

  RuntimeProfile::Counter* finalization_timer_;

  void SetExecPlanFragmentParams(QuerySchedule& schedule,
      int backend_num, const TPlanFragment& fragment,
      int fragment_idx, const FragmentExecParams& params, int instance_idx,
      const TNetworkAddress& coord, TExecPlanFragmentParams* rpc_params);

  Status ExecRemoteFragment(void* exec_state);

  int GetFragmentNum(const TUniqueId& fragment_id);

  void PrintBackendInfo();

  void CreateAggregateCounters(const std::vector<TPlanFragment>& fragments);

  void CollectScanNodeCounters(RuntimeProfile*, FragmentInstanceCounters* result);

  int64_t ComputeTotalThroughput(int node_id);

  int64_t ComputeTotalScanRangesComplete(int node_id);

  void CancelInternal();

  void CancelRemoteFragments();

  Status UpdateStatus(const Status& status, const TUniqueId* failed_fragment);

  Status WaitForAllBackends();

  Status FinalizeQuery();

  Status FinalizeSuccessfulInsert();

  void InitExecProfile(const TQueryExecRequest& request);

  void UpdateAverageProfile(BackendExecState* backend_exec_state);

  void ComputeFragmentSummaryStats(BackendExecState* backend_exec_state);

  void ReportQuerySummary();

  void UpdateExecSummary(int fragment_idx, int instance_idx, RuntimeProfile* profile);

  typedef boost::unordered_map<std::string, std::pair<bool, short> > PermissionCache;
  void PopulatePathPermissionCache(hdfsFS fs, const std::string& path_str,
      PermissionCache* permissions_cache);
};

}

#endif