change writeset type to tbb::concurrent_unordered_set

Author: Yingjun Wu

src/codegen/updater.cpp

@@ -38,7 +38,7 @@ void Updater::Init(storage::DataTable *table,
   target_list_ =
       new TargetList(target_vector, target_vector + target_vector_size);
 
-  statement_write_set_ = new ReadWriteSet();
+  statement_write_set_ = new WriteSet();
 }
 
 char *Updater::GetDataPtr(uint32_t tile_group_id, uint32_t tuple_offset) {
@@ -78,7 +78,8 @@ char *Updater::Prepare(uint32_t tile_group_id, uint32_t tuple_offset) {
 char *Updater::PreparePK(uint32_t tile_group_id, uint32_t tuple_offset) {
 
   PELOTON_ASSERT(table_ != nullptr && executor_context_ != nullptr);
-  if (statement_write_set_->Contains(ItemPointer(tile_group_id, tuple_offset))) {
+
+  if (statement_write_set_->find(ItemPointer(tile_group_id, tuple_offset)) != statement_write_set_->end()) {
     return nullptr;
   }
 
@@ -173,7 +174,7 @@ void Updater::UpdatePK() {
     return;
   }
   txn_manager.PerformInsert(txn, new_location_, index_entry_ptr);
-  statement_write_set_->Insert(new_location_, RWType::INSERT);
+  statement_write_set_->insert(new_location_);
   executor_context_->num_processed++;
 }
 

src/executor/update_executor.cpp

@@ -53,7 +53,7 @@ bool UpdateExecutor::DInit() {
   PELOTON_ASSERT(target_table_);
   PELOTON_ASSERT(project_info_);
 
-  statement_write_set_.Clear();
+  statement_write_set_.clear();
 
   return true;
 }
@@ -63,7 +63,7 @@ bool UpdateExecutor::PerformUpdatePrimaryKey(
     storage::TileGroupHeader *tile_group_header, oid_t physical_tuple_id,
     ItemPointer &old_location) {
 
-  if (statement_write_set_.Contains(old_location)) {
+  if (statement_write_set_.find(old_location) != statement_write_set_.end()) {
     return true;
   }
 
@@ -143,8 +143,7 @@ bool UpdateExecutor::PerformUpdatePrimaryKey(
   }
 
   transaction_manager.PerformInsert(current_txn, location, index_entry_ptr);
-  statement_write_set_.Insert(location, RWType::INSERT);
-
+  statement_write_set_.insert(location);
   return true;
 }
 

src/include/codegen/updater.h

@@ -82,7 +82,7 @@ class Updater {
   // the same operation.
   // By maintaining the statement-level write set, an update operation will check 
   // whether the to-be-updated tuple is created by the same operation.
-  ReadWriteSet *statement_write_set_;
+  WriteSet *statement_write_set_;
 
   // Ownership information
   bool is_owner_;

src/include/common/internal_types.h

@@ -24,6 +24,7 @@
 #include <unistd.h>
 
 #include "tbb/concurrent_vector.h"
+#include "tbb/concurrent_unordered_set.h"
 
 #include "parser/pg_trigger.h"
 #include "type/type_id.h"
@@ -1209,6 +1210,8 @@ std::ostream &operator<<(std::ostream &os, const RWType &type);
 typedef CuckooMap<ItemPointer, RWType, ItemPointerHasher, ItemPointerComparator>
     ReadWriteSet;
 
+typedef tbb::concurrent_unordered_multiset<ItemPointer, ItemPointerHasher, ItemPointerComparator> WriteSet;
+
 // this enum is to identify why the version should be GC'd.
 enum class GCVersionType {
   INVALID,

src/include/common/item_pointer.h

@@ -44,6 +44,10 @@ class ItemPointer {
     }
   }
 
+  bool operator==(const ItemPointer &rhs) const {
+    return (block == rhs.block && offset == rhs.offset);
+  }
+
 } __attribute__((__aligned__(8))) __attribute__((__packed__));
 
 extern ItemPointer INVALID_ITEMPOINTER;

src/include/executor/update_executor.h

@@ -60,7 +60,7 @@ class UpdateExecutor : public AbstractExecutor {
   // the same operation.
   // By maintaining the statement-level write set, an update operation will check 
   // whether the to-be-updated tuple is created by the same operation.
-  ReadWriteSet statement_write_set_;
+  WriteSet statement_write_set_;
 };
 
 }  // namespace executor