数据库内核月报 － 2023 / 12

struct buf_pool_t {
  /** zip_hash mutex */
  BufListMutex zip_hash_mutex;

  /** 压缩页链表 */
  UT_LIST_BASE_NODE_T(buf_page_t, zip_list) zip_list;

  /** Hash table of buf_block_t blocks whose frames are allocated to the zip
  buddy system, indexed by block->frame */
  hash_table_t *zip_hash;

  /** 解压后的压缩页链表 */
  UT_LIST_BASE_NODE_T(buf_block_t, unzip_LRU) unzip_LRU;

  /** 记录没被修改过的压缩页，debug 模式下才有 */
  UT_LIST_BASE_NODE_T(buf_page_t, list) zip_clean;

  /** 支持压缩表的 free lists, 不同页面大小对应数组的不同值 */
  UT_LIST_BASE_NODE_T(buf_buddy_free_t, list) zip_free[BUF_BUDDY_SIZES_MAX];
}

struct buf_block_t {
    /* page 页的元信息 */
    buf_page_t page;
    /* 真正存储数据的 page */
    byte *frame;
};

/* page 的一些元信息 */
class buf_page_t {
  /* 压缩页 */
  page_zip_des_t zip;
  /* 是否在 buf_pool->zip_hash 链表中 */
  bool in_zip_hash;
};

/** 描述压缩页结构体 */
struct page_zip_des_t {
  /** 压缩页数据 */
  page_zip_t *data;
  /** modification log 起始偏移量 */
  uint16_t m_start;
  /** modification log 终结偏移量 */
  uint16_t m_end;
  /** modification log 是否为空 */
  bool m_nonempty;
  /** 表示压缩页大小 */
  uint8_t ssize;
};

rec_t *page_cur_insert_rec_zip(
    page_cur_t *cursor,  /*!< in/out: page cursor */
    dict_index_t *index, /*!< in: record descriptor */
    const rec_t *rec,    /*!< in: pointer to a physical record */
    ulint *offsets,      /*!< in/out: rec_get_offsets(rec, index) */
    mtr_t *mtr)          /*!< in: mini-transaction handle, or NULL */
{
  
  {
    ...
  }

  /* 1. Get the size of the physical record in the page */
  {
    ...
  }

  /* 2. Try to find suitable space from page memory management */
  {
    ...
  }
  
  /* 3. Create the record */
  insert_rec = rec_copy(insert_buf, rec, offsets);
  rec_offs_make_valid(insert_rec, index, offsets);

  /* 4. Insert the record in the linked list of records */
  ut_ad(cursor->rec != insert_rec);

  {
    /* next record after current before the insertion */
    const rec_t *next_rec = page_rec_get_next_low(cursor->rec, TRUE);
    ...

    page_rec_set_next(insert_rec, next_rec);
    page_rec_set_next(cursor->rec, insert_rec);
  }

  page_header_set_field(page, page_zip, PAGE_N_RECS, 1 + page_get_n_recs(page));

  /* 5. Set the n_owned field in the inserted record to zero,
  and set the heap_no field */
  rec_set_n_owned_new(insert_rec, nullptr, 0);
  rec_set_heap_no_new(insert_rec, heap_no);

  UNIV_MEM_ASSERT_RW(rec_get_start(insert_rec, offsets),
                     rec_offs_size(offsets));

  page_zip_dir_insert(page_zip, cursor->rec, free_rec, insert_rec);

  /* 6. Update the last insertion info in page header */

  last_insert = page_header_get_ptr(page, PAGE_LAST_INSERT);
  ut_ad(!last_insert || rec_get_node_ptr_flag(last_insert) ==
                            rec_get_node_ptr_flag(insert_rec));

  if (!dict_index_is_spatial(index)) {
    if (UNIV_UNLIKELY(last_insert == nullptr)) {
      page_header_set_field(page, page_zip, PAGE_DIRECTION, PAGE_NO_DIRECTION);
      page_header_set_field(page, page_zip, PAGE_N_DIRECTION, 0);

    } else if ((last_insert == cursor->rec) &&
               (page_header_get_field(page, PAGE_DIRECTION) != PAGE_LEFT)) {
      page_header_set_field(page, page_zip, PAGE_DIRECTION, PAGE_RIGHT);
      page_header_set_field(page, page_zip, PAGE_N_DIRECTION,
                            page_header_get_field(page, PAGE_N_DIRECTION) + 1);

    } else if ((page_rec_get_next(insert_rec) == last_insert) &&
               (page_header_get_field(page, PAGE_DIRECTION) != PAGE_RIGHT)) {
      page_header_set_field(page, page_zip, PAGE_DIRECTION, PAGE_LEFT);
      page_header_set_field(page, page_zip, PAGE_N_DIRECTION,
                            page_header_get_field(page, PAGE_N_DIRECTION) + 1);
    } else {
      page_header_set_field(page, page_zip, PAGE_DIRECTION, PAGE_NO_DIRECTION);
      page_header_set_field(page, page_zip, PAGE_N_DIRECTION, 0);
    }
  }

  page_header_set_ptr(page, page_zip, PAGE_LAST_INSERT, insert_rec);

  /* 7. It remains to update the owner record. */
  {
    rec_t *owner_rec = page_rec_find_owner_rec(insert_rec);
    ulint n_owned;

    n_owned = rec_get_n_owned_new(owner_rec);
    rec_set_n_owned_new(owner_rec, page_zip, n_owned + 1);

    /* 8. Now we have incremented the n_owned field of the owner
    record. If the number exceeds PAGE_DIR_SLOT_MAX_N_OWNED,
    we have to split the corresponding directory slot in two. */

    if (UNIV_UNLIKELY(n_owned == PAGE_DIR_SLOT_MAX_N_OWNED)) {
      page_dir_split_slot(page, page_zip, page_dir_find_owner_slot(owner_rec));
    }
  }

  /* 9. write compressed page */
  page_zip_write_rec(page_zip, insert_rec, index, offsets, 1);

  /* 10. Write log record of the insert */
  if (UNIV_LIKELY(mtr != nullptr)) {
    page_cur_insert_rec_write_log(insert_rec, rec_size, cursor->rec, index,
                                  mtr);
  }

  return (insert_rec);
}

