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/01 05:39:55 UTC
incubator-quickstep git commit: Updates
Repository: incubator-quickstep
Updated Branches:
refs/heads/collision-free-agg 963a60428 -> 12b112f85
Updates
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/12b112f8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/12b112f8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/12b112f8
Branch: refs/heads/collision-free-agg
Commit: 12b112f851608c7f423cb9bca780b1be0489e55d
Parents: 963a604
Author: Jianqiao Zhu <ji...@cs.wisc.edu>
Authored: Tue Jan 31 23:39:41 2017 -0600
Committer: Jianqiao Zhu <ji...@cs.wisc.edu>
Committed: Tue Jan 31 23:39:41 2017 -0600
----------------------------------------------------------------------
.../aggregation/AggregationConcreteHandle.cpp | 23 ++++----
.../aggregation/AggregationConcreteHandle.hpp | 14 ++---
expressions/aggregation/AggregationHandle.hpp | 11 ++--
.../aggregation/AggregationHandleSum.cpp | 4 +-
query_optimizer/ExecutionGenerator.cpp | 2 +-
.../InitializeAggregationStateOperator.cpp | 2 +-
storage/AggregationOperationState.cpp | 56 +++++++++++++++++---
storage/AggregationOperationState.hpp | 3 +-
.../CollisionFreeAggregationStateHashTable.cpp | 3 +-
9 files changed, 76 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/expressions/aggregation/AggregationConcreteHandle.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationConcreteHandle.cpp b/expressions/aggregation/AggregationConcreteHandle.cpp
index c3d133a..d28aa6e 100644
--- a/expressions/aggregation/AggregationConcreteHandle.cpp
+++ b/expressions/aggregation/AggregationConcreteHandle.cpp
@@ -48,21 +48,16 @@ AggregationStateHashTableBase* AggregationConcreteHandle::createDistinctifyHashT
}
void AggregationConcreteHandle::insertValueAccessorIntoDistinctifyHashTable(
+ const std::vector<attribute_id> &argument_ids,
+ const std::vector<attribute_id> &key_attr_ids,
+ AggregationStateHashTableBase *distinctify_hash_table,
ValueAccessor *accessor,
- const std::vector<attribute_id> &key_ids,
- 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 */);
-// }
+ ColumnVectorsValueAccessor *aux_accessor) const {
+ std::vector<attribute_id> combined_ids(key_attr_ids);
+ combined_ids.insert(combined_ids.end(), argument_ids.begin(), argument_ids.end());
+
+ static_cast<PackedPayloadSeparateChainingAggregationStateHashTable *>(distinctify_hash_table)
+ ->upsertValueAccessor({}, combined_ids, accessor, aux_accessor);
}
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/expressions/aggregation/AggregationConcreteHandle.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationConcreteHandle.hpp b/expressions/aggregation/AggregationConcreteHandle.hpp
index 04be232..c49f597 100644
--- a/expressions/aggregation/AggregationConcreteHandle.hpp
+++ b/expressions/aggregation/AggregationConcreteHandle.hpp
@@ -110,9 +110,11 @@ class AggregationConcreteHandle : public AggregationHandle {
StorageManager *storage_manager) const override;
void insertValueAccessorIntoDistinctifyHashTable(
+ const std::vector<attribute_id> &argument_ids,
+ const std::vector<attribute_id> &key_attr_ids,
+ AggregationStateHashTableBase *distinctify_hash_table,
ValueAccessor *accessor,
- const std::vector<attribute_id> &key_ids,
- AggregationStateHashTableBase *distinctify_hash_table) const override;
+ ColumnVectorsValueAccessor *aux_accessor = nullptr) const override;
void blockUpdate() override {
block_update_ = true;
@@ -127,11 +129,11 @@ class AggregationConcreteHandle : public AggregationHandle {
: AggregationHandle(agg_id) {}
template <typename HandleT, typename StateT>
- StateT* aggregateOnDistinctifyHashTableForSingleUnaryHelperFast(
+ StateT* aggregateOnDistinctifyHashTableForSingleUnaryHelper(
const AggregationStateHashTableBase &distinctify_hash_table) const;
template <typename HandleT, typename HashTableT>
- void aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast(
+ void aggregateOnDistinctifyHashTableForGroupByUnaryHelper(
const AggregationStateHashTableBase &distinctify_hash_table,
AggregationStateHashTableBase *hash_table,
std::size_t index) const;
@@ -200,7 +202,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());
@@ -226,7 +228,7 @@ StateT* AggregationConcreteHandle::
template <typename HandleT, typename HashTableT>
void AggregationConcreteHandle::
- aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast(
+ aggregateOnDistinctifyHashTableForGroupByUnaryHelper(
const AggregationStateHashTableBase &distinctify_hash_table,
AggregationStateHashTableBase *aggregation_hash_table,
std::size_t index) const {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/expressions/aggregation/AggregationHandle.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandle.hpp b/expressions/aggregation/AggregationHandle.hpp
index bc9c27f..fd7f7af 100644
--- a/expressions/aggregation/AggregationHandle.hpp
+++ b/expressions/aggregation/AggregationHandle.hpp
@@ -249,11 +249,6 @@ class AggregationHandle {
* @brief Inserts the GROUP BY expressions and aggregation arguments together
* 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 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
@@ -261,9 +256,11 @@ class AggregationHandle {
* by calling createDistinctifyHashTable();
*/
virtual void insertValueAccessorIntoDistinctifyHashTable(
+ const std::vector<attribute_id> &argument_ids,
+ const std::vector<attribute_id> &key_attr_ids,
+ AggregationStateHashTableBase *distinctify_hash_table,
ValueAccessor *accessor,
- const std::vector<attribute_id> &key_ids,
- AggregationStateHashTableBase *distinctify_hash_table) const = 0;
+ ColumnVectorsValueAccessor *aux_accessor = nullptr) const = 0;
/**
* @brief Perform single (i.e. without GROUP BY) aggregation on the keys from
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/expressions/aggregation/AggregationHandleSum.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleSum.cpp b/expressions/aggregation/AggregationHandleSum.cpp
index 29a986f..ce36e79 100644
--- a/expressions/aggregation/AggregationHandleSum.cpp
+++ b/expressions/aggregation/AggregationHandleSum.cpp
@@ -152,7 +152,7 @@ ColumnVector* AggregationHandleSum::finalizeHashTable(
AggregationState* AggregationHandleSum::aggregateOnDistinctifyHashTableForSingle(
const AggregationStateHashTableBase &distinctify_hash_table) const {
- return aggregateOnDistinctifyHashTableForSingleUnaryHelperFast<
+ return aggregateOnDistinctifyHashTableForSingleUnaryHelper<
AggregationHandleSum,
AggregationStateSum>(distinctify_hash_table);
}
@@ -161,7 +161,7 @@ void AggregationHandleSum::aggregateOnDistinctifyHashTableForGroupBy(
const AggregationStateHashTableBase &distinctify_hash_table,
AggregationStateHashTableBase *aggregation_hash_table,
std::size_t index) const {
- aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast<
+ aggregateOnDistinctifyHashTableForGroupByUnaryHelper<
AggregationHandleSum,
PackedPayloadSeparateChainingAggregationStateHashTable>(
distinctify_hash_table, aggregation_hash_table, index);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/query_optimizer/ExecutionGenerator.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp
index d32505b..76522dc 100644
--- a/query_optimizer/ExecutionGenerator.cpp
+++ b/query_optimizer/ExecutionGenerator.cpp
@@ -1657,7 +1657,7 @@ void ExecutionGenerator::convertAggregate(
execution_plan_->addDirectDependency(aggregation_operator_index,
initialize_aggregation_state_operator_index,
- true);
+ true /* is_pipeline_breaker */);
}
// Create InsertDestination proto.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/relational_operators/InitializeAggregationStateOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/InitializeAggregationStateOperator.cpp b/relational_operators/InitializeAggregationStateOperator.cpp
index dfee459..b041aef 100644
--- a/relational_operators/InitializeAggregationStateOperator.cpp
+++ b/relational_operators/InitializeAggregationStateOperator.cpp
@@ -53,7 +53,7 @@ bool InitializeAggregationStateOperator::getAllWorkOrders(
}
started_ = true;
}
- return started_;
+ return true;
}
bool InitializeAggregationStateOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/storage/AggregationOperationState.cpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.cpp b/storage/AggregationOperationState.cpp
index 1bc5832..6e7d2ae 100644
--- a/storage/AggregationOperationState.cpp
+++ b/storage/AggregationOperationState.cpp
@@ -136,6 +136,9 @@ AggregationOperationState::AggregationOperationState(
std::vector<std::vector<std::unique_ptr<const Scalar>>>::iterator
args_it = arguments.begin();
std::vector<bool>::const_iterator is_distinct_it = is_distinct_.begin();
+ std::vector<HashTableImplType>::const_iterator
+ distinctify_hash_table_impl_types_it =
+ distinctify_hash_table_impl_types.begin();
for (; agg_func_it != aggregate_functions.end();
++agg_func_it, ++args_it, ++is_distinct_it) {
// Get the Types of this aggregate's arguments so that we can create an
@@ -174,13 +177,35 @@ AggregationOperationState::AggregationOperationState(
// Aggregation with GROUP BY: combined payload is partially updated in
// the presence of DISTINCT.
if (*is_distinct_it) {
- LOG(FATAL) << "Distinct aggregation not supported";
+ handles_.back()->blockUpdate();
}
group_by_handles.emplace_back(handles_.back().get());
} else {
// Aggregation without GROUP BY: create a single global state.
single_states_.emplace_back(handles_.back()->createInitialState());
}
+
+ // Initialize the corresponding distinctify hash table if this is a
+ // DISTINCT aggregation.
+ if (*is_distinct_it) {
+ std::vector<const Type *> key_types(group_by_types_);
+ key_types.insert(
+ key_types.end(), argument_types.begin(), argument_types.end());
+ // TODO(jianqiao): estimated_num_entries is quite inaccurate for
+ // estimating the number of entries in the distinctify hash table.
+ // We need to estimate for each distinct aggregation an
+ // estimated_num_distinct_keys value during query optimization.
+ distinctify_hashtables_.emplace_back(
+ AggregationStateHashTableFactory::CreateResizable(
+ *distinctify_hash_table_impl_types_it,
+ key_types,
+ estimated_num_entries,
+ {},
+ storage_manager));
+ ++distinctify_hash_table_impl_types_it;
+ } else {
+ distinctify_hashtables_.emplace_back(nullptr);
+ }
}
// Aggregation with GROUP BY: create a HashTable pool.
@@ -439,14 +464,23 @@ void AggregationOperationState::aggregateBlockSingleState(
const auto &argument_ids = argument_ids_[agg_idx];
const auto &handle = handles_[agg_idx];
- AggregationState *state;
- if (argument_ids.empty()) {
- // Special case. This is a nullary aggregate (i.e. COUNT(*)).
- state = handle->accumulateNullary(matches == nullptr ? tuple_store.numTuples()
- : matches->size());
+ AggregationState *state = nullptr;
+ if (is_distinct_[agg_idx]) {
+ handle->insertValueAccessorIntoDistinctifyHashTable(
+ argument_ids,
+ {},
+ distinctify_hashtables_[agg_idx].get(),
+ accessor.get(),
+ &non_trivial_results);
} else {
- // Have the AggregationHandle actually do the aggregation.
- state = handle->accumulate(accessor.get(), &non_trivial_results, argument_ids);
+ if (argument_ids.empty()) {
+ // Special case. This is a nullary aggregate (i.e. COUNT(*)).
+ state = handle->accumulateNullary(matches == nullptr ? tuple_store.numTuples()
+ : matches->size());
+ } else {
+ // Have the AggregationHandle actually do the aggregation.
+ state = handle->accumulate(accessor.get(), &non_trivial_results, argument_ids);
+ }
}
local_state.emplace_back(state);
}
@@ -616,6 +650,12 @@ void AggregationOperationState::finalizeSingleState(
std::vector<TypedValue> attribute_values;
for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
+ if (is_distinct_[agg_idx]) {
+ single_states_[agg_idx].reset(
+ handles_[agg_idx]->aggregateOnDistinctifyHashTableForSingle(
+ *distinctify_hashtables_[agg_idx]));
+ }
+
attribute_values.emplace_back(
handles_[agg_idx]->finalize(*single_states_[agg_idx]));
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/storage/AggregationOperationState.hpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.hpp b/storage/AggregationOperationState.hpp
index 44803fc..5ee675a 100644
--- a/storage/AggregationOperationState.hpp
+++ b/storage/AggregationOperationState.hpp
@@ -261,8 +261,7 @@ class AggregationOperationState {
std::vector<const Type *> group_by_types_;
// Hash table for obtaining distinct (i.e. unique) arguments.
-// std::vector<std::unique_ptr<AggregationStateHashTableBase>>
-// distinctify_hashtables_;
+ std::vector<std::unique_ptr<AggregationStateHashTableBase>> distinctify_hashtables_;
// Per-aggregate global states for aggregation without GROUP BY.
std::vector<std::unique_ptr<AggregationState>> single_states_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/12b112f8/storage/CollisionFreeAggregationStateHashTable.cpp
----------------------------------------------------------------------
diff --git a/storage/CollisionFreeAggregationStateHashTable.cpp b/storage/CollisionFreeAggregationStateHashTable.cpp
index 15d4dfe..39560cc 100644
--- a/storage/CollisionFreeAggregationStateHashTable.cpp
+++ b/storage/CollisionFreeAggregationStateHashTable.cpp
@@ -111,7 +111,8 @@ CollisionFreeAggregationStateHashTable::CollisionFreeAggregationStateHashTable(
}
memory_size_ = required_memory;
- num_init_partitions_ = std::min(memory_size_ / (4uL * 1024u * 1024u), 80uL);
+ num_init_partitions_ =
+ std::max(1uL, std::min(memory_size_ / (4uL * 1024u * 1024u), 80uL));
}
CollisionFreeAggregationStateHashTable::~CollisionFreeAggregationStateHashTable() {