MySQL redo log 恢复原理 | StoneDB 技术分享会 #5

Type (1) + Space ID (4) + Page Number (4) + Body

version (1): 版本信息flag (1)n (2): 字段的数量inst_cols (2)n_uniq (2): 主键的数量n个字段的长度 (n * 2)offset (2): 前一条记录在页面中的偏移量end_seg_len (compressed): 从mismatch_index开始的记录长度，最低位是标志位info_and_status_bits (1)origin_offset (compressed): record header的长度mismatch_index (compressed): 和前一个记录相比第一个不一样的位置data (end_seg_len >> 1): 该redo log对应的记录从mismatch_index开始的数据

version (1): 版本信息flag (1)n (2): 字段的数量inst_cols (2)n_uniq (2): 主键的数量n个字段的长度 (n * 2)flags (1)val (1): 设置还是取消delete flagpos (compressed): trx_id在记录中的偏移量roll_ptr (7)trx_id (compressed)offset (2): 记录origin offset的位置在页面中的偏移量

version (1): 版本信息flag (1)n (2): 字段的数量inst_cols (2)n_uniq (2): 主键的数量n个字段的长度 (n * 2)flags (1)pos (compressed): trx_id在记录中的偏移量roll_ptr (7)trx_id (compressed)rec_offset (2): 记录在页面中的偏移量info_bits (1)n_fields (compressed): 修改的字段的数量对n_fields个修改字段的描述 field_no (compressed): 字段的编号 len (compressed): 字段的长度 data (len): 数据

// storage/innobase/srv/srv0start.ccsrv_start  // 从系统表的第一个页中获取flushed_lsn  // 如果是正常shutdown的话，会做一次同步的全量checkpoint，会在系统表的第一个页中写入checkpoint的lsn  srv_sys_space.open_or_create(false, create_new_db, &sum_of_new_sizes, &flushed_lsn);    read_lsn_and_check_flags(flush_lsn);      it->validate_first_page(it->m_space_id, flushed_lsn, false);        *flush_lsn = mach_read_from_8(m_first_page + FIL_PAGE_FILE_FLUSH_LSN);
  recv_recovery_from_checkpoint_start(*log_sys, flushed_lsn);    // 每个ib_logfile文件有2KB的header，在header的第2个log block和第4个log block中的8字节偏移量处分别存有checkpoint1和checkpoint2    // 当checkpoint_no为偶数时，写入checkpoint1，为奇数时，写入checkpoint2    // 遍历所有的ib_logfile文件，分别从其header中取出两个checkpoint lsn，取最大值返回    // note: 其实在第一个ib_logfile中寻找checkpoint lsn即可，因为做checkpoint的时候只会往第一个ib_logfile中写入    Log_checkpoint_location checkpoint;    recv_find_max_checkpoint(log, checkpoint)
    // 从checkpoint lsn开始解析redo log并且apply    recv_recovery_begin      recv_read_log_seg      recv_scan_log_recs        recv_parse_log_recs          recv_single_rec            recv_parse_log_rec              mlog_parse_initial_log_record              recv_parse_or_apply_log_rec_body          recv_multi_rec            recv_parse_log_rec              mlog_parse_initial_log_record              recv_parse_or_apply_log_rec_body
  // 将哈希表中的redo log进行重放  recv_apply_hashed_log_recs

// storage/innobase/log/log0recv.ccstatic dberr_t recv_recovery_begin(log_t &log, const lsn_t checkpoint_lsn) {
  // 初始化recv_sys  recv_sys->len = 0;  ...

  // checkpoint_lsn向下向512KB对齐  lsn_t start_lsn = ut_uint64_align_down(checkpoint_lsn, OS_FILE_LOG_BLOCK_SIZE);
  bool finished = false;
  // 循环读取ib_logfile中的内容到  while (!finished) {    // 读取从start_lsn开始的64KB的数据到log.buf中    const lsn_t end_lsn =        recv_read_log_seg(log, log.buf, start_lsn, start_lsn + RECV_SCAN_SIZE);
    if (end_lsn == start_lsn) {      /* This could happen if we crashed just after completing file,      and before next file has been successfully created. */      break;    }
    dberr_t err;
    finished = recv_scan_log_recs(log, max_mem, log.buf, end_lsn - start_lsn,                                  start_lsn, &log.m_scanned_lsn, err);
    if (err != DB_SUCCESS) {      return err;    }
    start_lsn = end_lsn;  }
  return DB_SUCCESS;}

