p4: more comments and more tests (#658)

* p4: more comments and more tests

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

* add nc-shell

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

* add abort/serializable tests

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

* fix shell

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

* fix compile error

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

* update shell

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

* fix comments

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

* fix comments

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

* txn support for bustub shell

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

* support wasm shell

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

* adjust txn

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

---------

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

Author: skyzh

CMakeLists.txt

@@ -369,11 +369,12 @@ add_custom_target(submit-p3
 )
 
 set(P4_FILES
-        "src/include/concurrency/lock_manager.h"
-        "src/concurrency/lock_manager.cpp"
         "src/include/concurrency/transaction_manager.h"
         "src/concurrency/transaction_manager.cpp"
-        "tools/terrier_bench/terrier_bench_config.h"
+        "src/include/concurrency/watermark.h"
+        "src/concurrency/watermark.cpp"
+        "src/include/execution/execution_common.h"
+        "src/execution/execution_common.cpp"
         ${P3_FILES}
 )
 

src/binder/bind_variable.cpp

@@ -5,6 +5,7 @@
 #include "binder/expressions/bound_constant.h"
 #include "binder/statement/set_show_statement.h"
 #include "common/exception.h"
+#include "nodes/parsenodes.hpp"
 namespace bustub {
 
 auto Binder::BindVariableSet(duckdb_libpgquery::PGVariableSetStmt *stmt) -> std::unique_ptr<VariableSetStatement> {
@@ -23,4 +24,17 @@ auto Binder::BindVariableShow(duckdb_libpgquery::PGVariableShowStmt *stmt) -> st
   return std::make_unique<VariableShowStatement>(stmt->name);
 }
 
+auto Binder::BindTransaction(duckdb_libpgquery::PGTransactionStmt *stmt) -> std::unique_ptr<TransactionStatement> {
+  switch (stmt->kind) {
+    case duckdb_libpgquery::PG_TRANS_STMT_COMMIT:
+      return std::make_unique<TransactionStatement>("commit");
+    case duckdb_libpgquery::PG_TRANS_STMT_ROLLBACK:
+      return std::make_unique<TransactionStatement>("abort");
+    case duckdb_libpgquery::PG_TRANS_STMT_BEGIN:
+      return std::make_unique<TransactionStatement>("begin");
+    default:
+      throw bustub::NotImplementedException("unsupported txn statement kind");
+  }
+}
+
 }  // namespace bustub

src/binder/transformer.cpp

@@ -72,6 +72,8 @@ auto Binder::BindStatement(duckdb_libpgquery::PGNode *stmt) -> std::unique_ptr<B
       return BindVariableSet(reinterpret_cast<duckdb_libpgquery::PGVariableSetStmt *>(stmt));
     case duckdb_libpgquery::T_PGVariableShowStmt:
       return BindVariableShow(reinterpret_cast<duckdb_libpgquery::PGVariableShowStmt *>(stmt));
+    case duckdb_libpgquery::T_PGTransactionStmt:
+      return BindTransaction(reinterpret_cast<duckdb_libpgquery::PGTransactionStmt *>(stmt));
     default:
       throw NotImplementedException(NodeTagToString(stmt->type));
   }

src/common/bustub_ddl.cpp

@@ -21,6 +21,7 @@
 #include "common/bustub_instance.h"
 #include "common/enums/statement_type.h"
 #include "common/exception.h"
+#include "common/macros.h"
 #include "common/util/string_util.h"
 #include "concurrency/lock_manager.h"
 #include "concurrency/transaction.h"
@@ -169,4 +170,57 @@ void BustubInstance::HandleVariableSetStatement(Transaction *txn, const Variable
   session_variables_[stmt.variable_] = stmt.value_;
 }
 
