exec-node.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_EXEC_EXEC_NODE_H
#define IMPALA_EXEC_EXEC_NODE_H

#include <vector>
#include <sstream>

#include "common/status.h"
#include "runtime/descriptors.h"  // for RowDescriptor
#include "util/runtime-profile.h"
#include "util/blocking-queue.h"
#include "gen-cpp/PlanNodes_types.h"

namespace impala {

class Expr;
class ExprContext;
class ObjectPool;
class Counters;
class SortExecExprs;
class RowBatch;
class RuntimeState;
class TPlan;
class TupleRow;
class DataSink;
class MemTracker;

class ExecNode {
 public:
  ExecNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs);

  virtual ~ExecNode();

  virtual Status Init(const TPlanNode& tnode);

  virtual Status Prepare(RuntimeState* state);

  virtual Status Open(RuntimeState* state);

  virtual Status GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) = 0;

  virtual Status Reset(RuntimeState* state);

  virtual void Close(RuntimeState* state);

  static Status CreateTree(ObjectPool* pool, const TPlan& plan,
                           const DescriptorTbl& descs, ExecNode** root);

  static void SetDebugOptions(int node_id, TExecNodePhase::type phase,
                              TDebugAction::type action, ExecNode* tree);

  void CollectNodes(TPlanNodeType::type node_type, std::vector<ExecNode*>* nodes);

  void CollectScanNodes(std::vector<ExecNode*>* nodes);

  static bool EvalConjuncts(ExprContext* const* ctxs, int num_ctxs, TupleRow* row);

  static llvm::Function* CodegenEvalConjuncts(
      RuntimeState* state, const std::vector<ExprContext*>& conjunct_ctxs,
      const char* name = "EvalConjuncts");

  std::string DebugString() const;

  virtual void DebugString(int indentation_level, std::stringstream* out) const;
  const std::vector<ExprContext*>& conjunct_ctxs() const { return conjunct_ctxs_; }

  int id() const { return id_; }
  TPlanNodeType::type type() const { return type_; }
  const RowDescriptor& row_desc() const { return row_descriptor_; }
  int64_t rows_returned() const { return num_rows_returned_; }
  int64_t limit() const { return limit_; }
  bool ReachedLimit() { return limit_ != -1 && num_rows_returned_ >= limit_; }

  RuntimeProfile* runtime_profile() { return runtime_profile_.get(); }
  MemTracker* mem_tracker() { return mem_tracker_.get(); }
  MemTracker* expr_mem_tracker() { return expr_mem_tracker_.get(); }

  static int GetNodeIdFromProfile(RuntimeProfile* p);

  static const std::string ROW_THROUGHPUT_COUNTER;

 protected:
  friend class DataSink;

  class RowBatchQueue : public BlockingQueue<RowBatch*> {
   public:
    RowBatchQueue(int max_batches);
    ~RowBatchQueue();

    void AddBatch(RowBatch* batch);

    RowBatch* GetBatch();

    int Cleanup();

   private:
    SpinLock lock_;

    std::list<RowBatch*> cleanup_queue_;
  };

  int id_;  // unique w/in single plan tree
  TPlanNodeType::type type_;
  ObjectPool* pool_;
  std::vector<ExprContext*> conjunct_ctxs_;

  std::vector<ExecNode*> children_;
  RowDescriptor row_descriptor_;

  TExecNodePhase::type debug_phase_;
  TDebugAction::type debug_action_;

  int64_t limit_;  // -1: no limit
  int64_t num_rows_returned_;

  boost::scoped_ptr<RuntimeProfile> runtime_profile_;
  RuntimeProfile::Counter* rows_returned_counter_;
  RuntimeProfile::Counter* rows_returned_rate_;

  boost::scoped_ptr<MemTracker> mem_tracker_;

  boost::scoped_ptr<MemTracker> expr_mem_tracker_;

  boost::mutex exec_options_lock_;
  std::string runtime_exec_options_;

  ExecNode* child(int i) { return children_[i]; }
  bool is_closed() { return is_closed_; }

  static Status CreateNode(ObjectPool* pool, const TPlanNode& tnode,
                           const DescriptorTbl& descs, ExecNode** node);

  static Status CreateTreeHelper(ObjectPool* pool, const std::vector<TPlanNode>& tnodes,
      const DescriptorTbl& descs, ExecNode* parent, int* node_idx, ExecNode** root);

  virtual bool IsScanNode() const { return false; }

  void InitRuntimeProfile(const std::string& name);

  Status ExecDebugAction(TExecNodePhase::type phase, RuntimeState* state);

  void AddRuntimeExecOption(const std::string& option);

  //
  virtual Status QueryMaintenance(RuntimeState* state);

  void AddExprCtxToFree(ExprContext* ctx) { expr_ctxs_to_free_.push_back(ctx); }
  void AddExprCtxsToFree(const std::vector<ExprContext*>& ctxs);
  void AddExprCtxsToFree(const SortExecExprs& sort_exec_exprs);

 private:
  bool is_closed_;

  std::vector<ExprContext*> expr_ctxs_to_free_;
};

}
#endif