压缩页创建主要函数：
  /* 创建一个压缩页 */
  -->page_create_zip
    /* 创建一个页的底层函数，填充文件头等 */
    -->page_create_low
    /* 写入一条 MLOG_ZIP_PAGE_COMPRESS 类型的 redolog */
    --> page_zip_compress

/** Compress a page.
 @return true on success, false on failure; page_zip will be left
 intact on failure. */
ibool page_zip_compress(page_zip_des_t *page_zip, /*!< in: size; out: data,
                                                  n_blobs, m_start, m_end,
                                                  m_nonempty */
                        const page_t *page,       /*!< in: uncompressed page */
                        dict_index_t *index,      /*!< in: index tree */
                        ulint level,              /*!< in: commpression level */
                        mtr_t *mtr)               
{
  ...

  err = deflateInit2(&c_stream, static_cast<int>(level), Z_DEFLATED,
                     UNIV_PAGE_SIZE_SHIFT, MAX_MEM_LEVEL, Z_DEFAULT_STRATEGY);
  ...
  
  if (mtr) {
    page_zip_compress_write_log(page_zip, page, index, mtr);
  }

  ...

  const auto time_diff = std::chrono::duration_cast<std::chrono::microseconds>(
      std::chrono::steady_clock::now() - start_time);
  page_zip_stat[page_zip->ssize - 1].compressed_ok++;
  page_zip_stat[page_zip->ssize - 1].compress_time += time_diff;
  if (cmp_per_index_enabled) {
    mutex_enter(&page_zip_stat_per_index_mutex);
    page_zip_stat_per_index[ind_id].compressed_ok++;
    page_zip_stat_per_index[ind_id].compress_time += time_diff;
    mutex_exit(&page_zip_stat_per_index_mutex);
  }

  return (TRUE);
}

static 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) {

...
...

    case MLOG_ZIP_WRITE_NODE_PTR:

      ut_ad(!page || fil_page_type_is_index(page_type));

      ptr = page_zip_parse_write_node_ptr(ptr, end_ptr, page, page_zip);

      break;

    case MLOG_ZIP_WRITE_BLOB_PTR:

      ut_ad(!page || fil_page_type_is_index(page_type));

      ptr = page_zip_parse_write_blob_ptr(ptr, end_ptr, page, page_zip);

      break;

    case MLOG_ZIP_WRITE_HEADER:

      ut_ad(!page || fil_page_type_is_index(page_type));

      ptr = page_zip_parse_write_header(ptr, end_ptr, page, page_zip);

      break;

    case MLOG_ZIP_PAGE_COMPRESS:

      /* Allow anything in page_type when creating a page. */
      ptr = page_zip_parse_compress(ptr, end_ptr, page, page_zip);
      break;

    case MLOG_ZIP_PAGE_COMPRESS_NO_DATA:

      if (nullptr != (ptr = mlog_parse_index(ptr, end_ptr, true, &index))) {
        ut_a(!page || ((ibool) !!page_is_comp(page) ==
                       dict_table_is_comp(index->table)));

        ptr = page_zip_parse_compress_no_data(ptr, end_ptr, page, page_zip,
                                              index);
      }

      break;
...

}

/* zip 算法默认的压缩级别 */
#define DEFAULT_COMPRESSION_LEVEL 6

/* zip 算法的压缩级别，默认是上面的 6 ，可以取 0-9 */
uint page_zip_level = DEFAULT_COMPRESSION_LEVEL;