+void BustubInstance::HandleTxnStatement(Transaction *txn, const TransactionStatement &stmt, ResultWriter &writer) {
+  if (managed_txn_mode_ && current_txn_ != nullptr) {
+    BUSTUB_ASSERT(current_txn_ == txn, "txn mismatched??");
+  }
+  auto dump_current_txn = [&](const std::string &prefix) {
+    writer.OneCell(fmt::format("{}txn_id={} txn_real_id={} read_ts={} commit_ts={} status={} iso_lvl={}", prefix,
+                               current_txn_->GetTransactionIdHumanReadable(), current_txn_->GetTransactionId(),
+                               current_txn_->GetReadTs(), current_txn_->GetCommitTs(),
+                               current_txn_->GetTransactionState(), current_txn_->GetIsolationLevel()));
+  };
+  if (txn == nullptr) {
+    writer.OneCell("commit / rollback can only be used with txn");
+    return;
+  }
+  if (stmt.type_ == "begin") {
+    if (!managed_txn_mode_) {
+      writer.OneCell("begin statement is only supported in managed txn mode, please use bustub-shell");
+      return;
+    }
+    bool txn_activated = current_txn_ != nullptr;
+    auto iso_lvl = StringUtil::Lower(GetSessionVariable("global_isolation_level"));
+    if (iso_lvl == "serializable") {
+      current_txn_ = txn_manager_->Begin(IsolationLevel::SERIALIZABLE);
+    } else if (iso_lvl == "snapshot_isolation" || iso_lvl.empty()) {
+      current_txn_ = txn_manager_->Begin(IsolationLevel::SNAPSHOT_ISOLATION);
+    } else {
+      throw Exception("unsupported global_isolation_level");
+    }
+    dump_current_txn(txn_activated ? "pause current txn and begin new txn " : "begin txn ");
+    return;
+  }
+  if (stmt.type_ == "commit") {
+    auto res = txn_manager_->Commit(txn);
+    if (res) {
+      writer.OneCell(fmt::format("txn committed, txn_id={}, status={}, read_ts={}, commit_ts={}",
+                                 txn->GetTransactionIdHumanReadable(), txn->GetTransactionState(), txn->GetReadTs(),
+                                 txn->GetCommitTs()));
+    } else {
+      writer.OneCell(fmt::format("txn failed to commit, txn_id={}, status={}, read_ts={}, commit_ts={}",
+                                 txn->GetTransactionIdHumanReadable(), txn->GetTransactionState(), txn->GetReadTs(),
+                                 txn->GetCommitTs()));
+    }
+    current_txn_ = nullptr;
+    return;
+  }
+  if (stmt.type_ == "abort") {
+    txn_manager_->Abort(txn);
+    writer.OneCell(fmt::format("txn aborted, txn_id={}, status={}, read_ts={}", txn->GetTransactionIdHumanReadable(),
+                               txn->GetTransactionState(), txn->GetReadTs()));
+    current_txn_ = nullptr;
+    return;
+  }
+}
 }  // namespace bustub

src/common/bustub_instance.cpp

@@ -16,11 +16,13 @@
 #include "catalog/schema.h"
 #include "catalog/table_generator.h"
 #include "common/bustub_instance.h"
+#include "common/config.h"
 #include "common/exception.h"
 #include "common/util/string_util.h"
 #include "concurrency/lock_manager.h"
 #include "concurrency/transaction.h"
 #include "execution/check_options.h"
+#include "execution/execution_common.h"
 #include "execution/execution_engine.h"
 #include "execution/executor_context.h"
 #include "execution/executors/mock_scan_executor.h"
@@ -147,6 +149,22 @@ BustubInstance::BustubInstance() {
   execution_engine_ = std::make_unique<ExecutionEngine>(buffer_pool_manager_.get(), txn_manager_.get(), catalog_.get());
 }
 
