hash-table.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_TABLE_H
#define IMPALA_EXEC_HASH_TABLE_H

#include <vector>
#include <boost/cstdint.hpp>
#include <boost/scoped_ptr.hpp>
#include "codegen/impala-ir.h"
#include "common/logging.h"
#include "runtime/buffered-block-mgr.h"
#include "runtime/buffered-tuple-stream.h"
#include "runtime/buffered-tuple-stream.inline.h"
#include "runtime/mem-tracker.h"
#include "runtime/tuple-row.h"
#include "util/bitmap.h"
#include "util/hash-util.h"

namespace llvm {
  class Function;
}

namespace impala {

class Expr;
class ExprContext;
class LlvmCodeGen;
class MemTracker;
class RowDescriptor;
class RuntimeState;
class Tuple;
class TupleRow;
class HashTable;

//
//
//
//
//

class HashTableCtx {
 public:
  HashTableCtx(const std::vector<ExprContext*>& build_expr_ctxs,
      const std::vector<ExprContext*>& probe_expr_ctxs, bool stores_nulls,
      bool finds_nulls, int32_t initial_seed, int max_levels,
      int num_build_tuples);

  void Close();

  void set_level(int level);
  int level() const { return level_; }
  uint32_t seed(int level) { return seeds_.at(level); }

  TupleRow* row() const { return row_; }

  void* last_expr_value(int expr_idx) const {
    return expr_values_buffer_ + expr_values_buffer_offsets_[expr_idx];
  }

  bool last_expr_value_null(int expr_idx) const {
    return expr_value_null_bits_[expr_idx];
  }

  bool IR_ALWAYS_INLINE EvalAndHashBuild(TupleRow* row, uint32_t* hash);
  bool IR_ALWAYS_INLINE EvalAndHashProbe(TupleRow* row, uint32_t* hash);

  int results_buffer_size() const { return results_buffer_size_; }

  llvm::Function* CodegenEvalRow(RuntimeState* state, bool build_row);

  llvm::Function* CodegenEquals(RuntimeState* state);

  llvm::Function* CodegenHashCurrentRow(RuntimeState* state, bool use_murmur);

  static const char* LLVM_CLASS_NAME;

 private:
  friend class HashTable;

  uint32_t IR_NO_INLINE HashCurrentRow() {
    DCHECK_LT(level_, seeds_.size());
    if (var_result_begin_ == -1) {
      return Hash(expr_values_buffer_, results_buffer_size_, seeds_[level_]);
    } else {
      return HashTableCtx::HashVariableLenRow();
    }
  }

  uint32_t inline Hash(const void* input, int len, int32_t hash) {
    if (level_ == 0) return HashUtil::Hash(input, len, hash);
    return HashUtil::MurmurHash2_64(input, len, hash);
  }

  bool IR_NO_INLINE EvalBuildRow(TupleRow* row) {
    return EvalRow(row, build_expr_ctxs_);
  }

  bool IR_NO_INLINE EvalProbeRow(TupleRow* row) {
    return EvalRow(row, probe_expr_ctxs_);
  }

  uint32_t HashVariableLenRow();

  bool EvalRow(TupleRow* row, const std::vector<ExprContext*>& ctxs);

  bool IR_NO_INLINE Equals(TupleRow* build_row);

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

  const bool stores_nulls_;
  const bool finds_nulls_;

  int level_;

  std::vector<uint32_t> seeds_;

  std::vector<int> expr_values_buffer_offsets_;

  int var_result_begin_;

  int results_buffer_size_;

  uint8_t* expr_values_buffer_;

  uint8_t* expr_value_null_bits_;

  TupleRow* row_;

  uint32_t GetHashSeed() const;
};

class HashTable {
 private:

  union HtData {
    BufferedTupleStream::RowIdx idx;
    Tuple* tuple;
  };

  struct DuplicateNode {
    bool matched;

    DuplicateNode* next; // Chain to next duplicate node, NULL when end of list.
    HtData htdata;
  };

  struct Bucket {
    bool filled;

    bool matched;

    bool hasDuplicates;

    uint32_t hash;

    union {
      HtData htdata;
      DuplicateNode* duplicates;
    } bucketData;
  };

 public:
  class Iterator;

  HashTable(RuntimeState* state, BufferedBlockMgr::Client* client,
      int num_build_tuples, BufferedTupleStream* tuple_stream,
      int64_t max_num_buckets, int64_t initial_num_buckets = 1024);

  HashTable(MemPool* pool, bool quadratic_probing, int num_buckets);

  bool Init();

  void Close();

  bool IR_ALWAYS_INLINE Insert(HashTableCtx* ht_ctx,
      const BufferedTupleStream::RowIdx& idx, TupleRow* row, uint32_t hash);