// storage/innobase/log/log0recv.ccstatic lsn_t recv_read_log_seg(log_t &log, byte *buf, lsn_t start_lsn,                               const lsn_t end_lsn) {
  // 找到start_lsn所在的ib_logfile文件  auto file = log.m_files.find(start_lsn);
  if (file == log.m_files.end()) {    /* Missing valid file ! */    return start_lsn;  }
  do {
    os_offset_t source_offset;
    // 计算start_lsn在ib_logfile文件中的偏移量    // LOG_FILE_HDR_SIZE + (lsn - file_start_lsn);    source_offset = file->offset(start_lsn);
    os_offset_t len = end_lsn - start_lsn;
    bool switch_to_next_file = false;
    if (source_offset + len > file->m_size_in_bytes) {      len = file->m_size_in_bytes - source_offset;      switch_to_next_file = true;    }
    // 读取文件    const dberr_t err =        log_data_blocks_read(file_handle, source_offset, len, buf);
    start_lsn += len;    buf += len;
    if (switch_to_next_file) {
      // 切换到下一个文件      ...
    }
  } while (start_lsn != end_lsn);
  return end_lsn;}
// 每个ib_logfile文件的header中记录有该文件起始的file_start_lsnos_offset_t offset(lsn_t lsn)  os_offset_t offset(lsn_t lsn, lsn_t file_start_lsn)    return LOG_FILE_HDR_SIZE + (lsn - file_start_lsn);

struct Log_data_block_header {  ...
  /** Offset up to which this block has data inside, computed from the  beginning of the block. */  // 该log block中前m_data_len个字节是有内容的  uint16_t m_data_len;
  /** Offset to the first mtr starting in this block, or 0 if there is no  mtr starting in this block. */  // 该log block中第一个从该block中开始的mtr的起始位置  uint16_t m_first_rec_group;};
static bool recv_scan_log_recs(log_t &log,                               size_t max_memory, const byte *buf, size_t len,                               lsn_t start_lsn, lsn_t *read_upto_lsn,                               dberr_t &err) {
  const byte *log_block = buf;  lsn_t scanned_lsn = start_lsn;  bool finished = false;  bool more_data = false;
  // 每个log block有header和trail，导致跨block的日志是不连续的，不能直接解析  // 所以需要先将每个block的header和trail去掉，将所有block的主体内容拼起来  do {
    // 解析log block header    Log_data_block_header block_header;    log_data_block_header_deserialize(log_block, block_header);
    ...
    const auto data_len = block_header.m_data_len;
    ...
    // 如果解析redo log的起点位置还没确定并且存在mtr从该block中开始，就确定解析的起点    if (!recv_sys->parse_start_lsn && block_header.m_first_rec_group > 0) {
      recv_sys->parse_start_lsn = scanned_lsn + block_header.m_first_rec_group;
      if (recv_sys->parse_start_lsn < recv_sys->checkpoint_lsn) {        recv_sys->bytes_to_ignore_before_checkpoint =            recv_sys->checkpoint_lsn - recv_sys->parse_start_lsn;      }
      recv_sys->scanned_lsn = recv_sys->parse_start_lsn;      recv_sys->recovered_lsn = recv_sys->parse_start_lsn;    }
    scanned_lsn += data_len;
    if (scanned_lsn > recv_sys->scanned_lsn) {
      // buf空间不够用，扩容      if (recv_sys->len + 4 * OS_FILE_LOG_BLOCK_SIZE >= recv_sys->buf_len) {        recv_sys_resize_buf();      }
      if (!recv_sys->found_corrupt_log) {        // 将该log block去掉header和trail后接到recv_sys->buf的尾部        more_data = recv_sys_add_to_parsing_buf(log_block, scanned_lsn);      }
      recv_sys->scanned_lsn = scanned_lsn;    }
    // 该log block没有满，那么解析redo log的终点就是这个block    if (data_len < OS_FILE_LOG_BLOCK_SIZE) {      /* Log data for this group ends here */      finished = true;
      break;
    } else {      log_block += OS_FILE_LOG_BLOCK_SIZE;    }
  } while (log_block < buf + len);
  if (more_data && !recv_sys->found_corrupt_log) {
    // 解析redo log    recv_parse_log_recs();
    if (recv_sys->recovered_offset > recv_sys->buf_len / 4) {      /* Move parsing buffer data to the buffer start */
      recv_reset_buffer();    }
  }
  return finished;}

