You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by zu...@apache.org on 2017/02/10 14:45:43 UTC
[22/31] incubator-quickstep git commit: Fuse Aggregate with
LeftOuterJoin to accelerate evaluation.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/relational_operators/BuildAggregationExistenceMapOperator.hpp
----------------------------------------------------------------------
diff --git a/relational_operators/BuildAggregationExistenceMapOperator.hpp b/relational_operators/BuildAggregationExistenceMapOperator.hpp
new file mode 100644
index 0000000..e2928a8
--- /dev/null
+++ b/relational_operators/BuildAggregationExistenceMapOperator.hpp
@@ -0,0 +1,177 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ **/
+
+#ifndef QUICKSTEP_RELATIONAL_OPERATORS_BUILD_AGGREGATION_EXISTENCE_MAP_OPERATOR_HPP_
+#define QUICKSTEP_RELATIONAL_OPERATORS_BUILD_AGGREGATION_EXISTENCE_MAP_OPERATOR_HPP_
+
+#include <cstddef>
+
+#include <string>
+#include <vector>
+
+#include "catalog/CatalogRelation.hpp"
+#include "catalog/CatalogTypedefs.hpp"
+#include "query_execution/QueryContext.hpp"
+#include "relational_operators/RelationalOperator.hpp"
+#include "relational_operators/WorkOrder.hpp"
+#include "storage/StorageBlockInfo.hpp"
+#include "utility/Macros.hpp"
+
+#include "glog/logging.h"
+
+#include "tmb/id_typedefs.h"
+
+namespace tmb { class MessageBus; }
+
+namespace quickstep {
+
+class AggregationOperationState;
+class CatalogRelationSchema;
+class StorageManager;
+class WorkOrderProtosContainer;
+class WorkOrdersContainer;
+
+namespace serialization { class WorkOrder; }
+
+/** \addtogroup RelationalOperators
+ * @{
+ */
+
+/**
+ * @brief An operator which builds a bit vector on the input relation's one
+ * attribute where the bit vector serves as the existence map for an
+ * AggregationOperationState's CollisionFreeVectorTable.
+ **/
+class BuildAggregationExistenceMapOperator : public RelationalOperator {
+ public:
+ /**
+ * @brief Constructor.
+ *
+ * @param query_id The ID of the query to which this operator belongs.
+ * @param input_relation The relation to build the existence map on.
+ * @param build_attribute The ID of the attribute to build the existence map on.
+ * @param input_relation_is_stored If input_relation is a stored relation and
+ * is fully available to the operator before it can start generating
+ * workorders.
+ * @param aggr_state_index The index of the AggregationState in QueryContext.
+ **/
+ BuildAggregationExistenceMapOperator(const std::size_t query_id,
+ const CatalogRelation &input_relation,
+ const attribute_id build_attribute,
+ const bool input_relation_is_stored,
+ const QueryContext::aggregation_state_id aggr_state_index)
+ : RelationalOperator(query_id),
+ input_relation_(input_relation),
+ build_attribute_(build_attribute),
+ input_relation_is_stored_(input_relation_is_stored),
+ input_relation_block_ids_(input_relation_is_stored ? input_relation.getBlocksSnapshot()
+ : std::vector<block_id>()),
+ aggr_state_index_(aggr_state_index),
+ num_workorders_generated_(0),
+ started_(false) {}
+
+ ~BuildAggregationExistenceMapOperator() override {}
+
+ std::string getName() const override {
+ return "BuildAggregationExistenceMapOperator";
+ }
+
+ /**
+ * @return The input relation.
+ */
+ const CatalogRelation& input_relation() const {
+ return input_relation_;
+ }
+
+ bool getAllWorkOrders(WorkOrdersContainer *container,
+ QueryContext *query_context,
+ StorageManager *storage_manager,
+ const tmb::client_id scheduler_client_id,
+ tmb::MessageBus *bus) override;
+
+ bool getAllWorkOrderProtos(WorkOrderProtosContainer *container) override;
+
+ void feedInputBlock(const block_id input_block_id, const relation_id input_relation_id,
+ const partition_id part_id) override {
+ input_relation_block_ids_.push_back(input_block_id);
+ }
+
+ private:
+ serialization::WorkOrder* createWorkOrderProto(const block_id block);
+
+ const CatalogRelation &input_relation_;
+ const attribute_id build_attribute_;
+ const bool input_relation_is_stored_;
+ std::vector<block_id> input_relation_block_ids_;
+ const QueryContext::aggregation_state_id aggr_state_index_;
+
+ std::vector<block_id>::size_type num_workorders_generated_;
+ bool started_;
+
+ DISALLOW_COPY_AND_ASSIGN(BuildAggregationExistenceMapOperator);
+};
+
+/**
+ * @brief A WorkOrder produced by BuildAggregationExistenceMapOperator.
+ **/
+class BuildAggregationExistenceMapWorkOrder : public WorkOrder {
+ public:
+ /**
+ * @brief Constructor
+ *
+ * @param query_id The ID of this query.
+ * @param input_relation The relation to build the existence map on.
+ * @param build_block_id The block id.
+ * @param build_attribute The ID of the attribute to build on.
+ * @param state The AggregationState to use.
+ * @param storage_manager The StorageManager to use.
+ **/
+ BuildAggregationExistenceMapWorkOrder(const std::size_t query_id,
+ const CatalogRelationSchema &input_relation,
+ const block_id build_block_id,
+ const attribute_id build_attribute,
+ AggregationOperationState *state,
+ StorageManager *storage_manager)
+ : WorkOrder(query_id),
+ input_relation_(input_relation),
+ build_block_id_(build_block_id),
+ build_attribute_(build_attribute),
+ state_(DCHECK_NOTNULL(state)),
+ storage_manager_(DCHECK_NOTNULL(storage_manager)) {}
+
+ ~BuildAggregationExistenceMapWorkOrder() override {}
+
+ void execute() override;
+
+ private:
+ const CatalogRelationSchema &input_relation_;
+ const block_id build_block_id_;
+ const attribute_id build_attribute_;
+ AggregationOperationState *state_;
+
+ StorageManager *storage_manager_;
+
+ DISALLOW_COPY_AND_ASSIGN(BuildAggregationExistenceMapWorkOrder);
+};
+
+/** @} */
+
+} // namespace quickstep
+
+#endif // QUICKSTEP_RELATIONAL_OPERATORS_BUILD_AGGREGATION_EXISTENCE_MAP_OPERATOR_HPP_
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/relational_operators/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/relational_operators/CMakeLists.txt b/relational_operators/CMakeLists.txt
index df4114d..457d58a 100644
--- a/relational_operators/CMakeLists.txt
+++ b/relational_operators/CMakeLists.txt
@@ -33,6 +33,9 @@ set_gflags_lib_name ()
# Declare micro-libs:
add_library(quickstep_relationaloperators_AggregationOperator AggregationOperator.cpp AggregationOperator.hpp)
+add_library(quickstep_relationaloperators_BuildAggregationExistenceMapOperator
+ BuildAggregationExistenceMapOperator.cpp
+ BuildAggregationExistenceMapOperator.hpp)
add_library(quickstep_relationaloperators_BuildHashOperator BuildHashOperator.cpp BuildHashOperator.hpp)
add_library(quickstep_relationaloperators_BuildLIPFilterOperator BuildLIPFilterOperator.cpp BuildLIPFilterOperator.hpp)
add_library(quickstep_relationaloperators_CreateIndexOperator CreateIndexOperator.cpp CreateIndexOperator.hpp)
@@ -95,6 +98,31 @@ target_link_libraries(quickstep_relationaloperators_AggregationOperator
quickstep_utility_lipfilter_LIPFilterAdaptiveProber
quickstep_utility_lipfilter_LIPFilterUtil
tmb)
+target_link_libraries(quickstep_relationaloperators_BuildAggregationExistenceMapOperator
+ glog
+ quickstep_catalog_CatalogAttribute
+ quickstep_catalog_CatalogRelation
+ quickstep_catalog_CatalogRelationSchema
+ quickstep_catalog_CatalogTypedefs
+ quickstep_queryexecution_QueryContext
+ quickstep_queryexecution_WorkOrderProtosContainer
+ quickstep_queryexecution_WorkOrdersContainer
+ quickstep_relationaloperators_RelationalOperator
+ quickstep_relationaloperators_WorkOrder
+ quickstep_relationaloperators_WorkOrder_proto
+ quickstep_storage_AggregationOperationState
+ quickstep_storage_CollisionFreeVectorTable
+ quickstep_storage_StorageBlock
+ quickstep_storage_StorageBlockInfo
+ quickstep_storage_StorageManager
+ quickstep_storage_TupleStorageSubBlock
+ quickstep_storage_ValueAccessor
+ quickstep_storage_ValueAccessorUtil
+ quickstep_types_Type
+ quickstep_types_TypeID
+ quickstep_utility_BarrieredReadWriteConcurrentBitVector
+ quickstep_utility_Macros
+ tmb)
target_link_libraries(quickstep_relationaloperators_BuildHashOperator
glog
quickstep_catalog_CatalogRelation
@@ -518,6 +546,7 @@ target_link_libraries(quickstep_relationaloperators_WorkOrderFactory
quickstep_catalog_CatalogTypedefs
quickstep_queryexecution_QueryContext
quickstep_relationaloperators_AggregationOperator
+ quickstep_relationaloperators_BuildAggregationExistenceMapOperator
quickstep_relationaloperators_BuildHashOperator
quickstep_relationaloperators_BuildLIPFilterOperator
quickstep_relationaloperators_DeleteOperator
@@ -552,6 +581,7 @@ target_link_libraries(quickstep_relationaloperators_WorkOrder_proto
add_library(quickstep_relationaloperators ../empty_src.cpp RelationalOperatorsModule.hpp)
target_link_libraries(quickstep_relationaloperators
quickstep_relationaloperators_AggregationOperator
+ quickstep_relationaloperators_BuildAggregationExistenceMapOperator
quickstep_relationaloperators_BuildLIPFilterOperator
quickstep_relationaloperators_BuildHashOperator
quickstep_relationaloperators_CreateIndexOperator
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/relational_operators/WorkOrder.proto
----------------------------------------------------------------------
diff --git a/relational_operators/WorkOrder.proto b/relational_operators/WorkOrder.proto
index 76753d2..d0d0753 100644
--- a/relational_operators/WorkOrder.proto
+++ b/relational_operators/WorkOrder.proto
@@ -44,6 +44,7 @@ enum WorkOrderType {
UPDATE = 20;
WINDOW_AGGREGATION = 21;
DESTROY_AGGREGATION_STATE = 22;
+ BUILD_AGGREGATION_EXISTENCE_MAP = 23;
}
message WorkOrder {
@@ -278,6 +279,15 @@ message WindowAggregationWorkOrder {
message DestroyAggregationStateWorkOrder {
extend WorkOrder {
- optional uint32 aggr_state_index = 339;
+ optional uint32 aggr_state_index = 352;
+ }
+}
+
+message BuildAggregationExistenceMapWorkOrder {
+ extend WorkOrder {
+ optional int32 relation_id = 368;
+ optional fixed64 build_block_id = 369;
+ optional int32 build_attribute = 370;
+ optional uint32 aggr_state_index = 371;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/relational_operators/WorkOrderFactory.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/WorkOrderFactory.cpp b/relational_operators/WorkOrderFactory.cpp
index bd2a0f8..d2c8251 100644
--- a/relational_operators/WorkOrderFactory.cpp
+++ b/relational_operators/WorkOrderFactory.cpp
@@ -29,6 +29,7 @@
#include "catalog/CatalogTypedefs.hpp"
#include "query_execution/QueryContext.hpp"
#include "relational_operators/AggregationOperator.hpp"
+#include "relational_operators/BuildAggregationExistenceMapOperator.hpp"
#include "relational_operators/BuildHashOperator.hpp"
#include "relational_operators/BuildLIPFilterOperator.hpp"
#include "relational_operators/DeleteOperator.hpp"
@@ -91,6 +92,19 @@ WorkOrder* WorkOrderFactory::ReconstructFromProto(const serialization::WorkOrder
CreateLIPFilterAdaptiveProberHelper(
proto.GetExtension(serialization::AggregationWorkOrder::lip_deployment_index), query_context));
}
+ case serialization::BUILD_AGGREGATION_EXISTENCE_MAP: {
+ LOG(INFO) << "Creating BuildAggregationExistenceMapWorkOrder in Shiftboss " << shiftboss_index;
+
+ return new BuildAggregationExistenceMapWorkOrder(
+ proto.query_id(),
+ catalog_database->getRelationSchemaById(
+ proto.GetExtension(serialization::BuildAggregationExistenceMapWorkOrder::relation_id)),
+ proto.GetExtension(serialization::BuildAggregationExistenceMapWorkOrder::build_block_id),
+ proto.GetExtension(serialization::BuildAggregationExistenceMapWorkOrder::build_attribute),
+ query_context->getAggregationState(
+ proto.GetExtension(serialization::AggregationWorkOrder::aggr_state_index)),
+ storage_manager);
+ }
case serialization::BUILD_LIP_FILTER: {
LOG(INFO) << "Creating BuildLIPFilterWorkOrder in Shiftboss " << shiftboss_index;
@@ -525,6 +539,29 @@ bool WorkOrderFactory::ProtoIsValid(const serialization::WorkOrder &proto,
query_context.isValidAggregationStateId(
proto.GetExtension(serialization::AggregationWorkOrder::aggr_state_index));
}
+ case serialization::BUILD_AGGREGATION_EXISTENCE_MAP: {
+ if (!proto.HasExtension(serialization::BuildAggregationExistenceMapWorkOrder::relation_id)) {
+ return false;
+ }
+
+ const relation_id rel_id =
+ proto.GetExtension(serialization::BuildAggregationExistenceMapWorkOrder::relation_id);
+ if (!catalog_database.hasRelationWithId(rel_id)) {
+ return false;
+ }
+
+ const CatalogRelationSchema &relation = catalog_database.getRelationSchemaById(rel_id);
+ const attribute_id build_attribute =
+ proto.GetExtension(serialization::BuildAggregationExistenceMapWorkOrder::build_attribute);
+ if (!relation.hasAttributeWithId(build_attribute)) {
+ return false;
+ }
+
+ return proto.HasExtension(serialization::BuildAggregationExistenceMapWorkOrder::build_block_id) &&
+ proto.HasExtension(serialization::BuildAggregationExistenceMapWorkOrder::aggr_state_index) &&
+ query_context.isValidAggregationStateId(
+ proto.GetExtension(serialization::BuildAggregationExistenceMapWorkOrder::aggr_state_index));
+ }
case serialization::BUILD_HASH: {
if (!proto.HasExtension(serialization::BuildHashWorkOrder::relation_id)) {
return false;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/storage/AggregationOperationState.cpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.cpp b/storage/AggregationOperationState.cpp
index 0b34908..0f39b41 100644
--- a/storage/AggregationOperationState.cpp
+++ b/storage/AggregationOperationState.cpp
@@ -412,12 +412,18 @@ std::size_t AggregationOperationState::getNumFinalizationPartitions() const {
}
}
+CollisionFreeVectorTable* AggregationOperationState
+ ::getCollisionFreeVectorTable() const {
+ return static_cast<CollisionFreeVectorTable *>(
+ collision_free_hashtable_.get());
+}
+
void AggregationOperationState::initialize(const std::size_t partition_id) {
if (is_aggregate_collision_free_) {
static_cast<CollisionFreeVectorTable *>(
collision_free_hashtable_.get())->initialize(partition_id);
} else {
- LOG(FATAL) << "AggregationOperationState::initializeState() "
+ LOG(FATAL) << "AggregationOperationState::initialize() "
<< "is not supported by this aggregation";
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/storage/AggregationOperationState.hpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.hpp b/storage/AggregationOperationState.hpp
index 13ee377..c8930ee 100644
--- a/storage/AggregationOperationState.hpp
+++ b/storage/AggregationOperationState.hpp
@@ -41,6 +41,7 @@ namespace serialization { class AggregationOperationState; }
class AggregateFunction;
class CatalogDatabaseLite;
class CatalogRelationSchema;
+class CollisionFreeVectorTable;
class InsertDestination;
class LIPFilterAdaptiveProber;
class StorageManager;
@@ -198,6 +199,14 @@ class AggregationOperationState {
void finalizeAggregate(const std::size_t partition_id,
InsertDestination *output_destination);
+ /**
+ * @brief Get the collision-free vector table used by this aggregation.
+ *
+ * @return The collision-free vector table used by this aggregation.
+ * Returns NULL if collision-free vector table is not used.
+ */
+ CollisionFreeVectorTable* getCollisionFreeVectorTable() const;
+
private:
// Check whether partitioned aggregation can be applied.
bool checkAggregatePartitioned(
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/storage/CollisionFreeVectorTable.hpp
----------------------------------------------------------------------
diff --git a/storage/CollisionFreeVectorTable.hpp b/storage/CollisionFreeVectorTable.hpp
index 4f3e238..772d47d 100644
--- a/storage/CollisionFreeVectorTable.hpp
+++ b/storage/CollisionFreeVectorTable.hpp
@@ -105,6 +105,15 @@ class CollisionFreeVectorTable : public AggregationStateHashTableBase {
}
/**
+ * @brief Get the existence map for this vector table.
+ *
+ * @return The existence map for this vector table.
+ */
+ inline BarrieredReadWriteConcurrentBitVector* getExistenceMap() const {
+ return existence_map_.get();
+ }
+
+ /**
* @brief Initialize the specified partition of this aggregation table.
*
* @param partition_id ID of the partition to be initialized.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/utility/lip_filter/BitVectorExactFilter.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/BitVectorExactFilter.hpp b/utility/lip_filter/BitVectorExactFilter.hpp
index 6ad0567..48fd5e1 100644
--- a/utility/lip_filter/BitVectorExactFilter.hpp
+++ b/utility/lip_filter/BitVectorExactFilter.hpp
@@ -20,17 +20,16 @@
#ifndef QUICKSTEP_UTILITY_LIP_FILTER_BIT_VECTOR_EXACT_FILTER_HPP_
#define QUICKSTEP_UTILITY_LIP_FILTER_BIT_VECTOR_EXACT_FILTER_HPP_
-#include <atomic>
+#include <cstddef>
#include <cstdint>
-#include <cstring>
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
#include "storage/StorageBlockInfo.hpp"
-#include "storage/StorageConstants.hpp"
#include "storage/ValueAccessor.hpp"
#include "storage/ValueAccessorUtil.hpp"
#include "types/Type.hpp"
+#include "utility/BarrieredReadWriteConcurrentBitVector.hpp"
#include "utility/Macros.hpp"
#include "utility/lip_filter/LIPFilter.hpp"
@@ -64,14 +63,10 @@ class BitVectorExactFilter : public LIPFilter {
: LIPFilter(LIPFilterType::kBitVectorExactFilter),
min_value_(static_cast<CppType>(min_value)),
max_value_(static_cast<CppType>(max_value)),
- bit_array_(GetByteSize(max_value - min_value + 1)) {
+ bit_vector_(max_value - min_value + 1) {
DCHECK_EQ(min_value, static_cast<std::int64_t>(min_value_));
DCHECK_EQ(max_value, static_cast<std::int64_t>(max_value_));
DCHECK_GE(max_value_, min_value_);
-
- std::memset(bit_array_.data(),
- 0x0,
- sizeof(std::atomic<std::uint8_t>) * GetByteSize(max_value - min_value + 1));
}
void insertValueAccessor(ValueAccessor *accessor,
@@ -109,13 +104,6 @@ class BitVectorExactFilter : public LIPFilter {
private:
/**
- * @brief Round up bit_size to multiples of 8.
- */
- inline static std::size_t GetByteSize(const std::size_t bit_size) {
- return (bit_size + 7u) / 8u;
- }
-
- /**
* @brief Iterate through the accessor and hash values into the internal bit
* array.
*/
@@ -164,8 +152,7 @@ class BitVectorExactFilter : public LIPFilter {
DCHECK_GE(value, min_value_);
DCHECK_LE(value, max_value_);
- const CppType loc = value - min_value_;
- bit_array_[loc >> 3u].fetch_or(1u << (loc & 7u), std::memory_order_relaxed);
+ bit_vector_.setBit(value - min_value_);
}
/**
@@ -177,9 +164,7 @@ class BitVectorExactFilter : public LIPFilter {
return is_anti_filter;
}
- const CppType loc = value - min_value_;
- const bool is_bit_set =
- (bit_array_[loc >> 3u].load(std::memory_order_relaxed) & (1u << (loc & 7u))) != 0;
+ const bool is_bit_set = bit_vector_.getBit(value - min_value_);
if (is_anti_filter) {
return !is_bit_set;
@@ -190,7 +175,7 @@ class BitVectorExactFilter : public LIPFilter {
const CppType min_value_;
const CppType max_value_;
- alignas(kCacheLineBytes) std::vector<std::atomic<std::uint8_t>> bit_array_;
+ BarrieredReadWriteConcurrentBitVector bit_vector_;
DISALLOW_COPY_AND_ASSIGN(BitVectorExactFilter);
};
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/utility/lip_filter/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/utility/lip_filter/CMakeLists.txt b/utility/lip_filter/CMakeLists.txt
index edd0d24..519d3e9 100644
--- a/utility/lip_filter/CMakeLists.txt
+++ b/utility/lip_filter/CMakeLists.txt
@@ -35,12 +35,12 @@ add_library(quickstep_utility_lipfilter_SingleIdentityHashFilter ../../empty_src
target_link_libraries(quickstep_utility_lipfilter_BitVectorExactFilter
quickstep_catalog_CatalogTypedefs
quickstep_storage_StorageBlockInfo
- quickstep_storage_StorageConstants
quickstep_storage_ValueAccessor
quickstep_storage_ValueAccessorUtil
quickstep_types_Type
- quickstep_utility_lipfilter_LIPFilter
- quickstep_utility_Macros)
+ quickstep_utility_BarrieredReadWriteConcurrentBitVector
+ quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilter)
target_link_libraries(quickstep_utility_lipfilter_LIPFilter
quickstep_catalog_CatalogTypedefs
quickstep_storage_StorageBlockInfo
@@ -79,9 +79,9 @@ target_link_libraries(quickstep_utility_lipfilter_LIPFilter_proto
target_link_libraries(quickstep_utility_lipfilter_SingleIdentityHashFilter
quickstep_catalog_CatalogTypedefs
quickstep_storage_StorageBlockInfo
- quickstep_storage_StorageConstants
quickstep_storage_ValueAccessor
quickstep_storage_ValueAccessorUtil
quickstep_types_Type
- quickstep_utility_lipfilter_LIPFilter
- quickstep_utility_Macros)
+ quickstep_utility_BarrieredReadWriteConcurrentBitVector
+ quickstep_utility_Macros
+ quickstep_utility_lipfilter_LIPFilter)
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a28b1e4d/utility/lip_filter/SingleIdentityHashFilter.hpp
----------------------------------------------------------------------
diff --git a/utility/lip_filter/SingleIdentityHashFilter.hpp b/utility/lip_filter/SingleIdentityHashFilter.hpp
index 5c0e8a2..d7e3475 100644
--- a/utility/lip_filter/SingleIdentityHashFilter.hpp
+++ b/utility/lip_filter/SingleIdentityHashFilter.hpp
@@ -20,18 +20,15 @@
#ifndef QUICKSTEP_UTILITY_LIP_FILTER_SINGLE_IDENTITY_HASH_FILTER_HPP_
#define QUICKSTEP_UTILITY_LIP_FILTER_SINGLE_IDENTITY_HASH_FILTER_HPP_
-#include <atomic>
#include <cstddef>
-#include <cstdint>
-#include <cstring>
#include <vector>
#include "catalog/CatalogTypedefs.hpp"
#include "storage/StorageBlockInfo.hpp"
-#include "storage/StorageConstants.hpp"
#include "storage/ValueAccessor.hpp"
#include "storage/ValueAccessorUtil.hpp"
#include "types/Type.hpp"
+#include "utility/BarrieredReadWriteConcurrentBitVector.hpp"
#include "utility/Macros.hpp"
#include "utility/lip_filter/LIPFilter.hpp"
@@ -65,11 +62,8 @@ class SingleIdentityHashFilter : public LIPFilter {
explicit SingleIdentityHashFilter(const std::size_t filter_cardinality)
: LIPFilter(LIPFilterType::kSingleIdentityHashFilter),
filter_cardinality_(filter_cardinality),
- bit_array_(GetByteSize(filter_cardinality)) {
+ bit_vector_(filter_cardinality) {
DCHECK_GE(filter_cardinality, 1u);
- std::memset(bit_array_.data(),
- 0x0,
- sizeof(std::atomic<std::uint8_t>) * GetByteSize(filter_cardinality));
}
void insertValueAccessor(ValueAccessor *accessor,
@@ -158,8 +152,9 @@ class SingleIdentityHashFilter : public LIPFilter {
* @brief Inserts a given value into the hash filter.
*/
inline void insert(const void *key_begin) {
- const CppType hash = *reinterpret_cast<const CppType *>(key_begin) % filter_cardinality_;
- bit_array_[hash >> 3u].fetch_or(1u << (hash & 7u), std::memory_order_relaxed);
+ const CppType hash =
+ *reinterpret_cast<const CppType *>(key_begin) % filter_cardinality_;
+ bit_vector_.setBit(hash);
}
/**
@@ -168,12 +163,13 @@ class SingleIdentityHashFilter : public LIPFilter {
* If false is returned, a value is certainly not present in the hash filter.
*/
inline bool contains(const void *key_begin) const {
- const CppType hash = *reinterpret_cast<const CppType *>(key_begin) % filter_cardinality_;
- return (bit_array_[hash >> 3u].load(std::memory_order_relaxed) & (1u << (hash & 7u)));
+ const CppType hash =
+ *reinterpret_cast<const CppType *>(key_begin) % filter_cardinality_;
+ return bit_vector_.getBit(hash);
}
std::size_t filter_cardinality_;
- alignas(kCacheLineBytes) std::vector<std::atomic<std::uint8_t>> bit_array_;
+ BarrieredReadWriteConcurrentBitVector bit_vector_;
DISALLOW_COPY_AND_ASSIGN(SingleIdentityHashFilter);
};