You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by ra...@apache.org on 2016/07/04 07:57:00 UTC
[1/5] incubator-quickstep git commit: Initial commit for QUICKSTEP-28
and QUICKSTEP-29. Code refactoring and cleanup,
some more optimizations are pending.
Repository: incubator-quickstep
Updated Branches:
refs/heads/quickstep-28-29 [created] d5e7d09fc
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/StorageBlock.hpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.hpp b/storage/StorageBlock.hpp
index 3ae3812..8a0bd42 100644
--- a/storage/StorageBlock.hpp
+++ b/storage/StorageBlock.hpp
@@ -466,6 +466,14 @@ class StorageBlock : public StorageBlockBase {
std::vector<std::unique_ptr<ColumnVector>>
*reuse_group_by_vectors) const;
+
+ void aggregateGroupByFast(const std::vector<std::vector<std::unique_ptr<const Scalar>>> &arguments,
+ const std::vector<std::unique_ptr<const Scalar>> &group_by,
+ const Predicate *predicate,
+ AggregationStateHashTableBase *hash_table,
+ std::unique_ptr<TupleIdSequence> *reuse_matches,
+ std::vector<std::unique_ptr<ColumnVector>>
+ *reuse_group_by_vectors) const;
/**
* @brief Inserts the GROUP BY expressions and aggregation arguments together
* as keys into the distinctify hash table.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/threading/SpinMutex.hpp
----------------------------------------------------------------------
diff --git a/threading/SpinMutex.hpp b/threading/SpinMutex.hpp
index b64a6ca..c9838e4 100644
--- a/threading/SpinMutex.hpp
+++ b/threading/SpinMutex.hpp
@@ -42,6 +42,8 @@ class SpinMutex {
SpinMutex() : locked_(false) {
}
+ explicit SpinMutex(uint8_t *ptr): locked_(*ptr) {}
+
/**
* @note This call does NOT yield when contended. SpinMutex is intended
* mainly for cases where locks are held briefly and it is better to
[4/5] incubator-quickstep git commit: Initial commit for QUICKSTEP-28
and QUICKSTEP-29. Code refactoring and cleanup,
some more optimizations are pending.
Posted by ra...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/FastHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/FastHashTable.hpp b/storage/FastHashTable.hpp
new file mode 100644
index 0000000..12e447f
--- /dev/null
+++ b/storage/FastHashTable.hpp
@@ -0,0 +1,2640 @@
+/**
+ * Copyright 2011-2015 Quickstep Technologies LLC.
+ * Copyright 2015-2016 Pivotal Software, Inc.
+ * Copyright 2016, Quickstep Research Group, Computer Sciences Department,
+ * University of Wisconsin\u2014Madison.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+#ifndef QUICKSTEP_STORAGE_FAST_HASH_TABLE_HPP_
+#define QUICKSTEP_STORAGE_FAST_HASH_TABLE_HPP_
+
+#include <atomic>
+#include <cstddef>
+#include <cstdlib>
+#include <type_traits>
+#include <vector>
+
+#include "catalog/CatalogTypedefs.hpp"
+#include "storage/HashTableBase.hpp"
+#include "storage/StorageBlob.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "storage/StorageConstants.hpp"
+#include "storage/StorageManager.hpp"
+#include "storage/TupleReference.hpp"
+#include "storage/ValueAccessor.hpp"
+#include "storage/ValueAccessorUtil.hpp"
+#include "expressions/aggregation/AggregationHandleAvg.hpp"
+#include "threading/SpinSharedMutex.hpp"
+#include "threading/SpinMutex.hpp"
+#include "types/Type.hpp"
+#include "types/TypedValue.hpp"
+#include "utility/BloomFilter.hpp"
+#include "utility/HashPair.hpp"
+#include "utility/Macros.hpp"
+#include "storage/HashTable.hpp"
+
+namespace quickstep {
+
+/** \addtogroup Storage
+ * @{
+ */
+
+/**
+ * @brief Base class for hash table.
+ *
+ * This class is templated so that the core hash-table logic can be reused in
+ * different contexts requiring different value types and semantics (e.g.
+ * hash-joins vs. hash-based grouping for aggregates vs. hash-based indices).
+ * The base template defines the interface that HashTables provide to clients
+ * and implements some common functionality for all HashTables. There a few
+ * different (also templated) implementation classes that inherit from this
+ * base class and have different physical layouts with different performance
+ * characteristics. As of this writing, they are:
+ * 1. LinearOpenAddressingHashTable - All keys/values are stored directly
+ * in a single array of buckets. Collisions are handled by simply
+ * advancing to the "next" adjacent bucket until an empty bucket is
+ * found. This implementation is vulnerable to performance degradation
+ * due to the formation of bucket chains when there are many duplicate
+ * and/or consecutive keys.
+ * 2. SeparateChainingHashTable - Keys/values are stored in a separate
+ * region of memory from the base hash table slot array. Every bucket
+ * has a "next" pointer so that entries that collide (i.e. map to the
+ * same base slot) form chains of pointers with each other. Although
+ * this implementation has some extra indirection compared to
+ * LinearOpenAddressingHashTable, it does not have the same
+ * vulnerabilities to key skew, and it additionally supports a very
+ * efficient bucket-preallocation mechanism that minimizes cache
+ * coherency overhead when multiple threads are building a HashTable
+ * as part of a hash-join.
+ * 3. SimpleScalarSeparateChainingHashTable - A simplified version of
+ * SeparateChainingHashTable that is only usable for single, scalar
+ * keys with a reversible hash function. This implementation exploits
+ * the reversible hash to avoid storing separate copies of keys at all,
+ * and to skip an extra key comparison when hash codes collide.
+ *
+ * @note If you need to create a HashTable and not just use it as a client, see
+ * HashTableFactory, which simplifies the process of creating a
+ * HashTable.
+ *
+ * @param ValueT The mapped value in this hash table. Must be
+ * copy-constructible. For a serializable hash table, ValueT must also
+ * be trivially copyable and trivially destructible (and beware of
+ * pointers to external memory).
+ * @param resizable Whether this hash table is resizable (using memory from a
+ * StorageManager) or not (using a private, fixed memory allocation).
+ * @param serializable If true, this hash table can safely be saved to and
+ * loaded from disk. If false, some out of band memory may be used (e.g.
+ * to store variable length keys).
+ * @param force_key_copy If true, inserted keys are always copied into this
+ * HashTable's memory. If false, pointers to external values may be
+ * stored instead. force_key_copy should be true if the hash table will
+ * outlive the external key values which are inserted into it. Note that
+ * if serializable is true and force_key_copy is false, then relative
+ * offsets will be used instead of absolute pointers to keys, meaning
+ * that the pointed-to keys must be serialized and deserialized in
+ * exactly the same relative byte order (e.g. as part of the same
+ * StorageBlock), and keys must not change position relative to this
+ * HashTable (beware TupleStorageSubBlocks that may self-reorganize when
+ * modified). If serializable and resizable are both true, then
+ * force_key_copy must also be true.
+ * @param allow_duplicate_keys If true, multiple values can be mapped to the
+ * same key. If false, one and only one value may be mapped.
+ **/
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+class FastHashTable : public HashTableBase<resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys> {
+ static_assert(!(serializable && resizable && !force_key_copy),
+ "A HashTable must have force_key_copy=true when serializable "
+ "and resizable are both true.");
+
+ // TODO(chasseur): GCC 4.8.3 doesn't yet implement
+ // std::is_trivially_copyable. In the future, we should include a
+ // static_assert that prevents a serializable HashTable from being used with
+ // a ValueT which is not trivially copyable.
+
+ public:
+ // Shadow template parameters. This is useful for shared test harnesses.
+// typedef ValueT value_type;
+ static constexpr bool template_resizable = resizable;
+ static constexpr bool template_serializable = serializable;
+ static constexpr bool template_force_key_copy = force_key_copy;
+ static constexpr bool template_allow_duplicate_keys = allow_duplicate_keys;
+
+ // Some HashTable implementations (notably LinearOpenAddressingHashTable)
+ // use a special hash code to represent an empty bucket, and another special
+ // code to indicate that a bucket is currently being inserted into. For those
+ // HashTables, this is a surrogate hash value for empty buckets. Keys which
+ // actually hash to this value should have their hashes mutated (e.g. by
+ // adding 1). We use zero, since we will often be using memory which is
+ // already zeroed-out and this saves us the trouble of a memset. This has
+ // some downside, as the hash function we use is the identity hash for
+ // integers, and the integer 0 is common in many data sets and must be
+ // adjusted (and will then spuriously collide with 1). Nevertheless, this
+ // expense is outweighed by no longer having to memset large regions of
+ // memory when initializing a HashTable.
+ static constexpr unsigned char kEmptyHashByte = 0x0;
+ static constexpr std::size_t kEmptyHash = 0x0;
+
+ // A surrogate hash value for a bucket which is currently being inserted
+ // into. As with kEmptyHash, keys which actually hash to this value should
+ // have their hashes adjusted.
+ static constexpr std::size_t kPendingHash = ~kEmptyHash;
+
+ /**
+ * @brief Virtual destructor.
+ **/
+ virtual ~FastHashTable() {
+ if (resizable) {
+ if (blob_.valid()) {
+ if (serializable) {
+ DEV_WARNING("Destroying a resizable serializable HashTable's underlying "
+ "StorageBlob.");
+ }
+ const block_id blob_id = blob_->getID();
+ blob_.release();
+ storage_manager_->deleteBlockOrBlobFile(blob_id);
+ }
+ }
+ }
+
+ /**
+ * @brief Get the ID of the StorageBlob used to store a resizable HashTable.
+ *
+ * @warning This method must not be used for a non-resizable HashTable.
+ *
+ * @return The ID of the StorageBlob used to store this HashTable.
+ **/
+ inline block_id getBlobId() const {
+ DEBUG_ASSERT(resizable);
+ return blob_->getID();
+ }
+
+ /**
+ * @brief Erase all entries in this hash table.
+ *
+ * @warning This method is not guaranteed to be threadsafe.
+ **/
+ virtual void clear() = 0;
+
+ /**
+ * @brief Add a new entry into the hash table.
+ *
+ * @warning The key must not be null.
+ * @warning This method is threadsafe with regard to other calls to put(),
+ * putCompositeKey(), putValueAccessor(), and
+ * putValueAccessorCompositeKey(), but should not be used
+ * simultaneously with upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey().
+ * @note This version is for single scalar keys, see also putCompositeKey().
+ * @note If the hash table is (close to) full and resizable is true, this
+ * routine might result in rebuilding the entire hash table.
+ *
+ * @param key The key.
+ * @param value The value payload.
+ * @return HashTablePutResult::kOK if an entry was successfully inserted,
+ * HashTablePutResult::kDuplicateKey if allow_duplicate_keys is false
+ * and key was a duplicate, or HashTablePutResult::kOutOfSpace if
+ * resizable is false and storage space for the hash table has been
+ * exhausted.
+ **/
+ HashTablePutResult put(const TypedValue &key,
+ const uint8_t &value);
+
+ /**
+ * @brief Add a new entry into the hash table (composite key version).
+ *
+ * @warning No component of the key may be null.
+ * @warning This method is threadsafe with regard to other calls to put(),
+ * putCompositeKey(), putValueAccessor(), and
+ * putValueAccessorCompositeKey(), but should not be used
+ * simultaneously with upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey().
+ * @note This version is for composite keys, see also put().
+ * @note If the hash table is (close to) full and resizable is true, this
+ * routine might result in rebuilding the entire hash table.
+ *
+ * @param key The components of the key.
+ * @param value The value payload.
+ * @return HashTablePutResult::kOK if an entry was successfully inserted,
+ * HashTablePutResult::kDuplicateKey if allow_duplicate_keys is false
+ * and key was a duplicate, or HashTablePutResult::kOutOfSpace if
+ * resizable is false and storage space for the hash table has been
+ * exhausted.
+ **/
+ HashTablePutResult putCompositeKey(const std::vector<TypedValue> &key,
+ const uint8_t &value);
+
+ HashTablePutResult putCompositeKeyFast(const std::vector<TypedValue> &key,
+ const uint8_t *value_ptr);
+
+ /**
+ * @brief Add (multiple) new entries into the hash table from a
+ * ValueAccessor.
+ *
+ * @warning This method is threadsafe with regard to other calls to put(),
+ * putCompositeKey(), putValueAccessor(), and
+ * putValueAccessorCompositeKey(), but should not be used
+ * simultaneously with upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey().
+ * @note This version is for scalar keys, see also
+ * putValueAccessorCompositeKey().
+ * @note If the hash table fills up while this call is in progress and
+ * resizable is true, this might result in rebuilding the entire hash
+ * table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before inserting it (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator that takes const ValueAccessor& as an argument (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as an argument) and returns either
+ * a ValueT or a reference to a ValueT. The functor should generate
+ * the appropriate mapped value for the current tuple the accessor is
+ * iterating on.
+ * @return HashTablePutResult::kOK if all keys and generated values from
+ * accessor were successfully inserted.
+ * HashTablePutResult::kOutOfSpace is returned if this hash-table is
+ * non-resizable and ran out of space (note that some entries may
+ * still have been inserted, and accessor's iteration will be left on
+ * the first tuple which could not be inserted).
+ * HashTablePutResult::kDuplicateKey is returned if
+ * allow_duplicate_keys is false and a duplicate key is encountered
+ * (as with HashTablePutResult::kOutOfSpace, some entries may have
+ * been inserted, and accessor will be left on the tuple with a
+ * duplicate key).
+ **/
+ template <typename FunctorT>
+ HashTablePutResult putValueAccessor(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor);
+
+ /**
+ * @brief Add (multiple) new entries into the hash table from a
+ * ValueAccessor (composite key version).
+ *
+ * @warning This method is threadsafe with regard to other calls to put(),
+ * putCompositeKey(), putValueAccessor(), and
+ * putValueAccessorCompositeKey(), but should not be used
+ * simultaneously with upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey().
+ * @note This version is for composite keys, see also putValueAccessor().
+ * @note If the hash table fills up while this call is in progress and
+ * resizable is true, this might result in rebuilding the entire hash
+ * table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_ids The attribute IDs of each key component to be read
+ * from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * has a null component before inserting it (null keys are skipped).
+ * This must be set to true if some of the keys that will be read from
+ * accessor may be null.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator that takes const ValueAccessor& as an argument (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as an argument) and returns either
+ * a ValueT or a reference to a ValueT. The functor should generate
+ * the appropriate mapped value for the current tuple the accessor is
+ * iterating on.
+ * @return HashTablePutResult::kOK if all keys and generated values from
+ * accessor were successfully inserted.
+ * HashTablePutResult::kOutOfSpace is returned if this hash-table is
+ * non-resizable and ran out of space (note that some entries may
+ * still have been inserted, and accessor's iteration will be left on
+ * the first tuple which could not be inserted).
+ * HashTablePutResult::kDuplicateKey is returned if
+ * allow_duplicate_keys is false and a duplicate key is encountered
+ * (as with HashTablePutResult::kOutOfSpace, some entries may have
+ * been inserted, and accessor will be left on the tuple with a
+ * duplicate key).
+ **/
+ template <typename FunctorT>
+ HashTablePutResult putValueAccessorCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor);
+
+ /**
+ * @brief Apply a functor to the value mapped to a key, first inserting a new
+ * value if one is not already present.
+ *
+ * @warning The key must not be null.
+ * @warning This method is only usable if allow_duplicate_keys is false.
+ * @warning This method is threadsafe with regard to other calls to upsert(),
+ * upsertCompositeKey(), upsertValueAccessor(), and
+ * upsertValueAccessorCompositeKey(), but should not be used
+ * simultaneously with put(), putCompositeKey(), putValueAccessor(),
+ * or putValueAccessorCompositeKey().
+ * @warning The ValueT* pointer passed to functor's call operator is only
+ * guaranteed to be valid for the duration of the call. The functor
+ * should not store a copy of the pointer and assume that it remains
+ * valid.
+ * @warning Although this method itself is threadsafe, the ValueT object
+ * accessed by functor is not guaranteed to be (although it is
+ * guaranteed that its initial insertion will be atomic). If it is
+ * possible for multiple threads to call upsert() with the same key
+ * at the same time, then their access to ValueT should be made
+ * threadsafe (e.g. with the use of atomic types, mutexes, or some
+ * other external synchronization).
+ * @note This version is for single scalar keys, see also
+ * upsertCompositeKey().
+ * @note If the hash table is (close to) full and resizable is true, this
+ * routine might result in rebuilding the entire hash table.
+ *
+ * @param key The key.
+ * @param initial_value If there was not already a preexisting entry in this
+ * HashTable for the specified key, then the value will be initialized
+ * with a copy of initial_value. This parameter is ignored if a value
+ * is already present for key.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator which takes ValueT* as an argument. The call operator will
+ * be invoked once on the value corresponding to key (which may be
+ * newly inserted and default-constructed).
+ * @return True on success, false if upsert failed because there was not
+ * enough space to insert a new entry in this HashTable.
+ **/
+ template <typename FunctorT>
+ bool upsert(const TypedValue &key,
+ const uint8_t &initial_value,
+ FunctorT *functor);
+
+ /**
+ * @brief Apply a functor to the value mapped to a key, first inserting a new
+ * value if one is not already present.
+ *
+ * @warning The key must not be null.
+ * @warning This method is only usable if allow_duplicate_keys is false.
+ * @warning This method is threadsafe with regard to other calls to upsert(),
+ * upsertCompositeKey(), upsertValueAccessor(), and
+ * upsertValueAccessorCompositeKey(), but should not be used
+ * simultaneously with put(), putCompositeKey(), putValueAccessor(),
+ * or putValueAccessorCompositeKey().
+ * @warning The ValueT* pointer passed to functor's call operator is only
+ * guaranteed to be valid for the duration of the call. The functor
+ * should not store a copy of the pointer and assume that it remains
+ * valid.
+ * @warning Although this method itself is threadsafe, the ValueT object
+ * accessed by functor is not guaranteed to be (although it is
+ * guaranteed that its initial insertion will be atomic). If it is
+ * possible for multiple threads to call upsertCompositeKey() with
+ * the same key at the same time, then their access to ValueT should
+ * be made threadsafe (e.g. with the use of atomic types, mutexes,
+ * or some other external synchronization).
+ * @note This version is for composite keys, see also upsert().
+ * @note If the hash table is (close to) full and resizable is true, this
+ * routine might result in rebuilding the entire hash table.
+ *
+ * @param key The key.
+ * @param initial_value If there was not already a preexisting entry in this
+ * HashTable for the specified key, then the value will be initialized
+ * with a copy of initial_value. This parameter is ignored if a value
+ * is already present for key.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator which takes ValueT* as an argument. The call operator will
+ * be invoked once on the value corresponding to key (which may be
+ * newly inserted and default-constructed).
+ * @return True on success, false if upsert failed because there was not
+ * enough space to insert a new entry in this HashTable.
+ **/
+ template <typename FunctorT>
+ bool upsertCompositeKey(const std::vector<TypedValue> &key,
+ const uint8_t &initial_value,
+ FunctorT *functor);
+
+
+ template <typename FunctorT>
+ bool upsertCompositeKeyFast(const std::vector<TypedValue> &key,
+ const uint8_t *init_value_ptr,
+ FunctorT *functor);
+
+ bool upsertCompositeKeyNewFast(const std::vector<TypedValue> &key,
+ const uint8_t *init_value_ptr,
+ const uint8_t *source_state);
+
+ /**
+ * @brief Apply a functor to (multiple) entries in this hash table, with keys
+ * drawn from a ValueAccessor. New values are first inserted if not
+ * already present.
+ *
+ * @warning This method is only usable if allow_duplicate_keys is false.
+ * @warning This method is threadsafe with regard to other calls to upsert(),
+ * upsertCompositeKey(), upsertValueAccessor(), and
+ * upsertValueAccessorCompositeKey(), but should not be used
+ * simultaneously with put(), putCompositeKey(), putValueAccessor(),
+ * or putValueAccessorCompositeKey().
+ * @warning The ValueAccessor reference and ValueT* pointer passed to
+ * functor's call operator are only guaranteed to be valid for the
+ * duration of the call. The functor should not store a copy of
+ * these pointers and assume that they remain valid.
+ * @warning Although this method itself is threadsafe, the ValueT object
+ * accessed by functor is not guaranteed to be (although it is
+ * guaranteed that its initial insertion will be atomic). If it is
+ * possible for multiple threads to call upsertValueAccessor() with
+ * the same key at the same time, then their access to ValueT should
+ * be made threadsafe (e.g. with the use of atomic types, mutexes,
+ * or some other external synchronization).
+ * @note This version is for single scalar keys, see also
+ * upsertValueAccessorCompositeKey().
+ * @note If the hash table is (close to) full and resizable is true, this
+ * routine might result in rebuilding the entire hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before upserting it (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator that takes two arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and ValueT*.
+ * The call operator will be invoked once for every tuple with a
+ * non-null key in accessor.
+ * @return True on success, false if upsert failed because there was not
+ * enough space to insert new entries for all the keys in accessor
+ * (note that some entries may still have been upserted, and
+ * accessor's iteration will be left on the first tuple which could
+ * not be inserted).
+ **/
+ template <typename FunctorT>
+ bool upsertValueAccessor(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ const uint8_t &initial_value,
+ FunctorT *functor);
+
+
+ bool upsertValueAccessorFast(const std::vector<std::vector<attribute_id>> &argument_ids,
+ ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys);
+
+ /**
+ * @brief Apply a functor to (multiple) entries in this hash table, with keys
+ * drawn from a ValueAccessor. New values are first inserted if not
+ * already present. Composite key version.
+ *
+ * @warning This method is only usable if allow_duplicate_keys is false.
+ * @warning This method is threadsafe with regard to other calls to upsert(),
+ * upsertCompositeKey(), upsertValueAccessor(), and
+ * upsertValueAccessorCompositeKey(), but should not be used
+ * simultaneously with put(), putCompositeKey(), putValueAccessor(),
+ * or putValueAccessorCompositeKey().
+ * @warning The ValueAccessor reference and ValueT* pointer passed to
+ * functor's call operator are only guaranteed to be valid for the
+ * duration of the call. The functor should not store a copy of
+ * these pointers and assume that they remain valid.
+ * @warning Although this method itself is threadsafe, the ValueT object
+ * accessed by functor is not guaranteed to be (although it is
+ * guaranteed that its initial insertion will be atomic). If it is
+ * possible for multiple threads to call upsertValueAccessor() with
+ * the same key at the same time, then their access to ValueT should
+ * be made threadsafe (e.g. with the use of atomic types, mutexes,
+ * or some other external synchronization).
+ * @note This version is for composite keys, see also upsertValueAccessor().
+ * @note If the hash table is (close to) full and resizable is true, this
+ * routine might result in rebuilding the entire hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_ids The attribute IDs of each key component to be read
+ * from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before upserting it (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator that takes two arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and ValueT*.
+ * The call operator will be invoked once for every tuple with a
+ * non-null key in accessor.
+ * @return True on success, false if upsert failed because there was not
+ * enough space to insert new entries for all the keys in accessor
+ * (note that some entries may still have been upserted, and
+ * accessor's iteration will be left on the first tuple which could
+ * not be inserted).
+ **/
+ template <typename FunctorT>
+ bool upsertValueAccessorCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ const uint8_t &initial_value,
+ FunctorT *functor);
+
+ bool upsertValueAccessorCompositeKeyFast(
+ const std::vector<std::vector<attribute_id>> &argument,
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys);
+
+ /**
+ * @brief Determine the number of entries (key-value pairs) contained in this
+ * HashTable.
+ * @note For some HashTable implementations, this is O(1), but for others it
+ * may be O(n) where n is the number of buckets.
+ *
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ *
+ * @return The number of entries in this HashTable.
+ **/
+ virtual std::size_t numEntries() const = 0;
+
+ /**
+ * @brief Lookup a key against this hash table to find a matching entry.
+ *
+ * @warning Only usable with the hash table that does not allow duplicate
+ * keys.
+ * @warning The key must not be null.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for single scalar keys. See also
+ * getSingleCompositeKey().
+ *
+ * @param key The key to look up.
+ * @return The value of a matched entry if a matching key is found.
+ * Otherwise, return NULL.
+ **/
+ virtual const uint8_t* getSingle(const TypedValue &key) const = 0;
+
+ /**
+ * @brief Lookup a composite key against this hash table to find a matching
+ * entry.
+ *
+ * @warning Only usable with the hash table that does not allow duplicate
+ * keys.
+ * @warning The key must not be null.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for composite keys. See also getSingle().
+ *
+ * @param key The key to look up.
+ * @return The value of a matched entry if a matching key is found.
+ * Otherwise, return NULL.
+ **/
+ virtual const uint8_t* getSingleCompositeKey(const std::vector<TypedValue> &key) const = 0;
+ virtual const uint8_t* getSingleCompositeKey(const std::vector<TypedValue> &key, int index) const = 0;
+
+ /**
+ * @brief Lookup a key against this hash table to find matching entries.
+ *
+ * @warning The key must not be null.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note It is more efficient to call getSingle() if the hash table does not
+ * allow duplicate keys.
+ * @note This version is for single scalar keys. See also
+ * getAllCompositeKey().
+ *
+ * @param key The key to look up.
+ * @param values A vector to hold values of all matching entries. Matches
+ * will be appended to the vector.
+ **/
+ virtual void getAll(const TypedValue &key, std::vector<const uint8_t*> *values) const = 0;
+
+ /**
+ * @brief Lookup a composite key against this hash table to find matching
+ * entries.
+ *
+ * @warning The key must not be null.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note It is more efficient to call getSingleCompositeKey() if the hash
+ * table does not allow duplicate keys.
+ * @note This version is for composite keys. See also getAll().
+ *
+ * @param key The key to look up.
+ * @param values A vector to hold values of all matching entries. Matches
+ * will be appended to the vector.
+ **/
+ virtual void getAllCompositeKey(const std::vector<TypedValue> &key,
+ std::vector<const uint8_t*> *values) const = 0;
+
+ /**
+ * @brief Lookup (multiple) keys from a ValueAccessor and apply a functor to
+ * the matching values.
+ *
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for single scalar keys. See also
+ * getAllFromValueAccessorCompositeKey().
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator that takes 2 arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and
+ * const ValueT&. The functor will be invoked once for each pair of a
+ * key taken from accessor and matching value.
+ **/
+ template <typename FunctorT>
+ void getAllFromValueAccessor(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ /**
+ * @brief Lookup (multiple) keys from a ValueAccessor, apply a functor to the
+ * matching values and additionally call a recordMatch() function of
+ * the functor when the first match for a key is found.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for single scalar keys. See also
+ * getAllFromValueAccessorCompositeKeyWithExtraWorkForFirstMatch().
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide two functions:
+ * 1) An operator that takes 2 arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and
+ * const ValueT&. The operator will be invoked once for each pair of a
+ * key taken from accessor and matching value.
+ * 2) A function hasMatch that takes 1 argument: const ValueAccessor&.
+ * The function will be called only once for a key from accessor when
+ * the first match is found.
+ */
+ template <typename FunctorT>
+ void getAllFromValueAccessorWithExtraWorkForFirstMatch(
+ ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ /**
+ * @brief Lookup (multiple) keys from a ValueAccessor, apply a functor to the
+ * matching values and additionally call a recordMatch() function of
+ * the functor when the first match for a key is found. Composite key
+ * version.
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor, which should provide two functions:
+ * 1) An operator that takes 2 arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and
+ * const ValueT&. The operator will be invoked once for each pair of a
+ * key taken from accessor and matching value.
+ * 2) A function hasMatch that takes 1 argument: const ValueAccessor&.
+ * The function will be called only once for a key from accessor when
+ * the first match is found.
+ */
+ template <typename FunctorT>
+ void getAllFromValueAccessorCompositeKeyWithExtraWorkForFirstMatch(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ /**
+ * @brief Lookup (multiple) keys from a ValueAccessor and apply a functor to
+ * the matching values. Composite key version.
+ *
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for composite keys. See also
+ * getAllFromValueAccessor().
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_ids The attribute IDs of each key component to be read
+ * from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * has a null component before inserting it (null keys are skipped).
+ * This must be set to true if some of the keys that will be read from
+ * accessor may be null.
+ * @param functor A pointer to a functor, which should provide a call
+ * operator that takes 2 arguments: const ValueAccessor& (or better
+ * yet, a templated call operator which takes a const reference to
+ * some subclass of ValueAccessor as its first argument) and
+ * const ValueT&. The functor will be invoked once for each pair of a
+ * key taken from accessor and matching value.
+ **/
+ template <typename FunctorT>
+ void getAllFromValueAccessorCompositeKey(ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ /**
+ * @brief Apply the functor to each key with a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is a match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchFound(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessor<true>(accessor,
+ key_attr_id,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
+ * @brief Apply the functor to each key with a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is a match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchFoundCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessorCompositeKey<true>(accessor,
+ key_attr_ids,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
+ * @brief Apply the functor to each key without a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is no match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchNotFound(
+ ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessor<false>(accessor,
+ key_attr_id,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
+ * @brief Apply the functor to each key without a match in the hash table.
+ *
+ * @param accessor A ValueAccessor which will be used to access keys.
+ * beginIteration() should be called on accessor before calling this
+ * method.
+ * @param key_attr_id The attribute ID of the keys to be read from accessor.
+ * @param check_for_null_keys If true, each key will be checked to see if it
+ * is null before looking it up (null keys are skipped). This must be
+ * set to true if some of the keys that will be read from accessor may
+ * be null.
+ * @param functor A pointer to a functor which should provide an operator that
+ * takes 1 argument: const ValueAccessor&. The operator will be called
+ * only once for a key from accessor if there is no match.
+ */
+ template <typename FunctorT>
+ void runOverKeysFromValueAccessorIfMatchNotFoundCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const {
+ return runOverKeysFromValueAccessorCompositeKey<false>(accessor,
+ key_attr_ids,
+ check_for_null_keys,
+ functor);
+ }
+
+ /**
+ * @brief Apply a functor to each key, value pair in this hash table.
+ *
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for single scalar keys. See also
+ * forEachCompositeKey().
+ *
+ * @param functor A pointer to a functor, which should provide a call
+ * operator which takes 2 arguments: const TypedValue&, const ValueT&.
+ * The call operator will be invoked once on each key, value pair in
+ * this hash table (note that if allow_duplicate_keys is true,
+ * the call may occur multiple times for the same key with different
+ * values).
+ * @return The number of key-value pairs visited.
+ **/
+ template <typename FunctorT>
+ std::size_t forEach(FunctorT *functor) const;
+
+ /**
+ * @brief Apply a functor to each key, value pair in this hash table.
+ *
+ * @warning This method assumes that no concurrent calls to put(),
+ * putCompositeKey(), putValueAccessor(),
+ * putValueAccessorCompositeKey(), upsert(), upsertCompositeKey(),
+ * upsertValueAccessor(), or upsertValueAccessorCompositeKey() are
+ * taking place (i.e. that this HashTable is immutable for the
+ * duration of the call and as long as the returned pointer may be
+ * dereferenced). Concurrent calls to getSingle(),
+ * getSingleCompositeKey(), getAll(), getAllCompositeKey(),
+ * getAllFromValueAccessor(), getAllFromValueAccessorCompositeKey(),
+ * forEach(), and forEachCompositeKey() are safe.
+ * @note This version is for composite keys. See also forEach().
+ *
+ * @param functor A pointer to a functor, which should provide a call
+ * operator which takes 2 arguments: const std::vector<TypedValue>&,
+ * const ValueT&. The call operator will be invoked once on each key,
+ * value pair in this hash table (note that if allow_duplicate_keys is
+ * true, the call may occur multiple times for the same key with
+ * different values).
+ * @return The number of key-value pairs visited.
+ **/
+ template <typename FunctorT>
+ std::size_t forEachCompositeKey(FunctorT *functor) const;
+
+ template <typename FunctorT>
+ std::size_t forEachCompositeKeyFast(FunctorT *functor) const;
+
+ template <typename FunctorT>
+ std::size_t forEachCompositeKeyFast(FunctorT *functor, int index) const;
+ /**
+ * @brief A call to this function will cause a bloom filter to be built
+ * during the build phase of this hash table.
+ **/
+ inline void enableBuildSideBloomFilter() {
+ has_build_side_bloom_filter_ = true;
+ }
+
+ /**
+ * @brief A call to this function will cause a set of bloom filters to be
+ * probed during the probe phase of this hash table.
+ **/
+ inline void enableProbeSideBloomFilter() {
+ has_probe_side_bloom_filter_ = true;
+ }
+
+ /**
+ * @brief This function sets the pointer to the bloom filter to be
+ * used during the build phase of this hash table.
+ * @warning Should call enable_build_side_bloom_filter() first to enable
+ * bloom filter usage during build phase.
+ * @note The ownership of the bloom filter lies with the caller.
+ *
+ * @param bloom_filter The pointer to the bloom filter.
+ **/
+ inline void setBuildSideBloomFilter(BloomFilter *bloom_filter) {
+ build_bloom_filter_ = bloom_filter;
+ }
+
+ /**
+ * @brief This function adds a pointer to the list of bloom filters to be
+ * used during the probe phase of this hash table.
+ * @warning Should call enable_probe_side_bloom_filter() first to enable
+ * bloom filter usage during probe phase.
+ * @note The ownership of the bloom filter lies with the caller.
+ *
+ * @param bloom_filter The pointer to the bloom filter.
+ **/
+ inline void addProbeSideBloomFilter(const BloomFilter *bloom_filter) {
+ probe_bloom_filters_.emplace_back(bloom_filter);
+ }
+
+ /**
+ * @brief This function adds a vector of attribute ids corresponding to a
+ * bloom filter used during the probe phase of this hash table.
+ * @warning Should call enable_probe_side_bloom_filter() first to enable
+ * bloom filter usage during probe phase.
+ *
+ * @param probe_attribute_ids The vector of attribute ids to use for probing
+ * the bloom filter.
+ **/
+ inline void addProbeSideAttributeIds(std::vector<attribute_id> &&probe_attribute_ids) {
+ probe_attribute_ids_.push_back(probe_attribute_ids);
+ }
+
+ protected:
+ /**
+ * @brief Constructor for new resizable hash table.
+ *
+ * @param key_types A vector of one or more types (>1 indicates a composite
+ * key).
+ * @param num_entries The estimated number of entries this hash table will
+ * hold.
+ * @param storage_manager The StorageManager to use (a StorageBlob will be
+ * allocated to hold this hash table's contents).
+ * @param adjust_hashes If true, the hash of a key should be modified by
+ * applying AdjustHash() so that it does not collide with one of the
+ * special values kEmptyHash or kPendingHash. If false, the hash is
+ * used as-is.
+ * @param use_scalar_literal_hash If true, the key is a single scalar literal
+ * (non-composite) that it is safe to use the simplified hash function
+ * TypedValue::getHashScalarLiteral() on. If false, the generic
+ * TypedValue::getHash() method will be used.
+ * @param preallocate_supported If true, this HashTable overrides
+ * preallocateForBulkInsert() to allow bulk-allocation of resources
+ * (i.e. buckets and variable-length key storage) in a single up-front
+ * pass when bulk-inserting entries. If false, resources are allocated
+ * on the fly for each entry.
+ **/
+ FastHashTable(const std::vector<const Type*> &key_types,
+ const std::size_t num_entries,
+ const std::vector<AggregationHandle *> &handles,
+ const std::vector<std::size_t> &payload_sizes,
+ StorageManager *storage_manager,
+ const bool adjust_hashes,
+ const bool use_scalar_literal_hash,
+ const bool preallocate_supported)
+ : key_types_(key_types),
+ scalar_key_inline_(true),
+ key_inline_(nullptr),
+ adjust_hashes_(adjust_hashes),
+ use_scalar_literal_hash_(use_scalar_literal_hash),
+ preallocate_supported_(preallocate_supported),
+ handles_(handles),
+ total_payload_size_(std::accumulate(payload_sizes.begin(), payload_sizes.end(), sizeof(SpinMutex))),
+ storage_manager_(storage_manager),
+ hash_table_memory_(nullptr),
+ hash_table_memory_size_(0) {
+ DEBUG_ASSERT(resizable);
+ std::size_t running_sum = sizeof(SpinMutex);
+ for (auto size : payload_sizes) {
+ payload_offsets_.emplace_back(running_sum);
+ running_sum+=size;
+ }
+ }
+
+ /**
+ * @brief Constructor for non-resizable hash table.
+ *
+ * @param key_types A vector of one or more types (>1 indicates a composite
+ * key).
+ * @param hash_table_memory A pointer to memory to use for this hash table.
+ * @param hash_table_memory_size The size of hash_table_memory in bytes.
+ * @param new_hash_table If true, this hash table is being constructed for
+ * the first time and hash_table_memory will be cleared. If false,
+ * reload a pre-existing hash table.
+ * @param hash_table_memory_zeroed If new_hash_table is true, setting this to
+ * true means that this HashTable will assume that hash_table_memory
+ * has already been zeroed-out (any newly-allocated block or blob
+ * memory from StorageManager is zeroed-out). If false, this HashTable
+ * will explicitly zero-fill its memory as neccessary. This parameter
+ * has no effect when new_hash_table is false.
+ * @param adjust_hashes If true, the hash of a key should be modified by
+ * applying AdjustHash() so that it does not collide with one of the
+ * special values kEmptyHash or kPendingHash. If false, the hash is
+ * used as-is.
+ * @param use_scalar_literal_hash If true, the key is a single scalar literal
+ * (non-composite) that it is safe to use the simplified hash function
+ * TypedValue::getHashScalarLiteral() on. If false, the generic
+ * TypedValue::getHash() method will be used.
+ * @param preallocate_supported If true, this HashTable overrides
+ * preallocateForBulkInsert() to allow bulk-allocation of resources
+ * (i.e. buckets and variable-length key storage) in a single up-front
+ * pass when bulk-inserting entries. If false, resources are allocated
+ * on the fly for each entry.
+ **/
+ FastHashTable(const std::vector<const Type*> &key_types,
+ void *hash_table_memory,
+ const std::size_t hash_table_memory_size,
+ const bool new_hash_table,
+ const bool hash_table_memory_zeroed,
+ const bool adjust_hashes,
+ const bool use_scalar_literal_hash,
+ const bool preallocate_supported)
+ : key_types_(key_types),
+ scalar_key_inline_(true),
+ key_inline_(nullptr),
+ adjust_hashes_(adjust_hashes),
+ use_scalar_literal_hash_(use_scalar_literal_hash),
+ preallocate_supported_(preallocate_supported),
+ storage_manager_(nullptr),
+ hash_table_memory_(hash_table_memory),
+ hash_table_memory_size_(hash_table_memory_size) {
+ DEBUG_ASSERT(!resizable);
+ }
+
+ // Adjust 'hash' so that it is not exactly equal to either of the special
+ // values kEmptyHash or kPendingHash.
+ inline constexpr static std::size_t AdjustHash(const std::size_t hash) {
+ return hash + (hash == kEmptyHash) - (hash == kPendingHash);
+ }
+
+ // Set information about which key components are stored inline. This usually
+ // comes from a HashTableKeyManager, and is set by the constructor of a
+ // subclass of HashTable.
+ inline void setKeyInline(const std::vector<bool> *key_inline) {
+ scalar_key_inline_ = key_inline->front();
+ key_inline_ = key_inline;
+ }
+
+ // Generate a hash for a composite key by hashing each component of 'key' and
+ // mixing their bits with CombineHashes().
+ inline std::size_t hashCompositeKey(const std::vector<TypedValue> &key) const;
+
+ // If 'force_key_copy' is true and some part of a composite key is
+ // variable-length, calculate the total number of bytes for variable-length
+ // key components that need to be copied. Otherwise, return 0 to indicate
+ // that no variable-length copy is required.
+ inline std::size_t calculateVariableLengthCompositeKeyCopySize(
+ const std::vector<TypedValue> &key) const;
+
+ // Helpers for put. If this HashTable is resizable, 'resize_shared_mutex_'
+ // should be locked in shared mode before calling either of these methods.
+ virtual HashTablePutResult putInternal(const TypedValue &key,
+ const std::size_t variable_key_size,
+ const uint8_t &value,
+ HashTablePreallocationState *prealloc_state) = 0;
+ virtual HashTablePutResult putCompositeKeyInternal(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const uint8_t &value,
+ HashTablePreallocationState *prealloc_state) = 0;
+
+ virtual HashTablePutResult putCompositeKeyInternalFast(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const std::uint8_t *init_value_ptr,
+ HashTablePreallocationState *prealloc_state) = 0;
+
+
+ // Helpers for upsert. Both return a pointer to the value corresponding to
+ // 'key'. If this HashTable is resizable, 'resize_shared_mutex_' should be
+ // locked in shared mode while calling and using the returned pointer. May
+ // return NULL if there is not enough space to insert a new key, in which
+ // case a resizable HashTable should release the 'resize_shared_mutex_' and
+ // call resize(), then try again.
+ virtual uint8_t* upsertInternal(const TypedValue &key,
+ const std::size_t variable_key_size,
+ const uint8_t &initial_value) = 0;
+ virtual uint8_t* upsertInternalFast(const TypedValue &key,
+ const std::uint8_t *init_value_ptr,
+ const std::size_t variable_key_size) = 0;
+ virtual uint8_t* upsertCompositeKeyInternal(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const uint8_t &initial_value) = 0;
+
+ virtual uint8_t* upsertCompositeKeyInternalFast(const std::vector<TypedValue> &key,
+ const std::uint8_t *init_value_ptr,
+ const std::size_t variable_key_size) = 0;
+
+ // Helpers for forEach. Each return true on success, false if no more entries
+ // exist to iterate over. After a successful call, '*key' is overwritten with
+ // the key of the next entry, '*value' points to the associated value, and
+ // '*entry_num' is incremented to the next (implementation defined) entry to
+ // check ('*entry_num' should initially be set to zero).
+ virtual bool getNextEntry(TypedValue *key,
+ const uint8_t **value,
+ std::size_t *entry_num) const = 0;
+ virtual bool getNextEntryCompositeKey(std::vector<TypedValue> *key,
+ const uint8_t **value,
+ std::size_t *entry_num) const = 0;
+
+ // Helpers for getAllFromValueAccessor. Each return true on success, false if
+ // no more entries exist for the specified key. After a successful call,
+ // '*value' points to the associated value, and '*entry_num' is incremented
+ // to the next (implementation defined) entry to check ('*entry_num' should
+ // initially be set to zero).
+ virtual bool getNextEntryForKey(const TypedValue &key,
+ const std::size_t hash_code,
+ const uint8_t **value,
+ std::size_t *entry_num) const = 0;
+ virtual bool getNextEntryForCompositeKey(const std::vector<TypedValue> &key,
+ const std::size_t hash_code,
+ const uint8_t **value,
+ std::size_t *entry_num) const = 0;
+
+ // Return true if key exists in the hash table.
+ virtual bool hasKey(const TypedValue &key) const = 0;
+ virtual bool hasCompositeKey(const std::vector<TypedValue> &key) const = 0;
+
+ // For a resizable HashTable, grow to accomodate more entries. If
+ // 'extra_buckets' is not zero, it may serve as a "hint" to implementations
+ // that at least the requested number of extra buckets are required when
+ // resizing (mainly used in putValueAccessor() and
+ // putValueAccessorCompositeKey() when 'preallocate_supported_' is true).
+ // Implementations are free to ignore 'extra_buckets'. If
+ // 'extra_variable_storage' is not zero, implementations will attempt to
+ // allocate at least enough additional variable-key storage space to
+ // accomodate the number of bytes specified. 'retry_num' is intended ONLY for
+ // when resize() recursively calls itself and should not be set to nonzero by
+ // any other caller.
+ virtual void resize(const std::size_t extra_buckets,
+ const std::size_t extra_variable_storage,
+ const std::size_t retry_num = 0) = 0;
+
+ // In the case where 'allow_duplicate_keys' is true, it is possible to
+ // pre-calculate the number of key-value entries and the amount of
+ // variable-length key storage that will be needed to insert all the
+ // entries from a ValueAccessor in putValueAccessor() or
+ // putValueAccessorCompositeKey() before actually inserting anything. Some
+ // HashTable implemetations (notably SeparateChainingHashTable) can achieve
+ // better performance by ammortizing the cost of allocating certain resources
+ // (buckets and variable-length key storage) in one up-front allocation. This
+ // method is intended to support that. Returns true and fills in
+ // '*prealloc_state' if pre-allocation was successful. Returns false if a
+ // resize() is needed.
+ virtual bool preallocateForBulkInsert(const std::size_t total_entries,
+ const std::size_t total_variable_key_size,
+ HashTablePreallocationState *prealloc_state) {
+ FATAL_ERROR("Called HashTable::preallocateForBulkInsert() on a HashTable "
+ "implementation that does not support preallocation.");
+ }
+
+ // Type(s) of keys.
+ const std::vector<const Type*> key_types_;
+
+ // Information about whether key components are stored inline or in a
+ // separate variable-length storage region. This is usually determined by a
+ // HashTableKeyManager and set by calling setKeyInline().
+ bool scalar_key_inline_;
+ const std::vector<bool> *key_inline_;
+
+ // Whether hashes should be adjusted by AdjustHash() before being used.
+ const bool adjust_hashes_;
+ // Whether it is safe to use the simplified TypedValue::getHashScalarLiteral()
+ // method instead of the generic TypedValue::getHash() method.
+ const bool use_scalar_literal_hash_;
+ // Whether preallocateForBulkInsert() is supported by this HashTable.
+ const bool preallocate_supported_;
+
+ const std::vector<AggregationHandle *> handles_;
+ const std::size_t total_payload_size_;
+ std::vector<std::size_t> payload_offsets_;
+
+ // Used only when resizable is true:
+ StorageManager *storage_manager_;
+ MutableBlobReference blob_;
+ // Locked in shared mode for most operations, exclusive mode during resize.
+ // Not locked at all for non-resizable HashTables.
+ alignas(kCacheLineBytes) SpinSharedMutex<true> resize_shared_mutex_;
+
+ // Used only when resizable is false:
+ void *hash_table_memory_;
+ const std::size_t hash_table_memory_size_;
+virtual size_t get_buckets_allocated() const {return 0;}
+
+ private:
+ // Assign '*key_vector' with the attribute values specified by 'key_attr_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 <typename ValueAccessorT>
+ inline static bool GetCompositeKeyFromValueAccessor(
+ const ValueAccessorT &accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ std::vector<TypedValue> *key_vector) {
+ for (std::vector<attribute_id>::size_type key_idx = 0;
+ key_idx < key_attr_ids.size();
+ ++key_idx) {
+ (*key_vector)[key_idx] = accessor.getTypedValue(key_attr_ids[key_idx]);
+ if (check_for_null_keys && (*key_vector)[key_idx].isNull()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ // If run_if_match_found is true, apply the functor to each key if a match is
+ // found; otherwise, apply the functor if no match is found.
+ template <bool run_if_match_found, typename FunctorT>
+ void runOverKeysFromValueAccessor(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ template <bool run_if_match_found, typename FunctorT>
+ void runOverKeysFromValueAccessorCompositeKey(
+ ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) const;
+
+ // Method containing the actual logic implementing getAllFromValueAccessor().
+ // Has extra template parameters that control behavior to avoid some
+ // inner-loop branching.
+ template <typename FunctorT,
+ bool check_for_null_keys,
+ bool adjust_hashes_template,
+ bool use_scalar_literal_hash_template>
+ void getAllFromValueAccessorImpl(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ FunctorT *functor) const;
+
+ // Data structures used for bloom filter optimized semi-joins.
+ bool has_build_side_bloom_filter_ = false;
+ bool has_probe_side_bloom_filter_ = false;
+ BloomFilter *build_bloom_filter_;
+ std::vector<const BloomFilter*> probe_bloom_filters_;
+ std::vector<std::vector<attribute_id>> probe_attribute_ids_;
+ DISALLOW_COPY_AND_ASSIGN(FastHashTable);
+};
+
+
+/**
+ * @brief An instantiation of the HashTable template for use in aggregations.
+ * @note This has force_key_copy = true, so that we don't have dangling pointers
+ * to blocks that are evicted.
+ **/
+using AggregationStateFastHashTable = FastHashTable<true, false, true, false>;
+
+/** @} */
+
+// ----------------------------------------------------------------------------
+// Implementations of template class methods follow.
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+HashTablePutResult FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::put(const TypedValue &key,
+ const uint8_t &value) {
+ const std::size_t variable_size = (force_key_copy && !scalar_key_inline_) ? key.getDataSize()
+ : 0;
+ if (resizable) {
+ HashTablePutResult result = HashTablePutResult::kOutOfSpace;
+ while (result == HashTablePutResult::kOutOfSpace) {
+ {
+ SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+ result = putInternal(key, variable_size, value, nullptr);
+ }
+ if (result == HashTablePutResult::kOutOfSpace) {
+ resize(0, variable_size);
+ }
+ }
+ return result;
+ } else {
+ return putInternal(key, variable_size, value, nullptr);
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+HashTablePutResult FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::putCompositeKey(const std::vector<TypedValue> &key,
+ const uint8_t& value) {
+ const std::size_t variable_size = calculateVariableLengthCompositeKeyCopySize(key);
+ if (resizable) {
+ HashTablePutResult result = HashTablePutResult::kOutOfSpace;
+ while (result == HashTablePutResult::kOutOfSpace) {
+ {
+ SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+ result = putCompositeKeyInternal(key, variable_size, value, nullptr);
+ }
+ if (result == HashTablePutResult::kOutOfSpace) {
+ resize(0, variable_size);
+ }
+ }
+ return result;
+ } else {
+ return putCompositeKeyInternal(key, variable_size, value, nullptr);
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+HashTablePutResult FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::putCompositeKeyFast(const std::vector<TypedValue> &key,
+ const std::uint8_t* init_value_ptr) {
+ const std::size_t variable_size = calculateVariableLengthCompositeKeyCopySize(key);
+ if (resizable) {
+ HashTablePutResult result = HashTablePutResult::kOutOfSpace;
+ while (result == HashTablePutResult::kOutOfSpace) {
+ {
+ SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+ result = putCompositeKeyInternalFast(key, variable_size, init_value_ptr, nullptr);
+ }
+ if (result == HashTablePutResult::kOutOfSpace) {
+ resize(0, variable_size);
+ }
+ }
+ return result;
+ } else {
+ return putCompositeKeyInternalFast(key, variable_size, init_value_ptr, nullptr);
+ }
+}
+
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <typename FunctorT>
+HashTablePutResult FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::putValueAccessor(ValueAccessor *accessor,
+ const attribute_id key_attr_id,
+ const bool check_for_null_keys,
+ FunctorT *functor) {
+ HashTablePutResult result = HashTablePutResult::kOutOfSpace;
+ std::size_t variable_size;
+ HashTablePreallocationState prealloc_state;
+ bool using_prealloc = allow_duplicate_keys && preallocate_supported_;
+ return InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> HashTablePutResult { // NOLINT(build/c++11)
+ if (using_prealloc) {
+ std::size_t total_entries = 0;
+ std::size_t total_variable_key_size = 0;
+ if (check_for_null_keys || (force_key_copy && !scalar_key_inline_)) {
+ // If we need to filter out nulls OR make variable copies, make a
+ // prepass over the ValueAccessor.
+ while (accessor->next()) {
+ TypedValue key = accessor->getTypedValue(key_attr_id);
+ if (check_for_null_keys && key.isNull()) {
+ continue;
+ }
+ ++total_entries;
+ total_variable_key_size += (force_key_copy && !scalar_key_inline_) ? key.getDataSize() : 0;
+ }
+ accessor->beginIteration();
+ } else {
+ total_entries = accessor->getNumTuples();
+ }
+ if (resizable) {
+ bool prealloc_succeeded = false;
+ while (!prealloc_succeeded) {
+ {
+ SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+ prealloc_succeeded = this->preallocateForBulkInsert(total_entries,
+ total_variable_key_size,
+ &prealloc_state);
+ }
+ if (!prealloc_succeeded) {
+ this->resize(total_entries, total_variable_key_size);
+ }
+ }
+ } else {
+ using_prealloc = this->preallocateForBulkInsert(total_entries,
+ total_variable_key_size,
+ &prealloc_state);
+ }
+ }
+ std::unique_ptr<BloomFilter> thread_local_bloom_filter;
+ if (has_build_side_bloom_filter_) {
+ thread_local_bloom_filter.reset(new BloomFilter(build_bloom_filter_->getRandomSeed(),
+ build_bloom_filter_->getNumberOfHashes(),
+ build_bloom_filter_->getBitArraySize()));
+ }
+ if (resizable) {
+ while (result == HashTablePutResult::kOutOfSpace) {
+ {
+ result = HashTablePutResult::kOK;
+ SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+ while (accessor->next()) {
+ TypedValue key = accessor->getTypedValue(key_attr_id);
+ if (check_for_null_keys && key.isNull()) {
+ continue;
+ }
+ variable_size = (force_key_copy && !scalar_key_inline_) ? key.getDataSize() : 0;
+ result = this->putInternal(key,
+ variable_size,
+ (*functor)(*accessor),
+ using_prealloc ? &prealloc_state : nullptr);
+ // Insert into bloom filter, if enabled.
+ if (has_build_side_bloom_filter_) {
+ thread_local_bloom_filter->insertUnSafe(static_cast<const std::uint8_t *>(key.getDataPtr()),
+ key.getDataSize());
+ }
+ if (result == HashTablePutResult::kDuplicateKey) {
+ DEBUG_ASSERT(!using_prealloc);
+ return result;
+ } else if (result == HashTablePutResult::kOutOfSpace) {
+ DEBUG_ASSERT(!using_prealloc);
+ break;
+ }
+ }
+ }
+ if (result == HashTablePutResult::kOutOfSpace) {
+ this->resize(0, variable_size);
+ accessor->previous();
+ }
+ }
+ } else {
+ while (accessor->next()) {
+ TypedValue key = accessor->getTypedValue(key_attr_id);
+ if (check_for_null_keys && key.isNull()) {
+ continue;
+ }
+ variable_size = (force_key_copy && !scalar_key_inline_) ? key.getDataSize() : 0;
+ result = this->putInternal(key,
+ variable_size,
+ (*functor)(*accessor),
+ using_prealloc ? &prealloc_state : nullptr);
+ // Insert into bloom filter, if enabled.
+ if (has_build_side_bloom_filter_) {
+ thread_local_bloom_filter->insertUnSafe(static_cast<const std::uint8_t *>(key.getDataPtr()),
+ key.getDataSize());
+ }
+ if (result != HashTablePutResult::kOK) {
+ return result;
+ }
+ }
+ }
+ // Update the build side bloom filter with thread local copy, if available.
+ if (has_build_side_bloom_filter_) {
+ build_bloom_filter_->bitwiseOr(thread_local_bloom_filter.get());
+ }
+
+ return HashTablePutResult::kOK;
+ });
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <typename FunctorT>
+HashTablePutResult FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::putValueAccessorCompositeKey(ValueAccessor *accessor,
+ const std::vector<attribute_id> &key_attr_ids,
+ const bool check_for_null_keys,
+ FunctorT *functor) {
+ DEBUG_ASSERT(key_types_.size() == key_attr_ids.size());
+ HashTablePutResult result = HashTablePutResult::kOutOfSpace;
+ std::size_t variable_size;
+ HashTablePreallocationState prealloc_state;
+ bool using_prealloc = allow_duplicate_keys && preallocate_supported_;
+ std::vector<TypedValue> key_vector;
+ key_vector.resize(key_attr_ids.size());
+ return InvokeOnAnyValueAccessor(
+ accessor,
+ [&](auto *accessor) -> HashTablePutResult { // NOLINT(build/c++11)
+ if (using_prealloc) {
+ std::size_t total_entries = 0;
+ std::size_t total_variable_key_size = 0;
+ if (check_for_null_keys || force_key_copy) {
+ // If we need to filter out nulls OR make variable copies, make a
+ // prepass over the ValueAccessor.
+ while (accessor->next()) {
+ if (this->GetCompositeKeyFromValueAccessor(*accessor,
+ key_attr_ids,
+ check_for_null_keys,
+ &key_vector)) {
+ continue;
+ }
+ ++total_entries;
+ total_variable_key_size += this->calculateVariableLengthCompositeKeyCopySize(key_vector);
+ }
+ accessor->beginIteration();
+ } else {
+ total_entries = accessor->getNumTuples();
+ }
+ if (resizable) {
+ bool prealloc_succeeded = false;
+ while (!prealloc_succeeded) {
+ {
+ SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+ prealloc_succeeded = this->preallocateForBulkInsert(total_entries,
+ total_variable_key_size,
+ &prealloc_state);
+ }
+ if (!prealloc_succeeded) {
+ this->resize(total_entries, total_variable_key_size);
+ }
+ }
+ } else {
+ using_prealloc = this->preallocateForBulkInsert(total_entries,
+ total_variable_key_size,
+ &prealloc_state);
+ }
+ }
+ if (resizable) {
+ while (result == HashTablePutResult::kOutOfSpace) {
+ {
+ result = HashTablePutResult::kOK;
+ SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+ while (accessor->next()) {
+ if (this->GetCompositeKeyFromValueAccessor(*accessor,
+ key_attr_ids,
+ check_for_null_keys,
+ &key_vector)) {
+ continue;
+ }
+ variable_size = this->calculateVariableLengthCompositeKeyCopySize(key_vector);
+ result = this->putCompositeKeyInternal(key_vector,
+ variable_size,
+ (*functor)(*accessor),
+ using_prealloc ? &prealloc_state : nullptr);
+ if (result == HashTablePutResult::kDuplicateKey) {
+ DEBUG_ASSERT(!using_prealloc);
+ return result;
+ } else if (result == HashTablePutResult::kOutOfSpace) {
+ DEBUG_ASSERT(!using_prealloc);
+ break;
+ }
+ }
+ }
+ if (result == HashTablePutResult::kOutOfSpace) {
+ this->resize(0, variable_size);
+ accessor->previous();
+ }
+ }
+ } else {
+ while (accessor->next()) {
+ if (this->GetCompositeKeyFromValueAccessor(*accessor,
+ key_attr_ids,
+ check_for_null_keys,
+ &key_vector)) {
+ continue;
+ }
+ variable_size = this->calculateVariableLengthCompositeKeyCopySize(key_vector);
+ result = this->putCompositeKeyInternal(key_vector,
+ variable_size,
+ (*functor)(*accessor),
+ using_prealloc ? &prealloc_state : nullptr);
+ if (result != HashTablePutResult::kOK) {
+ return result;
+ }
+ }
+ }
+
+ return HashTablePutResult::kOK;
+ });
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <typename FunctorT>
+bool FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::upsert(const TypedValue &key,
+ const uint8_t &initial_value,
+ FunctorT *functor) {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ const std::size_t variable_size = (force_key_copy && !scalar_key_inline_) ? key.getDataSize() : 0;
+ if (resizable) {
+ for (;;) {
+ {
+ SpinSharedMutexSharedLock<true> resize_lock(resize_shared_mutex_);
+ uint8_t *value = upsertInternal(key, variable_size, initial_value);
+ if (value != nullptr) {
+ (*functor)(value);
+ return true;
+ }
+ }
+ resize(0, force_key_copy && !scalar_key_inline_ ? key.getDataSize() : 0);
+ }
+ } else {
+ uint8_t *value = upsertInternal(key, variable_size, initial_value);
+ if (value == nullptr) {
+ return false;
+ } else {
+ (*functor)(value);
+ return true;
+ }
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+template <typename FunctorT>
+bool FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::upsertCompositeKey(const std::vector<TypedValue> &key,
+ const uint8_t &initial_value,
+ FunctorT *functor) {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ const std::size_t variable_size = calculateVariableLengthCompositeKeyCopySize(key);
+ if (resizable) {
+ for (;;) {
+ {
+ SpinSharedMutexSharedLock<true> resize_lock(resize_shared_mutex_);
+ uint8_t *value = upsertCompositeKeyInternal(key, variable_size, initial_value);
+ if (value
<TRUNCATED>
[2/5] incubator-quickstep git commit: Initial commit for QUICKSTEP-28
and QUICKSTEP-29. Code refactoring and cleanup,
some more optimizations are pending.
Posted by ra...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/FastSeparateChainingHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/FastSeparateChainingHashTable.hpp b/storage/FastSeparateChainingHashTable.hpp
new file mode 100644
index 0000000..64c4979
--- /dev/null
+++ b/storage/FastSeparateChainingHashTable.hpp
@@ -0,0 +1,1761 @@
+/**
+ * Copyright 2011-2015 Quickstep Technologies LLC.
+ * Copyright 2015-2016 Pivotal Software, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+#ifndef QUICKSTEP_STORAGE_FAST_SEPARATE_CHAINING_HASH_TABLE_HPP_
+#define QUICKSTEP_STORAGE_FAST_SEPARATE_CHAINING_HASH_TABLE_HPP_
+
+#include <algorithm>
+#include <atomic>
+#include <cstddef>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "storage/HashTable.hpp"
+#include "storage/FastHashTable.hpp"
+#include "storage/HashTableBase.hpp"
+#include "storage/HashTableKeyManager.hpp"
+#include "storage/StorageBlob.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "storage/StorageConstants.hpp"
+#include "storage/StorageManager.hpp"
+#include "threading/SpinSharedMutex.hpp"
+#include "types/Type.hpp"
+#include "types/TypedValue.hpp"
+#include "utility/Alignment.hpp"
+#include "utility/Macros.hpp"
+#include "utility/PrimeNumber.hpp"
+
+namespace quickstep {
+
+/** \addtogroup Storage
+ * @{
+ */
+
+/**
+ * @brief A hash table implementation which uses separate chaining for buckets.
+ **/
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+class FastSeparateChainingHashTable : public FastHashTable<resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys> {
+ public:
+ FastSeparateChainingHashTable(const std::vector<const Type*> &key_types,
+ const std::size_t num_entries,
+ const std::vector<std::size_t> &payload_sizes,
+ const std::vector<AggregationHandle *> &handles,
+ StorageManager *storage_manager);
+
+ FastSeparateChainingHashTable(const std::vector<const Type*> &key_types,
+ void *hash_table_memory,
+ const std::size_t hash_table_memory_size,
+ const bool new_hash_table,
+ const bool hash_table_memory_zeroed);
+
+ // Delegating constructors for single scalar keys.
+ FastSeparateChainingHashTable(const Type &key_type,
+ const std::size_t num_entries,
+ StorageManager *storage_manager)
+ : FastSeparateChainingHashTable(std::vector<const Type*>(1, &key_type),
+ num_entries,
+ storage_manager) {
+ }
+
+ FastSeparateChainingHashTable(const Type &key_type,
+ void *hash_table_memory,
+ const std::size_t hash_table_memory_size,
+ const bool new_hash_table,
+ const bool hash_table_memory_zeroed)
+ : FastSeparateChainingHashTable(std::vector<const Type*>(1, &key_type),
+ hash_table_memory,
+ hash_table_memory_size,
+ new_hash_table,
+ hash_table_memory_zeroed) {
+ }
+
+ ~FastSeparateChainingHashTable() override {
+ DestroyValues(buckets_,
+ header_->buckets_allocated.load(std::memory_order_relaxed),
+ bucket_size_);
+ std::free(init_payload_);
+ }
+
+ void clear() override;
+
+ std::size_t numEntries() const override {
+ return header_->buckets_allocated.load(std::memory_order_relaxed);
+ }
+
+ const uint8_t* getSingle(const TypedValue &key) const override;
+ const uint8_t* getSingleCompositeKey(const std::vector<TypedValue> &key) const override;
+ const uint8_t* getSingleCompositeKey(const std::vector<TypedValue> &key, int index) const override;
+
+ void getAll(const TypedValue &key,
+ std::vector<const uint8_t*> *values) const override;
+ void getAllCompositeKey(const std::vector<TypedValue> &key,
+ std::vector<const uint8_t*> *values) const override;
+
+ protected:
+ HashTablePutResult putInternal(const TypedValue &key,
+ const std::size_t variable_key_size,
+ const uint8_t &value,
+ HashTablePreallocationState *prealloc_state) override;
+ HashTablePutResult putCompositeKeyInternal(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const uint8_t &value,
+ HashTablePreallocationState *prealloc_state) override;
+ HashTablePutResult putCompositeKeyInternalFast(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const std::uint8_t *init_value_ptr,
+ HashTablePreallocationState *prealloc_state) override;
+
+ uint8_t* upsertInternal(const TypedValue &key,
+ const std::size_t variable_key_size,
+ const uint8_t &initial_value) override;
+ uint8_t* upsertInternalFast(const TypedValue &key,
+ const std::uint8_t *init_value_ptr,
+ const std::size_t variable_key_size) override;
+
+ uint8_t* upsertCompositeKeyInternal(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const uint8_t &initial_value) override;
+
+ uint8_t* upsertCompositeKeyInternalFast(const std::vector<TypedValue> &key,
+ const std::uint8_t *init_value_ptr,
+ const std::size_t variable_key_size) override;
+
+ bool getNextEntry(TypedValue *key,
+ const uint8_t **value,
+ std::size_t *entry_num) const override;
+ bool getNextEntryCompositeKey(std::vector<TypedValue> *key,
+ const uint8_t **value,
+ std::size_t *entry_num) const override;
+
+ bool getNextEntryForKey(const TypedValue &key,
+ const std::size_t hash_code,
+ const uint8_t **value,
+ std::size_t *entry_num) const override;
+ bool getNextEntryForCompositeKey(const std::vector<TypedValue> &key,
+ const std::size_t hash_code,
+ const uint8_t **value,
+ std::size_t *entry_num) const override;
+
+ bool hasKey(const TypedValue &key) const override;
+ bool hasCompositeKey(const std::vector<TypedValue> &key) const override;
+
+ void resize(const std::size_t extra_buckets,
+ const std::size_t extra_variable_storage,
+ const std::size_t retry_num = 0) override;
+
+ bool preallocateForBulkInsert(const std::size_t total_entries,
+ const std::size_t total_variable_key_size,
+ HashTablePreallocationState *prealloc_state) override;
+
+ size_t get_buckets_allocated() const override {return header_->buckets_allocated;}
+
+ private:
+ struct Header {
+ std::size_t num_slots;
+ std::size_t num_buckets;
+ alignas(kCacheLineBytes)
+ std::atomic<std::size_t> buckets_allocated;
+ alignas(kCacheLineBytes)
+ std::atomic<std::size_t> variable_length_bytes_allocated;
+ };
+
+ std::uint8_t *init_payload_;
+ std::size_t kBucketAlignment;
+
+ // Value's offset in a bucket is the first alignof(ValueT) boundary after the
+ // next pointer and hash code.
+ std::size_t kValueOffset;
+
+ // Round bucket size up to a multiple of kBucketAlignment.
+ constexpr std::size_t ComputeBucketSize(const std::size_t fixed_key_size) {
+ return (((kValueOffset + this->total_payload_size_ + fixed_key_size - 1) / kBucketAlignment) + 1)
+ * kBucketAlignment;
+ }
+ // If ValueT is not trivially destructible, invoke its destructor for all
+ // values held in the specified buckets (including those in "empty" buckets
+ // that were default constructed). If ValueT is trivially destructible, this
+ // is a no-op.
+ void DestroyValues(void *buckets,
+ const std::size_t num_buckets,
+ const std::size_t bucket_size);
+
+ // Attempt to find an empty bucket to insert 'hash_code' into, starting after
+ // '*bucket' in the chain (or, if '*bucket' is NULL, starting from the slot
+ // array). Returns true and stores SIZE_T_MAX in '*pending_chain_ptr' if an
+ // empty bucket is found. Returns false if 'allow_duplicate_keys' is false
+ // and a hash collision is found (caller should then check whether there is a
+ // genuine key collision or the hash collision is spurious). Returns false
+ // and sets '*bucket' to NULL if there are no more empty buckets in the hash
+ // table. If 'variable_key_allocation_required' is nonzero, this method will
+ // attempt to allocate storage for a variable-length key BEFORE allocating a
+ // bucket, so that no bucket number below 'header_->num_buckets' is ever
+ // 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,
+ HashTablePreallocationState *prealloc_state);
+
+ // Write a scalar 'key' and its 'hash_code' into the '*bucket', which was
+ // found by locateBucketForInsertion(). Assumes that storage for a
+ // variable-length key copy (if any) was already allocated by a successful
+ // call to allocateVariableLengthKeyStorage().
+ inline void writeScalarKeyToBucket(const TypedValue &key,
+ const std::size_t hash_code,
+ void *bucket,
+ HashTablePreallocationState *prealloc_state);
+
+ // Write a composite 'key' and its 'hash_code' into the '*bucket', which was
+ // found by locateBucketForInsertion(). Assumes that storage for
+ // variable-length key copies (if any) was already allocated by a successful
+ // call to allocateVariableLengthKeyStorage().
+ inline void writeCompositeKeyToBucket(const std::vector<TypedValue> &key,
+ const std::size_t hash_code,
+ void *bucket,
+ HashTablePreallocationState *prealloc_state);
+
+ // Determine whether it is actually necessary to resize this hash table.
+ // Checks that there is at least one unallocated bucket, and that there is
+ // at least 'extra_variable_storage' bytes of variable-length storage free.
+ bool isFull(const std::size_t extra_variable_storage) const;
+
+ // Helper object to manage key storage.
+ HashTableKeyManager<serializable, force_key_copy> key_manager_;
+
+ // In-memory structure is as follows:
+ // - SeparateChainingHashTable::Header
+ // - Array of slots, interpreted as follows:
+ // - 0 = Points to nothing (empty)
+ // - SIZE_T_MAX = Pending (some thread is starting a chain from this
+ // slot and will overwrite it soon)
+ // - Anything else = The number of the first bucket in the chain for
+ // this slot PLUS ONE (i.e. subtract one to get the actual bucket
+ // number).
+ // - Array of buckets, each of which is:
+ // - atomic size_t "next" pointer, interpreted the same as slots above.
+ // - size_t hash value
+ // - possibly some unused bytes as needed so that ValueT's alignment
+ // requirement is met
+ // - ValueT value slot
+ // - fixed-length key storage (which may include pointers to external
+ // memory or offsets of variable length keys stored within this hash
+ // table)
+ // - possibly some additional unused bytes so that bucket size is a
+ // multiple of both alignof(std::atomic<std::size_t>) and
+ // alignof(ValueT)
+ // - Variable-length key storage region (referenced by offsets stored in
+ // fixed-length keys).
+ Header *header_;
+
+ std::atomic<std::size_t> *slots_;
+ void *buckets_;
+ const std::size_t bucket_size_;
+
+ DISALLOW_COPY_AND_ASSIGN(FastSeparateChainingHashTable);
+};
+
+/** @} */
+
+// ----------------------------------------------------------------------------
+// Implementations of template class methods follow.
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::FastSeparateChainingHashTable(const std::vector<const Type*> &key_types,
+ const std::size_t num_entries,
+ const std::vector<std::size_t> &payload_sizes,
+ const std::vector<AggregationHandle *> &handles,
+ StorageManager *storage_manager)
+ : FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>(
+ key_types,
+ num_entries,
+ handles,
+ payload_sizes,
+ storage_manager,
+ false,
+ false,
+ true),
+ kBucketAlignment(alignof(std::atomic<std::size_t>)),
+ kValueOffset(sizeof(std::atomic<std::size_t>) + sizeof(std::size_t)),
+ key_manager_(this->key_types_, kValueOffset + this->total_payload_size_),
+ bucket_size_(ComputeBucketSize(key_manager_.getFixedKeySize())) {
+ init_payload_ = static_cast<std::uint8_t *>(calloc(this->total_payload_size_, 1));
+ for (auto handle : handles)
+ handle->initPayload(init_payload_);
+ // 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.
+ //
+ // Give base HashTable information about what key components are stored
+ // inline from 'key_manager_'.
+ this->setKeyInline(key_manager_.getKeyInline());
+
+ // Pick out a prime number of slots and calculate storage requirements.
+ std::size_t num_slots_tmp = get_next_prime_number(num_entries * kHashTableLoadFactor);
+ std::size_t required_memory = sizeof(Header)
+ + num_slots_tmp * sizeof(std::atomic<std::size_t>)
+ + (num_slots_tmp / kHashTableLoadFactor)
+ * (bucket_size_ + key_manager_.getEstimatedVariableKeySize());
+ std::size_t num_storage_slots = this->storage_manager_->SlotsNeededForBytes(required_memory);
+ if (num_storage_slots == 0) {
+ FATAL_ERROR("Storage requirement for SeparateChainingHashTable "
+ "exceeds maximum allocation size.");
+ }
+
+ // Get a StorageBlob to hold the hash table.
+ const block_id blob_id = this->storage_manager_->createBlob(num_storage_slots);
+ this->blob_ = this->storage_manager_->getBlobMutable(blob_id);
+
+ void *aligned_memory_start = this->blob_->getMemoryMutable();
+ std::size_t available_memory = num_storage_slots * kSlotSizeBytes;
+ if (align(alignof(Header),
+ sizeof(Header),
+ aligned_memory_start,
+ available_memory)
+ == nullptr) {
+ // With current values from StorageConstants.hpp, this should be
+ // impossible. A blob is at least 1 MB, while a Header has alignment
+ // requirement of just kCacheLineBytes (64 bytes).
+ FATAL_ERROR("StorageBlob used to hold resizable "
+ "SeparateChainingHashTable is too small to meet alignment "
+ "requirements of SeparateChainingHashTable::Header.");
+ } else if (aligned_memory_start != this->blob_->getMemoryMutable()) {
+ // This should also be impossible, since the StorageManager allocates slots
+ // aligned to kCacheLineBytes.
+ DEV_WARNING("StorageBlob memory adjusted by "
+ << (num_storage_slots * kSlotSizeBytes - available_memory)
+ << " bytes to meet alignment requirement for "
+ << "SeparateChainingHashTable::Header.");
+ }
+
+ // Locate the header.
+ header_ = static_cast<Header*>(aligned_memory_start);
+ aligned_memory_start = static_cast<char*>(aligned_memory_start) + sizeof(Header);
+ available_memory -= sizeof(Header);
+
+ // Recompute the number of slots & buckets using the actual available memory.
+ // Most likely, we got some extra free bucket space due to "rounding up" to
+ // the storage blob's size. It's also possible (though very unlikely) that we
+ // will wind up with fewer buckets than we initially wanted because of screwy
+ // alignment requirements for ValueT.
+ std::size_t num_buckets_tmp
+ = available_memory / (kHashTableLoadFactor * sizeof(std::atomic<std::size_t>)
+ + bucket_size_
+ + key_manager_.getEstimatedVariableKeySize());
+ num_slots_tmp = get_previous_prime_number(num_buckets_tmp * kHashTableLoadFactor);
+ num_buckets_tmp = num_slots_tmp / kHashTableLoadFactor;
+ DEBUG_ASSERT(num_slots_tmp > 0);
+ DEBUG_ASSERT(num_buckets_tmp > 0);
+
+ // Locate the slot array.
+ slots_ = static_cast<std::atomic<std::size_t>*>(aligned_memory_start);
+ aligned_memory_start = static_cast<char*>(aligned_memory_start)
+ + sizeof(std::atomic<std::size_t>) * num_slots_tmp;
+ available_memory -= sizeof(std::atomic<std::size_t>) * num_slots_tmp;
+
+ // Locate the buckets.
+ buckets_ = aligned_memory_start;
+ // Extra-paranoid: If ValueT has an alignment requirement greater than that
+ // of std::atomic<std::size_t>, we may need to adjust the start of the bucket
+ // array.
+ if (align(kBucketAlignment,
+ bucket_size_,
+ buckets_,
+ available_memory)
+ == nullptr) {
+ FATAL_ERROR("StorageBlob used to hold resizable "
+ "SeparateChainingHashTable is too small to meet "
+ "alignment requirements of buckets.");
+ } else if (buckets_ != aligned_memory_start) {
+ DEV_WARNING("Bucket array start position adjusted to meet alignment "
+ "requirement for SeparateChainingHashTable's value type.");
+ if (num_buckets_tmp * bucket_size_ > available_memory) {
+ --num_buckets_tmp;
+ }
+ }
+
+ // Fill in the header.
+ header_->num_slots = num_slots_tmp;
+ header_->num_buckets = num_buckets_tmp;
+ header_->buckets_allocated.store(0, std::memory_order_relaxed);
+ header_->variable_length_bytes_allocated.store(0, std::memory_order_relaxed);
+ available_memory -= bucket_size_ * (header_->num_buckets);
+
+ // Locate variable-length key storage region, and give it all the remaining
+ // bytes in the blob.
+ key_manager_.setVariableLengthStorageInfo(
+ static_cast<char*>(buckets_) + header_->num_buckets * bucket_size_,
+ available_memory,
+ &(header_->variable_length_bytes_allocated));
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::FastSeparateChainingHashTable(const std::vector<const Type*> &key_types,
+ void *hash_table_memory,
+ const std::size_t hash_table_memory_size,
+ const bool new_hash_table,
+ const bool hash_table_memory_zeroed)
+ : FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>(
+ key_types,
+ hash_table_memory,
+ hash_table_memory_size,
+ new_hash_table,
+ hash_table_memory_zeroed,
+ false,
+ false,
+ true),
+ kBucketAlignment(alignof(std::atomic<std::size_t>) < alignof(uint8_t) ? alignof(uint8_t)
+ : alignof(std::atomic<std::size_t>)),
+ kValueOffset((((sizeof(std::atomic<std::size_t>) + sizeof(std::size_t) - 1) /
+ alignof(uint8_t)) + 1) * alignof(uint8_t)),
+ key_manager_(this->key_types_, kValueOffset + sizeof(uint8_t)),
+ bucket_size_(ComputeBucketSize(key_manager_.getFixedKeySize())) {
+ // 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.
+ //
+ // Make sure that the larger of the two alignment requirements also satisfies
+ // the smaller.
+ static_assert(alignof(std::atomic<std::size_t>) < alignof(uint8_t)
+ ? alignof(uint8_t) % alignof(std::atomic<std::size_t>) == 0
+ : alignof(std::atomic<std::size_t>) % alignof(uint8_t) == 0,
+ "Alignment requirement of std::atomic<std::size_t> does not "
+ "evenly divide with alignment requirement of ValueT.");
+
+ // Give base HashTable information about what key components are stored
+ // inline from 'key_manager_'.
+ this->setKeyInline(key_manager_.getKeyInline());
+
+ // FIXME(chasseur): If we are reconstituting a HashTable using a block of
+ // memory whose start was aligned differently than the memory block that was
+ // originally used (modulo alignof(Header)), we could wind up with all of our
+ // data structures misaligned. If memory is inside a
+ // StorageBlock/StorageBlob, this will never occur, since the StorageManager
+ // always allocates slots aligned to kCacheLineBytes. Similarly, this isn't
+ // a problem for memory inside any other allocation aligned to at least
+ // alignof(Header) == kCacheLineBytes.
+
+ void *aligned_memory_start = this->hash_table_memory_;
+ std::size_t available_memory = this->hash_table_memory_size_;
+
+ if (align(alignof(Header),
+ sizeof(Header),
+ aligned_memory_start,
+ available_memory)
+ == nullptr) {
+ FATAL_ERROR("Attempted to create a non-resizable "
+ << "SeparateChainingHashTable with "
+ << available_memory << " bytes of memory at "
+ << aligned_memory_start << " which either can not fit a "
+ << "SeparateChainingHashTable::Header or meet its alignement "
+ << "requirement.");
+ } else if (aligned_memory_start != this->hash_table_memory_) {
+ // In general, we could get memory of any alignment, although at least
+ // cache-line aligned would be nice.
+ DEV_WARNING("StorageBlob memory adjusted by "
+ << (this->hash_table_memory_size_ - available_memory)
+ << " bytes to meet alignment requirement for "
+ << "SeparateChainingHashTable::Header.");
+ }
+
+ header_ = static_cast<Header*>(aligned_memory_start);
+ aligned_memory_start = static_cast<char*>(aligned_memory_start) + sizeof(Header);
+ available_memory -= sizeof(Header);
+
+ if (new_hash_table) {
+ std::size_t estimated_bucket_capacity
+ = available_memory / (kHashTableLoadFactor * sizeof(std::atomic<std::size_t>)
+ + bucket_size_
+ + key_manager_.getEstimatedVariableKeySize());
+ std::size_t num_slots = get_previous_prime_number(estimated_bucket_capacity * kHashTableLoadFactor);
+
+ // Fill in the header.
+ header_->num_slots = num_slots;
+ header_->num_buckets = num_slots / kHashTableLoadFactor;
+ header_->buckets_allocated.store(0, std::memory_order_relaxed);
+ header_->variable_length_bytes_allocated.store(0, std::memory_order_relaxed);
+ }
+
+ // Locate the slot array.
+ slots_ = static_cast<std::atomic<std::size_t>*>(aligned_memory_start);
+ aligned_memory_start = static_cast<char*>(aligned_memory_start)
+ + sizeof(std::atomic<std::size_t>) * header_->num_slots;
+ available_memory -= sizeof(std::atomic<std::size_t>) * header_->num_slots;
+
+ if (new_hash_table && !hash_table_memory_zeroed) {
+ std::memset(slots_, 0x0, sizeof(std::atomic<std::size_t>) * header_->num_slots);
+ }
+
+ // Locate the buckets.
+ buckets_ = aligned_memory_start;
+ // Extra-paranoid: sizeof(Header) should almost certainly be a multiple of
+ // kBucketAlignment, unless ValueT has some members with seriously big
+ // (> kCacheLineBytes) alignment requirements specified using alignas().
+ if (align(kBucketAlignment,
+ bucket_size_,
+ buckets_,
+ available_memory)
+ == nullptr) {
+ FATAL_ERROR("Attempted to create a non-resizable "
+ << "SeparateChainingHashTable with "
+ << this->hash_table_memory_size_ << " bytes of memory at "
+ << this->hash_table_memory_ << ", which can hold an aligned "
+ << "SeparateChainingHashTable::Header but does not have "
+ << "enough remaining space for even a single hash bucket.");
+ } else if (buckets_ != aligned_memory_start) {
+ DEV_WARNING("Bucket array start position adjusted to meet alignment "
+ "requirement for SeparateChainingHashTable's value type.");
+ if (header_->num_buckets * bucket_size_ > available_memory) {
+ DEBUG_ASSERT(new_hash_table);
+ --(header_->num_buckets);
+ }
+ }
+ available_memory -= bucket_size_ * header_->num_buckets;
+
+ // Make sure "next" pointers in buckets are zeroed-out.
+ if (new_hash_table && !hash_table_memory_zeroed) {
+ std::memset(buckets_, 0x0, header_->num_buckets * bucket_size_);
+ }
+
+ // Locate variable-length key storage region.
+ key_manager_.setVariableLengthStorageInfo(
+ static_cast<char*>(buckets_) + header_->num_buckets * bucket_size_,
+ available_memory,
+ &(header_->variable_length_bytes_allocated));
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+void FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::clear() {
+ const std::size_t used_buckets = header_->buckets_allocated.load(std::memory_order_relaxed);
+ // Destroy existing values, if necessary.
+ DestroyValues(buckets_,
+ used_buckets,
+ bucket_size_);
+
+ // Zero-out slot array.
+ std::memset(slots_, 0x0, sizeof(std::atomic<std::size_t>) * header_->num_slots);
+
+ // Zero-out used buckets.
+ std::memset(buckets_, 0x0, used_buckets * bucket_size_);
+
+ header_->buckets_allocated.store(0, std::memory_order_relaxed);
+ header_->variable_length_bytes_allocated.store(0, std::memory_order_relaxed);
+ key_manager_.zeroNextVariableLengthKeyOffset();
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+const uint8_t* FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getSingle(const TypedValue &key) const {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ const std::size_t hash_code = key.getHash();
+ 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_.scalarKeyCollisionCheck(key, bucket)) {
+ // Match located.
+ return reinterpret_cast<const uint8_t*>(bucket + kValueOffset);
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+
+ // Reached the end of the chain and didn't find a match.
+ return nullptr;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+const uint8_t* FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getSingleCompositeKey(const std::vector<TypedValue> &key) const {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ 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 uint8_t*>(bucket + kValueOffset);
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+
+ // Reached the end of the chain and didn't find a match.
+ return nullptr;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+const uint8_t* FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getSingleCompositeKey(const std::vector<TypedValue> &key, int index) const {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ 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 uint8_t*>(bucket + kValueOffset)+this->payload_offsets_[index];
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+
+ // Reached the end of the chain and didn't find a match.
+ return nullptr;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+void FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getAll(const TypedValue &key, std::vector<const uint8_t*> *values) const {
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ const std::size_t hash_code = key.getHash();
+ 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_.scalarKeyCollisionCheck(key, bucket)) {
+ // Match located.
+ values->push_back(reinterpret_cast<const uint8_t*>(bucket + kValueOffset));
+ if (!allow_duplicate_keys) {
+ return;
+ }
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+void FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getAllCompositeKey(const std::vector<TypedValue> &key, std::vector<const uint8_t*> *values) 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.
+ values->push_back(reinterpret_cast<const uint8_t*>(bucket + kValueOffset));
+ if (!allow_duplicate_keys) {
+ return;
+ }
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+HashTablePutResult
+ FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::putInternal(const TypedValue &key,
+ const std::size_t variable_key_size,
+ const uint8_t &value,
+ HashTablePreallocationState *prealloc_state) {
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ if (prealloc_state == nullptr) {
+ // Early check for a free bucket.
+ if (header_->buckets_allocated.load(std::memory_order_relaxed) >= header_->num_buckets) {
+ return HashTablePutResult::kOutOfSpace;
+ }
+
+ // TODO(chasseur): If allow_duplicate_keys is true, avoid storing more than
+ // one copy of the same variable-length key.
+ if (!key_manager_.allocateVariableLengthKeyStorage(variable_key_size)) {
+ // Ran out of variable-length key storage space.
+ return HashTablePutResult::kOutOfSpace;
+ }
+ }
+
+ 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,
+ 0,
+ &bucket,
+ &pending_chain_ptr,
+ &pending_chain_ptr_finish_value,
+ prealloc_state)) {
+ // Found an empty bucket.
+ break;
+ } else if (bucket == nullptr) {
+ // Ran out of buckets. Deallocate any variable space that we were unable
+ // to use.
+ DEBUG_ASSERT(prealloc_state == nullptr);
+ key_manager_.deallocateVariableLengthKeyStorage(variable_key_size);
+ return HashTablePutResult::kOutOfSpace;
+ } else {
+ // Hash collision found, and duplicates aren't allowed.
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(prealloc_state == nullptr);
+ if (key_manager_.scalarKeyCollisionCheck(key, bucket)) {
+ // Duplicate key. Deallocate any variable storage space and return.
+ key_manager_.deallocateVariableLengthKeyStorage(variable_key_size);
+ return HashTablePutResult::kDuplicateKey;
+ }
+ }
+ }
+
+ // Write the key and hash.
+ writeScalarKeyToBucket(key, hash_code, bucket, prealloc_state);
+
+ // Store the value by using placement new with ValueT's copy constructor.
+ new(static_cast<char*>(bucket) + kValueOffset) uint8_t(value);
+
+ // Update the previous chain pointer to point to the new bucket.
+ pending_chain_ptr->store(pending_chain_ptr_finish_value, std::memory_order_release);
+
+ // We're all done.
+ return HashTablePutResult::kOK;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+HashTablePutResult
+ FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::putCompositeKeyInternal(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const uint8_t &value,
+ HashTablePreallocationState *prealloc_state) {
+ DEBUG_ASSERT(this->key_types_.size() == key.size());
+
+ if (prealloc_state == nullptr) {
+ // Early check for a free bucket.
+ if (header_->buckets_allocated.load(std::memory_order_relaxed) >= header_->num_buckets) {
+ return HashTablePutResult::kOutOfSpace;
+ }
+
+ // TODO(chasseur): If allow_duplicate_keys is true, avoid storing more than
+ // one copy of the same variable-length key.
+ if (!key_manager_.allocateVariableLengthKeyStorage(variable_key_size)) {
+ // Ran out of variable-length key storage space.
+ return HashTablePutResult::kOutOfSpace;
+ }
+ }
+
+ 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,
+ 0,
+ &bucket,
+ &pending_chain_ptr,
+ &pending_chain_ptr_finish_value,
+ prealloc_state)) {
+ // Found an empty bucket.
+ break;
+ } else if (bucket == nullptr) {
+ // Ran out of buckets. Deallocate any variable space that we were unable
+ // to use.
+ DEBUG_ASSERT(prealloc_state == nullptr);
+ key_manager_.deallocateVariableLengthKeyStorage(variable_key_size);
+ return HashTablePutResult::kOutOfSpace;
+ } else {
+ // Hash collision found, and duplicates aren't allowed.
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(prealloc_state == nullptr);
+ if (key_manager_.compositeKeyCollisionCheck(key, bucket)) {
+ // Duplicate key. Deallocate any variable storage space and return.
+ key_manager_.deallocateVariableLengthKeyStorage(variable_key_size);
+ return HashTablePutResult::kDuplicateKey;
+ }
+ }
+ }
+
+ // Write the key and hash.
+ writeCompositeKeyToBucket(key, hash_code, bucket, prealloc_state);
+
+ // Store the value by using placement new with ValueT's copy constructor.
+ new(static_cast<char*>(bucket) + kValueOffset) uint8_t(value);
+
+ // Update the previous chain pointer to point to the new bucket.
+ pending_chain_ptr->store(pending_chain_ptr_finish_value, std::memory_order_release);
+
+ // We're all done.
+ return HashTablePutResult::kOK;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+HashTablePutResult
+ FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::putCompositeKeyInternalFast(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const uint8_t *init_value_ptr,
+ HashTablePreallocationState *prealloc_state) {
+ DEBUG_ASSERT(this->key_types_.size() == key.size());
+
+ if (prealloc_state == nullptr) {
+ // Early check for a free bucket.
+ if (header_->buckets_allocated.load(std::memory_order_relaxed) >= header_->num_buckets) {
+ return HashTablePutResult::kOutOfSpace;
+ }
+
+ // TODO(chasseur): If allow_duplicate_keys is true, avoid storing more than
+ // one copy of the same variable-length key.
+ if (!key_manager_.allocateVariableLengthKeyStorage(variable_key_size)) {
+ // Ran out of variable-length key storage space.
+ return HashTablePutResult::kOutOfSpace;
+ }
+ }
+
+ 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,
+ 0,
+ &bucket,
+ &pending_chain_ptr,
+ &pending_chain_ptr_finish_value,
+ prealloc_state)) {
+ // Found an empty bucket.
+ break;
+ } else if (bucket == nullptr) {
+ // Ran out of buckets. Deallocate any variable space that we were unable
+ // to use.
+ DEBUG_ASSERT(prealloc_state == nullptr);
+ key_manager_.deallocateVariableLengthKeyStorage(variable_key_size);
+ return HashTablePutResult::kOutOfSpace;
+ } else {
+ // Hash collision found, and duplicates aren't allowed.
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(prealloc_state == nullptr);
+ if (key_manager_.compositeKeyCollisionCheck(key, bucket)) {
+ // Duplicate key. Deallocate any variable storage space and return.
+ key_manager_.deallocateVariableLengthKeyStorage(variable_key_size);
+ return HashTablePutResult::kDuplicateKey;
+ }
+ }
+ }
+
+ // Write the key and hash.
+ writeCompositeKeyToBucket(key, hash_code, bucket, prealloc_state);
+
+ // Store the value by using placement new with ValueT's copy constructor.
+// new(static_cast<char*>(bucket) + kValueOffset) uint8_t(value);
+ uint8_t *value = static_cast<uint8_t*>(bucket) + kValueOffset;
+ memcpy(value, init_value_ptr, 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);
+
+ // We're all done.
+ return HashTablePutResult::kOK;
+}
+
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+uint8_t* FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::upsertInternal(const TypedValue &key,
+ const std::size_t variable_key_size,
+ const uint8_t &initial_value) {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ 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())) {
+ return nullptr;
+ }
+ }
+
+ 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,
+ variable_key_size,
+ &bucket,
+ &pending_chain_ptr,
+ &pending_chain_ptr_finish_value,
+ nullptr)) {
+ // 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<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, nullptr);
+
+ // Copy the supplied 'initial_value' into place.
+ uint8_t *value = new(static_cast<char*>(bucket) + kValueOffset) uint8_t(initial_value);
+
+ // 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;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+uint8_t* FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::upsertInternalFast(const TypedValue &key,
+ const std::uint8_t *init_value_ptr,
+ const std::size_t variable_key_size) {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ 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())) {
+ return nullptr;
+ }
+ }
+
+ 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,
+ variable_key_size,
+ &bucket,
+ &pending_chain_ptr,
+ &pending_chain_ptr_finish_value,
+ nullptr)) {
+ // 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<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, nullptr);
+
+ // Copy the supplied 'initial_value' into place.
+// uint8_t *value = new(static_cast<char*>(bucket) + kValueOffset) uint8_t(initial_value);
+
+ uint8_t *value = static_cast<unsigned char*>(bucket) + kValueOffset;
+ if (init_value_ptr == nullptr)
+ memcpy(value, init_payload_, this->total_payload_size_);
+ else
+ memcpy(value, init_value_ptr, 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;
+}
+
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+uint8_t* FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::upsertCompositeKeyInternal(const std::vector<TypedValue> &key,
+ const std::size_t variable_key_size,
+ const uint8_t &initial_value) {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(this->key_types_.size() == 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())) {
+ return nullptr;
+ }
+ }
+
+ 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,
+ nullptr)) {
+ // Found an empty bucket.
+ break;
+ } else if (bucket == nullptr) {
+ // Ran out of buckets or variable-key space.
+ return nullptr;
+ } else if (key_manager_.compositeKeyCollisionCheck(key, bucket)) {
+ // Found an already-existing entry for this key.
+ return reinterpret_cast<uint8_t*>(static_cast<char*>(bucket) + kValueOffset);
+ }
+ }
+
+ // We are now writing to an empty bucket.
+ // Write the key and hash.
+ writeCompositeKeyToBucket(key, hash_code, bucket, nullptr);
+
+ // Copy the supplied 'initial_value' into place.
+ uint8_t *value = new(static_cast<char*>(bucket) + kValueOffset) uint8_t(initial_value);
+
+ // Update the previous chaing 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;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+uint8_t* FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::upsertCompositeKeyInternalFast(const std::vector<TypedValue> &key,
+ const std::uint8_t *init_value_ptr,
+ const std::size_t variable_key_size) {
+ DEBUG_ASSERT(!allow_duplicate_keys);
+ DEBUG_ASSERT(this->key_types_.size() == 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())) {
+ return nullptr;
+ }
+ }
+
+ 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,
+ nullptr)) {
+ // Found an empty bucket.
+ break;
+ } else if (bucket == nullptr) {
+ // Ran out of buckets or variable-key space.
+ return nullptr;
+ } else if (key_manager_.compositeKeyCollisionCheck(key, bucket)) {
+ // Found an already-existing entry for this key.
+ return reinterpret_cast<uint8_t*>(static_cast<char*>(bucket) + kValueOffset);
+ }
+ }
+
+ // We are now writing to an empty bucket.
+ // Write the key and hash.
+ writeCompositeKeyToBucket(key, hash_code, bucket, nullptr);
+
+// uint8_t *value;
+// value = static_cast<unsigned char*>(bucket) + kValueOffset;
+ uint8_t *value = static_cast<unsigned char*>(bucket) + kValueOffset;
+ if (init_value_ptr == nullptr)
+ memcpy(value, init_payload_, this->total_payload_size_);
+ else
+ memcpy(value, init_value_ptr, this->total_payload_size_);
+
+ // Update the previous chaing 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;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getNextEntry(TypedValue *key, const uint8_t **value, std::size_t *entry_num) const {
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ if (*entry_num < header_->buckets_allocated.load(std::memory_order_relaxed)) {
+ const char *bucket = static_cast<const char*>(buckets_) + (*entry_num) * bucket_size_;
+ *key = key_manager_.getKeyComponentTyped(bucket, 0);
+ *value = reinterpret_cast<const uint8_t*>(bucket + kValueOffset);
+ ++(*entry_num);
+ return true;
+ } else {
+ return false;
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getNextEntryCompositeKey(std::vector<TypedValue> *key,
+ const uint8_t **value,
+ std::size_t *entry_num) const {
+ if (*entry_num < header_->buckets_allocated.load(std::memory_order_relaxed)) {
+ const char *bucket = static_cast<const char*>(buckets_) + (*entry_num) * bucket_size_;
+ for (std::vector<const Type*>::size_type key_idx = 0;
+ key_idx < this->key_types_.size();
+ ++key_idx) {
+ key->emplace_back(key_manager_.getKeyComponentTyped(bucket, key_idx));
+ }
+ *value = reinterpret_cast<const uint8_t*>(bucket + kValueOffset);
+ ++(*entry_num);
+ return true;
+ } else {
+ return false;
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getNextEntryForKey(const TypedValue &key,
+ const std::size_t hash_code,
+ const uint8_t **value,
+ std::size_t *entry_num) const {
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ if (*entry_num == 0) {
+ *entry_num = slots_[hash_code % header_->num_slots].load(std::memory_order_relaxed);
+ } else if (*entry_num == std::numeric_limits<std::size_t>::max()) {
+ return false;
+ }
+
+ while (*entry_num != 0) {
+ DEBUG_ASSERT(*entry_num != std::numeric_limits<std::size_t>::max());
+ const char *bucket = static_cast<const char*>(buckets_) + (*entry_num - 1) * bucket_size_;
+ *entry_num = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ 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_.scalarKeyCollisionCheck(key, bucket)) {
+ // Match located.
+ *value = reinterpret_cast<const uint8_t*>(bucket + kValueOffset);
+ if (*entry_num == 0) {
+ // If this is the last bucket in the chain, prevent the next call from
+ // starting over again.
+ *entry_num = std::numeric_limits<std::size_t>::max();
+ }
+ return true;
+ }
+ }
+
+ // Reached the end of the chain.
+ return false;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::getNextEntryForCompositeKey(const std::vector<TypedValue> &key,
+ const std::size_t hash_code,
+ const uint8_t **value,
+ std::size_t *entry_num) const {
+ DEBUG_ASSERT(this->key_types_.size() == key.size());
+
+ if (*entry_num == 0) {
+ *entry_num = slots_[hash_code % header_->num_slots].load(std::memory_order_relaxed);
+ } else if (*entry_num == std::numeric_limits<std::size_t>::max()) {
+ return false;
+ }
+
+ while (*entry_num != 0) {
+ DEBUG_ASSERT(*entry_num != std::numeric_limits<std::size_t>::max());
+ const char *bucket = static_cast<const char*>(buckets_) + (*entry_num - 1) * bucket_size_;
+ *entry_num = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ 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.
+ *value = reinterpret_cast<const uint8_t*>(bucket + kValueOffset);
+ if (*entry_num == 0) {
+ // If this is the last bucket in the chain, prevent the next call from
+ // starting over again.
+ *entry_num = std::numeric_limits<std::size_t>::max();
+ }
+ return true;
+ }
+ }
+
+ // Reached the end of the chain.
+ return false;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::hasKey(const TypedValue &key) const {
+ DEBUG_ASSERT(this->key_types_.size() == 1);
+ DEBUG_ASSERT(key.isPlausibleInstanceOf(this->key_types_.front()->getSignature()));
+
+ const std::size_t hash_code = key.getHash();
+ 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_.scalarKeyCollisionCheck(key, bucket)) {
+ // Find a match.
+ return true;
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+ return false;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::hasCompositeKey(const std::vector<TypedValue> &key) 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)) {
+ // Find a match.
+ return true;
+ }
+ bucket_ref = reinterpret_cast<const std::atomic<std::size_t>*>(bucket)->load(std::memory_order_relaxed);
+ }
+ return false;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+void FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::resize(const std::size_t extra_buckets,
+ const std::size_t extra_variable_storage,
+ const std::size_t retry_num) {
+ DEBUG_ASSERT(resizable);
+
+ // A retry should never be necessary with this implementation of HashTable.
+ // Separate chaining ensures that any resized hash table with more buckets
+ // than the original table will be able to hold more entries than the
+ // original.
+ DEBUG_ASSERT(retry_num == 0);
+
+ SpinSharedMutexExclusiveLock<true> write_lock(this->resize_shared_mutex_);
+
+ // Recheck whether the hash table is still full. Note that multiple threads
+ // might wait to rebuild this hash table simultaneously. Only the first one
+ // should do the rebuild.
+ if (!isFull(extra_variable_storage)) {
+ return;
+ }
+
+ // Approximately double the number of buckets and slots.
+ //
+ // TODO(chasseur): It may be worth it to more than double the number of
+ // buckets here so that we can maintain a good, sparse fill factor for a
+ // longer time as more values are inserted. Such behavior should take into
+ // account kHashTableLoadFactor.
+ std::size_t resized_num_slots = get_next_prime_number(
+ (header_->num_buckets + extra_buckets / 2) * kHashTableLoadFactor * 2);
+ std::size_t variable_storage_required
+ = (resized_num_slots / kHashTableLoadFactor) * key_manager_.getEstimatedVariableKeySize();
+ const std::size_t original_variable_storage_used
+ = header_->variable_length_bytes_allocated.load(std::memory_order_relaxed);
+ // If this resize was triggered by a too-large variable-length key, bump up
+ // the variable-length storage requirement.
+ if ((extra_variable_storage > 0)
+ && (extra_variable_storage + original_variable_storage_used
+ > key_manager_.getVariableLengthKeyStorageSize())) {
+ variable_storage_required += extra_variable_storage;
+ }
+
+ const std::size_t resized_memory_required
+ = sizeof(Header)
+ + resized_num_slots * sizeof(std::atomic<std::size_t>)
+ + (resized_num_slots / kHashTableLoadFactor) * bucket_size_
+ + variable_storage_required;
+ const std::size_t resized_storage_slots
+ = this->storage_manager_->SlotsNeededForBytes(resized_memory_required);
+ if (resized_storage_slots == 0) {
+ FATAL_ERROR("Storage requirement for resized SeparateChainingHashTable "
+ "exceeds maximum allocation size.");
+ }
+
+ // Get a new StorageBlob to hold the resized hash table.
+ const block_id resized_blob_id = this->storage_manager_->createBlob(resized_storage_slots);
+ MutableBlobReference resized_blob = this->storage_manager_->getBlobMutable(resized_blob_id);
+
+ // Locate data structures inside the new StorageBlob.
+ void *aligned_memory_start = resized_blob->getMemoryMutable();
+ std::size_t available_memory = resized_storage_slots * kSlotSizeBytes;
+ if (align(alignof(Header),
+ sizeof(Header),
+ aligned_memory_start,
+ available_memory)
+ == nullptr) {
+ // Should be impossible, as noted in constructor.
+ FATAL_ERROR("StorageBlob used to hold resized SeparateChainingHashTable "
+ "is too small to meet alignment requirements of "
+ "LinearOpenAddressingHashTable::Header.");
+ } else if (aligned_memory_start != resized_blob->getMemoryMutable()) {
+ // Again, should be impossible.
+ DEV_WARNING("In SeparateChainingHashTable::resize(), StorageBlob "
+ << "memory adjusted by "
+ << (resized_num_slots * kSlotSizeBytes - available_memory)
+ << " bytes to meet alignment requirement for "
+ << "LinearOpenAddressingHashTable::Header.");
+ }
+
+ Header *resized_header = static_cast<Header*>(aligned_memory_start);
+ aligned_memory_start = static_cast<char*>(aligned_memory_start) + sizeof(Header);
+ available_memory -= sizeof(Header);
+
+ // As in constructor, recompute the number of slots and buckets using the
+ // actual available memory.
+ std::size_t resized_num_buckets
+ = (available_memory - extra_variable_storage)
+ / (kHashTableLoadFactor * sizeof(std::atomic<std::size_t>)
+ + bucket_size_
+ + key_manager_.getEstimatedVariableKeySize());
+ resized_num_slots = get_previous_prime_number(resized_num_buckets * kHashTableLoadFactor);
+ resized_num_buckets = resized_num_slots / kHashTableLoadFactor;
+
+ // Locate slot array.
+ std::atomic<std::size_t> *resized_slots = static_cast<std::atomic<std::size_t>*>(aligned_memory_start);
+ aligned_memory_start = static_cast<char*>(aligned_memory_start)
+ + sizeof(std::atomic<std::size_t>) * resized_num_slots;
+ available_memory -= sizeof(std::atomic<std::size_t>) * resized_num_slots;
+
+ // As in constructor, we will be extra paranoid and use align() to locate the
+ // start of the array of buckets, as well.
+ void *resized_buckets = aligned_memory_start;
+ if (align(kBucketAlignment,
+ bucket_size_,
+ resized_buckets,
+ available_memory)
+ == nullptr) {
+ FATAL_ERROR("StorageBlob used to hold resized SeparateChainingHashTable "
+ "is too small to meet alignment requirements of buckets.");
+ } else if (resized_buckets != aligned_memory_start) {
+ DEV_WARNING("Bucket array start position adjusted to meet alignment "
+ "requirement for SeparateChainingHashTable's value type.");
+ if (resized_num_buckets * bucket_size_ + variable_storage_required > available_memory) {
+ --resized_num_buckets;
+ }
+ }
+ aligned_memory_start = static_cast<char*>(aligned_memory_start)
+ + resized_num_buckets * bucket_size_;
+ available_memory -= resized_num_buckets * bucket_size_;
+
+ void *resized_variable_length_key_storage = aligned_memory_start;
+ const std::size_t resized_variable_length_key_storage_size = available_memory;
+
+ const std::size_t original_buckets_used = header_->buckets_allocated.load(std::memory_order_relaxed);
+
+ // Initialize the header.
+ resized_header->num_slots = resized_num_slots;
+ resized_header->num_buckets = resized_num_buckets;
+ resized_header->buckets_allocated.store(original_buckets_used, std::memory_order_relaxed);
+ resized_header->variable_length_bytes_allocated.store(
+ original_variable_storage_used,
+ std::memory_order_relaxed);
+
+ // Bulk-copy buckets. This is safe because:
+ // 1. The "next" pointers will be adjusted when rebuilding chains below.
+ // 2. The hash codes will stay the same.
+ // 3. For key components:
+ // a. Inline keys will stay exactly the same.
+ // b. Offsets into variable-length storage will remain valid, because
+ // we also do a byte-for-byte copy of variable-length storage below.
+ // c. Absolute external pointers will still point to the same address.
+ // d. Relative pointers are not used with resizable hash tables.
+ // 4. If values are not trivially copyable, then we invoke ValueT's copy
+ // or move constructor with placement new.
+ std::memcpy(resized_buckets, buckets_, original_buckets_used * bucket_size_);
+
+ // TODO(chasseur): std::is_trivially_copyable is not yet implemented in
+ // GCC 4.8.3, so we assume we need to invoke ValueT's copy or move
+ // constructor, even though the plain memcpy above could suffice for many
+ // possible ValueTs.
+ void *current_value_original = static_cast<char*>(buckets_) + kValueOffset;
+ void *current_value_resized = static_cast<char*>(resized_buckets) + kValueOffset;
+ for (std::size_t bucket_num = 0; bucket_num < original_buckets_used; ++bucket_num) {
+ // Use a move constructor if available to avoid a deep-copy, since resizes
+ // always succeed.
+ new (current_value_resized) uint8_t(std::move(*static_cast<uint8_t*>(current_value_original)));
+ current_value_original = static_cast<char*>(current_value_original) + bucket_size_;
+ current_value_resized = static_cast<char*>(current_value_resized) + bucket_size_;
+ }
+
+ // Copy over variable-length key components, if any.
+ if (original_variable_storage_used > 0) {
+ DEBUG_ASSERT(original_variable_storage_used
+ == key_manager_.getNextVariableLengthKeyOffset());
+ DEBUG_ASSERT(original_variable_storage_used <= resized_variable_length_key_storage_size);
+ std::memcpy(resized_variable_length_key_storage,
+ key_manager_.getVariableLengthKeyStorage(),
+ original_variable_storage_used);
+ }
+
+ // Destroy values in the original hash table, if neccesary,
+ DestroyValues(buckets_,
+ original_buckets_used,
+ bucket_size_);
+
+ // Make resized structures active.
+ std::swap(this->blob_, resized_blob);
+ header_ = resized_header;
+ slots_ = resized_slots;
+ buckets_ = resized_buckets;
+ key_manager_.setVariableLengthStorageInfo(
+ resized_variable_length_key_storage,
+ resized_variable_length_key_storage_size,
+ &(resized_header->variable_length_bytes_allocated));
+
+ // Drop the old blob.
+ const block_id old_blob_id = resized_blob->getID();
+ resized_blob.release();
+ this->storage_manager_->deleteBlockOrBlobFile(old_blob_id);
+
+ // Rebuild chains.
+ void *current_bucket = buckets_;
+ for (std::size_t bucket_num = 0; bucket_num < original_buckets_used; ++bucket_num) {
+ std::atomic<std::size_t> *next_ptr
+ = static_cast<std::atomic<std::size_t>*>(current_bucket);
+ const std::size_t hash_code = *reinterpret_cast<const std::size_t*>(
+ static_cast<const char*>(current_bucket) + sizeof(std::atomic<std::size_t>));
+
+ const std::size_t slot_number = hash_code % header_->num_slots;
+ std::size_t slot_ptr_value = 0;
+ if (slots_[slot_number].compare_exchange_strong(slot_ptr_value,
+ bucket_num + 1,
+ std::memory_order_relaxed)) {
+ // This bucket is the first in the chain for this block, so reset its
+ // next pointer to 0.
+ next_ptr->store(0, std::memory_order_relaxed);
+ } else {
+ // A chain already exists starting from this slot, so put this bucket at
+ // the head.
+ next_ptr->store(slot_ptr_value, std::memory_order_relaxed);
+ slots_[slot_number].store(bucket_num + 1, std::memory_order_relaxed);
+ }
+ current_bucket = static_cast<char*>(current_bucket) + bucket_size_;
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::preallocateForBulkInsert(const std::size_t total_entries,
+ const std::size_t total_variable_key_size,
+ HashTablePreallocationState *prealloc_state) {
+ DEBUG_ASSERT(allow_duplicate_keys);
+ if (!key_manager_.allocateVariableLengthKeyStorage(total_variable_key_size)) {
+ return false;
+ }
+
+ // We use load then compare-exchange here instead of simply fetch-add,
+ // because if multiple threads are simultaneously trying to allocate more
+ // than one bucket and exceed 'header_->num_buckets', their respective
+ // rollbacks might happen in such an order that some bucket ranges get
+ // skipped, while others might get double-allocated later.
+ std::size_t original_buckets_allocated = header_->buckets_allocated.load(std::memory_order_relaxed);
+ std::size_t buckets_post_allocation = original_buckets_allocated + total_entries;
+ while ((buckets_post_allocation <= header_->num_buckets)
+ && !header_->buckets_allocated.compare_exchange_weak(original_buckets_allocated,
+ buckets_post_allocation,
+ std::memory_order_relaxed)) {
+ buckets_post_allocation = original_buckets_allocated + total_entries;
+ }
+
+ if (buckets_post_allocation > header_->num_buckets) {
+ key_manager_.deallocateVariableLengthKeyStorage(total_variable_key_size);
+ return false;
+ }
+
+ prealloc_state->bucket_position = original_buckets_allocated;
+ if (total_variable_key_size != 0) {
+ prealloc_state->variable_length_key_position
+ = key_manager_.incrementNextVariableLengthKeyOffset(total_variable_key_size);
+ }
+ return true;
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+void FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::DestroyValues(void *hash_buckets,
+ const std::size_t num_buckets,
+ const std::size_t bucket_size) {
+ if (!std::is_trivially_destructible<uint8_t>::value) {
+ void *value_ptr = static_cast<char*>(hash_buckets) + kValueOffset;
+ for (std::size_t bucket_num = 0;
+ bucket_num < num_buckets;
+ ++bucket_num) {
+ static_cast<uint8_t*>(value_ptr)->~uint8_t();
+ value_ptr = static_cast<char*>(value_ptr) + bucket_size;
+ }
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+inline bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::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,
+ HashTablePreallocationState *prealloc_state) {
+ DEBUG_ASSERT((prealloc_state == nullptr) || allow_duplicate_keys);
+ if (*bucket == nullptr) {
+ *pending_chain_ptr = &(slots_[hash_code % header_->num_slots]);
+ } else {
+ *pending_chain_ptr = static_cast<std::atomic<std::size_t>*>(*bucket);
+ }
+ for (;;) {
+ std::size_t existing_chain_ptr = 0;
+ if ((*pending_chain_ptr)->compare_exchange_strong(existing_chain_ptr,
+ 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 ((prealloc_state == nullptr)
+ && !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
+ = (prealloc_state == nullptr)
+ ? header_->buckets_allocated.fetch_add(1, std::memory_order_relaxed)
+ : (prealloc_state->bucket_position)++;
+ if (allocated_bucket_num >= header_->num_buckets) {
+ // Ran out of buckets.
+ DEBUG_ASSERT(prealloc_state == nullptr);
+ header_->buckets_allocated.fetch_sub(1, std::memory_order_relaxed);
+ (*pending_chain_ptr)->store(0, std::memory_order_release);
+ *bucket = nullptr;
+ return false;
+ } else {
+ *bucket = static_cast<char*>(buckets_) + allocated_bucket_num * bucket_size_;
+ *pending_chain_ptr_finish_value = allocated_bucket_num + 1;
+ return true;
+ }
+ }
+ // Spin until the real "next" pointer is available.
+ while (existing_chain_ptr == std::numeric_limits<std::size_t>::max()) {
+ existing_chain_ptr = (*pending_chain_ptr)->load(std::memory_order_acquire);
+ }
+ if (existing_chain_ptr == 0) {
+ // Other thread had to roll back, so try again.
+ continue;
+ }
+ // Chase the next pointer.
+ *bucket = static_cast<char*>(buckets_) + (existing_chain_ptr - 1) * bucket_size_;
+ *pending_chain_ptr = static_cast<std::atomic<std::size_t>*>(*bucket);
+ if (!allow_duplicate_keys) {
+ const std::size_t hash_in_bucket
+ = *reinterpret_cast<const std::size_t*>(static_cast<const char*>(*bucket)
+ + sizeof(std::atomic<std::size_t>));
+ if (hash_in_bucket == hash_code) {
+ return false;
+ }
+ }
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+inline void FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::writeScalarKeyToBucket(const TypedValue &key,
+ const std::size_t hash_code,
+ void *bucket,
+ HashTablePreallocationState *prealloc_state) {
+ *reinterpret_cast<std::size_t*>(static_cast<char*>(bucket) + sizeof(std::atomic<std::size_t>))
+ = hash_code;
+ key_manager_.writeKeyComponentToBucket(key, 0, bucket, prealloc_state);
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+inline void FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::writeCompositeKeyToBucket(const std::vector<TypedValue> &key,
+ const std::size_t hash_code,
+ void *bucket,
+ HashTablePreallocationState *prealloc_state) {
+ DEBUG_ASSERT(key.size() == this->key_types_.size());
+ *reinterpret_cast<std::size_t*>(static_cast<char*>(bucket) + sizeof(std::atomic<std::size_t>))
+ = hash_code;
+ for (std::size_t idx = 0;
+ idx < this->key_types_.size();
+ ++idx) {
+ key_manager_.writeKeyComponentToBucket(key[idx], idx, bucket, prealloc_state);
+ }
+}
+
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+bool FastSeparateChainingHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>
+ ::isFull(const std::size_t extra_variable_storage) const {
+ if (header_->buckets_allocated.load(std::memory_order_relaxed) >= header_->num_buckets) {
+ // All buckets are allocated.
+ return true;
+ }
+
+ if (extra_variable_storage > 0) {
+ if (extra_variable_storage
+ + header_->variable_length_bytes_allocated.load(std::memory_order_relaxed)
+ > key_manager_.getVariableLengthKeyStorageSize()) {
+ // Not enough variable-length key storage space.
+ return true;
+ }
+ }
+
+ return false;
+}
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_STORAGE_SEPARATE_CHAINING_HASH_TABLE_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/HashTableBase.hpp
----------------------------------------------------------------------
diff --git a/storage/HashTableBase.hpp b/storage/HashTableBase.hpp
index c37c1ec..ebdf464 100644
--- a/storage/HashTableBase.hpp
+++ b/storage/HashTableBase.hpp
@@ -63,7 +63,7 @@ class HashTableBase {
public:
virtual ~HashTableBase() {
}
-
+virtual size_t get_buckets_allocated() const {return 0;}
protected:
HashTableBase() {
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/HashTablePool.hpp
----------------------------------------------------------------------
diff --git a/storage/HashTablePool.hpp b/storage/HashTablePool.hpp
index c16d0f1..909b233 100644
--- a/storage/HashTablePool.hpp
+++ b/storage/HashTablePool.hpp
@@ -25,6 +25,8 @@
#include "expressions/aggregation/AggregationHandle.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
+#include "storage/FastHashTableFactory.hpp"
#include "threading/SpinMutex.hpp"
#include "utility/Macros.hpp"
#include "utility/StringUtil.hpp"
@@ -79,6 +81,19 @@ class HashTablePool {
agg_handle_(DCHECK_NOTNULL(agg_handle)),
storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+ HashTablePool(const std::size_t estimated_num_entries,
+ const HashTableImplType hash_table_impl_type,
+ const std::vector<const Type *> &group_by_types,
+ const std::vector<std::size_t> &payload_sizes,
+ const std::vector<AggregationHandle *> &handles,
+ StorageManager *storage_manager)
+ : estimated_num_entries_(reduceEstimatedCardinality(estimated_num_entries)),
+ hash_table_impl_type_(hash_table_impl_type),
+ group_by_types_(group_by_types),
+ payload_sizes_(payload_sizes),
+ handles_(handles),
+ storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+
/**
* @brief Check out a hash table for insertion.
*
@@ -98,6 +113,20 @@ class HashTablePool {
return createNewHashTable();
}
+ AggregationStateHashTableBase* getHashTableFast() {
+ {
+ SpinMutexLock lock(mutex_);
+ if (!hash_tables_.empty()) {
+ std::unique_ptr<AggregationStateHashTableBase> ret_hash_table(
+ std::move(hash_tables_.back()));
+ hash_tables_.pop_back();
+ DCHECK(ret_hash_table != nullptr);
+ return ret_hash_table.release();
+ }
+ }
+ return createNewHashTableFast();
+ }
+
/**
* @brief Return a previously checked out hash table.
*
@@ -132,6 +161,16 @@ class HashTablePool {
storage_manager_);
}
+ AggregationStateHashTableBase* createNewHashTableFast() {
+ return AggregationStateFastHashTableFactory::CreateResizable(
+ hash_table_impl_type_,
+ group_by_types_,
+ estimated_num_entries_,
+ payload_sizes_,
+ handles_,
+ storage_manager_);
+ }
+
inline std::size_t reduceEstimatedCardinality(
const std::size_t original_estimate) const {
if (original_estimate < kEstimateReductionFactor) {
@@ -151,7 +190,10 @@ class HashTablePool {
const std::vector<const Type *> group_by_types_;
+ std::vector<std::size_t> payload_sizes_;
+
AggregationHandle *agg_handle_;
+ const std::vector<AggregationHandle *> handles_;
StorageManager *storage_manager_;
SpinMutex mutex_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/StorageBlock.cpp
----------------------------------------------------------------------
diff --git a/storage/StorageBlock.cpp b/storage/StorageBlock.cpp
index fdd438d..b9faa2a 100644
--- a/storage/StorageBlock.cpp
+++ b/storage/StorageBlock.cpp
@@ -38,6 +38,7 @@
#include "storage/CompressedPackedRowStoreTupleStorageSubBlock.hpp"
#include "storage/CountedReference.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
#include "storage/IndexSubBlock.hpp"
#include "storage/InsertDestinationInterface.hpp"
#include "storage/PackedRowStoreTupleStorageSubBlock.hpp"
@@ -494,6 +495,92 @@ void StorageBlock::aggregateGroupBy(
hash_table);
}
+
+void StorageBlock::aggregateGroupByFast(
+ const std::vector<std::vector<std::unique_ptr<const Scalar>>> &arguments,
+ const std::vector<std::unique_ptr<const Scalar>> &group_by,
+ const Predicate *predicate,
+ AggregationStateHashTableBase *hash_table,
+ std::unique_ptr<TupleIdSequence> *reuse_matches,
+ std::vector<std::unique_ptr<ColumnVector>> *reuse_group_by_vectors) const {
+ DCHECK_GT(group_by.size(), 0u)
+ << "Called aggregateGroupBy() with zero GROUP BY expressions";
+
+ SubBlocksReference sub_blocks_ref(*tuple_store_,
+ indices_,
+ indices_consistent_);
+
+ // IDs of 'arguments' as attributes in the ValueAccessor we create below.
+ std::vector<attribute_id> arg_ids;
+ std::vector<std::vector<attribute_id>> argument_ids;
+
+ // IDs of GROUP BY key element(s) in the ValueAccessor we create below.
+ std::vector<attribute_id> key_ids;
+
+ // An intermediate ValueAccessor that stores the materialized 'arguments' for
+ // this aggregate, as well as the GROUP BY expression values.
+ ColumnVectorsValueAccessor temp_result;
+ {
+ std::unique_ptr<ValueAccessor> accessor;
+ if (predicate) {
+ if (!*reuse_matches) {
+ // If there is a filter predicate that hasn't already been evaluated,
+ // evaluate it now and save the results for other aggregates on this
+ // same block.
+ reuse_matches->reset(getMatchesForPredicate(predicate));
+ }
+
+ // Create a filtered ValueAccessor that only iterates over predicate
+ // matches.
+ accessor.reset(tuple_store_->createValueAccessor(reuse_matches->get()));
+ } else {
+ // Create a ValueAccessor that iterates over all tuples in this block
+ accessor.reset(tuple_store_->createValueAccessor());
+ }
+
+ attribute_id attr_id = 0;
+
+ // First, put GROUP BY keys into 'temp_result'.
+ if (reuse_group_by_vectors->empty()) {
+ // Compute GROUP BY values from group_by Scalars, and store them in
+ // reuse_group_by_vectors for reuse by other aggregates on this same
+ // block.
+ reuse_group_by_vectors->reserve(group_by.size());
+ for (const std::unique_ptr<const Scalar> &group_by_element : group_by) {
+ reuse_group_by_vectors->emplace_back(
+ group_by_element->getAllValues(accessor.get(), &sub_blocks_ref));
+ temp_result.addColumn(reuse_group_by_vectors->back().get(), false);
+ key_ids.push_back(attr_id++);
+ }
+ } else {
+ // Reuse precomputed GROUP BY values from reuse_group_by_vectors.
+ DCHECK_EQ(group_by.size(), reuse_group_by_vectors->size())
+ << "Wrong number of reuse_group_by_vectors";
+ for (const std::unique_ptr<ColumnVector> &reuse_cv : *reuse_group_by_vectors) {
+ temp_result.addColumn(reuse_cv.get(), false);
+ key_ids.push_back(attr_id++);
+ }
+ }
+
+ // Compute argument vectors and add them to 'temp_result'.
+ for (const std::vector<std::unique_ptr<const Scalar>> &argument : arguments) {
+ arg_ids.clear();
+ for (const std::unique_ptr<const Scalar> &args : argument) {
+ temp_result.addColumn(args->getAllValues(accessor.get(), &sub_blocks_ref));
+ arg_ids.push_back(attr_id++);
+ }
+ argument_ids.push_back(arg_ids);
+ }
+ }
+
+ static_cast<AggregationStateFastHashTable *>(hash_table)->upsertValueAccessorCompositeKeyFast(
+ argument_ids,
+ &temp_result,
+ key_ids,
+ true);
+}
+
+
void StorageBlock::aggregateDistinct(
const AggregationHandle &handle,
const std::vector<std::unique_ptr<const Scalar>> &arguments,
@@ -582,7 +669,6 @@ void StorageBlock::aggregateDistinct(
&temp_result, key_ids, distinctify_hash_table);
}
-
// TODO(chasseur): Vectorization for updates.
StorageBlock::UpdateResult StorageBlock::update(
const unordered_map<attribute_id, unique_ptr<const Scalar>> &assignments,
[3/5] incubator-quickstep git commit: Initial commit for QUICKSTEP-28
and QUICKSTEP-29. Code refactoring and cleanup,
some more optimizations are pending.
Posted by ra...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/FastHashTableFactory.hpp
----------------------------------------------------------------------
diff --git a/storage/FastHashTableFactory.hpp b/storage/FastHashTableFactory.hpp
new file mode 100644
index 0000000..6ad3212
--- /dev/null
+++ b/storage/FastHashTableFactory.hpp
@@ -0,0 +1,300 @@
+/**
+ * Copyright 2015-2016 Pivotal Software, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+#ifndef QUICKSTEP_STORAGE_FAST_HASH_TABLE_FACTORY_HPP_
+#define QUICKSTEP_STORAGE_FAST_HASH_TABLE_FACTORY_HPP_
+
+#include <cstddef>
+#include <string>
+#include <vector>
+
+#include "storage/HashTable.hpp"
+#include "storage/FastHashTable.hpp"
+#include "storage/HashTableBase.hpp"
+#include "storage/HashTableFactory.hpp"
+#include "storage/HashTable.pb.h"
+#include "storage/LinearOpenAddressingHashTable.hpp"
+#include "storage/SeparateChainingHashTable.hpp"
+#include "storage/FastSeparateChainingHashTable.hpp"
+#include "storage/SimpleScalarSeparateChainingHashTable.hpp"
+#include "storage/TupleReference.hpp"
+#include "types/TypeFactory.hpp"
+#include "utility/BloomFilter.hpp"
+#include "utility/Macros.hpp"
+
+#include "glog/logging.h"
+
+namespace quickstep {
+
+class StorageManager;
+class Type;
+
+/** \addtogroup Storage
+ * @{
+ */
+
+/**
+ * @brief Templated all-static factory class that makes it easier to
+ * instantiate HashTables with the particular HashTable implementation
+ * chosen at runtime. All template parameters are exactly the same as
+ * those of HashTable.
+ **/
+template <bool resizable,
+ bool serializable,
+ bool force_key_copy,
+ bool allow_duplicate_keys>
+class FastHashTableFactory {
+ public:
+ /**
+ * @brief Create a new resizable HashTable, with the type selected by
+ * hash_table_type. Other parameters are forwarded to the HashTable's
+ * constructor.
+ *
+ * @param hash_table_type The specific HashTable implementation that should
+ * be used.
+ * @param key_types A vector of one or more types (>1 indicates a composite
+ * key). Forwarded as-is to the HashTable's constructor.
+ * @param num_entries The estimated number of entries the HashTable will
+ * hold. Forwarded as-is to the HashTable's constructor.
+ * @param storage_manager The StorageManager to use (a StorageBlob will be
+ * allocated to hold the HashTable's contents). Forwarded as-is to the
+ * HashTable's constructor.
+ * @return A new resizable HashTable.
+ **/
+ static FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>*
+ CreateResizable(const HashTableImplType hash_table_type,
+ const std::vector<const Type*> &key_types,
+ const std::size_t num_entries,
+ const std::vector<std::size_t> &payload_sizes,
+ const std::vector<AggregationHandle *> &handles,
+ StorageManager *storage_manager) {
+ DCHECK(resizable);
+
+ switch (hash_table_type) {
+ case HashTableImplType::kSeparateChaining:
+ return new FastSeparateChainingHashTable<
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types, num_entries, payload_sizes, handles, storage_manager);
+ case HashTableImplType::kLinearOpenAddressing:
+/* return new LinearOpenAddressingHashTable<
+ ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types, num_entries, storage_manager);*/
+ return new FastSeparateChainingHashTable<
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types, num_entries, payload_sizes, handles, storage_manager);
+ case HashTableImplType::kSimpleScalarSeparateChaining:
+ return new FastSeparateChainingHashTable<
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types, num_entries, payload_sizes, handles, storage_manager);
+/* return new SimpleScalarSeparateChainingHashTable<
+ ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types, num_entries, storage_manager);*/
+ default: {
+ LOG(FATAL) << "Unrecognized HashTableImplType in HashTableFactory::createResizable()\n";
+ }
+ }
+ }
+
+ /**
+ * @brief Create a new fixed-sized HashTable, with the type selected by
+ * hash_table_type. Other parameters are forwarded to the HashTables's
+ * constructor.
+ *
+ * @param hash_table_type The specific HashTable implementation that should
+ * be used.
+ * @param key_types A vector of one or more types (>1 indicates a composite
+ * key). Forwarded as-is to the HashTable's constructor.
+ * @param hash_table_memory A pointer to memory to use for the HashTable.
+ * Forwarded as-is to the HashTable's constructor.
+ * @param hash_table_memory_size The size of hash_table_memory in bytes.
+ * Forwarded as-is to the HashTable's constructor.
+ * @param new_hash_table If true, the HashTable is being constructed for the
+ * first time and hash_table_memory will be cleared. If false, reload
+ * a pre-existing HashTable. Forwarded as-is to the HashTable's
+ * constructor.
+ * @param hash_table_memory_zeroed If new_hash_table is true, setting this to
+ * true means that the HashTable will assume that hash_table_memory
+ * has already been zeroed-out (any newly-allocated block or blob
+ * memory from StorageManager is zeroed-out). If false, the HashTable
+ * will explicitly zero-fill its memory as neccessary. This parameter
+ * has no effect when new_hash_table is false. Forwarded as-is to the
+ * HashTable's constructor.
+ * @return A new (or reloaded) fixed-size HashTable.
+ **/
+ static FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>*
+ CreateFixedSize(const HashTableImplType hash_table_type,
+ const std::vector<const Type*> &key_types,
+ void *hash_table_memory,
+ const std::size_t hash_table_memory_size,
+ const bool new_hash_table,
+ const bool hash_table_memory_zeroed) {
+ DCHECK(!resizable);
+
+ switch (hash_table_type) {
+ case HashTableImplType::kSeparateChaining:
+ return new SeparateChainingHashTable<
+ int,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types,
+ hash_table_memory,
+ hash_table_memory_size,
+ new_hash_table,
+ hash_table_memory_zeroed);
+ case HashTableImplType::kLinearOpenAddressing:
+/* return new LinearOpenAddressingHashTable<
+ ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types,
+ hash_table_memory,
+ hash_table_memory_size,
+ new_hash_table,
+ hash_table_memory_zeroed);*/
+ case HashTableImplType::kSimpleScalarSeparateChaining:
+/* return new SimpleScalarSeparateChainingHashTable<
+ ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>(key_types,
+ hash_table_memory,
+ hash_table_memory_size,
+ new_hash_table,
+ hash_table_memory_zeroed);*/
+ default: {
+ LOG(FATAL) << "Unrecognized HashTableImplType\n";
+ }
+ }
+ }
+
+ /**
+ * @brief Check whether a serialization::HashTable describing a resizable
+ * HashTable is fully-formed and all parts are valid.
+ *
+ * @param proto A serialized Protocol Buffer description of a HashTable,
+ * originally generated by the optimizer.
+ * @return Whether proto is fully-formed and valid.
+ **/
+ static bool ProtoIsValid(const serialization::HashTable &proto) {
+ if (!proto.IsInitialized() ||
+ !serialization::HashTableImplType_IsValid(
+ proto.hash_table_impl_type())) {
+ return false;
+ }
+
+ for (int i = 0; i < proto.key_types_size(); ++i) {
+ if (!TypeFactory::ProtoIsValid(proto.key_types(i))) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * @brief Create a new resizable HashTable according to a protobuf
+ * description.
+ *
+ * @param proto A protobuf description of a resizable HashTable.
+ * @param storage_manager The StorageManager to use (a StorageBlob will be
+ * allocated to hold the HashTable's contents).
+ * @param bloom_filters A vector of pointers to bloom filters that may be used
+ * during hash table construction in build/probe phase.
+ * @return A new resizable HashTable with parameters specified by proto.
+ **/
+ static FastHashTable<resizable, serializable, force_key_copy, allow_duplicate_keys>*
+ CreateResizableFromProto(const serialization::HashTable &proto,
+ StorageManager *storage_manager,
+ const std::vector<std::unique_ptr<BloomFilter>> &bloom_filters) {
+ DCHECK(ProtoIsValid(proto))
+ << "Attempted to create HashTable from invalid proto description:\n"
+ << proto.DebugString();
+
+ std::vector<const Type*> key_types;
+ for (int i = 0; i < proto.key_types_size(); ++i) {
+ key_types.emplace_back(&TypeFactory::ReconstructFromProto(proto.key_types(i)));
+ }
+
+ auto hash_table = CreateResizable(HashTableImplTypeFromProto(proto.hash_table_impl_type()),
+ key_types,
+ proto.estimated_num_entries(),
+ storage_manager);
+
+ // TODO(ssaurabh): These lazy initializations can be moved from here and pushed to the
+ // individual implementations of the hash table constructors.
+
+ // Check if there are any build side bloom filter defined on the hash table.
+ if (proto.build_side_bloom_filter_id_size() > 0) {
+ hash_table->enableBuildSideBloomFilter();
+ hash_table->setBuildSideBloomFilter(bloom_filters[proto.build_side_bloom_filter_id(0)].get());
+ }
+
+ // Check if there are any probe side bloom filters defined on the hash table.
+ if (proto.probe_side_bloom_filters_size() > 0) {
+ hash_table->enableProbeSideBloomFilter();
+ // Add as many probe bloom filters as defined by the proto.
+ for (int j = 0; j < proto.probe_side_bloom_filters_size(); ++j) {
+ // Add the pointer to the probe bloom filter within the list of probe bloom filters to use.
+ const auto probe_side_bloom_filter = proto.probe_side_bloom_filters(j);
+ hash_table->addProbeSideBloomFilter(bloom_filters[probe_side_bloom_filter.probe_side_bloom_filter_id()].get());
+
+ // Add the attribute ids corresponding to this probe bloom filter.
+ std::vector<attribute_id> probe_attribute_ids;
+ for (int k = 0; k < probe_side_bloom_filter.probe_side_attr_ids_size(); ++k) {
+ const attribute_id probe_attribute_id = probe_side_bloom_filter.probe_side_attr_ids(k);
+ probe_attribute_ids.push_back(probe_attribute_id);
+ }
+ hash_table->addProbeSideAttributeIds(std::move(probe_attribute_ids));
+ }
+ }
+
+ return hash_table;
+ }
+
+ private:
+ // Class is all-static and should not be instantiated.
+ FastHashTableFactory();
+
+ DISALLOW_COPY_AND_ASSIGN(FastHashTableFactory);
+};
+
+/**
+ * @brief Convenient alias that provides a HashTableFactory whose only template
+ * parameter is the aggregate state type.
+ **/
+using AggregationStateFastHashTableFactory
+ = FastHashTableFactory<true, false, true, false>;
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_STORAGE_HASH_TABLE_FACTORY_HPP_
[5/5] incubator-quickstep git commit: Initial commit for QUICKSTEP-28
and QUICKSTEP-29. Code refactoring and cleanup,
some more optimizations are pending.
Posted by ra...@apache.org.
Initial commit for QUICKSTEP-28 and QUICKSTEP-29. Code refactoring and cleanup, some more optimizations are pending.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/d5e7d09f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/d5e7d09f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/d5e7d09f
Branch: refs/heads/quickstep-28-29
Commit: d5e7d09fc704a33fb761bc818bac01ead1c273de
Parents: 040a511
Author: rathijit <ra...@node-2.hashtable.quickstep-pg0.wisc.cloudlab.us>
Authored: Mon Jul 4 02:44:48 2016 -0500
Committer: rathijit <ra...@node-2.hashtable.quickstep-pg0.wisc.cloudlab.us>
Committed: Mon Jul 4 02:44:48 2016 -0500
----------------------------------------------------------------------
.../aggregation/AggregationConcreteHandle.cpp | 29 +-
.../aggregation/AggregationConcreteHandle.hpp | 223 ++
expressions/aggregation/AggregationHandle.hpp | 8 +-
.../aggregation/AggregationHandleAvg.cpp | 40 +-
.../aggregation/AggregationHandleAvg.hpp | 62 +-
.../aggregation/AggregationHandleCount.cpp | 38 +-
.../aggregation/AggregationHandleCount.hpp | 50 +-
.../aggregation/AggregationHandleDistinct.cpp | 2 +-
.../aggregation/AggregationHandleDistinct.hpp | 2 +-
.../aggregation/AggregationHandleMax.cpp | 29 +-
.../aggregation/AggregationHandleMax.hpp | 39 +-
.../aggregation/AggregationHandleMin.cpp | 30 +-
.../aggregation/AggregationHandleMin.hpp | 44 +-
.../aggregation/AggregationHandleSum.cpp | 31 +-
.../aggregation/AggregationHandleSum.hpp | 52 +-
expressions/aggregation/CMakeLists.txt | 7 +
storage/AggregationOperationState.cpp | 95 +-
storage/AggregationOperationState.hpp | 7 +-
storage/CMakeLists.txt | 58 +
storage/FastHashTable.hpp | 2640 ++++++++++++++++++
storage/FastHashTableFactory.hpp | 300 ++
storage/FastSeparateChainingHashTable.hpp | 1761 ++++++++++++
storage/HashTableBase.hpp | 2 +-
storage/HashTablePool.hpp | 42 +
storage/StorageBlock.cpp | 88 +-
storage/StorageBlock.hpp | 8 +
threading/SpinMutex.hpp | 2 +
27 files changed, 5587 insertions(+), 102 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationConcreteHandle.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationConcreteHandle.cpp b/expressions/aggregation/AggregationConcreteHandle.cpp
index 05ca58d..d808302 100644
--- a/expressions/aggregation/AggregationConcreteHandle.cpp
+++ b/expressions/aggregation/AggregationConcreteHandle.cpp
@@ -22,6 +22,7 @@
#include "catalog/CatalogTypedefs.hpp"
#include "storage/HashTable.hpp"
+#include "storage/FastHashTable.hpp"
#include "storage/HashTableFactory.hpp"
namespace quickstep {
@@ -49,22 +50,24 @@ void AggregationConcreteHandle::insertValueAccessorIntoDistinctifyHashTable(
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.
- const auto noop_upserter = [](const auto &accessor, const bool *value) -> void {};
+// const auto noop_upserter = [](const auto &accessor, const bool *value) -> void {};
- AggregationStateHashTable<bool> *hash_table =
- static_cast<AggregationStateHashTable<bool>*>(distinctify_hash_table);
+ AggregationStateFastHashTable *hash_table =
+ static_cast<AggregationStateFastHashTable *>(distinctify_hash_table);
if (key_ids.size() == 1) {
- hash_table->upsertValueAccessor(accessor,
- key_ids[0],
- true /* check_for_null_keys */,
- true /* initial_value */,
- &noop_upserter);
+// TODO(rathijit): fix
+// hash_table->upsertValueAccessor(accessor,
+// key_ids[0],
+// true /* check_for_null_keys */,
+// true /* initial_value */,
+// &noop_upserter);
} else {
- hash_table->upsertValueAccessorCompositeKey(accessor,
- key_ids,
- true /* check_for_null_keys */,
- true /* initial_value */,
- &noop_upserter);
+ std::vector<std::vector<attribute_id>> empty_args;
+ empty_args.resize(1);
+ hash_table->upsertValueAccessorCompositeKeyFast(empty_args,
+ accessor,
+ key_ids,
+ true /* check_for_null_keys */);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationConcreteHandle.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationConcreteHandle.hpp b/expressions/aggregation/AggregationConcreteHandle.hpp
index 0267e17..c5d7b3c 100644
--- a/expressions/aggregation/AggregationConcreteHandle.hpp
+++ b/expressions/aggregation/AggregationConcreteHandle.hpp
@@ -31,6 +31,7 @@
#include "types/TypedValue.hpp"
#include "types/containers/ColumnVector.hpp"
#include "utility/Macros.hpp"
+#include "threading/SpinMutex.hpp"
#include "glog/logging.h"
@@ -79,6 +80,37 @@ class HashTableStateUpserter {
DISALLOW_COPY_AND_ASSIGN(HashTableStateUpserter);
};
+template <typename HandleT>
+class HashTableStateUpserterFast {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param handle The aggregation handle being used.
+ * @param source_state The aggregation state in the source aggregation hash
+ * table. The corresponding state (for the same key) in the destination
+ * hash table will be upserted.
+ **/
+ HashTableStateUpserterFast(const HandleT &handle, const uint8_t *source_state)
+ : handle_(handle), source_state_(source_state) {}
+
+ /**
+ * @brief The operator for the functor required for the upsert.
+ *
+ * @param destination_state The aggregation state in the aggregation hash
+ * table that is being upserted.
+ **/
+ void operator()(uint8_t *destination_state) {
+ handle_.mergeStatesFast(source_state_, destination_state);
+ }
+
+ private:
+ const HandleT &handle_;
+ const uint8_t *source_state_;
+
+ DISALLOW_COPY_AND_ASSIGN(HashTableStateUpserterFast);
+};
+
/**
* @brief A class to support the functor for merging group by hash tables.
**/
@@ -129,6 +161,53 @@ class HashTableMerger {
DISALLOW_COPY_AND_ASSIGN(HashTableMerger);
};
+template <typename HandleT, typename HashTableT>
+class HashTableMergerFast {
+ public:
+ /**
+ * @brief Constructor
+ *
+ * @param handle The Aggregation handle being used.
+ * @param destination_hash_table The destination hash table to which other
+ * hash tables will be merged.
+ **/
+ HashTableMergerFast(const HandleT &handle,
+ AggregationStateHashTableBase *destination_hash_table)
+ : handle_(handle),
+ destination_hash_table_(
+ static_cast<HashTableT *>(destination_hash_table)) {}
+
+ /**
+ * @brief The operator for the functor.
+ *
+ * @param group_by_key The group by key being merged.
+ * @param source_state The aggregation state for the given key in the source
+ * aggregation hash table.
+ **/
+ inline void operator()(const std::vector<TypedValue> &group_by_key,
+ const uint8_t *source_state) {
+ const uint8_t *original_state =
+ destination_hash_table_->getSingleCompositeKey(group_by_key);
+ if (original_state != nullptr) {
+ HashTableStateUpserterFast<HandleT> upserter(
+ handle_, source_state);
+ // The CHECK is required as upsertCompositeKey can return false if the
+ // hash table runs out of space during the upsert process. The ideal
+ // solution will be to retry again if the upsert fails.
+ CHECK(destination_hash_table_->upsertCompositeKeyFast(
+ group_by_key, original_state, &upserter));
+ } else {
+ destination_hash_table_->putCompositeKeyFast(group_by_key, source_state);
+ }
+ }
+
+ private:
+ const HandleT &handle_;
+ HashTableT *destination_hash_table_;
+
+ DISALLOW_COPY_AND_ASSIGN(HashTableMergerFast);
+};
+
/**
* @brief The helper intermediate subclass of AggregationHandle that provides
* virtual method implementations as well as helper methods that are
@@ -208,11 +287,26 @@ class AggregationConcreteHandle : public AggregationHandle {
template <typename HandleT,
typename HashTableT>
+ void aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast(
+ const AggregationStateHashTableBase &distinctify_hash_table,
+ AggregationStateHashTableBase *hash_table) const;
+
+
+ template <typename HandleT,
+ typename HashTableT>
ColumnVector* finalizeHashTableHelper(
const Type &result_type,
const AggregationStateHashTableBase &hash_table,
std::vector<std::vector<TypedValue>> *group_by_keys) const;
+ template <typename HandleT,
+ typename HashTableT>
+ ColumnVector* finalizeHashTableHelperFast(
+ const Type &result_type,
+ const AggregationStateHashTableBase &hash_table,
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const;
+
template <typename HandleT, typename HashTableT>
inline TypedValue finalizeGroupInHashTable(
const AggregationStateHashTableBase &hash_table,
@@ -224,11 +318,29 @@ class AggregationConcreteHandle : public AggregationHandle {
return static_cast<const HandleT*>(this)->finalizeHashTableEntry(*group_state);
}
+ template <typename HandleT, typename HashTableT>
+ inline TypedValue finalizeGroupInHashTableFast(
+ const AggregationStateHashTableBase &hash_table,
+ const std::vector<TypedValue> &group_key,
+ int index) const {
+ const std::uint8_t *group_state
+ = static_cast<const HashTableT&>(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);
+ }
+
template <typename HandleT, typename StateT, typename HashTableT>
void mergeGroupByHashTablesHelper(
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const;
+ template <typename HandleT, typename HashTableT>
+ void mergeGroupByHashTablesHelperFast(
+ const AggregationStateHashTableBase &source_hash_table,
+ AggregationStateHashTableBase *destination_hash_table) const;
+
+
private:
DISALLOW_COPY_AND_ASSIGN(AggregationConcreteHandle);
};
@@ -302,6 +414,12 @@ class HashTableAggregateFinalizer {
output_column_vector_->appendTypedValue(handle_.finalizeHashTableEntry(group_state));
}
+ inline void operator()(const std::vector<TypedValue> &group_by_key,
+ const unsigned char *byte_ptr) {
+ group_by_keys_->emplace_back(group_by_key);
+ output_column_vector_->appendTypedValue(handle_.finalizeHashTableEntryFast(byte_ptr));
+ }
+
private:
const HandleT &handle_;
std::vector<std::vector<TypedValue>> *group_by_keys_;
@@ -414,6 +532,42 @@ void AggregationConcreteHandle::aggregateOnDistinctifyHashTableForGroupByUnaryHe
template <typename HandleT,
typename HashTableT>
+void AggregationConcreteHandle::aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast(
+ const AggregationStateHashTableBase &distinctify_hash_table,
+ AggregationStateHashTableBase *aggregation_hash_table) const {
+ const HandleT& handle = static_cast<const HandleT&>(*this);
+ HashTableT *target_hash_table = static_cast<HashTableT*>(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](
+ std::vector<TypedValue> &key,
+ const bool &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.
+ const TypedValue argument(std::move(key.back()));
+ key.pop_back();
+
+ // 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+sizeof(SpinMutex));
+ };
+
+ target_hash_table->upsertCompositeKeyFast(key, nullptr, &upserter);
+ };
+
+ const HashTableT &source_hash_table =
+ static_cast<const HashTableT&>(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);
+}
+
+
+template <typename HandleT,
+ typename HashTableT>
ColumnVector* AggregationConcreteHandle::finalizeHashTableHelper(
const Type &result_type,
const AggregationStateHashTableBase &hash_table,
@@ -463,6 +617,59 @@ ColumnVector* AggregationConcreteHandle::finalizeHashTableHelper(
}
template <typename HandleT,
+ typename HashTableT>
+ColumnVector* AggregationConcreteHandle::finalizeHashTableHelperFast(
+ const Type &result_type,
+ const AggregationStateHashTableBase &hash_table,
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const {
+ const HandleT &handle = static_cast<const HandleT&>(*this);
+ const HashTableT &hash_table_concrete = static_cast<const HashTableT&>(hash_table);
+
+ if (group_by_keys->empty()) {
+ if (NativeColumnVector::UsableForType(result_type)) {
+ NativeColumnVector *result = new NativeColumnVector(result_type,
+ hash_table_concrete.numEntries());
+ HashTableAggregateFinalizer<HandleT, NativeColumnVector> finalizer(
+ handle,
+ group_by_keys,
+ result);
+ hash_table_concrete.forEachCompositeKeyFast(&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);
+ return result;
+ }
+ } else {
+ if (NativeColumnVector::UsableForType(result_type)) {
+ NativeColumnVector *result = new NativeColumnVector(result_type,
+ group_by_keys->size());
+ for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
+ result->appendTypedValue(finalizeGroupInHashTableFast<HandleT, HashTableT>(hash_table,
+ group_by_key,
+ index));
+ }
+ return result;
+ } else {
+ IndirectColumnVector *result = new IndirectColumnVector(result_type,
+ hash_table_concrete.numEntries());
+ for (const std::vector<TypedValue> &group_by_key : *group_by_keys) {
+ result->appendTypedValue(finalizeGroupInHashTableFast<HandleT, HashTableT>(hash_table,
+ group_by_key,
+ index));
+ }
+ return result;
+ }
+ }
+}
+
+template <typename HandleT,
typename StateT,
typename HashTableT>
void AggregationConcreteHandle::mergeGroupByHashTablesHelper(
@@ -478,6 +685,22 @@ void AggregationConcreteHandle::mergeGroupByHashTablesHelper(
source_hash_table_concrete.forEachCompositeKey(&merger);
}
+template <typename HandleT,
+ typename HashTableT>
+void AggregationConcreteHandle::mergeGroupByHashTablesHelperFast(
+ const AggregationStateHashTableBase &source_hash_table,
+ AggregationStateHashTableBase *destination_hash_table) const {
+ const HandleT &handle = static_cast<const HandleT &>(*this);
+ const HashTableT &source_hash_table_concrete =
+ static_cast<const HashTableT &>(source_hash_table);
+
+ HashTableMergerFast<HandleT, HashTableT> merger(handle,
+ destination_hash_table);
+
+ source_hash_table_concrete.forEachCompositeKeyFast(&merger);
+}
+
+
} // namespace quickstep
#endif // QUICKSTEP_EXPRESSIONS_AGGREGATION_AGGREGATION_CONCRETE_HANDLE_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandle.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandle.hpp b/expressions/aggregation/AggregationHandle.hpp
index cdebb03..d9c3897 100644
--- a/expressions/aggregation/AggregationHandle.hpp
+++ b/expressions/aggregation/AggregationHandle.hpp
@@ -265,7 +265,7 @@ class AggregationHandle {
**/
virtual ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const = 0;
+ std::vector<std::vector<TypedValue>> *group_by_keys, int index) const = 0;
/**
* @brief Create a new HashTable for the distinctify step for DISTINCT aggregation.
@@ -362,6 +362,12 @@ class AggregationHandle {
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const = 0;
+ virtual size_t getPayloadSize() const {return 8;}
+ virtual void setPayloadOffset(std::size_t) {}
+ virtual void iterateInlFast(const std::vector<TypedValue> &arguments, uint8_t *byte_ptr) {}
+ virtual void mergeStatesFast(const uint8_t *src, uint8_t *dst) const {}
+ virtual void initPayload(uint8_t *byte_ptr) {}
+
protected:
AggregationHandle() {
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleAvg.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleAvg.cpp b/expressions/aggregation/AggregationHandleAvg.cpp
index 42a2fb9..d0bd3b8 100644
--- a/expressions/aggregation/AggregationHandleAvg.cpp
+++ b/expressions/aggregation/AggregationHandleAvg.cpp
@@ -137,8 +137,7 @@ void AggregationHandleAvg::aggregateValueAccessorIntoHashTable(
AggregationStateHashTableBase *hash_table) const {
DCHECK_EQ(1u, argument_ids.size())
<< "Got wrong number of arguments for AVG: " << argument_ids.size();
-
- aggregateValueAccessorIntoHashTableUnaryHelper<
+/* aggregateValueAccessorIntoHashTableUnaryHelper<
AggregationHandleAvg,
AggregationStateAvg,
AggregationStateHashTable<AggregationStateAvg>>(
@@ -146,7 +145,14 @@ void AggregationHandleAvg::aggregateValueAccessorIntoHashTable(
argument_ids.front(),
group_by_key_ids,
blank_state_,
- hash_table);
+ hash_table); */
+
+/* static_cast<AggregationStateFastHashTable *>(hash_table)->upsertValueAccessorCompositeKeyFast(
+ argument_ids.front(),
+ accessor,
+ group_by_key_ids,
+ true,
+ const_cast<AggregationHandleAvg *>(this));*/
}
void AggregationHandleAvg::mergeStates(
@@ -161,6 +167,19 @@ void AggregationHandleAvg::mergeStates(
avg_source.sum_);
}
+void AggregationHandleAvg::mergeStatesFast(
+ const uint8_t *source,
+ 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 *>(source + blank_state_.count_offset);
+ TypedValue *dst_sum_ptr = reinterpret_cast<TypedValue *>(destination+blank_state_.sum_offset);
+ std::int64_t *dst_count_ptr = reinterpret_cast<std::int64_t *>(destination + blank_state_.count_offset);
+ (*dst_count_ptr) += (*src_count_ptr);
+ *dst_sum_ptr = merge_add_operator_->applyToTypedValues(*dst_sum_ptr, *src_sum_ptr);
+}
+
+
+
TypedValue AggregationHandleAvg::finalize(const AggregationState &state) const {
const AggregationStateAvg &agg_state = static_cast<const AggregationStateAvg&>(state);
if (agg_state.count_ == 0) {
@@ -175,12 +194,14 @@ TypedValue AggregationHandleAvg::finalize(const AggregationState &state) const {
ColumnVector* AggregationHandleAvg::finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const {
- return finalizeHashTableHelper<AggregationHandleAvg,
- AggregationStateHashTable<AggregationStateAvg>>(
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const {
+ return finalizeHashTableHelperFast<AggregationHandleAvg,
+ AggregationStateFastHashTable>(
*result_type_,
hash_table,
- group_by_keys);
+ group_by_keys,
+ index);
}
AggregationState* AggregationHandleAvg::aggregateOnDistinctifyHashTableForSingle(
@@ -206,9 +227,8 @@ void AggregationHandleAvg::aggregateOnDistinctifyHashTableForGroupBy(
void AggregationHandleAvg::mergeGroupByHashTables(
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const {
- mergeGroupByHashTablesHelper<AggregationHandleAvg,
- AggregationStateAvg,
- AggregationStateHashTable<AggregationStateAvg>>(
+ mergeGroupByHashTablesHelperFast<AggregationHandleAvg,
+ AggregationStateFastHashTable>(
source_hash_table, destination_hash_table);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleAvg.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleAvg.hpp b/expressions/aggregation/AggregationHandleAvg.hpp
index 4ad4b21..fe178f6 100644
--- a/expressions/aggregation/AggregationHandleAvg.hpp
+++ b/expressions/aggregation/AggregationHandleAvg.hpp
@@ -29,6 +29,7 @@
#include "expressions/aggregation/AggregationConcreteHandle.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
#include "threading/SpinMutex.hpp"
#include "types/Type.hpp"
#include "types/TypedValue.hpp"
@@ -57,7 +58,10 @@ class AggregationStateAvg : public AggregationState {
*/
AggregationStateAvg(const AggregationStateAvg &orig)
: sum_(orig.sum_),
- count_(orig.count_) {
+ count_(orig.count_),
+ sum_offset(orig.sum_offset),
+ count_offset(orig.count_offset),
+ mutex_offset(orig.mutex_offset) {
}
/**
@@ -65,11 +69,19 @@ class AggregationStateAvg : public AggregationState {
*/
~AggregationStateAvg() override {}
+ size_t getPayloadSize() const {
+ size_t p1 = reinterpret_cast<size_t>(&sum_);
+ size_t p2 = reinterpret_cast<size_t>(&mutex_);
+ return (p2-p1);
+ }
+
private:
friend class AggregationHandleAvg;
AggregationStateAvg()
- : sum_(0), count_(0) {
+ : sum_(0), count_(0), sum_offset(0),
+ count_offset(reinterpret_cast<uint8_t *>(&count_)-reinterpret_cast<uint8_t *>(&sum_)),
+ mutex_offset(reinterpret_cast<uint8_t *>(&mutex_)-reinterpret_cast<uint8_t *>(&sum_)) {
}
// TODO(shoban): We might want to specialize sum_ and count_ to use atomics
@@ -77,6 +89,8 @@ class AggregationStateAvg : public AggregationState {
TypedValue sum_;
std::int64_t count_;
SpinMutex mutex_;
+
+ int sum_offset, count_offset, mutex_offset;
};
/**
@@ -109,6 +123,26 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
++state->count_;
}
+ inline void iterateUnaryInlFast(const TypedValue &value, uint8_t *byte_ptr) {
+ DCHECK(value.isPlausibleInstanceOf(argument_type_.getSignature()));
+ if (value.isNull()) return;
+ TypedValue *sum_ptr = reinterpret_cast<TypedValue *>(byte_ptr + blank_state_.sum_offset);
+ std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(byte_ptr + blank_state_.count_offset);
+ *sum_ptr = fast_add_operator_->applyToTypedValues(*sum_ptr, value);
+ ++(*count_ptr);
+ }
+
+ inline void iterateInlFast(const std::vector<TypedValue> &arguments, uint8_t *byte_ptr) override {
+ iterateUnaryInlFast(arguments.front(), byte_ptr);
+ }
+
+ void initPayload(uint8_t *byte_ptr) override {
+ TypedValue *sum_ptr = reinterpret_cast<TypedValue *>(byte_ptr + blank_state_.sum_offset);
+ std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(byte_ptr + blank_state_.count_offset);
+ *sum_ptr = blank_state_.sum_;
+ *count_ptr = blank_state_.count_;
+ }
+
AggregationState* accumulateColumnVectors(
const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const override;
@@ -127,6 +161,9 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
void mergeStates(const AggregationState &source,
AggregationState *destination) const override;
+ void mergeStatesFast(const uint8_t *source,
+ uint8_t *destination) const override;
+
TypedValue finalize(const AggregationState &state) const override;
inline TypedValue finalizeHashTableEntry(const AggregationState &state) const {
@@ -139,9 +176,24 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
TypedValue(static_cast<double>(agg_state.count_)));
}
+ inline TypedValue finalizeHashTableEntryFast(const uint8_t *byte_ptr) 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.
+
+ uint8_t *value_ptr = const_cast<uint8_t*>(byte_ptr);
+ TypedValue *sum_ptr = reinterpret_cast<TypedValue *>(value_ptr + blank_state_.sum_offset);
+ std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(value_ptr + blank_state_.count_offset);
+ return divide_operator_->applyToTypedValues(*sum_ptr,
+ TypedValue(static_cast<double>(*count_ptr)));
+ }
+
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const override;
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const override;
/**
* @brief Implementation of AggregationHandle::aggregateOnDistinctifyHashTableForSingle()
@@ -162,6 +214,10 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const override;
+ size_t getPayloadSize() const override {
+ return blank_state_.getPayloadSize();
+ }
+
private:
friend class AggregateFunctionAvg;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleCount.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleCount.cpp b/expressions/aggregation/AggregationHandleCount.cpp
index 964b7c2..dc11dac 100644
--- a/expressions/aggregation/AggregationHandleCount.cpp
+++ b/expressions/aggregation/AggregationHandleCount.cpp
@@ -135,18 +135,18 @@ template <bool count_star, bool nullable_type>
if (count_star) {
DCHECK_EQ(0u, argument_ids.size())
<< "Got wrong number of arguments for COUNT(*): " << argument_ids.size();
- aggregateValueAccessorIntoHashTableNullaryHelper<
+/* aggregateValueAccessorIntoHashTableNullaryHelper<
AggregationHandleCount<count_star, nullable_type>,
AggregationStateCount,
AggregationStateHashTable<AggregationStateCount>>(
accessor,
group_by_key_ids,
AggregationStateCount(),
- hash_table);
+ hash_table);*/
} else {
DCHECK_EQ(1u, argument_ids.size())
<< "Got wrong number of arguments for COUNT: " << argument_ids.size();
- aggregateValueAccessorIntoHashTableUnaryHelper<
+/* aggregateValueAccessorIntoHashTableUnaryHelper<
AggregationHandleCount<count_star, nullable_type>,
AggregationStateCount,
AggregationStateHashTable<AggregationStateCount>>(
@@ -154,7 +154,7 @@ template <bool count_star, bool nullable_type>
argument_ids.front(),
group_by_key_ids,
AggregationStateCount(),
- hash_table);
+ hash_table); */
}
}
@@ -170,14 +170,25 @@ template <bool count_star, bool nullable_type>
}
template <bool count_star, bool nullable_type>
+void AggregationHandleCount<count_star, nullable_type>::mergeStatesFast(
+ const uint8_t *source,
+ uint8_t *destination) const {
+ const std::int64_t *src_count_ptr = reinterpret_cast<const std::int64_t *>(source);
+ std::int64_t *dst_count_ptr = reinterpret_cast<std::int64_t *>(destination);
+ (*dst_count_ptr) += (*src_count_ptr);
+}
+
+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) const {
- return finalizeHashTableHelper<AggregationHandleCount<count_star, nullable_type>,
- AggregationStateHashTable<AggregationStateCount>>(
+ 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);
+ group_by_keys,
+ index);
}
template <bool count_star, bool nullable_type>
@@ -197,12 +208,10 @@ void AggregationHandleCount<count_star, nullable_type>
const AggregationStateHashTableBase &distinctify_hash_table,
AggregationStateHashTableBase *aggregation_hash_table) const {
DCHECK_EQ(count_star, false);
- aggregateOnDistinctifyHashTableForGroupByUnaryHelper<
+ aggregateOnDistinctifyHashTableForGroupByUnaryHelperFast<
AggregationHandleCount<count_star, nullable_type>,
- AggregationStateCount,
- AggregationStateHashTable<AggregationStateCount>>(
+ AggregationStateFastHashTable>(
distinctify_hash_table,
- AggregationStateCount(),
aggregation_hash_table);
}
@@ -210,10 +219,9 @@ template <bool count_star, bool nullable_type>
void AggregationHandleCount<count_star, nullable_type>::mergeGroupByHashTables(
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const {
- mergeGroupByHashTablesHelper<
+ mergeGroupByHashTablesHelperFast<
AggregationHandleCount,
- AggregationStateCount,
- AggregationStateHashTable<AggregationStateCount>>(source_hash_table,
+ AggregationStateFastHashTable>(source_hash_table,
destination_hash_table);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleCount.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleCount.hpp b/expressions/aggregation/AggregationHandleCount.hpp
index 50138b9..4046106 100644
--- a/expressions/aggregation/AggregationHandleCount.hpp
+++ b/expressions/aggregation/AggregationHandleCount.hpp
@@ -30,6 +30,7 @@
#include "expressions/aggregation/AggregationConcreteHandle.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
#include "types/TypedValue.hpp"
#include "utility/Macros.hpp"
@@ -62,6 +63,10 @@ class AggregationStateCount : public AggregationState {
*/
~AggregationStateCount() override {}
+ size_t getPayloadSize() const {
+ return sizeof(count_);
+ }
+
private:
friend class AggregationHandleCount<false, false>;
friend class AggregationHandleCount<false, true>;
@@ -108,6 +113,11 @@ class AggregationHandleCount : public AggregationConcreteHandle {
state->count_.fetch_add(1, std::memory_order_relaxed);
}
+ inline void iterateNullaryInlFast(uint8_t *byte_ptr) {
+ std::int64_t *count_ptr = reinterpret_cast<std::int64_t *>(byte_ptr);
+ (*count_ptr)++;
+ }
+
/**
* @brief Iterate with count aggregation state.
*/
@@ -117,6 +127,25 @@ class AggregationHandleCount : public AggregationConcreteHandle {
}
}
+ inline void iterateUnaryInlFast(const TypedValue &value, 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 iterateInlFast(const std::vector<TypedValue> &arguments, uint8_t *byte_ptr) override {
+ if (arguments.size())
+ iterateUnaryInlFast(arguments.front(), byte_ptr);
+ else
+ iterateNullaryInlFast(byte_ptr);
+ }
+
+ void initPayload(uint8_t *byte_ptr) 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);
}
@@ -139,6 +168,9 @@ class AggregationHandleCount : public AggregationConcreteHandle {
void mergeStates(const AggregationState &source,
AggregationState *destination) const override;
+ void mergeStatesFast(const uint8_t *source,
+ 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));
}
@@ -147,9 +179,21 @@ class AggregationHandleCount : public AggregationConcreteHandle {
return TypedValue(static_cast<const AggregationStateCount&>(state).count_.load(std::memory_order_relaxed));
}
+ inline TypedValue finalizeHashTableEntryFast(const uint8_t *byte_ptr) 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.
+
+ const std::int64_t *count_ptr = reinterpret_cast<const std::int64_t *>(byte_ptr);
+ return TypedValue(*count_ptr);
+ }
+
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const override;
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const override;
/**
* @brief Implementation of AggregationHandle::aggregateOnDistinctifyHashTableForSingle()
@@ -170,6 +214,10 @@ class AggregationHandleCount : public AggregationConcreteHandle {
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const override;
+ size_t getPayloadSize() const override {
+ return sizeof(std::int64_t);
+ }
+
private:
friend class AggregateFunctionCount;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleDistinct.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleDistinct.cpp b/expressions/aggregation/AggregationHandleDistinct.cpp
index fe8ffcf..0088239 100644
--- a/expressions/aggregation/AggregationHandleDistinct.cpp
+++ b/expressions/aggregation/AggregationHandleDistinct.cpp
@@ -63,7 +63,7 @@ void AggregationHandleDistinct::aggregateValueAccessorIntoHashTable(
ColumnVector* AggregationHandleDistinct::finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const {
+ 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,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleDistinct.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleDistinct.hpp b/expressions/aggregation/AggregationHandleDistinct.hpp
index 6342c2b..1e36845 100644
--- a/expressions/aggregation/AggregationHandleDistinct.hpp
+++ b/expressions/aggregation/AggregationHandleDistinct.hpp
@@ -107,7 +107,7 @@ class AggregationHandleDistinct : public AggregationConcreteHandle {
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const override;
+ std::vector<std::vector<TypedValue>> *group_by_keys, int index) const override;
void mergeGroupByHashTables(
const AggregationStateHashTableBase &source_hash_table,
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleMax.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMax.cpp b/expressions/aggregation/AggregationHandleMax.cpp
index a7a4a52..f0eebde 100644
--- a/expressions/aggregation/AggregationHandleMax.cpp
+++ b/expressions/aggregation/AggregationHandleMax.cpp
@@ -87,7 +87,7 @@ void AggregationHandleMax::aggregateValueAccessorIntoHashTable(
DCHECK_EQ(1u, argument_ids.size())
<< "Got wrong number of arguments for MAX: " << argument_ids.size();
- aggregateValueAccessorIntoHashTableUnaryHelper<
+/* aggregateValueAccessorIntoHashTableUnaryHelper<
AggregationHandleMax,
AggregationStateMax,
AggregationStateHashTable<AggregationStateMax>>(
@@ -95,7 +95,7 @@ void AggregationHandleMax::aggregateValueAccessorIntoHashTable(
argument_ids.front(),
group_by_key_ids,
AggregationStateMax(type_),
- hash_table);
+ hash_table);*/
}
void AggregationHandleMax::mergeStates(
@@ -109,14 +109,26 @@ void AggregationHandleMax::mergeStates(
}
}
+void AggregationHandleMax::mergeStatesFast(
+ 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);
+ }
+}
+
ColumnVector* AggregationHandleMax::finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const {
- return finalizeHashTableHelper<AggregationHandleMax,
- AggregationStateHashTable<AggregationStateMax>>(
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const {
+ return finalizeHashTableHelperFast<AggregationHandleMax,
+ AggregationStateFastHashTable>(
type_.getNullableVersion(),
hash_table,
- group_by_keys);
+ group_by_keys,
+ index);
}
AggregationState* AggregationHandleMax::aggregateOnDistinctifyHashTableForSingle(
@@ -142,9 +154,8 @@ void AggregationHandleMax::aggregateOnDistinctifyHashTableForGroupBy(
void AggregationHandleMax::mergeGroupByHashTables(
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const {
- mergeGroupByHashTablesHelper<AggregationHandleMax,
- AggregationStateMax,
- AggregationStateHashTable<AggregationStateMax>>(
+ mergeGroupByHashTablesHelperFast<AggregationHandleMax,
+ AggregationStateFastHashTable>(
source_hash_table, destination_hash_table);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleMax.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMax.hpp b/expressions/aggregation/AggregationHandleMax.hpp
index 5af5a12..ce7c702 100644
--- a/expressions/aggregation/AggregationHandleMax.hpp
+++ b/expressions/aggregation/AggregationHandleMax.hpp
@@ -29,6 +29,7 @@
#include "expressions/aggregation/AggregationConcreteHandle.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
#include "threading/SpinMutex.hpp"
#include "types/Type.hpp"
#include "types/TypedValue.hpp"
@@ -105,6 +106,22 @@ class AggregationHandleMax : public AggregationConcreteHandle {
compareAndUpdate(static_cast<AggregationStateMax*>(state), value);
}
+ inline void iterateUnaryInlFast(const TypedValue &value, std::uint8_t *byte_ptr) const {
+ DCHECK(value.isPlausibleInstanceOf(type_.getSignature()));
+ TypedValue *max_ptr = reinterpret_cast<TypedValue *>(byte_ptr);
+ compareAndUpdateFast(max_ptr, value);
+ }
+
+ inline void iterateInlFast(const std::vector<TypedValue> &arguments, uint8_t *byte_ptr) override {
+ iterateUnaryInlFast(arguments.front(), byte_ptr);
+ }
+
+ void initPayload(uint8_t *byte_ptr) override {
+ TypedValue *max_ptr = reinterpret_cast<TypedValue *>(byte_ptr);
+ TypedValue t1 = (type_.getNullableVersion().makeNullValue());
+ *max_ptr = t1;
+ }
+
AggregationState* accumulateColumnVectors(
const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const override;
@@ -123,6 +140,9 @@ class AggregationHandleMax : public AggregationConcreteHandle {
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 AggregationStateMax&>(state).max_);
}
@@ -131,9 +151,15 @@ class AggregationHandleMax : public AggregationConcreteHandle {
return TypedValue(static_cast<const AggregationStateMax&>(state).max_);
}
+ inline TypedValue finalizeHashTableEntryFast(const std::uint8_t *byte_ptr) const {
+ const TypedValue *max_ptr = reinterpret_cast<const TypedValue *>(byte_ptr);
+ return TypedValue(*max_ptr);
+ }
+
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const override;
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const override;
/**
* @brief Implementation of AggregationHandle::aggregateOnDistinctifyHashTableForSingle()
@@ -155,6 +181,10 @@ class AggregationHandleMax : public AggregationConcreteHandle {
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const override;
+ size_t getPayloadSize() const override {
+ return sizeof(TypedValue);
+ }
+
private:
friend class AggregateFunctionMax;
@@ -181,6 +211,13 @@ class AggregationHandleMax : public AggregationConcreteHandle {
}
}
+ inline void compareAndUpdateFast(TypedValue *max_ptr, const TypedValue &value) const {
+ if (value.isNull()) return;
+ if (max_ptr->isNull() || fast_comparator_->compareTypedValues(value, *max_ptr)) {
+ *max_ptr = value;
+ }
+ }
+
const Type &type_;
std::unique_ptr<UncheckedComparator> fast_comparator_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleMin.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMin.cpp b/expressions/aggregation/AggregationHandleMin.cpp
index ca9b163..cbedd9b 100644
--- a/expressions/aggregation/AggregationHandleMin.cpp
+++ b/expressions/aggregation/AggregationHandleMin.cpp
@@ -89,7 +89,7 @@ void AggregationHandleMin::aggregateValueAccessorIntoHashTable(
DCHECK_EQ(1u, argument_ids.size())
<< "Got wrong number of arguments for MIN: " << argument_ids.size();
- aggregateValueAccessorIntoHashTableUnaryHelper<
+/* aggregateValueAccessorIntoHashTableUnaryHelper<
AggregationHandleMin,
AggregationStateMin,
AggregationStateHashTable<AggregationStateMin>>(
@@ -97,7 +97,7 @@ void AggregationHandleMin::aggregateValueAccessorIntoHashTable(
argument_ids.front(),
group_by_key_ids,
AggregationStateMin(type_),
- hash_table);
+ hash_table);*/
}
void AggregationHandleMin::mergeStates(
@@ -111,14 +111,27 @@ void AggregationHandleMin::mergeStates(
}
}
+void AggregationHandleMin::mergeStatesFast(
+ const std::uint8_t *source,
+ std::uint8_t *destination) const {
+ const TypedValue *src_min_ptr = reinterpret_cast<const TypedValue *>(source);
+ TypedValue *dst_min_ptr = reinterpret_cast<TypedValue *>(destination);
+
+ if (!(src_min_ptr->isNull())) {
+ compareAndUpdateFast(dst_min_ptr, *src_min_ptr);
+ }
+}
+
ColumnVector* AggregationHandleMin::finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const {
- return finalizeHashTableHelper<AggregationHandleMin,
- AggregationStateHashTable<AggregationStateMin>>(
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const {
+ return finalizeHashTableHelperFast<AggregationHandleMin,
+ AggregationStateFastHashTable>(
type_.getNonNullableVersion(),
hash_table,
- group_by_keys);
+ group_by_keys,
+ index);
}
AggregationState* AggregationHandleMin::aggregateOnDistinctifyHashTableForSingle(
@@ -144,9 +157,8 @@ void AggregationHandleMin::aggregateOnDistinctifyHashTableForGroupBy(
void AggregationHandleMin::mergeGroupByHashTables(
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const {
- mergeGroupByHashTablesHelper<AggregationHandleMin,
- AggregationStateMin,
- AggregationStateHashTable<AggregationStateMin>>(
+ mergeGroupByHashTablesHelperFast<AggregationHandleMin,
+ AggregationStateFastHashTable>(
source_hash_table, destination_hash_table);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleMin.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMin.hpp b/expressions/aggregation/AggregationHandleMin.hpp
index f68bb9d..7c7869e 100644
--- a/expressions/aggregation/AggregationHandleMin.hpp
+++ b/expressions/aggregation/AggregationHandleMin.hpp
@@ -29,6 +29,7 @@
#include "expressions/aggregation/AggregationConcreteHandle.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
#include "threading/SpinMutex.hpp"
#include "types/Type.hpp"
#include "types/TypedValue.hpp"
@@ -64,6 +65,11 @@ class AggregationStateMin : public AggregationState {
*/
~AggregationStateMin() override {}
+ size_t getPayloadSize() const {
+ return sizeof(TypedValue);
+ }
+
+
private:
friend class AggregationHandleMin;
@@ -104,6 +110,22 @@ class AggregationHandleMin : public AggregationConcreteHandle {
compareAndUpdate(state, value);
}
+ inline void iterateUnaryInlFast(const TypedValue &value, uint8_t *byte_ptr) {
+ DCHECK(value.isPlausibleInstanceOf(type_.getSignature()));
+ TypedValue *min_ptr = reinterpret_cast<TypedValue *>(byte_ptr);
+ compareAndUpdateFast(min_ptr, value);
+ }
+
+ inline void iterateInlFast(const std::vector<TypedValue> &arguments, uint8_t *byte_ptr) override {
+ iterateUnaryInlFast(arguments.front(), byte_ptr);
+ }
+
+ void initPayload(uint8_t *byte_ptr) override {
+ TypedValue *min_ptr = reinterpret_cast<TypedValue *>(byte_ptr);
+ TypedValue t1 = (type_.getNullableVersion().makeNullValue());
+ *min_ptr = t1;
+ }
+
AggregationState* accumulateColumnVectors(
const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const override;
@@ -122,6 +144,9 @@ class AggregationHandleMin : public AggregationConcreteHandle {
void mergeStates(const AggregationState &source,
AggregationState *destination) const override;
+ void mergeStatesFast(const uint8_t *source,
+ uint8_t *destination) const override;
+
TypedValue finalize(const AggregationState &state) const override {
return static_cast<const AggregationStateMin&>(state).min_;
}
@@ -130,9 +155,15 @@ class AggregationHandleMin : public AggregationConcreteHandle {
return static_cast<const AggregationStateMin&>(state).min_;
}
+ inline TypedValue finalizeHashTableEntryFast(const std::uint8_t *byte_ptr) const {
+ const TypedValue *min_ptr = reinterpret_cast<const TypedValue *>(byte_ptr);
+ return TypedValue(*min_ptr);
+ }
+
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const override;
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const override;
/**
* @brief Implementation of AggregationHandle::aggregateOnDistinctifyHashTableForSingle()
@@ -153,6 +184,10 @@ class AggregationHandleMin : public AggregationConcreteHandle {
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const override;
+ size_t getPayloadSize() const override {
+ return sizeof(TypedValue);
+ }
+
private:
friend class AggregateFunctionMin;
@@ -178,6 +213,13 @@ class AggregationHandleMin : public AggregationConcreteHandle {
}
}
+ inline void compareAndUpdateFast(TypedValue *min_ptr, const TypedValue &value) const {
+ if (value.isNull()) return;
+ if (min_ptr->isNull() || fast_comparator_->compareTypedValues(value, *min_ptr)) {
+ *min_ptr = value;
+ }
+ }
+
const Type &type_;
std::unique_ptr<UncheckedComparator> fast_comparator_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleSum.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleSum.cpp b/expressions/aggregation/AggregationHandleSum.cpp
index 691ff39..ad486eb 100644
--- a/expressions/aggregation/AggregationHandleSum.cpp
+++ b/expressions/aggregation/AggregationHandleSum.cpp
@@ -93,7 +93,6 @@ AggregationState* AggregationHandleSum::accumulateColumnVectors(
const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
DCHECK_EQ(1u, column_vectors.size())
<< "Got wrong number of ColumnVectors for SUM: " << column_vectors.size();
-
std::size_t num_tuples = 0;
TypedValue cv_sum = fast_operator_->accumulateColumnVector(
blank_state_.sum_,
@@ -127,7 +126,7 @@ void AggregationHandleSum::aggregateValueAccessorIntoHashTable(
DCHECK_EQ(1u, argument_ids.size())
<< "Got wrong number of arguments for SUM: " << argument_ids.size();
- aggregateValueAccessorIntoHashTableUnaryHelper<
+/* aggregateValueAccessorIntoHashTableUnaryHelper<
AggregationHandleSum,
AggregationStateSum,
AggregationStateHashTable<AggregationStateSum>>(
@@ -135,7 +134,7 @@ void AggregationHandleSum::aggregateValueAccessorIntoHashTable(
argument_ids.front(),
group_by_key_ids,
blank_state_,
- hash_table);
+ hash_table);*/
}
void AggregationHandleSum::mergeStates(
@@ -150,6 +149,17 @@ void AggregationHandleSum::mergeStates(
sum_destination->null_ = sum_destination->null_ && sum_source.null_;
}
+void AggregationHandleSum::mergeStatesFast(
+ const uint8_t *source,
+ uint8_t *destination) const {
+ const TypedValue *src_sum_ptr = reinterpret_cast<const TypedValue *>(source+blank_state_.sum_offset);
+ const bool *src_null_ptr = reinterpret_cast<const bool *>(source+blank_state_.null_offset);
+ TypedValue *dst_sum_ptr = reinterpret_cast<TypedValue *>(destination+blank_state_.sum_offset);
+ bool *dst_null_ptr = reinterpret_cast<bool *>(destination+blank_state_.null_offset);
+ *dst_sum_ptr = merge_operator_->applyToTypedValues(*dst_sum_ptr, *src_sum_ptr);
+ *dst_null_ptr = (*dst_null_ptr) && (*src_null_ptr);
+}
+
TypedValue AggregationHandleSum::finalize(const AggregationState &state) const {
const AggregationStateSum &agg_state = static_cast<const AggregationStateSum&>(state);
if (agg_state.null_) {
@@ -162,12 +172,14 @@ TypedValue AggregationHandleSum::finalize(const AggregationState &state) const {
ColumnVector* AggregationHandleSum::finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const {
- return finalizeHashTableHelper<AggregationHandleSum,
- AggregationStateHashTable<AggregationStateSum>>(
+ std::vector<std::vector<TypedValue>> *group_by_keys,
+ int index) const {
+ return finalizeHashTableHelperFast<AggregationHandleSum,
+ AggregationStateFastHashTable>(
*result_type_,
hash_table,
- group_by_keys);
+ group_by_keys,
+ index);
}
AggregationState* AggregationHandleSum::aggregateOnDistinctifyHashTableForSingle(
@@ -193,9 +205,8 @@ void AggregationHandleSum::aggregateOnDistinctifyHashTableForGroupBy(
void AggregationHandleSum::mergeGroupByHashTables(
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const {
- mergeGroupByHashTablesHelper<AggregationHandleSum,
- AggregationStateSum,
- AggregationStateHashTable<AggregationStateSum>>(
+ mergeGroupByHashTablesHelperFast<AggregationHandleSum,
+ AggregationStateFastHashTable>(
source_hash_table, destination_hash_table);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/AggregationHandleSum.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleSum.hpp b/expressions/aggregation/AggregationHandleSum.hpp
index fdc0884..a8d2b5a 100644
--- a/expressions/aggregation/AggregationHandleSum.hpp
+++ b/expressions/aggregation/AggregationHandleSum.hpp
@@ -29,6 +29,7 @@
#include "expressions/aggregation/AggregationConcreteHandle.hpp"
#include "expressions/aggregation/AggregationHandle.hpp"
#include "storage/HashTableBase.hpp"
+#include "storage/FastHashTable.hpp"
#include "threading/SpinMutex.hpp"
#include "types/Type.hpp"
#include "types/TypedValue.hpp"
@@ -57,27 +58,39 @@ class AggregationStateSum : public AggregationState {
*/
AggregationStateSum(const AggregationStateSum &orig)
: sum_(orig.sum_),
- null_(orig.null_) {
+ null_(orig.null_),
+ sum_offset(orig.sum_offset),
+ null_offset(orig.null_offset) {
}
private:
friend class AggregationHandleSum;
AggregationStateSum()
- : sum_(0), null_(true) {
+ : sum_(0), null_(true), sum_offset(0),
+ null_offset(reinterpret_cast<uint8_t *>(&null_)-reinterpret_cast<uint8_t *>(&sum_)) {
}
AggregationStateSum(TypedValue &&sum, const bool is_null)
: sum_(std::move(sum)), null_(is_null) {
}
+ size_t getPayloadSize() const {
+ size_t p1 = reinterpret_cast<size_t>(&sum_);
+ size_t p2 = reinterpret_cast<size_t>(&mutex_);
+ return (p2-p1);
+ }
+
// TODO(shoban): We might want to specialize sum_ to use atomics for int types
// similar to in AggregationStateCount.
TypedValue sum_;
bool null_;
SpinMutex mutex_;
+
+ int sum_offset, null_offset;
};
+
/**
* @brief An aggregationhandle for sum.
**/
@@ -105,6 +118,26 @@ class AggregationHandleSum : public AggregationConcreteHandle {
state->null_ = false;
}
+ inline void iterateUnaryInlFast(const TypedValue &value, uint8_t *byte_ptr) {
+ DCHECK(value.isPlausibleInstanceOf(argument_type_.getSignature()));
+ if (value.isNull()) return;
+ TypedValue *sum_ptr = reinterpret_cast<TypedValue *>(byte_ptr + blank_state_.sum_offset);
+ bool *null_ptr = reinterpret_cast<bool *>(byte_ptr + blank_state_.null_offset);
+ *sum_ptr = fast_operator_->applyToTypedValues(*sum_ptr, value);
+ *null_ptr = false;
+ }
+
+ inline void iterateInlFast(const std::vector<TypedValue> &arguments, uint8_t *byte_ptr) override {
+ iterateUnaryInlFast(arguments.front(), byte_ptr);
+ }
+
+ void initPayload(uint8_t *byte_ptr) override {
+ TypedValue *sum_ptr = reinterpret_cast<TypedValue *>(byte_ptr + blank_state_.sum_offset);
+ bool *null_ptr = reinterpret_cast<bool *>(byte_ptr + blank_state_.null_offset);
+ *sum_ptr = blank_state_.sum_;
+ *null_ptr = true;
+ }
+
AggregationState* accumulateColumnVectors(
const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const override;
@@ -123,15 +156,24 @@ class AggregationHandleSum : public AggregationConcreteHandle {
void mergeStates(const AggregationState &source,
AggregationState *destination) const override;
+ void mergeStatesFast(const uint8_t *source,
+ uint8_t *destination) const override;
+
TypedValue finalize(const AggregationState &state) const override;
inline TypedValue finalizeHashTableEntry(const AggregationState &state) const {
return static_cast<const AggregationStateSum&>(state).sum_;
}
+ inline TypedValue finalizeHashTableEntryFast(const uint8_t *byte_ptr) const {
+ uint8_t *value_ptr = const_cast<uint8_t*>(byte_ptr);
+ TypedValue *sum_ptr = reinterpret_cast<TypedValue *>(value_ptr + blank_state_.sum_offset);
+ return *sum_ptr;
+ }
+
ColumnVector* finalizeHashTable(
const AggregationStateHashTableBase &hash_table,
- std::vector<std::vector<TypedValue>> *group_by_keys) const override;
+ std::vector<std::vector<TypedValue>> *group_by_keys, int index) const override;
/**
* @brief Implementation of AggregationHandle::aggregateOnDistinctifyHashTableForSingle()
@@ -152,6 +194,10 @@ class AggregationHandleSum : public AggregationConcreteHandle {
const AggregationStateHashTableBase &source_hash_table,
AggregationStateHashTableBase *destination_hash_table) const override;
+ size_t getPayloadSize() const override {
+ return blank_state_.getPayloadSize();
+ }
+
private:
friend class AggregateFunctionSum;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/expressions/aggregation/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/expressions/aggregation/CMakeLists.txt b/expressions/aggregation/CMakeLists.txt
index 5744c52..6f259fa 100644
--- a/expressions/aggregation/CMakeLists.txt
+++ b/expressions/aggregation/CMakeLists.txt
@@ -144,9 +144,11 @@ target_link_libraries(quickstep_expressions_aggregation_AggregationConcreteHandl
glog
quickstep_catalog_CatalogTypedefs
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
quickstep_storage_HashTable
quickstep_storage_HashTableBase
quickstep_storage_HashTableFactory
+ quickstep_threading_SpinMutex
quickstep_types_TypedValue
quickstep_types_containers_ColumnVector
quickstep_utility_Macros)
@@ -161,6 +163,7 @@ target_link_libraries(quickstep_expressions_aggregation_AggregationHandleAvg
quickstep_catalog_CatalogTypedefs
quickstep_expressions_aggregation_AggregationConcreteHandle
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
quickstep_storage_HashTable
quickstep_storage_HashTableBase
quickstep_storage_HashTableFactory
@@ -178,6 +181,7 @@ target_link_libraries(quickstep_expressions_aggregation_AggregationHandleCount
quickstep_catalog_CatalogTypedefs
quickstep_expressions_aggregation_AggregationConcreteHandle
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
quickstep_storage_HashTable
quickstep_storage_HashTableBase
quickstep_storage_HashTableFactory
@@ -202,6 +206,7 @@ target_link_libraries(quickstep_expressions_aggregation_AggregationHandleMax
quickstep_catalog_CatalogTypedefs
quickstep_expressions_aggregation_AggregationConcreteHandle
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
quickstep_storage_HashTable
quickstep_storage_HashTableBase
quickstep_storage_HashTableFactory
@@ -218,6 +223,7 @@ target_link_libraries(quickstep_expressions_aggregation_AggregationHandleMin
quickstep_catalog_CatalogTypedefs
quickstep_expressions_aggregation_AggregationConcreteHandle
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
quickstep_storage_HashTable
quickstep_storage_HashTableBase
quickstep_storage_HashTableFactory
@@ -234,6 +240,7 @@ target_link_libraries(quickstep_expressions_aggregation_AggregationHandleSum
quickstep_catalog_CatalogTypedefs
quickstep_expressions_aggregation_AggregationConcreteHandle
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
quickstep_storage_HashTable
quickstep_storage_HashTableBase
quickstep_storage_HashTableFactory
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/AggregationOperationState.cpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.cpp b/storage/AggregationOperationState.cpp
index 4878cf1..185c6b1 100644
--- a/storage/AggregationOperationState.cpp
+++ b/storage/AggregationOperationState.cpp
@@ -83,6 +83,9 @@ AggregationOperationState::AggregationOperationState(
group_by_types.emplace_back(&group_by_element->getType());
}
+ std::vector<AggregationHandle *> group_by_handles;
+ group_by_handles.clear();
+
if (aggregate_functions.size() == 0) {
// If there is no aggregation function, then it is a distinctify operation
// on the group-by expressions.
@@ -92,11 +95,17 @@ AggregationOperationState::AggregationOperationState(
arguments_.push_back({});
is_distinct_.emplace_back(false);
- group_by_hashtable_pools_.emplace_back(std::unique_ptr<HashTablePool>(
+ /* group_by_hashtable_pools_.emplace_back(std::unique_ptr<HashTablePool>(
new HashTablePool(estimated_num_entries,
hash_table_impl_type,
group_by_types,
handles_.back().get(),
+ storage_manager)));*/
+ group_by_hashtable_pools_.emplace_back(std::unique_ptr<HashTablePool>(
+ new HashTablePool(estimated_num_entries,
+ hash_table_impl_type,
+ group_by_types,
+ handles_.back(),
storage_manager)));
} else {
// Set up each individual aggregate in this operation.
@@ -107,6 +116,7 @@ AggregationOperationState::AggregationOperationState(
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();
+ std::vector<std::size_t> payload_sizes;
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
// AggregationHandle.
@@ -126,12 +136,15 @@ AggregationOperationState::AggregationOperationState(
if (!group_by_list_.empty()) {
// Aggregation with GROUP BY: create a HashTable pool for per-group states.
- group_by_hashtable_pools_.emplace_back(std::unique_ptr<HashTablePool>(
+ /* group_by_hashtable_pools_.emplace_back(std::unique_ptr<HashTablePool>(
new HashTablePool(estimated_num_entries,
hash_table_impl_type,
group_by_types,
handles_.back().get(),
- storage_manager)));
+ storage_manager)));*/
+ group_by_handles.emplace_back(handles_.back());
+ payload_sizes.emplace_back(handles_.back()->getPayloadSize());
+
} else {
// Aggregation without GROUP BY: create a single global state.
single_states_.emplace_back(handles_.back()->createInitialState());
@@ -166,17 +179,40 @@ AggregationOperationState::AggregationOperationState(
// the number of entries in the distinctify hash table. We may estimate
// for each distinct aggregation an estimated_num_distinct_keys value during
// query optimization, if it worths.
- distinctify_hashtables_.emplace_back(
+ /* distinctify_hashtables_.emplace_back(
handles_.back()->createDistinctifyHashTable(
*distinctify_hash_table_impl_types_it,
key_types,
estimated_num_entries,
+ storage_manager));*/
+
+std::vector<AggregationHandle *> local;
+local.emplace_back(handles_.back());
+ distinctify_hashtables_.emplace_back(
+AggregationStateFastHashTableFactory::CreateResizable(
+ *distinctify_hash_table_impl_types_it,
+ key_types,
+ estimated_num_entries,
+ {0},
+ local,
storage_manager));
+
++distinctify_hash_table_impl_types_it;
} else {
distinctify_hashtables_.emplace_back(nullptr);
}
}
+
+ if (!group_by_handles.empty()) {
+ // Aggregation with GROUP BY: create a HashTable pool for per-group states.
+ group_by_hashtable_pools_.emplace_back(std::unique_ptr<HashTablePool>(
+ new HashTablePool(estimated_num_entries,
+ hash_table_impl_type,
+ group_by_types,
+ payload_sizes,
+ group_by_handles,
+ storage_manager)));
+ }
}
}
@@ -410,17 +446,24 @@ void AggregationOperationState::aggregateBlockHashTable(const block_id input_blo
// Call StorageBlock::aggregateGroupBy() to aggregate this block's values
// directly into the (threadsafe) shared global HashTable for this
// aggregate.
- DCHECK(group_by_hashtable_pools_[agg_idx] != nullptr);
- AggregationStateHashTableBase *agg_hash_table = group_by_hashtable_pools_[agg_idx]->getHashTable();
+ DCHECK(group_by_hashtable_pools_[0] != nullptr);
+ AggregationStateHashTableBase *agg_hash_table = group_by_hashtable_pools_[0]->getHashTableFast();
DCHECK(agg_hash_table != nullptr);
- block->aggregateGroupBy(*handles_[agg_idx],
+ /* block->aggregateGroupBy(*handles_[agg_idx],
arguments_[agg_idx],
group_by_list_,
predicate_.get(),
agg_hash_table,
&reuse_matches,
+ &reuse_group_by_vectors);*/
+ block->aggregateGroupByFast(arguments_,
+ group_by_list_,
+ predicate_.get(),
+ agg_hash_table,
+ &reuse_matches,
&reuse_group_by_vectors);
- group_by_hashtable_pools_[agg_idx]->returnHashTable(agg_hash_table);
+ group_by_hashtable_pools_[0]->returnHashTable(agg_hash_table);
+ break;
}
}
}
@@ -444,6 +487,12 @@ void AggregationOperationState::finalizeSingleState(InsertDestination *output_de
output_destination->insertTuple(Tuple(std::move(attribute_values)));
}
+void AggregationOperationState::mergeGroupByHashTables(AggregationStateHashTableBase *src,
+ AggregationStateHashTableBase *dst) {
+ HashTableMergerNewFast merger(dst);
+ (static_cast<FastHashTable<true, false, true, false> *>(src))->forEachCompositeKeyFast(&merger);
+}
+
void AggregationOperationState::finalizeHashTable(InsertDestination *output_destination) {
// 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).
@@ -455,18 +504,21 @@ void AggregationOperationState::finalizeHashTable(InsertDestination *output_dest
// TODO(harshad) - Find heuristics for faster merge, even in a single thread.
// e.g. Keep merging entries from smaller hash tables to larger.
+// auto *hash_tables = group_by_hashtable_pools_[0]->getAllHashTables();
+
+ auto *hash_tables = group_by_hashtable_pools_[0]->getAllHashTables();
for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
- auto *hash_tables = group_by_hashtable_pools_[agg_idx]->getAllHashTables();
if (hash_tables->size() > 1) {
for (int hash_table_index = 0;
hash_table_index < static_cast<int>(hash_tables->size() - 1);
++hash_table_index) {
// Merge each hash table to the last hash table.
- handles_[agg_idx]->mergeGroupByHashTables(
- (*(*hash_tables)[hash_table_index]),
+ mergeGroupByHashTables(
+ (*hash_tables)[hash_table_index].get(),
hash_tables->back().get());
}
}
+ break;
}
// Collect per-aggregate finalized values.
@@ -475,16 +527,16 @@ void AggregationOperationState::finalizeHashTable(InsertDestination *output_dest
agg_idx < handles_.size();
++agg_idx) {
if (is_distinct_[agg_idx]) {
- DCHECK(group_by_hashtable_pools_[agg_idx] != nullptr);
- auto *hash_tables = group_by_hashtable_pools_[agg_idx]->getAllHashTables();
+ DCHECK(group_by_hashtable_pools_[0] != nullptr);
+ auto *hash_tables = group_by_hashtable_pools_[0]->getAllHashTables();
DCHECK(hash_tables != nullptr);
if (hash_tables->empty()) {
// We may have a case where hash_tables is empty, e.g. no input blocks.
// However for aggregateOnDistinctifyHashTableForGroupBy to work
// correctly, we should create an empty group by hash table.
- AggregationStateHashTableBase *new_hash_table = group_by_hashtable_pools_[agg_idx]->getHashTable();
- group_by_hashtable_pools_[agg_idx]->returnHashTable(new_hash_table);
- hash_tables = group_by_hashtable_pools_[agg_idx]->getAllHashTables();
+ AggregationStateHashTableBase *new_hash_table = group_by_hashtable_pools_[0]->getHashTableFast();
+ group_by_hashtable_pools_[0]->returnHashTable(new_hash_table);
+ hash_tables = group_by_hashtable_pools_[0]->getAllHashTables();
}
DCHECK(hash_tables->back() != nullptr);
AggregationStateHashTableBase *agg_hash_table = hash_tables->back().get();
@@ -494,21 +546,22 @@ void AggregationOperationState::finalizeHashTable(InsertDestination *output_dest
agg_hash_table);
}
- auto *hash_tables = group_by_hashtable_pools_[agg_idx]->getAllHashTables();
+ auto *hash_tables = group_by_hashtable_pools_[0]->getAllHashTables();
DCHECK(hash_tables != nullptr);
if (hash_tables->empty()) {
// We may have a case where hash_tables is empty, e.g. no input blocks.
// However for aggregateOnDistinctifyHashTableForGroupBy to work
// correctly, we should create an empty group by hash table.
- AggregationStateHashTableBase *new_hash_table = group_by_hashtable_pools_[agg_idx]->getHashTable();
- group_by_hashtable_pools_[agg_idx]->returnHashTable(new_hash_table);
- hash_tables = group_by_hashtable_pools_[agg_idx]->getAllHashTables();
+ AggregationStateHashTableBase *new_hash_table = group_by_hashtable_pools_[0]->getHashTable();
+ group_by_hashtable_pools_[0]->returnHashTable(new_hash_table);
+ hash_tables = group_by_hashtable_pools_[0]->getAllHashTables();
}
AggregationStateHashTableBase *agg_hash_table = hash_tables->back().get();
DCHECK(agg_hash_table != nullptr);
ColumnVector* agg_result_col =
handles_[agg_idx]->finalizeHashTable(*agg_hash_table,
- &group_by_keys);
+ &group_by_keys,
+ agg_idx);
if (agg_result_col != nullptr) {
final_values.emplace_back(agg_result_col);
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/AggregationOperationState.hpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.hpp b/storage/AggregationOperationState.hpp
index 0199749..8934cda 100644
--- a/storage/AggregationOperationState.hpp
+++ b/storage/AggregationOperationState.hpp
@@ -165,6 +165,8 @@ class AggregationOperationState {
**/
void finalizeAggregate(InsertDestination *output_destination);
+ int dflag;
+
private:
// Merge locally (per storage block) aggregated states with global aggregation
// states.
@@ -185,7 +187,8 @@ class AggregationOperationState {
// Each individual aggregate in this operation has an AggregationHandle and
// some number of Scalar arguments.
- std::vector<std::unique_ptr<AggregationHandle>> handles_;
+// std::vector<std::unique_ptr<AggregationHandle>> handles_;
+ std::vector<AggregationHandle *> handles_;
std::vector<std::vector<std::unique_ptr<const Scalar>>> arguments_;
// For each aggregate, whether DISTINCT should be applied to the aggregate's
@@ -215,6 +218,8 @@ class AggregationOperationState {
StorageManager *storage_manager_;
+ void mergeGroupByHashTables(AggregationStateHashTableBase *src, AggregationStateHashTableBase *dst);
+
DISALLOW_COPY_AND_ASSIGN(AggregationOperationState);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/d5e7d09f/storage/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/storage/CMakeLists.txt b/storage/CMakeLists.txt
index b536411..eec0bfe 100644
--- a/storage/CMakeLists.txt
+++ b/storage/CMakeLists.txt
@@ -193,6 +193,9 @@ if (ENABLE_DISTRIBUTED)
endif()
add_library(quickstep_storage_EvictionPolicy EvictionPolicy.cpp EvictionPolicy.hpp)
+add_library(quickstep_storage_FastHashTable ../empty_src.cpp FastHashTable.hpp)
+add_library(quickstep_storage_FastHashTableFactory ../empty_src.cpp FastHashTableFactory.hpp)
+add_library(quickstep_storage_FastSeparateChainingHashTable ../empty_src.cpp FastSeparateChainingHashTable.hpp)
add_library(quickstep_storage_FileManager ../empty_src.cpp FileManager.hpp)
if (QUICKSTEP_HAVE_FILE_MANAGER_HDFS)
add_library(quickstep_storage_FileManagerHdfs FileManagerHdfs.cpp FileManagerHdfs.hpp)
@@ -616,6 +619,55 @@ target_link_libraries(quickstep_storage_EvictionPolicy
quickstep_threading_SpinMutex
quickstep_threading_SpinSharedMutex
quickstep_utility_Macros)
+target_link_libraries(quickstep_storage_FastHashTable
+ quickstep_catalog_CatalogTypedefs
+ quickstep_expressions_aggregation_AggregationHandleAvg
+ quickstep_storage_HashTable
+ quickstep_storage_HashTableBase
+ quickstep_storage_StorageBlob
+ quickstep_storage_StorageBlockInfo
+ quickstep_storage_StorageConstants
+ quickstep_storage_StorageManager
+ quickstep_storage_TupleReference
+ quickstep_storage_ValueAccessor
+ quickstep_storage_ValueAccessorUtil
+ quickstep_threading_SpinMutex
+ quickstep_threading_SpinSharedMutex
+ quickstep_types_Type
+ quickstep_types_TypedValue
+ quickstep_utility_BloomFilter
+ quickstep_utility_HashPair
+ quickstep_utility_Macros)
+target_link_libraries(quickstep_storage_FastHashTableFactory
+ glog
+ quickstep_storage_FastHashTable
+ quickstep_storage_FastSeparateChainingHashTable
+ quickstep_storage_HashTable
+ quickstep_storage_HashTable_proto
+ quickstep_storage_HashTableBase
+ quickstep_storage_HashTableFactory
+ quickstep_storage_LinearOpenAddressingHashTable
+ quickstep_storage_SeparateChainingHashTable
+ quickstep_storage_SimpleScalarSeparateChainingHashTable
+ quickstep_storage_TupleReference
+ quickstep_types_TypeFactory
+ quickstep_utility_BloomFilter
+ quickstep_utility_Macros)
+target_link_libraries(quickstep_storage_FastSeparateChainingHashTable
+ quickstep_storage_FastHashTable
+ quickstep_storage_HashTable
+ quickstep_storage_HashTableBase
+ quickstep_storage_HashTableKeyManager
+ quickstep_storage_StorageBlob
+ quickstep_storage_StorageBlockInfo
+ quickstep_storage_StorageConstants
+ quickstep_storage_StorageManager
+ quickstep_threading_SpinSharedMutex
+ quickstep_types_Type
+ quickstep_types_TypedValue
+ quickstep_utility_Alignment
+ quickstep_utility_Macros
+ quickstep_utility_PrimeNumber)
target_link_libraries(quickstep_storage_FileManager
quickstep_storage_StorageBlockInfo
quickstep_utility_Macros
@@ -701,6 +753,8 @@ target_link_libraries(quickstep_storage_HashTableKeyManager
target_link_libraries(quickstep_storage_HashTablePool
glog
quickstep_expressions_aggregation_AggregationHandle
+ quickstep_storage_FastHashTable
+ quickstep_storage_FastHashTableFactory
quickstep_storage_HashTableBase
quickstep_threading_SpinMutex
quickstep_utility_Macros
@@ -905,6 +959,7 @@ target_link_libraries(quickstep_storage_StorageBlock
quickstep_storage_CompressedColumnStoreTupleStorageSubBlock
quickstep_storage_CompressedPackedRowStoreTupleStorageSubBlock
quickstep_storage_CountedReference
+ quickstep_storage_FastHashTable
quickstep_storage_HashTableBase
quickstep_storage_IndexSubBlock
quickstep_storage_InsertDestinationInterface
@@ -1060,6 +1115,9 @@ target_link_libraries(quickstep_storage
quickstep_storage_EvictionPolicy
quickstep_storage_FileManager
quickstep_storage_FileManagerLocal
+ quickstep_storage_FastHashTable
+ quickstep_storage_FastHashTableFactory
+ quickstep_storage_FastSeparateChainingHashTable
quickstep_storage_HashTable
quickstep_storage_HashTable_proto
quickstep_storage_HashTableBase