old-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_OLD_HASH_TABLE_H
#define IMPALA_EXEC_OLD_HASH_TABLE_H

#include <vector>
#include <boost/cstdint.hpp>
#include "codegen/impala-ir.h"
#include "common/logging.h"
#include "runtime/mem-pool.h"
#include "util/bitmap.h"
#include "util/hash-util.h"
#include "util/runtime-profile.h"

namespace llvm {
  class Function;
}

namespace impala {

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


//
//
//
//
class OldHashTable {
 private:
  struct Node;

 public:
  class Iterator;

  OldHashTable(RuntimeState* state, const std::vector<ExprContext*>& build_expr_ctxs,
      const std::vector<ExprContext*>& probe_expr_ctxs, int num_build_tuples,
      bool stores_nulls, bool finds_nulls, int32_t initial_seed,
      MemTracker* mem_tracker, bool stores_tuples = false, int64_t num_buckets = 1024);

  void Close();

  bool IR_ALWAYS_INLINE Insert(TupleRow* row) {
    if (UNLIKELY(mem_limit_exceeded_)) return false;
    bool has_null = EvalBuildRow(row);
    if (!stores_nulls_ && has_null) return true;

    if (UNLIKELY(num_filled_buckets_ > num_buckets_till_resize_)) {
      ResizeBuckets(num_buckets_ * 2);
      if (UNLIKELY(mem_limit_exceeded_)) return false;
    }
    return InsertImpl(row);
  }

  bool IR_ALWAYS_INLINE Insert(Tuple* tuple) {
    if (UNLIKELY(mem_limit_exceeded_)) return false;
    bool has_null = EvalBuildRow(reinterpret_cast<TupleRow*>(&tuple));
    if (!stores_nulls_ && has_null) return true;

    if (UNLIKELY(num_filled_buckets_ > num_buckets_till_resize_)) {
      ResizeBuckets(num_buckets_ * 2);
      if (UNLIKELY(mem_limit_exceeded_)) return false;
    }
    return InsertImpl(tuple);
  }

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

  Iterator IR_ALWAYS_INLINE Find(TupleRow* probe_row);

  int64_t size() const { return num_nodes_; }

  int64_t num_buckets() const { return buckets_.size(); }

  float load_factor() const {
    return num_filled_buckets_ / static_cast<float>(buckets_.size());
  }

  static int64_t EstimateSize(int64_t num_rows) {
    int64_t num_buckets = num_rows * 2;
    return num_buckets * sizeof(Bucket) + num_rows * sizeof(Node);
  }

  int64_t byte_size() const {
    int64_t nodes_mem = (num_nodes_ + node_remaining_current_page_) * sizeof(Node);
    return nodes_mem + sizeof(Bucket) * buckets_.capacity();
  }

  bool mem_limit_exceeded() const { return mem_limit_exceeded_; }

  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];
  }

  void AddBitmapFilters();

  Iterator Begin();

  Iterator FirstUnmatched();

  Iterator End() { return Iterator(); }

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

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

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

  static const char* LLVM_CLASS_NAME;

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

  class Iterator {
   public:
    Iterator() : table_(NULL), bucket_idx_(-1), node_(NULL) {
    }

    template<bool check_match>
    void IR_ALWAYS_INLINE Next();

    bool NextUnmatched();

    TupleRow* GetRow() {
      DCHECK(!AtEnd());
      DCHECK(!table_->stores_tuples_);
      return reinterpret_cast<TupleRow*>(node_->data);
    }

    Tuple* GetTuple() {
      DCHECK(!AtEnd());
      DCHECK(table_->stores_tuples_);
      return reinterpret_cast<Tuple*>(node_->data);
    }

    void set_matched(bool v) {
      DCHECK(!AtEnd());
      node_->matched = v;
    }

    bool matched() const {
      DCHECK(!AtEnd());
      return node_->matched;
    }

    void reset() {
      bucket_idx_ = -1;
      node_ = NULL;
    }

    bool AtEnd() const { return node_ == NULL; }
    bool operator!=(const Iterator& rhs) { return !(*this == rhs); }

    bool operator==(const Iterator& rhs) {
      return bucket_idx_ == rhs.bucket_idx_ && node_ == rhs.node_;
    }

   private:
    friend class OldHashTable;

    Iterator(OldHashTable* table, int bucket_idx, Node* node, uint32_t hash) :
      table_(table),
      bucket_idx_(bucket_idx),
      node_(node),
      scan_hash_(hash) {
    }

    OldHashTable* table_;

    int64_t bucket_idx_;

    Node* node_;

    uint32_t scan_hash_;
  };

 private:
  friend class Iterator;
  friend class OldHashTableTest;

  struct Node {
    bool matched;

    uint32_t hash;   // Cache of the hash for data_
    Node* next;      // Chain to next node for collisions
    void* data;      // Either the Tuple* or TupleRow*
  };

  struct Bucket {
    Node* node;
    Bucket() : node(NULL) { }
  };

  Bucket* NextBucket(int64_t* bucket_idx);

  void ResizeBuckets(int64_t num_buckets);

  bool IR_ALWAYS_INLINE InsertImpl(void* data);

  void AddToBucket(Bucket* bucket, Node* node);

  void MoveNode(Bucket* from_bucket, Bucket* to_bucket, Node* node,
                Node* previous_node);

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

  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 IR_NO_INLINE HashCurrentRow() {
    if (var_result_begin_ == -1) {
      return HashUtil::Hash(expr_values_buffer_, results_buffer_size_, initial_seed_);
    } else {
      return HashVariableLenRow();
    }
  }

  TupleRow* GetRow(Node* node) const {
    if (stores_tuples_) {
      return reinterpret_cast<TupleRow*>(&node->data);
    } else {
      return reinterpret_cast<TupleRow*>(node->data);
    }
  }

  uint32_t HashVariableLenRow();

  bool Equals(TupleRow* build_row);

  void GrowNodeArray();

  void MemLimitExceeded(int64_t allocation_size);

  static const float MAX_BUCKET_OCCUPANCY_FRACTION;

  RuntimeState* state_;

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

  const int num_build_tuples_;

  const bool stores_nulls_;
  const bool finds_nulls_;
  const bool stores_tuples_;

  const int32_t initial_seed_;

  int64_t num_filled_buckets_;

  int64_t num_nodes_;

  boost::scoped_ptr<MemPool> mem_pool_;

  int num_data_pages_;

  Node* next_node_;

  int node_remaining_current_page_;

  MemTracker* mem_tracker_;

  bool mem_limit_exceeded_;

  std::vector<Bucket> buckets_;

  int64_t num_buckets_;

  int64_t num_buckets_till_resize_;

  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_;
};

}

#endif