quickstep-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jianq...@apache.org
Subject [4/5] incubator-quickstep git commit: Fuse Aggregate with LeftOuterJoin to accelerate evaluation.
Date Thu, 09 Feb 2017 05:37:43 GMT
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2cf6a2c8/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/2cf6a2c8/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/2cf6a2c8/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/2cf6a2c8/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/2cf6a2c8/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/2cf6a2c8/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/2cf6a2c8/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/2cf6a2c8/third_party/src/iwyu/iwyu_helper.py
----------------------------------------------------------------------
diff --git a/third_party/src/iwyu/iwyu_helper.py b/third_party/src/iwyu/iwyu_helper.py
index dff4d55..93ddbae 100755
--- a/third_party/src/iwyu/iwyu_helper.py
+++ b/third_party/src/iwyu/iwyu_helper.py
@@ -22,15 +22,15 @@ QUICKSTEP_INCLUDES = [ '.',
                        './build/third_party/gflags/include',
                        './build/third_party/protobuf/include',
                        './build/third_party/tmb/include',
-                       './third_party/benchmark/include',
-                       './third_party/glog/src',
-                       './third_party/googletest/googletest/include',
-                       './third_party/protobuf/src',
-                       './third_party/re2',
-                       './third_party/tmb/include']
+                       './third_party/src/benchmark/include',
+                       './third_party/src/glog/src',
+                       './third_party/src/googletest/googletest/include',
+                       './third_party/src/protobuf/src',
+                       './third_party/src/re2',
+                       './third_party/src/tmb/include']
 QUICKSTEP_DEFINES = [ '-DQUICKSTEP_DEBUG',
                       '-DQUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION', ]
-CLANG_FLAGS = [ '-std=c++14', '-x', 'c++', ]
+CLANG_FLAGS = [ '-std=c++14', '-x', 'c++', '-ferror-limit=-1' ]
 
 # Custom configuration filenames.
 CUSTOM_IWYU_MAPPINGS = '.iwyu.imp'

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/2cf6a2c8/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/2cf6a2c8/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/2cf6a2c8/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);
 };


Mime
View raw message