hash-join-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_HASH_JOIN_NODE_H
#define IMPALA_EXEC_HASH_JOIN_NODE_H

#include <boost/scoped_ptr.hpp>
#include <boost/thread.hpp>
#include <string>

#include "exec/exec-node.h"
#include "exec/old-hash-table.h"
#include "exec/blocking-join-node.h"
#include "util/promise.h"

#include "gen-cpp/PlanNodes_types.h"  // for TJoinOp

namespace impala {

class MemPool;
class RowBatch;
class TupleRow;

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

  virtual Status Init(const TPlanNode& tnode);
  virtual Status Prepare(RuntimeState* state);
  // Open() implemented in BlockingJoinNode
  virtual Status GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos);
  virtual Status Reset(RuntimeState* state);
  virtual void Close(RuntimeState* state);

  static const char* LLVM_CLASS_NAME;

 protected:
  virtual void AddToDebugString(int indentation_level, std::stringstream* out) const;
  virtual Status InitGetNext(TupleRow* first_probe_row);
  virtual Status ConstructBuildSide(RuntimeState* state);

 private:
  boost::scoped_ptr<OldHashTable> hash_tbl_;
  OldHashTable::Iterator hash_tbl_iterator_;

    std::vector<ExprContext*> probe_expr_ctxs_;
    std::vector<ExprContext*> build_expr_ctxs_;

  std::vector<ExprContext*> other_join_conjunct_ctxs_;

  bool match_all_probe_;

  bool match_one_build_;

  bool match_all_build_;

  llvm::Function* codegen_process_build_batch_fn_;

  typedef void (*ProcessBuildBatchFn)(HashJoinNode*, RowBatch*);
  ProcessBuildBatchFn process_build_batch_fn_;

  typedef int (*ProcessProbeBatchFn)(HashJoinNode*, RowBatch*, RowBatch*, int);
  ProcessProbeBatchFn process_probe_batch_fn_;

  RuntimeProfile::Counter* build_buckets_counter_;   // num buckets in hash table
  RuntimeProfile::Counter* hash_tbl_load_factor_counter_;

  Status LeftJoinGetNext(RuntimeState* state, RowBatch* row_batch, bool* eos);

  int ProcessProbeBatch(RowBatch* out_batch, RowBatch* probe_batch, int max_added_rows);

  void ProcessBuildBatch(RowBatch* build_batch);

  llvm::Function* CodegenCreateOutputRow(LlvmCodeGen* codegen);

  llvm::Function* CodegenProcessBuildBatch(RuntimeState* state, llvm::Function* hash_fn);

  llvm::Function* CodegenProcessProbeBatch(RuntimeState* state, llvm::Function* hash_fn);
};

}

#endif