src/backend/logging/loggers/wal_frontend_logger.cpp
/**
 * @brief Recovery system based on log file
 */
@brief Recovery system based on log file
@brief Recovery system based on log file
void WriteAheadFrontendLogger::DoRecovery() {
  cid_t start_commit_id = 0;
  if (peloton_checkpoint_mode == CHECKPOINT_TYPE_NORMAL) {
    start_commit_id = this->checkpoint.DoRecovery();
  }
  log_file_cursor_ = 0;
  this->OpenNextLogFile();
  if (log_file_size > 0) {
    bool reached_end_of_file = false;
    __attribute__((unused)) oid_t recovery_log_record_count = 0;
    while (reached_end_of_file == false) {
      auto record_type =
          this->GetNextLogRecordTypeForRecovery(log_file, log_file_size);
      cid_t commit_id = INVALID_CID;
      TupleRecord *tuple_record;
      switch (record_type) {
        case LOGRECORD_TYPE_TRANSACTION_BEGIN:
        case LOGRECORD_TYPE_TRANSACTION_COMMIT: {
          TransactionRecord txn_rec(record_type);
          if (ReadTransactionRecordHeader(txn_rec, log_file, log_file_size) ==
              false) {
            this->log_file_fd = -1;
            return;
          }
          commit_id = txn_rec.GetTransactionId();
          if (commit_id <= start_commit_id) {
            continue;
          }
          break;
        }
        case LOGRECORD_TYPE_WAL_TUPLE_INSERT:
        case LOGRECORD_TYPE_WAL_TUPLE_UPDATE: {
          tuple_record = new TupleRecord(record_type);
          if (ReadTupleRecordHeader(*tuple_record, log_file, log_file_size) ==
              false) {
            LOG_ERROR("Could not read tuple record header.");
            this->log_file_fd = -1;
            return;
          }
          auto cid = tuple_record->GetTransactionId();
          if (recovery_txn_table.find(cid) == recovery_txn_table.end()) {
            LOG_ERROR("Insert txd id %d not found in recovery txn table",
                      (int)cid);
            this->log_file_fd = -1;
            return;
          }
          auto table = GetTable(*tuple_record);
          if (!table || cid <= start_commit_id) {
            SkipTupleRecordBody(log_file, log_file_size);
            delete tuple_record;
            continue;
          }
          tuple_record->SetTuple(ReadTupleRecordBody(
              table->GetSchema(), recovery_pool, log_file, log_file_size));
          break;
        }
        case LOGRECORD_TYPE_WAL_TUPLE_DELETE: {
          tuple_record = new TupleRecord(record_type);
          if (ReadTupleRecordHeader(*tuple_record, log_file, log_file_size) ==
              false) {
            this->log_file_fd = -1;
            return;
          }
          auto cid = tuple_record->GetTransactionId();
          if (cid <= start_commit_id) {
            delete tuple_record;
            continue;
          }
          if (recovery_txn_table.find(cid) == recovery_txn_table.end()) {
            LOG_TRACE("Delete txd id %d not found in recovery txn table",
                      (int)cid);
            this->log_file_fd = -1;
            return;
          }
          break;
        }
        default:
          reached_end_of_file = true;
          break;
      }
      if (!reached_end_of_file) {
        switch (record_type) {
          case LOGRECORD_TYPE_TRANSACTION_BEGIN:
            assert(commit_id != INVALID_CID);
            StartTransactionRecovery(commit_id);
            break;
          case LOGRECORD_TYPE_TRANSACTION_COMMIT:
            assert(commit_id != INVALID_CID);
            CommitTransactionRecovery(commit_id);
            break;
          case LOGRECORD_TYPE_WAL_TUPLE_INSERT:
          case LOGRECORD_TYPE_WAL_TUPLE_DELETE:
          case LOGRECORD_TYPE_WAL_TUPLE_UPDATE:
            recovery_txn_table[tuple_record->GetTransactionId()].push_back(
                tuple_record);
            break;
          default:
            LOG_INFO("Got Type as TXN_INVALID");
            reached_end_of_file = true;
            break;
        }
      }
    }
    AbortActiveTransactions();
    auto &manager = catalog::Manager::GetInstance();
    if (max_oid > manager.GetNextOid()) {
      manager.SetNextOid(max_oid);
    }
    concurrency::TransactionManagerFactory::GetInstance().SetNextCid(max_cid);
  }
  this->log_file_fd = -1;
}
void WriteAheadFrontendLogger :: DoRecovery ( ) { cid_t start_commit_id = 0 ; if ( peloton_checkpoint_mode == CHECKPOINT_TYPE_NORMAL ) { start_commit_id = this -> checkpoint . DoRecovery ( ) ; } log_file_cursor_ = 0 ; this -> OpenNextLogFile ( ) ; if ( log_file_size > 0 ) { bool reached_end_of_file = false ; __attribute__ ( ( unused ) ) oid_t recovery_log_record_count = 0 ; while ( reached_end_of_file == false ) { auto record_type = this -> GetNextLogRecordTypeForRecovery ( log_file , log_file_size ) ; cid_t commit_id = INVALID_CID ; TupleRecord * tuple_record ; switch ( record_type ) { case LOGRECORD_TYPE_TRANSACTION_BEGIN : case LOGRECORD_TYPE_TRANSACTION_COMMIT : { TransactionRecord txn_rec ( record_type ) ; if ( ReadTransactionRecordHeader ( txn_rec , log_file , log_file_size ) == false ) { this -> log_file_fd = -1 ; return ; } commit_id = txn_rec . GetTransactionId ( ) ; if ( commit_id <= start_commit_id ) { continue ; } break ; } case LOGRECORD_TYPE_WAL_TUPLE_INSERT : case LOGRECORD_TYPE_WAL_TUPLE_UPDATE : { tuple_record = new TupleRecord ( record_type ) ; if ( ReadTupleRecordHeader ( * tuple_record , log_file , log_file_size ) == false ) { LOG_ERROR ( " " ) ; this -> log_file_fd = -1 ; return ; } auto cid = tuple_record -> GetTransactionId ( ) ; if ( recovery_txn_table . find ( cid ) == recovery_txn_table . end ( ) ) { LOG_ERROR ( " " , ( int ) cid ) ; this -> log_file_fd = -1 ; return ; } auto table = GetTable ( * tuple_record ) ; if ( ! table || cid <= start_commit_id ) { SkipTupleRecordBody ( log_file , log_file_size ) ; delete tuple_record ; continue ; } tuple_record -> SetTuple ( ReadTupleRecordBody ( table -> GetSchema ( ) , recovery_pool , log_file , log_file_size ) ) ; break ; } case LOGRECORD_TYPE_WAL_TUPLE_DELETE : { tuple_record = new TupleRecord ( record_type ) ; if ( ReadTupleRecordHeader ( * tuple_record , log_file , log_file_size ) == false ) { this -> log_file_fd = -1 ; return ; } auto cid = tuple_record -> GetTransactionId ( ) ; if ( cid <= start_commit_id ) { delete tuple_record ; continue ; } if ( recovery_txn_table . find ( cid ) == recovery_txn_table . end ( ) ) { LOG_TRACE ( " " , ( int ) cid ) ; this -> log_file_fd = -1 ; return ; } break ; } default : reached_end_of_file = true ; break ; } if ( ! reached_end_of_file ) { switch ( record_type ) { case LOGRECORD_TYPE_TRANSACTION_BEGIN : assert ( commit_id != INVALID_CID ) ; StartTransactionRecovery ( commit_id ) ; break ; case LOGRECORD_TYPE_TRANSACTION_COMMIT : assert ( commit_id != INVALID_CID ) ; CommitTransactionRecovery ( commit_id ) ; break ; case LOGRECORD_TYPE_WAL_TUPLE_INSERT : case LOGRECORD_TYPE_WAL_TUPLE_DELETE : case LOGRECORD_TYPE_WAL_TUPLE_UPDATE : recovery_txn_table [ tuple_record -> GetTransactionId ( ) ] . push_back ( tuple_record ) ; break ; default : LOG_INFO ( " " ) ; reached_end_of_file = true ; break ; } } } AbortActiveTransactions ( ) ; auto & manager = catalog :: Manager :: GetInstance ( ) ; if ( max_oid > manager . GetNextOid ( ) ) { manager . SetNextOid ( max_oid ) ; } concurrency :: TransactionManagerFactory :: GetInstance ( ) . SetNextCid ( max_cid ) ; } this -> log_file_fd = -1 ; }
@brief Recovery system based on log file
@brief Recovery system based on log file
// Set log file size // log_file_size = GetLogFileSize(log_file_fd); // Go over the log size if needed // Go over each log record in the log file // Read the first byte to identify log record type // If that is not possible, then wrap up recovery // Check for torn log write // Check for torn log write // Read off the tuple record body from the log // Check for torn log write // Finally, abort ACTIVE transactions in recovery_txn_table // After finishing recovery, set the next oid with maximum oid // observed during the recovery