MDEV-12353: Replace MLOG_*_END_COPY_CREATED

Instead of writing the high-level redo log records
MLOG_LIST_END_COPY_CREATED, MLOG_COMP_LIST_END_COPY_CREATED
write log for each individual insert of a record.

page_copy_rec_list_end_to_created_page(): Remove.

This will improve the fill factor of some pages.
Adjust some tests accordingly.

PageBulk::init(), PageBulk::finish(): Avoid setting bogus limits
to PAGE_HEAP_TOP and PAGE_N_DIR_SLOTS. Avoid accessor functions
that would enforce these limits before the correct ones are set
at the end of PageBulk::finish().

mysql-test/suite/encryption/r/innodb-discard-import-change.result

@@ -100,5 +100,5 @@ NOT FOUND /verysecretmessage/ in t3.ibd
 # t4 page compressed and encrypted expecting NOT FOUND
 NOT FOUND /verysecretmessage/ in t4.ibd
 # t5 normal expecting FOUND
-FOUND 289 /verysecretmessage/ in t5.ibd
+FOUND 256 /verysecretmessage/ in t5.ibd
 DROP TABLE t1,t2,t3,t4,t5,t6;

mysql-test/suite/innodb/r/index_tree_operation.result

@@ -16,9 +16,10 @@ information_schema.innodb_buffer_page s2
 WHERE s1.space = s2.space AND name = 'test/t1'
 AND page_type = "INDEX" ORDER BY page_number;
 page_number	number_records
-3	2
+3	3
 4	3
-5	3
+5	1
+6	2
 INSERT INTO t1 VALUES (999, REPEAT('a', 4096));
 SELECT page_number, number_records
 FROM information_schema.innodb_sys_tablespaces s1,
@@ -28,8 +29,8 @@ AND page_type = "INDEX" ORDER BY page_number;
 page_number	number_records
 3	3
 4	3
-5	3
-6	1
+5	1
+6	3
 INSERT INTO t1 VALUES (998, REPEAT('a', 4096));
 SELECT page_number, number_records
 FROM information_schema.innodb_sys_tablespaces s1,
@@ -37,10 +38,11 @@ information_schema.innodb_buffer_page s2
 WHERE s1.space = s2.space AND name = 'test/t1'
 AND page_type = "INDEX" ORDER BY page_number;
 page_number	number_records
-3	3
+3	4
 4	3
-5	3
-6	2
+5	1
+6	3
+7	1
 INSERT INTO t1 VALUES (997, REPEAT('a', 4096));
 SELECT page_number, number_records
 FROM information_schema.innodb_sys_tablespaces s1,
@@ -48,8 +50,9 @@ information_schema.innodb_buffer_page s2
 WHERE s1.space = s2.space AND name = 'test/t1'
 AND page_type = "INDEX" ORDER BY page_number;
 page_number	number_records
-3	3
+3	4
 4	3
-5	3
+5	1
 6	3
+7	2
 DROP TABLE t1;

mysql-test/suite/innodb/r/innodb-change-buffer-recovery.result

@@ -13,6 +13,7 @@ c INT,
 INDEX(b))
 ENGINE=InnoDB STATS_PERSISTENT=0;
 SET GLOBAL innodb_change_buffering_debug = 1;
+SET GLOBAL innodb_limit_optimistic_insert_debug=700;
 INSERT INTO t1 SELECT 0,'x',1 FROM seq_1_to_8192;
 BEGIN;
 SELECT b FROM t1 LIMIT 3;
@@ -50,5 +51,5 @@ Table	Op	Msg_type	Msg_text
 test.t1	check	status	OK
 SHOW ENGINE INNODB STATUS;
 Type	Name	Status
-InnoDB		insert 79, delete mark 1
+InnoDB		insert 139, delete mark 1
 DROP TABLE t1;

mysql-test/suite/innodb/r/innodb-page_compression_default.result