static void recv_parse_log_recs() {  ut_ad(recv_sys->parse_start_lsn != 0);
  // 解析redo log以mtr为基本单位  for (;;) {    byte *ptr = recv_sys->buf + recv_sys->recovered_offset;
    byte *end_ptr = recv_sys->buf + recv_sys->len;
    if (ptr == end_ptr) {      return;    }
    bool single_rec;
    switch (*ptr) {      case MLOG_DUMMY_RECORD:        single_rec = true;        break;      default:        // 解析Type最高位的标志位，看该mtr是由单条redo log构成还是多条redo log构成        single_rec = !!(*ptr & MLOG_SINGLE_REC_FLAG);    }
    if (single_rec) {      if (recv_single_rec(ptr, end_ptr)) { // 单条redo log构成的mtr的解析入口        return;      }
    } else if (recv_multi_rec(ptr, end_ptr)) { // 多条redo log构成的mtr的解析入口      return;    }  }}

static bool recv_single_rec(byte *ptr, byte *end_ptr) {
  lsn_t old_lsn = recv_sys->recovered_lsn;
  byte *body;  mlog_id_t type;  page_no_t page_no;  space_id_t space_id;
  // 解析单条redo log  ulint len =      recv_parse_log_rec(&type, ptr, end_ptr, &space_id, &page_no, &body);
  lsn_t new_recovered_lsn;
  new_recovered_lsn = recv_calc_lsn_on_data_add(old_lsn, len);
  if (new_recovered_lsn > recv_sys->scanned_lsn) {    /* The log record filled a log block, and we    require that also the next log block should    have been scanned in */
    return true;  }
  ...
  recv_sys->recovered_offset += len;  recv_sys->recovered_lsn = new_recovered_lsn;
  if (recv_recovery_on) {
    // 将redo log加入到哈希表中    // 不直接重放的原因是可能会有很多redo log作用在同一个page上，将这些redo log使用一次IO进行重放，可以加快重放的速度    // 哈希表包括两层，第一层以space_id为key，第二层以page_no为key    recv_add_to_hash_table(type, space_id, page_no, body, ptr + len,                                 old_lsn, recv_sys->recovered_lsn);
  }
  return false;}

static bool recv_multi_rec(byte *ptr, byte *end_ptr) {
  ulint n_recs = 0;  ulint total_len = 0;
  // 先找到mtr的终点，即MLOG_MULTI_REC_END类型的记录  for (;;) {    mlog_id_t type = MLOG_BIGGEST_TYPE;    byte *body;    page_no_t page_no = 0;    space_id_t space_id = 0;
    ulint len =        recv_parse_log_rec(&type, ptr, end_ptr, &space_id, &page_no, &body);
    // 将部分解析的redo log缓存起来    recv_sys->save_rec(n_recs, space_id, page_no, type, body, len);
    total_len += len;    ++n_recs;
    ptr += len;
    if (type == MLOG_MULTI_REC_END) {      break;    }  }
  lsn_t new_recovered_lsn =      recv_calc_lsn_on_data_add(recv_sys->recovered_lsn, total_len);
  // 重置ptr的位置，开始扫第二遍  ptr = recv_sys->buf + recv_sys->recovered_offset;
  for (ulint i = 0; i < n_recs; i++) {
    lsn_t old_lsn = recv_sys->recovered_lsn;
    space_id_t space_id = 0;    page_no_t page_no = 0;
    mlog_id_t type = MLOG_BIGGEST_TYPE;
    byte *body = nullptr;    size_t len = 0;
    // 从第一遍扫的缓存中取出一条redo log    recv_sys->get_saved_rec(i, space_id, page_no, type, body, len);
    recv_sys->recovered_offset += len;
    recv_sys->recovered_lsn = recv_calc_lsn_on_data_add(old_lsn, len);
    if (recv_recovery_on) {      // 将redo log加入到哈希表中      recv_add_to_hash_table(type, space_id, page_no, body, ptr + len,                                   old_lsn, new_recovered_lsn);    }
    ptr += len;  }
  return false;}

static ulint recv_parse_log_rec(mlog_id_t *type, byte *ptr, byte *end_ptr,                                space_id_t *space_id, page_no_t *page_no,                                byte **body) {
  byte *new_ptr;
  *body = nullptr;
  switch (*ptr) {    case MLOG_MULTI_REC_END:    case MLOG_DUMMY_RECORD:      *page_no = FIL_NULL;      *space_id = SPACE_UNKNOWN;      *type = static_cast<mlog_id_t>(*ptr);      return 1;
    ...  }
  // 解析Type、Space ID、Page Number  new_ptr =      mlog_parse_initial_log_record(ptr, end_ptr, type, space_id, page_no);
  *body = new_ptr;
  if (new_ptr == nullptr) {    return 0;  }
  // 解析body部分  new_ptr = recv_parse_or_apply_log_rec_body(      *type, new_ptr, end_ptr, *space_id, *page_no, nullptr, nullptr,      new_ptr - ptr, recv_sys->recovered_lsn);
  if (new_ptr == nullptr) {    return 0;  }
  return new_ptr - ptr;}

// storage/innobase/mtr/mtr0log.ccbyte *mlog_parse_initial_log_record(    const byte *ptr,     /*!< in: buffer */    const byte *end_ptr, /*!< in: buffer end */    mlog_id_t *type,     /*!< out: log record type: MLOG_1BYTE, ... */    space_id_t *space,   /*!< out: space id */    page_no_t *page_no)  /*!< out: page number */{  if (end_ptr < ptr + 1) {    return (nullptr);  }
  // 解析Type  *type = (mlog_id_t)((ulint)*ptr & ~MLOG_SINGLE_REC_FLAG);  ut_ad(*type <= MLOG_BIGGEST_TYPE);
  ptr++;
  if (end_ptr < ptr + 2) {    return (nullptr);  }
  // 解析Space ID  *space = mach_parse_compressed(&ptr, end_ptr);
  if (ptr != nullptr) {    // 解析Page Number    *page_no = mach_parse_compressed(&ptr, end_ptr);  }
  return (const_cast<byte *>(ptr));}

// storage/innobase/log/log0recv.ccstatic byte *recv_parse_or_apply_log_rec_body(    mlog_id_t type, byte *ptr, byte *end_ptr, space_id_t space_id,    page_no_t page_no, buf_block_t *block, mtr_t *mtr, ulint parsed_bytes,    lsn_t start_lsn) {
  ...
  dict_index_t *index = nullptr;
  ...
  // 这里枚举了所有的redo log类型  switch (type) {
    ...
    case MLOG_REC_INSERT:      // 解析字段数量、主键数量、字段长度等信息，构建出索引字典      if (nullptr != (ptr = mlog_parse_index(ptr, end_ptr, &index))) {
        // 解析剩余的部分，构建出完整的记录，插入到对应的页中        ptr = page_cur_parse_insert_rec(false, ptr, end_ptr, block, index, mtr);      }      break;
    ...  }
  if (index != nullptr) {    dict_table_t *table = index->table;
    dict_mem_index_free(index);    dict_mem_table_free(table);  }
  return ptr;}

byte *mlog_parse_index(byte *ptr, const byte *end_ptr, dict_index_t **index) {
  /* Read the 1 byte for index log version */  uint8_t index_log_version = 0;  ptr = parse_index_log_version(ptr, end_ptr, index_log_version);
  /* Read the 1 byte flag */  uint8_t flag = 0;  ptr = parse_index_flag(ptr, end_ptr, flag);
  /* Read n and n_uniq */  // 解析字段的数量n和主键的数量n_uniq  uint16_t n = 0;  uint16_t n_uniq = 0;  uint16_t inst_cols = 0;  ptr = parse_index_column_counts(ptr, end_ptr, is_comp, is_versioned,                                  is_instant, n, n_uniq, inst_cols);
  /* Create a dummy dict_table_t */  dict_table_t *table =      dict_mem_table_create(RECOVERY_INDEX_TABLE_NAME, DICT_HDR_SPACE, n, 0, 0,                            is_comp ? DICT_TF_COMPACT : 0, 0);
  /* Create a dummy dict_index_t */  dict_index_t *ind =      dict_mem_index_create(RECOVERY_INDEX_TABLE_NAME,                            RECOVERY_INDEX_TABLE_NAME, DICT_HDR_SPACE, 0, n);
  ind->table = table;  ind->n_uniq = (unsigned int)n_uniq;  if (n_uniq != n) {    ind->type = DICT_CLUSTERED;  }
  /* Read each index field info */  // 解析每个字段的长度，填充index的feild信息  ptr = parse_index_fields(ptr, end_ptr, n, n_uniq, is_versioned, ind, table);  if (ptr == nullptr) {    *index = ind;    return ptr;  }
  ...
  *index = ind;  return (ptr);}

static byte *parse_index_fields(byte *ptr, const byte *end_ptr, uint16_t n,                                uint16_t n_uniq, bool is_versioned,                                dict_index_t *&ind, dict_table_t *&table) {
  for (size_t i = 0; i < n; i++) {
    uint16_t len = 0;    // 读取字段的长度信息    ptr = read_2_bytes(ptr, end_ptr, len);
    // 这里构建出来的field字段的类型并不是准确的，只能区分出是变长还是定长，因为redo log中只有字段长度相关的信息，并没有类型相关的信息    dict_mem_table_add_col(        table, nullptr, nullptr,        ((len + 1) & 0x7fff) <= 1 ? DATA_BINARY : DATA_FIXBINARY,        len & 0x8000 ? DATA_NOT_NULL : 0, len & 0x7fff, true, phy_pos, v_added,        v_dropped);
    dict_index_add_col(ind, table, table->get_col(i), 0, true);  }
  // 加上trx_id和roll_ptr的列  dict_table_add_system_columns(table, table->heap);
  /* Identify DB_TRX_ID and DB_ROLL_PTR in the index. */  // index中字段的顺序和物理记录保持一致  // 如果是聚簇索引，trx_id和roll_ptr放在主键的后面  if (is_versioned || (n_uniq != n)) {    size_t i = 0;    i = DATA_TRX_ID - 1 + n_uniq;    ind->fields[i].col = &table->cols[n + DATA_TRX_ID];    ind->fields[i].col->set_phy_pos(table->cols[i].get_phy_pos());
    i = DATA_ROLL_PTR - 1 + n_uniq;    ind->fields[i].col = &table->cols[n + DATA_ROLL_PTR];    ind->fields[i].col->set_phy_pos(table->cols[i].get_phy_pos());  }
  return ptr;}

// storage/innobase/page/page0cur.ccbyte *page_cur_parse_insert_rec(    bool is_short,       /*!< in: true if short inserts */    const byte *ptr,     /*!< in: buffer */    const byte *end_ptr, /*!< in: buffer end */    buf_block_t *block,  /*!< in: page or NULL */    dict_index_t *index, /*!< in: record descriptor */    mtr_t *mtr)          /*!< in: mtr or NULL */{  ulint origin_offset = 0; /* remove warning */  ulint end_seg_len;  ulint mismatch_index = 0; /* remove warning */  page_t *page;  rec_t *cursor_rec{nullptr};  byte buf1[1024];  // buf描述待插入记录  byte *buf;  const byte *ptr2 = ptr;  ulint info_and_status_bits = 0; /* remove warning */  page_cur_t cursor;  mem_heap_t *heap = nullptr;  ulint offsets_[REC_OFFS_NORMAL_SIZE];  // offsets描述每个字段在物理记录中的偏移量  ulint *offsets = offsets_;  // offsets[0]存offsets数组占用的内存大小  rec_offs_init(offsets_);

  page = block ? buf_block_get_frame(block) : nullptr;
  ulint offset;  // 前一条记录在页面中的偏移量  offset = mach_read_from_2(ptr);  ptr += 2;
  if (page != nullptr) cursor_rec = page + offset;
  // 该redo log对应的记录和前一条记录不一样的部分的长度，最低位是一个标志位  end_seg_len = mach_parse_compressed(&ptr, end_ptr);
  info_and_status_bits = mach_read_from_1(ptr);  ptr++;
  // 该redo log对应的record header的长度  origin_offset = mach_parse_compressed(&ptr, end_ptr);
  // 和前一个记录相比第一个不一样的位置  mismatch_index = mach_parse_compressed(&ptr, end_ptr);
  if (!block) {    return (const_cast<byte *>(ptr + (end_seg_len >> 1)));  }
  ...
  // end_seg_len的最低位是一个标志位，所以真实的大小还需要除以2  end_seg_len >>= 1;
  // 如果buf在栈上分配的内存不够，就从堆上分配进行扩容  if (mismatch_index + end_seg_len < sizeof buf1) {    buf = buf1;  } else {    buf = static_cast<byte *>(ut::malloc_withkey(UT_NEW_THIS_FILE_PSI_KEY,                                                 mismatch_index + end_seg_len));  }
  // 待插入记录 = 前一条记录的前mismatch_index个字节 + 从ptr开始的eng_seg_len个字节  if (mismatch_index) {    ut_memcpy(buf, rec_get_start(cursor_rec, offsets), mismatch_index);  }  ut_memcpy(buf + mismatch_index, ptr, end_seg_len);
  ...
  // 将cursor定位到前一条记录的位置  page_cur_position(cursor_rec, block, &cursor);
  // 构建offsets数组，用于描述每个字段在记录中的偏移量  offsets = rec_get_offsets(buf + origin_offset, index, offsets,                            ULINT_UNDEFINED, UT_LOCATION_HERE, &heap);
  // 插入到B+树中  page_cur_rec_insert(&cursor, buf + origin_offset, index,                                         offsets, mtr);
  if (buf != buf1) {    ut::free(buf);  }
  if (UNIV_LIKELY_NULL(heap)) {    mem_heap_free(heap);  }
  return (const_cast<byte *>(ptr + end_seg_len));}