partitioned-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_PARTITIONED_HASH_JOIN_NODE_H
#define IMPALA_EXEC_PARTITIONED_HASH_JOIN_NODE_H

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

#include "exec/blocking-join-node.h"
#include "exec/exec-node.h"
#include "exec/hash-table.h"
#include "runtime/buffered-block-mgr.h"

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

namespace impala {

class BufferedBlockMgr;
class MemPool;
class RowBatch;
class TupleRow;
class BufferedTupleStream;

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

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

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

 private:
  class Partition;

  //
  enum State {
    PARTITIONING_BUILD,

    PROCESSING_PROBE,

    PROBING_SPILLED_PARTITION,

    REPARTITIONING,
  };

  static const int PARTITION_FANOUT = 16;

  static const int NUM_PARTITIONING_BITS = 4;

  static const int MAX_PARTITION_DEPTH = 16;

  static const int MAX_IN_MEM_BUILD_TABLES = PARTITION_FANOUT;

  bool AppendRow(BufferedTupleStream* stream, TupleRow* row);

  bool AppendRowStreamFull(BufferedTupleStream* stream, TupleRow* row);

  Status SpillPartition(Partition** spilled_partition);

  Status ProcessBuildInput(RuntimeState* state, int level);

  Status ProcessBuildBatch(RowBatch* build_batch);

  Status BuildHashTables(RuntimeState* state);

  template<int const JoinOp>
  int ProcessProbeBatch(RowBatch* out_batch, HashTableCtx* ht_ctx);

  int ProcessProbeBatch(
      const TJoinOp::type join_op, RowBatch* out_batch, HashTableCtx* ht_ctx);

  void OutputUnmatchedBuild(RowBatch* out_batch);

  Status PrepareNullAwarePartition();

  Status OutputNullAwareProbeRows(RuntimeState* state, RowBatch* out_batch);

  Status EvaluateNullProbe(BufferedTupleStream* build);

  Status PrepareNullAwareNullProbe();

  Status OutputNullAwareNullProbe(RuntimeState* state, RowBatch* out_batch);

  Status CleanUpHashPartitions(RowBatch* batch);

  Status ReserveTupleStreamBlocks();

  Status NextProbeRowBatch(RuntimeState*, RowBatch* out_batch);

  Status NextSpilledProbeRowBatch(RuntimeState*, RowBatch* out_batch);

  Status PrepareNextPartition(RuntimeState*);

  int64_t LargestSpilledPartition() const;

  void ResetForProbe();

  bool AllocateProbeFilters(RuntimeState* state);

  bool AttachProbeFilters(RuntimeState* state);

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

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

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

  std::string PrintState() const;

  void UpdateState(State s);

  std::string NodeDebugString() const;

  int MinRequiredBuffers() const {
    int num_reserved_buffers = PARTITION_FANOUT * 2 + 2;
    num_reserved_buffers += join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN ? 3 : 0;
    return num_reserved_buffers;
  }

  RuntimeState* runtime_state_;

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

  std::vector<ExprContext*> other_join_conjunct_ctxs_;

  bool using_small_buffers_;

  State state_;
  Status status_;

  BufferedBlockMgr::Client* block_mgr_client_;

  boost::scoped_ptr<HashTableCtx> ht_ctx_;

  HashTable::Iterator hash_tbl_iterator_;

  RuntimeProfile::Counter* partition_build_timer_;

  RuntimeProfile::Counter* num_hash_buckets_;

  RuntimeProfile::Counter* partitions_created_;

  RuntimeProfile::HighWaterMarkCounter* max_partition_level_;

  RuntimeProfile::Counter* num_build_rows_partitioned_;
  RuntimeProfile::Counter* num_probe_rows_partitioned_;

  RuntimeProfile::Counter* num_repartitions_;

  RuntimeProfile::Counter* num_spilled_partitions_;

  RuntimeProfile::HighWaterMarkCounter* largest_partition_percent_;

  RuntimeProfile::Counter* null_aware_eval_timer_;

  class Partition {
   public:
    Partition(RuntimeState* state, PartitionedHashJoinNode* parent, int level,
        bool use_small_buffers);
    ~Partition();

    BufferedTupleStream* build_rows() { return build_rows_; }
    BufferedTupleStream* probe_rows() { return probe_rows_; }
    HashTable* hash_tbl() const { return hash_tbl_.get(); }

    bool is_closed() const { return is_closed_; }
    bool is_spilled() const { return is_spilled_; }

    void Close(RowBatch* batch);

    int64_t EstimatedInMemSize() const;

    int64_t InMemSize() const;

    template<bool const AddProbeFilters>
    Status BuildHashTableInternal(RuntimeState* state, bool* built);

    Status BuildHashTable(RuntimeState* state, bool* built, const bool add_probe_filters);

    Status Spill(bool unpin_all_build);

   private:
    friend class PartitionedHashJoinNode;

    PartitionedHashJoinNode* parent_;

    bool is_closed_;

    bool is_spilled_;

    int level_;

    boost::scoped_ptr<HashTable> hash_tbl_;

    BufferedTupleStream* build_rows_;
    BufferedTupleStream* probe_rows_;
  };

  typedef Status (*ProcessBuildBatchFn)(PartitionedHashJoinNode*, RowBatch*);
  ProcessBuildBatchFn process_build_batch_fn_;
  ProcessBuildBatchFn process_build_batch_fn_level0_;

  typedef int (*ProcessProbeBatchFn)(
      PartitionedHashJoinNode*, RowBatch*, HashTableCtx*);
  ProcessProbeBatchFn process_probe_batch_fn_;
  ProcessProbeBatchFn process_probe_batch_fn_level0_;

  std::list<Partition*> spilled_partitions_;

  std::vector<Partition*> hash_partitions_;

  HashTable* hash_tbls_[PARTITION_FANOUT];

  Partition* input_partition_;

  std::list<Partition*> output_build_partitions_;

  std::vector<std::pair<SlotId, Bitmap*> > probe_filters_;

  Partition* null_aware_partition_;

  boost::scoped_ptr<RowBatch> nulls_build_batch_;

  bool non_empty_build_;

  BufferedTupleStream* null_probe_rows_;

  std::vector<bool> matched_null_probe_;

  int64_t null_probe_output_idx_;
};

}

#endif