mariadb/extra/mariabackup/fil_cur.cc
Marko Mäkelä 6b6fa3cdb1 MDEV-18644: Support full_crc32 for page_compressed
This is a follow-up task to MDEV-12026, which introduced
innodb_checksum_algorithm=full_crc32 and a simpler page format.
MDEV-12026 did not enable full_crc32 for page_compressed tables,
which we will be doing now.

This is joint work with Thirunarayanan Balathandayuthapani.

For innodb_checksum_algorithm=full_crc32 we change the
page_compressed format as follows:

FIL_PAGE_TYPE: The most significant bit will be set to indicate
page_compressed format. The least significant bits will contain
the compressed page size, rounded up to a multiple of 256 bytes.

The checksum will be stored in the last 4 bytes of the page
(whether it is the full page or a page_compressed page whose
size is determined by FIL_PAGE_TYPE), covering all preceding
bytes of the page. If encryption is used, then the page will
be encrypted between compression and computing the checksum.
For page_compressed, FIL_PAGE_LSN will not be repeated at
the end of the page.

FSP_SPACE_FLAGS (already implemented as part of MDEV-12026):
We will store the innodb_compression_algorithm that may be used
to compress pages. Previously, the choice of algorithm was written
to each compressed data page separately, and one would be unable
to know in advance which compression algorithm(s) are used.

fil_space_t::full_crc32_page_compressed_len(): Determine if the
page_compressed algorithm of the tablespace needs to know the
exact length of the compressed data. If yes, we will reserve and
write an extra byte for this right before the checksum.

buf_page_is_compressed(): Determine if a page uses page_compressed
(in any innodb_checksum_algorithm).

fil_page_decompress(): Pass also fil_space_t::flags so that the
format can be determined.

buf_page_is_zeroes(): Check if a page is full of zero bytes.

buf_page_full_crc32_is_corrupted(): Renamed from
buf_encrypted_full_crc32_page_is_corrupted(). For full_crc32,
we always simply validate the checksum to the page contents,
while the physical page size is explicitly specified by an
unencrypted part of the page header.

buf_page_full_crc32_size(): Determine the size of a full_crc32 page.

buf_dblwr_check_page_lsn(): Make this a debug-only function, because
it involves potentially costly lookups of fil_space_t.

create_table_info_t::check_table_options(),
ha_innobase::check_if_supported_inplace_alter(): Do allow the creation
of SPATIAL INDEX with full_crc32 also when page_compressed is used.

commit_cache_norebuild(): Preserve the compression algorithm when
updating the page_compression_level.

dict_tf_to_fsp_flags(): Set the flags for page compression algorithm.
FIXME: Maybe there should be a table option page_compression_algorithm
and a session variable to back it?
2019-03-18 14:08:43 +02:00

505 lines
14 KiB
C++

