MDEV-21098: Assertion failure in rec_get_offsets_func()

The function rec_get_offsets_func() used to hit ut_error
due to an invalid rec_get_status() value of a
ROW_FORMAT!=REDUNDANT record. This fix is twofold:
We will not only avoid a crash on corruption in this case,
but we will also make more effort to validate each record
every time we are iterating over index page records.

rec_get_offsets_func(): Do not crash on a corrupted record.

page_rec_get_nth(): Return nullptr on error.

page_dir_slot_get_rec_validate(): Like page_dir_slot_get_rec(),
but validate the pointer and return nullptr on error.

page_cur_search_with_match(), page_cur_search_with_match_bytes(),
page_dir_split_slot(), page_cur_move_to_next():
Indicate failure in a return value.

page_cur_search(): Replaced with page_cur_search_with_match().

rec_get_next_ptr_const(), rec_get_next_ptr(): Replaced with
page_rec_get_next_low().

TODO: rtr_page_split_initialize_nodes(), rtr_update_mbr_field(),
and possibly other SPATIAL INDEX functions fail to properly handle
errors.

Reviewed by: Thirunarayanan Balathandayuthapani
Tested by: Matthias Leich
Performance tested by: Axel Schwenke

storage/innobase/btr/btr0bulk.cc

@@ -1199,6 +1199,9 @@ BtrBulk::finish(dberr_t	err)
 
 		first_rec = page_rec_get_next(
 			page_get_infimum_rec(last_block->page.frame));
+		/* Because this index tree is being created by this thread,
+		we assume that it cannot be corrupted. */
+		ut_ad(first_rec);
 		ut_ad(page_rec_is_user_rec(first_rec));
 
 		/* Copy last page to root page. */

storage/innobase/btr/btr0cur.cc

@@ -403,15 +403,13 @@ static dberr_t btr_cur_instant_init_low(dict_index_t* index, mtr_t* mtr)
 	ut_ad(page_cur_is_before_first(&cur.page_cur));
 	ut_ad(page_is_leaf(cur.page_cur.block->page.frame));
 
-	page_cur_move_to_next(&cur.page_cur);
-
-	const rec_t* rec = cur.page_cur.rec;
+	const rec_t* rec = page_cur_move_to_next(&cur.page_cur);
 	const ulint comp = dict_table_is_comp(index->table);
-	const ulint info_bits = rec_get_info_bits(rec, comp);
+	const ulint info_bits = rec ? rec_get_info_bits(rec, comp) : 0;
 
 	if (page_rec_is_supremum(rec)
 	    || !(info_bits & REC_INFO_MIN_REC_FLAG)) {
-		if (!index->is_instant()) {
+		if (rec && !index->is_instant()) {
 			/* The FIL_PAGE_TYPE_INSTANT and PAGE_INSTANT may be
 			assigned even if instant ADD COLUMN was not
 			committed. Changes to these page header fields are not
@@ -884,6 +882,21 @@ btr_cur_latch_for_root_leaf(
 	return(RW_NO_LATCH); /* avoid compiler warnings */
 }
 
+/** @return whether the distance between two records is at most the
+specified value */
+static bool
+page_rec_distance_is_at_most(const rec_t *left, const rec_t *right, ulint val)
+{
+  do
+  {
+    if (left == right)
+      return true;
+    left= page_rec_get_next_const(left);
+  }
+  while (left && val--);
+  return false;
+}
+
 /** Detects whether the modifying record might need a modifying tree structure.
 @param[in]	index		index
 @param[in]	page		page
@@ -1914,22 +1927,28 @@ btr_cur_search_to_nth_level_func(
 #ifdef BTR_CUR_HASH_ADAPT
 	} else if (height == 0 && btr_search_enabled
 		   && !(tuple->info_bits & REC_INFO_MIN_REC_FLAG)
-		   && !dict_index_is_spatial(index)) {
+		   && index->is_btree()) {
 		/* The adaptive hash index is only used when searching
 		for leaf pages (height==0), but not in r-trees.
 		We only need the byte prefix comparison for the purpose
 		of updating the adaptive hash index. */
-		page_cur_search_with_match_bytes(
+		if (page_cur_search_with_match_bytes(
 			block, index, tuple, page_mode, &up_match, &up_bytes,
-			&low_match, &low_bytes, page_cursor);
+			&low_match, &low_bytes, page_cursor)) {
+			err = DB_CORRUPTION;
+			goto func_exit;
+		}
 #endif /* BTR_CUR_HASH_ADAPT */
 	} else {
 		/* Search for complete index fields. */
 		up_bytes = low_bytes = 0;
-		page_cur_search_with_match(
+		if (page_cur_search_with_match(
 			block, index, tuple, page_mode, &up_match,
 			&low_match, page_cursor,
-			need_path ? cursor->rtr_info : NULL);
+			need_path ? cursor->rtr_info : nullptr)) {
+			err = DB_CORRUPTION;
+			goto func_exit;
+		}
 	}
 
 	/* If this is the desired level, leave the loop */
@@ -2097,6 +2116,11 @@ btr_cur_search_to_nth_level_func(
 
 			ut_ad(upper_rw_latch == RW_X_LATCH);
 
+			if (UNIV_UNLIKELY(!first_rec)) {
+			corrupted:
+				err = DB_CORRUPTION;
+				goto func_exit;
+			}
 			if (node_ptr == first_rec
 			    || page_rec_is_last(node_ptr, page)) {
 				detected_same_key_root = true;
@@ -2131,8 +2155,7 @@ btr_cur_search_to_nth_level_func(
 						detected_same_key_root = true;
 					}
 				} else {
-					err = DB_CORRUPTION;
-					goto func_exit;
+					goto corrupted;
 				}
 			}
 		}
