MDEV-21174: Clean up record insertion

page_cur_insert_rec_low(): Take page_cur_t* as a parameter,
and do not tolerate mtr=NULL.

page_cur_insert_rec_zip(): Do not tolerate mtr=NULL.

storage/innobase/btr/btr0btr.cc

@@ -4068,11 +4068,13 @@ btr_discard_only_page_on_level(
 
 	if (index->is_primary()) {
 		if (rec) {
+			page_cur_t cur;
+			page_cur_set_before_first(block, &cur);
 			DBUG_ASSERT(index->table->instant);
 			DBUG_ASSERT(rec_is_alter_metadata(rec, *index));
 			btr_set_instant(block, *index, mtr);
 			rec = page_cur_insert_rec_low(
-				page_get_infimum_rec(block->frame),
+				&cur,
 				index, rec, offsets, mtr);
 			ut_ad(rec);
 			mem_heap_free(heap);

storage/innobase/gis/gis0rtree.cc

@@ -840,7 +840,7 @@ rtr_split_page_move_rec_list(
 			ut_ad(!is_leaf || cur_split_node->key != first_rec);
 
 			rec = page_cur_insert_rec_low(
-				page_cur_get_rec(&new_page_cursor),
+				&new_page_cursor,
 				index, cur_split_node->key, offsets, mtr);
 
 			ut_a(rec);
@@ -1460,7 +1460,7 @@ rtr_page_copy_rec_list_end_no_locks(
 		offsets1 = rec_get_offsets(cur1_rec, index, offsets1, is_leaf,
 					   ULINT_UNDEFINED, &heap);
 
-		ins_rec = page_cur_insert_rec_low(cur_rec, index,
+		ins_rec = page_cur_insert_rec_low(&page_cur, index,
 						  cur1_rec, offsets1, mtr);
 		if (UNIV_UNLIKELY(!ins_rec)) {
 			fprintf(stderr, "page number %ld and %ld\n",
@@ -1582,14 +1582,11 @@ rtr_page_copy_rec_list_start_no_locks(
 		offsets1 = rec_get_offsets(cur1_rec, index, offsets1, is_leaf,
 					   ULINT_UNDEFINED, &heap);
 
-		ins_rec = page_cur_insert_rec_low(cur_rec, index,
+		ins_rec = page_cur_insert_rec_low(&page_cur, index,
 						  cur1_rec, offsets1, mtr);
 		if (UNIV_UNLIKELY(!ins_rec)) {
-			fprintf(stderr, "page number %ld and %ld\n",
-				(long)new_block->page.id.page_no(),
-				(long)block->page.id.page_no());
-
-			ib::fatal() << "rec offset " << page_offset(rec)
+			ib::fatal() << new_block->page.id
+				<< "rec offset " << page_offset(rec)
 				<< ", cur1 offset "
 				<<  page_offset(page_cur_get_rec(&cur1))
 				<< ", cur_rec offset "

storage/innobase/include/page0cur.h

@@ -181,13 +181,12 @@ space available, NULL otherwise. The cursor stays at the same position.
 rec_t*
 page_cur_insert_rec_low(
 /*====================*/
-	rec_t*		current_rec,/*!< in: pointer to current record after
-				which the new record is inserted */
+	const page_cur_t*cur,	/*!< in: page cursor */
 	dict_index_t*	index,	/*!< in: record descriptor */
-	const rec_t*	rec,	/*!< in: pointer to a physical record */
+	const rec_t*	rec,	/*!< in: record to insert after cur */
 	ulint*		offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr)	/*!< in: mini-transaction handle, or NULL */
-	MY_ATTRIBUTE((nonnull(1,2,3,4), warn_unused_result));
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
+	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
 /***********************************************************//**
 Inserts a record next to page cursor on a compressed and uncompressed
@@ -208,8 +207,8 @@ page_cur_insert_rec_zip(
 	dict_index_t*	index,	/*!< in: record descriptor */
 	const rec_t*	rec,	/*!< in: pointer to a physical record */
 	ulint*		offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr)	/*!< in: mini-transaction handle, or NULL */
-	MY_ATTRIBUTE((nonnull(1,2,3,4), warn_unused_result));
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
+	MY_ATTRIBUTE((nonnull, warn_unused_result));
 /*************************************************************//**
 Copies records from page to a newly created page, from a given record onward,
 including that record. Infimum and supremum records are not copied.

storage/innobase/include/page0cur.ic

 /*****************************************************************************
 
 Copyright (c) 1994, 2014, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2018, MariaDB Corporation.
+Copyright (c) 2015, 2019, 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
@@ -281,7 +281,7 @@ page_cur_tuple_insert(
 		rec = page_cur_insert_rec_zip(
 			cursor, index, rec, *offsets, mtr);
 	} else {
-		rec = page_cur_insert_rec_low(cursor->rec,
+		rec = page_cur_insert_rec_low(cursor,
 					      index, rec, *offsets, mtr);
 	}
 
@@ -315,7 +315,7 @@ page_cur_rec_insert(
 		return(page_cur_insert_rec_zip(
 			       cursor, index, rec, offsets, mtr));
 	} else {
-		return(page_cur_insert_rec_low(cursor->rec,
-					       index, rec, offsets, mtr));
+		return(page_cur_insert_rec_low(
+			       cursor, index, rec, offsets, mtr));
 	}
 }

storage/innobase/page/page0cur.cc

@@ -1385,12 +1385,11 @@ space available, NULL otherwise. The cursor stays at the same position.
 rec_t*
 page_cur_insert_rec_low(
 /*====================*/
-	rec_t*		current_rec,/*!< in: pointer to current record after
-				which the new record is inserted */
+	const page_cur_t*cur,	/*!< in: page cursor */
 	dict_index_t*	index,	/*!< in: record descriptor */
-	const rec_t*	rec,	/*!< in: pointer to a physical record */
+	const rec_t*	rec,	/*!< in: record to insert after cur */
 	ulint*		offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr)	/*!< in: mini-transaction handle, or NULL */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	byte*		insert_buf;
 	ulint		rec_size;
@@ -1403,6 +1402,8 @@ page_cur_insert_rec_low(
 	ulint		heap_no;	/*!< heap number of the inserted
 					record */
 
+	rec_t* current_rec = cur->rec;
+
 	ut_ad(rec_offs_validate(rec, index, offsets));
 
 	page = page_align(current_rec);
@@ -1411,7 +1412,7 @@ page_cur_insert_rec_low(
 	ut_ad(fil_page_index_page_check(page));
 	ut_ad(mach_read_from_8(page + PAGE_HEADER + PAGE_INDEX_ID) == index->id
 	      || index->is_dummy
-	      || (mtr ? mtr->is_inside_ibuf() : dict_index_is_ibuf(index)));
+	      || mtr->is_inside_ibuf());
 
 	ut_ad(!page_rec_is_supremum(current_rec));
 
@@ -1613,7 +1614,7 @@ page_cur_insert_rec_zip(
 	dict_index_t*	index,	/*!< in: record descriptor */
 	const rec_t*	rec,	/*!< in: pointer to a physical record */
 	ulint*		offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr)	/*!< in: mini-transaction handle, or NULL */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	byte*		insert_buf;
 	ulint		rec_size;
@@ -1638,7 +1639,7 @@ page_cur_insert_rec_zip(
 	ut_ad(fil_page_index_page_check(page));
 	ut_ad(mach_read_from_8(page + PAGE_HEADER + PAGE_INDEX_ID) == index->id
 	      || index->is_dummy
-	      || (mtr ? mtr->is_inside_ibuf() : dict_index_is_ibuf(index)));
+	      || mtr->is_inside_ibuf());
 	ut_ad(!page_get_instant(page));
 	ut_ad(!page_cur_is_after_last(cursor));
 #ifdef UNIV_ZIP_DEBUG
@@ -1731,8 +1732,10 @@ page_cur_insert_rec_zip(
 		}
 
 		/* Try compressing the whole page afterwards. */
+		const mtr_log_t log_mode = mtr->set_log_mode(MTR_LOG_NONE);
 		insert_rec = page_cur_insert_rec_low(
-			cursor->rec, index, rec, offsets, NULL);
+			cursor, index, rec, offsets, mtr);
+		mtr->set_log_mode(log_mode);
 
 		/* If recovery is on, this implies that the compression
 		of the page was successful during runtime. Had that not
@@ -2039,10 +2042,8 @@ page_cur_insert_rec_zip(
 	page_zip_write_rec(page_zip, insert_rec, index, offsets, 1);
 
 	/* 9. Write log record of the insert */
-	if (UNIV_LIKELY(mtr != NULL)) {
-		page_cur_insert_rec_write_log(insert_rec, rec_size,
-					      cursor->rec, index, mtr);
-	}
+	page_cur_insert_rec_write_log(insert_rec, rec_size,
+				      cursor->rec, index, mtr);
 
 	return(insert_rec);
 }

storage/innobase/page/page0page.cc

@@ -503,7 +503,7 @@ page_copy_rec_list_end_no_locks(
 {
 	page_t*		new_page	= buf_block_get_frame(new_block);
 	page_cur_t	cur1;
-	rec_t*		cur2;
+	page_cur_t	cur2;
 	mem_heap_t*	heap		= NULL;
 	ulint		offsets_[REC_OFFS_NORMAL_SIZE];
 	ulint*		offsets		= offsets_;
@@ -522,7 +522,7 @@ page_copy_rec_list_end_no_locks(
 	     (page_is_comp(new_page) ? PAGE_NEW_INFIMUM : PAGE_OLD_INFIMUM));
 	const bool is_leaf = page_is_leaf(block->frame);
 
-	cur2 = page_get_infimum_rec(buf_block_get_frame(new_block));
+	page_cur_set_before_first(new_block, &cur2);
 
 	/* Copy records from the original page to the new page */
 
@@ -530,18 +530,18 @@ page_copy_rec_list_end_no_locks(
 		rec_t*	ins_rec;
 		offsets = rec_get_offsets(cur1.rec, index, offsets, is_leaf,
 					  ULINT_UNDEFINED, &heap);
-		ins_rec = page_cur_insert_rec_low(cur2, index,
+		ins_rec = page_cur_insert_rec_low(&cur2, index,
 						  cur1.rec, offsets, mtr);
 		if (UNIV_UNLIKELY(!ins_rec)) {
 			ib::fatal() << "Rec offset " << page_offset(rec)
 				<< ", cur1 offset " << page_offset(cur1.rec)
-				<< ", cur2 offset " << page_offset(cur2);
+				<< ", cur2 offset " << page_offset(cur2.rec);
 		}
 
 		page_cur_move_to_next(&cur1);
 		ut_ad(!(rec_get_info_bits(cur1.rec, page_is_comp(new_page))
 			& REC_INFO_MIN_REC_FLAG));
-		cur2 = ins_rec;
+		cur2.rec = ins_rec;
 	}
 
 	if (UNIV_LIKELY_NULL(heap)) {
@@ -730,7 +730,7 @@ page_copy_rec_list_start(
 	page_t*		new_page	= buf_block_get_frame(new_block);
 	page_zip_des_t*	new_page_zip	= buf_block_get_page_zip(new_block);
 	page_cur_t	cur1;
-	rec_t*		cur2;
+	page_cur_t	cur2;
 	mem_heap_t*	heap		= NULL;
 	ulint		num_moved	= 0;
 	rtr_rec_move_t*	rec_move	= NULL;
@@ -756,7 +756,7 @@ page_copy_rec_list_start(
 	page_cur_set_before_first(block, &cur1);
 	page_cur_move_to_next(&cur1);
 
-	cur2 = ret;
+	page_cur_position(ret, new_block, &cur2);
 
 	const bool is_leaf = page_rec_is_leaf(rec);
 
@@ -782,9 +782,10 @@ page_copy_rec_list_start(
 			offsets = rec_get_offsets(cur1.rec, index, offsets,
 						  is_leaf,
 						  ULINT_UNDEFINED, &heap);
-			cur2 = page_cur_insert_rec_low(cur2, index,
-						       cur1.rec, offsets, mtr);
-			ut_a(cur2);
+			cur2.rec = page_cur_insert_rec_low(&cur2, index,
+							   cur1.rec, offsets,
+							   mtr);
+			ut_a(cur2.rec);
 
 			page_cur_move_to_next(&cur1);
 			ut_ad(!(rec_get_info_bits(cur1.rec,