aggregate_hash_table.h

/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */
 
#pragma once
 
#include "common/lang/vector.h"
#include "common/lang/unordered_map.h"
#include "common/math/simd_util.h"
#include "common/sys/rc.h"
#include "sql/expr/expression.h"
 
class AggregateHashTable
{
public:
  class Scanner
  {
  public:
    explicit Scanner(AggregateHashTable *hash_table) : hash_table_(hash_table) {}
    virtual ~Scanner() = default;
 
    virtual void open_scan() = 0;
 
    virtual RC next(Chunk &chunk) = 0;
 
    virtual void close_scan() {}
 
  protected:
    AggregateHashTable *hash_table_;
  };
 
  virtual RC add_chunk(Chunk &groups_chunk, Chunk &aggrs_chunk) = 0;
 
  virtual ~AggregateHashTable() = default;
  vector<AggregateExpr::Type> aggr_types_;
  vector<AttrType>            aggr_child_types_;
};
 
class StandardAggregateHashTable : public AggregateHashTable
{
private:
  struct VectorHash
  {
    size_t operator()(const vector<Value> &vec) const;
  };
 
  struct VectorEqual
  {
    bool operator()(const vector<Value> &lhs, const vector<Value> &rhs) const;
  };
 
public:
  using StandardHashTable = unordered_map<vector<Value>, vector<void *>, VectorHash, VectorEqual>;
  class Scanner : public AggregateHashTable::Scanner
  {
  public:
    explicit Scanner(AggregateHashTable *hash_table) : AggregateHashTable::Scanner(hash_table) {}
    ~Scanner() = default;
 
    void open_scan() override;
 
    RC next(Chunk &chunk) override;
 
  private:
    StandardHashTable::iterator end_;
    StandardHashTable::iterator it_;
  };
  StandardAggregateHashTable(const vector<Expression *> aggregations)
  {
    for (auto &expr : aggregations) {
      ASSERT(expr->type() == ExprType::AGGREGATION, "expect aggregate expression");
      auto *aggregation_expr = static_cast<AggregateExpr *>(expr);
      aggr_types_.push_back(aggregation_expr->aggregate_type());
      aggr_child_types_.push_back(aggregation_expr->value_type());
    }
  }
  virtual ~StandardAggregateHashTable()
  {
    for (auto &aggr : aggr_values_) {
      for (auto &state : aggr.second) {
        free(state);
      }
    }
  }
 
  RC add_chunk(Chunk &groups_chunk, Chunk &aggrs_chunk) override;
 
  StandardHashTable::iterator begin() { return aggr_values_.begin(); }
  StandardHashTable::iterator end() { return aggr_values_.end(); }
 
  StandardHashTable aggr_values_;
};
 
#ifdef USE_SIMD
template <typename V>
class LinearProbingAggregateHashTable : public AggregateHashTable
{
public:
  class Scanner : public AggregateHashTable::Scanner
  {
  public:
    explicit Scanner(AggregateHashTable *hash_table) : AggregateHashTable::Scanner(hash_table) {}
    ~Scanner() = default;
 
    void open_scan() override;
 
    RC next(Chunk &chunk) override;
 
    void close_scan() override;
 
  private:
    int capacity_   = -1;
    int size_       = -1;
    int scan_pos_   = -1;
    int scan_count_ = 0;
  };
 
  LinearProbingAggregateHashTable(AggregateExpr::Type aggregate_type, int capacity = DEFAULT_CAPACITY)
      : keys_(capacity, EMPTY_KEY), values_(capacity, 0), capacity_(capacity), aggregate_type_(aggregate_type)
  {}
  virtual ~LinearProbingAggregateHashTable() {}
 
  RC get(int key, V &value);
 
  RC iter_get(int pos, int &key, V &value);
 
  RC add_chunk(Chunk &group_chunk, Chunk &aggr_chunk) override;
 
  int capacity() { return capacity_; }
  int size() { return size_; }
 
private:
  void add_batch(int *input_keys, V *input_values, int len);
 
  void aggregate(V *value, V value_to_aggregate);
 
  void resize();
 
  void resize_if_need();
 
private:
  static const int EMPTY_KEY;
  static const int DEFAULT_CAPACITY;
 
  vector<int>         keys_;
  vector<V>           values_;
  int                 size_     = 0;
  int                 capacity_ = 0;
  AggregateExpr::Type aggregate_type_;
};
#endif  // USE_SIMD