/******************************************************
MariaBackup: hot backup tool for InnoDB
(c) 2009-2013 Percona LLC and/or its affiliates.
Originally Created 3/3/2009 Yasufumi Kinoshita
Written by Alexey Kopytov, Aleksandr Kuzminsky, Stewart Smith, Vadim Tkachenko,
Yasufumi Kinoshita, Ignacio Nin and Baron Schwartz.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; version 2 of the License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
*******************************************************/
/* Source file cursor implementation */
#include <my_global.h>
#include <my_base.h>
#include <fil0fil.h>
#include <fsp0fsp.h>
#include <srv0start.h>
#include <trx0sys.h>
#include "fil_cur.h"
#include "fil0crypt.h"
#include "fil0pagecompress.h"
#include "common.h"
#include "read_filt.h"
#include "xtrabackup.h"
/* Size of read buffer in pages (640 pages = 10M for 16K sized pages) */
#define XB_FIL_CUR_PAGES 640
/***********************************************************************
Extracts the relative path ("database/table.ibd") of a tablespace from a
specified possibly absolute path.
For user tablespaces both "./database/table.ibd" and
"/remote/dir/database/table.ibd" result in "database/table.ibd".
For system tablepsaces (i.e. When is_system is TRUE) both "/remote/dir/ibdata1"
and "./ibdata1" yield "ibdata1" in the output. */
const char *
xb_get_relative_path(
/*=================*/
const char* path, /*!< in: tablespace path (either
relative or absolute) */
ibool is_system) /*!< in: TRUE for system tablespaces,
i.e. when only the filename must be
returned. */
{
const char *next;
const char *cur;
const char *prev;
prev = NULL;
cur = path;
while ((next = strchr(cur, OS_PATH_SEPARATOR)) != NULL) {
prev = cur;
cur = next + 1;
}
if (is_system) {
return(cur);
} else {
return((prev == NULL) ? cur : prev);
}
}
/**********************************************************************//**
Closes a file. */
static
void
xb_fil_node_close_file(
/*===================*/
fil_node_t* node) /*!< in: file node */
{
ibool ret;
mutex_enter(&fil_system.mutex);
ut_ad(node);
ut_a(node->n_pending == 0);
ut_a(node->n_pending_flushes == 0);
ut_a(!node->being_extended);
if (!node->is_open()) {
mutex_exit(&fil_system.mutex);
return;
}
ret = os_file_close(node->handle);
ut_a(ret);
node->handle = OS_FILE_CLOSED;
ut_a(fil_system.n_open > 0);
fil_system.n_open--;
if (node->space->purpose == FIL_TYPE_TABLESPACE &&
fil_is_user_tablespace_id(node->space->id)) {
ut_a(UT_LIST_GET_LEN(fil_system.LRU) > 0);
/* The node is in the LRU list, remove it */
UT_LIST_REMOVE(fil_system.LRU, node);
}
mutex_exit(&fil_system.mutex);
}
/************************************************************************
Open a source file cursor and initialize the associated read filter.
@return XB_FIL_CUR_SUCCESS on success, XB_FIL_CUR_SKIP if the source file must
be skipped and XB_FIL_CUR_ERROR on error. */
xb_fil_cur_result_t
xb_fil_cur_open(
/*============*/
xb_fil_cur_t* cursor, /*!< out: source file cursor */
xb_read_filt_t* read_filter, /*!< in/out: the read filter */
fil_node_t* node, /*!< in: source tablespace node */
uint thread_n, /*!< thread number for diagnostics */
ulonglong max_file_size)
{
bool success;
int err;
/* Initialize these first so xb_fil_cur_close() handles them correctly
in case of error */
cursor->orig_buf = NULL;
cursor->node = NULL;
cursor->space_id = node->space->id;
strncpy(cursor->abs_path, node->name, (sizeof cursor->abs_path) - 1);
cursor->abs_path[(sizeof cursor->abs_path) - 1] = '\0';
/* Get the relative path for the destination tablespace name, i.e. the
one that can be appended to the backup root directory. Non-system
tablespaces may have absolute paths for DATA DIRECTORY.
We want to make "local" copies for the backup. */
strncpy(cursor->rel_path,
xb_get_relative_path(cursor->abs_path, cursor->is_system()),
(sizeof cursor->rel_path) - 1);
cursor->rel_path[(sizeof cursor->rel_path) - 1] = '\0';
/* In the backup mode we should already have a tablespace handle created
by fil_ibd_load() unless it is a system
tablespace. Otherwise we open the file here. */
if (!node->is_open()) {
ut_ad(cursor->is_system()
|| srv_operation == SRV_OPERATION_RESTORE_DELTA
|| xb_close_files);
node->handle = os_file_create_simple_no_error_handling(
0, node->name,
OS_FILE_OPEN,
OS_FILE_READ_ALLOW_DELETE, true, &success);
if (!success) {
/* The following call prints an error message */
os_file_get_last_error(TRUE);
msg(thread_n, "mariabackup: error: cannot open "
"tablespace %s", cursor->abs_path);
return(XB_FIL_CUR_SKIP);
}
mutex_enter(&fil_system.mutex);
fil_system.n_open++;
if (node->space->purpose == FIL_TYPE_TABLESPACE &&
fil_is_user_tablespace_id(node->space->id)) {
/* Put the node to the LRU list */
UT_LIST_ADD_FIRST(fil_system.LRU, node);
}
mutex_exit(&fil_system.mutex);
}
ut_ad(node->is_open());
cursor->node = node;
cursor->file = node->handle;
#ifdef _WIN32
HANDLE hDup;
DuplicateHandle(GetCurrentProcess(),cursor->file.m_file,
GetCurrentProcess(), &hDup, 0, FALSE, DUPLICATE_SAME_ACCESS);
int filenr = _open_osfhandle((intptr_t)hDup, 0);
if (filenr < 0) {
err = EINVAL;
}
else {
err = _fstat64(filenr, &cursor->statinfo);
close(filenr);
}
#else
err = fstat(cursor->file.m_file, &cursor->statinfo);
#endif
if (max_file_size < (ulonglong)cursor->statinfo.st_size) {
cursor->statinfo.st_size = (ulonglong)max_file_size;
}
if (err) {
msg(thread_n, "mariabackup: error: cannot fstat %s",
cursor->abs_path);
xb_fil_cur_close(cursor);
return(XB_FIL_CUR_SKIP);
}
if (srv_file_flush_method == SRV_O_DIRECT
|| srv_file_flush_method == SRV_O_DIRECT_NO_FSYNC) {
os_file_set_nocache(cursor->file, node->name, "OPEN");
}
posix_fadvise(cursor->file, 0, 0, POSIX_FADV_SEQUENTIAL);
cursor->page_size = node->space->physical_size();
cursor->zip_size = node->space->zip_size();
/* Allocate read buffer */
cursor->buf_size = XB_FIL_CUR_PAGES * cursor->page_size;
cursor->orig_buf = static_cast<byte *>
(malloc(cursor->buf_size + srv_page_size));
cursor->buf = static_cast<byte *>
(ut_align(cursor->orig_buf, srv_page_size));
cursor->buf_read = 0;
cursor->buf_npages = 0;
cursor->buf_offset = 0;
cursor->buf_page_no = 0;
cursor->thread_n = thread_n;
if (!node->space->crypt_data
&& os_file_read(IORequestRead,
node->handle, cursor->buf, 0,
cursor->page_size)) {
mutex_enter(&fil_system.mutex);
if (!node->space->crypt_data) {
node->space->crypt_data = fil_space_read_crypt_data(
node->space->zip_size(), cursor->buf);
}
mutex_exit(&fil_system.mutex);
}
cursor->space_size = (ulint)(cursor->statinfo.st_size
/ cursor->page_size);
cursor->read_filter = read_filter;
cursor->read_filter->init(&cursor->read_filter_ctxt, cursor,
node->space->id);
return(XB_FIL_CUR_SUCCESS);
}
static bool page_is_corrupted(const byte *page, ulint page_no,
const xb_fil_cur_t *cursor,
const fil_space_t *space)
{
byte tmp_frame[UNIV_PAGE_SIZE_MAX];
byte tmp_page[UNIV_PAGE_SIZE_MAX];
const ulint page_size = cursor->page_size;
ulint page_type = mach_read_from_2(page + FIL_PAGE_TYPE);
/* We ignore the doublewrite buffer pages.*/
if (cursor->space_id == TRX_SYS_SPACE
&& page_no >= FSP_EXTENT_SIZE
&& page_no < FSP_EXTENT_SIZE * 3) {
return false;
}
/* Validate page number. */
if (mach_read_from_4(page + FIL_PAGE_OFFSET) != page_no
&& cursor->space_id != TRX_SYS_SPACE) {
/* On pages that are not all zero, the
page number must match.
There may be a mismatch on tablespace ID,
because files may be renamed during backup.
We disable the page number check
on the system tablespace, because it may consist
of multiple files, and here we count the pages
from the start of each file.)
The first 38 and last 8 bytes are never encrypted. */
const ulint* p = reinterpret_cast<const ulint*>(page);
const ulint* const end = reinterpret_cast<const ulint*>(
page + page_size);
do {
if (*p++) {
return true;
}
} while (p != end);
/* Whole zero page is valid. */
return false;
}
if (space->full_crc32()) {
return buf_page_is_corrupted(true, page, space->flags);
}
/* Validate encrypted pages. The first page is never encrypted.
In the system tablespace, the first page would be written with
FIL_PAGE_FILE_FLUSH_LSN at shutdown, and if the LSN exceeds
4,294,967,295, the mach_read_from_4() below would wrongly
interpret the page as encrypted. We prevent that by checking
page_no first. */
if (page_no
&& mach_read_from_4(page + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION)
&& (opt_encrypted_backup
|| (space->crypt_data
&& space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED))) {
if (!fil_space_verify_crypt_checksum(page, space->zip_size()))
return true;
/* Compressed encrypted need to be decrypted
and decompressed for verification. */
if (page_type != FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED
&& !opt_extended_validation)
return false;
memcpy(tmp_page, page, page_size);
bool decrypted = false;
if (!space->crypt_data
|| space->crypt_data->type == CRYPT_SCHEME_UNENCRYPTED
|| !fil_space_decrypt(space, tmp_frame, tmp_page,
&decrypted)) {
return true;
}
if (page_type != FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED) {
return buf_page_is_corrupted(true, tmp_page,
space->flags);
}
}
if (page_type == FIL_PAGE_PAGE_COMPRESSED) {
memcpy(tmp_page, page, page_size);
}
if (page_type == FIL_PAGE_PAGE_COMPRESSED
|| page_type == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED) {
ulint decomp = fil_page_decompress(tmp_frame, tmp_page,
space->flags);
page_type = mach_read_from_2(tmp_page + FIL_PAGE_TYPE);
return (!decomp
|| (decomp != srv_page_size
&& cursor->zip_size)
|| page_type == FIL_PAGE_PAGE_COMPRESSED
|| page_type == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED
|| buf_page_is_corrupted(true, tmp_page,
space->flags));
}
return buf_page_is_corrupted(true, page, space->flags);
}
/************************************************************************
Reads and verifies the next block of pages from the source
file. Positions the cursor after the last read non-corrupted page.
@return XB_FIL_CUR_SUCCESS if some have been read successfully, XB_FIL_CUR_EOF
if there are no more pages to read and XB_FIL_CUR_ERROR on error. */
xb_fil_cur_result_t
xb_fil_cur_read(
/*============*/
xb_fil_cur_t* cursor) /*!< in/out: source file cursor */
{
byte* page;
ulint i;
ulint npages;
ulint retry_count;
xb_fil_cur_result_t ret;
ib_int64_t offset;
ib_int64_t to_read;
const ulint page_size = cursor->page_size;
xb_ad(!cursor->is_system() || page_size == srv_page_size);
cursor->read_filter->get_next_batch(&cursor->read_filter_ctxt,
&offset, &to_read);
if (to_read == 0LL) {
return(XB_FIL_CUR_EOF);
}
if (to_read > (ib_int64_t) cursor->buf_size) {
to_read = (ib_int64_t) cursor->buf_size;
}
xb_a(to_read > 0 && to_read <= 0xFFFFFFFFLL);
if ((to_read & ~(page_size - 1))
&& offset + to_read == cursor->statinfo.st_size) {
if (to_read < (ib_int64_t) page_size) {
msg(cursor->thread_n, "Warning: junk at the end of "
"%s, offset = %llu, to_read = %llu",cursor->abs_path, (ulonglong) offset, (ulonglong) to_read);
return(XB_FIL_CUR_EOF);
}
to_read = (ib_int64_t) (((ulint) to_read) &
~(page_size - 1));
}
xb_a((to_read & (page_size - 1)) == 0);
npages = (ulint) (to_read / page_size);
retry_count = 10;
ret = XB_FIL_CUR_SUCCESS;
fil_space_t *space = fil_space_acquire_for_io(cursor->space_id);
if (!space) {
return XB_FIL_CUR_ERROR;
}
read_retry:
xtrabackup_io_throttling();
cursor->buf_read = 0;
cursor->buf_npages = 0;
cursor->buf_offset = offset;
cursor->buf_page_no = (ulint)(offset / page_size);
if (!os_file_read(IORequestRead, cursor->file, cursor->buf, offset,
(ulint) to_read)) {
ret = XB_FIL_CUR_ERROR;
goto func_exit;
}
/* check pages for corruption and re-read if necessary. i.e. in case of
partially written pages */
for (page = cursor->buf, i = 0; i < npages;
page += page_size, i++) {
ulint page_no = cursor->buf_page_no + i;
if (page_is_corrupted(page, page_no, cursor, space)){
retry_count--;
if (retry_count == 0) {
msg(cursor->thread_n,
"Error: failed to read page after "
"10 retries. File %s seems to be "
"corrupted.", cursor->abs_path);
ret = XB_FIL_CUR_ERROR;
ut_print_buf(stderr, page, page_size);
break;
}
msg(cursor->thread_n, "Database page corruption detected at page "
ULINTPF ", retrying...",
page_no);
os_thread_sleep(100000);
goto read_retry;
}
cursor->buf_read += page_size;
cursor->buf_npages++;
}
posix_fadvise(cursor->file, offset, to_read, POSIX_FADV_DONTNEED);
func_exit:
space->release_for_io();
return(ret);
}
/************************************************************************
Close the source file cursor opened with xb_fil_cur_open() and its
associated read filter. */
void
xb_fil_cur_close(
/*=============*/
xb_fil_cur_t *cursor) /*!< in/out: source file cursor */
{
if (cursor->read_filter) {
cursor->read_filter->deinit(&cursor->read_filter_ctxt);
}
free(cursor->orig_buf);
if (cursor->node != NULL) {
xb_fil_node_close_file(cursor->node);
cursor->file = OS_FILE_CLOSED;
}
}