expressions/aggregation/AggregationHandle.hpp - incubator-retired-quickstep - Git at Google

 /**
  *   Copyright 2011-2015 Quickstep Technologies LLC.
  *   Copyright 2015 Pivotal Software, 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 QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_HPP_
 #define QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_HPP_

 #include <cstddef>
 #include <memory>
 #include <vector>

 #include "catalog/CatalogTypedefs.hpp"
 #include "storage/HashTableBase.hpp"
 #include "types/TypedValue.hpp"
 #include "types/containers/ColumnVector.hpp"
 #include "utility/Macros.hpp"

 #include "glog/logging.h"

 namespace quickstep {

 class StorageManager;
 class Type;
 class ValueAccessor;

 /** \addtogroup Expressions
  *  @{
  */


 /**
  * @brief Abstract base class for aggregation state.
  **/
 class AggregationState {
  public:
   /**
    * @brief Default constructor.
    **/
   AggregationState() {}

   /**
    * @brief Pure virtual destructor.
    **/
   virtual ~AggregationState() = 0;
 };

 // Destructor should be defined. This will be called when derived class
 // destructor is called.
 inline AggregationState::~AggregationState() {}

 /**
  * @brief AggregationHandle encapsulates logic for actually computing
  *        aggregates with particular argument(s).
  * @note See also AggregateFunction, which represents a SQL aggregate function
  *       in the abstract sense.
  *
  * An AggregationHandle is created by calling
  * AggregateFunction::createHandle(). The AggregationHandle object provides
  * methods that are used to actually compute the aggregate, storing
  * intermediate results in AggregationState objects.
  *
  * The work-flow for computing an aggregate without GROUP BY is as follows:
  *     1. Create a global state for the aggregate with createInitialState().
  *     2. For each block in a relation (parallelizable):
  *        a. Call StorageBlock::aggregate() to accumulate results from the
  *           block (under the covers, this calls either
  *           accumulateColumnVectors() or accumulateValueAccessor() to do the
  *           actual per-block aggregation in a vectorized fashion).
  *        b. Merge the per-block results back with the global state by calling
  *           mergeStates() (this is threadsafe).
  *     3. Generate the final result by calling finalize() on the global state.
  *
  * The work-flow for computing an aggregate with GROUP BY is as follows:
  *     1. Create a HashTable to hold per-group states by calling
  *        createGroupByHashTable().
  *     2. For each block in a relation (parallelizable):
  *        a. Call StorageBlock::aggregateGroupBy() to update the states in the
  *           HashTable according to the values in the block (under the covers,
  *           this calls aggregateValueAccessorIntoHashTable() to aggregate over
  *           all the values/groups in a block in one shot; this is threadsafe).
  *     3. Generate the final set of groups and their corresponding results by
  *        calling finalizeHashTable().
  *
  * See also AggregationOperationState, which holds 1 or more global states or
  * HashTables for an aggregate query, and has some logic to re-use common
  * information across multiple aggregates being computed on the same block
  * (e.g. the set of matches for a predicate, or the values of computed GROUP BY
  * expressions). AggregationOperationState also has a method to write out
  * finalized aggregate values to an InsertDestination.
  **/
 class AggregationHandle {
  public:
   /**
    * @brief Virtual destructor.
    *
    **/
   virtual ~AggregationHandle() {
   }

   /**
    * @brief Create an initial "blank" state for this aggregation.
    *
    * @return An initial "blank" state for this particular aggregation.
    **/
   virtual AggregationState* createInitialState() const = 0;

   /**
    * @brief Create a new HashTable for aggregation with GROUP BY.
    *
    * @param hash_table_impl The choice of which concrete HashTable
    *        implementation to use.
    * @param group_by_types The types of the GROUP BY columns/expressions. These
    *        correspond to the (composite) key type for the HashTable.
    * @param estimated_num_groups The estimated number of distinct groups for
    *        the GROUP BY aggregation. This is used to size the initial
    *        HashTable. This is an estimate only, and the HashTable will be
    *        resized if it becomes over-full.
    * @param storage_manager The StorageManager to use to create the HashTable.
    *        A StorageBlob will be allocated to serve as the HashTable's
    *        in-memory storage.
    * @return A new HashTable instance with the appropriate state type for this
    *         aggregate as the ValueT.
    **/
   virtual AggregationStateHashTableBase* createGroupByHashTable(
       const HashTableImplType hash_table_impl,
       const std::vector<const Type*> &group_by_types,
       const std::size_t estimated_num_groups,
       StorageManager *storage_manager) const = 0;

   /**
    * @brief Accumulate over tuples for a nullary aggregate function (one that
    *        has zero arguments, i.e. COUNT(*)).
    *
    * @param num_tuples The number of tuples to "accumulate". No actual tuple
    *        data is accessed, the only thing that a nullary aggeregate can know
    *        about input is its cardinality.
    * @return A new AggregationState which contains the accumulated results from
    *         applying the (nullary) aggregate to the specified number of
    *         tuples.
    **/
   virtual AggregationState* accumulateNullary(
       const std::size_t num_tuples) const {
     LOG(FATAL) << "Called accumulateNullary on an AggregationHandle that "
                << "takes at least one argument.";
   }

   /**
    * @brief Accumulate (iterate over) all values in one or more ColumnVectors
    *        and return a new AggregationState which can be merged with other
    *        states or finalized.
    *
    * @param column_vectors One or more ColumnVectors that the aggregate will be
    *        applied to. These correspond to the aggregate function's arguments,
    *        in order.
    * @return A new AggregationState which contains the accumulated results from
    *         applying the aggregate to column_vectors. Caller is responsible
    *         for deleting the returned AggregationState.
    **/
   virtual AggregationState* accumulateColumnVectors(
       const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const = 0;

 #ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
   /**
    * @brief Accumulate (iterate over) all values in columns accessible through
    *        a ValueAccessor and return a new AggregationState which can be
    *        merged with other states or finalized.
    *
    * @param accessor A ValueAccessor that the columns to be aggregated can be
    *        accessed through.
    * @param accessor_ids The attribute_ids that correspond to the columns in
    *        accessor to aggeregate. These correspond to the aggregate
    *        function's arguments, in order.
    * @return A new AggregationState which contains the accumulated results from
    *         applying the aggregate to the specified columns in accessor.
    *         Caller is responsible for deleting the returned AggregationState.
    **/
   virtual AggregationState* accumulateValueAccessor(
       ValueAccessor *accessor,
       const std::vector<attribute_id> &accessor_ids) const = 0;
 #endif

   /**
    * @brief Perform an aggregation with GROUP BY over all the tuples accessible
    *        through a ValueAccessor, upserting states in a HashTable.
    *
    * @note Implementations of this method are threadsafe with respect to
    *       hash_table, and can be called concurrently from multiple threads
    *       with the same HashTable object.
    *
    * @param accessor The ValueAccessor that will be iterated over to read
    *        tuples.
    * @param argument_ids The attribute_ids of the arguments to this aggregate
    *        in accessor, in order.
    * @param group_by_key_ids The attribute_ids of the group-by
    *        columns/expressions in accessor.
    * @param hash_table The HashTable to upsert AggregationStates in. This
    *        should have been created by calling createGroupByHashTable() on
    *        this same AggregationHandle.
    **/
   virtual void aggregateValueAccessorIntoHashTable(
       ValueAccessor *accessor,
       const std::vector<attribute_id> &argument_ids,
       const std::vector<attribute_id> &group_by_key_ids,
       AggregationStateHashTableBase *hash_table) const = 0;

   /**
    * @brief Merge two AggregationStates, updating one in-place. This computes a
    *        new intermediate result and stores it in the destination
    *        AggregationState.
    *
    * @note The order of arguments (i.e. which AggregationState is source and
    *       which is destination) is not important for the result of the merging
    *       process, but it does determine which of the two states is
    *       overwritten (destination) and which remains unchanged (source).
    * @note Implementations of this method are threadsafe with respect to the
    *       destination state, and can be called concurrently from multiple
    *       threads with the same state object.
    *
    * @param source The AggregationState to merge "from".
    * @param destination The AggregationState to merge "to". The internal state
    *        will be overwritten with the merged result.
    **/
   virtual void mergeStates(const AggregationState &source,
                            AggregationState *destination) const = 0;

   /**
    * @brief Computes and returns the resulting aggregate by using intermediate
    *        result saved in this handle.
    *
    * @note Except for count, SQL89 aggeregates return NULL when no rows are
    *       selected.
    * @warning It is dangerous to assume that calling mergeStates() or some
    *          iterate method after previous finalize call will work correctly
    *          when the aggregate function is not one of SQL89 aggregates (SUM,
    *          COUNT, MIN, MAX, AVG).
    *
    * @return The result of this aggregation.
    **/
   virtual TypedValue finalize(const AggregationState &state) const = 0;

   /**
    * @brief Compute and return finalized aggregates for all groups in a
    *        HashTable.
    *
    * @param hash_table The HashTable to finalize states from. This should have
    *        have been created by calling createGroupByHashTable() on this same
    *        AggregationHandle.
    * @param group_by_keys A pointer to a vector of vectors of GROUP BY keys. If
    *        this is initially empty, it will be filled in with the GROUP BY
    *        keys visited by this method in the same order as the finalized
    *        values returned in the ColumnVector. If this is already filled in,
    *        then this method will visit the GROUP BY keys in the exact order
    *        specified.
    * @return A ColumnVector containing each group's finalized aggregate value.
    **/
   virtual ColumnVector* finalizeHashTable(
       const AggregationStateHashTableBase &hash_table,
       std::vector<std::vector<TypedValue>> *group_by_keys) const = 0;

  protected:
   AggregationHandle() {
   }

   template <typename HandleT,
             typename StateT,
             typename HashTableT>
   void aggregateValueAccessorIntoHashTableNullaryHelper(
       ValueAccessor *accessor,
       const std::vector<attribute_id> &group_by_key_ids,
       const StateT &default_state,
       AggregationStateHashTableBase *hash_table) const;

   template <typename HandleT,
             typename StateT,
             typename HashTableT>
   void aggregateValueAccessorIntoHashTableUnaryHelper(
       ValueAccessor *accessor,
       const attribute_id argument_id,
       const std::vector<attribute_id> &group_by_key_ids,
       const StateT &default_state,
       AggregationStateHashTableBase *hash_table) const;

   template <typename HandleT,
             typename HashTableT>
   ColumnVector* finalizeHashTableHelper(
       const Type &result_type,
       const AggregationStateHashTableBase &hash_table,
       std::vector<std::vector<TypedValue>> *group_by_keys) const;

   template <typename HandleT, typename HashTableT>
   inline TypedValue finalizeGroupInHashTable(
       const AggregationStateHashTableBase &hash_table,
       const std::vector<TypedValue> &group_key) const {
     const AggregationState *group_state
         = static_cast<const HashTableT&>(hash_table).getSingleCompositeKey(group_key);
     DCHECK(group_state != nullptr)
         << "Could not find entry for specified group_key in HashTable";
     return static_cast<const HandleT*>(this)->finalizeHashTableEntry(*group_state);
   }

  private:
   DISALLOW_COPY_AND_ASSIGN(AggregationHandle);
 };

 /**
  * @brief Templated class to implement value-accessor-based upserter for each
  *        aggregation state payload type. This version is for nullary
  *        aggregates (those that take no arguments).
  **/
 template <typename HandleT, typename StateT>
 class NullaryAggregationStateValueAccessorUpserter {
  public:
   explicit NullaryAggregationStateValueAccessorUpserter(const HandleT &handle)
       : handle_(handle) {
   }

   template <typename ValueAccessorT>
   inline void operator()(const ValueAccessorT &accessor, StateT *state) {
     handle_.iterateNullaryInl(state);
   }

  private:
   const HandleT &handle_;
 };

 /**
  * @brief Templated class to implement value-accessor-based upserter for each
  *        aggregation state payload type. This version is for unary aggregates
  *        (those that take a single argument).
  **/
 template <typename HandleT, typename StateT>
 class UnaryAggregationStateValueAccessorUpserter {
  public:
   UnaryAggregationStateValueAccessorUpserter(const HandleT &handle,
                                              attribute_id value_id)
     : handle_(handle),
       value_id_(value_id) {
   }

   template <typename ValueAccessorT>
   inline void operator()(const ValueAccessorT &accessor, StateT *state) {
     handle_.iterateUnaryInl(state, accessor.getTypedValue(value_id_));
   }

  private:
   const HandleT &handle_;
   const attribute_id value_id_;
 };

 /**
  * @brief Templated helper class used to implement
  *        AggregationHandle::finalizeHashTable() by visiting each entry (i.e.
  *        GROUP) in a HashTable, finalizing the aggregation for the GROUP, and
  *        collecting the GROUP BY key values and the final aggregate values in
  *        a ColumnVector.
  **/
 template <typename HandleT, typename ColumnVectorT>
 class HashTableAggregateFinalizer {
  public:
   HashTableAggregateFinalizer(const HandleT &handle,
                               std::vector<std::vector<TypedValue>> *group_by_keys,
                               ColumnVectorT *output_column_vector)
       : handle_(handle),
         group_by_keys_(group_by_keys),
         output_column_vector_(output_column_vector) {
   }

   inline void operator()(const std::vector<TypedValue> &group_by_key,
                          const AggregationState &group_state) {
     group_by_keys_->emplace_back(group_by_key);
     output_column_vector_->appendTypedValue(handle_.finalizeHashTableEntry(group_state));
   }

  private:
   const HandleT &handle_;
   std::vector<std::vector<TypedValue>> *group_by_keys_;
   ColumnVectorT *output_column_vector_;
 };

 /** @} */

 // ----------------------------------------------------------------------------
 // Implementations of templated methods follow:

 template <typename HandleT,
           typename StateT,
           typename HashTableT>
 void AggregationHandle::aggregateValueAccessorIntoHashTableNullaryHelper(
     ValueAccessor *accessor,
     const std::vector<attribute_id> &group_by_key_ids,
     const StateT &default_state,
     AggregationStateHashTableBase *hash_table) const {
   NullaryAggregationStateValueAccessorUpserter<HandleT, StateT>
       upserter(static_cast<const HandleT&>(*this));
   static_cast<HashTableT*>(hash_table)->upsertValueAccessorCompositeKey(
       accessor,
       group_by_key_ids,
       true,
       default_state,
       &upserter);
 }

 template <typename HandleT,
           typename StateT,
           typename HashTableT>
 void AggregationHandle::aggregateValueAccessorIntoHashTableUnaryHelper(
     ValueAccessor *accessor,
     const attribute_id argument_id,
     const std::vector<attribute_id> &group_by_key_ids,
     const StateT &default_state,
     AggregationStateHashTableBase *hash_table) const {
   UnaryAggregationStateValueAccessorUpserter<HandleT, StateT>
       upserter(static_cast<const HandleT&>(*this), argument_id);
   static_cast<HashTableT*>(hash_table)->upsertValueAccessorCompositeKey(
       accessor,
       group_by_key_ids,
       true,
       default_state,
       &upserter);
 }

 template <typename HandleT,
           typename HashTableT>
 ColumnVector* AggregationHandle::finalizeHashTableHelper(
     const Type &result_type,
     const AggregationStateHashTableBase &hash_table,
     std::vector<std::vector<TypedValue>> *group_by_keys) const {
   const HandleT &handle = static_cast<const HandleT&>(*this);
   const HashTableT &hash_table_concrete = static_cast<const HashTableT&>(hash_table);

   if (group_by_keys->empty()) {
     if (NativeColumnVector::UsableForType(result_type)) {
       NativeColumnVector *result = new NativeColumnVector(result_type,
                                                           hash_table_concrete.numEntries());
       HashTableAggregateFinalizer<HandleT, NativeColumnVector> finalizer(
           handle,
           group_by_keys,
           result);
       hash_table_concrete.forEachCompositeKey(&finalizer);
       return result;
     } else {
       IndirectColumnVector *result = new IndirectColumnVector(result_type,
                                                               hash_table_concrete.numEntries());
       HashTableAggregateFinalizer<HandleT, IndirectColumnVector> finalizer(
           handle,
           group_by_keys,
           result);
       hash_table_concrete.forEachCompositeKey(&finalizer);
       return result;
     }
   } else {
     if (NativeColumnVector::UsableForType(result_type)) {
       NativeColumnVector *result = new NativeColumnVector(result_type,
                                                           group_by_keys->size());
       for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
         result->appendTypedValue(finalizeGroupInHashTable<HandleT, HashTableT>(hash_table,
                                                                                group_by_key));
       }
       return result;
     } else {
       IndirectColumnVector *result = new IndirectColumnVector(result_type,
                                                               hash_table_concrete.numEntries());
       for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
         result->appendTypedValue(finalizeGroupInHashTable<HandleT, HashTableT>(hash_table,
                                                                                group_by_key));
       }
       return result;
     }
   }
 }

 }  // namespace quickstep

 #endif  // QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_HPP_
	/**
	* Copyright 2011-2015 Quickstep Technologies LLC.
	* Copyright 2015 Pivotal Software, 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 QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_HPP_
	#define QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_HPP_

	#include <cstddef>
	#include <memory>
	#include <vector>

	#include "catalog/CatalogTypedefs.hpp"
	#include "storage/HashTableBase.hpp"
	#include "types/TypedValue.hpp"
	#include "types/containers/ColumnVector.hpp"
	#include "utility/Macros.hpp"

	#include "glog/logging.h"

	namespace quickstep {

	class StorageManager;
	class Type;
	class ValueAccessor;

	/** \addtogroup Expressions
	* @{
	*/


	/**
	* @brief Abstract base class for aggregation state.
	**/
	class AggregationState {
	public:
	/**
	* @brief Default constructor.
	**/
	AggregationState() {}

	/**
	* @brief Pure virtual destructor.
	**/
	virtual ~AggregationState() = 0;
	};

	// Destructor should be defined. This will be called when derived class
	// destructor is called.
	inline AggregationState::~AggregationState() {}

	/**
	* @brief AggregationHandle encapsulates logic for actually computing
	* aggregates with particular argument(s).
	* @note See also AggregateFunction, which represents a SQL aggregate function
	* in the abstract sense.
	*
	* An AggregationHandle is created by calling
	* AggregateFunction::createHandle(). The AggregationHandle object provides
	* methods that are used to actually compute the aggregate, storing
	* intermediate results in AggregationState objects.
	*
	* The work-flow for computing an aggregate without GROUP BY is as follows:
	* 1. Create a global state for the aggregate with createInitialState().
	* 2. For each block in a relation (parallelizable):
	* a. Call StorageBlock::aggregate() to accumulate results from the
	* block (under the covers, this calls either
	* accumulateColumnVectors() or accumulateValueAccessor() to do the
	* actual per-block aggregation in a vectorized fashion).
	* b. Merge the per-block results back with the global state by calling
	* mergeStates() (this is threadsafe).
	* 3. Generate the final result by calling finalize() on the global state.
	*
	* The work-flow for computing an aggregate with GROUP BY is as follows:
	* 1. Create a HashTable to hold per-group states by calling
	* createGroupByHashTable().
	* 2. For each block in a relation (parallelizable):
	* a. Call StorageBlock::aggregateGroupBy() to update the states in the
	* HashTable according to the values in the block (under the covers,
	* this calls aggregateValueAccessorIntoHashTable() to aggregate over
	* all the values/groups in a block in one shot; this is threadsafe).
	* 3. Generate the final set of groups and their corresponding results by
	* calling finalizeHashTable().
	*
	* See also AggregationOperationState, which holds 1 or more global states or
	* HashTables for an aggregate query, and has some logic to re-use common
	* information across multiple aggregates being computed on the same block
	* (e.g. the set of matches for a predicate, or the values of computed GROUP BY
	* expressions). AggregationOperationState also has a method to write out
	* finalized aggregate values to an InsertDestination.
	**/
	class AggregationHandle {
	public:
	/**
	* @brief Virtual destructor.
	*
	**/
	virtual ~AggregationHandle() {
	}

	/**
	* @brief Create an initial "blank" state for this aggregation.
	*
	* @return An initial "blank" state for this particular aggregation.
	**/
	virtual AggregationState* createInitialState() const = 0;

	/**
	* @brief Create a new HashTable for aggregation with GROUP BY.
	*
	* @param hash_table_impl The choice of which concrete HashTable
	* implementation to use.
	* @param group_by_types The types of the GROUP BY columns/expressions. These
	* correspond to the (composite) key type for the HashTable.
	* @param estimated_num_groups The estimated number of distinct groups for
	* the GROUP BY aggregation. This is used to size the initial
	* HashTable. This is an estimate only, and the HashTable will be
	* resized if it becomes over-full.
	* @param storage_manager The StorageManager to use to create the HashTable.
	* A StorageBlob will be allocated to serve as the HashTable's
	* in-memory storage.
	* @return A new HashTable instance with the appropriate state type for this
	* aggregate as the ValueT.
	**/
	virtual AggregationStateHashTableBase* createGroupByHashTable(
	const HashTableImplType hash_table_impl,
	const std::vector<const Type*> &group_by_types,
	const std::size_t estimated_num_groups,
	StorageManager *storage_manager) const = 0;

	/**
	* @brief Accumulate over tuples for a nullary aggregate function (one that
	* has zero arguments, i.e. COUNT(*)).
	*
	* @param num_tuples The number of tuples to "accumulate". No actual tuple
	* data is accessed, the only thing that a nullary aggeregate can know
	* about input is its cardinality.
	* @return A new AggregationState which contains the accumulated results from
	* applying the (nullary) aggregate to the specified number of
	* tuples.
	**/
	virtual AggregationState* accumulateNullary(
	const std::size_t num_tuples) const {
	LOG(FATAL) << "Called accumulateNullary on an AggregationHandle that "
	<< "takes at least one argument.";
	}

	/**
	* @brief Accumulate (iterate over) all values in one or more ColumnVectors
	* and return a new AggregationState which can be merged with other
	* states or finalized.
	*
	* @param column_vectors One or more ColumnVectors that the aggregate will be
	* applied to. These correspond to the aggregate function's arguments,
	* in order.
	* @return A new AggregationState which contains the accumulated results from
	* applying the aggregate to column_vectors. Caller is responsible
	* for deleting the returned AggregationState.
	**/
	virtual AggregationState* accumulateColumnVectors(
	const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const = 0;

	#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
	/**
	* @brief Accumulate (iterate over) all values in columns accessible through
	* a ValueAccessor and return a new AggregationState which can be
	* merged with other states or finalized.
	*
	* @param accessor A ValueAccessor that the columns to be aggregated can be
	* accessed through.
	* @param accessor_ids The attribute_ids that correspond to the columns in
	* accessor to aggeregate. These correspond to the aggregate
	* function's arguments, in order.
	* @return A new AggregationState which contains the accumulated results from
	* applying the aggregate to the specified columns in accessor.
	* Caller is responsible for deleting the returned AggregationState.
	**/
	virtual AggregationState* accumulateValueAccessor(
	ValueAccessor *accessor,
	const std::vector<attribute_id> &accessor_ids) const = 0;
	#endif

	/**
	* @brief Perform an aggregation with GROUP BY over all the tuples accessible
	* through a ValueAccessor, upserting states in a HashTable.
	*
	* @note Implementations of this method are threadsafe with respect to
	* hash_table, and can be called concurrently from multiple threads
	* with the same HashTable object.
	*
	* @param accessor The ValueAccessor that will be iterated over to read
	* tuples.
	* @param argument_ids The attribute_ids of the arguments to this aggregate
	* in accessor, in order.
	* @param group_by_key_ids The attribute_ids of the group-by
	* columns/expressions in accessor.
	* @param hash_table The HashTable to upsert AggregationStates in. This
	* should have been created by calling createGroupByHashTable() on
	* this same AggregationHandle.
	**/
	virtual void aggregateValueAccessorIntoHashTable(
	ValueAccessor *accessor,
	const std::vector<attribute_id> &argument_ids,
	const std::vector<attribute_id> &group_by_key_ids,
	AggregationStateHashTableBase *hash_table) const = 0;

	/**
	* @brief Merge two AggregationStates, updating one in-place. This computes a
	* new intermediate result and stores it in the destination
	* AggregationState.
	*
	* @note The order of arguments (i.e. which AggregationState is source and
	* which is destination) is not important for the result of the merging
	* process, but it does determine which of the two states is
	* overwritten (destination) and which remains unchanged (source).
	* @note Implementations of this method are threadsafe with respect to the
	* destination state, and can be called concurrently from multiple
	* threads with the same state object.
	*
	* @param source The AggregationState to merge "from".
	* @param destination The AggregationState to merge "to". The internal state
	* will be overwritten with the merged result.
	**/
	virtual void mergeStates(const AggregationState &source,
	AggregationState *destination) const = 0;

	/**
	* @brief Computes and returns the resulting aggregate by using intermediate
	* result saved in this handle.
	*
	* @note Except for count, SQL89 aggeregates return NULL when no rows are
	* selected.
	* @warning It is dangerous to assume that calling mergeStates() or some
	* iterate method after previous finalize call will work correctly
	* when the aggregate function is not one of SQL89 aggregates (SUM,
	* COUNT, MIN, MAX, AVG).
	*
	* @return The result of this aggregation.
	**/
	virtual TypedValue finalize(const AggregationState &state) const = 0;

	/**
	* @brief Compute and return finalized aggregates for all groups in a
	* HashTable.
	*
	* @param hash_table The HashTable to finalize states from. This should have
	* have been created by calling createGroupByHashTable() on this same
	* AggregationHandle.
	* @param group_by_keys A pointer to a vector of vectors of GROUP BY keys. If
	* this is initially empty, it will be filled in with the GROUP BY
	* keys visited by this method in the same order as the finalized
	* values returned in the ColumnVector. If this is already filled in,
	* then this method will visit the GROUP BY keys in the exact order
	* specified.
	* @return A ColumnVector containing each group's finalized aggregate value.
	**/
	virtual ColumnVector* finalizeHashTable(
	const AggregationStateHashTableBase &hash_table,
	std::vector<std::vector<TypedValue>> *group_by_keys) const = 0;

	protected:
	AggregationHandle() {
	}

	template <typename HandleT,
	typename StateT,
	typename HashTableT>
	void aggregateValueAccessorIntoHashTableNullaryHelper(
	ValueAccessor *accessor,
	const std::vector<attribute_id> &group_by_key_ids,
	const StateT &default_state,
	AggregationStateHashTableBase *hash_table) const;

	template <typename HandleT,
	typename StateT,
	typename HashTableT>
	void aggregateValueAccessorIntoHashTableUnaryHelper(
	ValueAccessor *accessor,
	const attribute_id argument_id,
	const std::vector<attribute_id> &group_by_key_ids,
	const StateT &default_state,
	AggregationStateHashTableBase *hash_table) const;

	template <typename HandleT,
	typename HashTableT>
	ColumnVector* finalizeHashTableHelper(
	const Type &result_type,
	const AggregationStateHashTableBase &hash_table,
	std::vector<std::vector<TypedValue>> *group_by_keys) const;

	template <typename HandleT, typename HashTableT>
	inline TypedValue finalizeGroupInHashTable(
	const AggregationStateHashTableBase &hash_table,
	const std::vector<TypedValue> &group_key) const {
	const AggregationState *group_state
	= static_cast<const HashTableT&>(hash_table).getSingleCompositeKey(group_key);
	DCHECK(group_state != nullptr)
	<< "Could not find entry for specified group_key in HashTable";
	return static_cast<const HandleT>(this)->finalizeHashTableEntry(group_state);
	}

	private:
	DISALLOW_COPY_AND_ASSIGN(AggregationHandle);
	};

	/**
	* @brief Templated class to implement value-accessor-based upserter for each
	* aggregation state payload type. This version is for nullary
	* aggregates (those that take no arguments).
	**/
	template <typename HandleT, typename StateT>
	class NullaryAggregationStateValueAccessorUpserter {
	public:
	explicit NullaryAggregationStateValueAccessorUpserter(const HandleT &handle)
	: handle_(handle) {
	}

	template <typename ValueAccessorT>
	inline void operator()(const ValueAccessorT &accessor, StateT *state) {
	handle_.iterateNullaryInl(state);
	}

	private:
	const HandleT &handle_;
	};

	/**
	* @brief Templated class to implement value-accessor-based upserter for each
	* aggregation state payload type. This version is for unary aggregates
	* (those that take a single argument).
	**/
	template <typename HandleT, typename StateT>
	class UnaryAggregationStateValueAccessorUpserter {
	public:
	UnaryAggregationStateValueAccessorUpserter(const HandleT &handle,
	attribute_id value_id)
	: handle_(handle),
	value_id_(value_id) {
	}

	template <typename ValueAccessorT>
	inline void operator()(const ValueAccessorT &accessor, StateT *state) {
	handle_.iterateUnaryInl(state, accessor.getTypedValue(value_id_));
	}

	private:
	const HandleT &handle_;
	const attribute_id value_id_;
	};

	/**
	* @brief Templated helper class used to implement
	* AggregationHandle::finalizeHashTable() by visiting each entry (i.e.
	* GROUP) in a HashTable, finalizing the aggregation for the GROUP, and
	* collecting the GROUP BY key values and the final aggregate values in
	* a ColumnVector.
	**/
	template <typename HandleT, typename ColumnVectorT>
	class HashTableAggregateFinalizer {
	public:
	HashTableAggregateFinalizer(const HandleT &handle,
	std::vector<std::vector<TypedValue>> *group_by_keys,
	ColumnVectorT *output_column_vector)
	: handle_(handle),
	group_by_keys_(group_by_keys),
	output_column_vector_(output_column_vector) {
	}

	inline void operator()(const std::vector<TypedValue> &group_by_key,
	const AggregationState &group_state) {
	group_by_keys_->emplace_back(group_by_key);
	output_column_vector_->appendTypedValue(handle_.finalizeHashTableEntry(group_state));
	}

	private:
	const HandleT &handle_;
	std::vector<std::vector<TypedValue>> *group_by_keys_;
	ColumnVectorT *output_column_vector_;
	};

	/** @} */

	// ----------------------------------------------------------------------------
	// Implementations of templated methods follow:

	template <typename HandleT,
	typename StateT,
	typename HashTableT>
	void AggregationHandle::aggregateValueAccessorIntoHashTableNullaryHelper(
	ValueAccessor *accessor,
	const std::vector<attribute_id> &group_by_key_ids,
	const StateT &default_state,
	AggregationStateHashTableBase *hash_table) const {
	NullaryAggregationStateValueAccessorUpserter<HandleT, StateT>
	upserter(static_cast<const HandleT&>(*this));
	static_cast<HashTableT*>(hash_table)->upsertValueAccessorCompositeKey(
	accessor,
	group_by_key_ids,
	true,
	default_state,
	&upserter);
	}

	template <typename HandleT,
	typename StateT,
	typename HashTableT>
	void AggregationHandle::aggregateValueAccessorIntoHashTableUnaryHelper(
	ValueAccessor *accessor,
	const attribute_id argument_id,
	const std::vector<attribute_id> &group_by_key_ids,
	const StateT &default_state,
	AggregationStateHashTableBase *hash_table) const {
	UnaryAggregationStateValueAccessorUpserter<HandleT, StateT>
	upserter(static_cast<const HandleT&>(*this), argument_id);
	static_cast<HashTableT*>(hash_table)->upsertValueAccessorCompositeKey(
	accessor,
	group_by_key_ids,
	true,
	default_state,
	&upserter);
	}

	template <typename HandleT,
	typename HashTableT>
	ColumnVector* AggregationHandle::finalizeHashTableHelper(
	const Type &result_type,
	const AggregationStateHashTableBase &hash_table,
	std::vector<std::vector<TypedValue>> *group_by_keys) const {
	const HandleT &handle = static_cast<const HandleT&>(*this);
	const HashTableT &hash_table_concrete = static_cast<const HashTableT&>(hash_table);

	if (group_by_keys->empty()) {
	if (NativeColumnVector::UsableForType(result_type)) {
	NativeColumnVector *result = new NativeColumnVector(result_type,
	hash_table_concrete.numEntries());
	HashTableAggregateFinalizer<HandleT, NativeColumnVector> finalizer(
	handle,
	group_by_keys,
	result);
	hash_table_concrete.forEachCompositeKey(&finalizer);
	return result;
	} else {
	IndirectColumnVector *result = new IndirectColumnVector(result_type,
	hash_table_concrete.numEntries());
	HashTableAggregateFinalizer<HandleT, IndirectColumnVector> finalizer(
	handle,
	group_by_keys,
	result);
	hash_table_concrete.forEachCompositeKey(&finalizer);
	return result;
	}
	} else {
	if (NativeColumnVector::UsableForType(result_type)) {
	NativeColumnVector *result = new NativeColumnVector(result_type,
	group_by_keys->size());
	for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
	result->appendTypedValue(finalizeGroupInHashTable<HandleT, HashTableT>(hash_table,
	group_by_key));
	}
	return result;
	} else {
	IndirectColumnVector *result = new IndirectColumnVector(result_type,
	hash_table_concrete.numEntries());
	for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
	result->appendTypedValue(finalizeGroupInHashTable<HandleT, HashTableT>(hash_table,
	group_by_key));
	}
	return result;
	}
	}
	}

	} // namespace quickstep

	#endif // QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_HPP_