+void BustubInstance::CmdDbgMvcc(const std::vector<std::string> &params, ResultWriter &writer) {
+  if (params.size() != 2) {
+    writer.OneCell("please provide a table name");
+    return;
+  }
+  const auto &table = params[1];
+  writer.OneCell("please view the result in the BusTub console (or Chrome DevTools console), table=" + table);
+  std::shared_lock<std::shared_mutex> lck(catalog_lock_);
+  auto table_info = catalog_->GetTable(table);
+  if (table_info == nullptr) {
+    writer.OneCell("table " + table + " not found");
+    return;
+  }
+  TxnMgrDbg("\\dbgmvcc", txn_manager_.get(), table_info, table_info->table_.get());
+}
+
 void BustubInstance::CmdDisplayTables(ResultWriter &writer) {
   auto table_names = catalog_->GetTableNames();
   writer.BeginTable(false);
@@ -195,7 +213,12 @@ void BustubInstance::CmdDisplayHelp(ResultWriter &writer) {
 
 \dt: show all tables
 \di: show all indices
+\dbgmvcc <table>: show version chain of a table
 \help: show this message again
+\txn: show current txn information
+\txn <txn_id>: switch to txn
+\txn gc: run garbage collection
+\txn -1: exit txn mode
 
 BusTub shell currently only supports a small set of Postgres queries. We'll set
 up a doc describing the current status later. It will silently ignore some parts
@@ -210,15 +233,20 @@ see the execution plan of your query.
 
 auto BustubInstance::ExecuteSql(const std::string &sql, ResultWriter &writer,
                                 std::shared_ptr<CheckOptions> check_options) -> bool {
-  auto *txn = txn_manager_->Begin();
+  bool is_local_txn = current_txn_ != nullptr;
+  auto *txn = is_local_txn ? current_txn_ : txn_manager_->Begin();
   try {
     auto result = ExecuteSqlTxn(sql, writer, txn, std::move(check_options));
-    if (!txn_manager_->Commit(txn)) {
-      throw Exception("failed to commit txn");
+    if (!is_local_txn) {
+      auto res = txn_manager_->Commit(txn);
+      if (!res) {
+        throw Exception("failed to commit txn");
+      }
     }
     return result;
   } catch (bustub::Exception &ex) {
     txn_manager_->Abort(txn);
+    current_txn_ = nullptr;
     throw ex;
   }
 }
@@ -239,6 +267,16 @@ auto BustubInstance::ExecuteSqlTxn(const std::string &sql, ResultWriter &writer,
       CmdDisplayHelp(writer);
       return true;
     }
+    if (StringUtil::StartsWith(sql, "\\dbgmvcc")) {
+      auto split = StringUtil::Split(sql, " ");
+      CmdDbgMvcc(split, writer);
+      return true;
+    }
+    if (StringUtil::StartsWith(sql, "\\txn")) {
+      auto split = StringUtil::Split(sql, " ");
+      CmdTxn(split, writer);
+      return true;
+    }
     throw Exception(fmt::format("unsupported internal command: {}", sql));
   }
 
@@ -280,6 +318,11 @@ auto BustubInstance::ExecuteSqlTxn(const std::string &sql, ResultWriter &writer,
         HandleExplainStatement(txn, explain_stmt, writer);
         continue;
       }
+      case StatementType::TRANSACTION_STATEMENT: {
+        const auto &txn_stmt = dynamic_cast<const TransactionStatement &>(*statement);
+        HandleTxnStatement(txn, txn_stmt, writer);
+        continue;
+      }
       case StatementType::DELETE_STATEMENT:
       case StatementType::UPDATE_STATEMENT:
         is_delete = true;
@@ -373,4 +416,57 @@ BustubInstance::~BustubInstance() {
   }
 }
 
+/** Enable managed txn mode on this BusTub instance, allowing statements like `BEGIN`. */
+void BustubInstance::EnableManagedTxn() { managed_txn_mode_ = true; }
+
+/** Get the current transaction. */
+auto BustubInstance::CurrentManagedTxn() -> Transaction * { return current_txn_; }
+
+void BustubInstance::CmdTxn(const std::vector<std::string> &params, ResultWriter &writer) {
+  if (!managed_txn_mode_) {
+    writer.OneCell("only supported in managed mode, please use bustub-shell");
+    return;
+  }
+  auto dump_current_txn = [&](const std::string &prefix) {
+    writer.OneCell(fmt::format("{}txn_id={} txn_real_id={} read_ts={} commit_ts={} status={} iso_lvl={}", prefix,
+                               current_txn_->GetTransactionIdHumanReadable(), current_txn_->GetTransactionId(),
+                               current_txn_->GetReadTs(), current_txn_->GetCommitTs(),
+                               current_txn_->GetTransactionState(), current_txn_->GetIsolationLevel()));
+  };
+  if (params.size() == 1) {
+    if (current_txn_ != nullptr) {
+      dump_current_txn("");
+    } else {
+      writer.OneCell("no active txn, each statement starts a new txn.");
+    }
+    return;
+  }
+  if (params.size() == 2) {
+    const std::string &param1 = params[1];
+    if (param1 == "gc") {
+      txn_manager_->GarbageCollection();
+      writer.OneCell("GC complete");
+      return;
+    }
+    auto txn_id = std::stoi(param1);
+    if (txn_id == -1) {
+      dump_current_txn("pause current txn ");
+      current_txn_ = nullptr;
+      return;
+    }
+    auto iter = txn_manager_->txn_map_.find(txn_id);
+    if (iter == txn_manager_->txn_map_.end()) {
+      iter = txn_manager_->txn_map_.find(txn_id + TXN_START_ID);
+      if (iter == txn_manager_->txn_map_.end()) {
+        writer.OneCell("cannot find txn.");
+        return;
+      }
+    }
+    current_txn_ = iter->second.get();
+    dump_current_txn("switch to new txn ");
+    return;
+  }
+  writer.OneCell("unsupported txn cmd.");
+}
+
 }  // namespace bustub

src/include/binder/binder.h

