Perform page conversion by reorganize

page_copy_rec_list_end_no_locks(): If needed, convert the records
to the flexible format. FIXME: Introduce a separate function for that.

btr_page_reorganize_low(): Allow format conversion.
FIXME: The conversion can fail due to out of space.
Handle that as well!

btr_cur_optimistic_insert(): If conversion is needed, invoke
btr_page_reorganize().

mysql-test/main/disabled.def

@@ -22,4 +22,3 @@ innodb_bug12902967       : broken upstream
 file_contents            : MDEV-6526 these files are not installed anymore
 max_statement_time       : cannot possibly work, depends on timing
 connect-abstract         : waiting for libmariadb update
-function_defaults_innodb : MDEV-17520 WIP (returns garbage instead of NULL)

storage/innobase/btr/btr0btr.cc

@@ -1679,7 +1679,15 @@ btr_page_reorganize_low(
 	data_size2 = page_get_data_size(page);
 	max_ins_size2 = page_get_max_insert_size_after_reorganize(page, 1);
 
-	if (data_size1 != data_size2 || max_ins_size1 != max_ins_size2) {
+	if (data_size1 == data_size2 && max_ins_size1 == max_ins_size2) {
+		success = true;
+	} else if (page_is_comp(temp_page) && page_is_leaf(temp_page)
+		   && index->dual_format()) {
+		/* On format conversion, the size can differ. */
+		ut_ad(data_size2 >= data_size1);
+		ut_ad(max_ins_size2 <= max_ins_size1);
+		success = true; /* FIXME: return false on overflow */
+	} else {
 		ib::error()
 			<< "Page old data size " << data_size1
 			<< " new data size " << data_size2
@@ -1688,15 +1696,16 @@ btr_page_reorganize_low(
 
 		ib::error() << BUG_REPORT_MSG;
 		ut_ad(0);
-	} else {
-		success = true;
 	}
 
 	/* Restore the cursor position. */
 	if (pos > 0) {
 		cursor->rec = page_rec_get_nth(page, pos);
 	} else {
-		ut_ad(cursor->rec == page_get_infimum_rec(page));
+		ut_ad(cursor->rec == page_get_infimum_rec(page)
+		      || (!page_is_comp(page) && page_is_comp(temp_page)
+			  && page_is_leaf(page) && index->dual_format()));
+		cursor->rec = page_get_infimum_rec(page);
 	}
 
 #ifdef UNIV_ZIP_DEBUG
@@ -1719,6 +1728,12 @@ btr_page_reorganize_low(
 				       PAGE_NEW_INFIMUM + temp_page, 8);
 				memcpy(PAGE_NEW_SUPREMUM + page,
 				       PAGE_NEW_SUPREMUM + temp_page, 8);
+			} else if (page_is_comp(temp_page)) {
+				ut_ad(index->dual_format());
+				memcpy(PAGE_OLD_INFIMUM + page,
+				       PAGE_NEW_INFIMUM + temp_page, 8);
+				memcpy(PAGE_OLD_SUPREMUM + page,
+				       PAGE_NEW_SUPREMUM + temp_page, 8);
 			} else {
 				memcpy(PAGE_OLD_INFIMUM + page,
 				       PAGE_OLD_INFIMUM + temp_page, 8);

storage/innobase/btr/btr0cur.cc

@@ -3333,9 +3333,7 @@ btr_cur_optimistic_insert(
 	buf_block_t*	block;
 	page_t*		page;
 	rec_t*		dummy;
-	bool		reorg;
 	bool		inherit = true;
-	ulint		rec_size;
 	dberr_t		err;
 
 	ut_ad(thr || !(~flags & (BTR_NO_LOCKING_FLAG | BTR_NO_UNDO_LOG_FLAG)));
@@ -3361,31 +3359,22 @@ btr_cur_optimistic_insert(
 #endif /* UNIV_DEBUG_VALGRIND */
 
 	const bool leaf = page_is_leaf(page);
+	const rec_fmt_t format = leaf
+		? (index->dual_format() ? REC_FMT_LEAF_FLEXIBLE : REC_FMT_LEAF)
+		: REC_FMT_NODE_PTR;
 
-	if (leaf && page_is_comp(page) && index->dual_format()) {
-		/* The page must be converted into ROW_FORMAT=REDUNDANT
-		in a pessimistic operation. */
-		return DB_FAIL;
-	}
-
-	if (UNIV_UNLIKELY(entry->is_alter_metadata())) {
-		ut_ad(leaf);
-		rec_size = rec_get_converted_size(REC_FMT_LEAF,
-						  index, entry, n_ext);
-		goto convert_big_rec;
-	}
+	ut_ad(leaf || !entry->is_alter_metadata());
 
 	/* Calculate the record size when entry is converted to a record */
-	rec_size = rec_get_converted_size(leaf
-					  ? REC_FMT_LEAF : REC_FMT_NODE_PTR,
-					  index, entry, n_ext);
+	ulint rec_size = rec_get_converted_size(format, index, entry, n_ext);
 
-	if (page_zip_rec_needs_ext(rec_size, page_is_comp(page),
-				   dtuple_get_n_fields(entry), page_size)) {
+	if (UNIV_UNLIKELY(entry->is_alter_metadata())
+	    || page_zip_rec_needs_ext(rec_size, page_is_comp(page),
+				      dtuple_get_n_fields(entry), page_size)) {
 		if (!leaf || !index->is_primary()) {
 			return DB_TOO_BIG_RECORD;
 		}
-convert_big_rec:
+
 		/* The record is so big that we have to store some fields
 		externally on separate database pages */
 		big_rec_vec = dtuple_convert_big_rec(index, rec_size, NULL,
@@ -3396,7 +3385,7 @@ btr_cur_optimistic_insert(
 			return(DB_TOO_BIG_RECORD);
 		}
 
-		rec_size = rec_get_converted_size(REC_FMT_LEAF, index, entry, n_ext);
+		rec_size = rec_get_converted_size(format, index, entry, n_ext);
 	}
 
 	if (page_size.is_compressed() && page_zip_is_too_big(index, entry)) {
@@ -3467,8 +3456,12 @@ btr_cur_optimistic_insert(
 		 "insert " << index->name << " (" << index->id << ") by "
 		 << ib::hex(thr ? thr->graph->trx->id : 0)
 		 << ' ' << rec_printer(entry).str());
-	DBUG_EXECUTE_IF("do_page_reorganize",
-			btr_page_reorganize(page_cursor, index, mtr););
+
+	bool reorg = leaf && page_is_comp(page) && index->dual_format();
+	DBUG_EXECUTE_IF("do_page_reorganize", reorg = true; );
+	if (reorg) {
+		btr_page_reorganize(page_cursor, index, mtr);
+	}
 
 	/* Now, try the insert */
 	{
@@ -3512,7 +3505,11 @@ btr_cur_optimistic_insert(
 			page_cursor, entry, index, offsets, heap,
 			n_ext, mtr);
 
-		reorg = page_cursor_rec != page_cur_get_rec(page_cursor);
+		if (reorg) {
+			ut_ad(page_cursor_rec == page_cursor->rec);
+		} else {
+			reorg = page_cursor_rec != page_cursor->rec;
+		}
 	}
 
 	if (*rec) {

storage/innobase/page/page0page.cc

@@ -35,6 +35,7 @@ Created 2/2/1994 Heikki Tuuri
 #include "fut0lst.h"
 #include "btr0sea.h"
 #include "trx0sys.h"
+#include "row0row.h"
 
 /*			THE INDEX PAGE
 			==============
@@ -570,9 +571,6 @@ page_copy_rec_list_end_no_locks(
 	}
 
 	btr_assert_not_corrupted(new_block, index);
-	// FIXME: If needed, copy and convert to REC_FMT_LEAF_FLEXIBLE
-	ut_ad(page_is_comp(new_block->frame) == page_rec_is_comp(rec)
-	      || index->dual_format());
 	ut_ad(mach_read_from_2(new_block->frame + srv_page_size - 10) ==
 	      (page_is_comp(new_block->frame)
 	       ? PAGE_NEW_INFIMUM : PAGE_OLD_INFIMUM));
@@ -585,13 +583,47 @@ page_copy_rec_list_end_no_locks(
 
 	/* Copy records from the original page to the new page */
 
+	if (page_is_comp(new_block->frame) != page_rec_is_comp(rec)) {
+		DBUG_ASSERT(!page_is_comp(new_block->frame));
+		DBUG_ASSERT(index->dual_format());
+		/* FIXME: deduplicate code that is shared with
+		page_copy_rec_list_end_to_created_page() */
+		page_cur_t cursor;
+		page_cur_position(cur2, new_block, &cursor);
+		mem_heap_t* row_heap = mem_heap_create(1024);
+
+		while (!page_cur_is_after_last(&cur1)) {
+			offsets = rec_get_offsets(cur1.rec, index, offsets,
+						  format,
+						  ULINT_UNDEFINED, &heap);
+			ulint n_ext;
+			/* FIXME: skip rec_copy() here */
+			dtuple_t* dtuple = row_rec_to_index_entry(
+				cur1.rec, index, offsets, &n_ext, row_heap);
+			ut_ad(!!n_ext == rec_offs_any_extern(offsets));
+			/* TODO: ensure that we trim CHAR columns that
+			use variable-width character set encoding and that
+			we store no BLOB prefix in ROW_FORMAT=DYNAMIC */
+			cursor.rec = page_cur_tuple_insert(
+				&cursor, dtuple, index, &offsets,
+				&row_heap, n_ext, mtr);
+			if (!cursor.rec) {
+				ut_ad(!"page overflow");
+			}
+
+			mem_heap_empty(row_heap);
+			page_cur_move_to_next(&cur1);
+		}
+
+		mem_heap_free(row_heap);
+		goto func_exit;
+	}
+
 	while (!page_cur_is_after_last(&cur1)) {
-		rec_t*	cur1_rec = page_cur_get_rec(&cur1);
-		rec_t*	ins_rec;
-		offsets = rec_get_offsets(cur1_rec, index, offsets, format,
+		offsets = rec_get_offsets(cur1.rec, index, offsets, format,
 					  ULINT_UNDEFINED, &heap);
-		ins_rec = page_cur_insert_rec_low(cur2, index,
-						  cur1_rec, offsets, mtr);
+		rec_t* 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 "
@@ -603,6 +635,7 @@ page_copy_rec_list_end_no_locks(
 		cur2 = ins_rec;
 	}
 
+func_exit:
 	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
 	}