fix #1313 Remove read only isolation level (#1322)

* remove 'read_only' from isolation level and add a new 'read only' flag in transaction context

* set flag when starting a read only txn

* add missing isolation level parameter

* fix serializalbe txn test - readonly txn failure. 'read only' is actually of snapshot isolation level in the previous implementation

* fix spacing, add comment and remove wrapper function definition

* add a test case of two concurrent txns with one read-only one

* fix duplicate assertion

* add comments to test case

* change parameter name

* update read only snapshot test case to a more reasonable one and add some comments

* wait one epoch before updating snapshot epoch. make test case more reliable.

Author: Nov11

src/common/internal_types.cpp

@@ -2403,9 +2403,6 @@ std::string IsolationLevelTypeToString(IsolationLevelType type) {
     case IsolationLevelType::READ_COMMITTED: {
       return "READ_COMMITTED";
     }
-    case IsolationLevelType::READ_ONLY: {
-      return "READ_ONLY";
-    }
     default: {
       throw ConversionException(StringUtil::Format(
           "No string conversion for IsolationLevelType value '%d'",
@@ -2427,8 +2424,6 @@ IsolationLevelType StringToIsolationLevelType(const std::string &str) {
     return IsolationLevelType::REPEATABLE_READS;
   } else if (upper_str == "READ_COMMITTED") {
     return IsolationLevelType::READ_COMMITTED;
-  } else if (upper_str == "READ_ONLY") {
-    return IsolationLevelType::READ_ONLY;
   } else {
     throw ConversionException(
         StringUtil::Format("No IsolationLevelType conversion from string '%s'",

src/concurrency/timestamp_ordering_transaction_manager.cpp

@@ -178,7 +178,7 @@ bool TimestampOrderingTransactionManager::PerformRead(
   //////////////////////////////////////////////////////////
   //// handle READ_ONLY
   //////////////////////////////////////////////////////////
-  if (current_txn->GetIsolationLevel() == IsolationLevelType::READ_ONLY) {
+  if (current_txn->IsReadOnly()) {
     // do not update read set for read-only transactions.
     return true;
   }  // end READ ONLY
@@ -437,8 +437,7 @@ bool TimestampOrderingTransactionManager::PerformRead(
 void TimestampOrderingTransactionManager::PerformInsert(
     TransactionContext *const current_txn, const ItemPointer &location,
     ItemPointer *index_entry_ptr) {
-  PELOTON_ASSERT(current_txn->GetIsolationLevel() !=
-                 IsolationLevelType::READ_ONLY);
+  PELOTON_ASSERT(!current_txn->IsReadOnly());
 
   oid_t tile_group_id = location.block;
   oid_t tuple_id = location.offset;
@@ -477,8 +476,7 @@ void TimestampOrderingTransactionManager::PerformInsert(
 void TimestampOrderingTransactionManager::PerformUpdate(
     TransactionContext *const current_txn, const ItemPointer &location,
     const ItemPointer &new_location) {
-  PELOTON_ASSERT(current_txn->GetIsolationLevel() !=
-                 IsolationLevelType::READ_ONLY);
+  PELOTON_ASSERT(!current_txn->IsReadOnly());
 
   ItemPointer old_location = location;
 
@@ -562,8 +560,7 @@ void TimestampOrderingTransactionManager::PerformUpdate(
 void TimestampOrderingTransactionManager::PerformUpdate(
     TransactionContext *const current_txn UNUSED_ATTRIBUTE,
     const ItemPointer &location) {
-  PELOTON_ASSERT(current_txn->GetIsolationLevel() !=
-                 IsolationLevelType::READ_ONLY);
+  PELOTON_ASSERT(!current_txn->IsReadOnly());
 
   oid_t tile_group_id = location.block;
   UNUSED_ATTRIBUTE oid_t tuple_id = location.offset;
@@ -596,8 +593,7 @@ void TimestampOrderingTransactionManager::PerformUpdate(
 void TimestampOrderingTransactionManager::PerformDelete(
     TransactionContext *const current_txn, const ItemPointer &location,
     const ItemPointer &new_location) {
-  PELOTON_ASSERT(current_txn->GetIsolationLevel() !=
-                 IsolationLevelType::READ_ONLY);
+  PELOTON_ASSERT(!current_txn->IsReadOnly());
 
   ItemPointer old_location = location;
 
@@ -684,8 +680,7 @@ void TimestampOrderingTransactionManager::PerformDelete(
 
 void TimestampOrderingTransactionManager::PerformDelete(
     TransactionContext *const current_txn, const ItemPointer &location) {
-  PELOTON_ASSERT(current_txn->GetIsolationLevel() !=
-                 IsolationLevelType::READ_ONLY);
+  PELOTON_ASSERT(!current_txn->IsReadOnly());
 
   oid_t tile_group_id = location.block;
   oid_t tuple_id = location.offset;
@@ -725,7 +720,7 @@ ResultType TimestampOrderingTransactionManager::CommitTransaction(
   //////////////////////////////////////////////////////////
   //// handle READ_ONLY
   //////////////////////////////////////////////////////////
-  if (current_txn->GetIsolationLevel() == IsolationLevelType::READ_ONLY) {
+  if (current_txn->IsReadOnly()) {
     EndTransaction(current_txn);
     return ResultType::SUCCESS;
   }
@@ -914,8 +909,7 @@ ResultType TimestampOrderingTransactionManager::CommitTransaction(
 ResultType TimestampOrderingTransactionManager::AbortTransaction(
     TransactionContext *const current_txn) {
   // a pre-declared read-only transaction will never abort.
-  PELOTON_ASSERT(current_txn->GetIsolationLevel() !=
-                 IsolationLevelType::READ_ONLY);
+  PELOTON_ASSERT(!current_txn->IsReadOnly());
 
   LOG_TRACE("Aborting peloton txn : %" PRId64, current_txn->GetTransactionId());
   auto &manager = catalog::Manager::GetInstance();

src/concurrency/transaction_manager.cpp

@@ -31,16 +31,10 @@ ConflictAvoidanceType TransactionManager::conflict_avoidance_ =
     ConflictAvoidanceType::ABORT;
 
 TransactionContext *TransactionManager::BeginTransaction(
-    const size_t thread_id, const IsolationLevelType type) {
+    const size_t thread_id, const IsolationLevelType type, bool read_only) {
   TransactionContext *txn = nullptr;
 
-  if (type == IsolationLevelType::READ_ONLY) {
-    // transaction processing with decentralized epoch manager
-    cid_t read_id = EpochManagerFactory::GetInstance().EnterEpoch(
-        thread_id, TimestampType::SNAPSHOT_READ);
-    txn = new TransactionContext(thread_id, type, read_id);
-
-  } else if (type == IsolationLevelType::SNAPSHOT) {
+  if (type == IsolationLevelType::SNAPSHOT) {
     // transaction processing with decentralized epoch manager
     // the DBMS must acquire
     cid_t read_id = EpochManagerFactory::GetInstance().EnterEpoch(
@@ -66,6 +60,10 @@ TransactionContext *TransactionManager::BeginTransaction(
     txn = new TransactionContext(thread_id, type, read_id);
   }
 
+  if (read_only) {
+    txn->SetReadOnly();
+  }
+
   if (static_cast<StatsType>(settings::SettingsManager::GetInt(
       settings::SettingId::stats_mode)) != StatsType::INVALID) {
     stats::BackendStatsContext::GetInstance()

src/gc/transaction_level_gc_manager.cpp

@@ -91,7 +91,7 @@ void TransactionLevelGCManager::RecycleTransaction(
 
   epoch_manager.ExitEpoch(txn->GetThreadId(), txn->GetEpochId());
 
-  if (txn->GetIsolationLevel() != IsolationLevelType::READ_ONLY &&
+  if (!txn->IsReadOnly() && \
       txn->GetResult() != ResultType::SUCCESS && txn->IsGCSetEmpty() != true) {
     txn->SetEpochId(epoch_manager.GetNextEpochId());
   }
@@ -146,7 +146,7 @@ int TransactionLevelGCManager::Unlink(const int &thread_id,
 
     // Deallocate the Transaction Context of transactions that don't involve
     // any garbage collection
-    if (txn_ctx->GetIsolationLevel() == IsolationLevelType::READ_ONLY ||
+    if (txn_ctx->IsReadOnly() || \
         txn_ctx->IsGCSetEmpty()) {
       delete txn_ctx;
       continue;

src/include/common/internal_types.h

@@ -422,7 +422,6 @@ enum class IsolationLevelType {
   SNAPSHOT = 2,          // snapshot isolation
   REPEATABLE_READS = 3,  // repeatable reads
   READ_COMMITTED = 4,    // read committed
-  READ_ONLY = 5          // read only
 };
 std::string IsolationLevelTypeToString(IsolationLevelType type);
 IsolationLevelType StringToIsolationLevelType(const std::string &str);

src/include/concurrency/transaction_context.h

@@ -262,8 +262,16 @@ class TransactionContext : public Printable {
    *
    * @return     True if read only, False otherwise.
    */
-  inline bool IsReadOnly() const {
-    return is_written_ == false && insert_count_ == 0;
+  bool IsReadOnly() const {
+    return read_only_;
+  }
+
+  /**
+   * @brief      mark this context as read only
+   *
+   */
+  void SetReadOnly() {
+    read_only_ = true;
   }
 
   /**
@@ -335,6 +343,9 @@ class TransactionContext : public Printable {
   IsolationLevelType isolation_level_;
 
   std::unique_ptr<trigger::TriggerSet> on_commit_triggers_;
+
+  /** one default transaction is NOT 'read only' unless it is marked 'read only' explicitly*/
+  bool read_only_ = false;
 };
 
 }  // namespace concurrency

src/include/concurrency/transaction_manager.h

@@ -219,11 +219,12 @@ class TransactionManager {
   }
 
   TransactionContext *BeginTransaction(const IsolationLevelType type) {
-    return BeginTransaction(0, type);
+    return BeginTransaction(0, type, false);
   }
 
   TransactionContext *BeginTransaction(const size_t thread_id = 0,
-                                const IsolationLevelType type = isolation_level_);
+                                const IsolationLevelType type = isolation_level_,
+                                bool read_only = false);
 
   /**
    * @brief      Ends a transaction.

test/common/internal_types_test.cpp

@@ -646,7 +646,7 @@ TEST_F(InternalTypesTests, IsolationLevelTypeTest) {
   std::vector<IsolationLevelType> list = {
       IsolationLevelType::INVALID,        IsolationLevelType::SERIALIZABLE,
       IsolationLevelType::SNAPSHOT,       IsolationLevelType::REPEATABLE_READS,
-      IsolationLevelType::READ_COMMITTED, IsolationLevelType::READ_ONLY};
+      IsolationLevelType::READ_COMMITTED};
 
   // Make sure that ToString and FromString work
   for (auto val : list) {

test/concurrency/serializable_transaction_test.cpp

@@ -64,17 +64,96 @@ TEST_F(SerializableTransactionTests, ReadOnlyTransactionTest) {
     auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
     // Just scan the table
     {
+      //same as 'ConcurrentReadOnlyTransactionTest'
+      thread_pool.Initialize(0, CONNECTION_THREAD_COUNT + 3);
       concurrency::EpochManagerFactory::GetInstance().Reset();
+      concurrency::EpochManagerFactory::GetInstance().StartEpoch();
+      gc::GCManagerFactory::Configure();
+      gc::GCManagerFactory::GetInstance().StartGC();
+
+      //this consists of 2 txns. 1.catalog creation 2.test table creation
       storage::DataTable *table = TestingTransactionUtil::CreateTable();
-      
-      TransactionScheduler scheduler(1, table, &txn_manager, true);
+
+      //manually update snapshot epoch number, so later snapshot read must get a larger epoch than table creating txn
+      //or it may read nothing
+      //wait one epoch. so that global epoch is guaranteed to increase
+      std::this_thread::sleep_for(std::chrono::milliseconds(EPOCH_LENGTH));
+      concurrency::EpochManagerFactory::GetInstance().GetExpiredEpochId();
+
+      TransactionScheduler scheduler(1, table, &txn_manager, {0});
       scheduler.Txn(0).Scan(0);
       scheduler.Txn(0).Commit();
 
       scheduler.Run();
 
-      // Snapshot read cannot read the recent insert
-      EXPECT_EQ(0, scheduler.schedules[0].results.size());
+      //it should read all the 10 tuples
+      EXPECT_EQ(10, scheduler.schedules[0].results.size());
+
+      gc::GCManagerFactory::GetInstance().StopGC();
+      concurrency::EpochManagerFactory::GetInstance().StopEpoch();
+      thread_pool.Shutdown();
+      //reset to default value, other test cases
+      gc::GCManagerFactory::Configure(0);
+    }
+  }
+}
+
+// test r/w txn with a read-only txn runs concurrently
+TEST_F(SerializableTransactionTests, ConcurrentReadOnlyTransactionTest) {
+  for (auto protocol_type : PROTOCOL_TYPES) {
+    concurrency::TransactionManagerFactory::Configure(protocol_type, ISOLATION_LEVEL_TYPE);
+    auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
+//    Txn #0 | Txn #1
+//    ----------------
+//    BEGIN  |
+//    W(X)   |
+//           | BEGIN R/O
+//           | R(X)
+//    W(X)   |
+//    COMMIT |
+//           | R(X)
+//           | COMMIT
+
+    {
+      //if gc manager is active, finishing a txn will make the txn be removed from epoch list as well.
+      //epoch manager needs this behavior to find the largest expired txn id.
+      //that id is used to determine whether snapshot epoch falls behind and needs update.
+      //gc and epoch manager both depend on thread pool
+      thread_pool.Initialize(0, CONNECTION_THREAD_COUNT + 3);
+      concurrency::EpochManagerFactory::GetInstance().Reset();
+      concurrency::EpochManagerFactory::GetInstance().StartEpoch();
+      gc::GCManagerFactory::Configure();
+      gc::GCManagerFactory::GetInstance().StartGC();
+
+      //this contains 2 txns: 1.create catalog table 2.create test table
+      storage::DataTable *table = TestingTransactionUtil::CreateTable();
+
+      //force snapshot epoch to be updated. it should be larger than table creation txn's epoch
+      std::this_thread::sleep_for(std::chrono::milliseconds(EPOCH_LENGTH));
+      concurrency::EpochManagerFactory::GetInstance().GetExpiredEpochId();
+
+      TransactionScheduler scheduler(2, table, &txn_manager, {1});
+      scheduler.Txn(0).Update(0, 1);
+      scheduler.Txn(1).Read(0);
+      scheduler.Txn(0).Update(0, 2);
+      scheduler.Txn(0).Commit();
+      scheduler.Txn(1).Read(0);
+      scheduler.Txn(1).Commit();
+
+      scheduler.Run();
+
+      EXPECT_EQ(ResultType::SUCCESS, scheduler.schedules[0].txn_result);
+      EXPECT_EQ(ResultType::SUCCESS, scheduler.schedules[1].txn_result);
+
+      //read only txn should read the same snapshot that exists after table creation and before update txn commits
+      EXPECT_EQ(0, scheduler.schedules[1].results[0]);
+      EXPECT_EQ(0, scheduler.schedules[1].results[1]);
+
+      gc::GCManagerFactory::GetInstance().StopGC();
+      concurrency::EpochManagerFactory::GetInstance().StopEpoch();
+      thread_pool.Shutdown();
+      //reset it to default value for test cases
+      gc::GCManagerFactory::Configure(0);
     }
   }
 }

test/include/concurrency/testing_transaction_util.h

@@ -54,7 +54,7 @@
  *
  * See isolation_level_test.cpp for examples.
  */
-
+#include <initializer_list>
 #include "catalog/schema.h"
 #include "common/harness.h"
 #include "concurrency/transaction_context.h"
@@ -241,7 +241,8 @@ class TransactionThread {
 
     if (cur_seq == 0) {
       if (schedule->declared_ro == true) {
-        txn = txn_manager->BeginTransaction(IsolationLevelType::READ_ONLY);
+        /** starts a read only transaction*/
+        txn = txn_manager->BeginTransaction(0, IsolationLevelType::SNAPSHOT, true);
       } else {
         txn = txn_manager->BeginTransaction();
       }
@@ -348,13 +349,14 @@ class TransactionScheduler {
  public:
   TransactionScheduler(size_t num_txn, storage::DataTable *datatable_,
                        concurrency::TransactionManager *txn_manager_,
-                       bool first_as_ro = false)
+                       std::set<int> read_only_ = {})
       : txn_manager(txn_manager_),
         table(datatable_),
         time(0),
         concurrent(false) {
+
     for (size_t i = 0; i < num_txn; i++) {
-      if (first_as_ro && i == 0) {
+      if (read_only_.find(i) != read_only_.end()) {
         schedules.emplace_back(i, true);
       } else {
         schedules.emplace_back(i, false);