Skip to content

M
MariaDB
Commit 2ca904f0 authored by Marko Mäkelä's avatar Marko Mäkelä
Browse files
Options

MDEV-13103 Deal with page_compressed page corruption 

fil_page_decompress(): Replaces fil_decompress_page().
Allow the caller detect errors. Remove
duplicated code. Use the "safe" instead of "fast" variants of
decompression routines.

fil_page_compress(): Replaces fil_compress_page().
The length of the input buffer always was srv_page_size (innodb_page_size).
Remove printouts, and remove the fil_space_t* parameter.

buf_tmp_buffer_t::reserved: Make private; the accessors acquire()
and release() will use atomic memory access.

buf_pool_reserve_tmp_slot(): Make static. Remove the second parameter.
Do not acquire any mutex. Remove the allocation of the buffers.

buf_tmp_reserve_crypt_buf(), buf_tmp_reserve_compression_buf():
Refactored away from buf_pool_reserve_tmp_slot().

buf_page_decrypt_after_read(): Make static, and simplify the logic.
Use the encryption buffer also for decompressing.

buf_page_io_complete(), buf_dblwr_process(): Check more failures.

fil_space_encrypt(): Simplify the debug checks.

fil_space_t::printed_compression_failure: Remove.

fil_get_compression_alg_name(): Remove.

fil_iterate(): Allocate a buffer for compression and decompression
only once, instead of allocating and freeing it for every page
that uses compression, during IMPORT TABLESPACE. Also, validate the
page checksum before decryption, and reduce the scope of some variables.

fil_page_is_index_page(), fil_page_is_lzo_compressed(): Remove (unused).