@@ -2241,11 +2264,14 @@ btr_cur_search_to_nth_level_func(
 					? cursor->rtr_info : NULL;
 
 				for (ulint i = 0; i < n_blocks; i++) {
-					page_cur_search_with_match(
+					if (page_cur_search_with_match(
 						tree_blocks[i], index, tuple,
 						page_mode, &up_match,
 						&low_match, page_cursor,
-						rtr_info);
+						rtr_info)) {
+						err = DB_CORRUPTION;
+						goto func_exit;
+					}
 				}
 
 				goto search_loop;
@@ -2678,13 +2704,11 @@ btr_cur_open_at_index_side(
 
 		ut_ad(height > 0);
 
-		if (from_left) {
-			page_cur_move_to_next(page_cursor);
-		} else {
-			if (!page_cur_move_to_prev(page_cursor)) {
-				err = DB_CORRUPTION;
-				goto exit_loop;
-			}
+		if (from_left
+		    ? !page_cur_move_to_next(page_cursor)
+		    : !page_cur_move_to_prev(page_cursor)) {
+			err = DB_CORRUPTION;
+			goto exit_loop;
 		}
 
 		height--;
@@ -3117,7 +3141,7 @@ btr_cur_insert_if_possible(
 
 /*************************************************************//**
 For an insert, checks the locks and does the undo logging if desired.
-@return DB_SUCCESS, DB_WAIT_LOCK, DB_FAIL, or error number */
+@return DB_SUCCESS, DB_LOCK_WAIT, DB_FAIL, or error number */
 UNIV_INLINE MY_ATTRIBUTE((warn_unused_result, nonnull(2,3,5,6)))
 dberr_t
 btr_cur_ins_lock_and_undo(
@@ -3268,7 +3292,7 @@ It is assumed that mtr holds an x-latch on the page. The operation does
 not succeed if there is too little space on the page. If there is just
 one record on the page, the insert will always succeed; this is to
 prevent trying to split a page with just one record.
-@return DB_SUCCESS, DB_WAIT_LOCK, DB_FAIL, or error number */
+@return DB_SUCCESS, DB_LOCK_WAIT, DB_FAIL, or error number */
 dberr_t
 btr_cur_optimistic_insert(
 /*======================*/
@@ -3736,7 +3760,7 @@ btr_cur_pessimistic_insert(
 
 /*************************************************************//**
 For an update, checks the locks and does the undo logging.
-@return DB_SUCCESS, DB_WAIT_LOCK, or error number */
+@return DB_SUCCESS, DB_LOCK_WAIT, or error number */
 UNIV_INLINE MY_ATTRIBUTE((warn_unused_result))
 dberr_t
 btr_cur_upd_lock_and_undo(
@@ -4682,8 +4706,7 @@ btr_cur_optimistic_update(
 	rec = btr_cur_insert_if_possible(cursor, new_entry, offsets, heap,
 					 0/*n_ext*/, mtr);
 	if (UNIV_UNLIKELY(!rec)) {
-		err = DB_CORRUPTION;
-		goto func_exit;
+		goto corrupted;
 	}
 
 	if (UNIV_UNLIKELY(update->is_metadata())) {
@@ -4701,8 +4724,11 @@ btr_cur_optimistic_update(
 						   block->page.id());
 	}
 
-	page_cur_move_to_next(page_cursor);
 	ut_ad(err == DB_SUCCESS);
+	if (!page_cur_move_to_next(page_cursor)) {
+corrupted:
+		err = DB_CORRUPTION;
+	}
 
 func_exit:
 	if (!(flags & BTR_KEEP_IBUF_BITMAP)
@@ -5454,6 +5480,10 @@ btr_cur_optimistic_delete(
 		dict_index_t* index = cursor->index;
 		const rec_t* first_rec = page_rec_get_next_const(
 			page_get_infimum_rec(block->page.frame));
+		if (UNIV_UNLIKELY(!first_rec)) {
+			err = DB_CORRUPTION;
+			goto func_exit;
+		}
 		ut_ad(!index->is_instant()
 		      || rec_is_metadata(first_rec, *index));
 		const bool is_metadata = rec_is_metadata(rec, *index);
@@ -5683,6 +5713,10 @@ btr_cur_pessimistic_delete(
 
 			const rec_t* first_rec = page_rec_get_next_const(
 				page_get_infimum_rec(page));
+			if (UNIV_UNLIKELY(!first_rec)) {
+				*err = DB_CORRUPTION;
+				goto err_exit;
+			}
 			ut_ad(!index->is_instant()
 			      || rec_is_metadata(first_rec, *index));
 			if (is_metadata || !index->is_instant()
@@ -5729,7 +5763,11 @@ btr_cur_pessimistic_delete(
 			goto return_after_reservations;
 		}
 
-		next_rec = page_rec_get_next(rec);
+		if (UNIV_UNLIKELY(!(next_rec = page_rec_get_next(rec)))) {
+			ut_ad(!ret);
+			*err = DB_CORRUPTION;
+			goto err_exit;
+		}
 
 		if (!page_has_prev(page)) {
 			/* If we delete the leftmost node pointer on a
@@ -5996,17 +6034,19 @@ class btr_est_cur_t
     if (dtuple_get_n_fields(&m_tuple) > 0)
     {
       m_up_bytes= m_low_bytes= 0;
-      page_cur_search_with_match(m_block, index(), &m_tuple, m_page_mode,
-                                 &m_up_match, &m_low_match, &m_page_cur,
-                                 nullptr);
+      if (page_cur_search_with_match(m_block, index(), &m_tuple, m_page_mode,
+                                     &m_up_match, &m_low_match, &m_page_cur,
+                                     nullptr))
+        return false;
       m_nth_rec= page_rec_get_n_recs_before(page_cur_get_rec(&m_page_cur));
     }
     else if (left)
     {
       page_cur_set_before_first(m_block, &m_page_cur);
       if (level)
       {
-        page_cur_move_to_next(&m_page_cur);
+        if (!page_cur_move_to_next(&m_page_cur))
+          return false;
         m_nth_rec= 1;
       }
       else

storage/innobase/btr/btr0defragment.cc

@@ -259,9 +259,10 @@ btr_defragment_calc_n_recs_for_size(
 
 	const ulint n_core = page_is_leaf(page) ? index->n_core_fields : 0;
 	page_cur_set_before_first(block, &cur);
-	page_cur_move_to_next(&cur);
-	while (page_cur_get_rec(&cur) != page_get_supremum_rec(page)) {
-		rec_t* cur_rec = page_cur_get_rec(&cur);
+	while (rec_t* cur_rec = page_cur_move_to_next(&cur)) {
+		if (page_rec_is_supremum(cur_rec)) {
+			break;
+		}
 		offsets = rec_get_offsets(cur_rec, index, offsets, n_core,
 					  ULINT_UNDEFINED, &heap);
 		ulint rec_size = rec_offs_size(offsets);
@@ -271,7 +272,6 @@ btr_defragment_calc_n_recs_for_size(
 			break;
 		}
 		n_recs ++;
-		page_cur_move_to_next(&cur);
 	}
 	*n_recs_size = size;
 	if (UNIV_LIKELY_NULL(heap)) {
@@ -356,8 +356,9 @@ btr_defragment_merge_pages(
 	target_n_recs = n_recs_to_move;
 	dberr_t err;
 	while (n_recs_to_move > 0) {
-		rec = page_rec_get_nth(from_page,
-					n_recs_to_move + 1);
+		if (!(rec = page_rec_get_nth(from_page, n_recs_to_move + 1))) {
+			return nullptr;
+		}
 		orig_pred = page_copy_rec_list_start(
 			to_block, from_block, rec, index, mtr, &err);
 		if (orig_pred)
@@ -439,14 +440,17 @@ btr_defragment_merge_pages(
 			}
 			rec = page_rec_get_next(
 				page_get_infimum_rec(from_page));
+			if (!rec) {
+				return nullptr;
+			}
 			node_ptr = dict_index_build_node_ptr(
 				index, rec, page_get_page_no(from_page),
 				heap, level);
 			if (btr_insert_on_non_leaf_level(0, index, level+1,
-                                                         node_ptr, mtr)
-                            != DB_SUCCESS) {
+							 node_ptr, mtr)
+			    != DB_SUCCESS) {
 				return nullptr;
-                        }
+			}
 		}
 		to_block = from_block;
 	}

storage/innobase/btr/btr0pcur.cc

@@ -176,10 +176,16 @@ btr_pcur_store_position(
 	} else if (page_rec_is_infimum_low(offs)) {
 		rec = page_rec_get_next(rec);
 
+		if (UNIV_UNLIKELY(!rec)) {
+			ut_ad("corrupted page" == 0);
+			goto before_first;
+		}
+
 		if (rec_is_metadata(rec, *index)) {
 			ut_ad(!page_has_prev(block->page.frame));
 			rec = page_rec_get_next(rec);
-			if (page_rec_is_supremum(rec)) {
+			ut_ad(rec);
+			if (!rec || page_rec_is_supremum(rec)) {
 				goto before_first;
 			}
 		}