MDEV-28457 Crash in page_dir_find_owner_slot()

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().

23 files changed, 391 insertions, 338 deletions

diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 72375fc6c43..fc7aad3f84f 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -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;
 
diff --git a/storage/innobase/btr/btr0bulk.cc b/storage/innobase/btr/btr0bulk.cc
index 538cb06e654..7af5f862aab 100644
--- a/storage/innobase/btr/btr0bulk.cc
+++ b/storage/innobase/btr/btr0bulk.cc
@@ -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))
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 3499e9d84fe..981a80adba0 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -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 =========================*/
diff --git a/storage/innobase/btr/btr0defragment.cc b/storage/innobase/btr/btr0defragment.cc
index 23d93caecf5..5a278fbd9a2 100644
--- a/storage/innobase/btr/btr0defragment.cc
+++ b/storage/innobase/btr/btr0defragment.cc
@@ -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. */
diff --git a/storage/innobase/btr/btr0pcur.cc b/storage/innobase/btr/btr0pcur.cc
index 8c33fee9e61..c45ce3be89e 100644
--- a/storage/innobase/btr/btr0pcur.cc
+++ b/storage/innobase/btr/btr0pcur.cc
@@ -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;
 }
diff --git a/storage/innobase/btr/btr0sea.cc b/storage/innobase/btr/btr0sea.cc
index a9b4e9ff0d0..53cc464f18f 100644
--- a/storage/innobase/btr/btr0sea.cc
+++ b/storage/innobase/btr/btr0sea.cc
@@ -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;
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index e6a6bd8cbb7..e0632513584 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -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;);
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index de94ec5f0c1..18f68aa5dd6 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -2860,7 +2860,7 @@ fail:
 io_error:
 #endif
 				set_corrupted();
-				err = DB_IO_ERROR;
+				err = DB_CORRUPTION;
 				node = nullptr;
 				goto release;
 			}
diff --git a/storage/innobase/gis/gis0rtree.cc b/storage/innobase/gis/gis0rtree.cc
index 8a5e1dc3ac9..df5d6aa1156 100644
--- a/storage/innobase/gis/gis0rtree.cc
+++ b/storage/innobase/gis/gis0rtree.cc
@@ -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;
 }
 
 /****************************************************************//**
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index d63f5b33654..57eecd47ebb 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -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);
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index 94c01f5108a..6fb41cfbd81 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -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;
         }
diff --git a/storage/innobase/include/btr0pcur.h b/storage/innobase/include/btr0pcur.h
index 5fc4e28527f..f2a1f4220da 100644
--- a/storage/innobase/include/btr0pcur.h
+++ b/storage/innobase/include/btr0pcur.h
@@ -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 */
diff --git a/storage/innobase/include/btr0pcur.inl b/storage/innobase/include/btr0pcur.inl
index fd4eeb9392a..b21de209760 100644
--- a/storage/innobase/include/btr0pcur.inl
+++ b/storage/innobase/include/btr0pcur.inl
@@ -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));
 }
 
 /*********************************************************//**
diff --git a/storage/innobase/include/gis0rtree.h b/storage/innobase/include/gis0rtree.h
index 4e10b90173e..8cd5e384530 100644
--- a/storage/innobase/include/gis0rtree.h
+++ b/storage/innobase/include/gis0rtree.h
@@ -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 */
diff --git a/storage/innobase/include/page0cur.h b/storage/innobase/include/page0cur.h
index d80eb4567e5..e715df19741 100644
--- a/storage/innobase/include/page0cur.h
+++ b/storage/innobase/include/page0cur.h
@@ -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 */
diff --git a/storage/innobase/include/page0cur.inl b/storage/innobase/include/page0cur.inl
index 5ee96dd716d..6f7c633561f 100644
--- a/storage/innobase/include/page0cur.inl
+++ b/storage/innobase/include/page0cur.inl
@@ -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.
diff --git a/storage/innobase/include/page0page.h b/storage/innobase/include/page0page.h
index a4a86791cd8..4787ce36c7a 100644
--- a/storage/innobase/include/page0page.h
+++ b/storage/innobase/include/page0page.h
@@ -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. */
diff --git a/storage/innobase/include/page0page.inl b/storage/innobase/include/page0page.inl
index 861bf4a53df..76bc62e5eb2 100644
--- a/storage/innobase/include/page0page.inl
+++ b/storage/innobase/include/page0page.inl
@@ -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
diff --git a/storage/innobase/page/page0cur.cc b/storage/innobase/page/page0cur.cc
index bd0905e80dc..e608abaa361 100644
--- a/storage/innobase/page/page0cur.cc
+++ b/storage/innobase/page/page0cur.cc
@@ -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);
 
diff --git a/storage/innobase/page/page0page.cc b/storage/innobase/page/page0page.cc
index bb2c267c633..22d3983072b 100644
--- a/storage/innobase/page/page0page.cc
+++ b/storage/innobase/page/page0page.cc
@@ -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;
 }
 
 /*************************************************************//**
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 96202311198..9f1bcbb820e 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -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
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index f01231fda16..0a8655ee4c4 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -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);
 
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 47c4b87a7c9..ebd4ddce130 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -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: