Skip to content

M
mariadb
Commit e6bdb11f authored by unknown's avatar unknown
Browse files
Options

Use LOGREC_REDO_NEW_ROW_HEAD and LOGREC_REDO_NEW_ROW_TAIL when writing to a new page 

This makes REDO safer as we then know when it's safe to ignore HA_ERR_WRONG_CRC from page cache
Removed not used logged data for LOGREC_UNDO_ROW_DELETE and LOGREC_UNDO_ROW_UPDATE


storage/maria/ma_blockrec.c:
  Use LOGREC_REDO_NEW_ROW_HEAD and LOGREC_REDO_NEW_ROW_TAIL when writing to a new page
  This makes REDO safer as we then know when it's safe to ignore HA_ERR_WRONG_CRC from page cache
storage/maria/ma_blockrec.h:
  Fixed prototype
storage/maria/ma_key_recover.c:
  Removed not used logged data for LOGREC_UNDO_ROW_DELETE and LOGREC_UNDO_ROW_UPDATE
storage/maria/ma_loghandler.c:
  Use LOGREC_REDO_NEW_ROW_HEAD and LOGREC_REDO_NEW_ROW_TAIL when writing to a new page
storage/maria/ma_loghandler.h:
  Added LOGREC_REDO_NEW_ROW_HEAD and LOGREC_REDO_NEW_ROW_TAIL
storage/maria/ma_recovery.c:
  Added support for LOGREC_REDO_NEW_ROW_HEAD and LOGREC_REDO_NEW_ROW_TAIL