  bool IR_ALWAYS_INLINE Insert(HashTableCtx* ht_ctx, Tuple* tuple, uint32_t hash);

  Iterator IR_ALWAYS_INLINE Find(HashTableCtx* ht_ctx, uint32_t hash);

  int64_t size() const {
    return num_filled_buckets_ - num_buckets_with_duplicates_ + num_duplicate_nodes_;
  }

  int64_t EmptyBuckets() const { return num_buckets_ - num_filled_buckets_; }

  int64_t num_buckets() const { return num_buckets_; }

  double load_factor() const {
    return static_cast<double>(num_filled_buckets_) / num_buckets_;
  }

  static int64_t EstimateNumBuckets(int64_t num_rows) {
    return BitUtil::NextPowerOfTwo(3 * num_rows / 2);
  }
  static int64_t EstimateSize(int64_t num_rows) {
    int64_t num_buckets = EstimateNumBuckets(num_rows);
    return num_buckets * sizeof(Bucket);
  }

  int64_t CurrentMemSize() const;

  bool CheckAndResize(uint64_t buckets_to_fill, HashTableCtx* ht_ctx);

  int64_t byte_size() const { return total_data_page_size_; }

  Iterator Begin(HashTableCtx* ht_ctx);

  Iterator FirstUnmatched(HashTableCtx* ctx);

  bool HasMatches() const { return has_matches_; }

  Iterator End() { return Iterator(); }

  std::string DebugString(bool skip_empty, bool show_match,
      const RowDescriptor* build_desc);

  void DebugStringTuple(std::stringstream& ss, HtData& htdata, const RowDescriptor* desc);

  std::string PrintStats() const;

  class Iterator {
   private:
    static const int64_t BUCKET_NOT_FOUND = -1;

   public:

    Iterator() : table_(NULL), row_(NULL), bucket_idx_(BUCKET_NOT_FOUND), node_(NULL) { }

    void IR_ALWAYS_INLINE Next();

    void IR_ALWAYS_INLINE NextDuplicate();

    void NextUnmatched();

    TupleRow* GetRow() const;
    Tuple* GetTuple() const;

    void SetMatched();

    bool IsMatched() const;

    void SetAtEnd();

    bool AtEnd() const { return bucket_idx_ == BUCKET_NOT_FOUND; }

   private:
    friend class HashTable;

    Iterator(HashTable* table, TupleRow* row, int bucket_idx, DuplicateNode* node,
             uint32_t hash)
      : table_(table),
        row_(row),
        bucket_idx_(bucket_idx),
        node_(node) {
    }

    HashTable* table_;
    TupleRow* row_;

    int64_t bucket_idx_;

    DuplicateNode* node_;
  };

 private:
  friend class Iterator;
  friend class HashTableTest;

  //
  int64_t IR_ALWAYS_INLINE Probe(Bucket* buckets, int64_t num_buckets,
      HashTableCtx* ht_ctx, uint32_t hash,  bool* found);

  HtData* IR_ALWAYS_INLINE InsertInternal(HashTableCtx* ht_ctx, uint32_t hash);

  void NextFilledBucket(int64_t* bucket_idx, DuplicateNode** node);

  bool ResizeBuckets(int64_t num_buckets, HashTableCtx* ht_ctx);

  DuplicateNode* IR_ALWAYS_INLINE AppendNextNode(Bucket* bucket);

  DuplicateNode* IR_ALWAYS_INLINE InsertDuplicateNode(int64_t bucket_idx);

  void IR_ALWAYS_INLINE PrepareBucketForInsert(int64_t bucket_idx, uint32_t hash);

  TupleRow* GetRow(HtData& htdata, TupleRow* row) const;

  TupleRow* GetRow(Bucket* bucket, TupleRow* row) const;

  bool GrowNodeArray();

  static const double MAX_FILL_FACTOR;

  RuntimeState* state_;

  BufferedBlockMgr::Client* block_mgr_client_;

  BufferedTupleStream* tuple_stream_;

  MemPool* data_page_pool_;

  const bool stores_tuples_;

  const bool quadratic_probing_;

  std::vector<BufferedBlockMgr::Block*> data_pages_;

  int64_t total_data_page_size_;

  DuplicateNode* next_node_;

  int node_remaining_current_page_;

  int64_t num_duplicate_nodes_;

  const int64_t max_num_buckets_;

  Bucket* buckets_;

  int64_t num_buckets_;

  int64_t num_filled_buckets_;

  int64_t num_buckets_with_duplicates_;

  int num_build_tuples_;

  bool has_matches_;

  int64_t num_probes_;

  int64_t num_failed_probes_;

  int64_t travel_length_;

  int64_t num_hash_collisions_;

  int64_t num_resizes_;
};

}

#endif