AbstractCallback::operator()(): Remove the parameter 'offset'.
The check for it in FetchIndexRootPages::operator() was basically
redundant and dead code since the previous refactoring.
parent 2cdb483b
Hide whitespace changes
Inline Side-by-side
Showing with 509 additions and 863 deletions
mysql-test/suite/innodb/r/innodb-page_compression_default.result
View file @ 2ca904f0
call mtr.add_suppression("InnoDB: Compression failed for space [0-9]+ name test/innodb_page_compressed[0-9] len [0-9]+ err 2 write_size [0-9]+.");
set global innodb_file_format = `Barracuda`;
set global innodb_file_per_table = on;
create table innodb_normal (c1 int not null auto_increment primary key, b char(200)) engine=innodb;
......
mysql-test/suite/innodb/r/innodb-page_compression_snappy.result
View file @ 2ca904f0
call mtr.add_suppression("InnoDB: Compression failed for space [0-9]+ name test/innodb_page_compressed[0-9] len [0-9]+ err 2 write_size [0-9]+.");
set global innodb_compression_algorithm = snappy;
set global innodb_file_format = `Barracuda`;
set global innodb_file_per_table = on;
......
mysql-test/suite/innodb/t/innodb-page_compression_default.test
View file @ 2ca904f0
--source include/have_innodb.inc
--source include/not_embedded.inc
call mtr.add_suppression("InnoDB: Compression failed for space [0-9]+ name test/innodb_page_compressed[0-9] len [0-9]+ err 2 write_size [0-9]+.");
# All page compression test use the same
--source include/innodb-page-compression.inc
......
mysql-test/suite/innodb/t/innodb-page_compression_snappy.test
View file @ 2ca904f0
......@@ -2,8 +2,6 @@
-- source include/have_innodb_snappy.inc
--source include/not_embedded.inc
call mtr.add_suppression("InnoDB: Compression failed for space [0-9]+ name test/innodb_page_compressed[0-9] len [0-9]+ err 2 write_size [0-9]+.");
# snappy
set global innodb_compression_algorithm = snappy;
......
storage/innobase/buf/buf0buf.cc
View file @ 2ca904f0
......@@ -423,6 +423,53 @@ on the io_type */
? (counter##_READ) \
: (counter##_WRITTEN))
/** Reserve a buffer slot for encryption, decryption or page compression.
@param[in,out] buf_pool buffer pool
@return reserved buffer slot */
static buf_tmp_buffer_t* buf_pool_reserve_tmp_slot(buf_pool_t* buf_pool)
{
for (ulint i = 0; i < buf_pool->tmp_arr->n_slots; i++) {
buf_tmp_buffer_t* slot = &buf_pool->tmp_arr->slots[i];
if (slot->acquire()) {
return slot;
}
}
/* We assume that free slot is found */
ut_error;
return NULL;
}
/** Reserve a buffer for encryption, decryption or decompression.
@param[in,out] slot reserved slot */
static void buf_tmp_reserve_crypt_buf(buf_tmp_buffer_t* slot)
{
if (!slot->crypt_buf) {
slot->crypt_buf = static_cast<byte*>(
aligned_malloc(srv_page_size, srv_page_size));
}
}
/** Reserve a buffer for compression.
@param[in,out] slot reserved slot */
static void buf_tmp_reserve_compression_buf(buf_tmp_buffer_t* slot)
{
if (!slot->comp_buf) {
/* Both snappy and lzo compression methods require that
output buffer used for compression is bigger than input
buffer. Increase the allocated buffer size accordingly. */
ulint size = srv_page_size;
#ifdef HAVE_LZO
size += LZO1X_1_15_MEM_COMPRESS;
#elif defined HAVE_SNAPPY
size = snappy_max_compressed_length(size);
#endif
slot->comp_buf = static_cast<byte*>(
aligned_malloc(size, srv_page_size));
}
}
/** Registers a chunk to buf_pool_chunk_map
@param[in] chunk chunk of buffers */
static
......@@ -438,10 +485,91 @@ buf_pool_register_chunk(
@param[in,out] bpage Page control block
@param[in,out] space tablespace
@return whether the operation was successful */
static
bool
buf_page_decrypt_after_read(buf_page_t* bpage, fil_space_t* space)
MY_ATTRIBUTE((nonnull));
static bool buf_page_decrypt_after_read(buf_page_t* bpage, fil_space_t* space)
{
ut_ad(space->n_pending_ios > 0);
ut_ad(space->id == bpage->id.space());
byte* dst_frame = bpage->zip.data ? bpage->zip.data :
((buf_block_t*) bpage)->frame;
bool page_compressed = fil_page_is_compressed(dst_frame);
buf_pool_t* buf_pool = buf_pool_from_bpage(bpage);
if (bpage->id.page_no() == 0) {
/* File header pages are not encrypted/compressed */
return (true);
}
/* Page is encrypted if encryption information is found from
tablespace and page contains used key_version. This is true
also for pages first compressed and then encrypted. */
buf_tmp_buffer_t* slot;
if (page_compressed) {
/* the page we read is unencrypted */
/* Find free slot from temporary memory array */
decompress:
slot = buf_pool_reserve_tmp_slot(buf_pool);
/* For decompression, use crypt_buf. */
buf_tmp_reserve_crypt_buf(slot);
decompress_with_slot:
ut_d(fil_page_type_validate(dst_frame));
bpage->write_size = fil_page_decompress(slot->crypt_buf,
dst_frame);
slot->release();
ut_ad(!bpage->write_size || fil_page_type_validate(dst_frame));
ut_ad(space->n_pending_ios > 0);
return bpage->write_size != 0;
}
if (space->crypt_data
&& mach_read_from_4(FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION
+ dst_frame)) {
/* Verify encryption checksum before we even try to
decrypt. */
if (!fil_space_verify_crypt_checksum(
dst_frame, bpage->size, bpage->id.space(),
bpage->id.page_no())) {
decrypt_failed:
/* Mark page encrypted in case it should be. */
if (space->crypt_data->type
!= CRYPT_SCHEME_UNENCRYPTED) {
bpage->encrypted = true;
}
return false;
}
/* Find free slot from temporary memory array */
slot = buf_pool_reserve_tmp_slot(buf_pool);
buf_tmp_reserve_crypt_buf(slot);
ut_d(fil_page_type_validate(dst_frame));
/* decrypt using crypt_buf to dst_frame */
if (!fil_space_decrypt(space, slot->crypt_buf,
dst_frame, &bpage->encrypted)) {
slot->release();
goto decrypt_failed;
}
ut_d(fil_page_type_validate(dst_frame));
if (fil_page_is_compressed_encrypted(dst_frame)) {
goto decompress_with_slot;
}
slot->release();
} else if (fil_page_is_compressed_encrypted(dst_frame)) {
goto decompress;
}
ut_ad(space->n_pending_ios > 0);
return true;
}
/* prototypes for new functions added to ha_innodb.cc */
trx_t* innobase_get_trx();
......@@ -5931,21 +6059,23 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
ulint read_page_no = 0;
ulint read_space_id = 0;
uint key_version = 0;
ut_ad(bpage->zip.data != NULL || ((buf_block_t*)bpage)->frame != NULL);
byte* frame = bpage->zip.data
? bpage->zip.data
: reinterpret_cast<buf_block_t*>(bpage)->frame;
ut_ad(frame);
fil_space_t* space = fil_space_acquire_for_io(
bpage->id.space());
if (!space) {
return DB_TABLESPACE_DELETED;
}
buf_page_decrypt_after_read(bpage, space);
byte* frame = bpage->zip.data
? bpage->zip.data
: reinterpret_cast<buf_block_t*>(bpage)->frame;
dberr_t err;
if (!buf_page_decrypt_after_read(bpage, space)) {
err = DB_DECRYPTION_FAILED;
goto database_corrupted;
}
if (bpage->zip.data && uncompressed) {
my_atomic_addlint(&buf_pool->n_pend_unzip, 1);
ibool ok = buf_zip_decompress((buf_block_t*) bpage,
......@@ -6097,7 +6227,7 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
/* io_type == BUF_IO_WRITE */
if (bpage->slot) {
/* Mark slot free */
bpage->slot->reserved = false;
bpage->slot->release();
bpage->slot = NULL;
}
}
......@@ -7268,66 +7398,6 @@ operator<<(
return(out);
}
/********************************************************************//**
Reserve unused slot from temporary memory array and allocate necessary
temporary memory if not yet allocated.
@return reserved slot */
UNIV_INTERN
buf_tmp_buffer_t*
buf_pool_reserve_tmp_slot(
/*======================*/
buf_pool_t* buf_pool, /*!< in: buffer pool where to
reserve */
bool compressed) /*!< in: is file space compressed */
{
buf_tmp_buffer_t *free_slot=NULL;
/* Array is protected by buf_pool mutex */
buf_pool_mutex_enter(buf_pool);
for(ulint i = 0; i < buf_pool->tmp_arr->n_slots; i++) {
buf_tmp_buffer_t *slot = &buf_pool->tmp_arr->slots[i];
if(slot->reserved == false) {
free_slot = slot;
break;
}
}
/* We assume that free slot is found */
ut_a(free_slot != NULL);
free_slot->reserved = true;
/* Now that we have reserved this slot we can release
buf_pool mutex */
buf_pool_mutex_exit(buf_pool);
/* Allocate temporary memory for encryption/decryption */
if (free_slot->crypt_buf == NULL) {
free_slot->crypt_buf = static_cast<byte*>(aligned_malloc(UNIV_PAGE_SIZE, UNIV_PAGE_SIZE));
memset(free_slot->crypt_buf, 0, UNIV_PAGE_SIZE);
}
/* For page compressed tables allocate temporary memory for
compression/decompression */
if (compressed && free_slot->comp_buf == NULL) {
ulint size = UNIV_PAGE_SIZE;
/* Both snappy and lzo compression methods require that
output buffer used for compression is bigger than input
buffer. Increase the allocated buffer size accordingly. */
#if HAVE_SNAPPY
size = snappy_max_compressed_length(size);
#endif
#if HAVE_LZO
size += LZO1X_1_15_MEM_COMPRESS;
#endif
free_slot->comp_buf = static_cast<byte*>(aligned_malloc(size, UNIV_PAGE_SIZE));
memset(free_slot->comp_buf, 0, size);
}
return (free_slot);
}
/** Encryption and page_compression hook that is called just before
a page is written to disk.
@param[in,out] space tablespace
......@@ -7376,15 +7446,18 @@ buf_page_encrypt_before_write(
return src_frame;
}
ut_ad(!bpage->size.is_compressed() || !page_compressed);
buf_pool_t* buf_pool = buf_pool_from_bpage(bpage);
/* Find free slot from temporary memory array */
buf_tmp_buffer_t* slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
buf_tmp_buffer_t* slot = buf_pool_reserve_tmp_slot(buf_pool);
slot->out_buf = NULL;
bpage->slot = slot;
buf_tmp_reserve_crypt_buf(slot);
byte *dst_frame = slot->crypt_buf;
if (!page_compressed) {
not_compressed:
/* Encrypt page content */
byte* tmp = fil_space_encrypt(space,
bpage->id.page_no(),
......@@ -7392,33 +7465,30 @@ buf_page_encrypt_before_write(
src_frame,
dst_frame);
bpage->real_size = UNIV_PAGE_SIZE;
slot->out_buf = dst_frame = tmp;
ut_d(fil_page_type_validate(tmp));
} else {
/* First we compress the page content */
ulint out_len = 0;
byte *tmp = fil_compress_page(
space,
(byte *)src_frame,
slot->comp_buf,
srv_page_size,
buf_tmp_reserve_compression_buf(slot);
byte* tmp = slot->comp_buf;
ulint out_len = fil_page_compress(
src_frame, tmp,
fsp_flags_get_page_compression_level(space->flags),
fil_space_get_block_size(space, bpage->id.page_no()),
encrypted,
&out_len);
encrypted);
if (!out_len) {
goto not_compressed;
}
bpage->real_size = out_len;
/* Workaround for MDEV-15527. */
memset(tmp + out_len, 0 , srv_page_size - out_len);
#ifdef UNIV_DEBUG
fil_page_type_validate(tmp);
#endif
if(encrypted) {
ut_d(fil_page_type_validate(tmp));
if (encrypted) {
/* And then we encrypt the page content */
tmp = fil_space_encrypt(space,
bpage->id.page_no(),
......@@ -7436,112 +7506,6 @@ buf_page_encrypt_before_write(
return dst_frame;
}
/** Decrypt a page.
@param[in,out] bpage Page control block
@param[in,out] space tablespace
@return whether the operation was successful */
static
bool
buf_page_decrypt_after_read(buf_page_t* bpage, fil_space_t* space)
{
ut_ad(space->n_pending_ios > 0);
ut_ad(space->id == bpage->id.space());
bool compressed = bpage->size.is_compressed();
const page_size_t& size = bpage->size;
byte* dst_frame = compressed ? bpage->zip.data :
((buf_block_t*) bpage)->frame;
unsigned key_version =
mach_read_from_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
bool page_compressed = fil_page_is_compressed(dst_frame);
bool page_compressed_encrypted = fil_page_is_compressed_encrypted(dst_frame);
buf_pool_t* buf_pool = buf_pool_from_bpage(bpage);
bool success = true;
if (bpage->id.page_no() == 0) {
/* File header pages are not encrypted/compressed */
return (true);
}
/* Page is encrypted if encryption information is found from
tablespace and page contains used key_version. This is true
also for pages first compressed and then encrypted. */
if (!space->crypt_data) {
key_version = 0;
}
if (page_compressed) {
/* the page we read is unencrypted */
/* Find free slot from temporary memory array */
buf_tmp_buffer_t* slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
ut_d(fil_page_type_validate(dst_frame));
/* decompress using comp_buf to dst_frame */
fil_decompress_page(slot->comp_buf,
dst_frame,
ulong(size.logical()),
&bpage->write_size);
/* Mark this slot as free */
slot->reserved = false;
key_version = 0;
ut_d(fil_page_type_validate(dst_frame));
} else {
buf_tmp_buffer_t* slot = NULL;
if (key_version) {
/* Verify encryption checksum before we even try to
decrypt. */
if (!fil_space_verify_crypt_checksum(
dst_frame, size,
bpage->id.space(), bpage->id.page_no())) {
if (space->crypt_data->type
!= CRYPT_SCHEME_UNENCRYPTED) {
bpage->encrypted = true;
}
return (false);
}
/* Find free slot from temporary memory array */
slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
ut_d(fil_page_type_validate(dst_frame));
/* decrypt using crypt_buf to dst_frame */
if (!fil_space_decrypt(space, slot->crypt_buf,
dst_frame, &bpage->encrypted)) {
success = false;
}
ut_d(fil_page_type_validate(dst_frame));
}
if (page_compressed_encrypted && success) {
if (!slot) {
slot = buf_pool_reserve_tmp_slot(buf_pool, page_compressed);
}
ut_d(fil_page_type_validate(dst_frame));
/* decompress using comp_buf to dst_frame */
fil_decompress_page(slot->comp_buf,
dst_frame,
ulong(size.logical()),
&bpage->write_size);
ut_d(fil_page_type_validate(dst_frame));
}
/* Mark this slot as free */
if (slot) {
slot->reserved = false;
}
}
ut_ad(space->n_pending_ios > 0);
return (success);
}
/**
Should we punch hole to deallocate unused portion of the page.
@param[in] bpage Page control block
......@@ -7565,6 +7529,4 @@ buf_page_get_trim_length(
{
return (bpage->size.physical() - write_length);
}
#endif /* !UNIV_INNOCHECKSUM */
storage/innobase/buf/buf0dblwr.cc
View file @ 2ca904f0
......@@ -533,10 +533,11 @@ buf_dblwr_process()
}
unaligned_read_buf = static_cast<byte*>(
ut_malloc_nokey(2 * UNIV_PAGE_SIZE));
ut_malloc_nokey(3 * UNIV_PAGE_SIZE));
read_buf = static_cast<byte*>(
ut_align(unaligned_read_buf, UNIV_PAGE_SIZE));
byte* const buf = read_buf + UNIV_PAGE_SIZE;
for (recv_dblwr_t::list::iterator i = recv_dblwr.pages.begin();
i != recv_dblwr.pages.end();
......@@ -604,24 +605,24 @@ buf_dblwr_process()
ignore this page (there should be redo log
records to initialize it). */
} else {
if (fil_page_is_compressed_encrypted(read_buf) ||
fil_page_is_compressed(read_buf)) {
/* Decompress the page before
validating the checksum. */
fil_decompress_page(
NULL, read_buf, srv_page_size,
NULL, true);
/* Decompress the page before
validating the checksum. */
ulint decomp = fil_page_decompress(buf, read_buf);
if (!decomp || (decomp != srv_page_size
&& page_size.is_compressed())) {
goto bad;
}
if (fil_space_verify_crypt_checksum(
read_buf, page_size, space_id, page_no)
|| !buf_page_is_corrupted(
true, read_buf, page_size, space)) {
|| !buf_page_is_corrupted(
true, read_buf, page_size, space)) {
/* The page is good; there is no need
to consult the doublewrite buffer. */
continue;
}
bad:
/* We intentionally skip this message for
is_all_zero pages. */
ib::info()
......@@ -629,19 +630,16 @@ buf_dblwr_process()
<< " from the doublewrite buffer.";
}
/* Next, validate the doublewrite page. */
if (fil_page_is_compressed_encrypted(page) ||
fil_page_is_compressed(page)) {
/* Decompress the page before
validating the checksum. */
fil_decompress_page(
NULL, page, srv_page_size, NULL, true);
ulint decomp = fil_page_decompress(buf, page);
if (!decomp || (decomp != srv_page_size
&& page_size.is_compressed())) {
goto bad_doublewrite;
}
if (!fil_space_verify_crypt_checksum(page, page_size,
space_id, page_no)
&& buf_page_is_corrupted(true, page, page_size, space)) {
if (!is_all_zero) {
bad_doublewrite:
ib::warn() << "A doublewrite copy of page "
<< page_id << " is corrupted.";
}
......
storage/innobase/fil/fil0crypt.cc
View file @ 2ca904f0
......@@ -665,57 +665,38 @@ fil_space_encrypt(
#ifdef UNIV_DEBUG
if (tmp) {
/* Verify that encrypted buffer is not corrupted */
byte* tmp_mem = (byte *)malloc(UNIV_PAGE_SIZE);
dberr_t err = DB_SUCCESS;
byte* src = src_frame;
bool page_compressed_encrypted = (mach_read_from_2(tmp+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED);
byte* comp_mem = NULL;
byte* uncomp_mem = NULL;
byte uncomp_mem[UNIV_PAGE_SIZE_MAX];
byte tmp_mem[UNIV_PAGE_SIZE_MAX];
if (page_compressed_encrypted) {
comp_mem = (byte *)malloc(UNIV_PAGE_SIZE);
uncomp_mem = (byte *)malloc(UNIV_PAGE_SIZE);
memcpy(comp_mem, src_frame, UNIV_PAGE_SIZE);
fil_decompress_page(uncomp_mem, comp_mem,
srv_page_size, NULL);
src = uncomp_mem;
memcpy(uncomp_mem, src, srv_page_size);
ulint unzipped1 = fil_page_decompress(
tmp_mem, uncomp_mem);
ut_ad(unzipped1);
if (unzipped1 != srv_page_size) {
src = uncomp_mem;
}
}
bool corrupted1 = buf_page_is_corrupted(true, src, page_size, space);
bool ok = fil_space_decrypt(crypt_data, tmp_mem, page_size, tmp, &err);
ut_ad(!buf_page_is_corrupted(true, src, page_size, space));
ut_ad(fil_space_decrypt(crypt_data, tmp_mem, page_size, tmp,
&err));
ut_ad(err == DB_SUCCESS);
/* Need to decompress the page if it was also compressed */
if (page_compressed_encrypted) {
memcpy(comp_mem, tmp_mem, UNIV_PAGE_SIZE);
fil_decompress_page(tmp_mem, comp_mem,
srv_page_size, NULL);
}
bool corrupted = buf_page_is_corrupted(true, tmp_mem, page_size, space);
memcpy(tmp_mem+FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, src+FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 8);
bool different = memcmp(src, tmp_mem, page_size.physical());
if (!ok || corrupted || corrupted1 || err != DB_SUCCESS || different) {
fprintf(stderr, "ok %d corrupted %d corrupted1 %d err %d different %d\n",
ok , corrupted, corrupted1, err, different);
fprintf(stderr, "src_frame\n");
buf_page_print(src_frame, page_size);
fprintf(stderr, "encrypted_frame\n");
buf_page_print(tmp, page_size);
fprintf(stderr, "decrypted_frame\n");
buf_page_print(tmp_mem, page_size);
ut_ad(0);
byte buf[UNIV_PAGE_SIZE_MAX];
memcpy(buf, tmp_mem, srv_page_size);
ulint unzipped2 = fil_page_decompress(tmp_mem, buf);
ut_ad(unzipped2);
}
free(tmp_mem);
if (comp_mem) {
free(comp_mem);
}
if (uncomp_mem) {
free(uncomp_mem);
}
memcpy(tmp_mem + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION,
src + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 8);
ut_ad(!memcmp(src, tmp_mem, page_size.physical()));
}
#endif /* UNIV_DEBUG */
......
storage/innobase/fil/fil0pagecompress.cc
View file @ 2ca904f0
/*****************************************************************************
Copyright (C) 2013, 2017, MariaDB Corporation.
Copyright (C) 2013, 2018, MariaDB Corporation.
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
......@@ -74,73 +74,26 @@ Updated 14/02/2015
#include "snappy-c.h"
#endif
/* Used for debugging */
//#define UNIV_PAGECOMPRESS_DEBUG 1
/****************************************************************//**
For page compressed pages compress the page before actual write
operation.
@return compressed page to be written*/
UNIV_INTERN
byte*
fil_compress_page(
/*==============*/
fil_space_t* space, /*!< in,out: tablespace (NULL during IMPORT) */
byte* buf, /*!< in: buffer from which to write; in aio
this must be appropriately aligned */
byte* out_buf, /*!< out: compressed buffer */
ulint len, /*!< in: length of input buffer.*/
ulint level, /* in: compression level */
ulint block_size, /*!< in: block size */
bool encrypted, /*!< in: is page also encrypted */
ulint* out_len) /*!< out: actual length of compressed
page */
/** Compress a page_compressed page before writing to a data file.
@param[in] buf page to be compressed
@param[out] out_buf compressed page
@param[in] level compression level
@param[in] block_size file system block size
@param[in] encrypted whether the page will be subsequently encrypted
@return actual length of compressed page
@retval 0 if the page was not compressed */
ulint fil_page_compress(const byte* buf, byte* out_buf, ulint level,
ulint block_size, bool encrypted)
{
int err = Z_OK;
int comp_level = int(level);
ulint header_len = FIL_PAGE_DATA + FIL_PAGE_COMPRESSED_SIZE;
ulint write_size = 0;
#if HAVE_LZO
lzo_uint write_size_lzo = write_size;
#endif
/* Cache to avoid change during function execution */
ulint comp_method = innodb_compression_algorithm;
bool allocated = false;
/* page_compression does not apply to tables or tablespaces
that use ROW_FORMAT=COMPRESSED */
ut_ad(!space || !FSP_FLAGS_GET_ZIP_SSIZE(space->flags));
if (encrypted) {
header_len += FIL_PAGE_COMPRESSION_METHOD_SIZE;
}
if (!out_buf) {
allocated = true;
ulint size = UNIV_PAGE_SIZE;
/* Both snappy and lzo compression methods require that
output buffer used for compression is bigger than input
buffer. Increase the allocated buffer size accordingly. */
#if HAVE_SNAPPY
if (comp_method == PAGE_SNAPPY_ALGORITHM) {
size = snappy_max_compressed_length(size);
}
#endif
#if HAVE_LZO
if (comp_method == PAGE_LZO_ALGORITHM) {
size += LZO1X_1_15_MEM_COMPRESS;
}
#endif
out_buf = static_cast<byte *>(ut_malloc_nokey(size));
}
ut_ad(buf);
ut_ad(out_buf);
ut_ad(len);
ut_ad(out_len);
/* Let's not compress file space header or
extent descriptor */
switch (fil_page_get_type(buf)) {
......@@ -148,8 +101,7 @@ fil_compress_page(
case FIL_PAGE_TYPE_FSP_HDR:
case FIL_PAGE_TYPE_XDES:
case FIL_PAGE_PAGE_COMPRESSED:
*out_len = len;
goto err_exit;
return 0;
}
/* If no compression level was provided to this table, use system
......@@ -158,125 +110,113 @@ fil_compress_page(
comp_level = page_zip_level;
}
DBUG_LOG("compress", "Preparing for space "
<< (space ? space->id : 0) << " '"
<< (space ? space->name : "(import)") << "' len " << len);
write_size = UNIV_PAGE_SIZE - header_len;
ulint write_size = srv_page_size - header_len;
switch(comp_method) {
switch (comp_method) {
default:
ut_ad(!"unknown compression method");
/* fall through */
case PAGE_UNCOMPRESSED:
return 0;
case PAGE_ZLIB_ALGORITHM:
{
ulong len = uLong(write_size);
if (Z_OK == compress2(
out_buf + header_len, &len,
buf, uLong(srv_page_size), comp_level)) {
write_size = len;
goto success;
}
}
break;
#ifdef HAVE_LZ4
case PAGE_LZ4_ALGORITHM:
#ifdef HAVE_LZ4_COMPRESS_DEFAULT
err = LZ4_compress_default((const char *)buf,
(char *)out_buf+header_len, len, write_size);
#else
err = LZ4_compress_limitedOutput((const char *)buf,
(char *)out_buf+header_len, len, write_size);
#endif /* HAVE_LZ4_COMPRESS_DEFAULT */
write_size = err;
if (err == 0) {
goto err_exit;
# ifdef HAVE_LZ4_COMPRESS_DEFAULT
write_size = LZ4_compress_default(
reinterpret_cast<const char*>(buf),
reinterpret_cast<char*>(out_buf) + header_len,
int(srv_page_size), int(write_size));
# else
write_size = LZ4_compress_limitedOutput(
reinterpret_cast<const char*>(buf),
reinterpret_cast<char*>(out_buf) + header_len,
int(srv_page_size), int(write_size));
# endif
if (write_size) {
goto success;
}
break;
#endif /* HAVE_LZ4 */
#ifdef HAVE_LZO
case PAGE_LZO_ALGORITHM:
err = lzo1x_1_15_compress(
buf, len, out_buf+header_len, &write_size_lzo, out_buf+UNIV_PAGE_SIZE);
write_size = write_size_lzo;
if (err != LZO_E_OK || write_size > UNIV_PAGE_SIZE-header_len) {
goto err_exit;
case PAGE_LZO_ALGORITHM: {
lzo_uint len = write_size;
if (LZO_E_OK == lzo1x_1_15_compress(
buf, srv_page_size,
out_buf + header_len, &len,
out_buf + srv_page_size)
&& len <= write_size) {
write_size = len;
goto success;
}
break;
}
#endif /* HAVE_LZO */
#ifdef HAVE_LZMA
case PAGE_LZMA_ALGORITHM: {
size_t out_pos=0;
err = lzma_easy_buffer_encode(
comp_level,
LZMA_CHECK_NONE,
NULL, /* No custom allocator, use malloc/free */
reinterpret_cast<uint8_t*>(buf),
len,
reinterpret_cast<uint8_t*>(out_buf + header_len),
&out_pos,
(size_t)write_size);
if (err != LZMA_OK || out_pos > UNIV_PAGE_SIZE-header_len) {
size_t out_pos = 0;
if (LZMA_OK == lzma_easy_buffer_encode(
comp_level, LZMA_CHECK_NONE, NULL,
buf, srv_page_size, out_buf + header_len,
&out_pos, write_size)
&& out_pos <= write_size) {
write_size = out_pos;
goto err_exit;
goto success;
}
write_size = out_pos;
break;
}
#endif /* HAVE_LZMA */
#ifdef HAVE_BZIP2
case PAGE_BZIP2_ALGORITHM: {
err = BZ2_bzBuffToBuffCompress(
(char *)(out_buf + header_len),
(unsigned int *)&write_size,
(char *)buf,
len,
1,
0,
0);
if (err != BZ_OK || write_size > UNIV_PAGE_SIZE-header_len) {
goto err_exit;
unsigned len = unsigned(write_size);
if (BZ_OK == BZ2_bzBuffToBuffCompress(
reinterpret_cast<char*>(out_buf + header_len),
&len,
const_cast<char*>(
reinterpret_cast<const char*>(buf)),
unsigned(srv_page_size), 1, 0, 0)
&& len <= write_size) {
write_size = len;
goto success;
}
break;
}
#endif /* HAVE_BZIP2 */
#ifdef HAVE_SNAPPY
case PAGE_SNAPPY_ALGORITHM:
{
snappy_status cstatus;
write_size = snappy_max_compressed_length(UNIV_PAGE_SIZE);
cstatus = snappy_compress(
(const char *)buf,
(size_t)len,
(char *)(out_buf+header_len),
(size_t*)&write_size);
if (cstatus != SNAPPY_OK || write_size > UNIV_PAGE_SIZE-header_len) {
err = (int)cstatus;
goto err_exit;
case PAGE_SNAPPY_ALGORITHM: {
size_t len = snappy_max_compressed_length(srv_page_size);
if (SNAPPY_OK == snappy_compress(
reinterpret_cast<const char*>(buf),
srv_page_size,
reinterpret_cast<char*>(out_buf) + header_len,
&len)
&& len <= write_size) {
write_size = len;
goto success;
}
break;
}
#endif /* HAVE_SNAPPY */
case PAGE_ZLIB_ALGORITHM:
err = compress2(out_buf+header_len, (ulong*)&write_size, buf,
uLong(len), comp_level);
if (err != Z_OK) {
goto err_exit;
}
break;
case PAGE_UNCOMPRESSED:
*out_len = len;
return (buf);
break;
default:
ut_error;
break;
}
srv_stats.pages_page_compression_error.inc();
return 0;
success:
/* Set up the page header */
memcpy(out_buf, buf, FIL_PAGE_DATA);
/* Set up the checksum */
......@@ -307,23 +247,12 @@ fil_compress_page(
/* Verify that page can be decompressed */
{
byte *comp_page;
byte *uncomp_page;
comp_page = static_cast<byte *>(ut_malloc_nokey(UNIV_PAGE_SIZE));
uncomp_page = static_cast<byte *>(ut_malloc_nokey(UNIV_PAGE_SIZE));
memcpy(comp_page, out_buf, UNIV_PAGE_SIZE);
fil_decompress_page(uncomp_page, comp_page, ulong(len), NULL);
if (buf_page_is_corrupted(false, uncomp_page, univ_page_size,
space)) {
buf_page_print(uncomp_page, univ_page_size);
ut_ad(0);
}
ut_free(comp_page);
ut_free(uncomp_page);
page_t tmp_buf[UNIV_PAGE_SIZE_MAX];
page_t page[UNIV_PAGE_SIZE_MAX];
memcpy(page, out_buf, srv_page_size);
ut_ad(fil_page_decompress(tmp_buf, page));
ut_ad(!buf_page_is_corrupted(false, page, univ_page_size,
NULL));
}
#endif /* UNIV_DEBUG */
......@@ -347,317 +276,143 @@ fil_compress_page(
#endif
}
DBUG_LOG("compress", "Succeeded for space "
<< (space ? space->id : 0) << " '"
<< (space ? space->name : "(import)")
<< "' len " << len << " out_len " << write_size);
srv_stats.page_compression_saved.add((len - write_size));
srv_stats.page_compression_saved.add(srv_page_size - write_size);
srv_stats.pages_page_compressed.inc();
/* If we do not persistently trim rest of page, we need to write it
all */
if (!srv_use_trim) {
memset(out_buf+write_size,0,len-write_size);
write_size = len;
}
*out_len = write_size;
if (allocated) {
/* TODO: reduce number of memcpy's */
memcpy(buf, out_buf, len);
goto exit_free;
} else {
return(out_buf);
memset(out_buf + write_size, 0, srv_page_size - write_size);
}
err_exit:
/* If error we leave the actual page as it was */
#ifndef UNIV_PAGECOMPRESS_DEBUG
if (space && !space->printed_compression_failure) {
space->printed_compression_failure = true;
#endif
ib::warn() << "Compression failed for space: "
<< space->id << " name: "
<< space->name << " len: "
<< len << " err: " << err << " write_size: "
<< write_size
<< " compression method: "
<< fil_get_compression_alg_name(comp_method)
<< ".";
#ifndef UNIV_PAGECOMPRESS_DEBUG
}
#endif
srv_stats.pages_page_compression_error.inc();
*out_len = len;
exit_free:
if (allocated) {
ut_free(out_buf);
}
return (buf);
return write_size;
}
/****************************************************************//**
For page compressed pages decompress the page after actual read
operation. */
UNIV_INTERN
void
fil_decompress_page(
/*================*/
byte* page_buf, /*!< in: preallocated buffer or NULL */
byte* buf, /*!< out: buffer from which to read; in aio
this must be appropriately aligned */
ulong len, /*!< in: length of output buffer.*/
ulint* write_size, /*!< in/out: Actual payload size of
the compressed data. */
bool return_error) /*!< in: true if only an error should
be produced when decompression fails.
By default this parameter is false. */
/** Decompress a page that may be subject to page_compressed compression.
@param[in,out] tmp_buf temporary buffer (of innodb_page_size)
@param[in,out] buf possibly compressed page buffer
@return size of the compressed data
@retval 0 if decompression failed
@retval srv_page_size if the page was not compressed */
ulint fil_page_decompress(byte* tmp_buf, byte* buf)
{
int err = 0;
ulint actual_size = 0;
ib_uint64_t compression_alg = 0;
byte *in_buf;
ulint ptype;
const unsigned ptype = mach_read_from_2(buf+FIL_PAGE_TYPE);
ulint header_len;
ut_ad(buf);
ut_ad(len);
ptype = mach_read_from_2(buf+FIL_PAGE_TYPE);
uint64_t compression_alg;
switch (ptype) {
case FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED:
header_len = FIL_PAGE_DATA + FIL_PAGE_COMPRESSED_SIZE
+ FIL_PAGE_COMPRESSION_METHOD_SIZE;
compression_alg = mach_read_from_2(
FIL_PAGE_DATA + FIL_PAGE_COMPRESSED_SIZE + buf);
break;
case FIL_PAGE_PAGE_COMPRESSED:
header_len = FIL_PAGE_DATA + FIL_PAGE_COMPRESSED_SIZE;
compression_alg = mach_read_from_8(
FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION + buf);
break;
default:
/* The page is not in our format. */
return;
return srv_page_size;
}
// If no buffer was given, we need to allocate temporal buffer
if (page_buf == NULL) {
in_buf = static_cast<byte *>(ut_malloc_nokey(UNIV_PAGE_SIZE));
memset(in_buf, 0, UNIV_PAGE_SIZE);
} else {
in_buf = page_buf;
if (mach_read_from_4(buf + FIL_PAGE_SPACE_OR_CHKSUM)
!= BUF_NO_CHECKSUM_MAGIC) {
return 0;
}
/* Before actual decompress, make sure that page type is correct */
ulint actual_size = mach_read_from_2(buf + FIL_PAGE_DATA);
if (mach_read_from_4(buf+FIL_PAGE_SPACE_OR_CHKSUM)
!= BUF_NO_CHECKSUM_MAGIC
|| (ptype != FIL_PAGE_PAGE_COMPRESSED
&& ptype != FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED)) {
ib::error() << "Corruption: We try to uncompress corrupted "
"page CRC "
<< mach_read_from_4(buf+FIL_PAGE_SPACE_OR_CHKSUM)
<< " type " << ptype << " len " << len << ".";
if (return_error) {
goto error_return;
}
ut_error;
}
/* Get compression algorithm */
if (ptype == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED) {
compression_alg = static_cast<ib_uint64_t>(mach_read_from_2(buf+FIL_PAGE_DATA+FIL_PAGE_COMPRESSED_SIZE));
} else {
compression_alg = mach_read_from_8(buf+FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
}
/* Get the actual size of compressed page */
actual_size = mach_read_from_2(buf+FIL_PAGE_DATA);
/* Check if payload size is corrupted */
if (actual_size == 0 || actual_size > UNIV_PAGE_SIZE) {
ib::error() << "Corruption: We try to uncompress corrupted page"
<< " actual size: " << actual_size
<< " compression method: "
<< fil_get_compression_alg_name(compression_alg)
<< ".";
if (return_error) {
goto error_return;
}
ut_error;
}
/* Store actual payload size of the compressed data. This pointer
points to buffer pool. */
if (write_size) {
*write_size = actual_size;
if (actual_size == 0 || actual_size > srv_page_size - header_len) {
return 0;
}
DBUG_LOG("compress", "Preparing for decompress for len "
<< actual_size << ".");
switch(compression_alg) {
switch (compression_alg) {
default:
ib::error() << "Unknown compression algorithm "
<< compression_alg;
return 0;
case PAGE_ZLIB_ALGORITHM:
err= uncompress(in_buf, &len, buf+header_len, (unsigned long)actual_size);
/* If uncompress fails it means that page is corrupted */
if (err != Z_OK) {
goto err_exit;
if (return_error) {
goto error_return;
{
uLong len = srv_page_size;
if (Z_OK != uncompress(tmp_buf, &len,
buf + header_len,
uLong(actual_size))
&& len != srv_page_size) {
return 0;
}
}
break;
#ifdef HAVE_LZ4
case PAGE_LZ4_ALGORITHM:
err = LZ4_decompress_fast((const char *)buf+header_len, (char *)in_buf, len);
if (err != (int)actual_size) {
goto err_exit;
if (return_error) {
goto error_return;
}
if (LZ4_decompress_safe(reinterpret_cast<const char*>(buf)
+ header_len,
reinterpret_cast<char*>(tmp_buf),
actual_size, srv_page_size)
== int(srv_page_size)) {
break;
}
break;
return 0;
#endif /* HAVE_LZ4 */
#ifdef HAVE_LZO
case PAGE_LZO_ALGORITHM: {
ulint olen = 0;
lzo_uint olen_lzo = olen;
err = lzo1x_decompress((const unsigned char *)buf+header_len,
actual_size,(unsigned char *)in_buf, &olen_lzo, NULL);
olen = olen_lzo;
if (err != LZO_E_OK || (olen == 0 || olen > UNIV_PAGE_SIZE)) {
len = olen;
goto err_exit;
if (return_error) {
goto error_return;
}
lzo_uint len_lzo = srv_page_size;
if (LZO_E_OK == lzo1x_decompress_safe(
buf + header_len,
actual_size, tmp_buf, &len_lzo, NULL)
&& len_lzo == srv_page_size) {
break;
}
break;
return 0;
}
#endif /* HAVE_LZO */
#ifdef HAVE_LZMA
case PAGE_LZMA_ALGORITHM: {
lzma_ret ret;
size_t src_pos = 0;
size_t dst_pos = 0;
uint64_t memlimit = UINT64_MAX;
ret = lzma_stream_buffer_decode(
&memlimit,
0,
NULL,
buf+header_len,
&src_pos,
actual_size,
in_buf,
&dst_pos,
len);
if (ret != LZMA_OK || (dst_pos == 0 || dst_pos > UNIV_PAGE_SIZE)) {
len = dst_pos;
goto err_exit;
if (return_error) {
goto error_return;
}
if (LZMA_OK == lzma_stream_buffer_decode(
&memlimit, 0, NULL, buf + header_len,
&src_pos, actual_size, tmp_buf, &dst_pos,
srv_page_size)
&& dst_pos == srv_page_size) {
break;
}
break;
return 0;
}
#endif /* HAVE_LZMA */
#ifdef HAVE_BZIP2
case PAGE_BZIP2_ALGORITHM: {
unsigned int dst_pos = UNIV_PAGE_SIZE;
err = BZ2_bzBuffToBuffDecompress(
(char *)in_buf,
&dst_pos,
(char *)(buf+header_len),
actual_size,
1,
0);
if (err != BZ_OK || (dst_pos == 0 || dst_pos > UNIV_PAGE_SIZE)) {
len = dst_pos;
goto err_exit;
if (return_error) {
goto error_return;
}
unsigned int dst_pos = srv_page_size;
if (BZ_OK == BZ2_bzBuffToBuffDecompress(
reinterpret_cast<char*>(tmp_buf),
&dst_pos,
reinterpret_cast<char*>(buf) + header_len,
actual_size, 1, 0)
&& dst_pos == srv_page_size) {
break;
}
break;
return 0;
}
#endif /* HAVE_BZIP2 */
#ifdef HAVE_SNAPPY
case PAGE_SNAPPY_ALGORITHM:
{
snappy_status cstatus;
ulint olen = UNIV_PAGE_SIZE;
cstatus = snappy_uncompress(
(const char *)(buf+header_len),
(size_t)actual_size,
(char *)in_buf,
(size_t*)&olen);
if (cstatus != SNAPPY_OK || (olen == 0 || olen > UNIV_PAGE_SIZE)) {
err = (int)cstatus;
len = olen;
goto err_exit;
if (return_error) {
goto error_return;
}
case PAGE_SNAPPY_ALGORITHM: {
size_t olen = srv_page_size;
if (SNAPPY_OK == snappy_uncompress(
reinterpret_cast<const char*>(buf) + header_len,
actual_size,
reinterpret_cast<char*>(tmp_buf), &olen)
&& olen == srv_page_size) {
break;
}
break;
return 0;
}
#endif /* HAVE_SNAPPY */
default:
goto err_exit;
if (return_error) {
goto error_return;
}
break;
}
srv_stats.pages_page_decompressed.inc();
/* Copy the uncompressed page to the buffer pool, not
really any other options. */
memcpy(buf, in_buf, len);
error_return:
if (page_buf != in_buf) {
ut_free(in_buf);
}
return;
err_exit:
/* Note that as we have found the page is corrupted, so
all this could be incorrect. */
ulint space_id = mach_read_from_4(buf+FIL_PAGE_SPACE_ID);
fil_space_t* space = fil_space_acquire_for_io(space_id);
ib::error() << "Corruption: Page is marked as compressed"
<< " space: " << space_id << " name: "
<< (space ? space->name : "NULL")
<< " but uncompress failed with error: " << err
<< " size: " << actual_size
<< " len: " << len
<< " compression method: "
<< fil_get_compression_alg_name(compression_alg) << ".";
buf_page_print(buf, univ_page_size);
fil_space_release_for_io(space);
ut_ad(0);
memcpy(buf, tmp_buf, srv_page_size);
return actual_size;
}
storage/innobase/ibuf/ibuf0ibuf.cc
View file @ 2ca904f0
......@@ -5040,6 +5040,11 @@ ibuf_check_bitmap_on_import(
continue;
}
if (!bitmap_page) {
mutex_exit(&ibuf_mutex);
return DB_CORRUPTION;
}
for (i = FSP_IBUF_BITMAP_OFFSET + 1;
i < page_size.physical();
i++) {
......
storage/innobase/include/buf0buf.h
View file @ 2ca904f0
......@@ -41,6 +41,7 @@ Created 11/5/1995 Heikki Tuuri
#include "os0proc.h"
#include "log0log.h"
#include "srv0srv.h"
#include "my_atomic.h"
#include <ostream>
// Forward declaration
......@@ -1516,8 +1517,10 @@ NOTE! The definition appears here only for other modules of this
directory (buf) to see it. Do not use from outside! */
typedef struct {
bool reserved; /*!< true if this slot is reserved
private:
int32 reserved; /*!< true if this slot is reserved
*/
public:
byte* crypt_buf; /*!< for encryption the data needs to be
copied to a separate buffer before it's
encrypted&written. this as a page can be
......@@ -1528,6 +1531,21 @@ typedef struct {
byte* out_buf; /*!< resulting buffer after
encryption/compression. This is a
pointer and not allocated. */
/** Release the slot */
void release()
{
my_atomic_store32_explicit(&reserved, false,
MY_MEMORY_ORDER_RELAXED);
}
/** Acquire the slot
@return whether the slot was acquired */
bool acquire()
{
return !my_atomic_fas32_explicit(&reserved, true,
MY_MEMORY_ORDER_RELAXED);
}
} buf_tmp_buffer_t;
/** The common buffer control block structure
......
storage/innobase/include/fil0fil.h
View file @ 2ca904f0
......@@ -175,9 +175,6 @@ struct fil_space_t {
/** MariaDB encryption data */
fil_space_crypt_t* crypt_data;
/** True if we have already printed compression failure */
bool printed_compression_failure;
/** True if the device this filespace is on supports atomic writes */
bool atomic_write_supported;
......
storage/innobase/include/fil0pagecompress.h
View file @ 2ca904f0
/*****************************************************************************
Copyright (C) 2013, 2017 MariaDB Corporation. All Rights Reserved.
Copyright (C) 2013, 2018 MariaDB Corporation.
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
......@@ -30,40 +30,24 @@ atomic writes information to table space.
Created 11/12/2013 Jan Lindström jan.lindstrom@skysql.com
***********************************************************************/
/****************************************************************//**
For page compressed pages compress the page before actual write
operation.
@return compressed page to be written*/
UNIV_INTERN
byte*
fil_compress_page(
/*==============*/
fil_space_t* space, /*!< in,out: tablespace (NULL during IMPORT) */
byte* buf, /*!< in: buffer from which to write; in aio
this must be appropriately aligned */
byte* out_buf, /*!< out: compressed buffer */
ulint len, /*!< in: length of input buffer.*/
ulint level, /* in: compression level */
ulint block_size, /*!< in: block size */
bool encrypted, /*!< in: is page also encrypted */
ulint* out_len); /*!< out: actual length of compressed
page */
/****************************************************************//**
For page compressed pages decompress the page after actual read
operation. */
UNIV_INTERN
void
fil_decompress_page(
/*================*/
byte* page_buf, /*!< in: preallocated buffer or NULL */
byte* buf, /*!< out: buffer from which to read; in aio
this must be appropriately aligned */
ulong len, /*!< in: length of output buffer.*/
ulint* write_size, /*!< in/out: Actual payload size of
the compressed data. */
bool return_error=false);
/*!< in: true if only an error should
be produced when decompression fails.
By default this parameter is false. */
/** Compress a page_compressed page before writing to a data file.
@param[in] buf page to be compressed
@param[out] out_buf compressed page
@param[in] level compression level
@param[in] block_size file system block size
@param[in] encrypted whether the page will be subsequently encrypted
@return actual length of compressed page
@retval 0 if the page was not compressed */
ulint fil_page_compress(const byte* buf, byte* out_buf, ulint level,
ulint block_size, bool encrypted)
MY_ATTRIBUTE((nonnull, warn_unused_result));
/** Decompress a page that may be subject to page_compressed compression.
@param[in,out] tmp_buf temporary buffer (of innodb_page_size)
@param[in,out] buf compressed page buffer
@return size of the compressed data
@retval 0 if decompression failed
@retval srv_page_size if the page was not compressed */
ulint fil_page_decompress(byte* tmp_buf, byte* buf)
MY_ATTRIBUTE((nonnull, warn_unused_result));
#endif
storage/innobase/include/fsp0pagecompress.ic
View file @ 2ca904f0
/*****************************************************************************
Copyright (C) 2013, 2017, MariaDB Corporation. All Rights Reserved.
Copyright (C) 2013, 2018, MariaDB Corporation.
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
......@@ -38,18 +38,6 @@ fsp_flags_get_page_compression_level(
}
/*******************************************************************//**
Find out wheather the page is index page or not
@return true if page type index page, false if not */
UNIV_INLINE
bool
fil_page_is_index_page(
/*===================*/
byte* buf) /*!< in: page */
{
return(mach_read_from_2(buf+FIL_PAGE_TYPE) == FIL_PAGE_INDEX);
}
/*******************************************************************//**
Find out wheather the page is page compressed
@return true if page is page compressed, false if not */
......@@ -73,59 +61,3 @@ fil_page_is_compressed_encrypted(
{
return(mach_read_from_2(buf+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED);
}
/****************************************************************//**
Get the name of the compression algorithm used for page
compression.
@return compression algorithm name or "UNKNOWN" if not known*/
UNIV_INLINE
const char*
fil_get_compression_alg_name(
/*=========================*/
ib_uint64_t comp_alg) /*!<in: compression algorithm number */
{
switch(comp_alg) {
case PAGE_UNCOMPRESSED:
return ("uncompressed");
break;
case PAGE_ZLIB_ALGORITHM:
return ("ZLIB");
break;
case PAGE_LZ4_ALGORITHM:
return ("LZ4");
break;
case PAGE_LZO_ALGORITHM:
return ("LZO");
break;
case PAGE_LZMA_ALGORITHM:
return ("LZMA");
break;
case PAGE_BZIP2_ALGORITHM:
return ("BZIP2");
break;
case PAGE_SNAPPY_ALGORITHM:
return ("SNAPPY");
break;
/* No default to get compiler warning */
}
return ("NULL");
}
#ifndef UNIV_INNOCHECKSUM
/*******************************************************************//**
Find out wheather the page is page compressed with lzo method
@return true if page is page compressed with lzo method, false if not */
UNIV_INLINE
bool
fil_page_is_lzo_compressed(
/*=======================*/
byte* buf) /*!< in: page */
{
return((mach_read_from_2(buf+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED &&
mach_read_from_8(buf+FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION) == PAGE_LZO_ALGORITHM) ||
(mach_read_from_2(buf+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED &&
mach_read_from_2(buf+FIL_PAGE_DATA+FIL_PAGE_COMPRESSED_SIZE) == PAGE_LZO_ALGORITHM));
}
#endif /* UNIV_INNOCHECKSUM */
storage/innobase/row/row0import.cc
View file @ 2ca904f0
......@@ -40,6 +40,12 @@ Created 2012-02-08 by Sunny Bains.
#include "row0quiesce.h"
#include "fil0pagecompress.h"
#include "ut0new.h"
#ifdef HAVE_LZO
#include "lzo/lzo1x.h"
#endif
#ifdef HAVE_SNAPPY
#include "snappy-c.h"
#endif
#include <vector>
......@@ -406,12 +412,9 @@ class AbstractCallback
updated then its state must be set to BUF_PAGE_NOT_USED. For
compressed tables the page descriptor memory will be at offset:
block->frame + UNIV_PAGE_SIZE;
@param offset - physical offset within the file
@param block - block read from file, note it is not from the buffer pool
@param block block read from file, note it is not from the buffer pool
@retval DB_SUCCESS or error code. */
virtual dberr_t operator()(
os_offset_t offset,
buf_block_t* block) UNIV_NOTHROW = 0;
virtual dberr_t operator()(buf_block_t* block) UNIV_NOTHROW = 0;
/**
@return the space id of the tablespace */
......@@ -635,12 +638,9 @@ struct FetchIndexRootPages : public AbstractCallback {
}
/** Called for each block as it is read from the file.
@param offset physical offset in the file
@param block block to convert, it is not from the buffer pool.
@retval DB_SUCCESS or error code. */
virtual dberr_t operator() (
os_offset_t offset,
buf_block_t* block) UNIV_NOTHROW;
dberr_t operator()(buf_block_t* block) UNIV_NOTHROW;
/** Update the import configuration that will be used to import
the tablespace. */
......@@ -657,13 +657,9 @@ struct FetchIndexRootPages : public AbstractCallback {
determine the exact row format. We can't get that from the tablespace
header flags alone.
@param offset physical offset in the file
@param block block to convert, it is not from the buffer pool.
@retval DB_SUCCESS or error code. */
dberr_t
FetchIndexRootPages::operator() (
os_offset_t offset,
buf_block_t* block) UNIV_NOTHROW
dberr_t FetchIndexRootPages::operator()(buf_block_t* block) UNIV_NOTHROW
{
if (is_interrupted()) return DB_INTERRUPTED;
......@@ -671,15 +667,7 @@ FetchIndexRootPages::operator() (
ulint page_type = fil_page_get_type(page);
if (block->page.id.page_no() * m_page_size.physical() != offset) {
ib::error() << "Page offset doesn't match file offset:"
" page offset: " << block->page.id.page_no()
<< ", file offset: "
<< (offset / m_page_size.physical());
return DB_CORRUPTION;
} else if (page_type == FIL_PAGE_TYPE_XDES) {
if (page_type == FIL_PAGE_TYPE_XDES) {
return set_current_xdes(block->page.id.page_no(), page);
} else if (fil_page_index_page_check(page)
&& !is_free(block->page.id.page_no())
......@@ -825,12 +813,9 @@ class PageConverter : public AbstractCallback {
}
/** Called for each block as it is read from the file.
@param offset physical offset in the file
@param block block to convert, it is not from the buffer pool.
@retval DB_SUCCESS or error code. */
virtual dberr_t operator() (
os_offset_t offset,
buf_block_t* block) UNIV_NOTHROW;
dberr_t operator()(buf_block_t* block) UNIV_NOTHROW;
private:
/** Update the page, set the space id, max trx id and index id.
@param block block read from file
......@@ -1954,8 +1939,7 @@ PageConverter::update_page(
updated then its state must be set to BUF_PAGE_NOT_USED.
@param block block read from file, note it is not from the buffer pool
@retval DB_SUCCESS or error code. */
dberr_t
PageConverter::operator() (os_offset_t, buf_block_t* block) UNIV_NOTHROW
dberr_t PageConverter::operator()(buf_block_t* block) UNIV_NOTHROW
{
/* If we already had an old page with matching number
in the buffer pool, evict it now, because
......@@ -3299,15 +3283,29 @@ fil_iterate(
const ulint size = callback.get_page_size().physical();
ulint n_bytes = iter.n_io_buffers * size;
const ulint buf_size = srv_page_size
#ifdef HAVE_LZO
+ LZO1X_1_15_MEM_COMPRESS
#elif defined HAVE_SNAPPY
+ snappy_max_compressed_length(srv_page_size)
#endif
;
byte* page_compress_buf = static_cast<byte*>(malloc(buf_size));
ut_ad(!srv_read_only_mode);
if (!page_compress_buf) {
return DB_OUT_OF_MEMORY;
}
/* TODO: For ROW_FORMAT=COMPRESSED tables we do a lot of useless
copying for non-index pages. Unfortunately, it is
required by buf_zip_decompress() */
dberr_t err = DB_SUCCESS;
for (offset = iter.start; offset < iter.end; offset += n_bytes) {
if (callback.is_interrupted()) {
return DB_INTERRUPTED;
err = DB_INTERRUPTED;
goto func_exit;
}
byte* io_buffer = iter.io_buffer;
......@@ -3337,11 +3335,12 @@ fil_iterate(
IORequest read_request(IORequest::READ);
read_request.disable_partial_io_warnings();
dberr_t err = os_file_read_no_error_handling(
err = os_file_read_no_error_handling(
read_request, iter.file, readptr, offset, n_bytes, 0);
if (err != DB_SUCCESS) {
ib::error() << iter.filepath
<< ": os_file_read() failed";
goto func_exit;
}
bool updated = false;
......@@ -3352,18 +3351,9 @@ fil_iterate(
for (ulint i = 0; i < n_pages_read;
block->page.id.set_page_no(block->page.id.page_no() + 1),
++i, page_off += size, block->frame += size) {
bool decrypted = false;
err = DB_SUCCESS;
byte* src = readptr + i * size;
byte* dst = io_buffer + i * size;
bool frame_changed = false;
ulint page_type = mach_read_from_2(src+FIL_PAGE_TYPE);
const bool page_compressed
= page_type
== FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED
|| page_type == FIL_PAGE_PAGE_COMPRESSED;
const ulint page_no = page_get_page_no(src);
if (!page_no && page_off) {
if (!page_no && block->page.id.page_no()) {
const ulint* b = reinterpret_cast<const ulint*>
(src);
const ulint* const e = b + size / sizeof *b;
......@@ -3378,54 +3368,84 @@ fil_iterate(
continue;
}
if (page_no != page_off / size) {
if (page_no != block->page.id.page_no()) {
page_corrupted:
ib::warn() << callback.filename()
<< ": Page " << (offset / size)
<< " at offset " << offset
<< " looks corrupted.";
err = DB_CORRUPTION;
goto func_exit;
}
const bool page_compressed
= fil_page_is_compressed_encrypted(src)
|| fil_page_is_compressed(src);
if (page_compressed && block->page.zip.data) {
goto page_corrupted;
}
if (encrypted) {
bool decrypted = false;
byte* dst = io_buffer + i * size;
bool frame_changed = false;
if (!encrypted) {
} else if (!mach_read_from_4(
FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION
+ src)) {
not_encrypted:
if (!page_compressed
&& !block->page.zip.data) {
block->frame = src;
frame_changed = true;
} else {
ut_ad(dst != src);
memcpy(dst, src, size);
}
} else {
if (!fil_space_verify_crypt_checksum(
src, callback.get_page_size(),
block->page.id.space(),
block->page.id.page_no())) {
goto page_corrupted;
}
decrypted = fil_space_decrypt(
iter.crypt_data, dst,
callback.get_page_size(), src, &err);
if (err != DB_SUCCESS) {
return err;
goto func_exit;
}
if (decrypted) {
updated = true;
} else {
if (!page_compressed
&& !block->page.zip.data) {
block->frame = src;
frame_changed = true;
} else {
ut_ad(dst != src);
memcpy(dst, src, size);
}
if (!decrypted) {
goto not_encrypted;
}
updated = true;
}
/* If the original page is page_compressed, we need
to decompress it before adjusting further. */
if (page_compressed) {
fil_decompress_page(NULL, dst, ulong(size),
NULL);
ulint compress_length = fil_page_decompress(
page_compress_buf, dst);
ut_ad(compress_length != srv_page_size);
if (compress_length == 0) {
goto page_corrupted;
}
updated = true;
} else if (buf_page_is_corrupted(
false,
encrypted && !frame_changed
? dst : src,
callback.get_page_size(), NULL)) {
page_corrupted:
ib::warn() << callback.filename()
<< ": Page " << (offset / size)
<< " at offset " << offset
<< " looks corrupted.";
return DB_CORRUPTION;
goto page_corrupted;
}
if ((err = callback(page_off, block)) != DB_SUCCESS) {
return err;
if ((err = callback(block)) != DB_SUCCESS) {
goto func_exit;
} else if (!updated) {
updated = buf_block_get_state(block)
== BUF_BLOCK_FILE_PAGE;
......@@ -3475,20 +3495,18 @@ fil_iterate(
src = io_buffer + (i * size);
if (page_compressed) {
ulint len = 0;
fil_compress_page(
NULL,
src,
NULL,
size,
0,/* FIXME: compression level */
512,/* FIXME: use proper block size */
encrypted,
&len);
ut_ad(len <= size);
memset(src + len, 0, size - len);
updated = true;
if (ulint len = fil_page_compress(
src,
page_compress_buf,
0,/* FIXME: compression level */
512,/* FIXME: proper block size */
encrypted)) {
/* FIXME: remove memcpy() */
memcpy(src, page_compress_buf, len);
memset(src + len, 0,
srv_page_size - len);
}
}
/* Encrypt the page if encryption was used. */
......@@ -3520,12 +3538,14 @@ fil_iterate(
writeptr, offset, n_bytes);
if (err != DB_SUCCESS) {
return err;
goto func_exit;
}
}
}
return DB_SUCCESS;
func_exit:
free(page_compress_buf);
return err;
}
/********************************************************************//**
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7