@@ -36,7 +36,7 @@ select count(*) from innodb_page_compressed9;
 count(*)
 10000
 # innodb_normal expected FOUND
-FOUND 24084 /AaAaAaAa/ in innodb_normal.ibd
+FOUND 24000 /AaAaAaAa/ in innodb_normal.ibd
 # innodb_page_compressed1 page compressed expected NOT FOUND
 NOT FOUND /AaAaAaAa/ in innodb_page_compressed1.ibd
 # innodb_page_compressed2 page compressed expected NOT FOUND

mysql-test/suite/innodb/r/innodb-page_compression_snappy.result

@@ -37,7 +37,7 @@ select count(*) from innodb_page_compressed9;
 count(*)
 10000
 # innodb_normal expected FOUND
-FOUND 24084 /AaAaAaAa/ in innodb_normal.ibd
+FOUND 24000 /AaAaAaAa/ in innodb_normal.ibd
 # innodb_page_compressed1 page compressed expected NOT FOUND
 NOT FOUND /AaAaAaAa/ in innodb_page_compressed1.ibd
 # innodb_page_compressed2 page compressed expected NOT FOUND

mysql-test/suite/innodb/r/innodb_defragment_fill_factor.result

@@ -115,4 +115,5 @@ insert into t2 values (201, REPEAT('A', 16), REPEAT('B', 32));
 insert into t2 values (202, REPEAT('A', 16), REPEAT('B', 32));
 insert into t2 values (203, REPEAT('A', 16), REPEAT('B', 32));
 insert into t2 values (204, REPEAT('A', 16), REPEAT('B', 32));
+Too little space is reserved on second index.
 DROP TABLE t2;

mysql-test/suite/innodb/r/instant_alter,32k.rdiff

@@ -44,7 +44,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -7
-+5
++4
  connection default;
  ROLLBACK;
  CHECKSUM TABLE big;
@@ -71,7 +71,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -7
-+5
++4
  connection default;
  ROLLBACK;
  CHECKSUM TABLE big;
@@ -98,7 +98,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -7
-+5
++4
  connection default;
  ROLLBACK;
  CHECKSUM TABLE big;

mysql-test/suite/innodb/r/instant_alter,4k.rdiff

@@ -117,7 +117,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -7
-+28
++27
  connection default;
  ROLLBACK;
  CHECKSUM TABLE big;

mysql-test/suite/innodb/r/instant_alter,8k.rdiff

