fix bug in seq scan

Author: Yingjun Wu

src/codegen/updater.cpp

@@ -37,6 +37,8 @@ void Updater::Init(storage::DataTable *table,
   // Target list is kept since it is required at a new version update
   target_list_ =
       new TargetList(target_vector, target_vector + target_vector_size);
+
+  statement_write_set_ = new ReadWriteSet();
 }
 
 char *Updater::GetDataPtr(uint32_t tile_group_id, uint32_t tuple_offset) {
@@ -74,7 +76,12 @@ 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))) {
+    return nullptr;
+  }
+
   auto *txn = executor_context_->GetTransaction();
   auto tile_group = table_->GetTileGroupById(tile_group_id).get();
   auto *tile_group_header = tile_group->GetHeader();
@@ -166,13 +173,15 @@ void Updater::UpdatePK() {
     return;
   }
   txn_manager.PerformInsert(txn, new_location_, index_entry_ptr);
+  statement_write_set_->Insert(new_location_, RWType::INSERT);
   executor_context_->num_processed++;
 }
 
 void Updater::TearDown() {
   // Updater object does not destruct its own data structures
   tile_.reset();
   delete target_list_;
+  delete statement_write_set_;
 }
 
 }  // namespace codegen

src/executor/update_executor.cpp

@@ -53,13 +53,20 @@ bool UpdateExecutor::DInit() {
   PELOTON_ASSERT(target_table_);
   PELOTON_ASSERT(project_info_);
 
+  statement_write_set_.Clear();
+
   return true;
 }
 
 bool UpdateExecutor::PerformUpdatePrimaryKey(
     bool is_owner, storage::TileGroup *tile_group,
     storage::TileGroupHeader *tile_group_header, oid_t physical_tuple_id,
     ItemPointer &old_location) {
+  
+  if (statement_write_set_.Contains(old_location)) {
+    return true;
+  }
+
   auto &transaction_manager =
       concurrency::TransactionManagerFactory::GetInstance();
 
@@ -136,6 +143,7 @@ bool UpdateExecutor::PerformUpdatePrimaryKey(
   }
 
   transaction_manager.PerformInsert(current_txn, location, index_entry_ptr);
+  statement_write_set_.Insert(location, RWType::INSERT);
 
   return true;
 }

src/include/codegen/updater.h

@@ -75,6 +75,9 @@ class Updater {
   // Target list and direct map list pointer from the update translator
   TargetList *target_list_;
 
+  // Write set for tracking newly created tuples inserted by the same statement
+  ReadWriteSet *statement_write_set_;
+
   // Ownership information
   bool is_owner_;
   bool acquired_ownership_;

src/include/executor/update_executor.h

@@ -52,6 +52,9 @@ class UpdateExecutor : public AbstractExecutor {
  private:
   storage::DataTable *target_table_ = nullptr;
   const planner::ProjectInfo *project_info_ = nullptr;
+
+  // Write set for tracking newly created tuples inserted by the same statement
+  ReadWriteSet statement_write_set_;
 };
 
 }  // namespace executor