mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2026-04-25 01:35:31 +02:00
Code Issues Projects Releases Packages Wiki Activity

MDEV-28457 Crash in page_dir_find_owner_slot()

Browse source 
A prominent remaining source of crashes on corrupted index pages
is page directory corruption.

A frequent caller of page_dir_find_owner_slot() is page_rec_get_prev().
Some of those calls can be replaced with simpler logic that is less
prone to fail.

page_dir_find_owner_slot(),
page_rec_get_prev(), page_rec_get_prev_const(),
btr_pcur_move_to_prev(), btr_pcur_move_to_prev_on_page(),
btr_cur_upd_rec_sys(),
page_delete_rec_list_end(),
rtr_page_copy_rec_list_end_no_locks(),
rtr_page_copy_rec_list_start_no_locks(): Return an error code on failure.

fil_space_t::io(), buf_page_get_low(): Use DB_CORRUPTION for
out-of-bounds page reads.

PageBulk::getSplitRec(), PageBulk::copyOut(): Simplify the code.

btr_validate_level(): Prevent some more CHECK TABLE crashes on
corrupted pages.

btr_block_get(), btr_pcur_move_to_next_page(): Implement some checks that
were previously only part of IndexPurge::next().

IndexPurge::next(): Use btr_pcur_move_to_next_page().
This commit is contained in:
Marko Mäkelä 2022-06-08 14:53:24 +03:00
commit 77b3959b5c
23 changed files with 391 additions and 338 deletions
Download patch file Download diff file Expand all files Collapse all files

108
storage/innobase/btr/btr0btr.cc
View file