@@ -41,7 +41,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -3
-+5
++4
  connection default;
  ALTER TABLE big ADD COLUMN
  (d1 INT DEFAULT 0, d2 VARCHAR(20) DEFAULT 'abcde',
@@ -50,7 +50,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -7
-+13
++12
  connection default;
  ROLLBACK;
  CHECKSUM TABLE big;
@@ -59,7 +59,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -3
-+5
++4
  connection default;
  InnoDB		0 transactions not purged
  DROP TABLE t1,t2,t3,big;
@@ -104,7 +104,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -3
-+5
++4
  connection default;
  ALTER TABLE big ADD COLUMN
  (d1 INT DEFAULT 0, d2 VARCHAR(20) DEFAULT 'abcde',
@@ -113,7 +113,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -7
-+13
++12
  connection default;
  ROLLBACK;
  CHECKSUM TABLE big;
@@ -122,7 +122,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -3
-+5
++4
  connection default;
  InnoDB		0 transactions not purged
  DROP TABLE t1,t2,t3,big;
@@ -167,7 +167,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -3
-+5
++4
  connection default;
  ALTER TABLE big ADD COLUMN
  (d1 INT DEFAULT 0, d2 VARCHAR(20) DEFAULT 'abcde',
@@ -176,7 +176,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -7
-+13
++12
  connection default;
  ROLLBACK;
  CHECKSUM TABLE big;
@@ -185,7 +185,7 @@
  WHERE name = 'test/big';
  clust_index_size
 -3
-+5
++4
  connection default;
  InnoDB		0 transactions not purged
  DROP TABLE t1,t2,t3,big;

mysql-test/suite/innodb/r/leaf_page_corrupted_during_recovery.result

@@ -4,17 +4,23 @@ INSERT INTO t1 VALUES(1, 'sql'), (2, 'server'), (3, 'mariadb'),
 (4, 'mariadb'), (5, 'test1'), (6, 'test2'), (7, 'test3'),
 (8, 'test4'), (9, 'test5'), (10, 'test6'), (11, 'test7'),
 (12, 'test8');
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+SET GLOBAL innodb_log_checkpoint_now=ON;
 SELECT COUNT(*) FROM t1;
 COUNT(*)
 12
 UPDATE t1 SET c='best8' WHERE pk=12;
 # Kill the server
 # Corrupt the pages
+# restart
 SELECT * FROM t1 WHERE PK = 1;
 ERROR 42000: Unknown storage engine 'InnoDB'
+# restart: --innodb-force-recovery=1
 SELECT * FROM t1 WHERE PK = 1;
 pk	c
 1	sql
 SELECT * FROM t1 WHERE pk = 12;
-ERROR HY000: Index for table 't1' is corrupt; try to repair it
+ERROR HY000: Got error 168 "Unknown (generic) error from engine" from storage engine InnoDB
 DROP TABLE t1;
+# restart

mysql-test/suite/innodb/t/innodb-change-buffer-recovery.test

@@ -35,9 +35,16 @@ ENGINE=InnoDB STATS_PERSISTENT=0;
 SET GLOBAL innodb_change_buffering_debug = 1;
 let SEARCH_FILE = $MYSQLTEST_VARDIR/log/mysqld.1.err;
 
+# The removed function page_copy_rec_list_end_to_created_page() would create
+# less dense pages than the remaining part of page_copy_rec_list_end().
+# For the INSERT after the DELETE to be buffered, the leftmost page
+# must not be too densely packed.
+SET GLOBAL innodb_limit_optimistic_insert_debug=700;
+
 # Create enough rows for the table, so that the change buffer will be
 # used for modifying the secondary index page. There must be multiple
 # index pages, because changes to the root page are never buffered.
+
 INSERT INTO t1 SELECT 0,'x',1 FROM seq_1_to_8192;
 
 BEGIN;

mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test

@@ -3,8 +3,8 @@
 
 --disable_query_log
 call mtr.add_suppression("InnoDB: Database page corruption on disk or a failed file read of tablespace test/t1 page ");
-call mtr.add_suppression("InnoDB: Background Page read failed to read or decrypt \\[page id: space=\\d+, page number=19\\]");
-call mtr.add_suppression("\\[ERROR\\] InnoDB: Failed to read file '.*test.t1\\.ibd' at offset 19: Page read from tablespace is corrupted\\.");
+call mtr.add_suppression("InnoDB: Background Page read failed to read or decrypt \\[page id: space=\\d+, page number=14\\]");
+call mtr.add_suppression("\\[ERROR\\] InnoDB: Failed to read file '.*test.t1\\.ibd' at offset 14: Page read from tablespace is corrupted\\.");
 call mtr.add_suppression("\\[ERROR\\] InnoDB: Plugin initialization aborted at srv0start\\.cc.* with error Data structure corruption");
 call mtr.add_suppression("\\[ERROR\\] Plugin 'InnoDB' (init function|registration)");
 call mtr.add_suppression("\\[ERROR\\] InnoDB: We detected index corruption");
@@ -20,8 +20,12 @@ INSERT INTO t1 VALUES(1, 'sql'), (2, 'server'), (3, 'mariadb'),
 	(8, 'test4'), (9, 'test5'), (10, 'test6'), (11, 'test7'),
 	(12, 'test8');
 
-let $restart_noprint=2;
---source include/restart_mysqld.inc
+# Flush all pages of the table, and perform log checkpoint,
+# so that no page initialization can be replayed from the redo log
+# to undo the page corruption that we are going to inject.
+FLUSH TABLE t1 FOR EXPORT;
+UNLOCK TABLES;
+SET GLOBAL innodb_log_checkpoint_now=ON;
 
 let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
 let MYSQLD_DATADIR=`select @@datadir`;
@@ -38,7 +42,7 @@ perl;
 my $file = "$ENV{MYSQLD_DATADIR}/test/t1.ibd";
 open(FILE, "+<$file") || die "Unable to open $file";
 binmode FILE;
-seek (FILE, $ENV{INNODB_PAGE_SIZE} * 19 + 38, SEEK_SET) or die "seek";
+seek (FILE, $ENV{INNODB_PAGE_SIZE} * 14 + 38, SEEK_SET) or die "seek";
 print FILE "junk";
 close FILE or die "close";
 EOF
@@ -50,7 +54,7 @@ SELECT * FROM t1 WHERE PK = 1;
 let $restart_parameters=--innodb-force-recovery=1;
 --source include/restart_mysqld.inc
 SELECT * FROM t1 WHERE PK = 1;
---error ER_NOT_KEYFILE
+--error ER_GET_ERRNO
 SELECT * FROM t1 WHERE pk = 12;
 
 DROP TABLE t1;

mysql-test/suite/innodb_zip/r/wl6347_comp_indx_stat.result

@@ -962,7 +962,7 @@ AND compress_ops BETWEEN @inl_val AND 1000
 AND table_name='tab5' AND database_name='test'
 AND index_name like 'idx%' ;
 compress_stat	1
-The size of the tab5.ibd file: 163840
+The size of the tab5.ibd file: 159744
 # fetch the compressed page and check the stats
 ===============
 Fetch Records
@@ -986,7 +986,7 @@ AND compress_ops BETWEEN @inl_val AND 1000
 AND table_name='tab5' AND database_name='test'
 AND index_name like 'idx%' ;
 compress_stat	1
-The size of the tab5.ibd file: 163840
+The size of the tab5.ibd file: 159744
 # fetch the compressed same page once again and check the stats
 # the stat figures should be same as above query
 ===============
@@ -1011,7 +1011,7 @@ AND compress_ops BETWEEN @inl_val AND 1000
 AND table_name='tab5' AND database_name='test'
 AND index_name like 'idx%' ;
 compress_stat	1
-The size of the tab5.ibd file: 163840
+The size of the tab5.ibd file: 159744
 DROP TABLE tab5;
 #******************************************************************
 # Test 1-8K: innodb_cmp_per_index_enabled=ON and innodb_compression_level=0 with page size 8K

storage/innobase/btr/btr0bulk.cc

@@ -98,7 +98,7 @@ PageBulk::init()
 			page_zip_write_header(new_page_zip, index_id,
 					      8, &m_mtr);
 		} else {
-			ut_ad(!dict_index_is_spatial(m_index));
+			ut_ad(!m_index->is_spatial());
 			page_create(new_block, &m_mtr,
 				    m_index->table->not_redundant(),
 				    false);
@@ -153,9 +153,6 @@ PageBulk::init()
 	m_rec_no = page_header_get_field(new_page, PAGE_N_RECS);
 
 	ut_d(m_total_data = 0);
-	/* See page_copy_rec_list_end_to_created_page() */
-	ut_d(page_header_set_field(m_page, NULL, PAGE_HEAP_TOP,
-				   srv_page_size - 1));
 
 	return(DB_SUCCESS);
 }
@@ -178,7 +175,7 @@ inline void PageBulk::insertPage(const rec_t *rec, offset_t *offsets)
 
 #ifdef UNIV_DEBUG
 	/* Check whether records are in order. */
-	if (!page_rec_is_infimum(m_cur_rec)) {
+	if (!page_rec_is_infimum_low(page_offset(m_cur_rec))) {
 		rec_t*	old_rec = m_cur_rec;
 		offset_t* old_offsets = rec_get_offsets(
 			old_rec, m_index, NULL,	is_leaf,
@@ -193,21 +190,26 @@ inline void PageBulk::insertPage(const rec_t *rec, offset_t *offsets)
 
 	/* 1. Copy the record to page. */
 	rec_t*	insert_rec = rec_copy(m_heap_top, rec, offsets);
+	ut_ad(page_align(insert_rec) == m_page);
 	rec_offs_make_valid(insert_rec, m_index, is_leaf, offsets);
 
 	/* 2. Insert the record in the linked list. */
-	rec_t*	next_rec = page_rec_get_next(m_cur_rec);
-
-	page_rec_set_next(insert_rec, next_rec);
-	page_rec_set_next(m_cur_rec, insert_rec);
-
-	/* 3. Set the n_owned field in the inserted record to zero,
-	and set the heap_no field. */
 	if (fmt != REDUNDANT) {
+		rec_t* next_rec = m_page
+			+ page_offset(m_cur_rec
+				      + mach_read_from_2(m_cur_rec
+							 - REC_NEXT));
+		mach_write_to_2(insert_rec - REC_NEXT,
+				static_cast<uint16_t>(next_rec - insert_rec));
+		mach_write_to_2(m_cur_rec - REC_NEXT,
+				static_cast<uint16_t>(insert_rec - m_cur_rec));
 		rec_set_n_owned_new(insert_rec, NULL, 0);
 		rec_set_heap_no_new(insert_rec,
 				    PAGE_HEAP_NO_USER_LOW + m_rec_no);
 	} else {
+		mach_write_to_2(insert_rec - REC_NEXT,
+				mach_read_from_2(m_cur_rec - REC_NEXT));
+		mach_write_to_2(m_cur_rec - REC_NEXT, page_offset(insert_rec));
 		rec_set_n_owned_old(insert_rec, 0);
 		rec_set_heap_no_old(insert_rec,
 				    PAGE_HEAP_NO_USER_LOW + m_rec_no);
@@ -225,7 +227,7 @@ inline void PageBulk::insertPage(const rec_t *rec, offset_t *offsets)
 	m_heap_top += rec_size;
 	m_rec_no += 1;
 
-	if (!m_page_zip) {
+	if (fmt != COMPRESSED) {
 		/* For ROW_FORMAT=COMPRESSED, redo log may be written
 		in PageBulk::compress(). */
 		page_cur_insert_rec_write_log(insert_rec, rec_size,
@@ -250,110 +252,137 @@ inline void PageBulk::insert(const rec_t *rec, offset_t *offsets)
 
 /** Mark end of insertion to the page. Scan all records to set page dirs,
 and set page header members.
-@tparam fmt     the page format */
+@tparam fmt  page format */
 template<PageBulk::format fmt>
 inline void PageBulk::finishPage()
 {
   ut_ad(m_rec_no > 0);
   ut_ad((m_page_zip != nullptr) == (fmt == COMPRESSED));
   ut_ad((fmt != REDUNDANT) == m_is_comp);
-	ut_ad(m_total_data + page_dir_calc_reserved_space(m_rec_no)
-	      <= page_get_free_space_of_empty(fmt != REDUNDANT));
-	/* See page_copy_rec_list_end_to_created_page() */
-	ut_d(page_dir_set_n_slots(m_page, NULL, srv_page_size / 2));
-
-	ulint	count = 0;
-	ulint	n_recs = 0;
-	ulint	slot_index = 0;
-	rec_t*	insert_rec = page_rec_get_next(page_get_infimum_rec(m_page));
-	page_dir_slot_t* slot = NULL;
 
+  ulint count= 0;
+  ulint n_recs= 0;
+  byte *slot= my_assume_aligned<2>(m_page + srv_page_size -
+                                   (PAGE_DIR + PAGE_DIR_SLOT_SIZE));
+  const page_dir_slot_t *const slot0 = slot;
+  compile_time_assert(PAGE_DIR_SLOT_SIZE == 2);
+  if (fmt != REDUNDANT)
+  {
+    uint16_t offset= mach_read_from_2(PAGE_NEW_INFIMUM - REC_NEXT + m_page);
+    ut_ad(offset >= PAGE_NEW_SUPREMUM - PAGE_NEW_INFIMUM);
+    offset += PAGE_NEW_INFIMUM;
     /* Set owner & dir. */
-	do {
-
+    do
+    {
+      ut_ad(offset >= PAGE_NEW_SUPREMUM);
+      ut_ad(offset < page_offset(slot));
       count++;
       n_recs++;
 
-		if (count == (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2) {
+      if (count == (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2)
+      {
+        slot-= PAGE_DIR_SLOT_SIZE;
+        mach_write_to_2(slot, offset);
+        rec_set_n_owned_new(m_page + offset, nullptr, count);
+        count = 0;
+      }
 
-			slot_index++;
+      uint16_t next= (mach_read_from_2(m_page + offset - REC_NEXT) + offset) &
+        (srv_page_size - 1);
+      ut_ad(next);
+      offset= next;
+    }
+    while (offset != PAGE_NEW_SUPREMUM);
+  }
+  else
+  {
+    rec_t *insert_rec= m_page +
+      mach_read_from_2(PAGE_OLD_INFIMUM - REC_NEXT + m_page);
 
-			slot = page_dir_get_nth_slot(m_page, slot_index);
+    /* Set owner & dir. */
+    do
+    {
+      count++;
+      n_recs++;
 
-			page_dir_slot_set_rec(slot, insert_rec);
-			page_dir_slot_set_n_owned(slot, NULL, count);
+      if (count == (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2)
+      {
+        slot-= PAGE_DIR_SLOT_SIZE;
+        mach_write_to_2(slot, page_offset(insert_rec));
+        rec_set_n_owned_old(insert_rec, count);
 
-			count = 0;
+        count= 0;
       }
 
-		insert_rec = page_rec_get_next(insert_rec);
-	} while (!page_rec_is_supremum(insert_rec));
-
-	if (slot_index > 0
-	    && (count + 1 + (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2
-		<= PAGE_DIR_SLOT_MAX_N_OWNED)) {
-		/* We can merge the two last dir slots. This operation is
-		here to make this function imitate exactly the equivalent
-		task made using page_cur_insert_rec, which we use in database
-		recovery to reproduce the task performed by this function.
-		To be able to check the correctness of recovery, it is good
-		that it imitates exactly. */
+      insert_rec= m_page + mach_read_from_2(insert_rec - REC_NEXT);
+    }
+    while (insert_rec != m_page + PAGE_OLD_SUPREMUM);
+  }
 
-		count += (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2;
+  if (slot0 != slot && (count + 1 + (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2 <=
+                        PAGE_DIR_SLOT_MAX_N_OWNED)) {
+    /* We can merge the two last dir slots. This operation is here to
+    make this function imitate exactly the equivalent task made using
+    page_cur_insert_rec(), which we use in database recovery to
+    reproduce the task performed by this function. To be able to
+    check the correctness of recovery, it is good that it imitates exactly. */
 
-		page_dir_slot_set_n_owned(slot, NULL, 0);
+    count+= (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2;
 
-		slot_index--;
+    page_dir_slot_set_n_owned(slot, nullptr, 0);
+    slot+= PAGE_DIR_SLOT_SIZE;
   }
 
-	slot = page_dir_get_nth_slot(m_page, 1 + slot_index);
+  slot-= PAGE_DIR_SLOT_SIZE;
   page_dir_slot_set_rec(slot, page_get_supremum_rec(m_page));
-	page_dir_slot_set_n_owned(slot, NULL, count + 1);
+  page_dir_slot_set_n_owned(slot, nullptr, count + 1);
 
   ut_ad(!dict_index_is_spatial(m_index));
   ut_ad(!page_get_instant(m_page));
   ut_ad(!mach_read_from_2(PAGE_HEADER + PAGE_N_DIRECTION + m_page));
 
-	if (fmt != COMPRESSED) {
+  if (fmt != COMPRESSED)
+  {
     m_mtr.write<2,mtr_t::OPT>(*m_block,
-					  PAGE_HEADER + PAGE_N_DIR_SLOTS
-					  + m_page, 2 + slot_index);
+                              PAGE_HEADER + PAGE_N_DIR_SLOTS + m_page,
+                              1 + static_cast<ulint>(slot0 - slot) /
+                              PAGE_DIR_SLOT_SIZE);
     m_mtr.write<2>(*m_block, PAGE_HEADER + PAGE_HEAP_TOP + m_page,
-			       ulint(m_heap_top - m_page));
-		m_mtr.write<2>(*m_block,
-			       PAGE_HEADER + PAGE_N_HEAP + m_page,
-			       (PAGE_HEAP_NO_USER_LOW + m_rec_no)
-			       | uint16_t{fmt != REDUNDANT} << 15);
-		m_mtr.write<2>(*m_block,
-			       PAGE_HEADER + PAGE_N_RECS + m_page, m_rec_no);
-		m_mtr.write<2>(*m_block,
-			       PAGE_HEADER + PAGE_LAST_INSERT + m_page,
-			       ulint(m_cur_rec - m_page));
-		m_mtr.write<2>(*m_block,
-			       PAGE_HEADER + PAGE_DIRECTION_B - 1 + m_page,
+                   static_cast<ulint>(m_heap_top - m_page));
+    m_mtr.write<2>(*m_block, PAGE_HEADER + PAGE_N_HEAP + m_page,
+                   (PAGE_HEAP_NO_USER_LOW + m_rec_no) |
+                   uint16_t{fmt != REDUNDANT} << 15);
+    m_mtr.write<2>(*m_block, PAGE_HEADER + PAGE_N_RECS + m_page, m_rec_no);
+    m_mtr.write<2>(*m_block, PAGE_HEADER + PAGE_LAST_INSERT + m_page,
+                   static_cast<ulint>(m_cur_rec - m_page));
+    m_mtr.write<2>(*m_block, PAGE_HEADER + PAGE_DIRECTION_B - 1 + m_page,
                    PAGE_RIGHT);
-	} else {
-		/* For ROW_FORMAT=COMPRESSED, redo log may be written
-		in PageBulk::compress(). */
+  }
+  else
+  {
+    /* For ROW_FORMAT=COMPRESSED, redo log may be written in
+    PageBulk::compress(). */
     mach_write_to_2(PAGE_HEADER + PAGE_N_DIR_SLOTS + m_page,
-				2 + slot_index);
+                    1 + (slot0 - slot) / PAGE_DIR_SLOT_SIZE);
     mach_write_to_2(PAGE_HEADER + PAGE_HEAP_TOP + m_page,
-				ulint(m_heap_top - m_page));
+                    static_cast<ulint>(m_heap_top - m_page));
     mach_write_to_2(PAGE_HEADER + PAGE_N_HEAP + m_page,
-				(PAGE_HEAP_NO_USER_LOW + m_rec_no)
-				| uint16_t{fmt != REDUNDANT} << 15);
+                    (PAGE_HEAP_NO_USER_LOW + m_rec_no) |
+                    uint16_t{fmt != REDUNDANT} << 15);
     mach_write_to_2(PAGE_HEADER + PAGE_N_RECS + m_page, m_rec_no);
     mach_write_to_2(PAGE_HEADER + PAGE_LAST_INSERT + m_page,
-				ulint(m_cur_rec - m_page));
-		mach_write_to_2(PAGE_HEADER + PAGE_DIRECTION_B - 1 + m_page,
-				PAGE_RIGHT);
+                    static_cast<ulint>(m_cur_rec - m_page));
+    mach_write_to_2(PAGE_HEADER + PAGE_DIRECTION_B - 1 + m_page, PAGE_RIGHT);
   }
 
-	m_block->skip_flush_check = false;
+  ut_ad(m_total_data + page_dir_calc_reserved_space(m_rec_no) <=
+        page_get_free_space_of_empty(m_is_comp));
+  m_block->skip_flush_check= false;
 }
 
 /** Mark end of insertion to the page. Scan all records to set page dirs,
-and set page header members. */
+and set page header members.
+@tparam compressed  whether the page is in ROW_FORMAT=COMPRESSED */
 inline void PageBulk::finish()
 {
   if (UNIV_LIKELY_NULL(m_page_zip))

storage/innobase/include/page0cur.h

@@ -217,6 +217,7 @@ IMPORTANT: The caller will have to update IBUF_BITMAP_FREE
 if this 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(). */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 void
 page_copy_rec_list_end_to_created_page(
 /*===================================*/

storage/innobase/page/page0cur.cc

@@ -2041,6 +2041,7 @@ IMPORTANT: The caller will have to update IBUF_BITMAP_FREE
 if this 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(). */
+ATTRIBUTE_COLD /* only used when crash-upgrading */
 void
 page_copy_rec_list_end_to_created_page(
 /*===================================*/

storage/innobase/page/page0page.cc

@@ -599,24 +599,18 @@ page_copy_rec_list_end(
 	/* Here, "ret" may be pointing to a user record or the
 	predefined supremum record. */
 
-	mtr_log_t	log_mode = MTR_LOG_NONE;
+	const mtr_log_t log_mode = new_page_zip
+		? mtr->set_log_mode(MTR_LOG_NONE) : MTR_LOG_NONE;
+	ut_d(const bool was_empty = page_dir_get_n_heap(new_page)
+	     == PAGE_HEAP_NO_USER_LOW);
 
-	if (new_page_zip) {
-		log_mode = mtr_set_log_mode(mtr, MTR_LOG_NONE);
-	}
-
-	if (page_dir_get_n_heap(new_page) == PAGE_HEAP_NO_USER_LOW) {
-		page_copy_rec_list_end_to_created_page(new_page, rec,
-						       index, mtr);
-	} else {
-		if (dict_index_is_spatial(index)) {
+	if (index->is_spatial()) {
 		ulint	max_to_move = page_get_n_recs(
 			buf_block_get_frame(block));
 		heap = mem_heap_create(256);
 
-			rec_move = static_cast<rtr_rec_move_t*>(mem_heap_alloc(
-					heap,
-					sizeof (*rec_move) * max_to_move));
+		rec_move = static_cast<rtr_rec_move_t*>(
+			mem_heap_alloc(heap, max_to_move * sizeof *rec_move));
 
 		/* For spatial index, we need to insert recs one by one
 		to keep recs ordered. */
@@ -630,7 +624,6 @@ page_copy_rec_list_end(
 		page_copy_rec_list_end_no_locks(new_block, block, rec,
 						index, mtr);
 	}
-	}
 
 	/* Update PAGE_MAX_TRX_ID on the uncompressed page.
 	Modifications will be redo logged and copied to the compressed
@@ -642,6 +635,9 @@ page_copy_rec_list_end(
 	if (dict_index_is_sec_or_ibuf(index)
 	    && page_is_leaf(page)
 	    && !index->table->is_temporary()) {
+		ut_ad(!was_empty || page_dir_get_n_heap(new_page)
+		      == PAGE_HEAP_NO_USER_LOW
+		      + page_header_get_field(new_page, PAGE_N_RECS));
 		page_update_max_trx_id(new_block, NULL,
 				       page_get_max_trx_id(page), mtr);
 	}