plan-fragment-executor.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_PLAN_EXECUTOR_H
#define IMPALA_SERVICE_PLAN_EXECUTOR_H

#include <vector>
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/function.hpp>

#include "common/status.h"
#include "common/object-pool.h"
#include "runtime/runtime-state.h"
#include "util/thread.h"

namespace impala {

class HdfsFsCache;
class ExecNode;
class RowDescriptor;
class RowBatch;
class DataSink;
class DataStreamMgr;
class RuntimeProfile;
class RuntimeState;
class TRowBatch;
class TPlanExecRequest;
class TPlanFragment;
class TPlanFragmentExecParams;
class TPlanExecParams;

//
//
class PlanFragmentExecutor {
 public:
  typedef boost::function<
      void (const Status& status, RuntimeProfile* profile, bool done)>
      ReportStatusCallback;

  PlanFragmentExecutor(ExecEnv* exec_env, const ReportStatusCallback& report_status_cb);

  ~PlanFragmentExecutor();

  Status Prepare(const TExecPlanFragmentParams& request);

  Status Open();

  Status GetNext(RowBatch** batch);

  void Close();

  void Cancel();

  bool ReachedLimit();

  void ReleaseThreadToken();

  RuntimeState* runtime_state() { return runtime_state_.get(); }
  const RowDescriptor& row_desc();

  RuntimeProfile* profile();

  static const std::string PER_HOST_PEAK_MEM_COUNTER;

 private:
  ExecEnv* exec_env_;  // not owned
  ExecNode* plan_;  // lives in runtime_state_->obj_pool()
  TUniqueId query_id_;

  ReportStatusCallback report_status_cb_;
  boost::scoped_ptr<Thread> report_thread_;
  boost::mutex report_thread_lock_;

  boost::condition_variable stop_report_thread_cv_;

  boost::condition_variable report_thread_started_cv_;
  bool report_thread_active_;  // true if we started the thread

  bool done_;

  bool prepared_;

  bool closed_;

  bool has_thread_token_;

  Status status_;

  boost::mutex status_lock_;

  boost::scoped_ptr<DataSink> sink_;
  boost::scoped_ptr<RuntimeState> runtime_state_;
  boost::scoped_ptr<RowBatch> row_batch_;
  boost::scoped_ptr<TRowBatch> thrift_batch_;

  RuntimeProfile::Counter* per_host_mem_usage_;

  RuntimeProfile::Counter* rows_produced_counter_;

  RuntimeProfile::Counter* average_thread_tokens_;

  MonotonicStopWatch fragment_sw_;

  AtomicInt<int> completed_report_sent_;

  RuntimeProfile::TimeSeriesCounter* mem_usage_sampled_counter_;

  RuntimeProfile::TimeSeriesCounter* thread_usage_sampled_counter_;

  ObjectPool* obj_pool() { return runtime_state_->obj_pool(); }

  typedef std::map<TPlanNodeId, std::vector<TScanRangeParams> > PerNodeScanRanges;

  void ReportProfile();

  void SendReport(bool done);

  void UpdateStatus(const Status& status);

  void FragmentComplete();

  void OptimizeLlvmModule();

  Status OpenInternal();

  Status GetNextInternal(RowBatch** batch);

  void StopReportThread();

  void PrintVolumeIds(const TPlanExecParams& params);
  void PrintVolumeIds(const PerNodeScanRanges& per_node_scan_ranges);

  const DescriptorTbl& desc_tbl() { return runtime_state_->desc_tbl(); }
};

}

#endif