p4: verify total count for terrier bench + more comments for starter code (#656)

Signed-off-by: Alex Chi <[email protected]>

Author: skyzh

src/concurrency/transaction_manager.cpp

@@ -53,6 +53,8 @@ auto TransactionManager::VerifyTxn(Transaction *txn) -> bool { return true; }
 auto TransactionManager::Commit(Transaction *txn) -> bool {
   std::unique_lock<std::mutex> commit_lck(commit_mutex_);
 
+  // TODO(fall2023): acquire commit ts!
+
   if (txn->state_ != TransactionState::RUNNING) {
     throw Exception("txn not in running state");
   }

src/concurrency/transaction_manager_impl.cpp

@@ -25,6 +25,19 @@
 
 namespace bustub {
 
+auto TransactionManager::UpdateUndoLink(RID rid, std::optional<UndoLink> prev_link,
+                                        std::function<bool(std::optional<UndoLink>)> &&check) -> bool {
+  std::function<bool(std::optional<VersionUndoLink>)> wrapper_func =
+      [check](std::optional<VersionUndoLink> link) -> bool {
+    if (link.has_value()) {
+      return check(link->prev_);
+    }
+    return check(std::nullopt);
+  };
+  return UpdateVersionLink(rid, prev_link.has_value() ? std::make_optional(VersionUndoLink{*prev_link}) : std::nullopt,
+                           check != nullptr ? wrapper_func : nullptr);
+}
+
 auto TransactionManager::UpdateVersionLink(RID rid, std::optional<VersionUndoLink> prev_version,
                                            std::function<bool(std::optional<VersionUndoLink>)> &&check) -> bool {
   std::unique_lock<std::shared_mutex> lck(version_info_mutex_);

src/include/concurrency/transaction_manager.h

@@ -79,21 +79,40 @@ class TransactionManager {
    */
   void Abort(Transaction *txn);
 
+  /**
+   * @brief Use this function before task 4.2. Update an undo link that links table heap tuple to the first undo log.
+   * Before updating, `check` function will be called to ensure validity.
+   */
+  auto UpdateUndoLink(RID rid, std::optional<UndoLink> prev_link,
+                      std::function<bool(std::optional<UndoLink>)> &&check = nullptr) -> bool;
+
+  /**
+   * @brief Use this function after task 4.2. Update an undo link that links table heap tuple to the first undo log.
+   * Before updating, `check` function will be called to ensure validity.
+   */
   auto UpdateVersionLink(RID rid, std::optional<VersionUndoLink> prev_version,
                          std::function<bool(std::optional<VersionUndoLink>)> &&check = nullptr) -> bool;
 
-  /** The same as `GetVersionLink`, except that we extracted the undo link field out. */
+  /** @brief Get the first undo log of a table heap tuple. Use this before task 4.2 */
   auto GetUndoLink(RID rid) -> std::optional<UndoLink>;
 
-  /** You only need this starting task 4.2 */
+  /** @brief Get the first undo log of a table heap tuple. Use this after task 4.2 */
   auto GetVersionLink(RID rid) -> std::optional<VersionUndoLink>;
 
+  /** @brief Access the transaction undo log buffer and get the undo log. Return nullopt if the txn does not exist. Will
+   * still throw an exception if the index is out of range. */
   auto GetUndoLogOptional(UndoLink link) -> std::optional<UndoLog>;
 
+  /** @brief Access the transaction undo log buffer and get the undo log. Except when accessing the current txn buffer,
+   * you should always call this function to get the undo log instead of manually retrieve the txn shared_ptr and access
+   * the buffer. */
   auto GetUndoLog(UndoLink link) -> UndoLog;
 
+  /** @brief Get the lowest read timestamp in the system. */
   auto GetWatermark() -> timestamp_t { return running_txns_.GetWatermark(); }
 
+  /** @brief Stop-the-world garbage collection. Will be called only when all transactions are not accessing the table
+   * heap. */
   void GarbageCollection();
 
   /** protects txn map */
@@ -112,7 +131,8 @@ class TransactionManager {
 
   /** protects version info */
   std::shared_mutex version_info_mutex_;
-  /** Stores the previous version of each tuple in the table heap. */
+  /** Stores the previous version of each tuple in the table heap. Do not directly access this field. Use the helper
+   * functions in `transaction_manager_impl.cpp`. */
   std::unordered_map<page_id_t, std::shared_ptr<PageVersionInfo>> version_info_;
 
   /** Stores all the read_ts of running txns so as to facilitate garbage collection. */
@@ -129,6 +149,8 @@ class TransactionManager {
   std::atomic<txn_id_t> next_txn_id_{TXN_START_ID};
 
  private:
+  /** @brief Verify if a txn satisfies serializability. We will not test this function and you can change / remove it as
+   * you want. */
   auto VerifyTxn(Transaction *txn) -> bool;
 };
 

src/include/storage/table/table_heap.h

@@ -29,6 +29,8 @@
 
 namespace bustub {
 
+class TablePage;
+
 /**
  * TableHeap represents a physical table on disk.
  * This is just a doubly-linked list of pages.
@@ -77,20 +79,27 @@ class TableHeap {
    */
   auto GetTupleMeta(RID rid) -> TupleMeta;
 
-  /** @return the iterator of this table, use this for project 3 */
+  /** @return the iterator of this table. When this iterator is created, it will record the current last tuple in the
+   * table heap, and the iterator will stop at that point, in order to avoid halloween problem. You usually will need to
+   * use this function for project 3. Given that you have already implemented your project 4 update executor as a
+   * pipeline breaker, you may use `MakeEagerIterator` to test whether the update executor is implemented correctly.
+   * There should be no difference between this function and `MakeEagerIterator` in project 4 if everything is
+   * implemented correctly. */
   auto MakeIterator() -> TableIterator;
 
-  /** @return the iterator of this table, use this for project 4 except updates */
+  /** @return the iterator of this table. The iterator will stop at the last tuple at the time of iterating. */
   auto MakeEagerIterator() -> TableIterator;
 
   /** @return the id of the first page of this table */
   inline auto GetFirstPageId() const -> page_id_t { return first_page_id_; }
 
   /**
-   * Update a tuple in place. SHOULD NOT BE USED UNLESS YOU WANT TO OPTIMIZE FOR PROJECT 4.
+   * Update a tuple in place. Should NOT be used in project 3. Implement your project 3 update executor as delete and
+   * insert. You will need to use this function in project 4.
    * @param meta new tuple meta
    * @param tuple  new tuple
-   * @param[out] rid the rid of the tuple to be updated
+   * @param rid the rid of the tuple to be updated
+   * @param check the check to run before actually update.
    */
   auto UpdateTupleInPlace(const TupleMeta &meta, const Tuple &tuple, RID rid,
                           std::function<bool(const TupleMeta &meta, const Tuple &table, RID rid)> &&check = nullptr)
@@ -103,6 +112,19 @@ class TableHeap {
     return std::unique_ptr<TableHeap>(new TableHeap(create_table_heap));
   }
 
+  // The below functions are useful only when you want to implement abort in a way that removes an undo log from the
+  // version chain. DO NOT USE THEM if you are unsure what they are supposed to do.
+
+  auto AcquireTablePageReadLock(RID rid) -> ReadPageGuard;
+
+  auto AcquireTablePageWriteLock(RID rid) -> WritePageGuard;
+
+  void UpdateTupleInPlaceWithLockAcquired(const TupleMeta &meta, const Tuple &tuple, RID rid, TablePage *page);
+
+  auto GetTupleWithLockAcquired(RID rid, TablePage *page) -> std::pair<TupleMeta, Tuple>;
+
+  auto GetTupleMetaWithLockAcquired(RID rid, TablePage *page) -> TupleMeta;
+
  private:
   /** Used for binder tests */
   explicit TableHeap(bool create_table_heap = false);

src/storage/table/table_heap.cpp

@@ -134,4 +134,21 @@ auto TableHeap::UpdateTupleInPlace(const TupleMeta &meta, const Tuple &tuple, RI
   return false;
 }
 
+auto TableHeap::AcquireTablePageReadLock(RID rid) -> ReadPageGuard { return bpm_->FetchPageRead(rid.GetPageId()); }
+
+auto TableHeap::AcquireTablePageWriteLock(RID rid) -> WritePageGuard { return bpm_->FetchPageWrite(rid.GetPageId()); }
+
+void TableHeap::UpdateTupleInPlaceWithLockAcquired(const TupleMeta &meta, const Tuple &tuple, RID rid,
+                                                   TablePage *page) {
+  page->UpdateTupleInPlaceUnsafe(meta, tuple, rid);
+}
+
+auto TableHeap::GetTupleWithLockAcquired(RID rid, TablePage *page) -> std::pair<TupleMeta, Tuple> {
+  auto [meta, tuple] = page->GetTuple(rid);
+  tuple.rid_ = rid;
+  return std::make_pair(meta, std::move(tuple));
+}
+
+auto TableHeap::GetTupleMetaWithLockAcquired(RID rid, TablePage *page) -> TupleMeta { return page->GetTupleMeta(rid); }
+
 }  // namespace bustub

test/txn/txn_executor_test.cpp

@@ -1,8 +1,6 @@
 #include "execution/execution_common.h"
 #include "txn_common.h"  // NOLINT
 
-// NO HIDDEN TEST IN THIS FILE!!!
-
 namespace bustub {
 
 // NOLINTBEGIN(bugprone-unchecked-optional-access)

test/txn/txn_scan_test.cpp

@@ -32,29 +32,73 @@ TEST(TxnScanTest, DISABLED_TupleReconstructTest) {  // NOLINT
     }
   }
   {
-    fmt::println(stderr, "(partial) C: reconstruct 4 records, one of them is empty, one of them is full change");
+    fmt::println(stderr, "C: reconstruct 4 records, one of them is empty, one of them is full change");
     auto base_tuple = Tuple{{Int(0), Double(1), BoolNull()}, schema.get()};
     auto base_meta = TupleMeta{2333, false};
     auto undo_log_1_schema = ParseCreateStatement("");
     auto undo_log_1 = UndoLog{false, {false, false, false}, Tuple{{}, undo_log_1_schema.get()}};
+    auto undo_log_2_schema = ParseCreateStatement("b double");
+    auto undo_log_2 = UndoLog{false, {false, true, false}, Tuple{{Double(2)}, undo_log_2_schema.get()}};
+    auto undo_log_3_schema = ParseCreateStatement("a integer,c boolean");
+    auto undo_log_3 = UndoLog{false, {true, false, true}, Tuple{{Int(3), Bool(false)}, undo_log_3_schema.get()}};
+    auto undo_log_4 = UndoLog{false, {true, true, true}, Tuple{{Int(4), Double(4), Bool(true)}, schema.get()}};
     {
       fmt::println(stderr, "C1: verify 1st record");
       auto tuple = ReconstructTuple(schema.get(), base_tuple, base_meta, {undo_log_1});
       ASSERT_TRUE(tuple.has_value());
       VerifyTuple(schema.get(), *tuple, {Int(0), Double(1), BoolNull()});
     }
+    {
+      fmt::println(stderr, "C2: verify 2nd record");
+      auto tuple = ReconstructTuple(schema.get(), base_tuple, base_meta, {undo_log_1, undo_log_2});
+      ASSERT_TRUE(tuple.has_value());
+      VerifyTuple(schema.get(), *tuple, {Int(0), Double(2), BoolNull()});
+    }
+    {
+      fmt::println(stderr, "C3: verify 3rd record");
+      auto tuple = ReconstructTuple(schema.get(), base_tuple, base_meta, {undo_log_1, undo_log_2, undo_log_3});
+      ASSERT_TRUE(tuple.has_value());
+      VerifyTuple(schema.get(), *tuple, {Int(3), Double(2), Bool(false)});
+    }
+    {
+      fmt::println(stderr, "C4: verify 4th record");
+      auto tuple =
+          ReconstructTuple(schema.get(), base_tuple, base_meta, {undo_log_1, undo_log_2, undo_log_3, undo_log_4});
+      ASSERT_TRUE(tuple.has_value());
+      VerifyTuple(schema.get(), *tuple, {Int(4), Double(4), Bool(true)});
+    }
   }
   {
-    fmt::println(stderr, "(partial) D: reconstruct 4 records, two of them are delete");
+    fmt::println(stderr, "D: reconstruct 4 records, two of them are delete");
     auto base_tuple = Tuple{{Int(0), Double(1), BoolNull()}, schema.get()};
     auto base_meta = TupleMeta{2333, false};
     auto delete_schema = ParseCreateStatement("");
     auto delete_log = UndoLog{true, {false, false, false}, Tuple{{}, delete_schema.get()}};
+    auto undo_log_1 = UndoLog{false, {true, true, true}, Tuple{{Int(1), Double(1), Bool(false)}, schema.get()}};
+    auto undo_log_2 = UndoLog{false, {true, true, true}, Tuple{{IntNull(), DoubleNull(), BoolNull()}, schema.get()}};
     {
       fmt::println(stderr, "D1: apply delete record");
       auto tuple = ReconstructTuple(schema.get(), base_tuple, base_meta, {delete_log});
       ASSERT_FALSE(tuple.has_value());
     }
+    {
+      fmt::println(stderr, "D2: verify 2nd record");
+      auto tuple = ReconstructTuple(schema.get(), base_tuple, base_meta, {delete_log, undo_log_1});
+      ASSERT_TRUE(tuple.has_value());
+      VerifyTuple(schema.get(), *tuple, {Int(1), Double(1), Bool(false)});
+    }
+    {
+      fmt::println(stderr, "D3: apply delete record");
+      auto tuple = ReconstructTuple(schema.get(), base_tuple, base_meta, {delete_log, undo_log_1, delete_log});
+      ASSERT_FALSE(tuple.has_value());
+    }
+    {
+      fmt::println(stderr, "D4: verify 4th record");
+      auto tuple =
+          ReconstructTuple(schema.get(), base_tuple, base_meta, {delete_log, undo_log_1, delete_log, undo_log_2});
+      ASSERT_TRUE(tuple.has_value());
+      VerifyTuple(schema.get(), *tuple, {IntNull(), DoubleNull(), BoolNull()});
+    }
   }
 }
 
@@ -161,7 +205,18 @@ TEST(TxnScanTest, DISABLED_ScanTest) {  // NOLINT
   WithTxn(txn0, QueryShowResult(*bustub, _var, _txn, query, IntResult{}));
   fmt::println(stderr, "B: Verify txn1");
   WithTxn(txn1, QueryShowResult(*bustub, _var, _txn, query, IntResult{{2}, {4}, {7}}));
-  // hidden tests...
+
+  // hidden tests... this is the only hidden test case among task 1, 2, 3. We recommend you to implement `TxnMgrDbg`
+  // function, draw the version chain out, and think of what should be read by each txn.
+
+  // fmt::println(stderr, "C: Verify txn2");
+  // WithTxn(txn2, QueryHideResult(*bustub, _var, _txn, query, IntResult{}));
+  // fmt::println(stderr, "D: Verify txn3");
+  // WithTxn(txn3, QueryHideResult(*bustub, _var, _txn, query, IntResult{}));
+  // fmt::println(stderr, "E: Verify txn4");
+  // WithTxn(txn4, QueryHideResult(*bustub, _var, _txn, query, IntResult{}));
+  // fmt::println(stderr, "F: Verify txn5");
+  // WithTxn(txn5, QueryHideResult(*bustub, _var, _txn, query, IntResult{}));
 }
 
 // NOLINTEND(bugprone-unchecked-optional-access))

test/txn/txn_timestamp_test.cpp

@@ -1,4 +1,18 @@
-#include "txn_common.h"  // NOLINT
+#include <chrono>  // NOLINT
+#include <cstdio>
+#include <exception>
+#include <functional>
+#include <future>  // NOLINT
+#include <memory>
+#include <stdexcept>
+#include <thread>  // NOLINT
+
+#include "common/bustub_instance.h"
+#include "concurrency/transaction.h"
+#include "concurrency/transaction_manager.h"
+#include "concurrency/watermark.h"
+#include "fmt/core.h"
+#include "gtest/gtest.h"
 
 namespace bustub {
 
@@ -36,19 +50,96 @@ TEST(TxnTsTest, DISABLED_WatermarkPerformance) {  // NOLINT
 
 TEST(TxnTsTest, DISABLED_TimestampTracking) {  // NOLINT
   auto bustub = std::make_unique<BustubInstance>();
+
   auto txn0 = bustub->txn_manager_->Begin();
   ASSERT_EQ(txn0->GetReadTs(), 0);
   ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 0);
-  bustub->txn_manager_->Commit(txn0);
-  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 1);
+
+  {
+    auto txn_store_1 = bustub->txn_manager_->Begin();
+    ASSERT_EQ(txn_store_1->GetReadTs(), 0);
+    bustub->txn_manager_->Commit(txn_store_1);
+    ASSERT_EQ(txn_store_1->GetCommitTs(), 1);
+  }
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 0);
 
   auto txn1 = bustub->txn_manager_->Begin();
   ASSERT_EQ(txn1->GetReadTs(), 1);
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 0);
+
+  {
+    auto txn_store_2 = bustub->txn_manager_->Begin();
+    ASSERT_EQ(txn_store_2->GetReadTs(), 1);
+    bustub->txn_manager_->Commit(txn_store_2);
+    ASSERT_EQ(txn_store_2->GetCommitTs(), 2);
+  }
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 0);
+
+  auto txn2 = bustub->txn_manager_->Begin();
+  ASSERT_EQ(txn2->GetReadTs(), 2);
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 0);
+
+  bustub->txn_manager_->Abort(txn0);
   ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 1);
-  bustub->txn_manager_->Abort(txn1);
+
+  {
+    auto txn_store_3 = bustub->txn_manager_->Begin();
+    ASSERT_EQ(txn_store_3->GetReadTs(), 2);
+    bustub->txn_manager_->Commit(txn_store_3);
+    ASSERT_EQ(txn_store_3->GetCommitTs(), 3);
+  }
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 1);
+
+  auto txn3 = bustub->txn_manager_->Begin();
+  ASSERT_EQ(txn3->GetReadTs(), 3);
+
   ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 1);
 