@ -234,7 +234,9 @@ buf_block_t *btr_block_get(const dict_index_t &index,
{
if (!!page_is_comp(block->page.frame) != index.table->not_redundant() ||
btr_page_get_index_id(block->page.frame) != index.id ||
!fil_page_index_page_check(block->page.frame))
!fil_page_index_page_check(block->page.frame) ||
index.is_spatial() !=
(fil_page_get_type(block->page.frame) == FIL_PAGE_RTREE))
{
*err= DB_PAGE_CORRUPTED;
block= nullptr;
@ -2716,11 +2718,9 @@ page_move_rec_list_end(
ut_ad(new_data_size >= old_data_size);
page_delete_rec_list_end(split_rec, block, index,
new_n_recs - old_n_recs,
new_data_size - old_data_size, mtr);
return DB_SUCCESS;
return page_delete_rec_list_end(split_rec, block, index,
new_n_recs - old_n_recs,
new_data_size - old_data_size, mtr);
}
/*************************************************************//**
@ -2980,10 +2980,15 @@ insert_empty:
page_zip_copy_recs(new_block,
page_zip, page, cursor->index, mtr);
page_delete_rec_list_end(move_limit - page + new_page,
new_block, cursor->index,
ULINT_UNDEFINED,
ULINT_UNDEFINED, mtr);
*err = page_delete_rec_list_end(move_limit
- page + new_page,
new_block,
cursor->index,
ULINT_UNDEFINED,
ULINT_UNDEFINED, mtr);
if (*err != DB_SUCCESS) {
return nullptr;
}
/* Update the lock table and possible hash index. */
if (cursor->index->has_locking()) {
@ -3045,10 +3050,13 @@ insert_empty:
/* Delete the records from the source page. */
page_delete_rec_list_end(move_limit, block,
cursor->index,
ULINT_UNDEFINED,
ULINT_UNDEFINED, mtr);
*err = page_delete_rec_list_end(move_limit, block,
cursor->index,
ULINT_UNDEFINED,
ULINT_UNDEFINED, mtr);
if (*err != DB_SUCCESS) {
return nullptr;
}
}
left_block = block;
@ -4699,13 +4707,16 @@ btr_validate_level(
default:
err = e;
}
ut_a(index->table->space_id == block->page.id().space());
ut_a(block->page.id().space() == page_get_space_id(page));
ut_ad(index->table->space_id == block->page.id().space());
ut_ad(block->page.id().space() == page_get_space_id(page));
#ifdef UNIV_ZIP_DEBUG
page_zip = buf_block_get_page_zip(block);
ut_a(!page_zip || page_zip_validate(page_zip, page, index));
#endif /* UNIV_ZIP_DEBUG */
ut_a(!page_is_leaf(page));
if (page_is_leaf(page)) {
err = DB_CORRUPTION;
goto invalid_page;
}
page_cur_set_before_first(block, &cursor);
page_cur_move_to_next(&cursor);
@ -4833,7 +4844,11 @@ func_exit:
err = DB_CORRUPTION;
}
rec = page_rec_get_prev(page_get_supremum_rec(page));
if (!(rec = page_rec_get_prev(page_get_supremum_rec(page)))) {
btr_validate_report1(index, level, block);
fputs("InnoDB: broken record links\n", stderr);
goto invalid_page;
}
right_rec = page_rec_get_next(page_get_infimum_rec(
right_page));
offsets = rec_get_offsets(rec, index, offsets,
@ -4857,10 +4872,12 @@ func_exit:
fputs("InnoDB: records in wrong order"
" on adjacent pages\n", stderr);
fputs("InnoDB: record ", stderr);
rec = page_rec_get_prev(page_get_supremum_rec(page));
rec_print(stderr, rec, index);
putc('\n', stderr);
if (rec) {
fputs("InnoDB: record ", stderr);
rec_print(stderr, rec, index);
putc('\n', stderr);
}
fputs("InnoDB: record ", stderr);
rec = page_rec_get_next(
page_get_infimum_rec(right_page));
@ -4905,15 +4922,17 @@ func_exit:
rightmost_child = page_rec_is_supremum(
page_rec_get_next(node_ptr));
btr_cur_position(
index,
page_rec_get_prev(page_get_supremum_rec(page)),
block, &node_cur);
rec = page_rec_get_prev(page_get_supremum_rec(page));
if (rec) {
btr_cur_position(index, rec, block, &node_cur);
offsets = btr_page_get_father_node_ptr_for_validate(
offsets = btr_page_get_father_node_ptr_for_validate(
offsets, heap, &node_cur, &mtr);
} else {
offsets = nullptr;
}
if (node_ptr != btr_cur_get_rec(&node_cur)
if (!offsets || node_ptr != btr_cur_get_rec(&node_cur)
|| btr_node_ptr_get_child_page_no(node_ptr, offsets)
!= block->page.id().page_no()) {
@ -4925,14 +4944,17 @@ func_exit:
fputs("InnoDB: node ptr ", stderr);
rec_print(stderr, node_ptr, index);
rec = btr_cur_get_rec(&node_cur);
fprintf(stderr, "\n"
"InnoDB: node ptr child page n:o %u\n",
btr_node_ptr_get_child_page_no(rec, offsets));
if (offsets) {
rec = btr_cur_get_rec(&node_cur);
fprintf(stderr, "\n"
"InnoDB: node ptr child page n:o %u\n",
btr_node_ptr_get_child_page_no(
rec, offsets));
fputs("InnoDB: record on page ", stderr);
rec_print_new(stderr, rec, offsets);
putc('\n', stderr);
}
fputs("InnoDB: record on page ", stderr);
rec_print_new(stderr, rec, offsets);
putc('\n', stderr);
err = DB_CORRUPTION;
goto node_ptr_fails;
}
@ -4963,15 +4985,21 @@ func_exit:
}
if (left_page_no == FIL_NULL) {
ut_a(node_ptr == page_rec_get_next(
page_get_infimum_rec(father_page)));
ut_a(!page_has_prev(father_page));
if (page_has_prev(father_page)
|| node_ptr != page_rec_get_next(
page_get_infimum_rec(father_page))) {
err = DB_CORRUPTION;
goto node_ptr_fails;
}
}
if (right_page_no == FIL_NULL) {
ut_a(node_ptr == page_rec_get_prev(
page_get_supremum_rec(father_page)));
ut_a(!page_has_next(father_page));
if (page_has_next(father_page)
|| node_ptr != page_rec_get_prev(
page_get_supremum_rec(father_page))) {
err = DB_CORRUPTION;
goto node_ptr_fails;
}
} else {
const rec_t* right_node_ptr;

35
storage/innobase/btr/btr0bulk.cc
View file

@ -637,7 +637,7 @@ PageBulk::getSplitRec()
< total_used_size / 2);
/* Keep at least one record on left page */
if (page_rec_is_infimum(page_rec_get_prev(rec))) {
if (page_rec_is_second(rec, m_page)) {
rec = page_rec_get_next(rec);
ut_ad(page_rec_is_user_rec(rec));
}
@ -679,35 +679,40 @@ void
PageBulk::copyOut(
rec_t* split_rec)
{
rec_t* rec;
rec_t* last_rec;
ulint n;
/* Suppose before copyOut, we have 5 records on the page:
infimum->r1->r2->r3->r4->r5->supremum, and r3 is the split rec.
after copyOut, we have 2 records on the page:
infimum->r1->r2->supremum. slot ajustment is not done. */
rec = page_rec_get_next(page_get_infimum_rec(m_page));
last_rec = page_rec_get_prev(page_get_supremum_rec(m_page));
n = 0;
rec_t *rec = page_get_infimum_rec(m_page);
ulint n;
while (rec != split_rec) {
rec = page_rec_get_next(rec);
n++;
for (n = 0;; n++) {
rec_t *next = page_rec_get_next(rec);
if (next == split_rec) {
break;
}
rec = next;
}
ut_ad(n > 0);
const rec_t *last_rec = split_rec;
for (;;) {
const rec_t *next = page_rec_get_next_const(last_rec);
if (page_rec_is_supremum(next)) {
break;
}
last_rec = next;
}
/* Set last record's next in page */
rec_offs* offsets = NULL;
rec = page_rec_get_prev(split_rec);
const ulint n_core = page_rec_is_leaf(split_rec)
? m_index->n_core_fields : 0;
offsets = rec_get_offsets(rec, m_index, offsets, n_core,
ULINT_UNDEFINED, &m_heap);
rec_offs* offsets = rec_get_offsets(rec, m_index, nullptr, n_core,
ULINT_UNDEFINED, &m_heap);
mach_write_to_2(rec - REC_NEXT, m_is_comp
? static_cast<uint16_t>
(PAGE_NEW_SUPREMUM - page_offset(rec))

60
storage/innobase/btr/btr0cur.cc
View file

@ -2132,12 +2132,10 @@ need_opposite_intention:
if (matched_fields
>= rec_offs_n_fields(offsets) - 1) {
detected_same_key_root = true;
} else {
const rec_t* last_rec;
last_rec = page_rec_get_prev_const(
page_get_supremum_rec(page));
} else if (const rec_t* last_rec
= page_rec_get_prev_const(
page_get_supremum_rec(
page))) {
matched_fields = 0;
offsets2 = rec_get_offsets(
@ -2151,6 +2149,9 @@ need_opposite_intention:
>= rec_offs_n_fields(offsets) - 1) {
detected_same_key_root = true;
}
} else {
err = DB_CORRUPTION;
goto func_exit;
}
}
}
@ -2710,7 +2711,10 @@ btr_cur_open_at_index_side(
if (from_left) {
page_cur_move_to_next(page_cursor);
} else {
page_cur_move_to_prev(page_cursor);
if (!page_cur_move_to_prev(page_cursor)) {
err = DB_CORRUPTION;
goto exit_loop;
}
}
if (estimate) {
@ -2800,7 +2804,7 @@ btr_cur_open_at_index_side(
}
exit_loop:
if (heap) {
if (UNIV_LIKELY_NULL(heap)) {
mem_heap_free(heap);
}
@ -3845,6 +3849,7 @@ static void btr_cur_write_sys(
trx_write_roll_ptr(static_cast<byte*>(r->data), roll_ptr);
}
MY_ATTRIBUTE((warn_unused_result))
/** Update DB_TRX_ID, DB_ROLL_PTR in a clustered index record.
@param[in,out] block clustered index leaf page
@param[in,out] rec clustered index record
@ -3852,11 +3857,12 @@ static void btr_cur_write_sys(
@param[in] offsets rec_get_offsets(rec, index)
@param[in] trx transaction
@param[in] roll_ptr DB_ROLL_PTR value
@param[in,out] mtr mini-transaction */
static void btr_cur_upd_rec_sys(buf_block_t *block, rec_t *rec,
dict_index_t *index, const rec_offs *offsets,
const trx_t *trx, roll_ptr_t roll_ptr,
mtr_t *mtr)
@param[in,out] mtr mini-transaction
@return error code */
static dberr_t btr_cur_upd_rec_sys(buf_block_t *block, rec_t *rec,
dict_index_t *index, const rec_offs *offsets,
const trx_t *trx, roll_ptr_t roll_ptr,
mtr_t *mtr)
{
ut_ad(index->is_primary());
ut_ad(rec_offs_validate(rec, index, offsets));
@ -3865,7 +3871,7 @@ static void btr_cur_upd_rec_sys(buf_block_t *block, rec_t *rec,
{
page_zip_write_trx_id_and_roll_ptr(block, rec, offsets, index->db_trx_id(),
trx->id, roll_ptr, mtr);
return;
return DB_SUCCESS;
}
ulint offset= index->trx_id_offset;
@ -3895,8 +3901,8 @@ static void btr_cur_upd_rec_sys(buf_block_t *block, rec_t *rec,
if (UNIV_LIKELY(index->trx_id_offset))
{
const rec_t *prev= page_rec_get_prev_const(rec);
if (UNIV_UNLIKELY(prev == rec))
ut_ad(0);
if (UNIV_UNLIKELY(!prev || prev == rec))
return DB_CORRUPTION;
else if (page_rec_is_infimum(prev));
else
for (src= prev + offset; d < DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN; d++)
@ -3934,6 +3940,8 @@ static void btr_cur_upd_rec_sys(buf_block_t *block, rec_t *rec,
if (UNIV_LIKELY(len)) /* extra safety, to avoid corrupting the log */
mtr->memcpy<mtr_t::MAYBE_NOP>(*block, dest, sys + d, len);
return DB_SUCCESS;
}
/*************************************************************//**
@ -4239,8 +4247,11 @@ btr_cur_update_in_place(
}
if (!(flags & BTR_KEEP_SYS_FLAG)) {
btr_cur_upd_rec_sys(block, rec, index, offsets,
thr_get_trx(thr), roll_ptr, mtr);
err = btr_cur_upd_rec_sys(block, rec, index, offsets,
thr_get_trx(thr), roll_ptr, mtr);
if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
goto func_exit;
}
}
was_delete_marked = rec_get_deleted_flag(
@ -4694,7 +4705,9 @@ any_extern:
page_cur_delete_rec(page_cursor, index, *offsets, mtr);
page_cur_move_to_prev(page_cursor);
if (!page_cur_move_to_prev(page_cursor)) {
return DB_CORRUPTION;
}
if (!(flags & BTR_KEEP_SYS_FLAG)) {
btr_cur_write_sys(new_entry, index, trx_id, roll_ptr);
@ -5056,7 +5069,10 @@ btr_cur_pessimistic_update(
page_cur_delete_rec(page_cursor, index, *offsets, mtr);
page_cur_move_to_prev(page_cursor);
if (!page_cur_move_to_prev(page_cursor)) {
err = DB_CORRUPTION;
goto return_after_reservations;
}
rec = btr_cur_insert_if_possible(cursor, new_entry,
offsets, offsets_heap, n_ext, mtr);
@ -5354,8 +5370,8 @@ btr_cur_del_mark_set_clust_rec(
<< ib::hex(trx->id) << ": "
<< rec_printer(rec, offsets).str());
btr_cur_upd_rec_sys(block, rec, index, offsets, trx, roll_ptr, mtr);
return(err);
return btr_cur_upd_rec_sys(block, rec, index, offsets, trx, roll_ptr,
mtr);
}
/*==================== B-TREE RECORD REMOVE =========================*/

8
storage/innobase/btr/btr0defragment.cc
View file

@ -707,9 +707,11 @@ processed:
page_t* last_page = buf_block_get_frame(last_block);
rec_t* rec = page_rec_get_prev(
page_get_supremum_rec(last_page));
ut_a(page_rec_is_user_rec(rec));
page_cur_position(rec, last_block,
btr_pcur_get_page_cur(item->pcur));
if (rec && page_rec_is_user_rec(rec)) {
page_cur_position(rec, last_block,
btr_pcur_get_page_cur(
item->pcur));
}
btr_pcur_store_position(item->pcur, &mtr);
mtr_commit(&mtr);
/* Update the last_processed time of this index. */

27
storage/innobase/btr/btr0pcur.cc
View file

@ -148,6 +148,11 @@ before_first:
if (page_rec_is_supremum_low(offs)) {
rec = page_rec_get_prev(rec);
if (UNIV_UNLIKELY(!rec || page_rec_is_infimum(rec))) {
ut_ad("corrupted index" == 0);
cursor->rel_pos = BTR_PCUR_AFTER_LAST_IN_TREE;
return;
}
ut_ad(!page_rec_is_infimum(rec));
if (UNIV_UNLIKELY(rec_is_metadata(rec, *index))) {
@ -486,7 +491,17 @@ btr_pcur_move_to_next_page(
const page_t* page = btr_pcur_get_page(cursor);
const uint32_t next_page_no = btr_page_get_next(page);
ut_ad(next_page_no != FIL_NULL);
switch (next_page_no) {
case 0:
case 1:
case FIL_NULL:
return DB_CORRUPTION;
}
if (UNIV_UNLIKELY(next_page_no == btr_pcur_get_block(cursor)
->page.id().page_no())) {
return DB_CORRUPTION;
}
ulint mode = cursor->latch_mode;
switch (mode) {
@ -599,13 +614,9 @@ btr_pcur_move_to_prev(
cursor->old_stored = false;
if (btr_pcur_is_before_first_on_page(cursor)) {
if (btr_pcur_is_before_first_in_tree(cursor)
|| btr_pcur_move_backward_from_page(cursor, mtr)) {
return false;
}
} else {
btr_pcur_move_to_prev_on_page(cursor);
return (!btr_pcur_is_before_first_in_tree(cursor)
&& !btr_pcur_move_backward_from_page(cursor, mtr));
}
return true;
return btr_pcur_move_to_prev_on_page(cursor) != nullptr;
}

11
storage/innobase/btr/btr0sea.cc
View file

@ -781,7 +781,7 @@ btr_search_check_guess(
mem_heap_t* heap = NULL;
rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
rec_offs* offsets = offsets_;
ibool success = FALSE;
bool success = false;
rec_offs_init(offsets_);
n_unique = dict_index_get_n_unique_in_tree(cursor->index);
@ -806,7 +806,7 @@ btr_search_check_guess(
cursor->up_match = match;
if (match >= n_unique) {
success = TRUE;
success = true;
goto exit_func;
}
} else if (mode == PAGE_CUR_LE) {
@ -835,10 +835,13 @@ btr_search_check_guess(
match = 0;
if ((mode == PAGE_CUR_G) || (mode == PAGE_CUR_GE)) {
ut_ad(!page_rec_is_infimum(rec));
const rec_t* prev_rec = page_rec_get_prev(rec);
if (UNIV_UNLIKELY(!prev_rec)) {
ut_ad("corrupted index" == 0);
goto exit_func;
}
if (page_rec_is_infimum(prev_rec)) {
success = !page_has_prev(page_align(prev_rec));
goto exit_func;

3
storage/innobase/buf/buf0buf.cc
View file

@ -2587,7 +2587,8 @@ loop:
checksum cannot be decypted. */
if (dberr_t local_err = buf_read_page(page_id, zip_size)) {
if (mode != BUF_GET_POSSIBLY_FREED
if (local_err != DB_CORRUPTION
&& mode != BUF_GET_POSSIBLY_FREED
&& retries++ < BUF_PAGE_READ_MAX_RETRIES) {
DBUG_EXECUTE_IF("intermittent_read_failure",
retries = BUF_PAGE_READ_MAX_RETRIES;);

2
storage/innobase/fil/fil0fil.cc
View file

@ -2860,7 +2860,7 @@ fail:
io_error:
#endif
set_corrupted();
err = DB_IO_ERROR;
err = DB_CORRUPTION;
node = nullptr;
goto release;
}

103
storage/innobase/gis/gis0rtree.cc
View file

@ -718,7 +718,6 @@ rtr_split_page_move_rec_list(
page_zip_des_t* new_page_zip
= buf_block_get_page_zip(new_block);
rec_t* rec;
rec_t* ret;
ulint moved = 0;
ulint max_to_move = 0;
rtr_rec_move_t* rec_move = NULL;
@ -733,7 +732,6 @@ rtr_split_page_move_rec_list(
page = buf_block_get_frame(block);
new_page = buf_block_get_frame(new_block);
ret = page_rec_get_prev(page_get_supremum_rec(new_page));
end_split_node = node_array + page_get_n_recs(page);
@ -804,32 +802,15 @@ rtr_split_page_move_rec_list(
if (!page_zip_compress(new_block, index,
page_zip_level, mtr)) {
/* Before trying to reorganize the page,
store the number of preceding records on the page. */
ulint ret_pos = page_rec_get_n_recs_before(ret);
/* Before copying, "ret" was the predecessor
of the predefined supremum record. If it was
the predefined infimum record, then it would
still be the infimum, and we would have
ret_pos == 0. */
switch (dberr_t err =
if (dberr_t err =
page_zip_reorganize(new_block, index,
page_zip_level, mtr)) {
case DB_FAIL:
if (UNIV_UNLIKELY
(!page_zip_decompress(new_page_zip,
new_page, FALSE))) {
ut_error;
if (err == DB_FAIL) {
ut_a(page_zip_decompress(new_page_zip,
new_page,
FALSE));
}
#ifdef UNIV_GIS_DEBUG
ut_ad(page_validate(new_page, index));
#endif
/* fall through */
default:
return err;
case DB_SUCCESS:
ret = page_rec_get_nth(new_page, ret_pos);
}
}
}
@ -1284,14 +1265,9 @@ rtr_ins_enlarge_mbr(
/*************************************************************//**
Copy recs from a page to new_block of rtree.
Differs from page_copy_rec_list_end, because this function does not
touch the lock table and max trx id on page or compress the page.
IMPORTANT: The caller will have to update IBUF_BITMAP_FREE
if new_block is a compressed leaf page in a secondary index.
This has to be done either within the same mini-transaction,
or by invoking ibuf_reset_free_bits() before mtr_commit(). */
void
@return error code */
dberr_t
rtr_page_copy_rec_list_end_no_locks(
/*================================*/
buf_block_t* new_block, /*!< in: index page to copy to */
@ -1355,8 +1331,7 @@ rtr_page_copy_rec_list_end_no_locks(
offsets1, offsets2, index, false,
&cur_matched_fields);
if (cmp < 0) {
page_cur_move_to_prev(&page_cur);
break;
goto move_to_prev;
} else if (cmp > 0) {
/* Skip small recs. */
page_cur_move_to_next(&page_cur);
@ -1379,26 +1354,23 @@ rtr_page_copy_rec_list_end_no_locks(
/* If position is on suprenum rec, need to move to
previous rec. */
if (page_rec_is_supremum(cur_rec)) {
page_cur_move_to_prev(&page_cur);
move_to_prev:
cur_rec = page_cur_move_to_prev(&page_cur);
} else {
cur_rec = page_cur_get_rec(&page_cur);
}
cur_rec = page_cur_get_rec(&page_cur);
if (UNIV_UNLIKELY(!cur_rec)) {
return DB_CORRUPTION;
}
offsets1 = rec_get_offsets(cur1_rec, index, offsets1, n_core,
ULINT_UNDEFINED, &heap);
ins_rec = page_cur_insert_rec_low(&page_cur, index,
cur1_rec, offsets1, mtr);
if (UNIV_UNLIKELY(!ins_rec)) {
fprintf(stderr, "page number %u and %u\n",
new_block->page.id().page_no(),
block->page.id().page_no());
ib::fatal() << "rec offset " << page_offset(rec)
<< ", cur1 offset "
<< page_offset(page_cur_get_rec(&cur1))
<< ", cur_rec offset "
<< page_offset(cur_rec);
if (UNIV_UNLIKELY(!ins_rec || moved >= max_move)) {
return DB_CORRUPTION;
}
rec_move[moved].new_rec = ins_rec;
@ -1406,20 +1378,18 @@ rtr_page_copy_rec_list_end_no_locks(
rec_move[moved].moved = false;
moved++;
next:
if (moved > max_move) {
ut_ad(0);
break;
}
page_cur_move_to_next(&cur1);
}
*num_moved = moved;
return DB_SUCCESS;
}
/*************************************************************//**
Copy recs till a specified rec from a page to new_block of rtree. */
void
Copy recs till a specified rec from a page to new_block of rtree.
@return error code */
dberr_t
rtr_page_copy_rec_list_start_no_locks(
/*==================================*/
buf_block_t* new_block, /*!< in: index page to copy to */
@ -1474,9 +1444,7 @@ rtr_page_copy_rec_list_start_no_locks(
offsets1, offsets2, index, false,
&cur_matched_fields);
if (cmp < 0) {
page_cur_move_to_prev(&page_cur);
cur_rec = page_cur_get_rec(&page_cur);
break;
goto move_to_prev;
} else if (cmp > 0) {
/* Skip small recs. */
page_cur_move_to_next(&page_cur);
@ -1500,23 +1468,22 @@ rtr_page_copy_rec_list_start_no_locks(
/* If position is on suprenum rec, need to move to
previous rec. */
if (page_rec_is_supremum(cur_rec)) {
page_cur_move_to_prev(&page_cur);
move_to_prev:
cur_rec = page_cur_move_to_prev(&page_cur);
if (UNIV_UNLIKELY(!cur_rec)) {
return DB_CORRUPTION;
}
} else {
cur_rec = page_cur_get_rec(&page_cur);
}
cur_rec = page_cur_get_rec(&page_cur);
offsets1 = rec_get_offsets(cur1_rec, index, offsets1, n_core,
ULINT_UNDEFINED, &heap);
ins_rec = page_cur_insert_rec_low(&page_cur, index,
cur1_rec, offsets1, mtr);
if (UNIV_UNLIKELY(!ins_rec)) {
ib::fatal() << new_block->page.id()
<< "rec offset " << page_offset(rec)
<< ", cur1 offset "
<< page_offset(page_cur_get_rec(&cur1))
<< ", cur_rec offset "
<< page_offset(cur_rec);
if (UNIV_UNLIKELY(!ins_rec || moved >= max_move)) {
return DB_CORRUPTION;
}
rec_move[moved].new_rec = ins_rec;
@ -1524,15 +1491,11 @@ rtr_page_copy_rec_list_start_no_locks(
rec_move[moved].moved = false;
moved++;
next:
if (moved > max_move) {
ut_ad(0);
break;
}
page_cur_move_to_next(&cur1);
}
*num_moved = moved;
return DB_SUCCESS;
}
/****************************************************************//**

8
storage/innobase/handler/handler0alter.cc
View file

@ -1993,7 +1993,7 @@ static bool innobase_table_is_empty(const dict_table_t *table,
btr_pcur_t pcur;
buf_block_t *block;
page_cur_t *cur;
const rec_t *rec;
rec_t *rec;
bool next_page= false;
mtr.start();
@ -2004,9 +2004,9 @@ non_empty:
mtr.commit();
return false;
}
btr_pcur_move_to_next_user_rec(&pcur, &mtr);
if (!rec_is_metadata(btr_pcur_get_rec(&pcur), *clust_index))
btr_pcur_move_to_prev_on_page(&pcur);
rec= page_rec_get_next(btr_pcur_get_rec(&pcur));
if (rec_is_metadata(rec, *clust_index))
btr_pcur_get_page_cur(&pcur)->rec= rec;
scan_leaf:
cur= btr_pcur_get_page_cur(&pcur);
page_cur_move_to_next(cur);

56
storage/innobase/ibuf/ibuf0ibuf.cc
View file

@ -2035,23 +2035,25 @@ ibuf_get_merge_page_nos_func(
*n_stored = 0;
limit = ut_min(IBUF_MAX_N_PAGES_MERGED,
buf_pool_get_curr_size() / 4);
if (page_rec_is_supremum(rec)) {
rec = page_rec_get_prev_const(rec);
if (UNIV_UNLIKELY(!rec)) {
corruption:
ut_ad("corrupted page" == 0);
return 0;
}
}
if (page_rec_is_infimum(rec)) {
rec = page_rec_get_next_const(rec);
if (page_rec_is_supremum(rec)) {
return 0;
}
}
if (page_rec_is_supremum(rec)) {
return(0);
}
limit = ut_min(IBUF_MAX_N_PAGES_MERGED,
buf_pool_get_curr_size() / 4);
first_page_no = ibuf_rec_get_page_no(mtr, rec);
first_space_id = ibuf_rec_get_space(mtr, rec);
@ -2083,7 +2085,9 @@ ibuf_get_merge_page_nos_func(
prev_page_no = rec_page_no;
prev_space_id = rec_space_id;
rec = page_rec_get_prev_const(rec);
if (UNIV_UNLIKELY(!(rec = page_rec_get_prev_const(rec)))) {
goto corruption;
}
}
rec = page_rec_get_next_const(rec);
@ -2809,14 +2813,16 @@ ibuf_get_volume_buffered(
page = page_align(rec);
ut_ad(page_validate(page, ibuf.index));
if (page_rec_is_supremum(rec)) {
rec = page_rec_get_prev_const(rec);
if (page_rec_is_supremum(rec)
&& UNIV_UNLIKELY(!(rec = page_rec_get_prev_const(rec)))) {
corruption:
ut_ad("corrupted page" == 0);
return srv_page_size;
}
uint32_t prev_page_no;
for (; !page_rec_is_infimum(rec);
rec = page_rec_get_prev_const(rec)) {
for (; !page_rec_is_infimum(rec); ) {
ut_ad(page_align(rec) == page);
if (page_no != ibuf_rec_get_page_no(mtr, rec)
@ -2828,6 +2834,10 @@ ibuf_get_volume_buffered(
volume += ibuf_get_volume_buffered_count(
mtr, rec,
hash_bitmap, UT_ARR_SIZE(hash_bitmap), n_recs);
if (UNIV_UNLIKELY(!(rec = page_rec_get_prev_const(rec)))) {
goto corruption;
}
}
/* Look at the previous page */
@ -2853,13 +2863,16 @@ ibuf_get_volume_buffered(
if (UNIV_UNLIKELY(memcmp_aligned<4>(prev_page + FIL_PAGE_NEXT,
page + FIL_PAGE_OFFSET, 4))) {
return 0;
return srv_page_size;
}
rec = page_get_supremum_rec(prev_page);
rec = page_rec_get_prev_const(rec);
rec = page_rec_get_prev_const(page_get_supremum_rec(prev_page));
for (;; rec = page_rec_get_prev_const(rec)) {
if (UNIV_UNLIKELY(!rec)) {
goto corruption;
}
for (;;) {
ut_ad(page_align(rec) == prev_page);
if (page_rec_is_infimum(rec)) {
@ -2880,6 +2893,10 @@ ibuf_get_volume_buffered(
volume += ibuf_get_volume_buffered_count(
mtr, rec,
hash_bitmap, UT_ARR_SIZE(hash_bitmap), n_recs);
if (UNIV_UNLIKELY(!(rec = page_rec_get_prev_const(rec)))) {
goto corruption;
}
}
count_later:
@ -3801,7 +3818,10 @@ ibuf_insert_to_index_page(
buffered one. */
page_cur_delete_rec(&page_cur, index, offsets, mtr);
page_cur_move_to_prev(&page_cur);
if (!(page_cur_move_to_prev(&page_cur))) {
err = DB_CORRUPTION;
goto updated_in_place;
}
} else {
offsets = NULL;
}

3
storage/innobase/include/btr0pcur.h
View file

@ -331,10 +331,11 @@ void
btr_pcur_move_to_next_on_page(
/*==========================*/
btr_pcur_t* cursor);/*!< in/out: persistent cursor */
MY_ATTRIBUTE((nonnull, warn_unused_result))
/*********************************************************//**
Moves the persistent cursor to the previous record on the same page. */
UNIV_INLINE
void
rec_t*
btr_pcur_move_to_prev_on_page(
/*==========================*/
btr_pcur_t* cursor);/*!< in/out: persistent cursor */

7
storage/innobase/include/btr0pcur.inl
View file

@ -171,17 +171,16 @@ btr_pcur_move_to_next_on_page(
/*********************************************************//**
Moves the persistent cursor to the previous record on the same page. */
UNIV_INLINE
void
rec_t*
btr_pcur_move_to_prev_on_page(
/*==========================*/
btr_pcur_t* cursor) /*!< in/out: persistent cursor */
{
ut_ad(cursor->pos_state == BTR_PCUR_IS_POSITIONED);
ut_ad(cursor->latch_mode != BTR_NO_LATCHES);
page_cur_move_to_prev(btr_pcur_get_page_cur(cursor));
cursor->old_stored = false;
return page_cur_move_to_prev(btr_pcur_get_page_cur(cursor));
}
/*********************************************************//**

14
storage/innobase/include/gis0rtree.h
View file

@ -337,9 +337,12 @@ rtr_get_parent_cursor(
ulint level, /*!< in: index level of buffer page */
ulint is_insert); /*!< in: whether insert operation */
MY_ATTRIBUTE((warn_unused_result))
/*************************************************************//**
Copy recs from a page to new_block of rtree. */
void
Copy recs from a page to new_block of rtree.
@return error code */
dberr_t
rtr_page_copy_rec_list_end_no_locks(
/*================================*/
buf_block_t* new_block, /*!< in: index page to copy to */
@ -352,9 +355,12 @@ rtr_page_copy_rec_list_end_no_locks(
ulint* num_moved, /*!< out: num of rec to move */
mtr_t* mtr); /*!< in: mtr */
MY_ATTRIBUTE((warn_unused_result))
/*************************************************************//**
Copy recs till a specified rec from a page to new_block of rtree. */
void
Copy recs till a specified rec from a page to new_block of rtree.
@return error code */
dberr_t
rtr_page_copy_rec_list_start_no_locks(
/*==================================*/
buf_block_t* new_block, /*!< in: index page to copy to */

3
storage/innobase/include/page0cur.h
View file

@ -120,10 +120,11 @@ void
page_cur_move_to_next(
/*==================*/
page_cur_t* cur); /*!< in/out: cursor; must not be after last */
MY_ATTRIBUTE((nonnull, warn_unused_result))
/**********************************************************//**
Moves the cursor to the previous record on page. */
UNIV_INLINE
void
rec_t*
page_cur_move_to_prev(
/*==================*/
page_cur_t* cur); /*!< in/out: cursor; not before first */

4
storage/innobase/include/page0cur.inl
View file

@ -168,14 +168,14 @@ page_cur_move_to_next(
/**********************************************************//**
Moves the cursor to the previous record on page. */
UNIV_INLINE
void
rec_t*
page_cur_move_to_prev(
/*==================*/
page_cur_t* cur) /*!< in/out: page cursor, not before first */
{
ut_ad(!page_cur_is_before_first(cur));
cur->rec = page_rec_get_prev(cur->rec);
return cur->rec = page_rec_get_prev(cur->rec);
}
/** Search the right position for a page cursor.

24
storage/innobase/include/page0page.h
View file

@ -669,7 +669,8 @@ page_dir_calc_reserved_space(
ulint n_recs); /*!< in: number of records */
/***************************************************************//**
Looks for the directory slot which owns the given record.
@return the directory slot number */
@return the directory slot number
@retval ULINT_UNDEFINED on corruption */
ulint
page_dir_find_owner_slot(
/*=====================*/
@ -763,7 +764,8 @@ page_rec_get_next_non_del_marked(
const rec_t* rec); /*!< in: pointer to record */
/************************************************************//**
Gets the pointer to the previous record.
@return pointer to previous record */
@return pointer to previous record
@retval nullptr on error */
UNIV_INLINE
const rec_t*
page_rec_get_prev_const(
@ -772,13 +774,13 @@ page_rec_get_prev_const(
infimum */
/************************************************************//**
Gets the pointer to the previous record.
@return pointer to previous record */
UNIV_INLINE
rec_t*
page_rec_get_prev(
/*==============*/
rec_t* rec); /*!< in: pointer to record,
must not be page infimum */
@param rec record (not page infimum)
@return pointer to previous record
@retval nullptr on error */
inline rec_t *page_rec_get_prev(rec_t *rec)
{
return const_cast<rec_t*>(page_rec_get_prev_const(rec));
}
/************************************************************//**
true if the record is the first user record on a page.
@ -997,7 +999,7 @@ page_copy_rec_list_start(
/*************************************************************//**
Deletes records from a page from a given record onward, including that record.
The infimum and supremum records are not deleted. */
void
dberr_t
page_delete_rec_list_end(
/*=====================*/
rec_t* rec, /*!< in: pointer to record on page */
@ -1009,7 +1011,7 @@ page_delete_rec_list_end(
records in the end of the chain to
delete, or ULINT_UNDEFINED if not known */
mtr_t* mtr) /*!< in: mtr */
MY_ATTRIBUTE((nonnull));
MY_ATTRIBUTE((nonnull, warn_unused_result));
/*************************************************************//**
Deletes records from page, up to the given record, NOT including
that record. Infimum and supremum records are not deleted. */

34
storage/innobase/include/page0page.inl
View file

@ -1,7 +1,7 @@
/*****************************************************************************
Copyright (c) 1994, 2019, Oracle and/or its affiliates. All Rights Reserved.
Copyright (c) 2016, 2021, MariaDB Corporation.
Copyright (c) 2016, 2022, 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
@ -24,9 +24,6 @@ Index page routines
Created 2/2/1994 Heikki Tuuri
*******************************************************/
#ifndef page0page_ic
#define page0page_ic
#ifndef UNIV_INNOCHECKSUM
#include "rem0cmp.h"
#include "mtr0log.h"
@ -506,7 +503,8 @@ page_rec_get_next_non_del_marked(
/************************************************************//**
Gets the pointer to the previous record.
@return pointer to previous record */
@return pointer to previous record
@retval nullptr on error */
UNIV_INLINE
const rec_t*
page_rec_get_prev_const(
@ -528,42 +526,28 @@ page_rec_get_prev_const(
slot_no = page_dir_find_owner_slot(rec);
ut_a(slot_no != 0);
if (UNIV_UNLIKELY(!slot_no || slot_no == ULINT_UNDEFINED)) {
return nullptr;
}
slot = page_dir_get_nth_slot(page, slot_no - 1);
rec2 = page_dir_slot_get_rec(slot);
if (page_is_comp(page)) {
while (rec != rec2) {
while (rec2 && rec != rec2) {
prev_rec = rec2;
rec2 = page_rec_get_next_low(rec2, TRUE);
}
} else {
while (rec != rec2) {
while (rec2 && rec != rec2) {
prev_rec = rec2;
rec2 = page_rec_get_next_low(rec2, FALSE);
}
}
ut_a(prev_rec);
return(prev_rec);
}
/************************************************************//**
Gets the pointer to the previous record.
@return pointer to previous record */
UNIV_INLINE
rec_t*
page_rec_get_prev(
/*==============*/
rec_t* rec) /*!< in: pointer to record, must not be page
infimum */
{
return((rec_t*) page_rec_get_prev_const(rec));
}
#endif /* UNIV_INNOCHECKSUM */
/************************************************************//**
@ -720,5 +704,3 @@ page_get_instant(const page_t* page)
return static_cast<uint16_t>(i >> 3); /* i / 8 */
}
#endif /* !UNIV_INNOCHECKSUM */
#endif

22
storage/innobase/page/page0cur.cc
View file

@ -1621,7 +1621,9 @@ copied:
if (UNIV_UNLIKELY(n_owned == PAGE_DIR_SLOT_MAX_N_OWNED))
{
const auto owner= page_dir_find_owner_slot(next_rec);
const ulint owner= page_dir_find_owner_slot(next_rec);
if (UNIV_UNLIKELY(owner == ULINT_UNDEFINED))
return nullptr;
page_dir_split_slot(*block,
page_dir_get_nth_slot(block->page.frame, owner));
}
@ -2047,8 +2049,12 @@ inc_dir:
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_zip_dir_split_slot(cursor->block,
page_dir_find_owner_slot(next_rec), mtr);
{
const ulint owner= page_dir_find_owner_slot(next_rec);
if (UNIV_UNLIKELY(owner == ULINT_UNDEFINED))
return nullptr;
page_zip_dir_split_slot(cursor->block, owner, mtr);
}
page_zip_write_rec(cursor->block, insert_rec, index, offsets, 1, mtr);
return insert_rec;
@ -2144,7 +2150,6 @@ page_cur_delete_rec(
rec_t* current_rec;
rec_t* prev_rec = NULL;
rec_t* next_rec;
ulint cur_slot_no;
ulint cur_n_owned;
rec_t* rec;
@ -2188,8 +2193,13 @@ page_cur_delete_rec(
}
/* Save to local variables some data associated with current_rec */
cur_slot_no = page_dir_find_owner_slot(current_rec);
ut_ad(cur_slot_no > 0);
ulint cur_slot_no = page_dir_find_owner_slot(current_rec);
if (UNIV_UNLIKELY(!cur_slot_no || cur_slot_no == ULINT_UNDEFINED)) {
/* Avoid crashing due to a corrupted page. */
return;
}
cur_dir_slot = page_dir_get_nth_slot(block->page.frame, cur_slot_no);
cur_n_owned = page_dir_slot_get_n_owned(cur_dir_slot);

56
storage/innobase/page/page0page.cc
View file

@ -82,7 +82,8 @@ is 50 x 4 bytes = 200 bytes. */
/***************************************************************//**
Looks for the directory slot which owns the given record.
@return the directory slot number */
@return the directory slot number
@retval ULINT_UNDEFINED on corruption */
ulint
page_dir_find_owner_slot(
/*=====================*/
@ -135,7 +136,7 @@ page_dir_find_owner_slot(
+ mach_decode_2(rec_offs_bytes));
}
ut_error;
return ULINT_UNDEFINED;
}
slot += PAGE_DIR_SLOT_SIZE;
@ -589,12 +590,12 @@ page_copy_rec_list_end(
/* For spatial index, we need to insert recs one by one
to keep recs ordered. */
rtr_page_copy_rec_list_end_no_locks(new_block,
block, rec, index,
heap, rec_move,
max_to_move,
&num_moved,
mtr);
*err = rtr_page_copy_rec_list_end_no_locks(new_block,
block, rec, index,
heap, rec_move,
max_to_move,
&num_moved,
mtr);
} else {
*err = page_copy_rec_list_end_no_locks(new_block, block, rec,
index, mtr);
@ -719,6 +720,11 @@ page_copy_rec_list_start(
rec_offs* offsets = offsets_;
rec_offs_init(offsets_);
if (UNIV_UNLIKELY(!ret)) {
*err = DB_CORRUPTION;
return ret;
}
/* Here, "ret" may be pointing to a user record or the
predefined infimum record. */
@ -753,10 +759,14 @@ page_copy_rec_list_start(
/* For spatial index, we need to insert recs one by one
to keep recs ordered. */
rtr_page_copy_rec_list_start_no_locks(new_block,
block, rec, index, heap,
rec_move, max_to_move,
&num_moved, mtr);
*err = rtr_page_copy_rec_list_start_no_locks(new_block,
block, rec, index,
heap, rec_move,
max_to_move,
&num_moved, mtr);
if (*err != DB_SUCCESS) {
return nullptr;
}
} else {
while (page_cur_get_rec(&cur1) != rec) {
offsets = rec_get_offsets(cur1.rec, index, offsets,
@ -857,7 +867,7 @@ zip_reorganize:
/*************************************************************//**
Deletes records from a page from a given record onward, including that record.
The infimum and supremum records are not deleted. */
void
dberr_t
page_delete_rec_list_end(
/*=====================*/
rec_t* rec, /*!< in: pointer to record on page */
@ -884,7 +894,7 @@ page_delete_rec_list_end(
{
ut_ad(n_recs == 0 || n_recs == ULINT_UNDEFINED);
/* Nothing to do, there are no records bigger than the page supremum. */
return;
return DB_SUCCESS;
}
if (page_rec_is_infimum(rec) ||
@ -895,7 +905,7 @@ page_delete_rec_list_end(
{
/* We are deleting all records. */
page_create_empty(block, index, mtr);
return;
return DB_SUCCESS;
}
#if 0 // FIXME: consider deleting the last record as a special case
@ -903,7 +913,7 @@ page_delete_rec_list_end(
{
page_cur_t cursor= { index, rec, offsets, block };
page_cur_delete_rec(&cursor, index, offsets, mtr);
return;
return DB_SUCCESS;
}
#endif
@ -936,12 +946,16 @@ page_delete_rec_list_end(
if (UNIV_LIKELY_NULL(heap))
mem_heap_free(heap);
return;
return DB_SUCCESS;
}
#endif
byte *prev_rec= page_rec_get_prev(rec);
if (UNIV_UNLIKELY(!prev_rec))
return DB_CORRUPTION;
byte *last_rec= page_rec_get_prev(page_get_supremum_rec(page));
if (UNIV_UNLIKELY(!last_rec))
return DB_CORRUPTION;
// FIXME: consider a special case of shrinking PAGE_HEAP_TOP
@ -998,9 +1012,11 @@ page_delete_rec_list_end(
ut_ad(n_owned > count);
n_owned-= count;
slot_index= page_dir_find_owner_slot(owner_rec);
ut_ad(slot_index > 0);
}
if (UNIV_UNLIKELY(!slot_index || slot_index == ULINT_UNDEFINED))
return DB_CORRUPTION;
mtr->write<2,mtr_t::MAYBE_NOP>(*block, my_assume_aligned<2>
(PAGE_N_DIR_SLOTS + PAGE_HEADER + page),
slot_index + 1);
@ -1046,7 +1062,7 @@ page_delete_rec_list_end(
mach_write_to_2(last_rec - REC_NEXT, free
? static_cast<uint16_t>(free - page_offset(last_rec))
: 0U);
return;
return DB_SUCCESS;
}
#endif
mtr->write<1,mtr_t::MAYBE_NOP>(*block, owned, new_owned);
@ -1066,6 +1082,8 @@ page_delete_rec_list_end(
mtr->write<2>(*block, prev_rec - REC_NEXT, PAGE_OLD_SUPREMUM);
mtr->write<2>(*block, last_rec - REC_NEXT, free);
}
return DB_SUCCESS;
}
/*************************************************************//**

70
storage/innobase/row/row0import.cc
View file

@ -1531,15 +1531,10 @@ inline bool IndexPurge::open() noexcept
&m_pcur, true, 0, &m_mtr) != DB_SUCCESS)
return false;
btr_pcur_move_to_next_user_rec(&m_pcur, &m_mtr);
if (rec_is_metadata(btr_pcur_get_rec(&m_pcur), *m_index))
{
if (!btr_pcur_is_on_user_rec(&m_pcur))
return false;
rec_t *rec= page_rec_get_next(btr_pcur_get_rec(&m_pcur));
if (rec_is_metadata(rec, *m_index))
/* Skip the metadata pseudo-record. */
}
else
btr_pcur_move_to_prev_on_page(&m_pcur);
btr_pcur_get_page_cur(&m_pcur)->rec= rec;
return true;
}
@ -1582,55 +1577,10 @@ dberr_t IndexPurge::next() noexcept
return DB_END_OF_INDEX;
}
buf_block_t* block = btr_pcur_get_block(&m_pcur);
uint32_t next_page = btr_page_get_next(
block->page.frame);
/* MDEV-13542 FIXME: Make these checks part of
btr_pcur_move_to_next_page(), and introduce a
return status that will be checked in all callers! */
switch (next_page) {
default:
if (next_page != block->page.id().page_no()) {
break;
}
/* MDEV-20931 FIXME: Check that
next_page is within the tablespace
bounds! Also check that it is not a
change buffer bitmap page. */
/* fall through */
case 0:
case 1:
case FIL_NULL:
return DB_CORRUPTION;
if (dberr_t err = btr_pcur_move_to_next_page(&m_pcur,
&m_mtr)) {
return err;
}
dict_index_t* index = m_pcur.btr_cur.index;
buf_block_t* next_block = btr_block_get(
*index, next_page, BTR_MODIFY_LEAF, false,
&m_mtr);
if (UNIV_UNLIKELY(!next_block
|| !fil_page_index_page_check(
next_block->page.frame)
|| !!dict_index_is_spatial(index)
!= (fil_page_get_type(
next_block->page.frame)
== FIL_PAGE_RTREE)
|| page_is_comp(next_block->page.frame)
!= page_is_comp(block->page.frame)
|| btr_page_get_prev(
next_block->page.frame)
!= block->page.id().page_no())) {
return DB_CORRUPTION;
}
btr_leaf_page_release(block, BTR_MODIFY_LEAF, &m_mtr);
page_cur_set_before_first(next_block,
&m_pcur.btr_cur.page_cur);
ut_d(page_check_dir(next_block->page.frame));
} else {
btr_pcur_move_to_next_on_page(&m_pcur);
}
@ -2347,11 +2297,11 @@ row_import_set_sys_max_row_id(
if (btr_pcur_open_at_index_side(false, index, BTR_SEARCH_LEAF,
&pcur, true, 0, &mtr) == DB_SUCCESS) {
btr_pcur_move_to_prev_on_page(&pcur);
rec = btr_pcur_get_rec(&pcur);
rec = btr_pcur_move_to_prev_on_page(&pcur);
/* Check for empty table. */
if (page_rec_is_infimum(rec)) {
if (!rec) {
/* The table is corrupted. */
} else if (page_rec_is_infimum(rec)) {
/* The table is empty. */
} else if (rec_is_metadata(rec, *index)) {
/* The clustered index contains the metadata

51
storage/innobase/row/row0merge.cc
View file

@ -139,7 +139,10 @@ public:
if (log_sys.check_flush_or_checkpoint()) {
if (mtr_started) {
btr_pcur_move_to_prev_on_page(pcur);
if (!btr_pcur_move_to_prev_on_page(pcur)) {
error = DB_CORRUPTION;
break;
}
btr_pcur_store_position(pcur, scan_mtr);
scan_mtr->commit();
mtr_started = false;
@ -1839,14 +1842,27 @@ row_merge_read_clustered_index(
err_exit:
trx->error_key_num = 0;
goto func_exit;
}
btr_pcur_move_to_next_user_rec(&pcur, &mtr);
if (rec_is_metadata(btr_pcur_get_rec(&pcur), *clust_index)) {
ut_ad(btr_pcur_is_on_user_rec(&pcur));
/* Skip the metadata pseudo-record. */
} else {
ut_ad(!clust_index->is_instant());
btr_pcur_move_to_prev_on_page(&pcur);
rec_t* rec = page_rec_get_next(btr_pcur_get_rec(&pcur));
if (!rec) {
corrupted_metadata:
err = DB_CORRUPTION;
goto err_exit;
}
if (rec_get_info_bits(rec, page_rec_is_comp(rec))
& REC_INFO_MIN_REC_FLAG) {
if (!clust_index->is_instant()) {
goto corrupted_metadata;
}
if (page_rec_is_comp(rec)
&& rec_get_status(rec) != REC_STATUS_INSTANT) {
goto corrupted_metadata;
}
/* Skip the metadata pseudo-record. */
btr_pcur_get_page_cur(&pcur)->rec = rec;
} else if (clust_index->is_instant()) {
goto corrupted_metadata;
}
}
/* Check if the table is supposed to be empty for our read view.
@ -1986,13 +2002,16 @@ err_exit:
/* Store the cursor position on the last user
record on the page. */
btr_pcur_move_to_prev_on_page(&pcur);
if (!btr_pcur_move_to_prev_on_page(&pcur)) {
goto corrupted_index;
}
/* Leaf pages must never be empty, unless
this is the only page in the index tree. */
ut_ad(btr_pcur_is_on_user_rec(&pcur)
|| btr_pcur_get_block(
&pcur)->page.id().page_no()
== clust_index->page);
if (!btr_pcur_is_on_user_rec(&pcur)
&& btr_pcur_get_block(&pcur)->page.id()
.page_no() != clust_index->page) {
goto corrupted_index;
}
btr_pcur_store_position(&pcur, &mtr);
mtr.commit();
@ -2495,8 +2514,10 @@ write_buffers:
we must reread it on the next
loop iteration. */
if (mtr_started) {
btr_pcur_move_to_prev_on_page(
&pcur);
if (!btr_pcur_move_to_prev_on_page(&pcur)) {
err = DB_CORRUPTION;
goto func_exit;
}
btr_pcur_store_position(
&pcur, &mtr);

20
storage/innobase/row/row0sel.cc
View file

@ -4715,6 +4715,15 @@ wait_table_again:
pcur, moves_up, &mtr);
if (UNIV_UNLIKELY(need_to_process)) {
if (UNIV_UNLIKELY(!btr_pcur_get_rec(pcur))) {
mtr.commit();
trx->op_info = "";
if (UNIV_LIKELY_NULL(heap)) {
mem_heap_free(heap);
}
return DB_CORRUPTION;
}
if (UNIV_UNLIKELY(prebuilt->row_read_type
== ROW_READ_DID_SEMI_CONSISTENT)) {
/* We did a semi-consistent read,
@ -4732,7 +4741,7 @@ wait_table_again:
pessimistic locking read, the record
cannot be skipped. */
goto next_rec;
goto next_rec_after_check;
}
} else if (dtuple_get_n_fields(search_tuple) > 0) {
@ -5727,6 +5736,7 @@ next_rec:
== ROW_READ_DID_SEMI_CONSISTENT)) {
prebuilt->row_read_type = ROW_READ_TRY_SEMI_CONSISTENT;
}
next_rec_after_check:
did_semi_consistent_read = false;
prebuilt->new_rec_locks = 0;
vrow = NULL;
@ -5752,7 +5762,6 @@ next_rec:
/* No need to do store restore for R-tree */
mtr.commit();
mtr.start();
mtr_extra_clust_savepoint = 0;
} else if (mtr_extra_clust_savepoint) {
/* We must release any clustered index latches
if we are moving to the next non-clustered
@ -5760,9 +5769,10 @@ next_rec:
order if we would access a different clustered
index page right away without releasing the previous. */
mtr.rollback_to_savepoint(mtr_extra_clust_savepoint);
mtr_extra_clust_savepoint = 0;
}
mtr_extra_clust_savepoint = 0;
if (moves_up) {
if (UNIV_UNLIKELY(spatial_search)) {
if (rtr_pcur_move_to_next(
@ -5792,6 +5802,10 @@ next_rec:
if (btr_pcur_move_to_prev(pcur, &mtr)) {
goto rec_loop;
}
if (UNIV_UNLIKELY(!btr_pcur_get_rec(pcur))) {
err = DB_CORRUPTION;
goto normal_return;
}
}
not_moved: