You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by ji...@apache.org on 2017/02/07 22:35:31 UTC
[12/14] incubator-quickstep git commit: - Adds
CollisionFreeVectorTable to support specialized fast path aggregation for
range-bounded single integer group-by key. - Supports copy elision for
aggregation.
- Adds CollisionFreeVectorTable to support specialized fast path aggregation for range-bounded single integer group-by key.
- Supports copy elision for aggregation.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/2d89e4fb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/2d89e4fb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/2d89e4fb
Branch: refs/heads/aggregate-on-left-outer-join
Commit: 2d89e4fbf3b51d7768d928d56b25ec0ab52faabb
Parents: ec76096
Author: Jianqiao Zhu <ji...@cs.wisc.edu>
Authored: Mon Jan 30 14:46:39 2017 -0600
Committer: Jianqiao Zhu <ji...@cs.wisc.edu>
Committed: Tue Feb 7 15:29:19 2017 -0600
----------------------------------------------------------------------
.../aggregation/AggregateFunctionCount.cpp | 6 +-
.../aggregation/AggregationConcreteHandle.cpp | 32 +-
.../aggregation/AggregationConcreteHandle.hpp | 160 +-
expressions/aggregation/AggregationHandle.hpp | 193 +-
.../aggregation/AggregationHandleAvg.cpp | 95 +-
.../aggregation/AggregationHandleAvg.hpp | 115 +-
.../aggregation/AggregationHandleCount.cpp | 147 +-
.../aggregation/AggregationHandleCount.hpp | 154 +-
.../aggregation/AggregationHandleDistinct.cpp | 81 -
.../aggregation/AggregationHandleDistinct.hpp | 130 -
.../aggregation/AggregationHandleMax.cpp | 91 +-
.../aggregation/AggregationHandleMax.hpp | 111 +-
.../aggregation/AggregationHandleMin.cpp | 91 +-
.../aggregation/AggregationHandleMin.hpp | 120 +-
.../aggregation/AggregationHandleSum.cpp | 93 +-
.../aggregation/AggregationHandleSum.hpp | 114 +-
expressions/aggregation/CMakeLists.txt | 58 +-
.../tests/AggregationHandleAvg_unittest.cpp | 110 +-
.../tests/AggregationHandleCount_unittest.cpp | 145 +-
.../tests/AggregationHandleMax_unittest.cpp | 158 +-
.../tests/AggregationHandleMin_unittest.cpp | 158 +-
.../tests/AggregationHandleSum_unittest.cpp | 109 +-
query_execution/QueryContext.hpp | 14 -
query_optimizer/CMakeLists.txt | 4 +
query_optimizer/ExecutionGenerator.cpp | 147 +-
query_optimizer/ExecutionGenerator.hpp | 20 +-
relational_operators/CMakeLists.txt | 15 +
.../DestroyAggregationStateOperator.cpp | 7 -
.../FinalizeAggregationOperator.cpp | 16 +-
.../FinalizeAggregationOperator.hpp | 14 +-
.../InitializeAggregationOperator.cpp | 72 +
.../InitializeAggregationOperator.hpp | 122 +
relational_operators/WorkOrderFactory.cpp | 3 +
storage/AggregationOperationState.cpp | 834 +++---
storage/AggregationOperationState.hpp | 178 +-
storage/CMakeLists.txt | 131 +-
storage/CollisionFreeVectorTable.cpp | 285 +++
storage/CollisionFreeVectorTable.hpp | 730 ++++++
storage/FastHashTable.hpp | 2403 ------------------
storage/FastHashTableFactory.hpp | 224 --
storage/FastSeparateChainingHashTable.hpp | 1551 -----------
storage/HashTable.proto | 7 +-
storage/HashTableBase.hpp | 42 +-
storage/HashTableFactory.hpp | 63 +-
storage/HashTablePool.hpp | 79 +-
storage/PackedPayloadHashTable.cpp | 463 ++++
storage/PackedPayloadHashTable.hpp | 995 ++++++++
storage/PartitionedHashTablePool.hpp | 56 +-
storage/StorageBlock.cpp | 274 +-
storage/StorageBlock.hpp | 167 --
storage/ValueAccessorMultiplexer.hpp | 145 ++
.../BarrieredReadWriteConcurrentBitVector.hpp | 282 ++
utility/CMakeLists.txt | 7 +
53 files changed, 4722 insertions(+), 7099 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregateFunctionCount.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregateFunctionCount.cpp b/expressions/aggregation/AggregateFunctionCount.cpp
index 466ff2f..9795b4a 100644
--- a/expressions/aggregation/AggregateFunctionCount.cpp
+++ b/expressions/aggregation/AggregateFunctionCount.cpp
@@ -53,16 +53,16 @@ AggregationHandle* AggregateFunctionCount::createHandle(
if (argument_types.empty()) {
// COUNT(*)
- return new AggregationHandleCount<true, false>();
+ return new AggregationHandleCount<true, false>(nullptr);
} else if (argument_types.front()->isNullable()) {
// COUNT(some_nullable_argument)
- return new AggregationHandleCount<false, true>();
+ return new AggregationHandleCount<false, true>(argument_types.front());
} else {
// COUNT(non_nullable_argument)
//
// TODO(chasseur): Modify query optimizer to optimize-away COUNT with a
// non-nullable argument and convert it to COUNT(*).
- return new AggregationHandleCount<false, false>();
+ return new AggregationHandleCount<false, false>(argument_types.front());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationConcreteHandle.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationConcreteHandle.cpp b/expressions/aggregation/AggregationConcreteHandle.cpp
index e3fb520..fa21056 100644
--- a/expressions/aggregation/AggregationConcreteHandle.cpp
+++ b/expressions/aggregation/AggregationConcreteHandle.cpp
@@ -22,16 +22,14 @@
#include <cstddef>
#include <vector>
-#include "catalog/CatalogTypedefs.hpp"
-#include "storage/FastHashTable.hpp"
-#include "storage/HashTable.hpp"
#include "storage/HashTableFactory.hpp"
+#include "storage/PackedPayloadHashTable.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
namespace quickstep {
class StorageManager;
class Type;
-class ValueAccessor;
AggregationStateHashTableBase* AggregationConcreteHandle::createDistinctifyHashTable(
const HashTableImplType hash_table_impl,
@@ -39,30 +37,26 @@ AggregationStateHashTableBase* AggregationConcreteHandle::createDistinctifyHashT
const std::size_t estimated_num_distinct_keys,
StorageManager *storage_manager) const {
// Create a hash table with key types as key_types and value type as bool.
- return AggregationStateHashTableFactory<bool>::CreateResizable(
+ return AggregationStateHashTableFactory::CreateResizable(
hash_table_impl,
key_types,
estimated_num_distinct_keys,
+ {},
storage_manager);
}
void AggregationConcreteHandle::insertValueAccessorIntoDistinctifyHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &key_ids,
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const std::vector<MultiSourceAttributeId> &key_ids,
+ const ValueAccessorMultiplexer &accessor_mux,
AggregationStateHashTableBase *distinctify_hash_table) const {
- // If the key-value pair is already there, we don't need to update the value,
- // which should always be "true". I.e. the value is just a placeholder.
-
- AggregationStateFastHashTable *hash_table =
- static_cast<AggregationStateFastHashTable *>(distinctify_hash_table);
- if (key_ids.size() == 1) {
- hash_table->upsertValueAccessorFast(
- key_ids, accessor, key_ids[0], true /* check_for_null_keys */);
- } else {
- std::vector<attribute_id> empty_args {kInvalidAttributeID};
- hash_table->upsertValueAccessorCompositeKeyFast(
- empty_args, accessor, key_ids, true /* check_for_null_keys */);
+ std::vector<MultiSourceAttributeId> concatenated_ids(key_ids);
+ for (const MultiSourceAttributeId &arg_id : argument_ids) {
+ concatenated_ids.emplace_back(arg_id);
}
+
+ static_cast<PackedPayloadHashTable *>(distinctify_hash_table)
+ ->upsertValueAccessorCompositeKey({}, concatenated_ids, accessor_mux);
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationConcreteHandle.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationConcreteHandle.hpp b/expressions/aggregation/AggregationConcreteHandle.hpp
index 398a032..2287ca1 100644
--- a/expressions/aggregation/AggregationConcreteHandle.hpp
+++ b/expressions/aggregation/AggregationConcreteHandle.hpp
@@ -21,14 +21,15 @@
#define QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_CONCRETE_HANDLE_HPP_
#include <cstddef>
+#include <cstdint>
#include <utility>
#include <vector>
-#include "catalog/CatalogTypedefs.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
-#include "storage/FastHashTable.hpp"
-#include "storage/HashTable.hpp"
+#include "expressions/aggregation/AggregationID.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/PackedPayloadHashTable.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
#include "threading/SpinMutex.hpp"
#include "types/TypedValue.hpp"
#include "types/containers/ColumnVector.hpp"
@@ -40,7 +41,6 @@ namespace quickstep {
class StorageManager;
class Type;
-class ValueAccessor;
/** \addtogroup Expressions
* @{
@@ -51,7 +51,7 @@ class ValueAccessor;
* merging two group by hash tables.
**/
template <typename HandleT>
-class HashTableStateUpserterFast {
+class HashTableStateUpserter {
public:
/**
* @brief Constructor.
@@ -61,8 +61,8 @@ class HashTableStateUpserterFast {
* table. The corresponding state (for the same key) in the destination
* hash table will be upserted.
**/
- HashTableStateUpserterFast(const HandleT &handle,
- const std::uint8_t *source_state)
+ HashTableStateUpserter(const HandleT &handle,
+ const std::uint8_t *source_state)
: handle_(handle), source_state_(source_state) {}
/**
@@ -72,14 +72,14 @@ class HashTableStateUpserterFast {
* table that is being upserted.
**/
void operator()(std::uint8_t *destination_state) {
- handle_.mergeStatesFast(source_state_, destination_state);
+ handle_.mergeStates(source_state_, destination_state);
}
private:
const HandleT &handle_;
const std::uint8_t *source_state_;
- DISALLOW_COPY_AND_ASSIGN(HashTableStateUpserterFast);
+ DISALLOW_COPY_AND_ASSIGN(HashTableStateUpserter);
};
/**
@@ -93,74 +93,62 @@ class HashTableStateUpserterFast {
**/
class AggregationConcreteHandle : public AggregationHandle {
public:
- /**
- * @brief Default implementaion for AggregationHandle::accumulateNullary().
- */
- AggregationState* accumulateNullary(
- const std::size_t num_tuples) const override {
- LOG(FATAL) << "Called accumulateNullary on an AggregationHandle that "
- << "takes at least one argument.";
- }
-
- /**
- * @brief Implementaion for AggregationHandle::createDistinctifyHashTable()
- * that creates a new HashTable for the distinctify step for
- * DISTINCT aggregation.
- */
AggregationStateHashTableBase* createDistinctifyHashTable(
const HashTableImplType hash_table_impl,
const std::vector<const Type *> &key_types,
const std::size_t estimated_num_distinct_keys,
StorageManager *storage_manager) const override;
- /**
- * @brief Implementaion for
- * AggregationHandle::insertValueAccessorIntoDistinctifyHashTable()
- * that inserts the GROUP BY expressions and aggregation arguments together
- * as keys into the distinctify hash table.
- */
void insertValueAccessorIntoDistinctifyHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &key_ids,
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const std::vector<MultiSourceAttributeId> &key_ids,
+ const ValueAccessorMultiplexer &accessor_mux,
AggregationStateHashTableBase *distinctify_hash_table) const override;
+ void blockUpdate() override {
+ block_update_ = true;
+ }
+
+ void allowUpdate() override {
+ block_update_ = false;
+ }
+
protected:
- AggregationConcreteHandle() {}
+ explicit AggregationConcreteHandle(const AggregationID agg_id)
+ : AggregationHandle(agg_id),
+ block_update_(false) {}
template <typename HandleT, typename StateT>
- StateT* aggregateOnDistinctifyHashTableForSingleUnaryHelperFast(
+ StateT* aggregateOnDistinctifyHashTableForSingleUnaryHelper(
const AggregationStateHashTableBase &distinctify_hash_table) const;
- template <typename HandleT, typename HashTableT>
- void aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast(
+ template <typename HandleT>
+ void aggregateOnDistinctifyHashTableForGroupByUnaryHelper(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *hash_table,
- std::size_t index) const;
+ const std::size_t index,
+ AggregationStateHashTableBase *hash_table) const;
- template <typename HandleT, typename HashTableT>
- ColumnVector* finalizeHashTableHelperFast(
+ template <typename HandleT>
+ ColumnVector* finalizeHashTableHelper(
const Type &result_type,
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const;
+ const std::size_t index,
+ std::vector<std::vector<TypedValue>> *group_by_keys) const;
- template <typename HandleT, typename HashTableT>
- inline TypedValue finalizeGroupInHashTableFast(
+ template <typename HandleT>
+ inline TypedValue finalizeGroupInHashTable(
const AggregationStateHashTableBase &hash_table,
- const std::vector<TypedValue> &group_key,
- int index) const {
+ const std::size_t index,
+ const std::vector<TypedValue> &group_key) const {
const std::uint8_t *group_state =
- static_cast<const HashTableT &>(hash_table).getSingleCompositeKey(group_key, index);
+ static_cast<const PackedPayloadHashTable &>(hash_table)
+ .getSingleCompositeKey(group_key, index);
DCHECK(group_state != nullptr)
<< "Could not find entry for specified group_key in HashTable";
- return static_cast<const HandleT *>(this)->finalizeHashTableEntryFast(
- group_state);
+ return static_cast<const HandleT *>(this)->finalizeHashTableEntry(group_state);
}
- template <typename HandleT, typename HashTableT>
- void mergeGroupByHashTablesHelperFast(
- const AggregationStateHashTableBase &source_hash_table,
- AggregationStateHashTableBase *destination_hash_table) const;
+ bool block_update_;
private:
DISALLOW_COPY_AND_ASSIGN(AggregationConcreteHandle);
@@ -185,17 +173,10 @@ class HashTableAggregateFinalizer {
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));
- }
-
- inline void operator()(const std::vector<TypedValue> &group_by_key,
- const unsigned char *byte_ptr) {
+ const std::uint8_t *byte_ptr) {
group_by_keys_->emplace_back(group_by_key);
output_column_vector_->appendTypedValue(
- handle_.finalizeHashTableEntryFast(byte_ptr));
+ handle_.finalizeHashTableEntry(byte_ptr));
}
private:
@@ -211,7 +192,7 @@ class HashTableAggregateFinalizer {
template <typename HandleT, typename StateT>
StateT* AggregationConcreteHandle::
- aggregateOnDistinctifyHashTableForSingleUnaryHelperFast(
+ aggregateOnDistinctifyHashTableForSingleUnaryHelper(
const AggregationStateHashTableBase &distinctify_hash_table) const {
const HandleT &handle = static_cast<const HandleT &>(*this);
StateT *state = static_cast<StateT *>(createInitialState());
@@ -219,15 +200,14 @@ StateT* AggregationConcreteHandle::
// A lambda function which will be called on each key from the distinctify
// hash table.
const auto aggregate_functor = [&handle, &state](
- const TypedValue &key, const std::uint8_t &dumb_placeholder) {
+ const TypedValue &key, const std::uint8_t *dumb_placeholder) {
// For each (unary) key in the distinctify hash table, aggregate the key
// into "state".
handle.iterateUnaryInl(state, key);
};
- const AggregationStateFastHashTable &hash_table =
- static_cast<const AggregationStateFastHashTable &>(
- distinctify_hash_table);
+ const auto &hash_table =
+ static_cast<const PackedPayloadHashTable &>(distinctify_hash_table);
// Invoke the lambda function "aggregate_functor" on each key from the
// distinctify hash table.
hash_table.forEach(&aggregate_functor);
@@ -235,20 +215,20 @@ StateT* AggregationConcreteHandle::
return state;
}
-template <typename HandleT, typename HashTableT>
+template <typename HandleT>
void AggregationConcreteHandle::
- aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast(
+ aggregateOnDistinctifyHashTableForGroupByUnaryHelper(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *aggregation_hash_table,
- std::size_t index) const {
+ const std::size_t index,
+ AggregationStateHashTableBase *aggregation_hash_table) const {
const HandleT &handle = static_cast<const HandleT &>(*this);
- HashTableT *target_hash_table =
- static_cast<HashTableT *>(aggregation_hash_table);
+ PackedPayloadHashTable *target_hash_table =
+ static_cast<PackedPayloadHashTable *>(aggregation_hash_table);
// A lambda function which will be called on each key-value pair from the
// distinctify hash table.
const auto aggregate_functor = [&handle, &target_hash_table, &index](
- std::vector<TypedValue> &key, const bool &dumb_placeholder) {
+ std::vector<TypedValue> &key, const std::uint8_t *dumb_placeholder) {
// For each (composite) key vector in the distinctify hash table with size N.
// The first N-1 entries are GROUP BY columns and the last entry is the
// argument to be aggregated on.
@@ -258,28 +238,28 @@ void AggregationConcreteHandle::
// An upserter as lambda function for aggregating the argument into its
// GROUP BY group's entry inside aggregation_hash_table.
const auto upserter = [&handle, &argument](std::uint8_t *state) {
- handle.iterateUnaryInlFast(argument, state);
+ handle.iterateUnaryInl(argument, state);
};
- target_hash_table->upsertCompositeKeyFast(key, nullptr, &upserter, index);
+ target_hash_table->upsertCompositeKey(key, &upserter, index);
};
- const HashTableT &source_hash_table =
- static_cast<const HashTableT &>(distinctify_hash_table);
+ const PackedPayloadHashTable &source_hash_table =
+ static_cast<const PackedPayloadHashTable &>(distinctify_hash_table);
// Invoke the lambda function "aggregate_functor" on each composite key vector
// from the distinctify hash table.
- source_hash_table.forEachCompositeKeyFast(&aggregate_functor);
+ source_hash_table.forEachCompositeKey(&aggregate_functor);
}
-template <typename HandleT, typename HashTableT>
-ColumnVector* AggregationConcreteHandle::finalizeHashTableHelperFast(
+template <typename HandleT>
+ColumnVector* AggregationConcreteHandle::finalizeHashTableHelper(
const Type &result_type,
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const {
+ const std::size_t index,
+ 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);
+ const PackedPayloadHashTable &hash_table_concrete =
+ static_cast<const PackedPayloadHashTable &>(hash_table);
if (group_by_keys->empty()) {
if (NativeColumnVector::UsableForType(result_type)) {
@@ -287,14 +267,14 @@ ColumnVector* AggregationConcreteHandle::finalizeHashTableHelperFast(
new NativeColumnVector(result_type, hash_table_concrete.numEntries());
HashTableAggregateFinalizer<HandleT, NativeColumnVector> finalizer(
handle, group_by_keys, result);
- hash_table_concrete.forEachCompositeKeyFast(&finalizer, index);
+ hash_table_concrete.forEachCompositeKey(&finalizer, index);
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.forEachCompositeKeyFast(&finalizer, index);
+ hash_table_concrete.forEachCompositeKey(&finalizer, index);
return result;
}
} else {
@@ -303,8 +283,8 @@ ColumnVector* AggregationConcreteHandle::finalizeHashTableHelperFast(
new NativeColumnVector(result_type, group_by_keys->size());
for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
result->appendTypedValue(
- finalizeGroupInHashTableFast<HandleT, HashTableT>(
- hash_table, group_by_key, index));
+ finalizeGroupInHashTable<HandleT>(
+ hash_table, index, group_by_key));
}
return result;
} else {
@@ -312,8 +292,8 @@ ColumnVector* AggregationConcreteHandle::finalizeHashTableHelperFast(
result_type, hash_table_concrete.numEntries());
for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
result->appendTypedValue(
- finalizeGroupInHashTableFast<HandleT, HashTableT>(
- hash_table, group_by_key, index));
+ finalizeGroupInHashTable<HandleT>(
+ hash_table, index, group_by_key));
}
return result;
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandle.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandle.hpp b/expressions/aggregation/AggregationHandle.hpp
index 4b51179..9c7b166 100644
--- a/expressions/aggregation/AggregationHandle.hpp
+++ b/expressions/aggregation/AggregationHandle.hpp
@@ -21,20 +21,21 @@
#define QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_HPP_
#include <cstddef>
-#include <memory>
#include <vector>
-#include "catalog/CatalogTypedefs.hpp"
+#include "expressions/aggregation/AggregationID.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
#include "types/TypedValue.hpp"
#include "utility/Macros.hpp"
+#include "glog/logging.h"
+
namespace quickstep {
class ColumnVector;
class StorageManager;
class Type;
-class ValueAccessor;
/** \addtogroup Expressions
* @{
@@ -109,34 +110,34 @@ class AggregationHandle {
virtual ~AggregationHandle() {}
/**
- * @brief Create an initial "blank" state for this aggregation.
+ * @brief Get the ID of this aggregation.
*
- * @return An initial "blank" state for this particular aggregation.
+ * @return The AggregationID of this AggregationHandle.
**/
- virtual AggregationState* createInitialState() const = 0;
+ inline AggregationID getAggregationID() const {
+ return agg_id_;
+ }
/**
- * @brief Create a new HashTable for aggregation with GROUP BY.
+ * @brief Get the list of Types (in order) for arguments to this aggregation.
*
- * @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.
+ * @return The list of Types for arguments to this aggregation.
+ */
+ virtual std::vector<const Type *> getArgumentTypes() const = 0;
+
+ /**
+ * @brief Get the result Type of this aggregation.
+ *
+ * @return The result Type of this aggregation.
+ */
+ virtual const Type* getResultType() const = 0;
+
+ /**
+ * @brief Create an initial "blank" state for this aggregation.
+ *
+ * @return An initial "blank" state for this particular aggregation.
**/
- 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;
+ virtual AggregationState* createInitialState() const = 0;
/**
* @brief Accumulate over tuples for a nullary aggregate function (one that
@@ -146,70 +147,31 @@ class AggregationHandle {
* 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.
+ * applying the (nullary) aggregate to the specified number of tuples.
**/
virtual AggregationState* accumulateNullary(
- const std::size_t num_tuples) const = 0;
-
- /**
- * @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;
+ const std::size_t num_tuples) const {
+ LOG(FATAL) << "Called accumulateNullary on an AggregationHandle that "
+ << "takes at least one argument.";
+ }
-#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.
+ * the ValueAccessors from a ValueAccessorMultiplexer 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.
+ * @param argument_ids The multi-source attribute ids that correspond to the
+ * columns in \p accessor_mux to aggeregate. These correspond to the
+ * aggregate function's arguments, in order.
+ * @param accessor_mux A ValueAccessorMultiplexer object that contains the
+ * ValueAccessors.
* @return A new AggregationState which contains the accumulated results from
- * applying the aggregate to the specified columns in accessor.
+ * applying the aggregate to the specified columns.
* 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;
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const ValueAccessorMultiplexer &accessor_mux) const = 0;
/**
* @brief Merge two AggregationStates, updating one in-place. This computes a
@@ -253,24 +215,24 @@ class AggregationHandle {
* @param hash_table The HashTable to finalize states from. This should have
* have been created by calling createGroupByHashTable() on this same
* AggregationHandle.
+ * @param index The index of the AggregationHandle to be finalized.
* @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.
- * @param index The index of the AggregationHandle to be finalized.
*
* @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,
- int index) const = 0;
+ const std::size_t index,
+ std::vector<std::vector<TypedValue>> *group_by_keys) const = 0;
/**
* @brief Create a new HashTable for the distinctify step for DISTINCT
- * aggregation.
+ * aggregation.
*
* Distinctify is the first step for DISTINCT aggregation. This step inserts
* the GROUP BY expression values and aggregation arguments together as keys
@@ -283,8 +245,8 @@ class AggregationHandle {
* we simply treat it as a special GROUP BY case that the GROUP BY expression
* vector is empty.
*
- * @param hash_table_impl The choice of which concrete HashTable
- * implementation to use.
+ * @param hash_table_impl The choice of which concrete HashTable implementation
+ * to use.
* @param key_types The types of the GROUP BY expressions together with the
* types of the aggregation arguments.
* @param estimated_num_distinct_keys The estimated number of distinct keys
@@ -307,13 +269,14 @@ class AggregationHandle {
/**
* @brief Inserts the GROUP BY expressions and aggregation arguments together
- * as keys into the distinctify hash table.
+ * as keys into the distinctify hash table.
*
- * @param accessor The ValueAccessor that will be iterated over to read
- * tuples.
- * @param key_ids The attribute_ids of the GROUP BY expressions in accessor
- * together with the attribute_ids of the arguments to this aggregate
- * in accessor, in order.
+ * @param argument_ids The argument ids that correspond to the columns in
+ * \p accessor_mux.
+ * @param key_ids The group-by key ids that correspond to the columns in
+ * \p accessor_mux.
+ * @param accessor_mux A ValueAccessorMultiplexer object that contains the
+ * ValueAccessors to be iterated over to read tuples.
* @param distinctify_hash_table The HashTable to store the GROUP BY
* expressions and the aggregation arguments together as hash table
* keys and a bool constant \c true as hash table value (So the hash
@@ -321,13 +284,14 @@ class AggregationHandle {
* by calling createDistinctifyHashTable();
*/
virtual void insertValueAccessorIntoDistinctifyHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &key_ids,
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const std::vector<MultiSourceAttributeId> &key_ids,
+ const ValueAccessorMultiplexer &accessor_mux,
AggregationStateHashTableBase *distinctify_hash_table) const = 0;
/**
* @brief Perform single (i.e. without GROUP BY) aggregation on the keys from
- * the distinctify hash table to actually compute the aggregated results.
+ * the distinctify hash table to actually compute the aggregated results.
*
* @param distinctify_hash_table Hash table which stores the distinctified
* aggregation arguments as hash table keys. This should have been
@@ -346,25 +310,26 @@ class AggregationHandle {
* @param distinctify_hash_table Hash table which stores the GROUP BY
* expression values and aggregation arguments together as hash table
* keys.
+ * @param index The index of the AggregationHandle to perform aggregation.
* @param aggregation_hash_table The HashTable to upsert AggregationStates in.
* This should have been created by calling createGroupByHashTable() on
* this same AggregationHandle.
- * @param index The index of the distinctify hash table for which we perform
- * the DISTINCT aggregation.
*/
virtual void aggregateOnDistinctifyHashTableForGroupBy(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *aggregation_hash_table,
- std::size_t index) const = 0;
+ const std::size_t index,
+ AggregationStateHashTableBase *aggregation_hash_table) const = 0;
/**
* @brief Get the number of bytes needed to store the aggregation handle's
* state.
**/
- virtual std::size_t getPayloadSize() const { return 1; }
+ virtual std::size_t getPayloadSize() const {
+ return 1u;
+ }
/**
- * @brief Update the aggregation state for nullary aggregation function e.g.
+ * @brief Update the aggregation state for nullary aggregation function, e.g.
* COUNT(*).
*
* @note This function should be overloaded by those aggregation function
@@ -372,7 +337,10 @@ class AggregationHandle {
*
* @param byte_ptr The pointer where the aggregation state is stored.
**/
- virtual void updateStateNullary(std::uint8_t *byte_ptr) const {}
+ virtual void updateStateNullary(std::uint8_t *byte_ptr) const {
+ LOG(FATAL) << "Called updateStateNullary on an AggregationHandle that "
+ << "takes at least one argument.";
+ }
/**
* @brief Update the aggregation state for unary aggregation function e.g.
@@ -383,7 +351,7 @@ class AggregationHandle {
* @param byte_ptr The pointer where the aggregation state is stored.
**/
virtual void updateStateUnary(const TypedValue &argument,
- std::uint8_t *byte_ptr) const {}
+ std::uint8_t *byte_ptr) const = 0;
/**
* @brief Merge two aggregation states for this aggregation handle.
@@ -394,8 +362,8 @@ class AggregationHandle {
* @param src A pointer to the source aggregation state.
* @param dst A pointer to the destination aggregation state.
**/
- virtual void mergeStatesFast(const std::uint8_t *src,
- std::uint8_t *dst) const {}
+ virtual void mergeStates(const std::uint8_t *src,
+ std::uint8_t *dst) const = 0;
/**
* @brief Initialize the payload (in the aggregation hash table) for the given
@@ -403,7 +371,7 @@ class AggregationHandle {
*
* @param byte_ptr The pointer to the aggregation state in the hash table.
**/
- virtual void initPayload(std::uint8_t *byte_ptr) const {}
+ virtual void initPayload(std::uint8_t *byte_ptr) const = 0;
/**
* @brief Destroy the payload (in the aggregation hash table) for the given
@@ -411,22 +379,25 @@ class AggregationHandle {
*
* @param byte_ptr The pointer to the aggregation state in the hash table.
**/
- virtual void destroyPayload(std::uint8_t *byte_ptr) const {}
+ virtual void destroyPayload(std::uint8_t *byte_ptr) const = 0;
/**
* @brief Inform the aggregation handle to block (prohibit) updates on the
* aggregation state.
**/
- virtual void blockUpdate() {}
+ virtual void blockUpdate() = 0;
/**
- * @brief Inform the aggregation handle to allow updates on the
- * aggregation state.
+ * @brief Inform the aggregation handle to allow updates on the aggregation
+ * state.
**/
- virtual void allowUpdate() {}
+ virtual void allowUpdate() = 0;
protected:
- AggregationHandle() {}
+ explicit AggregationHandle(const AggregationID agg_id)
+ : agg_id_(agg_id) {}
+
+ const AggregationID agg_id_;
private:
DISALLOW_COPY_AND_ASSIGN(AggregationHandle);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandleAvg.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleAvg.cpp b/expressions/aggregation/AggregationHandleAvg.cpp
index 2481092..46bec1e 100644
--- a/expressions/aggregation/AggregationHandleAvg.cpp
+++ b/expressions/aggregation/AggregationHandleAvg.cpp
@@ -20,12 +20,13 @@
#include "expressions/aggregation/AggregationHandleAvg.hpp"
#include <cstddef>
+#include <cstdint>
#include <memory>
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
-#include "storage/HashTable.hpp"
-#include "storage/HashTableFactory.hpp"
+#include "expressions/aggregation/AggregationID.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
#include "threading/SpinMutex.hpp"
#include "types/Type.hpp"
#include "types/TypeFactory.hpp"
@@ -39,10 +40,11 @@
namespace quickstep {
-class StorageManager;
+class ColumnVector;
AggregationHandleAvg::AggregationHandleAvg(const Type &type)
- : argument_type_(type), block_update_(false) {
+ : AggregationConcreteHandle(AggregationID::kAvg),
+ argument_type_(type) {
// We sum Int as Long and Float as Double so that we have more headroom when
// adding many values.
TypeID type_precision_id;
@@ -87,52 +89,29 @@ AggregationHandleAvg::AggregationHandleAvg(const Type &type)
->getNullableVersion());
}
-AggregationStateHashTableBase* AggregationHandleAvg::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 {
- return AggregationStateHashTableFactory<AggregationStateAvg>::CreateResizable(
- hash_table_impl, group_by_types, estimated_num_groups, storage_manager);
-}
+AggregationState* AggregationHandleAvg::accumulateValueAccessor(
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const ValueAccessorMultiplexer &accessor_mux) const {
+ DCHECK_EQ(1u, argument_ids.size())
+ << "Got wrong number of attributes for AVG: " << argument_ids.size();
-AggregationState* AggregationHandleAvg::accumulateColumnVectors(
- const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
- DCHECK_EQ(1u, column_vectors.size())
- << "Got wrong number of ColumnVectors for AVG: " << column_vectors.size();
+ const ValueAccessorSource argument_source = argument_ids.front().source;
+ const attribute_id argument_id = argument_ids.front().attr_id;
- AggregationStateAvg *state = new AggregationStateAvg(blank_state_);
- std::size_t count = 0;
- state->sum_ = fast_add_operator_->accumulateColumnVector(
- state->sum_, *column_vectors.front(), &count);
- state->count_ = count;
- return state;
-}
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-AggregationState* AggregationHandleAvg::accumulateValueAccessor(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &accessor_ids) const {
- DCHECK_EQ(1u, accessor_ids.size())
- << "Got wrong number of attributes for AVG: " << accessor_ids.size();
+ DCHECK(argument_source != ValueAccessorSource::kInvalid);
+ DCHECK_NE(argument_id, kInvalidAttributeID);
AggregationStateAvg *state = new AggregationStateAvg(blank_state_);
std::size_t count = 0;
- state->sum_ = fast_add_operator_->accumulateValueAccessor(
- state->sum_, accessor, accessor_ids.front(), &count);
+ state->sum_ =
+ fast_add_operator_->accumulateValueAccessor(
+ state->sum_,
+ accessor_mux.getValueAccessorBySource(argument_source),
+ argument_id,
+ &count);
state->count_ = count;
return state;
}
-#endif
-
-void AggregationHandleAvg::aggregateValueAccessorIntoHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &argument_ids,
- const std::vector<attribute_id> &group_by_key_ids,
- AggregationStateHashTableBase *hash_table) const {
- DCHECK_EQ(1u, argument_ids.size())
- << "Got wrong number of arguments for AVG: " << argument_ids.size();
-}
void AggregationHandleAvg::mergeStates(const AggregationState &source,
AggregationState *destination) const {
@@ -147,8 +126,8 @@ void AggregationHandleAvg::mergeStates(const AggregationState &source,
avg_destination->sum_, avg_source.sum_);
}
-void AggregationHandleAvg::mergeStatesFast(const std::uint8_t *source,
- std::uint8_t *destination) const {
+void AggregationHandleAvg::mergeStates(const std::uint8_t *source,
+ std::uint8_t *destination) const {
const TypedValue *src_sum_ptr =
reinterpret_cast<const TypedValue *>(source + blank_state_.sum_offset_);
const std::int64_t *src_count_ptr = reinterpret_cast<const std::int64_t *>(
@@ -177,29 +156,25 @@ TypedValue AggregationHandleAvg::finalize(const AggregationState &state) const {
ColumnVector* AggregationHandleAvg::finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const {
- return finalizeHashTableHelperFast<AggregationHandleAvg,
- AggregationStateFastHashTable>(
- *result_type_, hash_table, group_by_keys, index);
+ const std::size_t index,
+ std::vector<std::vector<TypedValue>> *group_by_keys) const {
+ return finalizeHashTableHelper<AggregationHandleAvg>(
+ *result_type_, hash_table, index, group_by_keys);
}
-AggregationState*
-AggregationHandleAvg::aggregateOnDistinctifyHashTableForSingle(
+AggregationState* AggregationHandleAvg::aggregateOnDistinctifyHashTableForSingle(
const AggregationStateHashTableBase &distinctify_hash_table) const {
- return aggregateOnDistinctifyHashTableForSingleUnaryHelperFast<
- AggregationHandleAvg,
- AggregationStateAvg>(distinctify_hash_table);
+ return aggregateOnDistinctifyHashTableForSingleUnaryHelper<
+ AggregationHandleAvg, AggregationStateAvg>(
+ distinctify_hash_table);
}
void AggregationHandleAvg::aggregateOnDistinctifyHashTableForGroupBy(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *aggregation_hash_table,
- std::size_t index) const {
- aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast<
- AggregationHandleAvg,
- AggregationStateFastHashTable>(
- distinctify_hash_table, aggregation_hash_table, index);
+ const std::size_t index,
+ AggregationStateHashTableBase *aggregation_hash_table) const {
+ aggregateOnDistinctifyHashTableForGroupByUnaryHelper<AggregationHandleAvg>(
+ distinctify_hash_table, index, aggregation_hash_table);
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandleAvg.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleAvg.hpp b/expressions/aggregation/AggregationHandleAvg.hpp
index 47132c6..970561c 100644
--- a/expressions/aggregation/AggregationHandleAvg.hpp
+++ b/expressions/aggregation/AggregationHandleAvg.hpp
@@ -25,11 +25,9 @@
#include <memory>
#include <vector>
-#include "catalog/CatalogTypedefs.hpp"
#include "expressions/aggregation/AggregationConcreteHandle.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
-#include "storage/FastHashTable.hpp"
-#include "storage/HashTableBase.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
#include "threading/SpinMutex.hpp"
#include "types/Type.hpp"
#include "types/TypedValue.hpp"
@@ -40,9 +38,8 @@
namespace quickstep {
+class AggregationStateHashTableBase;
class ColumnVector;
-class StorageManager;
-class ValueAccessor;
/** \addtogroup Expressions
* @{
@@ -106,19 +103,18 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
public:
~AggregationHandleAvg() override {}
+ std::vector<const Type *> getArgumentTypes() const override {
+ return {&argument_type_};
+ }
+
+ const Type* getResultType() const override {
+ return result_type_;
+ }
+
AggregationState* createInitialState() const override {
return new AggregationStateAvg(blank_state_);
}
- 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 override;
-
- /**
- * @brief Iterate method with average aggregation state.
- **/
inline void iterateUnaryInl(AggregationStateAvg *state,
const TypedValue &value) const {
DCHECK(value.isPlausibleInstanceOf(argument_type_.getSignature()));
@@ -129,8 +125,8 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
++state->count_;
}
- inline void iterateUnaryInlFast(const TypedValue &value,
- std::uint8_t *byte_ptr) const {
+ inline void iterateUnaryInl(const TypedValue &value,
+ std::uint8_t *byte_ptr) const {
DCHECK(value.isPlausibleInstanceOf(argument_type_.getSignature()));
if (value.isNull()) return;
TypedValue *sum_ptr =
@@ -141,16 +137,18 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
++(*count_ptr);
}
- inline void updateStateUnary(const TypedValue &argument,
- std::uint8_t *byte_ptr) const override {
- if (!block_update_) {
- iterateUnaryInlFast(argument, byte_ptr);
- }
- }
+ AggregationState* accumulateValueAccessor(
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const ValueAccessorMultiplexer &accessor_mux) const override;
- void blockUpdate() override { block_update_ = true; }
+ void mergeStates(const AggregationState &source,
+ AggregationState *destination) const override;
- void allowUpdate() override { block_update_ = false; }
+ TypedValue finalize(const AggregationState &state) const override;
+
+ std::size_t getPayloadSize() const override {
+ return blank_state_.getPayloadSize();
+ }
void initPayload(std::uint8_t *byte_ptr) const override {
TypedValue *sum_ptr =
@@ -169,43 +167,17 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
}
}
- AggregationState* accumulateColumnVectors(
- const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
- const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
- AggregationState* accumulateValueAccessor(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &accessor_id) const override;
-#endif
-
- void aggregateValueAccessorIntoHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &argument_ids,
- const std::vector<attribute_id> &group_by_key_ids,
- AggregationStateHashTableBase *hash_table) const override;
-
- void mergeStates(const AggregationState &source,
- AggregationState *destination) const override;
-
- void mergeStatesFast(const std::uint8_t *source,
- std::uint8_t *destination) const override;
+ inline void updateStateUnary(const TypedValue &argument,
+ std::uint8_t *byte_ptr) const override {
+ if (!block_update_) {
+ iterateUnaryInl(argument, byte_ptr);
+ }
+ }
- TypedValue finalize(const AggregationState &state) const override;
+ void mergeStates(const std::uint8_t *source,
+ std::uint8_t *destination) const override;
inline TypedValue finalizeHashTableEntry(
- const AggregationState &state) const {
- const AggregationStateAvg &agg_state =
- static_cast<const AggregationStateAvg &>(state);
- // TODO(chasseur): Could improve performance further if we made a special
- // version of finalizeHashTable() that collects all the sums into one
- // ColumnVector and all the counts into another and then applies
- // '*divide_operator_' to them in bulk.
- return divide_operator_->applyToTypedValues(
- agg_state.sum_, TypedValue(static_cast<double>(agg_state.count_)));
- }
-
- inline TypedValue finalizeHashTableEntryFast(
const std::uint8_t *byte_ptr) const {
std::uint8_t *value_ptr = const_cast<std::uint8_t *>(byte_ptr);
TypedValue *sum_ptr =
@@ -218,31 +190,16 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const override;
+ const std::size_t index,
+ std::vector<std::vector<TypedValue>> *group_by_keys) const override;
- /**
- * @brief Implementation of
- * AggregationHandle::aggregateOnDistinctifyHashTableForSingle()
- * for AVG aggregation.
- */
AggregationState* aggregateOnDistinctifyHashTableForSingle(
- const AggregationStateHashTableBase &distinctify_hash_table)
- const override;
+ const AggregationStateHashTableBase &distinctify_hash_table) const override;
- /**
- * @brief Implementation of
- * AggregationHandle::aggregateOnDistinctifyHashTableForGroupBy()
- * for AVG aggregation.
- */
void aggregateOnDistinctifyHashTableForGroupBy(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *aggregation_hash_table,
- std::size_t index) const override;
-
- std::size_t getPayloadSize() const override {
- return blank_state_.getPayloadSize();
- }
+ const std::size_t index,
+ AggregationStateHashTableBase *aggregation_hash_table) const override;
private:
friend class AggregateFunctionAvg;
@@ -261,8 +218,6 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
std::unique_ptr<UncheckedBinaryOperator> merge_add_operator_;
std::unique_ptr<UncheckedBinaryOperator> divide_operator_;
- bool block_update_;
-
DISALLOW_COPY_AND_ASSIGN(AggregationHandleAvg);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandleCount.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleCount.cpp b/expressions/aggregation/AggregationHandleCount.cpp
index 034c942..cf92ec7 100644
--- a/expressions/aggregation/AggregationHandleCount.cpp
+++ b/expressions/aggregation/AggregationHandleCount.cpp
@@ -21,100 +21,48 @@
#include <atomic>
#include <cstddef>
-#include <memory>
+#include <cstdint>
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
-#include "storage/HashTable.hpp"
-#include "storage/HashTableFactory.hpp"
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
#include "storage/ValueAccessor.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
#include "storage/ValueAccessorUtil.hpp"
-#endif
-
#include "types/TypeFactory.hpp"
#include "types/TypeID.hpp"
#include "types/TypedValue.hpp"
-#include "types/containers/ColumnVector.hpp"
-#include "types/containers/ColumnVectorUtil.hpp"
#include "glog/logging.h"
namespace quickstep {
-class StorageManager;
-class Type;
-class ValueAccessor;
+class ColumnVector;
template <bool count_star, bool nullable_type>
-AggregationStateHashTableBase*
-AggregationHandleCount<count_star, nullable_type>::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 {
- return AggregationStateHashTableFactory<
- AggregationStateCount>::CreateResizable(hash_table_impl,
- group_by_types,
- estimated_num_groups,
- storage_manager);
-}
-
-template <bool count_star, bool nullable_type>
-AggregationState*
-AggregationHandleCount<count_star, nullable_type>::accumulateColumnVectors(
- const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
+AggregationState* AggregationHandleCount<count_star, nullable_type>
+ ::accumulateValueAccessor(
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const ValueAccessorMultiplexer &accessor_mux) const {
DCHECK(!count_star)
<< "Called non-nullary accumulation method on an AggregationHandleCount "
<< "set up for nullary COUNT(*)";
- DCHECK_EQ(1u, column_vectors.size())
- << "Got wrong number of ColumnVectors for COUNT: "
- << column_vectors.size();
+ DCHECK_EQ(1u, argument_ids.size())
+ << "Got wrong number of attributes for COUNT: " << argument_ids.size();
- std::size_t count = 0;
- InvokeOnColumnVector(
- *column_vectors.front(),
- [&](const auto &column_vector) -> void { // NOLINT(build/c++11)
- if (nullable_type) {
- // TODO(shoban): Iterating over the ColumnVector is a rather slow way
- // to do this. We should look at extending the ColumnVector interface
- // to do a quick count of the non-null values (i.e. the length minus
- // the population count of the null bitmap). We should do something
- // similar for ValueAccessor too.
- for (std::size_t pos = 0; pos < column_vector.size(); ++pos) {
- count += !column_vector.getTypedValue(pos).isNull();
- }
- } else {
- count = column_vector.size();
- }
- });
+ const ValueAccessorSource argument_source = argument_ids.front().source;
+ const attribute_id argument_id = argument_ids.front().attr_id;
- return new AggregationStateCount(count);
-}
+ DCHECK(argument_source != ValueAccessorSource::kInvalid);
+ DCHECK_NE(argument_id, kInvalidAttributeID);
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-template <bool count_star, bool nullable_type>
-AggregationState*
-AggregationHandleCount<count_star, nullable_type>::accumulateValueAccessor(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &accessor_ids) const {
- DCHECK(!count_star)
- << "Called non-nullary accumulation method on an AggregationHandleCount "
- << "set up for nullary COUNT(*)";
-
- DCHECK_EQ(1u, accessor_ids.size())
- << "Got wrong number of attributes for COUNT: " << accessor_ids.size();
-
- const attribute_id accessor_id = accessor_ids.front();
std::size_t count = 0;
InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
- accessor,
- [&accessor_id, &count](auto *accessor) -> void { // NOLINT(build/c++11)
+ accessor_mux.getValueAccessorBySource(argument_source),
+ [&argument_id, &count](auto *accessor) -> void { // NOLINT(build/c++11)
if (nullable_type) {
while (accessor->next()) {
- count += !accessor->getTypedValue(accessor_id).isNull();
+ count += !accessor->getTypedValue(argument_id).isNull();
}
} else {
count = accessor->getNumTuples();
@@ -123,24 +71,6 @@ AggregationHandleCount<count_star, nullable_type>::accumulateValueAccessor(
return new AggregationStateCount(count);
}
-#endif
-
-template <bool count_star, bool nullable_type>
-void AggregationHandleCount<count_star, nullable_type>::
- aggregateValueAccessorIntoHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &argument_ids,
- const std::vector<attribute_id> &group_by_key_ids,
- AggregationStateHashTableBase *hash_table) const {
- if (count_star) {
- DCHECK_EQ(0u, argument_ids.size())
- << "Got wrong number of arguments for COUNT(*): "
- << argument_ids.size();
- } else {
- DCHECK_EQ(1u, argument_ids.size())
- << "Got wrong number of arguments for COUNT: " << argument_ids.size();
- }
-}
template <bool count_star, bool nullable_type>
void AggregationHandleCount<count_star, nullable_type>::mergeStates(
@@ -156,7 +86,7 @@ void AggregationHandleCount<count_star, nullable_type>::mergeStates(
}
template <bool count_star, bool nullable_type>
-void AggregationHandleCount<count_star, nullable_type>::mergeStatesFast(
+void AggregationHandleCount<count_star, nullable_type>::mergeStates(
const std::uint8_t *source, std::uint8_t *destination) const {
const std::int64_t *src_count_ptr =
reinterpret_cast<const std::int64_t *>(source);
@@ -165,38 +95,35 @@ void AggregationHandleCount<count_star, nullable_type>::mergeStatesFast(
}
template <bool count_star, bool nullable_type>
-ColumnVector*
-AggregationHandleCount<count_star, nullable_type>::finalizeHashTable(
- const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const {
- return finalizeHashTableHelperFast<
- AggregationHandleCount<count_star, nullable_type>,
- AggregationStateFastHashTable>(
- TypeFactory::GetType(kLong), hash_table, group_by_keys, index);
+ColumnVector* AggregationHandleCount<count_star, nullable_type>
+ ::finalizeHashTable(
+ const AggregationStateHashTableBase &hash_table,
+ const std::size_t index,
+ std::vector<std::vector<TypedValue>> *group_by_keys) const {
+ return finalizeHashTableHelper<
+ AggregationHandleCount<count_star, nullable_type>>(
+ TypeFactory::GetType(kLong), hash_table, index, group_by_keys);
}
template <bool count_star, bool nullable_type>
-AggregationState* AggregationHandleCount<count_star, nullable_type>::
- aggregateOnDistinctifyHashTableForSingle(
+AggregationState* AggregationHandleCount<count_star, nullable_type>
+ ::aggregateOnDistinctifyHashTableForSingle(
const AggregationStateHashTableBase &distinctify_hash_table) const {
- DCHECK_EQ(count_star, false);
- return aggregateOnDistinctifyHashTableForSingleUnaryHelperFast<
+ return aggregateOnDistinctifyHashTableForSingleUnaryHelper<
AggregationHandleCount<count_star, nullable_type>,
- AggregationStateCount>(distinctify_hash_table);
+ AggregationStateCount>(
+ distinctify_hash_table);
}
template <bool count_star, bool nullable_type>
-void AggregationHandleCount<count_star, nullable_type>::
- aggregateOnDistinctifyHashTableForGroupBy(
+void AggregationHandleCount<count_star, nullable_type>
+ ::aggregateOnDistinctifyHashTableForGroupBy(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *aggregation_hash_table,
- std::size_t index) const {
- DCHECK_EQ(count_star, false);
- aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast<
- AggregationHandleCount<count_star, nullable_type>,
- AggregationStateFastHashTable>(
- distinctify_hash_table, aggregation_hash_table, index);
+ const std::size_t index,
+ AggregationStateHashTableBase *aggregation_hash_table) const {
+ aggregateOnDistinctifyHashTableForGroupByUnaryHelper<
+ AggregationHandleCount<count_star, nullable_type>>(
+ distinctify_hash_table, index, aggregation_hash_table);
}
// Explicitly instantiate and compile in the different versions of
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandleCount.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleCount.hpp b/expressions/aggregation/AggregationHandleCount.hpp
index 6aab0cd..72ea923 100644
--- a/expressions/aggregation/AggregationHandleCount.hpp
+++ b/expressions/aggregation/AggregationHandleCount.hpp
@@ -23,23 +23,21 @@
#include <atomic>
#include <cstddef>
#include <cstdint>
-#include <memory>
#include <vector>
-#include "catalog/CatalogTypedefs.hpp"
#include "expressions/aggregation/AggregationConcreteHandle.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
-#include "storage/FastHashTable.hpp"
-#include "storage/HashTableBase.hpp"
+#include "expressions/aggregation/AggregationID.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
+#include "types/LongType.hpp"
#include "types/TypedValue.hpp"
#include "utility/Macros.hpp"
namespace quickstep {
+class AggregationStateHashTableBase;
class ColumnVector;
-class StorageManager;
class Type;
-class ValueAccessor;
template <bool, bool>
class AggregationHandleCount;
@@ -98,28 +96,31 @@ class AggregationHandleCount : public AggregationConcreteHandle {
public:
~AggregationHandleCount() override {}
+ std::vector<const Type *> getArgumentTypes() const override {
+ if (argument_type_ == nullptr) {
+ return {};
+ } else {
+ return {argument_type_};
+ }
+ }
+
+ const Type* getResultType() const override {
+ return &LongType::InstanceNonNullable();
+ }
+
AggregationState* createInitialState() const override {
return new AggregationStateCount();
}
- 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 override;
-
inline void iterateNullaryInl(AggregationStateCount *state) const {
state->count_.fetch_add(1, std::memory_order_relaxed);
}
- inline void iterateNullaryInlFast(std::uint8_t *byte_ptr) const {
+ inline void iterateNullaryInl(std::uint8_t *byte_ptr) const {
std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(byte_ptr);
- (*count_ptr)++;
+ ++(*count_ptr);
}
- /**
- * @brief Iterate with count aggregation state.
- */
inline void iterateUnaryInl(AggregationStateCount *state,
const TypedValue &value) const {
if ((!nullable_type) || (!value.isNull())) {
@@ -127,118 +128,89 @@ class AggregationHandleCount : public AggregationConcreteHandle {
}
}
- inline void iterateUnaryInlFast(const TypedValue &value,
- std::uint8_t *byte_ptr) const {
+ inline void iterateUnaryInl(const TypedValue &value,
+ std::uint8_t *byte_ptr) const {
if ((!nullable_type) || (!value.isNull())) {
- std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(byte_ptr);
- (*count_ptr)++;
- }
- }
-
- inline void updateStateUnary(const TypedValue &argument,
- std::uint8_t *byte_ptr) const override {
- if (!block_update_) {
- iterateUnaryInlFast(argument, byte_ptr);
+ ++(*reinterpret_cast<std::int64_t *>(byte_ptr));
}
}
- inline void updateStateNullary(std::uint8_t *byte_ptr) const override {
- if (!block_update_) {
- iterateNullaryInlFast(byte_ptr);
- }
- }
-
- void blockUpdate() override { block_update_ = true; }
-
- void allowUpdate() override { block_update_ = false; }
-
- void initPayload(std::uint8_t *byte_ptr) const override {
- std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(byte_ptr);
- *count_ptr = 0;
- }
-
AggregationState* accumulateNullary(
const std::size_t num_tuples) const override {
return new AggregationStateCount(num_tuples);
}
- AggregationState* accumulateColumnVectors(
- const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
- const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
AggregationState* accumulateValueAccessor(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &accessor_id) const override;
-#endif
-
- void aggregateValueAccessorIntoHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &argument_ids,
- const std::vector<attribute_id> &group_by_key_ids,
- AggregationStateHashTableBase *hash_table) const override;
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const ValueAccessorMultiplexer &accessor_mux) const override;
void mergeStates(const AggregationState &source,
AggregationState *destination) const override;
- void mergeStatesFast(const std::uint8_t *source,
- std::uint8_t *destination) const override;
-
TypedValue finalize(const AggregationState &state) const override {
return TypedValue(
static_cast<const AggregationStateCount &>(state).count_.load(
std::memory_order_relaxed));
}
- inline TypedValue finalizeHashTableEntry(
- const AggregationState &state) const {
- return TypedValue(
- static_cast<const AggregationStateCount &>(state).count_.load(
- std::memory_order_relaxed));
+ std::size_t getPayloadSize() const override {
+ return sizeof(std::int64_t);
+ }
+
+ void initPayload(std::uint8_t *byte_ptr) const override {
+ std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(byte_ptr);
+ *count_ptr = 0;
}
- inline TypedValue finalizeHashTableEntryFast(
- const std::uint8_t *byte_ptr) const {
- const std::int64_t *count_ptr =
- reinterpret_cast<const std::int64_t *>(byte_ptr);
- return TypedValue(*count_ptr);
+ void destroyPayload(std::uint8_t *byte_ptr) const override {}
+
+ inline void updateStateNullary(std::uint8_t *byte_ptr) const override {
+ if (!block_update_) {
+ iterateNullaryInl(byte_ptr);
+ }
+ }
+
+ inline void updateStateUnary(const TypedValue &argument,
+ std::uint8_t *byte_ptr) const override {
+ if (!block_update_) {
+ iterateUnaryInl(argument, byte_ptr);
+ }
+ }
+
+ void mergeStates(const std::uint8_t *source,
+ std::uint8_t *destination) const override;
+
+ inline TypedValue finalizeHashTableEntry(const std::uint8_t *byte_ptr) const {
+ return TypedValue(*reinterpret_cast<const std::int64_t *>(byte_ptr));
}
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const override;
+ const std::size_t index,
+ std::vector<std::vector<TypedValue>> *group_by_keys) const override;
- /**
- * @brief Implementation of
- * AggregationHandle::aggregateOnDistinctifyHashTableForSingle()
- * for SUM aggregation.
- */
AggregationState* aggregateOnDistinctifyHashTableForSingle(
- const AggregationStateHashTableBase &distinctify_hash_table)
- const override;
+ const AggregationStateHashTableBase &distinctify_hash_table) const override;
- /**
- * @brief Implementation of
- * AggregationHandle::aggregateOnDistinctifyHashTableForGroupBy()
- * for SUM aggregation.
- */
void aggregateOnDistinctifyHashTableForGroupBy(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *aggregation_hash_table,
- std::size_t index) const override;
-
- std::size_t getPayloadSize() const override { return sizeof(std::int64_t); }
+ const std::size_t index,
+ AggregationStateHashTableBase *aggregation_hash_table) const override;
private:
friend class AggregateFunctionCount;
/**
- * @brief Constructor.
+ * @brief Initialize handle for count.
+ *
+ * @param argument_type Type of the value to be counted. The parameter should
+ * be nullptr for nullary aggregation (i.e. COUNT(*)).
**/
- AggregationHandleCount() : block_update_(false) {}
+ explicit AggregationHandleCount(const Type *argument_type)
+ : AggregationConcreteHandle(AggregationID::kCount),
+ argument_type_(argument_type) {}
- bool block_update_;
+ const Type *argument_type_;
DISALLOW_COPY_AND_ASSIGN(AggregationHandleCount);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandleDistinct.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleDistinct.cpp b/expressions/aggregation/AggregationHandleDistinct.cpp
deleted file mode 100644
index 0dc8b56..0000000
--- a/expressions/aggregation/AggregationHandleDistinct.cpp
+++ /dev/null
@@ -1,81 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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.
- **/
-
-#include "expressions/aggregation/AggregationHandleDistinct.hpp"
-
-#include <cstddef>
-#include <memory>
-#include <vector>
-#include <utility>
-
-#include "catalog/CatalogTypedefs.hpp"
-#include "storage/HashTable.hpp"
-
-#include "types/TypedValue.hpp"
-
-#include "glog/logging.h"
-
-namespace quickstep {
-
-class ColumnVector;
-class StorageManager;
-class Type;
-class ValueAccessor;
-
-AggregationStateHashTableBase* AggregationHandleDistinct::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 {
- return createDistinctifyHashTable(
- hash_table_impl,
- group_by_types,
- estimated_num_groups,
- storage_manager);
-}
-
-void AggregationHandleDistinct::aggregateValueAccessorIntoHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &argument_ids,
- const std::vector<attribute_id> &group_by_key_ids,
- AggregationStateHashTableBase *hash_table) const {
- DCHECK_EQ(argument_ids.size(), 0u);
-
- insertValueAccessorIntoDistinctifyHashTable(
- accessor,
- group_by_key_ids,
- hash_table);
-}
-
-ColumnVector* AggregationHandleDistinct::finalizeHashTable(
- const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const {
- DCHECK(group_by_keys->empty());
-
- const auto keys_retriever = [&group_by_keys](std::vector<TypedValue> &group_by_key,
- const bool &dumb_placeholder) -> void {
- group_by_keys->emplace_back(std::move(group_by_key));
- };
- static_cast<const AggregationStateFastHashTable&>(hash_table).forEachCompositeKeyFast(&keys_retriever);
-
- return nullptr;
-}
-
-} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandleDistinct.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleDistinct.hpp b/expressions/aggregation/AggregationHandleDistinct.hpp
deleted file mode 100644
index 838bfdd..0000000
--- a/expressions/aggregation/AggregationHandleDistinct.hpp
+++ /dev/null
@@ -1,130 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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_DISTINCT_HPP_
-#define QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_DISTINCT_HPP_
-
-#include <cstddef>
-#include <memory>
-#include <vector>
-
-#include "catalog/CatalogTypedefs.hpp"
-#include "expressions/aggregation/AggregationConcreteHandle.hpp"
-#include "storage/HashTableBase.hpp"
-#include "types/TypedValue.hpp"
-#include "utility/Macros.hpp"
-
-#include "glog/logging.h"
-
-namespace quickstep {
-
-class AggregationState;
-class ColumnVector;
-class StorageManager;
-class Type;
-class ValueAccessor;
-
-/** \addtogroup Expressions
- * @{
- */
-
-class AggregationHandleDistinct : public AggregationConcreteHandle {
- public:
- /**
- * @brief Constructor.
- **/
- AggregationHandleDistinct() {}
-
- AggregationState* createInitialState() const override {
- LOG(FATAL)
- << "AggregationHandleDistinct does not support createInitialState().";
- }
-
- AggregationState* accumulateNullary(
- const std::size_t num_tuples) const override {
- LOG(FATAL)
- << "AggregationHandleDistinct does not support accumulateNullary().";
- }
-
- AggregationState* accumulateColumnVectors(
- const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
- const override {
- LOG(FATAL) << "AggregationHandleDistinct does not support "
- "accumulateColumnVectors().";
- }
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
- AggregationState* accumulateValueAccessor(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &accessor_ids) const override {
- LOG(FATAL) << "AggregationHandleDistinct does not support "
- "accumulateValueAccessor().";
- }
-#endif
-
- void mergeStates(const AggregationState &source,
- AggregationState *destination) const override {
- LOG(FATAL) << "AggregationHandleDistinct does not support mergeStates().";
- }
-
- TypedValue finalize(const AggregationState &state) const override {
- LOG(FATAL) << "AggregationHandleDistinct does not support finalize().";
- }
-
- AggregationState* aggregateOnDistinctifyHashTableForSingle(
- const AggregationStateHashTableBase &distinctify_hash_table)
- const override {
- LOG(FATAL) << "AggregationHandleDistinct does not support "
- << "aggregateOnDistinctifyHashTableForSingle().";
- }
-
- void aggregateOnDistinctifyHashTableForGroupBy(
- const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *groupby_hash_table,
- std::size_t index) const override {
- LOG(FATAL) << "AggregationHandleDistinct does not support "
- << "aggregateOnDistinctifyHashTableForGroupBy().";
- }
-
- 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 override;
-
- void aggregateValueAccessorIntoHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &argument_ids,
- const std::vector<attribute_id> &group_by_key_ids,
- AggregationStateHashTableBase *hash_table) const override;
-
- ColumnVector* finalizeHashTable(
- const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const override;
-
- private:
- DISALLOW_COPY_AND_ASSIGN(AggregationHandleDistinct);
-};
-
-/** @} */
-
-} // namespace quickstep
-
-#endif // QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_HANDLE_DISTINCT_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2d89e4fb/expressions/aggregation/AggregationHandleMax.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMax.cpp b/expressions/aggregation/AggregationHandleMax.cpp
index c2d571b..fe1773f 100644
--- a/expressions/aggregation/AggregationHandleMax.cpp
+++ b/expressions/aggregation/AggregationHandleMax.cpp
@@ -19,15 +19,16 @@
#include "expressions/aggregation/AggregationHandleMax.hpp"
+#include <cstddef>
+#include <cstdint>
#include <memory>
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
-#include "storage/HashTable.hpp"
-#include "storage/HashTableFactory.hpp"
+#include "expressions/aggregation/AggregationID.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
#include "types/Type.hpp"
#include "types/TypedValue.hpp"
-#include "types/containers/ColumnVector.hpp"
#include "types/operations/comparisons/Comparison.hpp"
#include "types/operations/comparisons/ComparisonFactory.hpp"
#include "types/operations/comparisons/ComparisonID.hpp"
@@ -36,54 +37,32 @@
namespace quickstep {
-class StorageManager;
+class ColumnVector;
AggregationHandleMax::AggregationHandleMax(const Type &type)
- : type_(type), block_update_(false) {
+ : AggregationConcreteHandle(AggregationID::kMax),
+ type_(type) {
fast_comparator_.reset(
ComparisonFactory::GetComparison(ComparisonID::kGreater)
.makeUncheckedComparatorForTypes(type, type.getNonNullableVersion()));
}
-AggregationStateHashTableBase* AggregationHandleMax::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 {
- return AggregationStateHashTableFactory<AggregationStateMax>::CreateResizable(
- hash_table_impl, group_by_types, estimated_num_groups, storage_manager);
-}
+AggregationState* AggregationHandleMax::accumulateValueAccessor(
+ const std::vector<MultiSourceAttributeId> &argument_ids,
+ const ValueAccessorMultiplexer &accessor_mux) const {
+ DCHECK_EQ(1u, argument_ids.size())
+ << "Got wrong number of attributes for MAX: " << argument_ids.size();
-AggregationState* AggregationHandleMax::accumulateColumnVectors(
- const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
- DCHECK_EQ(1u, column_vectors.size())
- << "Got wrong number of ColumnVectors for MAX: " << column_vectors.size();
+ const ValueAccessorSource argument_source = argument_ids.front().source;
+ const attribute_id argument_id = argument_ids.front().attr_id;
- return new AggregationStateMax(fast_comparator_->accumulateColumnVector(
- type_.getNullableVersion().makeNullValue(), *column_vectors.front()));
-}
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-AggregationState* AggregationHandleMax::accumulateValueAccessor(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &accessor_ids) const {
- DCHECK_EQ(1u, accessor_ids.size())
- << "Got wrong number of attributes for MAX: " << accessor_ids.size();
+ DCHECK(argument_source != ValueAccessorSource::kInvalid);
+ DCHECK_NE(argument_id, kInvalidAttributeID);
return new AggregationStateMax(fast_comparator_->accumulateValueAccessor(
type_.getNullableVersion().makeNullValue(),
- accessor,
- accessor_ids.front()));
-}
-#endif // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-
-void AggregationHandleMax::aggregateValueAccessorIntoHashTable(
- ValueAccessor *accessor,
- const std::vector<attribute_id> &argument_ids,
- const std::vector<attribute_id> &group_by_key_ids,
- AggregationStateHashTableBase *hash_table) const {
- DCHECK_EQ(1u, argument_ids.size())
- << "Got wrong number of arguments for MAX: " << argument_ids.size();
+ accessor_mux.getValueAccessorBySource(argument_source),
+ argument_id));
}
void AggregationHandleMax::mergeStates(const AggregationState &source,
@@ -98,40 +77,36 @@ void AggregationHandleMax::mergeStates(const AggregationState &source,
}
}
-void AggregationHandleMax::mergeStatesFast(const std::uint8_t *source,
- std::uint8_t *destination) const {
+void AggregationHandleMax::mergeStates(const std::uint8_t *source,
+ std::uint8_t *destination) const {
const TypedValue *src_max_ptr = reinterpret_cast<const TypedValue *>(source);
TypedValue *dst_max_ptr = reinterpret_cast<TypedValue *>(destination);
if (!(src_max_ptr->isNull())) {
- compareAndUpdateFast(dst_max_ptr, *src_max_ptr);
+ compareAndUpdate(dst_max_ptr, *src_max_ptr);
}
}
ColumnVector* AggregationHandleMax::finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys,
- int index) const {
- return finalizeHashTableHelperFast<AggregationHandleMax,
- AggregationStateFastHashTable>(
- type_.getNullableVersion(), hash_table, group_by_keys, index);
+ const std::size_t index,
+ std::vector<std::vector<TypedValue>> *group_by_keys) const {
+ return finalizeHashTableHelper<AggregationHandleMax>(
+ type_, hash_table, index, group_by_keys);
}
-AggregationState*
-AggregationHandleMax::aggregateOnDistinctifyHashTableForSingle(
+AggregationState* AggregationHandleMax::aggregateOnDistinctifyHashTableForSingle(
const AggregationStateHashTableBase &distinctify_hash_table) const {
- return aggregateOnDistinctifyHashTableForSingleUnaryHelperFast<
- AggregationHandleMax,
- AggregationStateMax>(distinctify_hash_table);
+ return aggregateOnDistinctifyHashTableForSingleUnaryHelper<
+ AggregationHandleMax, AggregationStateMax>(
+ distinctify_hash_table);
}
void AggregationHandleMax::aggregateOnDistinctifyHashTableForGroupBy(
const AggregationStateHashTableBase &distinctify_hash_table,
- AggregationStateHashTableBase *aggregation_hash_table,
- std::size_t index) const {
- aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast<
- AggregationHandleMax,
- AggregationStateFastHashTable>(
- distinctify_hash_table, aggregation_hash_table, index);
+ const std::size_t index,
+ AggregationStateHashTableBase *aggregation_hash_table) const {
+ aggregateOnDistinctifyHashTableForGroupByUnaryHelper<AggregationHandleMax>(
+ distinctify_hash_table, index, aggregation_hash_table);
}
} // namespace quickstep