-  // test case continues on Gradescope...
+  bustub->txn_manager_->Abort(txn1);
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 2);
+  bustub->txn_manager_->Abort(txn2);
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 3);
+
+  {
+    auto txn_store_4 = bustub->txn_manager_->Begin();
+    ASSERT_EQ(txn_store_4->GetReadTs(), 3);
+    bustub->txn_manager_->Commit(txn_store_4);
+    ASSERT_EQ(txn_store_4->GetCommitTs(), 4);
+  }
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 3);
+
+  auto txn4 = bustub->txn_manager_->Begin();
+  ASSERT_EQ(txn4->GetReadTs(), 4);
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 3);
+  bustub->txn_manager_->Abort(txn3);
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 4);
+  bustub->txn_manager_->Abort(txn4);
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 4);
+
+  {
+    auto txn_store_5 = bustub->txn_manager_->Begin();
+    ASSERT_EQ(txn_store_5->GetTransactionState(), TransactionState::RUNNING);
+    ASSERT_EQ(txn_store_5->GetReadTs(), 4);
+    bustub->txn_manager_->Commit(txn_store_5);
+    ASSERT_EQ(txn_store_5->GetTransactionState(), TransactionState::COMMITTED);
+  }
+
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 5);
+
+  auto txn5 = bustub->txn_manager_->Begin();
+  ASSERT_EQ(txn5->GetTransactionState(), TransactionState::RUNNING);
+  ASSERT_EQ(txn5->GetReadTs(), 5);
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 5);
+  bustub->txn_manager_->Abort(txn5);
+  ASSERT_EQ(txn5->GetTransactionState(), TransactionState::ABORTED);
+  ASSERT_EQ(bustub->txn_manager_->GetWatermark(), 5);
 }
 
 // NOLINTEND(bugprone-unchecked-optional-access))