@@ -213,6 +213,8 @@ class Binder {
 
   auto BindVariableShow(duckdb_libpgquery::PGVariableShowStmt *stmt) -> std::unique_ptr<VariableShowStatement>;
 
+  auto BindTransaction(duckdb_libpgquery::PGTransactionStmt *stmt) -> std::unique_ptr<TransactionStatement>;
+
   class ContextGuard {
    public:
     explicit ContextGuard(const BoundTableRef **scope, const CTEList **cte_scope) {

src/include/binder/statement/set_show_statement.h

@@ -44,4 +44,14 @@ class VariableShowStatement : public BoundStatement {
   }
 };
 
+class TransactionStatement : public BoundStatement {
+ public:
+  explicit TransactionStatement(std::string type)
+      : BoundStatement(StatementType::TRANSACTION_STATEMENT), type_(std::move(type)) {}
+
+  std::string type_;
+
+  auto ToString() const -> std::string override { return fmt::format("BoundTransaction {{ type={} }}", type_); }
+};
+
 }  // namespace bustub

src/include/common/bustub_instance.h

@@ -47,6 +47,7 @@ class IndexStatement;
 class VariableSetStatement;
 class VariableShowStatement;
 class ExplainStatement;
+class TransactionStatement;
 
 class ResultWriter {
  public:
@@ -227,7 +228,7 @@ class FortTableWriter : public ResultWriter {
     tables_.emplace_back(table_.to_string());
     table_ = fort::utf8_table{};
   }
-  void OneCell(const std::string &cell) override { tables_.emplace_back(cell); }
+  void OneCell(const std::string &cell) override { tables_.emplace_back(cell + "\n"); }
   fort::utf8_table table_;
   std::vector<std::string> tables_;
 };
@@ -258,6 +259,12 @@ class BustubInstance {
   auto ExecuteSqlTxn(const std::string &sql, ResultWriter &writer, Transaction *txn,
                      std::shared_ptr<CheckOptions> check_options = nullptr) -> bool;
 
+  /** Enable managed txn mode on this BusTub instance, allowing statements like `BEGIN`. */
+  void EnableManagedTxn();
+
+  /** Get the current transaction. */
+  auto CurrentManagedTxn() -> Transaction *;
+
   /**
    * FOR TEST ONLY. Generate test tables in this BusTub instance.
    * It's used in the shell to predefine some tables, as we don't support
@@ -301,17 +308,22 @@ class BustubInstance {
 
  private:
   void CmdDisplayTables(ResultWriter &writer);
+  void CmdDbgMvcc(const std::vector<std::string> &params, ResultWriter &writer);
+  void CmdTxn(const std::vector<std::string> &params, ResultWriter &writer);
   void CmdDisplayIndices(ResultWriter &writer);
   void CmdDisplayHelp(ResultWriter &writer);
   void WriteOneCell(const std::string &cell, ResultWriter &writer);
 
   void HandleCreateStatement(Transaction *txn, const CreateStatement &stmt, ResultWriter &writer);
   void HandleIndexStatement(Transaction *txn, const IndexStatement &stmt, ResultWriter &writer);
   void HandleExplainStatement(Transaction *txn, const ExplainStatement &stmt, ResultWriter &writer);
+  void HandleTxnStatement(Transaction *txn, const TransactionStatement &stmt, ResultWriter &writer);
   void HandleVariableShowStatement(Transaction *txn, const VariableShowStatement &stmt, ResultWriter &writer);
   void HandleVariableSetStatement(Transaction *txn, const VariableSetStatement &stmt, ResultWriter &writer);
 
   std::unordered_map<std::string, std::string> session_variables_;
+  Transaction *current_txn_{nullptr};
+  bool managed_txn_mode_{false};
 };
 
 }  // namespace bustub

src/include/common/enums/statement_type.h

@@ -32,6 +32,7 @@ enum class StatementType : uint8_t {
   INDEX_STATEMENT,          // index statement type
   VARIABLE_SET_STATEMENT,   // set variable statement type
   VARIABLE_SHOW_STATEMENT,  // show variable statement type
+  TRANSACTION_STATEMENT,    // txn statement type
 };
 
 }  // namespace bustub
@@ -75,6 +76,9 @@ struct fmt::formatter<bustub::StatementType> : formatter<string_view> {
       case bustub::StatementType::VARIABLE_SET_STATEMENT:
         name = "VariableSet";
         break;
+      case bustub::StatementType::TRANSACTION_STATEMENT:
+        name = "Transaction";
+        break;
     }
     return formatter<string_view>::format(name, ctx);
   }