/* 值得关注的是，表压缩跟页压缩使用相同的参数控制 zip 算法的压缩级别 */
static MYSQL_SYSVAR_UINT(
    compression_level, page_zip_level, PLUGIN_VAR_RQCMDARG,
    "Compression level used for compressed row format.  0 is no compression"
    ", 1 is fastest, 9 is best compression and default is 6.",
    nullptr, nullptr, DEFAULT_COMPRESSION_LEVEL, 0, 9, 0);

/** Compress a data page
@param[in]	compression	Compression algorithm
@param[in]	block_size	File system block size
@param[in]	src		Source contents to compress
@param[in]	src_len		Length in bytes of the source
@param[out]	dst		Compressed page contents
@param[out]	dst_len		Length in bytes of dst contents
@return buffer data, dst_len will have the length of the data */
byte *os_file_compress_page(Compression compression, ulint block_size,
                            byte *src, ulint src_len, byte *dst,
                            ulint *dst_len) {
  ulint len = 0;
  ulint compression_level = page_zip_level;
  ulint page_type = mach_read_from_2(src + FIL_PAGE_TYPE);

  /* Must compress to <= N-1 FS blocks. */
  ulint out_len = src_len - (FIL_PAGE_DATA + block_size);

  /* This is the original data page size - the page header. */
  ulint content_len = src_len - FIL_PAGE_DATA;

  /* Only compress the data + trailer, leave the header alone */

  switch (compression.m_type) {
    case Compression::NONE:
      ut_error;

    case Compression::ZLIB: {
      uLongf zlen = static_cast<uLongf>(out_len);

      if (compress2(dst + FIL_PAGE_DATA, &zlen, src + FIL_PAGE_DATA,
                    static_cast<uLong>(content_len),
                    static_cast<int>(compression_level)) != Z_OK) {
        *dst_len = src_len;

        return (src);
      }

      len = static_cast<ulint>(zlen);

      break;
    }

    case Compression::LZ4:

      len = LZ4_compress_default(reinterpret_cast<char *>(src) + FIL_PAGE_DATA,
                                 reinterpret_cast<char *>(dst) + FIL_PAGE_DATA,
                                 static_cast<int>(content_len),
                                 static_cast<int>(out_len));

      ut_a(len <= src_len - FIL_PAGE_DATA);

      if (len == 0 || len >= out_len) {
        *dst_len = src_len;

        return (src);
      }

      break;

    case Compression::ZSTD:

      len = ZSTD_compress(dst + FIL_PAGE_DATA, out_len, src + FIL_PAGE_DATA, content_len, page_compress_zstd_level);

      break;
    default:
      *dst_len = src_len;
      return (src);
  }

  ut_a(len <= out_len);

  ut_ad(memcmp(src + FIL_PAGE_LSN + 4,
               src + src_len - FIL_PAGE_END_LSN_OLD_CHKSUM + 4, 4) == 0);

  /* Copy the header as is. */
  memmove(dst, src, FIL_PAGE_DATA);

  /* Add compression control information. Required for decompressing. */
  mach_write_to_2(dst + FIL_PAGE_TYPE, FIL_PAGE_COMPRESSED);

  mach_write_to_1(dst + FIL_PAGE_VERSION, Compression::FIL_PAGE_VERSION_2);

  mach_write_to_1(dst + FIL_PAGE_ALGORITHM_V1, compression.m_type);

  mach_write_to_2(dst + FIL_PAGE_ORIGINAL_TYPE_V1, page_type);

  mach_write_to_2(dst + FIL_PAGE_ORIGINAL_SIZE_V1, content_len);

  mach_write_to_2(dst + FIL_PAGE_COMPRESS_SIZE_V1, len);

  /* Round to the next full block size */

  len += FIL_PAGE_DATA;

  *dst_len = ut_calc_align(len, block_size);

  ut_ad(*dst_len >= len && *dst_len <= out_len + FIL_PAGE_DATA);

  /* Clear out the unused portion of the page. */
  if (len % block_size) {
    memset(dst + len, 0x0, block_size - (len % block_size));
  }v c sa

  return (dst);
}

/** Free storage space associated with a section of the file.
@param[in]	fh		Open file handle
@param[in]	off		Starting offset (SEEK_SET)
@param[in]	len		Size of the hole
@return DB_SUCCESS or error code */
dberr_t os_file_punch_hole(os_file_t fh, os_offset_t off, os_offset_t len) {
  /* In this debugging mode, we act as if punch hole is supported,
  and then skip any calls to actually punch a hole here.
  In this way, Transparent Page Compression is still being tested. */
  DBUG_EXECUTE_IF("ignore_punch_hole", return (DB_SUCCESS););

#ifdef _WIN32
  return (os_file_punch_hole_win32(fh, off, len));
#else
  return (os_file_punch_hole_posix(fh, off, len));
#endif /* _WIN32 */
}