Ensure that page_copy_rec_list cannot fail

btr_page_create(): Add the parameter 'bool flexible', similar
to btr_page_empty(). In this way, btr_root_raise_and_insert()
and btr_page_split_and_insert() can avoid converting the page
when splitting it. If a page were converted to flexible format
during the split, the page split could theoretically fail
due to the growth of the record header.

Remove some bogus checks for os_has_said_disk_full, to avoid
dereferencing NULL pointers.

btr_page_split_and_insert(): If needed, convert the page to flexible
format right before inserting the record. Do not convert pages when
inserting a metadata record, because we want to be able to roll back
the transaction.

btr_can_merge_with_page(): Convert the page to flexible format if needed.

btr_lift_page_up(): Preserve the format of the page.

4 files changed, 83 insertions, 40 deletions

diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 0207d2fdab9..97fa5ee46f9 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -415,17 +415,22 @@ btr_root_adjust_on_import(
 	return(err);
 }
 
-/**************************************************************//**
-Creates a new index page (not the root, and also not
-used in page reorganization).  @see btr_page_empty(). */
+/** Create an index page (not the root, not in page reorganization).
+@see btr_page_empty().
+@param[in,out]	block		page to be created
+@param[in,out]	page_zip	compressed page frame, or NULL
+@param[in]	index		index of the page
+@param[in]	flexible	whether to invoke index->dual_format()
+@param[in]	level		B-tree level of the page (0=leaf)
+@param[in,out]	mtr		mini-transaction */
 void
 btr_page_create(
-/*============*/
-	buf_block_t*	block,	/*!< in/out: page to be created */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	dict_index_t*	index,	/*!< in: index */
-	ulint		level,	/*!< in: the B-tree level of the page */
-	mtr_t*		mtr)	/*!< in: mtr */
+	buf_block_t*	block,
+	page_zip_des_t*	page_zip,
+	dict_index_t*	index,
+	bool		flexible,
+	ulint		level,
+	mtr_t*		mtr)
 {
 	page_t*		page = buf_block_get_frame(block);
 
@@ -436,7 +441,7 @@ btr_page_create(
 		page_create_zip(block, index, level, 0, mtr);
 	} else {
 		const bool comp = index->table->not_redundant()
-			&& (level || !index->dual_format());
+			&& (!flexible || level || !index->dual_format());
 		page_create(block, mtr, comp,
 			    dict_index_is_spatial(index));
 		/* Set the level of the new index page */
@@ -2079,7 +2084,7 @@ btr_root_raise_and_insert(
 
 	new_block = btr_page_alloc(index, 0, FSP_NO_DIR, level, mtr, mtr);
 
-	if (new_block == NULL && os_has_said_disk_full) {
+	if (!new_block) {
 		return(NULL);
 	}
 
@@ -2090,7 +2095,8 @@ btr_root_raise_and_insert(
 	     || page_zip_get_size(new_page_zip)
 	     == page_zip_get_size(root_page_zip));
 
-	btr_page_create(new_block, new_page_zip, index, level, mtr);
+	btr_page_create(new_block, new_page_zip, index,
+			!page_is_comp(root), level, mtr);
 
 	/* Set the next node and previous node fields of new page */
 	btr_page_set_next(new_page, new_page_zip, FIL_NULL, mtr);
@@ -2105,6 +2111,9 @@ btr_root_raise_and_insert(
 	    || !page_copy_rec_list_end(new_block, root_block,
 				       page_get_infimum_rec(root),
 				       index, mtr)) {
+		/* Because btr_page_create() did not use the flexible format
+		(unless the root page already was in the flexible format),
+		the copying can only fail if ROW_FORMAT=COMPRESSED. */
 		ut_a(new_page_zip);
 
 		/* Copy the page byte for byte. */
@@ -3137,13 +3146,14 @@ func_start:
 	new_block = btr_page_alloc(cursor->index, hint_page_no, direction,
 				   level, mtr, mtr);
 
-	if (new_block == NULL && os_has_said_disk_full) {
+	if (!new_block) {
 		return(NULL);
 	}
 
 	new_page = buf_block_get_frame(new_block);
 	new_page_zip = buf_block_get_page_zip(new_block);
-	btr_page_create(new_block, new_page_zip, cursor->index, level, mtr);
+	btr_page_create(new_block, new_page_zip, cursor->index,
+			!page_is_comp(page), level, mtr);
 	/* Only record the leaf level page splits. */
 	if (format != REC_FMT_NODE_PTR) {
 		cursor->index->stat_defrag_n_page_split ++;
@@ -3161,12 +3171,8 @@ func_start:
 		*offsets = rec_get_offsets(split_rec, cursor->index, *offsets,
 					   format, n_uniq, heap);
 
-		if (tuple != NULL) {
-			insert_left = cmp_dtuple_rec(
-				tuple, split_rec, *offsets) < 0;
-		} else {
-			insert_left = 1;
-		}
+		insert_left = !tuple
+			|| cmp_dtuple_rec(tuple, split_rec, *offsets) < 0;
 
 		if (!insert_left && new_page_zip && n_iterations > 0) {
 			/* If a compressed page has already been split,
@@ -3183,7 +3189,7 @@ func_start:
 insert_empty:
 		ut_ad(!split_rec);
 		ut_ad(!insert_left);
-		if (format == REC_FMT_LEAF && cursor->index->dual_format()) {
+		if (format == REC_FMT_LEAF && !page_is_comp(new_page)) {
 			format = REC_FMT_LEAF_FLEXIBLE;
 		}
 		buf = UT_NEW_ARRAY_NOKEY(
@@ -3252,6 +3258,9 @@ insert_empty:
 			and then delete the records from both pages
 			as appropriate.  Deleting will always succeed. */
 			ut_a(new_page_zip);
+			/* btr_page_create() created new_block in the
+			format of the old page, so we cannot get here due
+			to overflow when converting to flexible format. */
 
 			page_zip_copy_recs(new_page_zip, new_page,
 					   page_zip, page, cursor->index, mtr);
@@ -3295,6 +3304,9 @@ insert_empty:
 			and then delete the records from both pages
 			as appropriate.  Deleting will always succeed. */
 			ut_a(new_page_zip);
+			/* btr_page_create() created new_block in the
+			format of the old page, so we cannot get here due
+			to overflow when converting to flexible format. */
 
 			page_zip_copy_recs(new_page_zip, new_page,
 					   page_zip, page, cursor->index, mtr);
@@ -3355,6 +3367,15 @@ insert_empty:
 
 	page_cur_search(insert_block, cursor->index, tuple, page_cursor);
 
+	if (page_is_comp(insert_block->frame) && cursor->index->dual_format()
+	    && !tuple->is_metadata()) {
+		if (btr_page_reorganize(page_cursor, cursor->index, mtr)) {
+			goto reorganized;
+		}
+
+		goto insert_failed;
+	}
+
 	rec = page_cur_tuple_insert(page_cursor, tuple, cursor->index,
 				    offsets, heap, n_ext, mtr);
 
@@ -3387,6 +3408,7 @@ insert_empty:
 		goto insert_failed;
 	}
 
+reorganized:
 	rec = page_cur_tuple_insert(page_cursor, tuple, cursor->index,
 				    offsets, heap, n_ext, mtr);
 
@@ -3706,10 +3728,12 @@ btr_lift_page_up(
 	btr_search_drop_page_hash_index(block);
 
 	/* Make the father empty */
-	btr_page_empty(father_block, father_page_zip, index, true,
-		       page_level, mtr);
+	btr_page_empty(father_block, father_page_zip, index,
+		       !page_is_comp(block->frame), page_level, mtr);
 	/* btr_page_empty() is supposed to zero-initialize the field. */
 	ut_ad(!page_get_instant(father_block->frame));
+	ut_ad(page_is_comp(father_block->frame)
+	      == page_is_comp(block->frame));
 
 	if (page_level == 0 && index->is_instant()) {
 		ut_ad(!father_page_zip);
@@ -3726,6 +3750,10 @@ btr_lift_page_up(
 	    || !page_copy_rec_list_end(father_block, block,
 				       page_get_infimum_rec(page),
 				       index, mtr)) {
+		/* Because btr_page_empty() did not convert the
+		father block to flexible format (unless block already
+		was in flexible format), the copying can only fail
+		if ROW_FORMAT=COMPRESSED. */
 		const page_zip_des_t*	page_zip
 			= buf_block_get_page_zip(block);
 		ut_a(father_page_zip);
@@ -4114,6 +4142,9 @@ retry:
 						   cursor->index, mtr);
 
 		if (!orig_succ) {
+			/* btr_can_merge_with_page() already ensured that
+			both blocks are in the same format. Thus, the copying
+			can only fail if ROW_FORMAT=COMPRESSED. */
 			ut_a(merge_page_zip);
 #ifdef UNIV_BTR_DEBUG
 			if (left_page_no == FIL_NULL) {
@@ -5689,6 +5720,7 @@ btr_can_merge_with_page(
 	DBUG_ENTER("btr_can_merge_with_page");
 
 	if (page_no == FIL_NULL) {
+error:
 		*merge_block = NULL;
 		DBUG_RETURN(false);
 	}
@@ -5696,6 +5728,8 @@ btr_can_merge_with_page(
 	index = btr_cur_get_index(cursor);
 	page = btr_cur_get_page(cursor);
 
+	DBUG_ASSERT(!page_is_comp(page) || !index->dual_format());
+
 	const page_id_t		page_id(index->table->space_id, page_no);
 	const page_size_t	page_size(index->table->space->flags);
 
@@ -5705,8 +5739,16 @@ btr_can_merge_with_page(
 	n_recs = page_get_n_recs(page);
 	data_size = page_get_data_size(page);
 
-	max_ins_size_reorg = page_get_max_insert_size_after_reorganize(
-		mpage, n_recs);
+	if (!page_is_comp(page) && page_is_comp(mpage)) {
+		DBUG_ASSERT(index->dual_format());
+		if (!btr_page_reorganize_block(false, 0, mblock, index, mtr)) {
+			goto error;
+		}
+		max_ins_size_reorg = page_get_max_insert_size(mpage, n_recs);
+	} else {
+		max_ins_size_reorg = page_get_max_insert_size_after_reorganize(
+			mpage, n_recs);
+	}
 
 	if (data_size > max_ins_size_reorg) {
 		goto error;
@@ -5750,8 +5792,4 @@ btr_can_merge_with_page(
 
 	*merge_block = mblock;
 	DBUG_RETURN(true);
-
-error:
-	*merge_block = NULL;
-	DBUG_RETURN(false);
 }
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 215f6336a28..1c26ac0b17a 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -3765,7 +3765,7 @@ convert_big_rec:
 			flags, cursor, offsets, heap, entry, n_ext, mtr);
 	}
 
-	if (*rec == NULL && os_has_said_disk_full) {
+	if (*rec == NULL) {
 		return(DB_OUT_OF_FILE_SPACE);
 	}
 
diff --git a/storage/innobase/gis/gis0rtree.cc b/storage/innobase/gis/gis0rtree.cc
index 5bad5012840..090764e6981 100644
--- a/storage/innobase/gis/gis0rtree.cc
+++ b/storage/innobase/gis/gis0rtree.cc
@@ -1094,7 +1094,7 @@ func_start:
 	new_block = btr_page_alloc(cursor->index, hint_page_no, FSP_UP,
 				   page_level, mtr, mtr);
 	new_page_zip = buf_block_get_page_zip(new_block);
-	btr_page_create(new_block, new_page_zip, cursor->index,
+	btr_page_create(new_block, new_page_zip, cursor->index, false,
 			page_level, mtr);
 
 	new_page = buf_block_get_frame(new_block);
diff --git a/storage/innobase/include/btr0btr.h b/storage/innobase/include/btr0btr.h
index 7baa1762b1f..3ea4e138bb7 100644
--- a/storage/innobase/include/btr0btr.h
+++ b/storage/innobase/include/btr0btr.h
@@ -717,17 +717,22 @@ btr_page_empty(
 	ulint		level,
 	mtr_t*		mtr)
 	MY_ATTRIBUTE((nonnull(1, 3, 6)));
-/**************************************************************//**
-Creates a new index page (not the root, and also not
-used in page reorganization).  @see btr_page_empty(). */
+/** Create an index page (not the root, not in page reorganization).
+@see btr_page_empty().
+@param[in,out]	block		page to be created
+@param[in,out]	page_zip	compressed page frame, or NULL
+@param[in]	index		index of the page
+@param[in]	flexible	whether to invoke index->dual_format()
+@param[in]	level		B-tree level of the page (0=leaf)
+@param[in,out]	mtr		mini-transaction */
 void
 btr_page_create(
-/*============*/
-	buf_block_t*	block,	/*!< in/out: page to be created */
-	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
-	dict_index_t*	index,	/*!< in: index */
-	ulint		level,	/*!< in: the B-tree level of the page */
-	mtr_t*		mtr);	/*!< in: mtr */
+	buf_block_t*	block,
+	page_zip_des_t*	page_zip,
+	dict_index_t*	index,
+	bool		flexible,
+	ulint		level,
+	mtr_t*		mtr);
 /**************************************************************//**
 Frees a file page used in an index tree. Can be used also to BLOB
 external storage pages. */