parent 7e7c534b
Hide whitespace changes
Inline Side-by-side
Showing with 63 additions and 32 deletions
storage/maria/ma_blockrec.c
View file @ e6bdb11f
......@@ -1777,7 +1777,9 @@ static my_bool write_tail(MARIA_HA *info,
log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
log_array[TRANSLOG_INTERNAL_PARTS + 1].str= (char*) row_pos.data;
log_array[TRANSLOG_INTERNAL_PARTS + 1].length= length;
if (translog_write_record(&lsn, LOGREC_REDO_INSERT_ROW_TAIL,
if (translog_write_record(&lsn,
(block_is_read ? LOGREC_REDO_INSERT_ROW_TAIL :
LOGREC_REDO_NEW_ROW_TAIL),
info->trn, info, sizeof(log_data) + length,
TRANSLOG_INTERNAL_PARTS + 2, log_array,
log_data, NULL))
......@@ -2830,7 +2832,11 @@ static my_bool write_block_record(MARIA_HA *info,
log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
log_array[TRANSLOG_INTERNAL_PARTS + 1].str= (char*) row_pos->data;
log_array[TRANSLOG_INTERNAL_PARTS + 1].length= head_length;
if (translog_write_record(&lsn, LOGREC_REDO_INSERT_ROW_HEAD, info->trn,
if (translog_write_record(&lsn,
head_block_is_read ?
LOGREC_REDO_INSERT_ROW_HEAD :
LOGREC_REDO_NEW_ROW_HEAD,
info->trn,
info, sizeof(log_data) + head_length,
TRANSLOG_INTERNAL_PARTS + 2, log_array,
log_data, NULL))
......@@ -5721,17 +5727,22 @@ my_bool write_hook_for_file_id(enum translog_record_type type
***************************************************************************/
/*
Apply LOGREC_REDO_INSERT_ROW_HEAD & LOGREC_REDO_INSERT_ROW_TAIL
Apply changes to head and tail pages
SYNOPSIS
_ma_apply_redo_insert_row_head_or_tail()
info Maria handler
lsn LSN to put on page
page_type HEAD_PAGE or TAIL_PAGE
new_page True if this is first entry on page
header Header (without FILEID)
data Data to be put on page
data_length Length of data
NOTE
Handles LOGREC_REDO_INSERT_ROW_HEAD, LOGREC_REDO_INSERT_ROW_TAIL
LOGREC_REDO_NEW_ROW_HEAD and LOGREC_REDO_NEW_ROW_TAIL
RETURN
0 ok
# Error number
......@@ -5739,6 +5750,7 @@ my_bool write_hook_for_file_id(enum translog_record_type type
uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
uint page_type,
my_bool new_page,
const uchar *header,
const uchar *data,
size_t data_length)
......@@ -5783,8 +5795,8 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
unlock_method= PAGECACHE_LOCK_WRITE;
unpin_method= PAGECACHE_PIN;
DBUG_ASSERT(rownr == 0);
if (rownr != 0)
DBUG_ASSERT(rownr == 0 && new_page);
if (rownr != 0 || !new_page)
goto crashed_file;
buff= info->keyread_buff;
......@@ -5808,8 +5820,8 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
if (!buff)
{
/* Skip errors when reading outside of file and uninitialized pages */
if (my_errno != HA_ERR_FILE_TOO_SHORT &&
my_errno != HA_ERR_WRONG_CRC)
if (!new_page || (my_errno != HA_ERR_FILE_TOO_SHORT &&
my_errno != HA_ERR_WRONG_CRC))
goto err;
/* Create new page */
buff= pagecache_block_link_to_buffer(page_link.link);
......@@ -5834,8 +5846,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
This is a page that has been freed before and now should be
changed to new type.
*/
if (((buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK) != BLOB_PAGE &&
(buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK) != UNALLOCATED_PAGE))
if (!new_page)
goto crashed_file;
make_empty_page(info, buff, page_type, 0);
empty_space= block_size - PAGE_HEADER_SIZE - PAGE_SUFFIX_SIZE;
......@@ -5850,6 +5861,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
uint max_entry= (uint) buff[DIR_COUNT_OFFSET];
uint length;
DBUG_ASSERT(!new_page);
dir= dir_entry_pos(buff, block_size, rownr);
empty_space= uint2korr(buff + EMPTY_SPACE_OFFSET);
......
storage/maria/ma_blockrec.h
View file @ e6bdb11f
......@@ -222,6 +222,7 @@ void _ma_print_bitmap(MARIA_FILE_BITMAP *bitmap, uchar *data,
uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
uint page_type,
my_bool new_page,
const uchar *header,
const uchar *data,
size_t data_length);
......
storage/maria/ma_key_recover.c
View file @ e6bdb11f
......@@ -119,21 +119,6 @@ my_bool _ma_write_clr(MARIA_HA *info, LSN undo_lsn,
page_store(log_pos + KEY_NR_STORE_SIZE, page);
log_pos+= KEY_NR_STORE_SIZE + PAGE_STORE_SIZE;
}
if (undo_type == LOGREC_UNDO_ROW_DELETE ||
undo_type == LOGREC_UNDO_ROW_UPDATE)
{
/*
We need to store position to the row that was inserted to be
able to regenerate keys
*/
MARIA_RECORD_POS rowid= info->cur_row.lastpos;
ulonglong page= ma_recordpos_to_page(rowid);
uint dir_entry= ma_recordpos_to_dir_entry(rowid);
page_store(log_pos, page);
dirpos_store(log_pos+ PAGE_STORE_SIZE, dir_entry);
log_pos+= PAGE_STORE_SIZE + DIRPOS_STORE_SIZE;
}
log_array[TRANSLOG_INTERNAL_PARTS + 0].str= (char*) log_data;
log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - log_data);
......
storage/maria/ma_loghandler.c
View file @ e6bdb11f
......@@ -403,10 +403,17 @@ static LOG_DESC INIT_LOGREC_REDO_INSERT_ROW_TAIL=
write_hook_for_redo, NULL, 0,
"redo_insert_row_tail", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
/* Use this entry next time we need to add a new entry */
static LOG_DESC INIT_LOGREC_REDO_NOT_USED=
{LOGRECTYPE_VARIABLE_LENGTH, 0, 8, NULL, write_hook_for_redo, NULL, 0,
"redo_insert_row_blob", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
static LOG_DESC INIT_LOGREC_REDO_NEW_ROW_HEAD=
{LOGRECTYPE_VARIABLE_LENGTH, 0,
FILEID_STORE_SIZE + PAGE_STORE_SIZE + DIRPOS_STORE_SIZE, NULL,
write_hook_for_redo, NULL, 0,
"redo_new_row_head", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
static LOG_DESC INIT_LOGREC_REDO_NEW_ROW_TAIL=
{LOGRECTYPE_VARIABLE_LENGTH, 0,
FILEID_STORE_SIZE + PAGE_STORE_SIZE + DIRPOS_STORE_SIZE, NULL,
write_hook_for_redo, NULL, 0,
"redo_new_row_tail", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
static LOG_DESC INIT_LOGREC_REDO_INSERT_ROW_BLOBS=
{LOGRECTYPE_VARIABLE_LENGTH, 0, FILEID_STORE_SIZE, NULL,
......@@ -594,8 +601,10 @@ void translog_table_init()
INIT_LOGREC_REDO_INSERT_ROW_HEAD;
log_record_type_descriptor[LOGREC_REDO_INSERT_ROW_TAIL]=
INIT_LOGREC_REDO_INSERT_ROW_TAIL;
log_record_type_descriptor[LOGREC_REDO_NOT_USED]=
INIT_LOGREC_REDO_NOT_USED;
log_record_type_descriptor[LOGREC_REDO_NEW_ROW_HEAD]=
INIT_LOGREC_REDO_NEW_ROW_HEAD;
log_record_type_descriptor[LOGREC_REDO_NEW_ROW_TAIL]=
INIT_LOGREC_REDO_NEW_ROW_TAIL;
log_record_type_descriptor[LOGREC_REDO_INSERT_ROW_BLOBS]=
INIT_LOGREC_REDO_INSERT_ROW_BLOBS;
log_record_type_descriptor[LOGREC_REDO_PURGE_ROW_HEAD]=
......
storage/maria/ma_loghandler.h
View file @ e6bdb11f
......@@ -105,7 +105,8 @@ enum translog_record_type
LOGREC_RESERVED_FOR_CHUNKS23= 0,
LOGREC_REDO_INSERT_ROW_HEAD,
LOGREC_REDO_INSERT_ROW_TAIL,
LOGREC_REDO_NOT_USED, /* Reserver for next tag */
LOGREC_REDO_NEW_ROW_HEAD,
LOGREC_REDO_NEW_ROW_TAIL,
LOGREC_REDO_INSERT_ROW_BLOBS,
LOGREC_REDO_PURGE_ROW_HEAD,
LOGREC_REDO_PURGE_ROW_TAIL,
......
storage/maria/ma_recovery.c
View file @ e6bdb11f
......@@ -88,7 +88,9 @@ prototype_redo_exec_hook(INCOMPLETE_LOG);
prototype_redo_exec_hook_dummy(INCOMPLETE_GROUP);
prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL);
prototype_redo_exec_hook(REDO_INSERT_ROW_BLOBS);
prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
prototype_redo_exec_hook(REDO_NEW_ROW_TAIL);
prototype_redo_exec_hook(REDO_NEW_ROW_BLOBS);
prototype_redo_exec_hook(REDO_PURGE_ROW_HEAD);
prototype_redo_exec_hook(REDO_PURGE_ROW_TAIL);
prototype_redo_exec_hook(REDO_FREE_HEAD_OR_TAIL);
......@@ -1312,6 +1314,10 @@ end:
return error;
}
/*
NOTE
This is called for REDO_INSERT_ROW_HEAD and READ_NEW_ROW_HEAD
*/
prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD)
{
......@@ -1353,6 +1359,8 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD)
buff= log_record_buffer.str;
if (_ma_apply_redo_insert_row_head_or_tail(info, current_group_end_lsn,
HEAD_PAGE,
(rec->type ==
LOGREC_REDO_NEW_ROW_HEAD),
buff + FILEID_STORE_SIZE,
buff +
FILEID_STORE_SIZE +
......@@ -1368,6 +1376,10 @@ end:
return error;
}
/*
NOTE
This is called for REDO_INSERT_ROW_TAIL and READ_NEW_ROW_TAIL
*/
prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL)
{
......@@ -1388,6 +1400,8 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL)
buff= log_record_buffer.str;
if (_ma_apply_redo_insert_row_head_or_tail(info, current_group_end_lsn,
TAIL_PAGE,
(rec->type ==
LOGREC_REDO_NEW_ROW_TAIL),
buff + FILEID_STORE_SIZE,
buff +
FILEID_STORE_SIZE +
......@@ -2189,6 +2203,9 @@ static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply)
#define install_redo_exec_hook(R) \
log_record_type_descriptor[LOGREC_ ## R].record_execute_in_redo_phase= \
exec_REDO_LOGREC_ ## R;
#define install_redo_exec_hook_shared(R,S) \
log_record_type_descriptor[LOGREC_ ## R].record_execute_in_redo_phase= \
exec_REDO_LOGREC_ ## S;
#define install_undo_exec_hook(R) \
log_record_type_descriptor[LOGREC_ ## R].record_execute_in_undo_phase= \
exec_UNDO_LOGREC_ ## R;
......@@ -2226,6 +2243,10 @@ static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply)
install_undo_exec_hook(UNDO_KEY_INSERT);
install_undo_exec_hook(UNDO_KEY_DELETE);
install_undo_exec_hook(UNDO_KEY_DELETE_WITH_ROOT);
/* REDO_NEW_ROW_HEAD shares entry with REDO_INSERT_ROW_HEAD */
install_redo_exec_hook_shared(REDO_NEW_ROW_HEAD, REDO_INSERT_ROW_HEAD);
/* REDO_NEW_ROW_TAIL shares entry with REDO_INSERT_ROW_TAIL */
install_redo_exec_hook_shared(REDO_NEW_ROW_TAIL, REDO_INSERT_ROW_TAIL);
current_group_end_lsn= LSN_IMPOSSIBLE;
#ifndef DBUG_OFF
......@@ -2612,6 +2633,8 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
case LOGREC_REDO_INSERT_ROW_TAIL:
case LOGREC_REDO_PURGE_ROW_HEAD:
case LOGREC_REDO_PURGE_ROW_TAIL:
case LOGREC_REDO_NEW_ROW_HEAD:
case LOGREC_REDO_NEW_ROW_TAIL:
llstr(page, llbuf);
tprint(tracef, " For page %s of table of short id %u", llbuf, sid);
break;
......
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