feat: Add wal replay and full checkpoint

src/buffer/buffer_pool_manager_instance.cpp

@@ -127,6 +127,9 @@ auto BufferPoolManagerInstance::FlushPgImp(page_id_t page_id) -> bool {
     return false;
   }
   // flush regardless of dirty or not
+  if (enable_logging && log_manager_->GetPersistentLSN() < pages_[frame_id].GetLSN()) {
+    log_manager_->Flush(true);
+  }
   disk_manager_->WritePage(page_id, pages_[frame_id].GetData());
   pages_[frame_id].is_dirty_ = false;  // reset dirty flag regardless
 
@@ -138,6 +141,14 @@ void BufferPoolManagerInstance::FlushAllPgsImp() {
   for (size_t i = 0; i < pool_size_; i++) {
     if (pages_[i].page_id_ != INVALID_PAGE_ID) {
       // there is a valid physical page resides, regardless of dirty or not
+      frame_id_t frame_id;
+      if (!page_table_->Find(pages_[i].page_id_, frame_id)) {
+        // page id is not present in the buffer pool
+        return;
+      }
+      if (enable_logging && log_manager_->GetPersistentLSN() < pages_[frame_id].GetLSN()) {
+        log_manager_->Flush(true);
+      }
       disk_manager_->WritePage(pages_[i].page_id_, pages_[i].GetData());
       pages_[i].is_dirty_ = false;
     }
@@ -191,6 +202,9 @@ auto BufferPoolManagerInstance::FindVictim(frame_id_t *available_frame_id) -> bo
     if (pages_[*available_frame_id].IsDirty()) {
       // the page to be evicted is dirty, flush out
       // don't call FlushPgImp, otherwise lead to deadlock
+      if (enable_logging && log_manager_->GetPersistentLSN() < pages_[*available_frame_id].GetLSN()) {
+        log_manager_->Flush(true);
+      }
       disk_manager_->WritePage(pages_[*available_frame_id].page_id_, pages_[*available_frame_id].GetData());
       pages_[*available_frame_id].is_dirty_ = false;  // reset dirty flag regardless
     }

src/catalog/column.cpp

@@ -38,4 +38,30 @@ auto Column::ToString(bool simplified) const -> std::string {
   return (os.str());
 }
 
+void Column::SerializeTo(char *storage) const {
+  int column_name_size = column_name_.size();
+  memcpy(storage, &column_name_size, sizeof(int));
+  memcpy(storage + sizeof(int), column_name_.data(), column_name_.size());
+  memcpy(storage + sizeof(int) + column_name_.size(), &column_type_, sizeof(TypeId));
+  memcpy(storage + sizeof(int) + column_name_.size() + sizeof(TypeId), &fixed_length_, sizeof(u_int32_t));
+  memcpy(storage + sizeof(int) + column_name_.size() + sizeof(TypeId) + sizeof(u_int32_t), &variable_length_,
+         sizeof(u_int32_t));
+  memcpy(storage + sizeof(int) + column_name_.size() + sizeof(TypeId) + sizeof(u_int32_t) + sizeof(u_int32_t),
+         &column_offset_, sizeof(u_int32_t));
+}
+
+void Column::DeserializeFrom(const char *storage) {
+  int column_name_size = 0;
+  memcpy(&column_name_size, storage, sizeof(int));
+  this->column_name_.resize(column_name_size);
+  memcpy(column_name_.data(), storage + sizeof(int), column_name_size);
+  memcpy(&column_type_, storage + sizeof(int) + column_name_.size(), sizeof(TypeId));
+  memcpy(&fixed_length_, storage + sizeof(int) + column_name_.size() + sizeof(TypeId), sizeof(u_int32_t));
+  memcpy(&variable_length_, storage + sizeof(int) + column_name_.size() + sizeof(TypeId) + sizeof(u_int32_t),
+         sizeof(u_int32_t));
+  memcpy(&column_offset_,
+         storage + sizeof(int) + column_name_.size() + sizeof(TypeId) + sizeof(u_int32_t) + sizeof(u_int32_t),
+         sizeof(u_int32_t));
+}
+
 }  // namespace bustub

src/common/bustub_instance.cpp

@@ -15,6 +15,7 @@
 #include "catalog/schema.h"
 #include "catalog/table_generator.h"
 #include "common/bustub_instance.h"
+
 #include "common/enums/statement_type.h"
 #include "common/exception.h"
 #include "common/util/string_util.h"
@@ -31,6 +32,7 @@
 #include "planner/planner.h"
 #include "recovery/checkpoint_manager.h"
 #include "recovery/log_manager.h"
+#include "recovery/log_recovery.h"
 #include "storage/disk/disk_manager.h"
 #include "storage/disk/disk_manager_memory.h"
 #include "type/value_factory.h"
@@ -71,6 +73,9 @@ BustubInstance::BustubInstance(const std::string &db_file_name) {
 
   // Execution engine.
   execution_engine_ = new ExecutionEngine(buffer_pool_manager_, txn_manager_, catalog_);
+
+  // LogRec
+  log_recovery_ = new LogRecovery(disk_manager_, buffer_pool_manager_, catalog_);
 }
 
 BustubInstance::BustubInstance() {
@@ -219,6 +224,12 @@ auto BustubInstance::ExecuteSqlTxn(const std::string &sql, ResultWriter &writer,
           throw bustub::Exception("Failed to create table");
         }
         WriteOneCell(fmt::format("Table created with id = {}", info->oid_), writer);
+        {
+          LogRecord log_record(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::CREATETABLE,
+                               create_stmt.table_, create_stmt.columns_);
+          lsn_t lsn = log_manager_->AppendLogRecord(&log_record);
+          txn->SetPrevLSN(lsn);
+        }
         continue;
       }
       case StatementType::INDEX_STATEMENT: {
@@ -391,6 +402,9 @@ BustubInstance::~BustubInstance() {
   if (enable_logging) {
     log_manager_->StopFlushThread();
   }
+  if (enable_checkpointing) {
+    checkpoint_manager_->StopFlushThread();
+  }
   delete execution_engine_;
   delete catalog_;
   delete checkpoint_manager_;

src/common/config.cpp

@@ -15,9 +15,10 @@
 namespace bustub {
 
 std::atomic<bool> enable_logging(false);
+std::atomic<bool> enable_checkpointing(false);
 
 std::chrono::duration<int64_t> log_timeout = std::chrono::seconds(1);
-
+std::chrono::duration<int64_t> checkpoint_timeout = std::chrono::seconds(2);
 std::chrono::milliseconds cycle_detection_interval = std::chrono::milliseconds(50);
 
 }  // namespace bustub

src/concurrency/transaction_manager.cpp

@@ -33,8 +33,8 @@ auto TransactionManager::Begin(Transaction *txn, IsolationLevel isolation_level)
   }
 
   if (enable_logging) {
-    LogRecord record = LogRecord(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::BEGIN);
-    lsn_t lsn = log_manager_->AppendLogRecord(&record);
+    LogRecord log_record = LogRecord(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::BEGIN);
+    lsn_t lsn = log_manager_->AppendLogRecord(&log_record);
     txn->SetPrevLSN(lsn);
   }
 
@@ -59,6 +59,14 @@ void TransactionManager::Commit(Transaction *txn) {
   }
   write_set->clear();
 
+  if (enable_logging) {
+    LogRecord log_record(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::COMMIT);
+    auto lsn = log_manager_->AppendLogRecord(&log_record);
+    txn->SetPrevLSN(lsn);
+    // TODO:cwhen committing ,we need to group commit log_manager_->Flush(false);
+    log_manager_->Flush(false);
+  }
+
   // Release all the locks.
   ReleaseLocks(txn);
   // Release the global transaction latch.
@@ -109,6 +117,13 @@ void TransactionManager::Abort(Transaction *txn) {
   table_write_set->clear();
   index_write_set->clear();
 
+  if (enable_logging) {
+    LogRecord log_record(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::ABORT);
+    auto lsn = log_manager_->AppendLogRecord(&log_record);
+    txn->SetPrevLSN(lsn);
+    log_manager_->Flush(false);
+  }
+
   // Release all the locks.
   ReleaseLocks(txn);
   // Release the global transaction latch.

src/execution/seq_scan_executor.cpp

@@ -84,7 +84,7 @@ auto SeqScanExecutor::Next(Tuple *tuple, RID *rid) -> bool {
       obtain_lock_ = false;
       return true;
     }  // not what the filter wants, move to next tuple
-    cursor_++;
+    ++cursor_;
   }
   return false;
 }

src/include/catalog/catalog.h

@@ -351,10 +351,10 @@ class Catalog {
    *
    * NOTE: `tables_` owns all table metadata.
    */
-  std::unordered_map<table_oid_t, std::unique_ptr<TableInfo>> tables_;
+  std::unordered_map<table_oid_t, std::unique_ptr<TableInfo>> tables_{};
 
   /** Map table name -> table identifiers. */
-  std::unordered_map<std::string, table_oid_t> table_names_;
+  std::unordered_map<std::string, table_oid_t> table_names_{};
 
   /** The next table identifier to be used. */
   std::atomic<table_oid_t> next_table_oid_{0};
@@ -364,10 +364,10 @@ class Catalog {
    *
    * NOTE: that `indexes_` owns all index metadata.
    */
-  std::unordered_map<index_oid_t, std::unique_ptr<IndexInfo>> indexes_;
+  std::unordered_map<index_oid_t, std::unique_ptr<IndexInfo>> indexes_{};
 
   /** Map table name -> index names -> index identifiers. */
-  std::unordered_map<std::string, std::unordered_map<std::string, index_oid_t>> index_names_;
+  std::unordered_map<std::string, std::unordered_map<std::string, index_oid_t>> index_names_{};
 
   /** The next index identifier to be used. */
   std::atomic<index_oid_t> next_index_oid_{0};

src/include/catalog/column.h

@@ -30,6 +30,7 @@ class Column {
   friend class Schema;
 
  public:
+  Column() = default;
   /**
    * Non-variable-length constructor for creating a Column.
    * @param column_name name of the column
@@ -66,6 +67,16 @@ class Column {
         fixed_length_(column.fixed_length_),
         variable_length_(column.variable_length_),
         column_offset_(column.column_offset_) {}
+  /**
+   * From wal
+   */
+  Column(std::string column_name, const TypeId type, uint32_t fixed_length, u_int32_t variable_length,
+         uint32_t column_offset)
+      : column_name_(std::move(column_name)),
+        column_type_(type),
+        fixed_length_(fixed_length),
+        variable_length_(variable_length),
+        column_offset_(column_offset) {}
 
   /** @return column name */
   auto GetName() const -> std::string { return column_name_; }
@@ -96,6 +107,9 @@ class Column {
   /** @return a string representation of this column */
   auto ToString(bool simplified = true) const -> std::string;
 
+  void SerializeTo(char *storage) const;
+  void DeserializeFrom(const char *storage);
+
  private:
   /**
    * Return the size in bytes of the type.

src/include/common/bustub_instance.h

@@ -40,6 +40,7 @@ class LogManager;
 class CheckpointManager;
 class Catalog;
 class ExecutionEngine;
+class LogRecovery;
 
 class ResultWriter {
  public:
@@ -253,6 +254,7 @@ class BustubInstance {
   Catalog *catalog_;
   ExecutionEngine *execution_engine_;
   std::shared_mutex catalog_lock_;
+  LogRecovery *log_recovery_;
 
   auto GetSessionVariable(const std::string &key) -> std::string {
     if (session_variables_.find(key) != session_variables_.end()) {

src/include/common/config.h

@@ -27,11 +27,14 @@ extern std::atomic<bool> enable_logging;
 /** If ENABLE_LOGGING is true, the log should be flushed to disk every LOG_TIMEOUT. */
 extern std::chrono::duration<int64_t> log_timeout;
 
+extern std::atomic<bool> enable_checkpointing;
+extern std::chrono::duration<int64_t> checkpoint_timeout;
+
 static constexpr int INVALID_PAGE_ID = -1;                                           // invalid page id
 static constexpr int INVALID_TXN_ID = -1;                                            // invalid transaction id
 static constexpr int INVALID_LSN = -1;                                               // invalid log sequence number
 static constexpr int HEADER_PAGE_ID = 0;                                             // the header page id
-static constexpr int BUSTUB_PAGE_SIZE = 4096;                                        // size of a data page in byte
+static constexpr int BUSTUB_PAGE_SIZE = 1024;                                        // size of a data page in byte
 static constexpr int BUFFER_POOL_SIZE = 10;                                          // size of buffer pool
 static constexpr int LOG_BUFFER_SIZE = ((BUFFER_POOL_SIZE + 1) * BUSTUB_PAGE_SIZE);  // size of a log buffer in byte
 static constexpr int BUCKET_SIZE = 50;                                               // size of extendible hash bucket

src/include/common/util/string_util.h

@@ -14,6 +14,7 @@
 
 #include <string>
 #include <vector>
+#include "common/config.h"
 
 namespace bustub {
 

src/include/recovery/checkpoint_manager.h

@@ -33,11 +33,14 @@ class CheckpointManager {
 
   void BeginCheckpoint();
   void EndCheckpoint();
+  void RunCheckPointThread();
+  void StopFlushThread();
 
  private:
   TransactionManager *transaction_manager_ __attribute__((__unused__));
   LogManager *log_manager_ __attribute__((__unused__));
   BufferPoolManager *buffer_pool_manager_ __attribute__((__unused__));
+  std::thread *checkpoint_thread_;
 };
 
 }  // namespace bustub

src/include/recovery/log_manager.h

@@ -43,6 +43,7 @@ class LogManager {
 
   void RunFlushThread();
   void StopFlushThread();
+  void Flush(bool force);
 
   auto AppendLogRecord(LogRecord *log_record) -> lsn_t;
 
@@ -61,12 +62,16 @@ class LogManager {
 
   char *log_buffer_;
   char *flush_buffer_;
+  int log_buffer_offset_ = 0;
+  int flush_buffer_offset_ = 0;
 
   std::mutex latch_;
 
   std::thread *flush_thread_ __attribute__((__unused__));
 
   std::condition_variable cv_;
+  std::condition_variable cv_append_;
+  std::atomic_bool need_flush_ = false;
 
   DiskManager *disk_manager_ __attribute__((__unused__));
 };

src/include/recovery/log_record.h

@@ -32,6 +32,8 @@ enum class LogRecordType {
   ABORT,
   /** Creating a new page in the table heap. */
   NEWPAGE,
+
+  CREATETABLE
 };
 
 /**
@@ -57,6 +59,10 @@ enum class LogRecordType {
  *--------------------------
  * | HEADER | prev_page_id |
  *--------------------------
+ * For create table
+ *--------------------------
+ * | HEADER | prev_page_id |
+ *--------------------------
  */
 class LogRecord {
   friend class LogManager;
@@ -110,6 +116,15 @@ class LogRecord {
     size_ = HEADER_SIZE + sizeof(page_id_t) * 2;
   }
 
+  // constructor for CREATETABLE type
+  LogRecord(txn_id_t txn_id, lsn_t prev_lsn, LogRecordType log_record_type, const std::string &table_name,
+            std::vector<Column> columns)
+      : txn_id_(txn_id), prev_lsn_(prev_lsn), log_record_type_(log_record_type) {
+    size_ = HEADER_SIZE + table_name.size() + sizeof(Schema);
+    table_name_ = table_name;
+    columns_ = columns;
+  }
+
   ~LogRecord() = default;
 
   inline auto GetDeleteTuple() -> Tuple & { return delete_tuple_; }
@@ -177,6 +192,10 @@ class LogRecord {
   page_id_t prev_page_id_{INVALID_PAGE_ID};
   page_id_t page_id_{INVALID_PAGE_ID};
   static const int HEADER_SIZE = 20;
+
+  // case5: for create table operation
+  std::string table_name_;
+  std::vector<Column> columns_{};
 };  // namespace bustub
 
 }  // namespace bustub

src/include/recovery/log_recovery.h

@@ -20,6 +20,8 @@
 #include "concurrency/lock_manager.h"
 #include "recovery/log_record.h"
 
+#include <catalog/catalog.h>
+
 namespace bustub {
 
 /**
@@ -32,18 +34,24 @@ class LogRecovery {
     log_buffer_ = new char[LOG_BUFFER_SIZE];
   }
 
+  LogRecovery(DiskManager *disk_manager, BufferPoolManager *buffer_pool_manager, Catalog *catalog)
+      : disk_manager_(disk_manager), buffer_pool_manager_(buffer_pool_manager), catalog_(catalog), offset_(0) {
+    log_buffer_ = new char[LOG_BUFFER_SIZE];
+  }
+
   ~LogRecovery() {
     delete[] log_buffer_;
     log_buffer_ = nullptr;
   }
 
   void Redo();
   void Undo();
-  auto DeserializeLogRecord(const char *data, LogRecord *log_record) -> bool;
+  auto DeserializeLogRecord(const char *data, LogRecord *log_record) const -> bool;
 
  private:
   DiskManager *disk_manager_ __attribute__((__unused__));
   BufferPoolManager *buffer_pool_manager_ __attribute__((__unused__));
+  Catalog *catalog_ __attribute__((__unused__));
 
   /** Maintain active transactions and its corresponding latest lsn. */
   std::unordered_map<txn_id_t, lsn_t> active_txn_;