refactor p4 (#551)

* refactor p4

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

* add eager iter

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

* rm

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

* add checker

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

* fix

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

* use simplified protocol

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

* fix

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

* fix

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

* fix

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

---------

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

Author: skyzh

src/common/bustub_instance.cpp

@@ -38,8 +38,8 @@
 
 namespace bustub {
 
-auto BustubInstance::MakeExecutorContext(Transaction *txn) -> std::unique_ptr<ExecutorContext> {
-  return std::make_unique<ExecutorContext>(txn, catalog_, buffer_pool_manager_, txn_manager_, lock_manager_);
+auto BustubInstance::MakeExecutorContext(Transaction *txn, bool is_modify) -> std::unique_ptr<ExecutorContext> {
+  return std::make_unique<ExecutorContext>(txn, catalog_, buffer_pool_manager_, txn_manager_, lock_manager_, is_modify);
 }
 
 BustubInstance::BustubInstance(const std::string &db_file_name) {
@@ -70,6 +70,8 @@ BustubInstance::BustubInstance(const std::string &db_file_name) {
 
   txn_manager_ = new TransactionManager(lock_manager_, log_manager_);
 
+  lock_manager_->txn_manager_ = txn_manager_;
+
   // Checkpoint related.
   checkpoint_manager_ = new CheckpointManager(txn_manager_, log_manager_, buffer_pool_manager_);
 
@@ -108,6 +110,8 @@ BustubInstance::BustubInstance() {
 
   txn_manager_ = new TransactionManager(lock_manager_, log_manager_);
 
+  lock_manager_->txn_manager_ = txn_manager_;
+
   // Checkpoint related.
   checkpoint_manager_ = new CheckpointManager(txn_manager_, log_manager_, buffer_pool_manager_);
 
@@ -226,6 +230,8 @@ auto BustubInstance::ExecuteSqlTxn(const std::string &sql, ResultWriter &writer,
   binder.ParseAndSave(sql);
   l.unlock();
 
+  bool is_delete = false;
+
   for (auto *stmt : binder.statement_nodes_) {
     auto statement = binder.BindStatement(stmt);
     switch (statement->type_) {
@@ -254,6 +260,8 @@ auto BustubInstance::ExecuteSqlTxn(const std::string &sql, ResultWriter &writer,
         HandleExplainStatement(txn, explain_stmt, writer);
         continue;
       }
+      case StatementType::DELETE_STATEMENT:
+        is_delete = true;
       default:
         break;
     }
@@ -271,7 +279,7 @@ auto BustubInstance::ExecuteSqlTxn(const std::string &sql, ResultWriter &writer,
     l.unlock();
 
     // Execute the query.
-    auto exec_ctx = MakeExecutorContext(txn);
+    auto exec_ctx = MakeExecutorContext(txn, is_delete);
     if (check_options != nullptr) {
       exec_ctx->InitCheckOptions(std::move(check_options));
     }
@@ -310,7 +318,7 @@ auto BustubInstance::ExecuteSqlTxn(const std::string &sql, ResultWriter &writer,
  */
 void BustubInstance::GenerateTestTable() {
   auto txn = txn_manager_->Begin();
-  auto exec_ctx = MakeExecutorContext(txn);
+  auto exec_ctx = MakeExecutorContext(txn, false);
   TableGenerator gen{exec_ctx.get()};
 
   std::shared_lock<std::shared_mutex> l(catalog_lock_);

src/include/common/bustub_instance.h

@@ -215,7 +215,7 @@ class BustubInstance {
   /**
    * Get the executor context from the BusTub instance.
    */
-  auto MakeExecutorContext(Transaction *txn) -> std::unique_ptr<ExecutorContext>;
+  auto MakeExecutorContext(Transaction *txn, bool is_modify) -> std::unique_ptr<ExecutorContext>;
 
  public:
   explicit BustubInstance(const std::string &db_file_name);

src/include/common/exception.h

@@ -57,10 +57,13 @@ class Exception : public std::runtime_error {
    * Construct a new Exception instance.
    * @param message The exception message
    */
-  explicit Exception(const std::string &message) : std::runtime_error(message), type_(ExceptionType::INVALID) {
+  explicit Exception(const std::string &message, bool print = true)
+      : std::runtime_error(message), type_(ExceptionType::INVALID) {
 #ifndef NDEBUG
-    std::string exception_message = "Message :: " + message + "\n";
-    std::cerr << exception_message;
+    if (print) {
+      std::string exception_message = "Message :: " + message + "\n";
+      std::cerr << exception_message;
+    }
 #endif
   }
 
@@ -69,12 +72,14 @@ class Exception : public std::runtime_error {
    * @param exception_type The exception type
    * @param message The exception message
    */
-  Exception(ExceptionType exception_type, const std::string &message)
+  Exception(ExceptionType exception_type, const std::string &message, bool print = true)
       : std::runtime_error(message), type_(exception_type) {
 #ifndef NDEBUG
-    std::string exception_message =
-        "\nException Type :: " + ExceptionTypeToString(type_) + "\nMessage :: " + message + "\n";
-    std::cerr << exception_message;
+    if (print) {
+      std::string exception_message =
+          "\nException Type :: " + ExceptionTypeToString(type_) + "\nMessage :: " + message + "\n";
+      std::cerr << exception_message;
+    }
 #endif
   }
 
@@ -122,7 +127,7 @@ class NotImplementedException : public Exception {
 class ExecutionException : public Exception {
  public:
   ExecutionException() = delete;
-  explicit ExecutionException(const std::string &msg) : Exception(ExceptionType::EXECUTION, msg) {}
+  explicit ExecutionException(const std::string &msg) : Exception(ExceptionType::EXECUTION, msg, false) {}
 };
 
 }  // namespace bustub

src/include/concurrency/lock_manager.h

@@ -298,6 +298,8 @@ class LockManager {
    */
   auto RunCycleDetection() -> void;
 
+  TransactionManager *txn_manager_;
+
  private:
   /** Fall 2022 */
   /** Structure that holds lock requests for a given table oid */

src/include/concurrency/transaction.h

@@ -12,6 +12,7 @@
 
 #pragma once
 
+#include <fmt/format.h>
 #include <atomic>
 #include <deque>
 #include <memory>
@@ -372,3 +373,25 @@ class Transaction {
 };
 
 }  // namespace bustub
+
+template <>
+struct fmt::formatter<bustub::IsolationLevel> : formatter<std::string_view> {
+  // parse is inherited from formatter<string_view>.
+  template <typename FormatContext>
+  auto format(bustub::IsolationLevel x, FormatContext &ctx) const {
+    using bustub::IsolationLevel;
+    string_view name = "unknown";
+    switch (x) {
+      case IsolationLevel::READ_UNCOMMITTED:
+        name = "READ_UNCOMMITTED";
+        break;
+      case IsolationLevel::READ_COMMITTED:
+        name = "READ_COMMITTED";
+        break;
+      case IsolationLevel::REPEATABLE_READ:
+        name = "REPEATABLE_READ";
+        break;
+    }
+    return formatter<string_view>::format(name, ctx);
+  }
+};

src/include/execution/execution_engine.h

@@ -66,9 +66,6 @@ class ExecutionEngine {
       PollExecutor(executor.get(), plan, result_set);
       PerformChecks(exec_ctx);
     } catch (const ExecutionException &ex) {
-#ifndef NDEBUG
-      LOG_ERROR("Error Encountered in Executor Execution: %s", ex.what());
-#endif
       executor_succeeded = false;
       if (result_set != nullptr) {
         result_set->clear();

src/include/execution/executor_context.h

@@ -40,8 +40,13 @@ class ExecutorContext {
    * @param lock_mgr The lock manager that the executor uses
    */
   ExecutorContext(Transaction *transaction, Catalog *catalog, BufferPoolManager *bpm, TransactionManager *txn_mgr,
-                  LockManager *lock_mgr)
-      : transaction_(transaction), catalog_{catalog}, bpm_{bpm}, txn_mgr_(txn_mgr), lock_mgr_(lock_mgr) {
+                  LockManager *lock_mgr, bool is_delete)
+      : transaction_(transaction),
+        catalog_{catalog},
+        bpm_{bpm},
+        txn_mgr_(txn_mgr),
+        lock_mgr_(lock_mgr),
+        is_delete_(is_delete) {
     nlj_check_exec_set_ = std::deque<std::pair<AbstractExecutor *, AbstractExecutor *>>(
         std::deque<std::pair<AbstractExecutor *, AbstractExecutor *>>{});
     check_options_ = std::make_shared<CheckOptions>();
@@ -86,6 +91,8 @@ class ExecutorContext {
     check_options_ = std::move(check_options);
   }
 
+  auto IsDelete() const -> bool { return is_delete_; }
+
  private:
   /** The transaction context associated with this executor context */
   Transaction *transaction_;
@@ -101,6 +108,7 @@ class ExecutorContext {
   std::deque<std::pair<AbstractExecutor *, AbstractExecutor *>> nlj_check_exec_set_;
   /** The set of check options associated with this executor context */
   std::shared_ptr<CheckOptions> check_options_;
+  bool is_delete_;
 };
 
 }  // namespace bustub

src/include/storage/table/table_heap.h

@@ -18,6 +18,8 @@
 
 #include "buffer/buffer_pool_manager.h"
 #include "common/config.h"
+#include "concurrency/lock_manager.h"
+#include "concurrency/transaction.h"
 #include "recovery/log_manager.h"
 #include "storage/page/table_page.h"
 #include "storage/table/table_iterator.h"
@@ -48,7 +50,8 @@ class TableHeap {
    * @param tuple tuple to insert
    * @return rid of the inserted tuple
    */
-  auto InsertTuple(const TupleMeta &meta, const Tuple &tuple) -> std::optional<RID>;
+  auto InsertTuple(const TupleMeta &meta, const Tuple &tuple, LockManager *lock_mgr = nullptr,
+                   Transaction *txn = nullptr, table_oid_t oid = 0) -> std::optional<RID>;
 
   /**
    * Insert a tuple into the table. If the tuple is too large (>= page_size), return false.
@@ -72,9 +75,12 @@ class TableHeap {
    */
   auto GetTupleMeta(RID rid) -> TupleMeta;
 
-  /** @return the iterator of this table */
+  /** @return the iterator of this table, use this for project 3 */
   auto MakeIterator() -> TableIterator;
 
+  /** @return the iterator of this table, use this for project 4 except updates */
+  auto MakeEagerIterator() -> TableIterator;
+
   /** @return the id of the first page of this table */
   inline auto GetFirstPageId() const -> page_id_t { return first_page_id_; }
 

src/storage/table/table_heap.cpp

@@ -18,6 +18,7 @@
 #include "common/exception.h"
 #include "common/logger.h"
 #include "common/macros.h"
+#include "concurrency/transaction.h"
 #include "fmt/format.h"
 #include "storage/page/page_guard.h"
 #include "storage/page/table_page.h"
@@ -35,7 +36,8 @@ TableHeap::TableHeap(BufferPoolManager *bpm) : bpm_(bpm) {
   first_page->Init();
 }
 
-auto TableHeap::InsertTuple(const TupleMeta &meta, const Tuple &tuple) -> std::optional<RID> {
+auto TableHeap::InsertTuple(const TupleMeta &meta, const Tuple &tuple, LockManager *lock_mgr, Transaction *txn,
+                            table_oid_t oid) -> std::optional<RID> {
   std::unique_lock<std::mutex> guard(latch_);
   auto page_guard = bpm_->FetchPageWrite(last_page_id_);
   while (true) {
@@ -65,11 +67,17 @@ auto TableHeap::InsertTuple(const TupleMeta &meta, const Tuple &tuple) -> std::o
     page_guard = std::move(next_page_guard);
   }
   auto last_page_id = last_page_id_;
-  guard.unlock();
 
   auto page = page_guard.AsMut<TablePage>();
   auto slot_id = *page->InsertTuple(meta, tuple);
 
+  // only allow one insertion at a time, otherwise it will deadlock.
+  guard.unlock();
+
+  if (lock_mgr != nullptr) {
+    lock_mgr->LockRow(txn, LockManager::LockMode::EXCLUSIVE, oid, RID{last_page_id, slot_id});
+  }
+
   page_guard.Drop();
 
   return RID(last_page_id, slot_id);
@@ -105,6 +113,8 @@ auto TableHeap::MakeIterator() -> TableIterator {
   return {this, {first_page_id_, 0}, {last_page_id, page->GetNumTuples()}};
 }
 
+auto TableHeap::MakeEagerIterator() -> TableIterator { return {this, {first_page_id_, 0}, {INVALID_PAGE_ID, 0}}; }
+
 void TableHeap::UpdateTupleInPlaceUnsafe(const TupleMeta &meta, const Tuple &tuple, RID rid) {
   auto page_guard = bpm_->FetchPageWrite(rid.GetPageId());
   auto page = page_guard.AsMut<TablePage>();

src/storage/table/table_iterator.cpp

@@ -42,11 +42,13 @@ auto TableIterator::operator++() -> TableIterator & {
   auto page = page_guard.As<TablePage>();
   auto next_tuple_id = rid_.GetSlotNum() + 1;
 
-  BUSTUB_ASSERT(
-      /* case 1: cursor before the page of the stop tuple */ rid_.GetPageId() < stop_at_rid_.GetPageId() ||
-          /* case 2: cursor at the page before the tuple */
-          (rid_.GetPageId() == stop_at_rid_.GetPageId() && next_tuple_id <= stop_at_rid_.GetSlotNum()),
-      "iterate out of bound");
+  if (stop_at_rid_.GetPageId() != INVALID_PAGE_ID) {
+    BUSTUB_ASSERT(
+        /* case 1: cursor before the page of the stop tuple */ rid_.GetPageId() < stop_at_rid_.GetPageId() ||
+            /* case 2: cursor at the page before the tuple */
+            (rid_.GetPageId() == stop_at_rid_.GetPageId() && next_tuple_id <= stop_at_rid_.GetSlotNum()),
+        "iterate out of bound");
+  }
 
   rid_ = RID{rid_.GetPageId(), next_tuple_id};