Updates
diff --git a/expressions/aggregation/AggFunc.hpp b/expressions/aggregation/AggFunc.hpp
index 31f385e..98b19ba 100644
--- a/expressions/aggregation/AggFunc.hpp
+++ b/expressions/aggregation/AggFunc.hpp
@@ -69,6 +69,11 @@
typename AggState<ArgType>::AtomicT> {};
template <typename ArgType>
+ inline static void InitAtomic(typename AggState<ArgType>::AtomicT *state) {
+ state->store(0, std::memory_order_relaxed);
+ }
+
+ template <typename ArgType>
inline static void MergeArgAtomic(const typename ArgType::cpptype &value,
typename AggState<ArgType>::AtomicT *state) {
LOG(FATAL) << "Not implemented";
@@ -81,6 +86,11 @@
}
template <typename ArgType>
+ inline static void InitUnsafe(typename AggState<ArgType>::T *state) {
+ *state = 0;
+ }
+
+ template <typename ArgType>
inline static void MergeArgUnsafe(const typename ArgType::cpptype &value,
typename AggState<ArgType>::T *state) {
*state += value;
diff --git a/storage/AggregationOperationState.cpp b/storage/AggregationOperationState.cpp
index 00bb433..392e0f6 100644
--- a/storage/AggregationOperationState.cpp
+++ b/storage/AggregationOperationState.cpp
@@ -212,7 +212,7 @@
AggregationStateHashTableFactory::CreateResizable(
hash_table_impl_type,
group_by_types_,
- estimated_num_entries,
+ estimated_num_entries * 2,
group_by_handles,
storage_manager_));
} else {
@@ -384,12 +384,12 @@
}
}
- // There are GROUP BYs without DISTINCT. Check if the estimated number of
- // groups is large enough to warrant a partitioned aggregation.
- return estimated_num_groups >
- static_cast<std::size_t>(
- FLAGS_partition_aggregation_num_groups_threshold);
- return false;
+// // There are GROUP BYs without DISTINCT. Check if the estimated number of
+// // groups is large enough to warrant a partitioned aggregation.
+// return estimated_num_groups >
+// static_cast<std::size_t>(
+// FLAGS_partition_aggregation_num_groups_threshold);
+ return true;
}
std::size_t AggregationOperationState::getNumInitializationPartitions() const {
@@ -663,60 +663,10 @@
InsertDestination *output_destination) {
PackedPayloadHashTable *hash_table =
static_cast<PackedPayloadHashTable *>(partitioned_hashtable_.get());
+// std::cout << hash_table->numEntries() << "\n";
- // Each element of 'group_by_keys' is a vector of values for a particular
- // group (which is also the prefix of the finalized Tuple for that group).
- std::vector<std::vector<TypedValue>> group_by_keys;
-
- if (handles_.empty()) {
- hash_table->forEachCompositeKeyInPartition(
- partition_id,
- [&](std::vector<TypedValue> &group_by_key) -> void {
- group_by_keys.emplace_back(std::move(group_by_key));
- });
- }
-
- // Collect per-aggregate finalized values.
- std::vector<std::unique_ptr<ColumnVector>> final_values;
- for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
- ColumnVector *agg_result_col = handles_[agg_idx]->finalizeHashTable(
- *hash_table, agg_idx, &group_by_keys);
- if (agg_result_col != nullptr) {
- final_values.emplace_back(agg_result_col);
- }
- }
-// hash_table->destroyPayload();
-
- std::vector<std::unique_ptr<ColumnVector>> group_by_cvs;
- std::size_t group_by_element_idx = 0;
- for (const Type *group_by_type : group_by_types_) {
- if (NativeColumnVector::UsableForType(*group_by_type)) {
- NativeColumnVector *element_cv =
- new NativeColumnVector(*group_by_type, group_by_keys.size());
- group_by_cvs.emplace_back(element_cv);
- for (std::vector<TypedValue> &group_key : group_by_keys) {
- element_cv->appendTypedValue(std::move(group_key[group_by_element_idx]));
- }
- } else {
- IndirectColumnVector *element_cv =
- new IndirectColumnVector(*group_by_type, group_by_keys.size());
- group_by_cvs.emplace_back(element_cv);
- for (std::vector<TypedValue> &group_key : group_by_keys) {
- element_cv->appendTypedValue(std::move(group_key[group_by_element_idx]));
- }
- }
- ++group_by_element_idx;
- }
-
- // Stitch together a ColumnVectorsValueAccessor combining the GROUP BY keys
- // and the finalized aggregates.
ColumnVectorsValueAccessor complete_result;
- for (std::unique_ptr<ColumnVector> &group_by_cv : group_by_cvs) {
- complete_result.addColumn(group_by_cv.release());
- }
- for (std::unique_ptr<ColumnVector> &final_value_cv : final_values) {
- complete_result.addColumn(final_value_cv.release());
- }
+ hash_table->finalize(partition_id, &complete_result);
// Bulk-insert the complete result.
output_destination->bulkInsertTuples(&complete_result);
diff --git a/storage/AggregationUtil.hpp b/storage/AggregationUtil.hpp
new file mode 100644
index 0000000..74a9095
--- /dev/null
+++ b/storage/AggregationUtil.hpp
@@ -0,0 +1,197 @@
+/**
+ * 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_STORAGE_AGGREGATION_UTIL_HPP_
+#define QUICKSTEP_STORAGE_AGGREGATION_UTIL_HPP_
+
+#include <type_traits>
+
+#include "expressions/aggregation/AggregationID.hpp"
+#include "expressions/aggregation/AggFunc.hpp"
+#include "storage/ValueAccessor.hpp"
+#include "storage/ValueAccessorMultiplexer.hpp"
+#include "storage/ValueAccessorUtil.hpp"
+#include "types/TypeID.hpp"
+#include "types/Type.hpp"
+#include "types/containers/ColumnVectorsValueAccessor.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+/** \addtogroup Storage
+ * @{
+ */
+
+template <typename T>
+using remove_const_reference_t = std::remove_const_t<std::remove_reference_t<T>>;
+
+template <typename FunctorT>
+inline auto InvokeOnKeyType(const Type &type,
+ const FunctorT &functor) {
+ switch (type.getTypeID()) {
+ case TypeID::kInt:
+ return functor(static_cast<const IntType&>(type));
+ case TypeID::kLong:
+ return functor(static_cast<const LongType&>(type));
+ default:
+ LOG(FATAL) << "Not supported";
+ }
+}
+
+template <typename FunctorT>
+inline auto InvokeOnType(const Type &type,
+ const FunctorT &functor) {
+ switch (type.getTypeID()) {
+ case TypeID::kInt:
+ return functor(static_cast<const IntType&>(type));
+ case TypeID::kLong:
+ return functor(static_cast<const LongType&>(type));
+ case TypeID::kFloat:
+ return functor(static_cast<const FloatType&>(type));
+ case TypeID::kDouble:
+ return functor(static_cast<const DoubleType&>(type));
+ default:
+ LOG(FATAL) << "Not supported";
+ }
+}
+
+template <typename FunctorT>
+inline auto InvokeOnBool(const bool &val,
+ const FunctorT &functor) {
+ if (val) {
+ return functor(std::true_type());
+ } else {
+ return functor(std::false_type());
+ }
+}
+
+template <typename FunctorT>
+inline auto InvokeOnBools(const bool &val1,
+ const bool &val2,
+ const FunctorT &functor) {
+ if (val1) {
+ if (val2) {
+ return functor(std::true_type(), std::true_type());
+ } else {
+ return functor(std::true_type(), std::false_type());
+ }
+ } else {
+ if (val2) {
+ return functor(std::false_type(), std::true_type());
+ } else {
+ return functor(std::false_type(), std::false_type());
+ }
+ }
+}
+
+template <typename FunctorT>
+inline auto InvokeOnAggFunc(const AggregationID &agg_id,
+ const FunctorT &functor) {
+ switch (agg_id) {
+ case AggregationID::kSum: {
+ return functor(Sum());
+ }
+ default:
+ LOG(FATAL) << "Not supported";
+ }
+}
+
+template <typename FunctorT>
+inline auto InvokeIf(const std::true_type &val,
+ const FunctorT &functor) {
+ return functor();
+}
+
+template <typename FunctorT>
+inline void InvokeIf(const std::false_type &val,
+ const FunctorT &functor) {
+}
+
+//template <typename FunctorT>
+//inline void InvokeOnAggFuncIfApplicableToArgType(
+// const AggregationID &agg_id,
+// const Type &arg_type,
+// const FunctorT &functor) {
+// InvokeOnAggFunc(
+// agg_id,
+// [&](const auto &agg_func) -> void {
+// InvokeOnType(
+// arg_type,
+// [&](const auto &arg_type) -> void {
+// using AggFuncT = std::remove_reference_t<decltype(agg_func)>;
+// using ArgT = remove_const_reference_t<decltype(arg_type)>;
+//
+// InvokeIf(
+// typename AggFuncT::template HasAtomicImpl<ArgT>(),
+// [&]() -> void {
+// functor(agg_func, arg_type);
+// });
+// });
+// });
+//}
+
+template <typename FunctorT>
+inline void InvokeOnAggFuncWithArgType(
+ const AggregationID &agg_id,
+ const Type &arg_type,
+ const FunctorT &functor) {
+ InvokeOnAggFunc(
+ agg_id,
+ [&](const auto &agg_func) -> void {
+ InvokeOnType(
+ arg_type,
+ [&](const auto &arg_type) -> void {
+ functor(agg_func, arg_type);
+ });
+ });
+}
+
+template <typename FunctorT>
+inline auto InvokeOnTwoAccessors(
+ const ValueAccessorMultiplexer &accessor_mux,
+ const ValueAccessorSource &first_source,
+ const ValueAccessorSource &second_source,
+ const FunctorT &functor) {
+ ValueAccessor *base_accessor = accessor_mux.getBaseAccessor();
+ ColumnVectorsValueAccessor *derived_accessor =
+ static_cast<ColumnVectorsValueAccessor *>(accessor_mux.getDerivedAccessor());
+
+ InvokeOnAnyValueAccessor(
+ base_accessor,
+ [&](auto *accessor) {
+ if (first_source == ValueAccessorSource::kBase) {
+ if (second_source == ValueAccessorSource::kBase) {
+ return functor(std::false_type(), accessor, accessor);
+ } else {
+ return functor(std::true_type(), accessor, derived_accessor);
+ }
+ } else {
+ if (second_source == ValueAccessorSource::kBase) {
+ return functor(std::true_type(), derived_accessor, accessor);
+ } else {
+ return functor(std::false_type(), derived_accessor, derived_accessor);
+ }
+ }
+ });
+}
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_STORAGE_AGGREGATION_UTIL_HPP_
diff --git a/storage/CMakeLists.txt b/storage/CMakeLists.txt
index fcc069b..8ef3560 100644
--- a/storage/CMakeLists.txt
+++ b/storage/CMakeLists.txt
@@ -145,6 +145,7 @@
AggregationOperationState.cpp
AggregationOperationState.hpp)
add_library(quickstep_storage_AggregationOperationState_proto ${storage_AggregationOperationState_proto_srcs})
+add_library(quickstep_storage_AggregationUtil ../empty_src.cpp AggregationUtil.hpp)
add_library(quickstep_storage_BasicColumnStoreTupleStorageSubBlock
BasicColumnStoreTupleStorageSubBlock.cpp
BasicColumnStoreTupleStorageSubBlock.hpp)
@@ -304,6 +305,15 @@
quickstep_expressions_aggregation_AggregateFunction_proto
quickstep_storage_HashTable_proto
${PROTOBUF_LIBRARY})
+target_link_libraries(quickstep_storage_AggregationUtil
+ quickstep_expressions_aggregation_AggFunc
+ quickstep_expressions_aggregation_AggregationID
+ quickstep_storage_ValueAccessor
+ quickstep_storage_ValueAccessorMultiplexer
+ quickstep_storage_ValueAccessorUtil
+ quickstep_types_Type
+ quickstep_types_TypeID
+ quickstep_types_containers_ColumnVectorsValueAccessor)
target_link_libraries(quickstep_storage_BasicColumnStoreTupleStorageSubBlock
quickstep_catalog_CatalogAttribute
quickstep_catalog_CatalogRelationSchema
@@ -441,6 +451,7 @@
quickstep_expressions_aggregation_AggFunc
quickstep_expressions_aggregation_AggregationHandle
quickstep_expressions_aggregation_AggregationID
+ quickstep_storage_AggregationUtil
quickstep_storage_HashTableBase
quickstep_storage_StorageBlob
quickstep_storage_StorageBlockInfo
@@ -453,7 +464,6 @@
quickstep_types_Type
quickstep_types_TypeID
quickstep_types_containers_ColumnVector
- quickstep_types_containers_ColumnVectorsValueAccessor
quickstep_utility_BoolVector
quickstep_utility_Macros)
target_link_libraries(quickstep_storage_ColumnStoreUtil
@@ -801,8 +811,11 @@
quickstep_utility_Macros
quickstep_utility_PrimeNumber)
target_link_libraries(quickstep_storage_PackedPayloadHashTable
+ ${GFLAGS_LIB_NAME}
quickstep_catalog_CatalogTypedefs
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_expressions_aggregation_AggFunc
+ quickstep_storage_AggregationUtil
quickstep_storage_HashTableBase
quickstep_storage_HashTableKeyManager
quickstep_storage_StorageBlob
@@ -1111,6 +1124,7 @@
target_link_libraries(quickstep_storage
quickstep_storage_AggregationOperationState
quickstep_storage_AggregationOperationState_proto
+ quickstep_storage_AggregationUtil
quickstep_storage_BasicColumnStoreTupleStorageSubBlock
quickstep_storage_BasicColumnStoreValueAccessor
quickstep_storage_BloomFilterIndexSubBlock
diff --git a/storage/CollisionFreeVectorTable.cpp b/storage/CollisionFreeVectorTable.cpp
index c92f0ab..4c57cc9 100644
--- a/storage/CollisionFreeVectorTable.cpp
+++ b/storage/CollisionFreeVectorTable.cpp
@@ -29,14 +29,13 @@
#include "expressions/aggregation/AggregationHandle.hpp"
#include "expressions/aggregation/AggregationID.hpp"
+#include "storage/AggregationUtil.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "storage/StorageManager.hpp"
#include "storage/ValueAccessor.hpp"
#include "storage/ValueAccessorMultiplexer.hpp"
#include "storage/ValueAccessorUtil.hpp"
#include "threading/SpinMutex.hpp"
-#include "types/TypeID.hpp"
-#include "types/containers/ColumnVectorsValueAccessor.hpp"
#include "utility/BoolVector.hpp"
#include "glog/logging.h"
@@ -46,163 +45,6 @@
DEFINE_uint64(vt_threadprivate_threshold, 1000000L, "");
DEFINE_bool(use_latch, false, "");
-namespace {
-
-template <typename T>
-using remove_const_reference_t = std::remove_const_t<std::remove_reference_t<T>>;
-
-template <typename FunctorT>
-inline auto InvokeOnKeyType(const Type &type,
- const FunctorT &functor) {
- switch (type.getTypeID()) {
- case TypeID::kInt:
- return functor(static_cast<const IntType&>(type));
- case TypeID::kLong:
- return functor(static_cast<const LongType&>(type));
- default:
- LOG(FATAL) << "Not supported";
- }
-}
-
-template <typename FunctorT>
-inline auto InvokeOnType(const Type &type,
- const FunctorT &functor) {
- switch (type.getTypeID()) {
- case TypeID::kInt:
- return functor(static_cast<const IntType&>(type));
- case TypeID::kLong:
- return functor(static_cast<const LongType&>(type));
- case TypeID::kFloat:
- return functor(static_cast<const FloatType&>(type));
- case TypeID::kDouble:
- return functor(static_cast<const DoubleType&>(type));
- default:
- LOG(FATAL) << "Not supported";
- }
-}
-
-template <typename FunctorT>
-inline auto InvokeOnBool(const bool &val,
- const FunctorT &functor) {
- if (val) {
- return functor(std::true_type());
- } else {
- return functor(std::false_type());
- }
-}
-
-template <typename FunctorT>
-inline auto InvokeOnBools(const bool &val1,
- const bool &val2,
- const FunctorT &functor) {
- if (val1) {
- if (val2) {
- return functor(std::true_type(), std::true_type());
- } else {
- return functor(std::true_type(), std::false_type());
- }
- } else {
- if (val2) {
- return functor(std::false_type(), std::true_type());
- } else {
- return functor(std::false_type(), std::false_type());
- }
- }
-}
-
-template <typename FunctorT>
-inline auto InvokeOnAggFunc(const AggregationID &agg_id,
- const FunctorT &functor) {
- switch (agg_id) {
- case AggregationID::kSum: {
- return functor(Sum());
- }
- default:
- LOG(FATAL) << "Not supported";
- }
-}
-
-template <typename FunctorT>
-inline auto InvokeIf(const std::true_type &val,
- const FunctorT &functor) {
- return functor();
-}
-
-template <typename FunctorT>
-inline void InvokeIf(const std::false_type &val,
- const FunctorT &functor) {
-}
-
-//template <typename FunctorT>
-//inline void InvokeOnAggFuncIfApplicableToArgType(
-// const AggregationID &agg_id,
-// const Type &arg_type,
-// const FunctorT &functor) {
-// InvokeOnAggFunc(
-// agg_id,
-// [&](const auto &agg_func) -> void {
-// InvokeOnType(
-// arg_type,
-// [&](const auto &arg_type) -> void {
-// using AggFuncT = std::remove_reference_t<decltype(agg_func)>;
-// using ArgT = remove_const_reference_t<decltype(arg_type)>;
-//
-// InvokeIf(
-// typename AggFuncT::template HasAtomicImpl<ArgT>(),
-// [&]() -> void {
-// functor(agg_func, arg_type);
-// });
-// });
-// });
-//}
-
-template <typename FunctorT>
-inline void InvokeOnAggFuncWithArgType(
- const AggregationID &agg_id,
- const Type &arg_type,
- const FunctorT &functor) {
- InvokeOnAggFunc(
- agg_id,
- [&](const auto &agg_func) -> void {
- InvokeOnType(
- arg_type,
- [&](const auto &arg_type) -> void {
- functor(agg_func, arg_type);
- });
- });
-}
-
-template <typename FunctorT>
-inline auto InvokeOnTwoAccessors(
- const ValueAccessorMultiplexer &accessor_mux,
- const ValueAccessorSource &first_source,
- const ValueAccessorSource &second_source,
- const FunctorT &functor) {
- ValueAccessor *base_accessor = accessor_mux.getBaseAccessor();
- ColumnVectorsValueAccessor *derived_accessor =
- static_cast<ColumnVectorsValueAccessor *>(accessor_mux.getDerivedAccessor());
-
- InvokeOnAnyValueAccessor(
- base_accessor,
- [&](auto *accessor) {
- if (first_source == ValueAccessorSource::kBase) {
- if (second_source == ValueAccessorSource::kBase) {
- return functor(std::false_type(), accessor, accessor);
- } else {
- return functor(std::true_type(), accessor, derived_accessor);
- }
- } else {
- if (second_source == ValueAccessorSource::kBase) {
- return functor(std::true_type(), derived_accessor, accessor);
- } else {
- return functor(std::false_type(), derived_accessor, derived_accessor);
- }
- }
- });
-}
-
-} // namespace
-
CollisionFreeVectorTable::CollisionFreeVectorTable(
const Type *key_type,
const std::size_t num_entries,
diff --git a/storage/PackedPayloadHashTable.cpp b/storage/PackedPayloadHashTable.cpp
index bd7e960..c9d956c 100644
--- a/storage/PackedPayloadHashTable.cpp
+++ b/storage/PackedPayloadHashTable.cpp
@@ -26,6 +26,8 @@
#include <vector>
#include "expressions/aggregation/AggregationHandle.hpp"
+#include "expressions/aggregation/AggFunc.hpp"
+#include "storage/AggregationUtil.hpp"
#include "storage/HashTableKeyManager.hpp"
#include "storage/StorageBlob.hpp"
#include "storage/StorageBlockInfo.hpp"
@@ -53,28 +55,14 @@
: key_types_(key_types),
num_handles_(handles.size()),
handles_(handles),
- total_payload_size_(ComputeTotalPayloadSize(handles)),
+ state_sizes_(ComputeStateSizes(handles)),
+ total_payload_size_(ComputeTotalPayloadSize(state_sizes_)),
+ state_offsets_(ComputeStateOffsets(state_sizes_)),
storage_manager_(storage_manager),
kBucketAlignment(alignof(std::atomic<std::size_t>)),
kValueOffset(sizeof(std::atomic<std::size_t>) + sizeof(std::size_t)),
key_manager_(key_types_, kValueOffset + total_payload_size_),
bucket_size_(ComputeBucketSize(key_manager_.getFixedKeySize())) {
- std::size_t payload_offset_running_sum = sizeof(SpinMutex);
- for (const auto *handle : handles) {
- payload_offsets_.emplace_back(payload_offset_running_sum);
- payload_offset_running_sum += handle->getPayloadSize();
- }
-
- // NOTE(jianqiao): Potential memory leak / double freeing by copying from
- // init_payload to buckets if payload contains out of line data.
- init_payload_ =
- static_cast<std::uint8_t *>(calloc(this->total_payload_size_, 1));
- DCHECK(init_payload_ != nullptr);
-
- for (std::size_t i = 0; i < num_handles_; ++i) {
- handles_[i]->initPayload(init_payload_ + payload_offsets_[i]);
- }
-
// Bucket size always rounds up to the alignment requirement of the atomic
// size_t "next" pointer at the front or a ValueT, whichever is larger.
//
@@ -192,7 +180,6 @@
blob_.release();
storage_manager_->deleteBlockOrBlobFile(blob_id);
}
- std::free(init_payload_);
}
void PackedPayloadHashTable::clear() {
@@ -214,17 +201,6 @@
}
void PackedPayloadHashTable::destroyPayload() {
- const std::size_t num_buckets =
- header_->buckets_allocated.load(std::memory_order_relaxed);
- void *bucket_ptr = static_cast<char *>(buckets_) + kValueOffset;
- for (std::size_t bucket_num = 0; bucket_num < num_buckets; ++bucket_num) {
- for (std::size_t handle_id = 0; handle_id < num_handles_; ++handle_id) {
- void *value_internal_ptr =
- static_cast<char *>(bucket_ptr) + this->payload_offsets_[handle_id];
- handles_[handle_id]->destroyPayload(static_cast<std::uint8_t *>(value_internal_ptr));
- }
- bucket_ptr = static_cast<char *>(bucket_ptr) + bucket_size_;
- }
}
bool PackedPayloadHashTable::upsertValueAccessorCompositeKey(
@@ -232,24 +208,131 @@
const std::vector<MultiSourceAttributeId> &key_attr_ids,
const ValueAccessorMultiplexer &accessor_mux) {
ValueAccessor *base_accessor = accessor_mux.getBaseAccessor();
- ValueAccessor *derived_accessor = accessor_mux.getDerivedAccessor();
+ CHECK(accessor_mux.getDerivedAccessor() == nullptr);
- base_accessor->beginIterationVirtual();
- if (derived_accessor == nullptr) {
- return upsertValueAccessorCompositeKeyInternal<false>(
- argument_ids,
- key_attr_ids,
- base_accessor,
- nullptr);
- } else {
- DCHECK(derived_accessor->getImplementationType()
- == ValueAccessor::Implementation::kColumnVectors);
- derived_accessor->beginIterationVirtual();
- return upsertValueAccessorCompositeKeyInternal<true>(
- argument_ids,
- key_attr_ids,
- base_accessor,
- static_cast<ColumnVectorsValueAccessor *>(derived_accessor));
+ std::vector<attribute_id> key_ids;
+ for (const auto &key_attr_id : key_attr_ids) {
+ key_ids.emplace_back(key_attr_id.attr_id);
+ }
+
+ for (std::size_t i = 0; i < num_handles_; ++i) {
+ DCHECK_LE(argument_ids[i].size(), 1u);
+
+ const AggregationHandle *handle = handles_[i];
+ const auto &argument_types = handle->getArgumentTypes();
+ const auto &argument_ids_i = argument_ids[i];
+
+ attribute_id argument_id = kInvalidAttributeID;
+ const Type *argument_type = nullptr;
+
+ if (argument_ids_i.empty()) {
+ LOG(FATAL) << "Not supported";
+ } else {
+ DCHECK_EQ(1u, argument_ids_i.size());
+ argument_id = argument_ids_i.front().attr_id;
+
+ DCHECK_EQ(1u, argument_types.size());
+ argument_type = argument_types.front();
+ }
+
+ InvokeOnAggFuncWithArgType(
+ handle->getAggregationID(),
+ *argument_type,
+ [&](const auto &agg_func, const auto &arg_type) {
+ using AggFuncT = std::remove_reference_t<decltype(agg_func)>;
+ using ArgT = remove_const_reference_t<decltype(arg_type)>;
+
+ InvokeOnAnyValueAccessor(
+ base_accessor,
+ [&](auto *accessor) -> void { // NOLINT(build/c++11)
+
+ if (key_ids.size() == 1) {
+ if (FLAGS_use_latch) {
+ this->upsertValueAccessorInternalUnaryLatch<AggFuncT, ArgT>(
+ argument_id,
+ key_ids.front(),
+ state_offsets_[i],
+ accessor);
+ } else {
+ this->upsertValueAccessorInternalUnaryAtomic<AggFuncT, ArgT>(
+ argument_id,
+ key_ids.front(),
+ state_offsets_[i],
+ accessor);
+ }
+ } else {
+ if (FLAGS_use_latch) {
+ this->upsertValueAccessorCompositeKeyInternalUnaryLatch<AggFuncT, ArgT>(
+ argument_id,
+ key_ids,
+ state_offsets_[i],
+ accessor);
+ } else {
+ this->upsertValueAccessorCompositeKeyInternalUnaryAtomic<AggFuncT, ArgT>(
+ argument_id,
+ key_ids,
+ state_offsets_[i],
+ accessor);
+ }
+ }
+ });
+ });
+ }
+ return true;
+}
+
+void PackedPayloadHashTable::finalize(const std::size_t partition_id,
+ ColumnVectorsValueAccessor *results) {
+ const std::size_t start_position =
+ calculatePartitionStartPosition(partition_id);
+ const std::size_t end_position =
+ calculatePartitionEndPosition(partition_id);
+
+ const std::size_t num_entries = end_position - start_position;
+
+ for (std::size_t key_idx = 0; key_idx < key_types_.size(); ++key_idx) {
+ NativeColumnVector *key_cv =
+ new NativeColumnVector(*key_types_[key_idx], num_entries);
+ finalizeKey(start_position, end_position, key_idx, key_cv);
+ results->addColumn(key_cv);
+ }
+
+ for (std::size_t i = 0; i < num_handles_; ++i) {
+ const AggregationHandle *handle = handles_[i];
+ const auto &argument_types = handle->getArgumentTypes();
+
+ const Type *argument_type = nullptr;
+ if (argument_types.empty()) {
+ LOG(FATAL) << "Not supported";
+ } else {
+ DCHECK_EQ(1u, argument_types.size());
+ argument_type = argument_types.front();
+ }
+
+ NativeColumnVector *result_cv =
+ new NativeColumnVector(*handle->getResultType(), num_entries);
+
+ InvokeOnAggFuncWithArgType(
+ handle->getAggregationID(),
+ *argument_type,
+ [&](const auto &agg_func, const auto &arg_type) {
+ using AggFuncT = std::remove_reference_t<decltype(agg_func)>;
+ using ArgT = remove_const_reference_t<decltype(arg_type)>;
+
+ if (FLAGS_use_latch) {
+ this->finalizeStateLatch<AggFuncT, ArgT>(start_position,
+ end_position,
+ state_offsets_[i],
+ result_cv);
+ } else {
+ this->finalizeStateAtomic<AggFuncT, ArgT>(start_position,
+ end_position,
+ state_offsets_[i],
+ result_cv);
+ }
+ });
+
+ results->addColumn(result_cv);
}
}
diff --git a/storage/PackedPayloadHashTable.hpp b/storage/PackedPayloadHashTable.hpp
index 9ba5500..1a56f94 100644
--- a/storage/PackedPayloadHashTable.hpp
+++ b/storage/PackedPayloadHashTable.hpp
@@ -20,6 +20,7 @@
#ifndef QUICKSTEP_STORAGE_PACKED_PAYLOAD_HASH_TABLE_HPP_
#define QUICKSTEP_STORAGE_PACKED_PAYLOAD_HASH_TABLE_HPP_
+#include <algorithm>
#include <atomic>
#include <cstddef>
#include <cstdint>
@@ -29,6 +30,8 @@
#include "catalog/CatalogTypedefs.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
+#include "expressions/aggregation/AggFunc.hpp"
+#include "storage/AggregationUtil.hpp"
#include "storage/HashTableBase.hpp"
#include "storage/HashTableKeyManager.hpp"
#include "storage/StorageBlob.hpp"
@@ -38,14 +41,20 @@
#include "threading/SpinMutex.hpp"
#include "threading/SpinSharedMutex.hpp"
#include "types/TypedValue.hpp"
+#include "types/containers/ColumnVector.hpp"
#include "types/containers/ColumnVectorsValueAccessor.hpp"
#include "utility/HashPair.hpp"
#include "utility/Macros.hpp"
+#include "gflags/gflags.h"
+
#include "glog/logging.h"
namespace quickstep {
+DECLARE_int32(num_workers);
+DECLARE_bool(use_latch);
+
class StorageManager;
class Type;
class ValueAccessor;
@@ -126,6 +135,9 @@
const std::vector<MultiSourceAttributeId> &key_ids,
const ValueAccessorMultiplexer &accessor_mux) override;
+ void finalize(const std::size_t partition_id,
+ ColumnVectorsValueAccessor *results);
+
/**
* @return The ID of the StorageBlob used to store this hash table.
**/
@@ -364,16 +376,55 @@
const std::uint8_t **value,
std::size_t *entry_num) const;
- inline std::uint8_t* upsertCompositeKeyInternal(
- const std::vector<TypedValue> &key,
- const std::size_t variable_key_size);
+ inline std::uint8_t* upsertInternal(const TypedValue &key);
- template <bool use_two_accessors>
- inline bool upsertValueAccessorCompositeKeyInternal(
- const std::vector<std::vector<MultiSourceAttributeId>> &argument_ids,
- const std::vector<MultiSourceAttributeId> &key_ids,
- ValueAccessor *base_accessor,
- ColumnVectorsValueAccessor *derived_accessor);
+ inline std::uint8_t* upsertCompositeKeyInternal(
+ const std::vector<TypedValue> &key);
+
+ template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+ inline bool upsertValueAccessorInternalUnaryAtomic(
+ const attribute_id argument_ids,
+ const attribute_id key_id,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor);
+
+ template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+ inline bool upsertValueAccessorInternalUnaryLatch(
+ const attribute_id argument_ids,
+ const attribute_id key_id,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor);
+
+ template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+ inline bool upsertValueAccessorCompositeKeyInternalUnaryAtomic(
+ const attribute_id argument_id,
+ const std::vector<attribute_id> key_ids,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor);
+
+ template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+ inline bool upsertValueAccessorCompositeKeyInternalUnaryLatch(
+ const attribute_id argument_id,
+ const std::vector<attribute_id> key_ids,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor);
+
+ inline void finalizeKey(const std::size_t start_position,
+ const std::size_t end_position,
+ const std::size_t key_idx,
+ NativeColumnVector *output_cv);
+
+ template <typename AggFuncT, typename ArgT>
+ inline void finalizeStateAtomic(const std::size_t start_position,
+ const std::size_t end_position,
+ const std::size_t state_offset,
+ NativeColumnVector *results);
+
+ template <typename AggFuncT, typename ArgT>
+ inline void finalizeStateLatch(const std::size_t start_position,
+ const std::size_t end_position,
+ const std::size_t state_offset,
+ NativeColumnVector *results);
// Generate a hash for a composite key by hashing each component of 'key' and
// mixing their bits with CombineHashes().
@@ -387,33 +438,62 @@
key_inline_ = key_inline;
}
- inline static std::size_t ComputeTotalPayloadSize(
+ inline static std::vector<std::size_t> ComputeStateSizes(
const std::vector<AggregationHandle *> &handles) {
- std::size_t total_payload_size = sizeof(SpinMutex);
- for (const auto *handle : handles) {
- total_payload_size += handle->getPayloadSize();
+ std::vector<std::size_t> state_sizes;
+ for (std::size_t i = 0; i < handles.size(); ++i) {
+ const AggregationHandle *handle = handles[i];
+ InvokeOnAggFuncWithArgType(
+ handle->getAggregationID(),
+ *handle->getArgumentTypes().front(),
+ [&](const auto &agg_func, const auto &arg_type) {
+ using AggFuncT = std::remove_reference_t<decltype(agg_func)>;
+ using ArgT = remove_const_reference_t<decltype(arg_type)>;
+
+ if (FLAGS_use_latch) {
+ state_sizes.emplace_back(
+ sizeof(typename AggFuncT::template AggState<ArgT>::T));
+ } else {
+ state_sizes.emplace_back(
+ sizeof(typename AggFuncT::template AggState<ArgT>::AtomicT));
+ }
+ });
}
- return total_payload_size;
+ return state_sizes;
+ }
+
+ inline static std::size_t ComputeTotalPayloadSize(
+ const std::vector<std::size_t> &state_sizes) {
+ const std::size_t mutex_size =
+ FLAGS_use_latch ? sizeof(SpinMutex) : 0;
+ const std::size_t total_state_size =
+ std::accumulate(state_sizes.begin(), state_sizes.end(), 0);
+ return mutex_size + total_state_size;
+ }
+
+ inline static std::vector<std::size_t> ComputeStateOffsets(
+ const std::vector<std::size_t> &state_sizes) {
+ std::vector<std::size_t> state_offsets;
+ std::size_t state_offset =
+ FLAGS_use_latch ? sizeof(SpinMutex) : 0;
+ for (const std::size_t state_size : state_sizes) {
+ state_offsets.emplace_back(state_offset);
+ state_offset += state_size;
+ }
+ return state_offsets;
}
// Assign '*key_vector' with the attribute values specified by 'key_ids' at
// the current position of 'accessor'. If 'check_for_null_keys' is true, stops
// and returns true if any of the values is null, otherwise returns false.
- template <bool use_two_accessors,
- bool check_for_null_keys,
+ template <bool check_for_null_keys,
typename ValueAccessorT>
inline static bool GetCompositeKeyFromValueAccessor(
- const std::vector<MultiSourceAttributeId> &key_ids,
+ const std::vector<attribute_id> &key_ids,
const ValueAccessorT *accessor,
- const ColumnVectorsValueAccessor *derived_accessor,
std::vector<TypedValue> *key_vector) {
for (std::size_t key_idx = 0; key_idx < key_ids.size(); ++key_idx) {
- const MultiSourceAttributeId &key_id = key_ids[key_idx];
- if (use_two_accessors && key_id.source == ValueAccessorSource::kDerived) {
- (*key_vector)[key_idx] = derived_accessor->getTypedValue(key_id.attr_id);
- } else {
- (*key_vector)[key_idx] = accessor->getTypedValue(key_id.attr_id);
- }
+ (*key_vector)[key_idx] = accessor->getTypedValue(key_ids[key_idx]);
if (check_for_null_keys && (*key_vector)[key_idx].isNull()) {
return true;
}
@@ -441,9 +521,9 @@
const std::size_t num_handles_;
const std::vector<AggregationHandle *> handles_;
- std::size_t total_payload_size_;
- std::vector<std::size_t> payload_offsets_;
- std::uint8_t *init_payload_;
+ const std::vector<std::size_t> state_sizes_;
+ const std::size_t total_payload_size_;
+ const std::vector<std::size_t> state_offsets_;
StorageManager *storage_manager_;
MutableBlobReference blob_;
@@ -471,8 +551,8 @@
// Set finalization segment size as 4096 entries.
constexpr std::size_t kFinalizeSegmentSize = 4uL * 1024L;
- // At least 1 partition, at most 80 partitions.
- return std::max(1uL, std::min(num_entries / kFinalizeSegmentSize, 80uL));
+ return std::max(static_cast<std::size_t>(FLAGS_num_workers),
+ std::min(num_entries / kFinalizeSegmentSize, 80uL));
}
// Attempt to find an empty bucket to insert 'hash_code' into, starting after
@@ -488,7 +568,6 @@
// deallocated after being allocated.
inline bool locateBucketForInsertion(
const std::size_t hash_code,
- const std::size_t variable_key_allocation_required,
void **bucket,
std::atomic<std::size_t> **pending_chain_ptr,
std::size_t *pending_chain_ptr_finish_value);
@@ -636,7 +715,6 @@
inline bool PackedPayloadHashTable::locateBucketForInsertion(
const std::size_t hash_code,
- const std::size_t variable_key_allocation_required,
void **bucket,
std::atomic<std::size_t> **pending_chain_ptr,
std::size_t *pending_chain_ptr_finish_value) {
@@ -652,17 +730,6 @@
std::numeric_limits<std::size_t>::max(),
std::memory_order_acq_rel)) {
// Got to the end of the chain. Allocate a new bucket.
-
- // First, allocate variable-length key storage, if needed (i.e. if this
- // is an upsert and we didn't allocate up-front).
- if (!key_manager_.allocateVariableLengthKeyStorage(
- variable_key_allocation_required)) {
- // Ran out of variable-length storage.
- (*pending_chain_ptr)->store(0, std::memory_order_release);
- *bucket = nullptr;
- return false;
- }
-
const std::size_t allocated_bucket_num =
header_->buckets_allocated.fetch_add(1, std::memory_order_relaxed);
if (allocated_bucket_num >= header_->num_buckets) {
@@ -730,27 +797,7 @@
inline const std::uint8_t* PackedPayloadHashTable::getSingleCompositeKey(
const std::vector<TypedValue> &key,
const std::size_t index) const {
- DEBUG_ASSERT(this->key_types_.size() == key.size());
-
- const std::size_t hash_code = this->hashCompositeKey(key);
- std::size_t bucket_ref =
- slots_[hash_code % header_->num_slots].load(std::memory_order_relaxed);
- while (bucket_ref != 0) {
- DEBUG_ASSERT(bucket_ref != std::numeric_limits<std::size_t>::max());
- const char *bucket =
- static_cast<const char *>(buckets_) + (bucket_ref - 1) * bucket_size_;
- const std::size_t bucket_hash = *reinterpret_cast<const std::size_t *>(
- bucket + sizeof(std::atomic<std::size_t>));
- if ((bucket_hash == hash_code) &&
- key_manager_.compositeKeyCollisionCheck(key, bucket)) {
- // Match located.
- return reinterpret_cast<const std::uint8_t *>(bucket + kValueOffset) +
- this->payload_offsets_[index];
- }
- bucket_ref =
- reinterpret_cast<const std::atomic<std::size_t> *>(bucket)->load(
- std::memory_order_relaxed);
- }
+ LOG(FATAL) << "Not implemented";
// Reached the end of the chain and didn't find a match.
return nullptr;
@@ -759,24 +806,9 @@
inline bool PackedPayloadHashTable::upsertCompositeKey(
const std::vector<TypedValue> &key,
const std::uint8_t *source_state) {
- const std::size_t variable_size =
- calculateVariableLengthCompositeKeyCopySize(key);
- for (;;) {
- {
- SpinSharedMutexSharedLock<true> resize_lock(resize_shared_mutex_);
- std::uint8_t *value =
- upsertCompositeKeyInternal(key, variable_size);
- if (value != nullptr) {
- SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
- for (unsigned int k = 0; k < num_handles_; ++k) {
- handles_[k]->mergeStates(source_state + payload_offsets_[k],
- value + payload_offsets_[k]);
- }
- return true;
- }
- }
- resize(0, variable_size);
- }
+ LOG(FATAL) << "Not implemented";
+
+ return true;
}
template <typename FunctorT>
@@ -784,48 +816,56 @@
const std::vector<TypedValue> &key,
FunctorT *functor,
const std::size_t index) {
- const std::size_t variable_size =
- calculateVariableLengthCompositeKeyCopySize(key);
- for (;;) {
- {
- SpinSharedMutexSharedLock<true> resize_lock(resize_shared_mutex_);
- std::uint8_t *value =
- upsertCompositeKeyInternal(key, variable_size);
- if (value != nullptr) {
- (*functor)(value + payload_offsets_[index]);
- return true;
- }
- }
- resize(0, variable_size);
- }
+ LOG(FATAL) << "Not implemented";
+
+ return true;
}
-
-inline std::uint8_t* PackedPayloadHashTable::upsertCompositeKeyInternal(
- const std::vector<TypedValue> &key,
- const std::size_t variable_key_size) {
- if (variable_key_size > 0) {
- // Don't allocate yet, since the key may already be present. However, we
- // do check if either the allocated variable storage space OR the free
- // space is big enough to hold the key (at least one must be true: either
- // the key is already present and allocated, or we need to be able to
- // allocate enough space for it).
- std::size_t allocated_bytes = header_->variable_length_bytes_allocated.load(
- std::memory_order_relaxed);
- if ((allocated_bytes < variable_key_size) &&
- (allocated_bytes + variable_key_size >
- key_manager_.getVariableLengthKeyStorageSize())) {
+inline std::uint8_t* PackedPayloadHashTable::upsertInternal(
+ const TypedValue &key) {
+ const std::size_t hash_code = key.getHash();
+ void *bucket = nullptr;
+ std::atomic<std::size_t> *pending_chain_ptr;
+ std::size_t pending_chain_ptr_finish_value;
+ for (;;) {
+ if (locateBucketForInsertion(hash_code,
+ &bucket,
+ &pending_chain_ptr,
+ &pending_chain_ptr_finish_value)) {
+ // Found an empty bucket.
+ break;
+ } else if (bucket == nullptr) {
+ // Ran out of buckets or variable-key space.
return nullptr;
+ } else if (key_manager_.scalarKeyCollisionCheck(key, bucket)) {
+ // Found an already-existing entry for this key.
+ return reinterpret_cast<std::uint8_t *>(static_cast<char *>(bucket) +
+ kValueOffset);
}
}
+ // We are now writing to an empty bucket.
+ // Write the key and hash.
+ writeScalarKeyToBucket(key, hash_code, bucket);
+
+ std::uint8_t *value = static_cast<unsigned char *>(bucket) + kValueOffset;
+// std::memcpy(value, init_payload_, this->total_payload_size_);
+
+ // Update the previous chain pointer to point to the new bucket.
+ pending_chain_ptr->store(pending_chain_ptr_finish_value, std::memory_order_release);
+
+ // Return the value.
+ return value;
+}
+
+inline std::uint8_t* PackedPayloadHashTable::upsertCompositeKeyInternal(
+ const std::vector<TypedValue> &key) {
const std::size_t hash_code = this->hashCompositeKey(key);
void *bucket = nullptr;
std::atomic<std::size_t> *pending_chain_ptr;
std::size_t pending_chain_ptr_finish_value;
for (;;) {
if (locateBucketForInsertion(hash_code,
- variable_key_size,
&bucket,
&pending_chain_ptr,
&pending_chain_ptr_finish_value)) {
@@ -846,7 +886,8 @@
writeCompositeKeyToBucket(key, hash_code, bucket);
std::uint8_t *value = static_cast<unsigned char *>(bucket) + kValueOffset;
- std::memcpy(value, init_payload_, this->total_payload_size_);
+// std::memcpy(value, init_payload_, this->total_payload_size_);
+ std::memset(value, 0, this->total_payload_size_);
// Update the previous chaing pointer to point to the new bucket.
pending_chain_ptr->store(pending_chain_ptr_finish_value,
@@ -856,72 +897,165 @@
return value;
}
-template <bool use_two_accessors>
-inline bool PackedPayloadHashTable::upsertValueAccessorCompositeKeyInternal(
- const std::vector<std::vector<MultiSourceAttributeId>> &argument_ids,
- const std::vector<MultiSourceAttributeId> &key_ids,
- ValueAccessor *base_accessor,
- ColumnVectorsValueAccessor *derived_accessor) {
- std::size_t variable_size;
+template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+inline bool PackedPayloadHashTable
+ ::upsertValueAccessorInternalUnaryAtomic(
+ const attribute_id argument_id,
+ const attribute_id key_id,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor) {
+ using StateT = typename AggFuncT::template AggState<ArgT>;
+
+ accessor->beginIteration();
+ while (accessor->next()) {
+ TypedValue key = accessor->getTypedValue(key_id);
+ std::uint8_t *payload = this->upsertInternal(key);
+ const auto *argument = static_cast<const typename ArgT::cpptype *>(
+ accessor->template getUntypedValue<false>(argument_id));
+ auto *state =
+ reinterpret_cast<typename StateT::AtomicT *>(payload + state_offset);
+
+ AggFuncT::template MergeArgAtomic<ArgT>(*argument, state);
+ }
+ return true;
+}
+
+template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+inline bool PackedPayloadHashTable
+ ::upsertValueAccessorInternalUnaryLatch(
+ const attribute_id argument_id,
+ const attribute_id key_id,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor) {
+ using StateT = typename AggFuncT::template AggState<ArgT>;
+
+ accessor->beginIteration();
+ while (accessor->next()) {
+ TypedValue key = accessor->getTypedValue(key_id);
+ std::uint8_t *payload = this->upsertInternal(key);
+ const auto *argument = static_cast<const typename ArgT::cpptype *>(
+ accessor->template getUntypedValue<false>(argument_id));
+ auto *state =
+ reinterpret_cast<typename StateT::T *>(payload + state_offset);
+
+ SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(payload)));
+ AggFuncT::template MergeArgUnsafe<ArgT>(*argument, state);
+ }
+ return true;
+}
+
+template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+inline bool PackedPayloadHashTable::
+ upsertValueAccessorCompositeKeyInternalUnaryAtomic(
+ const attribute_id argument_id,
+ const std::vector<attribute_id> key_ids,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor) {
+ using StateT = typename AggFuncT::template AggState<ArgT>;
+
std::vector<TypedValue> key_vector;
key_vector.resize(key_ids.size());
- return InvokeOnAnyValueAccessor(
- base_accessor,
- [&](auto *accessor) -> bool { // NOLINT(build/c++11)
- bool continuing = true;
- while (continuing) {
- {
- continuing = false;
- SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
- while (accessor->next()) {
- if (use_two_accessors) {
- derived_accessor->next();
- }
- if (this->GetCompositeKeyFromValueAccessor<use_two_accessors, true>(
- key_ids,
- accessor,
- derived_accessor,
- &key_vector)) {
- continue;
- }
- variable_size = this->calculateVariableLengthCompositeKeyCopySize(key_vector);
- std::uint8_t *value = this->upsertCompositeKeyInternal(
- key_vector, variable_size);
- if (value == nullptr) {
- continuing = true;
- break;
- } else {
- SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
- for (unsigned int k = 0; k < num_handles_; ++k) {
- const auto &ids = argument_ids[k];
- if (ids.empty()) {
- handles_[k]->updateStateNullary(value + payload_offsets_[k]);
- } else {
- const MultiSourceAttributeId &arg_id = ids.front();
- if (use_two_accessors && arg_id.source == ValueAccessorSource::kDerived) {
- DCHECK_NE(arg_id.attr_id, kInvalidAttributeID);
- handles_[k]->updateStateUnary(derived_accessor->getTypedValue(arg_id.attr_id),
- value + payload_offsets_[k]);
- } else {
- handles_[k]->updateStateUnary(accessor->getTypedValue(arg_id.attr_id),
- value + payload_offsets_[k]);
- }
- }
- }
- }
- }
- }
- if (continuing) {
- this->resize(0, variable_size);
- accessor->previous();
- if (use_two_accessors) {
- derived_accessor->previous();
- }
- }
- }
- return true;
- });
+ accessor->beginIteration();
+ while (accessor->next()) {
+ this->GetCompositeKeyFromValueAccessor<false>(key_ids,
+ accessor,
+ &key_vector);
+ std::uint8_t *payload = this->upsertCompositeKeyInternal(key_vector);
+ const auto *argument = static_cast<const typename ArgT::cpptype *>(
+ accessor->template getUntypedValue<false>(argument_id));
+ auto *state =
+ reinterpret_cast<typename StateT::AtomicT *>(payload + state_offset);
+
+ AggFuncT::template MergeArgAtomic<ArgT>(*argument, state);
+ }
+ return true;
+}
+
+template <typename AggFuncT, typename ArgT, typename ValueAccessorT>
+inline bool PackedPayloadHashTable::
+ upsertValueAccessorCompositeKeyInternalUnaryLatch(
+ const attribute_id argument_id,
+ const std::vector<attribute_id> key_ids,
+ const std::size_t state_offset,
+ ValueAccessorT *accessor) {
+ using StateT = typename AggFuncT::template AggState<ArgT>;
+
+ std::vector<TypedValue> key_vector;
+ key_vector.resize(key_ids.size());
+
+ accessor->beginIteration();
+ while (accessor->next()) {
+ this->GetCompositeKeyFromValueAccessor<false>(key_ids,
+ accessor,
+ &key_vector);
+ std::uint8_t *payload = this->upsertCompositeKeyInternal(key_vector);
+ const auto *argument = static_cast<const typename ArgT::cpptype *>(
+ accessor->template getUntypedValue<false>(argument_id));
+ auto *state =
+ reinterpret_cast<typename StateT::T *>(payload + state_offset);
+
+ SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(payload)));
+ AggFuncT::template MergeArgUnsafe<ArgT>(*argument, state);
+ }
+ return true;
+}
+
+inline void PackedPayloadHashTable
+ ::finalizeKey(const std::size_t start_position,
+ const std::size_t end_position,
+ const std::size_t key_idx,
+ NativeColumnVector *output_cv) {
+ for (std::size_t i = start_position; i < end_position; ++i) {
+ const char *bucket =
+ static_cast<const char *>(buckets_) + i * bucket_size_;
+ output_cv->appendTypedValue(
+ key_manager_.getKeyComponentTyped(bucket, key_idx));
+ }
+}
+
+template <typename AggFuncT, typename ArgT>
+inline void PackedPayloadHashTable
+ ::finalizeStateAtomic(const std::size_t start_position,
+ const std::size_t end_position,
+ const std::size_t state_offset,
+ NativeColumnVector *output_cv) {
+ using StateT = typename AggFuncT::template AggState<ArgT>;
+ using ResultT = typename StateT::ResultT;
+
+ const std::size_t offset = kValueOffset + state_offset;
+ for (std::size_t i = start_position; i < end_position; ++i) {
+ const char *bucket =
+ static_cast<const char *>(buckets_) + i * bucket_size_;
+ const auto *state =
+ reinterpret_cast<const typename StateT::AtomicT *>(bucket + offset);
+
+ AggFuncT::template FinalizeAtomic<ArgT>(
+ *state,
+ static_cast<ResultT *>(output_cv->getPtrForDirectWrite()));
+ }
+}
+
+template <typename AggFuncT, typename ArgT>
+inline void PackedPayloadHashTable
+ ::finalizeStateLatch(const std::size_t start_position,
+ const std::size_t end_position,
+ const std::size_t state_offset,
+ NativeColumnVector *output_cv) {
+ using StateT = typename AggFuncT::template AggState<ArgT>;
+ using ResultT = typename StateT::ResultT;
+
+ const std::size_t offset = kValueOffset + state_offset;
+ for (std::size_t i = start_position; i < end_position; ++i) {
+ const char *bucket =
+ static_cast<const char *>(buckets_) + i * bucket_size_;
+ const auto *state =
+ reinterpret_cast<const typename StateT::T *>(bucket + offset);
+
+ AggFuncT::template FinalizeUnsafe<ArgT>(
+ *state,
+ static_cast<ResultT *>(output_cv->getPtrForDirectWrite()));
+ }
}
inline void PackedPayloadHashTable::writeScalarKeyToBucket(
@@ -990,7 +1124,7 @@
const std::uint8_t *value_ptr;
while (getNextEntry(&key, &value_ptr, &entry_num)) {
++entries_visited;
- (*functor)(key, value_ptr + payload_offsets_[index]);
+ (*functor)(key, value_ptr + state_offsets_[index]);
key.clear();
}
return entries_visited;
@@ -1044,7 +1178,7 @@
const std::uint8_t *value_ptr;
while (getNextEntryCompositeKey(&key, &value_ptr, &entry_num)) {
++entries_visited;
- (*functor)(key, value_ptr + payload_offsets_[index]);
+ (*functor)(key, value_ptr + state_offsets_[index]);
key.clear();
}
return entries_visited;
