MDEV-12353: Remove support for crash-upgradebb-10.5-MDEV-12353

We tighten some assertions regarding dict_index_t::is_dummy
and crash recovery, now that redo log processing will
no longer create dummy objects.

37 files changed, 296 insertions, 5222 deletions

diff --git a/extra/mariabackup/xtrabackup.cc b/extra/mariabackup/xtrabackup.cc
index 8bce6567233..70ddc933fe3 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -2649,12 +2649,8 @@ static lsn_t xtrabackup_copy_log(lsn_t start_lsn, lsn_t end_lsn, bool last)
 		}
 	}
 
-	store_t store = STORE_NO;
-
-	if (more_data && recv_parse_log_recs(0, &store, false)) {
-
+	if (more_data && recv_sys.parse(0, STORE_NO, false)) {
 		msg("Error: copying the log failed");
-
 		return(0);
 	}
 
diff --git a/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result b/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result
index 55c1bd718ef..2c2de7e5f88 100644
--- a/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result
+++ b/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result
@@ -56,7 +56,7 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND 1 /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122/ in mysqld.1.err
-FOUND 1 /InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
+FOUND 1 /InnoDB: Upgrade after a crash is not supported\. The redo log was created with malicious intentions, or perhaps, and it appears corrupted\./ in mysqld.1.err
 # same, but with current-version header
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=5
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
@@ -64,7 +64,6 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND 2 /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122/ in mysqld.1.err
-FOUND 2 /InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
 # --innodb-force-recovery=6 (skip the entire redo log)
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=6
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
@@ -93,12 +92,7 @@ SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /InnoDB: Starting crash recovery from checkpoint LSN=1213964/ in mysqld.1.err
-FOUND 1 /InnoDB: MLOG_FILE_NAME incorrect:bogus/ in mysqld.1.err
-FOUND 1 /InnoDB: ############### CORRUPT LOG RECORD FOUND ##################/ in mysqld.1.err
-FOUND 1 /InnoDB: Log record type 55, page 151:488\. Log parsing proceeded successfully up to 1213973\. Previous log record type 56, is multi 0 Recv offset 9, prev 0/ in mysqld.1.err
-FOUND 1 /len 22. hex 38000000000012860cb7809781e80006626f67757300. asc 8               bogus / in mysqld.1.err
-FOUND 1 /InnoDB: Set innodb_force_recovery to ignore this error/ in mysqld.1.err
+FOUND 1 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\./ in mysqld.1.err
 # Test a corrupted MLOG_FILE_NAME record.
 # valid header, invalid checkpoint 1, valid checkpoint 2, invalid block
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
@@ -107,28 +101,28 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND 1 /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 2454333373 found: 150151/ in mysqld.1.err
+FOUND 3 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1, and it appears corrupted\./ in mysqld.1.err
 # valid header, invalid checkpoint 1, valid checkpoint 2, invalid log record
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /InnoDB: MLOG_FILE_NAME incorrect:bigot/ in mysqld.1.err
-FOUND 1 /len 22; hex 38000000000012860cb7809781e800066269676f7400; asc 8               bigot ;/ in mysqld.1.err
+FOUND 2 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\./ in mysqld.1.err
 # 10.2 missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /InnoDB: Missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT for tablespace 42/ in mysqld.1.err
+FOUND 1 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with ibbackup was here!!!1!\./ in mysqld.1.err
 # 10.3 missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 2 /InnoDB: Missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT for tablespace 42/ in mysqld.1.err
+FOUND 3 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\./ in mysqld.1.err
 # Empty 10.3 redo log
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=5 --innodb-log-file-size=2m
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.ENGINES
@@ -136,7 +130,7 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 COUNT(*)
 1
-FOUND 1 /InnoDB: .* started; log sequence number 12139[78]\d; transaction id 0/ in mysqld.1.err
+FOUND 1 /InnoDB: .* started; log sequence number 1213964; transaction id 0/ in mysqld.1.err
 # Empty 10.2 redo log
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=5 --innodb-log-file-size=2m
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.ENGINES
diff --git a/mysql-test/suite/innodb/r/log_corruption.result b/mysql-test/suite/innodb/r/log_corruption.result
index ab33ea1b152..f1d625125b2 100644
--- a/mysql-test/suite/innodb/r/log_corruption.result
+++ b/mysql-test/suite/innodb/r/log_corruption.result
@@ -56,7 +56,7 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND 1 /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122/ in mysqld.1.err
-FOUND 1 /InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
+FOUND 1 /InnoDB: Upgrade after a crash is not supported\. The redo log was created with malicious intentions, or perhaps, and it appears corrupted\./ in mysqld.1.err
 # same, but with current-version header
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=5
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
@@ -64,7 +64,6 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND 2 /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122/ in mysqld.1.err
-FOUND 2 /InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
 # --innodb-force-recovery=6 (skip the entire redo log)
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=6
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
@@ -93,12 +92,7 @@ SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /InnoDB: Starting crash recovery from checkpoint LSN=1213964/ in mysqld.1.err
-FOUND 1 /InnoDB: MLOG_FILE_NAME incorrect:bogus/ in mysqld.1.err
-FOUND 1 /InnoDB: ############### CORRUPT LOG RECORD FOUND ##################/ in mysqld.1.err
-FOUND 1 /InnoDB: Log record type 55, page 151:488\. Log parsing proceeded successfully up to 1213973\. Previous log record type 56, is multi 0 Recv offset 9, prev 0/ in mysqld.1.err
-FOUND 1 /len 22. hex 38000000000012860cb7809781e80006626f67757300. asc 8               bogus / in mysqld.1.err
-FOUND 1 /InnoDB: Set innodb_force_recovery to ignore this error/ in mysqld.1.err
+FOUND 1 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\./ in mysqld.1.err
 # Test a corrupted MLOG_FILE_NAME record.
 # valid header, invalid checkpoint 1, valid checkpoint 2, invalid block
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
@@ -107,28 +101,28 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND 1 /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 2454333373 found: 150151/ in mysqld.1.err
+FOUND 3 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1, and it appears corrupted\./ in mysqld.1.err
 # valid header, invalid checkpoint 1, valid checkpoint 2, invalid log record
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /InnoDB: MLOG_FILE_NAME incorrect:bigot/ in mysqld.1.err
-FOUND 1 /len 22; hex 38000000000012860cb7809781e800066269676f7400; asc 8               bigot ;/ in mysqld.1.err
+FOUND 2 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\./ in mysqld.1.err
 # 10.2 missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /InnoDB: Missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT for tablespace 42/ in mysqld.1.err
+FOUND 1 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with ibbackup was here!!!1!\./ in mysqld.1.err
 # 10.3 missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 2 /InnoDB: Missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT for tablespace 42/ in mysqld.1.err
+FOUND 3 /\[ERROR\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\./ in mysqld.1.err
 # Empty 10.3 redo log
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=5 --innodb-log-file-size=2m
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.ENGINES
@@ -136,7 +130,7 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 COUNT(*)
 1
-FOUND 1 /InnoDB: .* started; log sequence number 12139[78]\d; transaction id 0/ in mysqld.1.err
+FOUND 1 /InnoDB: .* started; log sequence number 1213964; transaction id 0/ in mysqld.1.err
 # Empty 10.2 redo log
 # restart: --innodb-data-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-log-group-home-dir=MYSQLTEST_VARDIR/tmp/log_corruption --innodb-force-recovery=5 --innodb-log-file-size=2m
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.ENGINES
diff --git a/mysql-test/suite/innodb/t/log_corruption.test b/mysql-test/suite/innodb/t/log_corruption.test
index 6c2ef5db0bb..ea6d31b958d 100644
--- a/mysql-test/suite/innodb/t/log_corruption.test
+++ b/mysql-test/suite/innodb/t/log_corruption.test
@@ -230,7 +230,7 @@ eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
 let SEARCH_PATTERN=InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122;
 --source include/search_pattern_in_file.inc
-let SEARCH_PATTERN=InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\.;
+let SEARCH_PATTERN=InnoDB: Upgrade after a crash is not supported\. The redo log was created with malicious intentions, or perhaps, and it appears corrupted\.;
 --source include/search_pattern_in_file.inc
 
 --echo # same, but with current-version header
@@ -247,8 +247,6 @@ eval $check_no_innodb;
 
 let SEARCH_PATTERN=InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122;
 --source include/search_pattern_in_file.inc
-let SEARCH_PATTERN=InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\.;
---source include/search_pattern_in_file.inc
 --echo # --innodb-force-recovery=6 (skip the entire redo log)
 --let $restart_parameters= $dirs --innodb-force-recovery=6
 --source include/start_mysqld.inc
@@ -302,17 +300,7 @@ EOF
 --source include/start_mysqld.inc
 eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
-let SEARCH_PATTERN=InnoDB: Starting crash recovery from checkpoint LSN=1213964;
---source include/search_pattern_in_file.inc
-let SEARCH_PATTERN=InnoDB: MLOG_FILE_NAME incorrect:bogus;
---source include/search_pattern_in_file.inc
-let SEARCH_PATTERN=InnoDB: ############### CORRUPT LOG RECORD FOUND ##################;
---source include/search_pattern_in_file.inc
-let SEARCH_PATTERN=InnoDB: Log record type 55, page 151:488\. Log parsing proceeded successfully up to 1213973\. Previous log record type 56, is multi 0 Recv offset 9, prev 0;
---source include/search_pattern_in_file.inc
-let SEARCH_PATTERN= len 22. hex 38000000000012860cb7809781e80006626f67757300. asc 8               bogus ;
---source include/search_pattern_in_file.inc
-let SEARCH_PATTERN=InnoDB: Set innodb_force_recovery to ignore this error;
+let SEARCH_PATTERN=\\[ERROR\\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\.;
 --source include/search_pattern_in_file.inc
 
 --echo # Test a corrupted MLOG_FILE_NAME record.
@@ -343,6 +331,8 @@ eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
 let SEARCH_PATTERN=InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 2454333373 found: 150151;
 --source include/search_pattern_in_file.inc
+let SEARCH_PATTERN=\\[ERROR\\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1, and it appears corrupted\.;
+--source include/search_pattern_in_file.inc
 --echo # valid header, invalid checkpoint 1, valid checkpoint 2, invalid log record
 perl;
 die unless open OUT, "+<", "$ENV{bugdir}/ib_logfile0";
@@ -355,9 +345,7 @@ EOF
 --source include/start_mysqld.inc
 eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
-let SEARCH_PATTERN=InnoDB: MLOG_FILE_NAME incorrect:bigot;
---source include/search_pattern_in_file.inc
---let SEARCH_PATTERN= len 22; hex 38000000000012860cb7809781e800066269676f7400; asc 8               bigot ;
+let SEARCH_PATTERN=\\[ERROR\\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\.;
 --source include/search_pattern_in_file.inc
 
 --echo # 10.2 missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT
@@ -388,7 +376,7 @@ EOF
 --source include/start_mysqld.inc
 eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
---let SEARCH_PATTERN= InnoDB: Missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT for tablespace 42
+let SEARCH_PATTERN=\\[ERROR\\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with ibbackup was here!!!1!\.;
 --source include/search_pattern_in_file.inc
 
 --echo # 10.3 missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT
@@ -403,7 +391,7 @@ EOF
 --source include/start_mysqld.inc
 eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
---let SEARCH_PATTERN= InnoDB: Missing MLOG_FILE_NAME or MLOG_FILE_DELETE before MLOG_CHECKPOINT for tablespace 42
+let SEARCH_PATTERN=\\[ERROR\\] InnoDB: Upgrade after a crash is not supported\. The redo log was created with MariaDB 10\.3\.1\.;
 --source include/search_pattern_in_file.inc
 
 --echo # Empty 10.3 redo log
@@ -421,11 +409,7 @@ SELECT COUNT(*) FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 --source include/shutdown_mysqld.inc
-# In encryption.innodb_encrypt_log_corruption, we would convert the
-# log to encrypted format. Writing an extra log checkpoint before the
-# redo log conversion would advance the LSN by the size of a
-# FILE_CHECKPOINT record (12 bytes).
---let SEARCH_PATTERN= InnoDB: .* started; log sequence number 12139[78]\d; transaction id 0
+--let SEARCH_PATTERN= InnoDB: .* started; log sequence number 1213964; transaction id 0
 --source include/search_pattern_in_file.inc
 
 --echo # Empty 10.2 redo log
diff --git a/storage/innobase/CMakeLists.txt b/storage/innobase/CMakeLists.txt
index 593567bb097..b2cf1620ba4 100644
--- a/storage/innobase/CMakeLists.txt
+++ b/storage/innobase/CMakeLists.txt
@@ -1,5 +1,5 @@
 # Copyright (c) 2006, 2017, Oracle and/or its affiliates. All rights reserved.
-# Copyright (c) 2014, 2019, MariaDB Corporation.
+# Copyright (c) 2014, 2020, 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
@@ -84,9 +84,7 @@ SET(INNOBASE_SOURCES
 	log/log0log.cc
 	log/log0recv.cc
 	log/log0crypt.cc
-	mach/mach0data.cc
 	mem/mem0mem.cc
-	mtr/mtr0log.cc
 	mtr/mtr0mtr.cc
 	os/os0file.cc
 	os/os0proc.cc
diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 0c7e3e38d78..74980775f8e 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -1376,11 +1376,9 @@ btr_write_autoinc(dict_index_t* index, ib_uint64_t autoinc, bool reset)
 }
 
 /** Reorganize an index page.
-@tparam recovery   whether this is invoked by btr_parse_page_reorganize()
 @param cursor      index page cursor
 @param index       the index that the cursor belongs to
 @param mtr         mini-transaction */
-template<bool recovery= false>
 static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
                                     mtr_t *mtr)
 {
@@ -1413,9 +1411,7 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
 	memcpy_aligned<UNIV_PAGE_SIZE_MIN>(temp_block->frame, block->frame,
 					   srv_page_size);
 
-	if (!recovery) {
-		btr_search_drop_page_hash_index(block);
-	}
+	btr_search_drop_page_hash_index(block);
 
 	/* Save the cursor position. */
 	pos = page_rec_get_n_recs_before(page_cur_get_rec(cursor));
@@ -1447,9 +1443,9 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
 	if (trx_id_t trx_id = page_get_max_trx_id(temp_block->frame)) {
 		/* PAGE_MAX_TRX_ID must be zero on non-leaf pages other than
 		clustered index root pages. */
-		ut_ad(recovery || (dict_index_is_sec_or_ibuf(index)
-				   ? page_is_leaf(temp_block->frame)
-				   : block->page.id.page_no() == index->page));
+		ut_ad(dict_index_is_sec_or_ibuf(index)
+		      ? page_is_leaf(temp_block->frame)
+		      : block->page.id.page_no() == index->page);
 		page_set_max_trx_id(block, NULL, trx_id, mtr);
 	} else {
 		/* PAGE_MAX_TRX_ID is unused in clustered index pages
@@ -1458,7 +1454,7 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
 		It was always zero-initialized in page_create().
 		PAGE_MAX_TRX_ID must be nonzero on
 		dict_index_is_sec_or_ibuf() leaf pages. */
-		ut_ad(recovery || index->table->is_temporary()
+		ut_ad(index->table->is_temporary()
 		      || !page_is_leaf(temp_block->frame)
 		      || !dict_index_is_sec_or_ibuf(index));
 	}
@@ -1482,46 +1478,44 @@ static void btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
 		ut_ad(cursor->rec == page_get_infimum_rec(block->frame));
 	}
 
-	if (!recovery) {
-		if (block->page.id.page_no() == index->page
-		    && fil_page_get_type(temp_block->frame)
-		    == FIL_PAGE_TYPE_INSTANT) {
-			/* Preserve the PAGE_INSTANT information. */
-			ut_ad(index->is_instant());
-			static_assert(!(FIL_PAGE_TYPE % 2), "alignment");
-			memcpy_aligned<2>(FIL_PAGE_TYPE + block->frame,
-					  FIL_PAGE_TYPE + temp_block->frame, 2);
-			static_assert(!((PAGE_HEADER+PAGE_INSTANT) % 2), "");
-			memcpy_aligned<2>(PAGE_HEADER + PAGE_INSTANT
-					  + block->frame,
-					  PAGE_HEADER + PAGE_INSTANT
-					  + temp_block->frame, 2);
-			if (!index->table->instant) {
-			} else if (page_is_comp(block->frame)) {
-				memcpy(PAGE_NEW_INFIMUM + block->frame,
-				       PAGE_NEW_INFIMUM + temp_block->frame, 8);
-				memcpy(PAGE_NEW_SUPREMUM + block->frame,
-				       PAGE_NEW_SUPREMUM + temp_block->frame, 8);
-			} else {
-				memcpy(PAGE_OLD_INFIMUM + block->frame,
-				       PAGE_OLD_INFIMUM + temp_block->frame, 8);
-				memcpy(PAGE_OLD_SUPREMUM + block->frame,
-				       PAGE_OLD_SUPREMUM + temp_block->frame, 8);
-			}
+	if (block->page.id.page_no() == index->page
+	    && fil_page_get_type(temp_block->frame)
+	    == FIL_PAGE_TYPE_INSTANT) {
+		/* Preserve the PAGE_INSTANT information. */
+		ut_ad(index->is_instant());
+		static_assert(!(FIL_PAGE_TYPE % 2), "alignment");
+		memcpy_aligned<2>(FIL_PAGE_TYPE + block->frame,
+				  FIL_PAGE_TYPE + temp_block->frame, 2);
+		static_assert(!((PAGE_HEADER+PAGE_INSTANT) % 2), "");
+		memcpy_aligned<2>(PAGE_HEADER + PAGE_INSTANT
+				  + block->frame,
+				  PAGE_HEADER + PAGE_INSTANT
+				  + temp_block->frame, 2);
+		if (!index->table->instant) {
+		} else if (page_is_comp(block->frame)) {
+			memcpy(PAGE_NEW_INFIMUM + block->frame,
+			       PAGE_NEW_INFIMUM + temp_block->frame, 8);
+			memcpy(PAGE_NEW_SUPREMUM + block->frame,
+			       PAGE_NEW_SUPREMUM + temp_block->frame, 8);
+		} else {
+			memcpy(PAGE_OLD_INFIMUM + block->frame,
+			       PAGE_OLD_INFIMUM + temp_block->frame, 8);
+			memcpy(PAGE_OLD_SUPREMUM + block->frame,
+			       PAGE_OLD_SUPREMUM + temp_block->frame, 8);
 		}
+	}
 
-		if (!dict_table_is_locking_disabled(index->table)) {
-			/* Update the record lock bitmaps */
-			lock_move_reorganize_page(block, temp_block);
-		}
+	if (!dict_table_is_locking_disabled(index->table)) {
+		/* Update the record lock bitmaps */
+		lock_move_reorganize_page(block, temp_block);
 	}
 
 	buf_block_free(temp_block);
 
 	MONITOR_INC(MONITOR_INDEX_REORG_SUCCESSFUL);
 
-	if (!recovery && UNIV_UNLIKELY(fil_page_get_type(block->frame)
-				       == FIL_PAGE_TYPE_INSTANT)) {
+	if (UNIV_UNLIKELY(fil_page_get_type(block->frame)
+			  == FIL_PAGE_TYPE_INSTANT)) {
 		/* Log the PAGE_INSTANT information. */
 		ut_ad(index->is_instant());
 		mtr->write<2,mtr_t::FORCED>(*block, FIL_PAGE_TYPE
@@ -1610,54 +1604,6 @@ btr_page_reorganize(
 	return true;
 }
 
-/***********************************************************//**
-Parses a redo log record of reorganizing a page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_parse_page_reorganize(
-/*======================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	bool		compressed,/*!< in: true if compressed page */
-	buf_block_t*	block,	/*!< in: page to be reorganized, or NULL */
-	mtr_t*		mtr)	/*!< in: mtr or NULL */
-{
-	ulint	level = page_zip_level;
-
-	ut_ad(ptr != NULL);
-	ut_ad(end_ptr != NULL);
-	ut_ad(index != NULL);
-
-	/* If dealing with a compressed page the record has the
-	compression level used during original compression written in
-	one byte. Otherwise record is empty. */
-	if (compressed) {
-		if (ptr == end_ptr) {
-			return(NULL);
-		}
-
-		level = mach_read_from_1(ptr);
-
-		ut_a(level <= 9);
-		++ptr;
-	} else {
-		level = page_zip_level;
-	}
-
-	if (block == NULL) {
-	} else if (block->page.zip.data) {
-		page_zip_reorganize(block, index, level, mtr, true);
-	} else {
-		page_cur_t cur;
-		page_cur_set_before_first(block, &cur);
-		btr_page_reorganize_low<true>(&cur, index, mtr);
-	}
-
-	return(ptr);
-}
-
 /** Empty an index page (possibly the root page). @see btr_page_create().
 @param[in,out]	block		page to be emptied
 @param[in,out]	page_zip	compressed page frame, or NULL
@@ -3196,38 +3142,6 @@ void btr_level_list_remove(const buf_block_t& block, const dict_index_t& index,
 	}
 }
 
-/****************************************************************//**
-Parses the redo log record for setting an index record as the predefined
-minimum record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_parse_set_min_rec_mark(
-/*=======================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	ulint		comp,	/*!< in: nonzero=compact page format */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	mtr_t*		mtr)	/*!< in: mtr or NULL */
-{
-	rec_t*	rec;
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	if (block) {
-		ut_a(!page_is_comp(block->frame) == !comp);
-
-		rec = block->frame + mach_read_from_2(ptr);
-
-		btr_set_min_rec_mark(rec, *block, mtr);
-	}
-
-	return(ptr + 2);
-}
-
 /*************************************************************//**
 If page is the only on its level, this function moves its records to the
 father page, thus reducing the tree height.
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 22495ddbd88..b00d7c6659f 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -3988,313 +3988,6 @@ static void btr_cur_upd_rec_sys(buf_block_t *block, rec_t *rec,
     mtr->memcpy<mtr_t::OPT>(*block, dest, sys + d, len);
 }
 
-/*********************************************************************//**
-Parses the log data of system field values.
-@return log data end or NULL */
-static
-byte*
-row_upd_parse_sys_vals(
-/*===================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	ulint*		pos,	/*!< out: TRX_ID position in record */
-	trx_id_t*	trx_id,	/*!< out: trx id */
-	roll_ptr_t*	roll_ptr)/*!< out: roll ptr */
-{
-	*pos = mach_parse_compressed(&ptr, end_ptr);
-
-	if (ptr == NULL) {
-
-		return(NULL);
-	}
-
-	if (end_ptr < ptr + DATA_ROLL_PTR_LEN) {
-
-		return(NULL);
-	}
-
-	*roll_ptr = trx_read_roll_ptr(ptr);
-	ptr += DATA_ROLL_PTR_LEN;
-
-	*trx_id = mach_u64_parse_compressed(&ptr, end_ptr);
-
-	return(const_cast<byte*>(ptr));
-}
-
-/***********************************************************//**
-Sets the value of the ith field SQL null bit of an old-style record. */
-static
-void
-rec_set_nth_field_null_bit(
-/*=======================*/
-	rec_t*	rec,	/*!< in: record */
-	ulint	i,	/*!< in: ith field */
-	ibool	val)	/*!< in: value to set */
-{
-	ulint	info;
-
-	if (rec_get_1byte_offs_flag(rec)) {
-
-		info = rec_1_get_field_end_info(rec, i);
-
-		if (val) {
-			info = info | REC_1BYTE_SQL_NULL_MASK;
-		} else {
-			info = info & ~REC_1BYTE_SQL_NULL_MASK;
-		}
-
-		rec_1_set_field_end_info(rec, i, info);
-
-		return;
-	}
-
-	info = rec_2_get_field_end_info(rec, i);
-
-	if (val) {
-		info = info | REC_2BYTE_SQL_NULL_MASK;
-	} else {
-		info = info & ~REC_2BYTE_SQL_NULL_MASK;
-	}
-
-	rec_2_set_field_end_info(rec, i, info);
-}
-
-/***********************************************************//**
-Sets an old-style record field to SQL null.
-The physical size of the field is not changed. */
-static
-void
-rec_set_nth_field_sql_null(
-/*=======================*/
-	rec_t*	rec,	/*!< in: record */
-	ulint	n)	/*!< in: index of the field */
-{
-	ulint	offset;
-
-	offset = rec_get_field_start_offs(rec, n);
-
-	data_write_sql_null(rec + offset, rec_get_nth_field_size(rec, n));
-
-	rec_set_nth_field_null_bit(rec, n, TRUE);
-}
-
-/***********************************************************//**
-This is used to modify the value of an already existing field in a record.
-The previous value must have exactly the same size as the new value. If len
-is UNIV_SQL_NULL then the field is treated as an SQL null.
-For records in ROW_FORMAT=COMPACT (new-style records), len must not be
-UNIV_SQL_NULL unless the field already is SQL null. */
-static
-void
-rec_set_nth_field(
-/*==============*/
-	rec_t*		rec,	/*!< in: record */
-	const offset_t*	offsets,/*!< in: array returned by rec_get_offsets() */
-	ulint		n,	/*!< in: index number of the field */
-	const void*	data,	/*!< in: pointer to the data
-				if not SQL null */
-	ulint		len)	/*!< in: length of the data or UNIV_SQL_NULL */
-{
-	byte*	data2;
-	ulint	len2;
-
-	ut_ad(rec_offs_validate(rec, NULL, offsets));
-	ut_ad(!rec_offs_nth_default(offsets, n));
-
-	if (len == UNIV_SQL_NULL) {
-		if (!rec_offs_nth_sql_null(offsets, n)) {
-			ut_a(!rec_offs_comp(offsets));
-			rec_set_nth_field_sql_null(rec, n);
-		}
-
-		return;
-	}
-
-	data2 = (byte*)rec_get_nth_field(rec, offsets, n, &len2);
-	if (len2 == UNIV_SQL_NULL) {
-		ut_ad(!rec_offs_comp(offsets));
-		rec_set_nth_field_null_bit(rec, n, FALSE);
-		ut_ad(len == rec_get_nth_field_size(rec, n));
-	} else {
-		ut_ad(len2 == len);
-	}
-
-	memcpy(data2, data, len);
-}
-
-/***********************************************************//**
-Replaces the new column values stored in the update vector to the
-record given. No field size changes are allowed. This function is
-usually invoked on a clustered index. The only use case for a
-secondary index is row_ins_sec_index_entry_by_modify() or its
-counterpart in ibuf_insert_to_index_page(). */
-static
-void
-row_upd_rec_in_place(
-/*=================*/
-	buf_block_t*	block,	/*!< in/out: index page */
-	rec_t*		rec,	/*!< in/out: record where replaced */
-	dict_index_t*	index,	/*!< in: the index the record belongs to */
-	const offset_t*	offsets,/*!< in: array returned by rec_get_offsets() */
-	const upd_t*	update,	/*!< in: update vector */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction */
-{
-	const upd_field_t*	upd_field;
-	const dfield_t*		new_val;
-	ulint			n_fields;
-	ulint			i;
-
-	ut_ad(rec_offs_validate(rec, index, offsets));
-	ut_ad(!index->table->skip_alter_undo);
-
-	if (rec_offs_comp(offsets)) {
-#ifdef UNIV_DEBUG
-		switch (rec_get_status(rec)) {
-		case REC_STATUS_ORDINARY:
-			break;
-		case REC_STATUS_INSTANT:
-			ut_ad(index->is_instant());
-			break;
-		case REC_STATUS_NODE_PTR:
-			if (index->is_dummy
-			    && fil_page_get_type(page_align(rec))
-			    == FIL_PAGE_RTREE) {
-				/* The function rtr_update_mbr_field_in_place()
-				is generating MLOG_COMP_REC_UPDATE_IN_PLACE
-				and MLOG_REC_UPDATE_IN_PLACE records for
-				node pointer pages. */
-				break;
-			}
-			/* fall through */
-		case REC_STATUS_INFIMUM:
-		case REC_STATUS_SUPREMUM:
-			ut_ad(!"wrong record status in update");
-		}
-#endif /* UNIV_DEBUG */
-
-		rec_set_bit_field_1(rec, update->info_bits, REC_NEW_INFO_BITS,
-				    REC_INFO_BITS_MASK, REC_INFO_BITS_SHIFT);
-	} else {
-		rec_set_bit_field_1(rec, update->info_bits, REC_OLD_INFO_BITS,
-				    REC_INFO_BITS_MASK, REC_INFO_BITS_SHIFT);
-	}
-
-	n_fields = upd_get_n_fields(update);
-
-	for (i = 0; i < n_fields; i++) {
-		upd_field = upd_get_nth_field(update, i);
-
-		/* No need to update virtual columns for non-virtual index */
-		if (upd_fld_is_virtual_col(upd_field)
-		    && !dict_index_has_virtual(index)) {
-			continue;
-		}
-
-		new_val = &(upd_field->new_val);
-		ut_ad(!dfield_is_ext(new_val) ==
-		      !rec_offs_nth_extern(offsets, upd_field->field_no));
-
-		rec_set_nth_field(rec, offsets, upd_field->field_no,
-				  dfield_get_data(new_val),
-				  dfield_get_len(new_val));
-	}
-
-	if (UNIV_LIKELY_NULL(block->page.zip.data)) {
-		page_zip_write_rec(block, rec, index, offsets, 0, mtr);
-	}
-}
-
-/***********************************************************//**
-Parses a redo log record of updating a record in-place.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_cur_parse_update_in_place(
-/*==========================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: page or NULL */
-	dict_index_t*	index,	/*!< in: index corresponding to page */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction */
-{
-	ulint		flags;
-	upd_t*		update;
-	ulint		pos;
-	trx_id_t	trx_id;
-	roll_ptr_t	roll_ptr;
-	ulint		rec_offset;
-	mem_heap_t*	heap;
-	offset_t*	offsets;
-
-	if (end_ptr < ptr + 1) {
-
-		return(NULL);
-	}
-
-	flags = mach_read_from_1(ptr);
-	ptr++;
-
-	ptr = row_upd_parse_sys_vals(ptr, end_ptr, &pos, &trx_id, &roll_ptr);
-
-	if (ptr == NULL) {
-
-		return(NULL);
-	}
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	rec_offset = mach_read_from_2(ptr);
-	ptr += 2;
-
-	ut_a(rec_offset <= srv_page_size);
-
-	heap = mem_heap_create(256);
-
-	ptr = row_upd_index_parse(ptr, end_ptr, heap, &update);
-
-	if (!ptr || !block) {
-func_exit:
-		mem_heap_free(heap);
-		return ptr;
-	}
-
-	ut_a(!!page_is_comp(block->frame) == index->table->not_redundant());
-	rec_t* rec = block->frame + rec_offset;
-
-	/* We do not need to reserve search latch, as the page is only
-	being recovered, and there cannot be a hash index to it. */
-
-	/* The function rtr_update_mbr_field_in_place() is generating
-	these records on node pointer pages; therefore we have to
-	check if this is a leaf page. */
-
-	offsets = rec_get_offsets(rec, index, NULL,
-				  flags != (BTR_NO_UNDO_LOG_FLAG
-					    | BTR_NO_LOCKING_FLAG
-					    | BTR_KEEP_SYS_FLAG)
-				  || page_is_leaf(block->frame),
-				  ULINT_UNDEFINED, &heap);
-
-	if (flags & BTR_KEEP_SYS_FLAG) {
-	} else if (UNIV_LIKELY_NULL(block->page.zip.data)) {
-		page_zip_write_trx_id_and_roll_ptr(
-			block, rec, offsets, pos, trx_id, roll_ptr, mtr);
-	} else {
-		ulint len;
-		byte* field = rec_get_nth_field(rec, offsets, pos, &len);
-		ut_ad(len == DATA_TRX_ID_LEN);
-		compile_time_assert(DATA_TRX_ID + 1 == DATA_ROLL_PTR);
-		trx_write_trx_id(field, trx_id);
-		trx_write_roll_ptr(field + DATA_TRX_ID_LEN, roll_ptr);
-	}
-
-	row_upd_rec_in_place(block, rec, index, offsets, update, mtr);
-	goto func_exit;
-}
-
 /*************************************************************//**
 See if there is enough place in the page modification log to log
 an update-in-place.
@@ -5607,134 +5300,6 @@ void btr_rec_set_deleted(buf_block_t *block, rec_t *rec, mtr_t *mtr)
 template void btr_rec_set_deleted<false>(buf_block_t *, rec_t *, mtr_t *);
 template void btr_rec_set_deleted<true>(buf_block_t *, rec_t *, mtr_t *);
 
-/****************************************************************//**
-Parses the redo log record for delete marking or unmarking of a clustered
-index record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_cur_parse_del_mark_set_clust_rec(
-/*=================================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: page or NULL */
-	dict_index_t*	index,	/*!< in: index corresponding to page */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction */
-{
-	ulint		flags;
-	ulint		val;
-	ulint		pos;
-	trx_id_t	trx_id;
-	roll_ptr_t	roll_ptr;
-	ulint		offset;
-
-	ut_ad(!block
-	      || !!page_is_comp(block->frame)
-	      == index->table->not_redundant());
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	flags = mach_read_from_1(ptr);
-	ptr++;
-	val = mach_read_from_1(ptr);
-	ptr++;
-
-	ptr = row_upd_parse_sys_vals(ptr, end_ptr, &pos, &trx_id, &roll_ptr);
-
-	if (ptr == NULL) {
-
-		return(NULL);
-	}
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	offset = mach_read_from_2(ptr);
-	ptr += 2;
-
-	ut_a(offset <= srv_page_size);
-
-	/* In delete-marked records, DB_TRX_ID must
-	always refer to an existing undo log record. */
-	ut_ad(trx_id || (flags & BTR_KEEP_SYS_FLAG));
-
-	if (block) {
-		rec_t* rec = block->frame + offset;
-
-		/* We do not need to reserve search latch, as the page
-		is only being recovered, and there cannot be a hash index to
-		it. Besides, these fields are being updated in place
-		and the adaptive hash index does not depend on them. */
-		byte* b = rec - (page_is_comp(block->frame)
-				 ? REC_NEW_INFO_BITS
-				 : REC_OLD_INFO_BITS);
-
-		if (val) {
-			*b |= REC_INFO_DELETED_FLAG;
-		} else {
-			*b &= ~REC_INFO_DELETED_FLAG;
-		}
-
-		if (UNIV_LIKELY_NULL(block->page.zip.data)) {
-			page_zip_rec_set_deleted(block, rec, val, mtr);
-		}
-
-		/* pos is the offset of DB_TRX_ID in the clustered index.
-		Debug assertions may also access DB_ROLL_PTR at pos+1.
-		Therefore, we must compute offsets for the first pos+2
-		clustered index fields. */
-		ut_ad(pos <= MAX_REF_PARTS);
-
-		offset_t offsets_[REC_OFFS_HEADER_SIZE + MAX_REF_PARTS + 2];
-		rec_offs_init(offsets_);
-		mem_heap_t*	heap	= NULL;
-
-		if (!(flags & BTR_KEEP_SYS_FLAG)) {
-			offset_t* offsets = rec_get_offsets(rec, index,
-							    offsets_, true,
-							    pos + 2, &heap);
-			if (UNIV_LIKELY_NULL(block->page.zip.data)) {
-				page_zip_write_trx_id_and_roll_ptr(
-					block, rec, offsets, pos, trx_id,
-					roll_ptr, mtr);
-			} else {
-				ulint	len;
-
-				byte* field = rec_get_nth_field(
-					rec, offsets, pos, &len);
-				ut_ad(len == DATA_TRX_ID_LEN);
-				compile_time_assert(DATA_TRX_ID + 1
-						    == DATA_ROLL_PTR);
-				trx_write_trx_id(field, trx_id);
-				trx_write_roll_ptr(field + DATA_TRX_ID_LEN,
-						   roll_ptr);
-			}
-		} else {
-			/* In delete-marked records, DB_TRX_ID must
-			always refer to an existing undo log record. */
-			ut_ad(memcmp(rec_get_nth_field(
-					     rec,
-					     rec_get_offsets(rec, index,
-							     offsets_, true,
-							     pos, &heap),
-					     pos, &offset),
-				     field_ref_zero, DATA_TRX_ID_LEN));
-			ut_ad(offset == DATA_TRX_ID_LEN);
-		}
-
-		if (UNIV_LIKELY_NULL(heap)) {
-			mem_heap_free(heap);
-		}
-	}
-
-	return(ptr);
-}
-
 /***********************************************************//**
 Marks a clustered index record deleted. Writes an undo log record to
 undo log on this delete marking. Writes in the trx id field the id
@@ -5810,62 +5375,6 @@ btr_cur_del_mark_set_clust_rec(
 	return(err);
 }
 
-/****************************************************************//**
-Parses the redo log record for delete marking or unmarking of a secondary
-index record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_cur_parse_del_mark_set_sec_rec(
-/*===============================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: page or NULL */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction */
-{
-	ulint	val;
-	ulint	offset;
-
-	if (end_ptr < ptr + 3) {
-
-		return(NULL);
-	}
-
-	val = mach_read_from_1(ptr);
-	ptr++;
-
-	offset = mach_read_from_2(ptr);
-	ptr += 2;
-
-	ut_a(offset <= srv_page_size);
-
-	if (!block) {
-		return ptr;
-	}
-
-	rec_t* rec = block->frame + offset;
-
-	/* We do not need to reserve search latch, as the page
-	is only being recovered, and there cannot be a hash index to
-	it. Besides, the delete-mark flag is being updated in place
-	and the adaptive hash index does not depend on it. */
-	byte* b = rec - (page_is_comp(block->frame)
-			 ? REC_NEW_INFO_BITS
-			 : REC_OLD_INFO_BITS);
-
-	if (val) {
-		*b |= REC_INFO_DELETED_FLAG;
-	} else {
-		*b &= ~REC_INFO_DELETED_FLAG;
-	}
-
-	if (UNIV_LIKELY_NULL(block->page.zip.data)) {
-		page_zip_rec_set_deleted(block, rec, val, mtr);
-	}
-
-	return ptr;
-}
-
 /*==================== B-TREE RECORD REMOVE =========================*/
 
 /*************************************************************//**
diff --git a/storage/innobase/buf/buf0dblwr.cc b/storage/innobase/buf/buf0dblwr.cc
index b1f5aa50433..a4651f53f8d 100644
--- a/storage/innobase/buf/buf0dblwr.cc
+++ b/storage/innobase/buf/buf0dblwr.cc
@@ -242,8 +242,7 @@ too_small:
 		ut_ad(rw_lock_get_x_lock_count(&new_block->lock) == 1);
 		page_no = new_block->page.id.page_no();
 		/* We only do this in the debug build, to ensure that
-		both the check in buf_flush_init_for_writing() and
-		recv_parse_or_apply_log_rec_body() will see a valid
+		the check in buf_flush_init_for_writing() will see a valid
 		page type. The flushes of new_block are actually
 		unnecessary here.  */
 		ut_d(mtr.write<2>(*new_block,
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index fc2a26d8b04..bd3f08a8d97 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -439,97 +439,6 @@ void fil_space_crypt_t::write_page0(buf_block_t* block, mtr_t* mtr)
 	mtr->memcpy(*block, offset + MAGIC_SZ, b - start);
 }
 
-/******************************************************************
-Parse a MLOG_FILE_WRITE_CRYPT_DATA log entry
-@param[in]	ptr		Log entry start
-@param[in]	end_ptr		Log entry end
-@param[in]	block		buffer block
-@return position on log buffer */
-UNIV_INTERN
-const byte*
-fil_parse_write_crypt_data(
-	const byte*		ptr,
-	const byte*		end_ptr,
-	dberr_t*		err)
-{
-	/* check that redo log entry is complete */
-	uint entry_size =
-		4 + // size of space_id
-		2 + // size of offset
-		1 + // size of type
-		1 + // size of iv-len
-		4 +  // size of min_key_version
-		4 +  // size of key_id
-		1; // fil_encryption_t
-
-	*err = DB_SUCCESS;
-
-	if (ptr + entry_size > end_ptr) {
-		return NULL;
-	}
-
-	ulint space_id = mach_read_from_4(ptr);
-	ptr += 4;
-	// uint offset = mach_read_from_2(ptr);
-	ptr += 2;
-	uint type = mach_read_from_1(ptr);
-	ptr += 1;
-	uint len = mach_read_from_1(ptr);
-	ptr += 1;
-
-	if ((type != CRYPT_SCHEME_1 && type != CRYPT_SCHEME_UNENCRYPTED)
-	    || len != CRYPT_SCHEME_1_IV_LEN) {
-		*err = DB_CORRUPTION;
-		return NULL;
-	}
-
-	uint min_key_version = mach_read_from_4(ptr);
-	ptr += 4;
-
-	uint key_id = mach_read_from_4(ptr);
-	ptr += 4;
-
-	fil_encryption_t encryption = (fil_encryption_t)mach_read_from_1(ptr);
-	ptr +=1;
-
-	if (ptr + len > end_ptr) {
-		return NULL;
-	}
-
-	mutex_enter(&fil_system.mutex);
-
-	fil_space_t* space = fil_space_get_by_id(space_id);
-
-	if (!space) {
-		mutex_exit(&fil_system.mutex);
-		return ptr + len;
-	}
-
-	fil_space_crypt_t* crypt_data = fil_space_create_crypt_data(
-		encryption, key_id);
-
-	crypt_data->min_key_version = min_key_version;
-	crypt_data->type = type;
-	memcpy(crypt_data->iv, ptr, len);
-	ptr += len;
-
-	if (space->crypt_data) {
-		fil_space_merge_crypt_data(space->crypt_data, crypt_data);
-		fil_space_destroy_crypt_data(&crypt_data);
-		crypt_data = space->crypt_data;
-	} else {
-		space->crypt_data = crypt_data;
-	}
-
-	mutex_exit(&fil_system.mutex);
-
-	if (crypt_data->should_encrypt() && !crypt_data->is_key_found()) {
-		*err = DB_DECRYPTION_FAILED;
-	}
-
-	return ptr;
-}
-
 /** Encrypt a buffer for non full checksum.
 @param[in,out]		crypt_data		Crypt data
 @param[in]		space			space_id
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index a13035ef0a2..94a5af0015c 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -521,25 +521,6 @@ ibuf_max_size_update(
 	mutex_exit(&ibuf_mutex);
 }
 
-
-/** Apply MLOG_IBUF_BITMAP_INIT when crash-upgrading */
-ATTRIBUTE_COLD void ibuf_bitmap_init_apply(buf_block_t* block)
-{
-	page_t*	page;
-	ulint	byte_offset;
-
-	page = buf_block_get_frame(block);
-	fil_page_set_type(page, FIL_PAGE_IBUF_BITMAP);
-
-	/* Write all zeros to the bitmap */
-	compile_time_assert(!(IBUF_BITS_PER_PAGE % 2));
-
-	byte_offset = UT_BITS_IN_BYTES(block->physical_size()
-				       * IBUF_BITS_PER_PAGE);
-
-	memset(page + IBUF_BITMAP, 0, byte_offset);
-}
-
 # ifdef UNIV_DEBUG
 /** Gets the desired bits for a given page from a bitmap page.
 @param[in]	page		bitmap page
diff --git a/storage/innobase/include/btr0btr.h b/storage/innobase/include/btr0btr.h
index c59474f7d12..c5b67e4b3a6 100644
--- a/storage/innobase/include/btr0btr.h
+++ b/storage/innobase/include/btr0btr.h
@@ -572,34 +572,6 @@ btr_discard_page(
 	btr_cur_t*	cursor,	/*!< in: cursor on the page to discard: not on
 				the root page */
 	mtr_t*		mtr);	/*!< in: mtr */
-/****************************************************************//**
-Parses the redo log record for setting an index record as the predefined
-minimum record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-MY_ATTRIBUTE((nonnull(1,2), warn_unused_result))
-const byte*
-btr_parse_set_min_rec_mark(
-/*=======================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	ulint		comp,	/*!< in: nonzero=compact page format */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	mtr_t*		mtr);	/*!< in: mtr or NULL */
-/***********************************************************//**
-Parses a redo log record of reorganizing a page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_parse_page_reorganize(
-/*======================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	bool		compressed,/*!< in: true if compressed page */
-	buf_block_t*	block,	/*!< in: page to be reorganized, or NULL */
-	mtr_t*		mtr)	/*!< in: mtr or NULL */
-	MY_ATTRIBUTE((warn_unused_result));
 /**************************************************************//**
 Gets the number of pages in a B-tree.
 @return number of pages, or ULINT_UNDEFINED if the index is unavailable */
diff --git a/storage/innobase/include/btr0cur.h b/storage/innobase/include/btr0cur.h
index 0e9d31bf1bc..0a45c3bb45f 100644
--- a/storage/innobase/include/btr0cur.h
+++ b/storage/innobase/include/btr0cur.h
@@ -555,44 +555,6 @@ btr_cur_pessimistic_delete(
 @param[in,out]	mtr	mini-transaction */
 void btr_cur_node_ptr_delete(btr_cur_t* parent, mtr_t* mtr)
 	MY_ATTRIBUTE((nonnull));
-/***********************************************************//**
-Parses a redo log record of updating a record in-place.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_cur_parse_update_in_place(
-/*==========================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: page or NULL */
-	dict_index_t*	index,	/*!< in: index corresponding to page */
-	mtr_t*		mtr);	/*!< in/out: mini-transaction */
-/****************************************************************//**
-Parses the redo log record for delete marking or unmarking of a clustered
-index record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_cur_parse_del_mark_set_clust_rec(
-/*=================================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: page or NULL */
-	dict_index_t*	index,	/*!< in: index corresponding to page */
-	mtr_t*		mtr);	/*!< in/out: mini-transaction */
-/****************************************************************//**
-Parses the redo log record for delete marking or unmarking of a secondary
-index record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-btr_cur_parse_del_mark_set_sec_rec(
-/*===============================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: page or NULL */
-	mtr_t*		mtr);	/*!< in/out: mini-transaction */
-
 /** Estimates the number of rows in a given index range.
 @param[in]	index	index
 @param[in]	tuple1	range start, may also be empty tuple
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index c604c4f70d3..6bb97a4ed62 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -1,6 +1,6 @@
 /*****************************************************************************
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (c) 2015, 2019, MariaDB Corporation.
+Copyright (c) 2015, 2020, 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
@@ -291,20 +291,6 @@ void
 fil_space_destroy_crypt_data(
 	fil_space_crypt_t **crypt_data);
 
-/******************************************************************
-Parse a MLOG_FILE_WRITE_CRYPT_DATA log entry
-@param[in]	ptr		Log entry start
-@param[in]	end_ptr		Log entry end
-@param[out]	err		DB_SUCCESS or DB_DECRYPTION_FAILED
-@return position on log buffer */
-UNIV_INTERN
-const byte*
-fil_parse_write_crypt_data(
-	const byte*		ptr,
-	const byte*		end_ptr,
-	dberr_t*		err)
-	MY_ATTRIBUTE((warn_unused_result));
-
 /** Amend encryption information from redo log.
 @param[in]	space	tablespace
 @param[in]	data	encryption metadata */
diff --git a/storage/innobase/include/ibuf0ibuf.h b/storage/innobase/include/ibuf0ibuf.h
index 2fd6ed0e954..061f1bc4e8e 100644
--- a/storage/innobase/include/ibuf0ibuf.h
+++ b/storage/innobase/include/ibuf0ibuf.h
@@ -362,9 +362,6 @@ ibuf_merge_space(
 /*=============*/
 	ulint	space);	/*!< in: space id */
 
-/** Apply MLOG_IBUF_BITMAP_INIT when crash-upgrading */
-ATTRIBUTE_COLD void ibuf_bitmap_init_apply(buf_block_t* block);
-
 /******************************************************************//**
 Looks if the insert buffer is empty.
 @return true if empty */
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 90b6cfe69d8..be387e18e3a 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -99,15 +99,6 @@ recv_sys.parse_start_lsn is non-zero.
 @return true if more data added */
 bool recv_sys_add_to_parsing_buf(const byte* log_block, lsn_t scanned_lsn);
 
-/** Parse log records from a buffer and optionally store them to recv_sys.pages
-to wait merging to file pages.
-@param[in]	checkpoint_lsn		the LSN of the latest checkpoint
-@param[in]	store			whether to store page operations
-@param[in]	apply			whether to apply the records
-@return whether MLOG_CHECKPOINT or FILE_CHECKPOINT record
-was seen the first time, or corruption was noticed */
-bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t *store, bool apply);
-
 /** Moves the parsing buffer data left to the buffer start */
 void recv_sys_justify_left_parsing_buf();
 
@@ -276,7 +267,7 @@ struct recv_sys_t{
 				the file system contents is detected
 				during log scan or apply */
 	lsn_t		mlog_checkpoint_lsn;
-				/*!< the LSN of a MLOG_CHECKPOINT
+				/*!< the LSN of a FILE_CHECKPOINT
 				record, or 0 if none was parsed */
 	/** the time when progress was last reported */
 	time_t		progress_time;
@@ -328,27 +319,16 @@ public:
   /** whether all redo log in the current batch has been applied */
   bool after_apply= false;
 #endif
-	/** Initialize the redo log recovery subsystem. */
-	void create();
-
-	/** Free most recovery data structures. */
-	void debug_free();
+  /** Initialize the redo log recovery subsystem. */
+  void create();
 
-	/** Clean up after create() */
-	void close();
+  /** Free most recovery data structures. */
+  void debug_free();
 
-	bool is_initialised() const { return buf_size != 0; }
+  /** Clean up after create() */
+  void close();
 
-	/** Store a redo log record for applying.
-	@param type	record type
-	@param page_id	page identifier
-	@param body	record body
-	@param rec_end	end of record
-	@param lsn	start LSN of the mini-transaction
-	@param end_lsn	end LSN of the mini-transaction */
-	inline void add(mlog_id_t type, const page_id_t page_id,
-			const byte* body, const byte* rec_end, lsn_t lsn,
-			lsn_t end_lsn);
+  bool is_initialised() const { return buf_size != 0; }
 
   /** Register a redo log snippet for a page.
   @param page_id  page identifier
@@ -365,24 +345,23 @@ public:
   @param apply           whether to apply file-level log records
   @return whether FILE_CHECKPOINT record was seen the first time,
   or corruption was noticed */
-  inline bool parse(lsn_t checkpoint_lsn, store_t store, bool apply);
+  bool parse(lsn_t checkpoint_lsn, store_t store, bool apply);
 
   /** Clear a fully processed set of stored redo log records. */
   inline void clear();
 
-	/** Determine whether redo log recovery progress should be reported.
-	@param[in]	time	the current time
-	@return	whether progress should be reported
-		(the last report was at least 15 seconds ago) */
-	bool report(time_t time)
-	{
-		if (time - progress_time < 15) {
-			return false;
-		}
-
-		progress_time = time;
-		return true;
-	}
+  /** Determine whether redo log recovery progress should be reported.
+  @param time  the current time
+  @return whether progress should be reported
+  (the last report was at least 15 seconds ago) */
+  bool report(time_t time)
+  {
+    if (time - progress_time < 15)
+      return false;
+
+    progress_time= time;
+    return true;
+  }
 
   /** The alloc() memory alignment, in bytes */
   static constexpr size_t ALIGNMENT= sizeof(size_t);
@@ -390,7 +369,7 @@ public:
   /** Allocate memory for log_rec_t
   @param len  allocation size, in bytes
   @return pointer to len bytes of memory (never NULL) */
-  inline void *alloc(size_t len, bool store_recv= false);
+  inline void *alloc(size_t len);
 
   /** Free a redo log snippet.
   @param data buffer returned by alloc() */
diff --git a/storage/innobase/include/mach0data.h b/storage/innobase/include/mach0data.h
index 3d0e48253eb..88317a73048 100644
--- a/storage/innobase/include/mach0data.h
+++ b/storage/innobase/include/mach0data.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -251,25 +251,6 @@ mach_u64_read_much_compressed(
 /*==========================*/
 	const byte*	b)	/*!< in: pointer to memory from where to read */
 	MY_ATTRIBUTE((warn_unused_result));
-/** Read a 32-bit integer in a compressed form.
-@param[in,out]	ptr	pointer to memory where to read;
-advanced by the number of bytes consumed, or set NULL if out of space
-@param[in]	end_ptr	end of the buffer
-@return unsigned value */
-ib_uint32_t
-mach_parse_compressed(
-	const byte**	ptr,
-	const byte*	end_ptr);
-/** Read a 64-bit integer in a compressed form.
-@param[in,out]	ptr	pointer to memory where to read;
-advanced by the number of bytes consumed, or set NULL if out of space
-@param[in]	end_ptr	end of the buffer
-@return unsigned value */
-UNIV_INLINE
-ib_uint64_t
-mach_u64_parse_compressed(
-	const byte**	ptr,
-	const byte*	end_ptr);
 
 /*********************************************************//**
 Reads a double. It is stored in a little-endian format.
diff --git a/storage/innobase/include/mach0data.ic b/storage/innobase/include/mach0data.ic
index 80bd925d70b..bfccf611991 100644
--- a/storage/innobase/include/mach0data.ic
+++ b/storage/innobase/include/mach0data.ic
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -546,38 +546,6 @@ mach_read_next_much_compressed(
 	return(val);
 }
 
-/** Read a 64-bit integer in a compressed form.
-@param[in,out]	ptr	pointer to memory where to read;
-advanced by the number of bytes consumed, or set NULL if out of space
-@param[in]	end_ptr	end of the buffer
-@return unsigned value */
-UNIV_INLINE
-ib_uint64_t
-mach_u64_parse_compressed(
-	const byte**	ptr,
-	const byte*	end_ptr)
-{
-	ib_uint64_t	val = 0;
-
-	if (end_ptr < *ptr + 5) {
-		*ptr = NULL;
-		return(val);
-	}
-
-	val = mach_read_next_compressed(ptr);
-
-	if (end_ptr < *ptr + 4) {
-		*ptr = NULL;
-		return(val);
-	}
-
-	val <<= 32;
-	val |= mach_read_from_4(*ptr);
-	*ptr += 4;
-
-	return(val);
-}
-
 /*********************************************************//**
 Reads a double. It is stored in a little-endian format.
 @return double read */
diff --git a/storage/innobase/include/mtr0log.h b/storage/innobase/include/mtr0log.h
index 71faf119cf0..bbea1119d73 100644
--- a/storage/innobase/include/mtr0log.h
+++ b/storage/innobase/include/mtr0log.h
@@ -1,7 +1,6 @@
 /*****************************************************************************
 
-Copyright (c) 1995, 2014, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, 2020, MariaDB Corporation.
+Copyright (c) 2019, 2020, 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
@@ -17,21 +16,13 @@ this program; if not, write to the Free Software Foundation, Inc.,
 
 *****************************************************************************/
 
-/**************************************************//**
+/**
 @file include/mtr0log.h
-Mini-transaction logging routines
-
-Created 12/7/1995 Heikki Tuuri
+Mini-transaction log record encoding and decoding
 *******************************************************/
 
-#ifndef mtr0log_h
-#define mtr0log_h
-
+#pragma once
 #include "mtr0mtr.h"
-#include "dyn0buf.h"
-
-// Forward declaration
-struct dict_index_t;
 
 /** The minimum 2-byte integer (0b10xxxxxx xxxxxxxx) */
 constexpr uint32_t MIN_2BYTE= 1 << 7;
@@ -543,53 +534,3 @@ inline void mtr_t::page_create(const buf_block_t &block, bool comp)
   m_log.close(l);
   m_last_offset= FIL_PAGE_TYPE;
 }
-
-/********************************************************//**
-Parses an initial log record written by mlog_write_initial_log_record_low().
-@return parsed record end, NULL if not a complete record */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-mlog_parse_initial_log_record(
-/*==========================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	mlog_id_t*	type,	/*!< out: log record type: MLOG_1BYTE, ... */
-	ulint*		space,	/*!< out: space id */
-	ulint*		page_no);/*!< out: page number */
-/********************************************************//**
-Parses a log record written by mtr_t::write(), mtr_t::memset().
-@return parsed record end, NULL if not a complete record */
-const byte*
-mlog_parse_nbytes(
-/*==============*/
-	mlog_id_t	type,	/*!< in: log record type: MLOG_1BYTE, ... */
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	byte*		page,	/*!< in: page where to apply the log record,
-				or NULL */
-	void*		page_zip);/*!< in/out: compressed page, or NULL */
-/********************************************************//**
-Parses a log record written by mtr_t::memcpy().
-@return parsed record end, NULL if not a complete record */
-const byte*
-mlog_parse_string(
-/*==============*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	byte*		page,	/*!< in: page where to apply the log record,
-				or NULL */
-	void*		page_zip);/*!< in/out: compressed page, or NULL */
-
-/********************************************************//**
-Parses a log record written by mlog_open_and_write_index.
-@return parsed record end, NULL if not a complete record */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-mlog_parse_index(
-/*=============*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	bool		comp,	/*!< in: TRUE=compact record format */
-	dict_index_t**	index);	/*!< out, own: dummy index */
-
-#endif /* mtr0log_h */
diff --git a/storage/innobase/include/mtr0types.h b/storage/innobase/include/mtr0types.h
index bdcca691c2e..aa5469b085a 100644
--- a/storage/innobase/include/mtr0types.h
+++ b/storage/innobase/include/mtr0types.h
@@ -276,201 +276,6 @@ terminate the mini-transaction. */
 constexpr byte SIZE_OF_FILE_CHECKPOINT= 3/*type,page_id*/ + 8/*LSN*/ + 1;
 #endif
 
-/** @name Log item types
-The log items are declared 'byte' so that the compiler can warn if val
-and type parameters are switched in a call to mlog_write. NOTE!
-For 1 - 8 bytes, the flag value must give the length also! @{ */
-enum mlog_id_t {
-	/** if the mtr contains only one log record for one page,
-	i.e., write_initial_log_record has been called only once,
-	this flag is ORed to the type of that first log record */
-	MLOG_SINGLE_REC_FLAG = 128,
-
-	/** one byte is written */
-	MLOG_1BYTE = 1,
-
-	/** 2 bytes ... */
-	MLOG_2BYTES = 2,
-
-	/** 4 bytes ... */
-	MLOG_4BYTES = 4,
-
-	/** 8 bytes ... */
-	MLOG_8BYTES = 8,
-
-	/** Record insert */
-	MLOG_REC_INSERT = 9,
-
-	/** Mark clustered index record deleted */
-	MLOG_REC_CLUST_DELETE_MARK = 10,
-
-	/** Mark secondary index record deleted */
-	MLOG_REC_SEC_DELETE_MARK = 11,
-
-	/** update of a record, preserves record field sizes */
-	MLOG_REC_UPDATE_IN_PLACE = 13,
-
-	/*!< Delete a record from a page */
-	MLOG_REC_DELETE = 14,
-
-	/** Delete record list end on index page */
-	MLOG_LIST_END_DELETE = 15,
-
-	/** Delete record list start on index page */
-	MLOG_LIST_START_DELETE = 16,
-
-	/** Copy record list end to a new created index page */
-	MLOG_LIST_END_COPY_CREATED = 17,
-
-	/** Reorganize an index page in ROW_FORMAT=REDUNDANT */
-	MLOG_PAGE_REORGANIZE = 18,
-
-	/** Create an index page */
-	MLOG_PAGE_CREATE = 19,
-
-	/** insert an undo log record */
-	MLOG_UNDO_INSERT = 20,
-
-	/** erase an undo log page end (used in MariaDB 10.2) */
-	MLOG_UNDO_ERASE_END = 21,
-
-	/** initialize a page in an undo log */
-	MLOG_UNDO_INIT = 22,
-
-	/** reuse an insert undo log header (used in MariaDB 10.2) */
-	MLOG_UNDO_HDR_REUSE = 24,
-
-	/** create an undo log header */
-	MLOG_UNDO_HDR_CREATE = 25,
-
-	/** mark an index record as the predefined minimum record */
-	MLOG_REC_MIN_MARK = 26,
-
-	/** initialize an ibuf bitmap page (used in MariaDB 10.2 and 10.3) */
-	MLOG_IBUF_BITMAP_INIT = 27,
-
-	/** write a string to a page */
-	MLOG_WRITE_STRING = 30,
-
-	/** If a single mtr writes several log records, this log
-	record ends the sequence of these records */
-	MLOG_MULTI_REC_END = 31,
-
-	/** dummy log record used to pad a log block full */
-	MLOG_DUMMY_RECORD = 32,
-
-	/** log record about an .ibd file creation */
-	//MLOG_FILE_CREATE = 33,
-
-	/** rename databasename/tablename (no .ibd file name suffix) */
-	//MLOG_FILE_RENAME = 34,
-
-	/** delete a tablespace file that starts with (space_id,page_no) */
-	MLOG_FILE_DELETE = 35,
-
-	/** mark a compact index record as the predefined minimum record */
-	MLOG_COMP_REC_MIN_MARK = 36,
-
-	/** create a compact index page */
-	MLOG_COMP_PAGE_CREATE = 37,
-
-	/** compact record insert */
-	MLOG_COMP_REC_INSERT = 38,
-
-	/** mark compact clustered index record deleted */
-	MLOG_COMP_REC_CLUST_DELETE_MARK = 39,
-
-	/** update of a compact record, preserves record field sizes */
-	MLOG_COMP_REC_UPDATE_IN_PLACE = 41,
-
-	/** delete a compact record from a page */
-	MLOG_COMP_REC_DELETE = 42,
-
-	/** delete compact record list end on index page */
-	MLOG_COMP_LIST_END_DELETE = 43,
-
-	/*** delete compact record list start on index page */
-	MLOG_COMP_LIST_START_DELETE = 44,
-
-	/** copy compact record list end to a new created index page */
-	MLOG_COMP_LIST_END_COPY_CREATED = 45,
-
-	/** reorganize an index page */
-	MLOG_COMP_PAGE_REORGANIZE = 46,
-
-	/** log record about creating an .ibd file, with format */
-	MLOG_FILE_CREATE2 = 47,
-
-	/** write the node pointer of a record on a compressed
-	non-leaf B-tree page */
-	MLOG_ZIP_WRITE_NODE_PTR = 48,
-
-	/** write the BLOB pointer of an externally stored column
-	on a compressed page */
-	MLOG_ZIP_WRITE_BLOB_PTR = 49,
-
-	/** write to compressed page header */
-	MLOG_ZIP_WRITE_HEADER = 50,
-
-	/** compress an index page */
-	MLOG_ZIP_PAGE_COMPRESS = 51,
-
-	/** compress an index page without logging it's image */
-	MLOG_ZIP_PAGE_COMPRESS_NO_DATA = 52,
-
-	/** reorganize a compressed page */
-	MLOG_ZIP_PAGE_REORGANIZE = 53,
-
-	/** rename a tablespace file that starts with (space_id,page_no) */
-	MLOG_FILE_RENAME2 = 54,
-
-	/** note the first use of a tablespace file since checkpoint */
-	MLOG_FILE_NAME = 55,
-
-	/** note that all buffered log was written since a checkpoint */
-	MLOG_CHECKPOINT = 56,
-
-	/** Create a R-Tree index page */
-	MLOG_PAGE_CREATE_RTREE = 57,
-
-	/** create a R-tree compact page */
-	MLOG_COMP_PAGE_CREATE_RTREE = 58,
-
-	/** initialize a file page */
-	MLOG_INIT_FILE_PAGE2 = 59,
-
-	/** Table is being truncated. (Was used in 10.2 and 10.3;
-	not supported for crash-upgrade to 10.4 or later.) */
-	MLOG_TRUNCATE = 60,
-
-	/** write DB_TRX_ID,DB_ROLL_PTR to a clustered index leaf page
-	of a ROW_FORMAT=COMPRESSED table */
-	MLOG_ZIP_WRITE_TRX_ID = 62,
-
-	/** initialize a page with a string of identical bytes */
-	MLOG_MEMSET = 63,
-
-	/** Zero-fill a page that is not allocated. */
-	MLOG_INIT_FREE_PAGE = 64,
-
-	/** biggest value (used in assertions) */
-	MLOG_BIGGEST_TYPE = MLOG_INIT_FREE_PAGE,
-
-	/** log record for writing/updating crypt data of
-	a tablespace */
-	MLOG_FILE_WRITE_CRYPT_DATA = 100,
-};
-
-/* @} */
-
-#define EXTRA_CHECK_MLOG_NUMBER(x) \
-  ((x) == MLOG_FILE_WRITE_CRYPT_DATA)
-
-/** Size of a MLOG_CHECKPOINT record in bytes.
-The record consists of a MLOG_CHECKPOINT byte followed by
-mach_write_to_8(checkpoint_lsn). */
-#define SIZE_OF_MLOG_CHECKPOINT	9
-
 #ifndef UNIV_INNOCHECKSUM
 /** Types for the mlock objects to store in the mtr_t::m_memo */
 enum mtr_memo_type_t {
diff --git a/storage/innobase/include/page0cur.h b/storage/innobase/include/page0cur.h
index 669321c999b..36a401cf0db 100644
--- a/storage/innobase/include/page0cur.h
+++ b/storage/innobase/include/page0cur.h
@@ -285,45 +285,6 @@ page_cur_open_on_rnd_user_rec(
 /*==========================*/
 	buf_block_t*	block,	/*!< in: page */
 	page_cur_t*	cursor);/*!< out: page cursor */
-/***********************************************************//**
-Parses a log record of a record insert on a page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_cur_parse_insert_rec(
-/*======================*/
-	bool		is_short,/*!< in: true if short inserts */
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr);	/*!< in: mtr or NULL */
-/**********************************************************//**
-Parses a log record of copying a record list end to a new created page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_parse_copy_rec_list_to_created_page(
-/*=====================================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr);	/*!< in: mtr or NULL */
-/***********************************************************//**
-Parses log record of a record delete on a page.
-@return pointer to record end or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_cur_parse_delete_rec(
-/*======================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction,
-				or NULL if block=NULL */
-	MY_ATTRIBUTE((warn_unused_result, nonnull(1,2,4)));
 
 /** Index page cursor */
 
diff --git a/storage/innobase/include/page0page.h b/storage/innobase/include/page0page.h
index 46981c777cd..c3b80e3e196 100644
--- a/storage/innobase/include/page0page.h
+++ b/storage/innobase/include/page0page.h
@@ -1063,22 +1063,6 @@ page_move_rec_list_start(
 	dict_index_t*	index,		/*!< in: record descriptor */
 	mtr_t*		mtr)		/*!< in: mtr */
 	MY_ATTRIBUTE((nonnull(1, 2, 4, 5)));
-/**********************************************************//**
-Parses a log record of a record list end or start deletion.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_parse_delete_rec_list(
-/*=======================*/
-	mlog_id_t	type,	/*!< in: MLOG_LIST_END_DELETE,
-				MLOG_LIST_START_DELETE,
-				MLOG_COMP_LIST_END_DELETE or
-				MLOG_COMP_LIST_START_DELETE */
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: buffer block or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr);	/*!< in: mtr or NULL */
 /** Create an index page.
 @param[in,out]	block	buffer block
 @param[in]	comp	nonzero=compact page format */
diff --git a/storage/innobase/include/page0zip.h b/storage/innobase/include/page0zip.h
index 8b8a4e5b984..459f4b9cc77 100644
--- a/storage/innobase/include/page0zip.h
+++ b/storage/innobase/include/page0zip.h
@@ -243,18 +243,6 @@ void page_zip_write_rec(buf_block_t *block, const byte *rec,
                         ulint create, mtr_t *mtr)
   MY_ATTRIBUTE((nonnull));
 
-/***********************************************************//**
-Parses a log record of writing a BLOB pointer of a record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_blob_ptr(
-/*==========================*/
-	const byte*	ptr,	/*!< in: redo log buffer */
-	const byte*	end_ptr,/*!< in: redo log buffer end */
-	page_t*		page,	/*!< in/out: uncompressed page */
-	page_zip_des_t*	page_zip);/*!< in/out: compressed page */
-
 /**********************************************************************//**
 Write a BLOB pointer of a record on the leaf page of a clustered index.
 The information must already have been updated on the uncompressed page. */
@@ -270,18 +258,6 @@ page_zip_write_blob_ptr(
 	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 	MY_ATTRIBUTE((nonnull));
 
-/***********************************************************//**
-Parses a log record of writing the node pointer of a record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_node_ptr(
-/*==========================*/
-	const byte*	ptr,	/*!< in: redo log buffer */
-	const byte*	end_ptr,/*!< in: redo log buffer end */
-	page_t*		page,	/*!< in/out: uncompressed page */
-	page_zip_des_t*	page_zip);/*!< in/out: compressed page */
-
 /**********************************************************************//**
 Write the node pointer of a record on a non-leaf compressed page. */
 void
@@ -313,22 +289,6 @@ page_zip_write_trx_id_and_roll_ptr(
 	mtr_t*		mtr)
 	MY_ATTRIBUTE((nonnull));
 
-/** Parse a MLOG_ZIP_WRITE_TRX_ID record.
-@param[in]	ptr		redo log buffer
-@param[in]	end_ptr		end of redo log buffer
-@param[in,out]	page		uncompressed page
-@param[in,out]	page_zip	compressed page
-@return end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_trx_id(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	page_t*		page,
-	page_zip_des_t*	page_zip)
-	MY_ATTRIBUTE((nonnull(1,2), warn_unused_result));
-
 /** Modify the delete-mark flag of a ROW_FORMAT=COMPRESSED record.
 @param[in,out]  block   buffer block
 @param[in,out]  rec     record on a physical index page
@@ -363,22 +323,10 @@ void page_zip_dir_delete(buf_block_t *block, byte *rec,
                          const byte *free, mtr_t *mtr)
   MY_ATTRIBUTE((nonnull(1,2,3,4,6)));
 
-/***********************************************************//**
-Parses a log record of writing to the header of a page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_header(
-/*========================*/
-	const byte*	ptr,	/*!< in: redo log buffer */
-	const byte*	end_ptr,/*!< in: redo log buffer end */
-	page_t*		page,	/*!< in/out: uncompressed page */
-	page_zip_des_t*	page_zip);/*!< in/out: compressed page */
-
 /**********************************************************************//**
 Reorganize and compress a page.  This is a low-level operation for
 compressed pages, to be used when page_zip_compress() fails.
-On success, a redo log entry MLOG_ZIP_PAGE_COMPRESS will be written.
+On success, redo log will be written.
 The function btr_page_reorganize() should be preferred whenever possible.
 IMPORTANT: if page_zip_reorganize() is invoked on a leaf page of a
 non-clustered index, the caller must update the insert buffer free
@@ -410,17 +358,6 @@ page_zip_copy_recs(
 	const page_t*		src,		/*!< in: page */
 	dict_index_t*		index,		/*!< in: index of the B-tree */
 	mtr_t*			mtr);		/*!< in: mini-transaction */
-
-/** Parse and optionally apply MLOG_ZIP_PAGE_COMPRESS.
-@param[in]	ptr	log record
-@param[in]	end_ptr	end of log
-@param[in,out]	block	ROW_FORMAT=COMPRESSED block, or NULL for parsing only
-@return	end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte* page_zip_parse_compress(const byte* ptr, const byte* end_ptr,
-				    buf_block_t* block);
-
 #endif /* !UNIV_INNOCHECKSUM */
 
 /** Calculate the compressed page checksum.
diff --git a/storage/innobase/include/rem0types.h b/storage/innobase/include/rem0types.h
index 754781d62d1..5a9f5945d6a 100644
--- a/storage/innobase/include/rem0types.h
+++ b/storage/innobase/include/rem0types.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2019, MariaDB Corporation.
+Copyright (c) 2019, 2020, 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
@@ -40,11 +40,12 @@ typedef unsigned short int offset_t;
 
 /* Maximum number of user defined fields/columns. The reserved columns
 are the ones InnoDB adds internally: DB_ROW_ID, DB_TRX_ID, DB_ROLL_PTR.
-We need "* 2" because mlog_parse_index() creates a dummy table object
-possibly, with some of the system columns in it, and then adds the 3
-system columns (again) using dict_table_add_system_columns(). The problem
-is that mlog_parse_index() cannot recognize the system columns by
-just having n_fields, n_uniq and the lengths of the columns. */
+Before MariaDB Server 10.5, we needed "* 2" because mlog_parse_index()
+created a dummy table object possibly, with some of the system columns
+in it, and then adds the 3 system columns (again) using
+dict_table_add_system_columns().
+For now, we will keep this limitation to maintain file format compatibility
+with older versions. */
 #define REC_MAX_N_USER_FIELDS	(REC_MAX_N_FIELDS - DATA_N_SYS_COLS * 2)
 
 /* REC_ANTELOPE_MAX_INDEX_COL_LEN is measured in bytes and is the maximum
diff --git a/storage/innobase/include/row0upd.h b/storage/innobase/include/row0upd.h
index 86cb18231bc..68692b8e86f 100644
--- a/storage/innobase/include/row0upd.h
+++ b/storage/innobase/include/row0upd.h
@@ -316,19 +316,6 @@ que_thr_t*
 row_upd_step(
 /*=========*/
 	que_thr_t*	thr);	/*!< in: query thread */
-/*********************************************************************//**
-Parses the log data written by row_upd_index_write_log.
-@return log data end or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-row_upd_index_parse(
-/*================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	mem_heap_t*	heap,	/*!< in: memory heap where update vector is
-				built */
-	upd_t**		update_out);/*!< out: update vector */
-
 
 /* Update vector field */
 struct upd_field_t{
diff --git a/storage/innobase/include/trx0rec.h b/storage/innobase/include/trx0rec.h
index 6ee7a49cf10..467968553fe 100644
--- a/storage/innobase/include/trx0rec.h
+++ b/storage/innobase/include/trx0rec.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -240,19 +240,6 @@ trx_undo_prev_version_build(
 				into this function by purge thread or not.
 				And if we read "after image" of undo log */
 
-/** Parse MLOG_UNDO_INSERT.
-@param[in]	ptr	log record
-@param[in]	end_ptr	end of log record buffer
-@param[in,out]	page	page or NULL
-@return	end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_add_undo_rec(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	page_t*		page);
-
 /** Read from an undo log record a non-virtual column value.
 @param[in,out]	ptr		pointer to remaining part of the undo record
 @param[in,out]	field		stored field
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index f65681bc2e8..b68f34ff4da 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2020, 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
@@ -250,41 +250,6 @@ trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp);
 void
 trx_undo_free_at_shutdown(trx_t *trx);
 
-/** Parse MLOG_UNDO_INIT.
-@param[in]	ptr	log record
-@param[in]	end_ptr	end of log record buffer
-@param[in,out]	page	page or NULL
-@param[in,out]	mtr	mini-transaction
-@return	end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_page_init(const byte* ptr, const byte* end_ptr, page_t* page);
-/** Parse MLOG_UNDO_HDR_REUSE for crash-upgrade from MariaDB 10.2.
-@param[in]	ptr	redo log record
-@param[in]	end_ptr	end of log buffer
-@param[in,out]	page	undo page or NULL
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_page_header_reuse(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	page_t*		page);
-
-/** Parse the redo log entry of an undo log page header create.
-@param[in]	ptr	redo log record
-@param[in]	end_ptr	end of log buffer
-@param[in,out]	block	page frame or NULL
-@param[in,out]	mtr	mini-transaction or NULL
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_page_header(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	buf_block_t*	block,
-	mtr_t*		mtr);
 /** Read an undo log when starting up the database.
 @param[in,out]	rseg		rollback segment
 @param[in]	id		rollback segment slot
diff --git a/storage/innobase/log/log0log.cc b/storage/innobase/log/log0log.cc
index 37742bb2008..5a9975f5429 100644
--- a/storage/innobase/log/log0log.cc
+++ b/storage/innobase/log/log0log.cc
@@ -58,28 +58,9 @@ Created 12/9/1995 Heikki Tuuri
 /*
 General philosophy of InnoDB redo-logs:
 
-1) Every change to a contents of a data page must be done
-through mtr, which in mtr_commit() writes log records
-to the InnoDB redo log.
-
-2) Normally these changes are performed using a mlog_write()
-or similar function.
-
-3) In some page level operations only a code number of a
-c-function and its parameters are written to the log to
-reduce the size of the log.
-
-  3a) You should not add parameters to these kind of functions
-  (e.g. trx_undo_header_create())
-
-  3b) You should not add such functionality which either change
-  working when compared with the old or are dependent on data
-  outside of the page. These kind of functions should implement
-  self-contained page transformation and it should be unchanged
-  if you don't have very essential reasons to change log
-  semantics or format.
-
-*/
+Every change to a contents of a data page must be done
+through mtr_t, and mtr_t::commit() will write log records
+to the InnoDB redo log. */
 
 /** Redo log system */
 log_t	log_sys;
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 9a229d4bb20..37fd8681881 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -90,15 +90,6 @@ TRUE means that recovery is running and no operations on the log files
 are allowed yet: the variable name is misleading. */
 bool	recv_no_ibuf_operations;
 
-#if 1 /* MDEV-12353: only for parsing old redo log format */
-/** The type of the previous parsed redo log record */
-static mlog_id_t	recv_previous_parsed_rec_type;
-/** The offset of the previous parsed redo log record */
-static ulint	recv_previous_parsed_rec_offset;
-/** The 'multi' flag of the previous parsed redo log record */
-static ulint	recv_previous_parsed_rec_is_multi;
-#endif
-
 /** The maximum lsn we see for a page during the recovery process. If this
 is bigger than the lsn we are able to scan up to, that is an indication that
 the recovery failed and the database may be corrupt. */
@@ -111,69 +102,6 @@ mysql_pfs_key_t	recv_writer_thread_key;
 /** Is recv_writer_thread active? */
 bool	recv_writer_thread_active;
 
-
-/** Stored physiological log record with byte-oriented start/end LSN
-(before log_t::FORMAT_10_5) */
-struct recv_t : public log_rec_t
-{
-  /**
-    Constructor.
-    @param len       total length of the redo log record body, in bytes
-    @param type      redo log record chunk
-    @param start_lsn start LSN of the mini-transaction
-    @param end_lsn   end LSN of the mini-transaction
-  */
-  recv_t(uint32_t len, mlog_id_t type, lsn_t start_lsn, lsn_t end_lsn) :
-    log_rec_t(end_lsn), start_lsn(start_lsn), len(len), type(type), data(NULL)
-  {}
-
-  /** start LSN of the mini-transaction (not necessarily of this record) */
-  const lsn_t start_lsn;
-  /** log record body length in bytes */
-  const uint32_t len;
-  /** log record type */
-  const mlog_id_t type;
-  /** log record */
-  struct data_t
-  {
-    /** pointer to the next chunk, or NULL for the last chunk.  The
-    log record data is stored immediately after this field. */
-    data_t *next= NULL;
-
-    data_t() {}
-    /**
-      Constructor.
-      @param chunk   redo log record chunk
-      @param len     length of the chunk, in bytes
-    */
-    data_t(const void* chunk, size_t len)
-    {
-      memcpy(reinterpret_cast<void*>(this + 1), chunk, len);
-    }
-
-    /**
-      Append a log snippet.
-      @param d  log snippet
-    */
-    void append(data_t *d) { ut_ad(!next); ut_ad(!d->next); next= d; }
-  }* data;
-
-  /** Free the log snippet */
-  void free() const
-  {
-    data_t *d= data;
-    do
-    {
-      data_t *next= d->next;
-      recv_sys.free(d);
-      d= next;
-    }
-    while (d);
-    recv_sys.free(this);
-  }
-};
-
-
 /** Stored physical log record with logical LSN (@see log_t::FORMAT_10_5) */
 struct log_phys_t : public log_rec_t
 {
@@ -422,16 +350,9 @@ public:
 };
 
 
-#ifndef	DBUG_OFF
-/** Return string name of the redo log record type.
-@param[in]	type	record log record enum
-@return string name of record log record */
-static const char* get_mlog_string(mlog_id_t type);
-#endif /* !DBUG_OFF */
-
 /** Tablespace item during recovery */
 struct file_name_t {
-	/** Tablespace file name (MLOG_FILE_NAME or FILE_MODIFY) */
+	/** Tablespace file name (FILE_MODIFY) */
 	std::string	name;
 	/** Tablespace object (NULL if not valid or not found) */
 	fil_space_t*	space;
@@ -625,12 +546,11 @@ inline void recv_sys_t::trim(const page_id_t page_id, lsn_t lsn)
 	DBUG_VOID_RETURN;
 }
 
-/** Process a file name from a MLOG_FILE_* or FILE_* record.
+/** Process a file name from a FILE_* record.
 @param[in,out]	name		file name
 @param[in]	len		length of the file name
 @param[in]	space_id	the tablespace ID
-@param[in]	deleted		whether this is a MLOG_FILE_DELETE
-				or FILE_DELETE record */
+@param[in]	deleted		whether this is a FILE_DELETE record */
 static
 void
 fil_name_process(char* name, ulint len, ulint space_id, bool deleted)
@@ -644,7 +564,7 @@ fil_name_process(char* name, ulint len, ulint space_id, bool deleted)
 	      || srv_operation == SRV_OPERATION_RESTORE_EXPORT);
 
 	/* We will also insert space=NULL into the map, so that
-	further checks can ensure that a MLOG_FILE_NAME record was
+	further checks can ensure that a FILE_MODIFY record was
 	scanned before applying any page records for the space_id. */
 
 	os_normalize_path(name);
@@ -656,7 +576,7 @@ fil_name_process(char* name, ulint len, ulint space_id, bool deleted)
 	file_name_t&	f = p.first->second;
 
 	if (deleted) {
-		/* Got MLOG_FILE_DELETE oR FILE_DELETE */
+		/* Got FILE_DELETE */
 
 		if (!p.second && f.status != file_name_t::DELETED) {
 			f.status = file_name_t::DELETED;
@@ -667,9 +587,7 @@ fil_name_process(char* name, ulint len, ulint space_id, bool deleted)
 		}
 
 		ut_ad(f.space == NULL);
-	} else if (p.second
-		   /* the first MLOG_FILE_NAME or MLOG_FILE_RENAME2
-		   or FILE_MODIFY or FILE_RENAME */
+	} else if (p.second // the first FILE_MODIFY or FILE_RENAME
 		   || f.name != fname.name) {
 		fil_space_t*	space;
 
@@ -706,7 +624,7 @@ fil_name_process(char* name, ulint len, ulint space_id, bool deleted)
 		case FIL_LOAD_NOT_FOUND:
 			/* No matching tablespace was found; maybe it
 			was renamed, and we will find a subsequent
-			MLOG_FILE_* or FILE_* record. */
+			FILE_* record. */
 			ut_ad(space == NULL);
 
 			if (srv_force_recovery) {
@@ -760,184 +678,6 @@ fil_name_process(char* name, ulint len, ulint space_id, bool deleted)
 	}
 }
 
-/** Parse or process a MLOG_FILE_* record.
-@param[in,out]	ptr		redo log record
-@param[in]	end		end of the redo log buffer
-@param[in]	page_id		first page number in the file
-@param[in]	type		MLOG_FILE_NAME or MLOG_FILE_DELETE
-or MLOG_FILE_CREATE2 or MLOG_FILE_RENAME2
-@param[in]	apply		whether to apply the record
-@return pointer to next redo log record
-@retval NULL if this log record was truncated */
-static
-const byte*
-fil_name_parse(
-	byte*		ptr,
-	const byte*	end,
-	const page_id_t	page_id,
-	mlog_id_t	type,
-	bool		apply)
-{
-	if (type == MLOG_FILE_CREATE2) {
-		if (end < ptr + 4) {
-			return(NULL);
-		}
-		ptr += 4;
-	}
-
-	if (end < ptr + 2) {
-		return(NULL);
-	}
-
-	ulint	len = mach_read_from_2(ptr);
-	ptr += 2;
-	if (end < ptr + len) {
-		return(NULL);
-	}
-
-	/* MLOG_FILE_* records should only be written for
-	user-created tablespaces. The name must be long enough
-	and end in .ibd. */
-	bool corrupt = is_predefined_tablespace(page_id.space())
-		|| len < sizeof "/a.ibd\0"
-		|| (!page_id.page_no() != !memcmp(ptr + len - 5, DOT_IBD, 5));
-
-	if (!corrupt && !memchr(ptr, OS_PATH_SEPARATOR, len)) {
-		if (byte* c = static_cast<byte*>
-		    (memchr(ptr, OS_PATH_SEPARATOR_ALT, len))) {
-			ut_ad(c >= ptr);
-			ut_ad(c < ptr + len);
-			do {
-				*c = OS_PATH_SEPARATOR;
-			} while ((c = static_cast<byte*>
-				  (memchr(ptr, OS_PATH_SEPARATOR_ALT,
-					  len - ulint(c - ptr)))) != NULL);
-		} else {
-			corrupt = true;
-		}
-	}
-
-	byte*	end_ptr	= ptr + len--;
-
-	switch (type) {
-	default:
-		ut_ad(0); // the caller checked this
-		/* fall through */
-	case MLOG_FILE_NAME:
-		if (corrupt) {
-			ib::error() << "MLOG_FILE_NAME incorrect:" << ptr;
-			recv_sys.found_corrupt_log = true;
-			break;
-		}
-
-		fil_name_process(
-			reinterpret_cast<char*>(ptr), len, page_id.space(),
-			false);
-		break;
-	case MLOG_FILE_DELETE:
-		if (corrupt) {
-			ib::error() << "MLOG_FILE_DELETE incorrect:" << ptr;
-			recv_sys.found_corrupt_log = true;
-			break;
-		}
-
-		fil_name_process(reinterpret_cast<char*>(ptr), len,
-				 page_id.space(), true);
-		/* fall through */
-	case MLOG_FILE_CREATE2:
-		if (page_id.page_no()) {
-			ut_ad(page_id.page_no()
-			      == SRV_UNDO_TABLESPACE_SIZE_IN_PAGES);
-			ut_a(srv_is_undo_tablespace(page_id.space()));
-			compile_time_assert(
-				UT_ARR_SIZE(recv_sys.truncated_undo_spaces)
-				== TRX_SYS_MAX_UNDO_SPACES);
-			recv_sys_t::trunc& t = recv_sys.truncated_undo_spaces[
-				page_id.space() - srv_undo_space_id_start];
-			t.lsn = recv_sys.recovered_lsn;
-			t.pages = uint32_t(page_id.page_no());
-		} else if (log_file_op) {
-			log_file_op(page_id.space(),
-				    type == MLOG_FILE_CREATE2,
-				    ptr, len, NULL, 0);
-		}
-		break;
-	case MLOG_FILE_RENAME2:
-		if (corrupt) {
-			ib::error() << "MLOG_FILE_RENAME2 incorrect:" << ptr;
-			recv_sys.found_corrupt_log = true;
-		}
-
-		/* The new name follows the old name. */
-		byte*	new_name = end_ptr + 2;
-		if (end < new_name) {
-			return(NULL);
-		}
-
-		ulint	new_len = mach_read_from_2(end_ptr);
-
-		if (end < end_ptr + 2 + new_len) {
-			return(NULL);
-		}
-
-		end_ptr += 2 + new_len;
-
-		corrupt = corrupt
-			|| new_len < sizeof "/a.ibd\0"
-			|| memcmp(new_name + new_len - 5, DOT_IBD, 5) != 0;
-		new_len--;
-
-		if (!corrupt && !memchr(new_name, OS_PATH_SEPARATOR, new_len)) {
-			if (byte* c = static_cast<byte*>
-			    (memchr(new_name, OS_PATH_SEPARATOR_ALT,
-				    new_len))) {
-				ut_ad(c >= new_name);
-				ut_ad(c < new_name + new_len);
-				do {
-					*c = OS_PATH_SEPARATOR;
-				} while ((c = static_cast<byte*>
-					  (memchr(ptr, OS_PATH_SEPARATOR_ALT,
-						  new_len
-						  - ulint(c - new_name))))
-					 != NULL);
-			} else {
-				corrupt = true;
-			}
-		}
-
-		if (corrupt) {
-			ib::error() << "MLOG_FILE_RENAME2 new_name incorrect:" << ptr
-				    << " new_name: " << new_name;
-			recv_sys.found_corrupt_log = true;
-			break;
-		}
-
-		fil_name_process(
-			reinterpret_cast<char*>(ptr), len,
-			page_id.space(), false);
-		fil_name_process(
-			reinterpret_cast<char*>(new_name), new_len,
-			page_id.space(), false);
-
-		if (log_file_op) {
-			log_file_op(page_id.space(), false,
-				    ptr, len, new_name, new_len);
-		}
-
-		if (!apply) {
-			break;
-		}
-		if (!fil_op_replay_rename(
-			    page_id.space(), page_id.page_no(),
-			    reinterpret_cast<const char*>(ptr),
-			    reinterpret_cast<const char*>(new_name))) {
-			recv_sys.found_corrupt_fs = true;
-		}
-	}
-
-	return(end_ptr);
-}
-
 /** Clean up after recv_sys_t::create() */
 void recv_sys_t::close()
 {
@@ -982,9 +722,6 @@ recv_sys_var_init(void)
 	recv_needed_recovery = false;
 	recv_lsn_checks_on = false;
 	recv_no_ibuf_operations = false;
-	recv_previous_parsed_rec_type = MLOG_SINGLE_REC_FLAG;
-	recv_previous_parsed_rec_offset	= 0;
-	recv_previous_parsed_rec_is_multi = 0;
 	recv_max_page_lsn = 0;
 }
 
@@ -1128,7 +865,7 @@ void recv_sys_t::debug_free()
 	mutex_exit(&mutex);
 }
 
-inline void *recv_sys_t::alloc(size_t len, bool store_recv)
+inline void *recv_sys_t::alloc(size_t len)
 {
   ut_ad(mutex_own(&mutex));
   ut_ad(len);
@@ -1490,6 +1227,61 @@ static dberr_t recv_log_format_0_recover(lsn_t lsn, bool crypt)
 	return(DB_SUCCESS);
 }
 
+/** Determine if a redo log from MariaDB 10.2.2+, 10.3, or 10.4 is clean.
+@return	error code
+@retval	DB_SUCCESS	if the redo log is clean
+@retval	DB_CORRUPTION	if the redo log is corrupted
+@retval	DB_ERROR	if the redo log is not empty */
+static dberr_t recv_log_recover_10_4()
+{
+	const lsn_t	lsn = log_sys.log.get_lsn();
+	const lsn_t	source_offset = log_sys.log.calc_lsn_offset(lsn);
+	byte*		buf = log_sys.buf;
+
+	log_sys.log.read(source_offset & ~(OS_FILE_LOG_BLOCK_SIZE - 1),
+			 {buf, OS_FILE_LOG_BLOCK_SIZE});
+
+	ulint crc = log_block_calc_checksum_crc32(buf);
+	ulint cksum = log_block_get_checksum(buf);
+
+	if (crc != cksum) {
+		ib::error() << "Invalid log block checksum."
+			    << " block: "
+			    << log_block_get_hdr_no(buf)
+			    << " checkpoint no: "
+			    << log_block_get_checkpoint_no(buf)
+			    << " expected: " << crc
+			    << " found: " << cksum;
+		return DB_CORRUPTION;
+	}
+
+	if (log_sys.log.is_encrypted()
+	    && !log_crypt(buf, lsn & (OS_FILE_LOG_BLOCK_SIZE - 1),
+			  OS_FILE_LOG_BLOCK_SIZE, LOG_DECRYPT)) {
+		return DB_ERROR;
+	}
+
+	/* On a clean shutdown, the redo log will be logically empty
+	after the checkpoint lsn. */
+
+	if (log_block_get_data_len(buf)
+	    != (source_offset & (OS_FILE_LOG_BLOCK_SIZE - 1))) {
+		return DB_ERROR;
+	}
+
+	/* Mark the redo log for upgrading. */
+	srv_log_file_size = 0;
+	recv_sys.parse_start_lsn = recv_sys.recovered_lsn
+		= recv_sys.scanned_lsn
+		= recv_sys.mlog_checkpoint_lsn = lsn;
+	log_sys.last_checkpoint_lsn = log_sys.next_checkpoint_lsn
+		= log_sys.lsn = log_sys.write_lsn
+		= log_sys.current_flush_lsn = log_sys.flushed_to_disk_lsn
+		= lsn;
+	log_sys.next_checkpoint_no = 0;
+	return DB_SUCCESS;
+}
+
 /** Find the latest checkpoint in the log header.
 @param[out]	max_field	LOG_CHECKPOINT_1 or LOG_CHECKPOINT_2
 @return error code or DB_SUCCESS */
@@ -1600,503 +1392,22 @@ recv_find_max_checkpoint(ulint* max_field)
 		return(DB_ERROR);
 	}
 
-	return(DB_SUCCESS);
-}
-
-/** Try to parse a single log record body and also applies it if
-specified.
-@param[in]	type		redo log entry type
-@param[in]	ptr		redo log record body
-@param[in]	end_ptr		end of buffer
-@param[in]	page_id		page identifier
-@param[in]	apply		whether to apply the record
-@param[in,out]	block		buffer block, or NULL if
-a page log record should not be applied
-or if it is a MLOG_FILE_ operation
-@param[in,out]	mtr		mini-transaction, or NULL if
-a page log record should not be applied
-@return log record end, NULL if not a complete record */
-static
-const byte*
-recv_parse_or_apply_log_rec_body(
-	mlog_id_t	type,
-	const byte*	ptr,
-	const byte*	end_ptr,
-	const page_id_t	page_id,
-	bool		apply,
-	buf_block_t*	block,
-	mtr_t*		mtr)
-{
-	ut_ad(!block == !mtr);
-	ut_ad(!apply || recv_sys.mlog_checkpoint_lsn);
-
-	switch (type) {
-	case MLOG_FILE_NAME:
-	case MLOG_FILE_DELETE:
-	case MLOG_FILE_CREATE2:
-	case MLOG_FILE_RENAME2:
-		ut_ad(block == NULL);
-		/* Collect the file names when parsing the log,
-		before applying any log records. */
-		return fil_name_parse(const_cast<byte*>(ptr), end_ptr,
-				      page_id, type, apply);
-	case MLOG_TRUNCATE:
-		ib::error() << "Cannot crash-upgrade from "
-			"old-style TRUNCATE TABLE";
-		recv_sys.found_corrupt_log = true;
-		return NULL;
-	default:
-		break;
-	}
-
-	dict_index_t*	index	= NULL;
-	page_t*		page;
-	page_zip_des_t*	page_zip;
-#ifdef UNIV_DEBUG
-	uint16_t	page_type;
-#endif /* UNIV_DEBUG */
-
-	if (block) {
-		/* Applying a page log record. */
-		ut_ad(apply);
-		page = block->frame;
-		page_zip = buf_block_get_page_zip(block);
-		ut_d(page_type = fil_page_get_type(page));
-	} else if (apply
-		   && !is_predefined_tablespace(page_id.space())
-		   && recv_spaces.find(page_id.space()) == recv_spaces.end()) {
-		if (recv_sys.recovered_lsn < recv_sys.mlog_checkpoint_lsn) {
-			/* We have not seen all records between the
-			checkpoint and MLOG_CHECKPOINT. There should be
-			a MLOG_FILE_DELETE for this tablespace later. */
-			recv_spaces.insert(
-				std::make_pair(page_id.space(),
-					       file_name_t("", false)));
-			goto parse_log;
-		}
-
-		ib::error() << "Missing MLOG_FILE_NAME or MLOG_FILE_DELETE"
-			" for redo log record " << type << page_id << " at "
-			    << recv_sys.recovered_lsn << ".";
-		recv_sys.found_corrupt_log = true;
-		return(NULL);
-	} else {
-parse_log:
-		/* Parsing a page log record. */
-		page = NULL;
-		page_zip = NULL;
-		ut_d(page_type = FIL_PAGE_TYPE_ALLOCATED);
-	}
-
-	const byte*	old_ptr = ptr;
-
-	switch (type) {
-	case MLOG_1BYTE: case MLOG_2BYTES: case MLOG_4BYTES: case MLOG_8BYTES:
-		ut_ad(!page_zip
-		      || fil_page_get_type(page_zip->data)
-		      <= FIL_PAGE_TYPE_ZBLOB2);
-		/* fall through */
-	case MLOG_MEMSET:
-#ifdef UNIV_DEBUG
-		if (page && page_type == FIL_PAGE_TYPE_ALLOCATED
-		    && end_ptr >= ptr + 2) {
-			/* It is OK to set FIL_PAGE_TYPE and certain
-			list node fields on an empty page.  Any other
-			write is not OK. */
-
-			/* NOTE: There may be bogus assertion failures for
-			dict_hdr_create(), trx_rseg_header_create(),
-			trx_sys_create_doublewrite_buf(), and
-			trx_sysf_create().
-			These are only called during database creation. */
-			ulint	offs = mach_read_from_2(ptr);
-
-			switch (type) {
-			default:
-				ut_error;
-			case MLOG_1BYTE:
-				ut_ad(offs == FIL_PAGE_TYPE + 1);
-				break;
-			case MLOG_2BYTES:
-				/* Note that this can fail when the
-				redo log been written with something
-				older than InnoDB Plugin 1.0.4. */
-				ut_ad(offs == FIL_PAGE_TYPE
-				      || srv_is_undo_tablespace(
-					      page_id.space())
-				      || offs == IBUF_TREE_SEG_HEADER
-				      + IBUF_HEADER + FSEG_HDR_OFFSET
-				      || offs == PAGE_BTR_IBUF_FREE_LIST
-				      + PAGE_HEADER + FIL_ADDR_BYTE
-				      || offs == PAGE_BTR_IBUF_FREE_LIST
-				      + PAGE_HEADER + FIL_ADDR_BYTE
-				      + FIL_ADDR_SIZE
-				      || offs == PAGE_BTR_SEG_LEAF
-				      + PAGE_HEADER + FSEG_HDR_OFFSET
-				      || offs == PAGE_BTR_SEG_TOP
-				      + PAGE_HEADER + FSEG_HDR_OFFSET
-				      || offs == PAGE_BTR_IBUF_FREE_LIST_NODE
-				      + PAGE_HEADER + FIL_ADDR_BYTE
-				      + 0 /*FLST_PREV*/
-				      || offs == PAGE_BTR_IBUF_FREE_LIST_NODE
-				      + PAGE_HEADER + FIL_ADDR_BYTE
-				      + FIL_ADDR_SIZE /*FLST_NEXT*/);
-				break;
-			case MLOG_4BYTES:
-				/* Note that this can fail when the
-				redo log been written with something
-				older than InnoDB Plugin 1.0.4. */
-				ut_ad(0
-				      /* fil_crypt_rotate_page() writes this */
-				      || offs == FIL_PAGE_SPACE_ID
-				      || srv_is_undo_tablespace(
-					      page_id.space())
-				      || offs == IBUF_TREE_SEG_HEADER
-				      + IBUF_HEADER + FSEG_HDR_SPACE
-				      || offs == IBUF_TREE_SEG_HEADER
-				      + IBUF_HEADER + FSEG_HDR_PAGE_NO
-				      || offs == PAGE_BTR_IBUF_FREE_LIST
-				      + PAGE_HEADER/* flst_init */
-				      || offs == PAGE_BTR_IBUF_FREE_LIST
-				      + PAGE_HEADER + FIL_ADDR_PAGE
-				      || offs == PAGE_BTR_IBUF_FREE_LIST
-				      + PAGE_HEADER + FIL_ADDR_PAGE
-				      + FIL_ADDR_SIZE
-				      || offs == PAGE_BTR_SEG_LEAF
-				      + PAGE_HEADER + FSEG_HDR_PAGE_NO
-				      || offs == PAGE_BTR_SEG_LEAF
-				      + PAGE_HEADER + FSEG_HDR_SPACE
-				      || offs == PAGE_BTR_SEG_TOP
-				      + PAGE_HEADER + FSEG_HDR_PAGE_NO
-				      || offs == PAGE_BTR_SEG_TOP
-				      + PAGE_HEADER + FSEG_HDR_SPACE
-				      || offs == PAGE_BTR_IBUF_FREE_LIST_NODE
-				      + PAGE_HEADER + FIL_ADDR_PAGE
-				      + 0 /*FLST_PREV*/
-				      || offs == PAGE_BTR_IBUF_FREE_LIST_NODE
-				      + PAGE_HEADER + FIL_ADDR_PAGE
-				      + FIL_ADDR_SIZE /*FLST_NEXT*/);
-				break;
-			}
-		}
-#endif /* UNIV_DEBUG */
-		ptr = mlog_parse_nbytes(type, ptr, end_ptr, page, page_zip);
-		if (ptr && page && !page_id.page_no() && type == MLOG_4BYTES) {
-			switch (ulint offs = mach_read_from_2(old_ptr)) {
-				fil_space_t*	space;
-				ulint		val;
-			default:
-				break;
-			case FSP_HEADER_OFFSET + FSP_SPACE_FLAGS:
-			case FSP_HEADER_OFFSET + FSP_SIZE:
-			case FSP_HEADER_OFFSET + FSP_FREE_LIMIT:
-			case FSP_HEADER_OFFSET + FSP_FREE + FLST_LEN:
-				space = fil_space_get(page_id.space());
-				ut_a(space != NULL);
-				val = mach_read_from_4(page + offs);
-
-				switch (offs) {
-				case FSP_HEADER_OFFSET + FSP_SPACE_FLAGS:
-					space->flags = val;
-					break;
-				case FSP_HEADER_OFFSET + FSP_SIZE:
-					space->size_in_header = val;
-					break;
-				case FSP_HEADER_OFFSET + FSP_FREE_LIMIT:
-					space->free_limit = val;
-					break;
-				case FSP_HEADER_OFFSET + FSP_FREE + FLST_LEN:
-					space->free_len = val;
-					ut_ad(val == flst_get_len(
-						      page + offs));
-					break;
-				}
-			}
-		}
-		break;
-	case MLOG_REC_INSERT: case MLOG_COMP_REC_INSERT:
-		if (!page_zip) {
-		} else if (!page_zip_decompress(page_zip, page, true)) {
-			ib::error() << "corrupted page " << block->page.id;
-		} else {
-			ut_d(page_type = fil_page_get_type(page));
-		}
-
-		ut_ad(!page || fil_page_type_is_index(page_type));
-
-		if (NULL != (ptr = mlog_parse_index(
-				     ptr, end_ptr,
-				     type == MLOG_COMP_REC_INSERT,
-				     &index))) {
-			ut_a(!page
-			     || (ibool)!!page_is_comp(page)
-			     == dict_table_is_comp(index->table));
-			ptr = page_cur_parse_insert_rec(false, ptr, end_ptr,
-							block, index, mtr);
-		}
-		break;
-	case MLOG_REC_CLUST_DELETE_MARK: case MLOG_COMP_REC_CLUST_DELETE_MARK:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-
-		if (NULL != (ptr = mlog_parse_index(
-				     ptr, end_ptr,
-				     type == MLOG_COMP_REC_CLUST_DELETE_MARK,
-				     &index))) {
-			ut_a(!block
-			     || !!page_is_comp(block->frame)
-			     == index->table->not_redundant());
-			ptr = btr_cur_parse_del_mark_set_clust_rec(
-				ptr, end_ptr, block, index, mtr);
-		}
-		break;
-	case MLOG_REC_SEC_DELETE_MARK:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-		ptr = btr_cur_parse_del_mark_set_sec_rec(ptr, end_ptr,
-							 block, mtr);
-		break;
-	case MLOG_REC_UPDATE_IN_PLACE: case MLOG_COMP_REC_UPDATE_IN_PLACE:
-		if (!page_zip) {
-		} else if (!page_zip_decompress(page_zip, page, true)) {
-			ib::error() << "corrupted page " << block->page.id;
-		} else {
-			ut_d(page_type = fil_page_get_type(page));
-		}
-
-		ut_ad(!page || fil_page_type_is_index(page_type));
-
-		if (NULL != (ptr = mlog_parse_index(
-				     ptr, end_ptr,
-				     type == MLOG_COMP_REC_UPDATE_IN_PLACE,
-				     &index))) {
-			ut_a(!page
-			     || (ibool)!!page_is_comp(page)
-			     == dict_table_is_comp(index->table));
-			ptr = btr_cur_parse_update_in_place(ptr, end_ptr,
-							    block, index,
-							    mtr);
-		}
-		break;
-	case MLOG_LIST_END_DELETE: case MLOG_COMP_LIST_END_DELETE:
-	case MLOG_LIST_START_DELETE: case MLOG_COMP_LIST_START_DELETE:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-
-		if (NULL != (ptr = mlog_parse_index(
-				     ptr, end_ptr,
-				     type == MLOG_COMP_LIST_END_DELETE
-				     || type == MLOG_COMP_LIST_START_DELETE,
-				     &index))) {
-			ut_a(!page
-			     || (ibool)!!page_is_comp(page)
-			     == dict_table_is_comp(index->table));
-			ptr = page_parse_delete_rec_list(type, ptr, end_ptr,
-							 block, index, mtr);
-		}
-		break;
-	case MLOG_LIST_END_COPY_CREATED: case MLOG_COMP_LIST_END_COPY_CREATED:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-
-		if (NULL != (ptr = mlog_parse_index(
-				     ptr, end_ptr,
-				     type == MLOG_COMP_LIST_END_COPY_CREATED,
-				     &index))) {
-			ut_a(!page
-			     || (ibool)!!page_is_comp(page)
-			     == dict_table_is_comp(index->table));
-			ptr = page_parse_copy_rec_list_to_created_page(
-				ptr, end_ptr, block, index, mtr);
-		}
-		break;
-	case MLOG_PAGE_REORGANIZE:
-	case MLOG_COMP_PAGE_REORGANIZE:
-	case MLOG_ZIP_PAGE_REORGANIZE:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-
-		if (NULL != (ptr = mlog_parse_index(
-				     ptr, end_ptr,
-				     type != MLOG_PAGE_REORGANIZE,
-				     &index))) {
-			ut_a(!page
-			     || (ibool)!!page_is_comp(page)
-			     == dict_table_is_comp(index->table));
-			ptr = btr_parse_page_reorganize(
-				ptr, end_ptr, index,
-				type == MLOG_ZIP_PAGE_REORGANIZE,
-				block, mtr);
-		}
-		break;
-	case MLOG_PAGE_CREATE: case MLOG_COMP_PAGE_CREATE:
-	case MLOG_PAGE_CREATE_RTREE: case MLOG_COMP_PAGE_CREATE_RTREE:
-		/* Allow anything in page_type when creating a page. */
-		ut_a(!page_zip);
-		if (!block) {
-			break;
-		}
-		page_create_low(block, type == MLOG_COMP_PAGE_CREATE
-				|| type == MLOG_COMP_PAGE_CREATE_RTREE);
-		if (type == MLOG_PAGE_CREATE_RTREE
-		    || type == MLOG_COMP_PAGE_CREATE_RTREE) {
-			static_assert(((FIL_PAGE_INDEX & 0xff00)
-				       | byte(FIL_PAGE_RTREE))
-				      == FIL_PAGE_RTREE, "compatibility");
-			page[FIL_PAGE_TYPE] = byte(FIL_PAGE_RTREE);
-			memset(page + FIL_RTREE_SPLIT_SEQ_NUM, 0, 8);
-		}
-
-		break;
-	case MLOG_UNDO_INSERT:
-		ut_ad(!page || page_type == FIL_PAGE_UNDO_LOG);
-		ptr = trx_undo_parse_add_undo_rec(ptr, end_ptr, page);
-		break;
-	case MLOG_UNDO_ERASE_END:
-		if (page) {
-			ut_ad(page_type == FIL_PAGE_UNDO_LOG);
-			uint16_t first_free = mach_read_from_2(
-				TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE + page);
-			memset(page + first_free, 0,
-			       (srv_page_size - FIL_PAGE_DATA_END)
-			       - first_free);
-		}
-		break;
-	case MLOG_UNDO_INIT:
-		/* Allow anything in page_type when creating a page. */
-		ptr = trx_undo_parse_page_init(ptr, end_ptr, page);
-		break;
-	case MLOG_UNDO_HDR_REUSE:
-		ut_ad(!page || page_type == FIL_PAGE_UNDO_LOG);
-		ptr = trx_undo_parse_page_header_reuse(ptr, end_ptr, page);
-		break;
-	case MLOG_UNDO_HDR_CREATE:
-		ut_ad(!page || page_type == FIL_PAGE_UNDO_LOG);
-		ptr = trx_undo_parse_page_header(ptr, end_ptr, block, mtr);
-		break;
-	case MLOG_REC_MIN_MARK: case MLOG_COMP_REC_MIN_MARK:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-		/* On a compressed page, MLOG_COMP_REC_MIN_MARK
-		will be followed by MLOG_COMP_REC_DELETE
-		or MLOG_ZIP_WRITE_HEADER(FIL_PAGE_PREV, FIL_NULL)
-		in the same mini-transaction. */
-		ut_a(type == MLOG_COMP_REC_MIN_MARK || !page_zip);
-		ptr = btr_parse_set_min_rec_mark(
-			ptr, end_ptr, type == MLOG_COMP_REC_MIN_MARK,
-			block, mtr);
-		break;
-	case MLOG_REC_DELETE: case MLOG_COMP_REC_DELETE:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-
-		if (NULL != (ptr = mlog_parse_index(
-				     ptr, end_ptr,
-				     type == MLOG_COMP_REC_DELETE,
-				     &index))) {
-			ut_a(!page
-			     || (ibool)!!page_is_comp(page)
-			     == dict_table_is_comp(index->table));
-			ptr = page_cur_parse_delete_rec(ptr, end_ptr,
-							block, index, mtr);
-		}
-		break;
-	case MLOG_IBUF_BITMAP_INIT:
-		/* Allow anything in page_type when creating a page. */
-		if (block) ibuf_bitmap_init_apply(block);
-		break;
-	case MLOG_INIT_FILE_PAGE2:
-		/* Allow anything in page_type when creating a page. */
-		if (block) fsp_apply_init_file_page(block);
-		break;
-	case MLOG_INIT_FREE_PAGE:
-		/* The page can be zero-filled and its previous
-		contents can be ignored. We do not write or apply
-		this record yet. */
-		break;
-	case MLOG_WRITE_STRING:
-		ut_ad(!page_zip
-		      || fil_page_get_type(page_zip->data)
-		      <= FIL_PAGE_TYPE_ZBLOB2);
-		if (page_id.page_no() || mach_read_from_2(ptr + 2)
-		    != 11 + MY_AES_BLOCK_SIZE) {
-			/* Not writing crypt_info */
-		} else if (fil_space_t* space
-			   = fil_space_acquire_silent(page_id.space())) {
-			if (mach_read_from_2(ptr)
-			    == FSP_HEADER_OFFSET + XDES_ARR_OFFSET + MAGIC_SZ
-			    + space->physical_size() * XDES_SIZE
-			    / FSP_EXTENT_SIZE
-			    && (ptr[4] == CRYPT_SCHEME_UNENCRYPTED
-				|| ptr[4] == CRYPT_SCHEME_1)
-			    && ptr[5] == MY_AES_BLOCK_SIZE
-			    && ptr[6 + MY_AES_BLOCK_SIZE + 4 + 4]
-			    <= FIL_ENCRYPTION_OFF) {
-				/* from fil_space_crypt_t::write_page0() */
-				fil_crypt_parse(space, ptr + 4);
-			}
-			space->release();
-		}
-		ptr = mlog_parse_string(ptr, end_ptr, page, page_zip);
-		break;
-	case MLOG_ZIP_WRITE_NODE_PTR:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-		ptr = page_zip_parse_write_node_ptr(ptr, end_ptr,
-						    page, page_zip);
-		break;
-	case MLOG_ZIP_WRITE_BLOB_PTR:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-		ptr = page_zip_parse_write_blob_ptr(ptr, end_ptr,
-						    page, page_zip);
-		break;
-	case MLOG_ZIP_WRITE_HEADER:
-		ut_ad(!page || fil_page_type_is_index(page_type));
-		ptr = page_zip_parse_write_header(ptr, end_ptr,
-						  page, page_zip);
-		break;
-	case MLOG_ZIP_PAGE_COMPRESS:
-		/* Allow anything in page_type when creating a page. */
-		ptr = page_zip_parse_compress(ptr, end_ptr, block);
-		break;
-	case MLOG_ZIP_PAGE_COMPRESS_NO_DATA:
-		if (NULL != (ptr = mlog_parse_index(
-				ptr, end_ptr, TRUE, &index))) {
-
-			ut_a(!page || ((ibool)!!page_is_comp(page)
-				== dict_table_is_comp(index->table)));
-			if (end_ptr == ptr) {
-				ptr = NULL;
-				break;
-			}
-			if (page &&
-			    !page_zip_compress(block, index, *ptr, mtr)) {
-				ut_error;
-			}
-			ptr++;
-		}
-		break;
-	case MLOG_ZIP_WRITE_TRX_ID:
-		/* This must be a clustered index leaf page. */
-		ut_ad(!page || page_type == FIL_PAGE_INDEX);
-		ptr = page_zip_parse_write_trx_id(ptr, end_ptr,
-						  page, page_zip);
-		break;
-	case MLOG_FILE_WRITE_CRYPT_DATA:
-		dberr_t err;
-		ptr = fil_parse_write_crypt_data(ptr, end_ptr, &err);
-
-		if (err != DB_SUCCESS) {
-			recv_sys.found_corrupt_log = TRUE;
-		}
+	switch (log_sys.log.format) {
+	case log_t::FORMAT_10_5:
+	case log_t::FORMAT_10_5 | log_t::FORMAT_ENCRYPTED:
 		break;
 	default:
-		ib::error() << "Incorrect log record type "
-			<< ib::hex(unsigned(type));
-		recv_sys.found_corrupt_log = true;
-		ptr = NULL;
-	}
-
-	if (index) {
-		dict_table_t*	table = index->table;
-
-		dict_mem_index_free(index);
-		dict_mem_table_free(table);
+		if (dberr_t err = recv_log_recover_10_4()) {
+			ib::error()
+				<< "Upgrade after a crash is not supported."
+				" The redo log was created with " << creator
+				<< (err == DB_ERROR
+				    ? "." : ", and it appears corrupted.");
+			return err;
+		}
 	}
 
-	return(ptr);
+	return(DB_SUCCESS);
 }
 
 /*******************************************************//**
@@ -2119,84 +1430,17 @@ recv_calc_lsn_on_data_add(
 	return(lsn + lsn_len);
 }
 
-/** Store a redo log record for applying.
-@param type	record type
-@param page_id	page identifier
-@param body	record body
-@param rec_end	end of record
-@param lsn	start LSN of the mini-transaction
-@param end_lsn	end LSN of the mini-transaction */
-inline void recv_sys_t::add(mlog_id_t type, const page_id_t page_id,
-                            const byte* body, const byte* rec_end, lsn_t lsn,
-                            lsn_t end_lsn)
-{
-  ut_ad(type != MLOG_FILE_DELETE);
-  ut_ad(type != MLOG_FILE_CREATE2);
-  ut_ad(type != MLOG_FILE_RENAME2);
-  ut_ad(type != MLOG_FILE_NAME);
-  ut_ad(type != MLOG_DUMMY_RECORD);
-  ut_ad(type != MLOG_CHECKPOINT);
-  ut_ad(type != MLOG_TRUNCATE);
-  ut_ad(!log_sys.is_physical());
-
-  std::pair<map::iterator, bool> p= pages.insert(map::value_type
-                                                 (page_id, page_recv_t()));
-  page_recv_t& recs= p.first->second;
-  ut_ad(p.second == recs.log.empty());
-
-  /* Store the log record body in limited-size chunks, because the
-  heap grows into the buffer pool. */
-  size_t len= static_cast<size_t>(rec_end - body);
-
-  recv_t *recv= new (alloc(sizeof *recv, true))
-    recv_t(static_cast<uint32_t>(len), type, lsn, end_lsn);
-  recs.log.append(recv);
-
-  for (recv_t::data_t *prev= nullptr;;)
-  {
-    const size_t used= static_cast<uint16_t>
-      (UT_LIST_GET_FIRST(blocks)->page.access_time);
-    ut_ad(used || srv_page_size == 65536);
-    const size_t l= std::min(len, srv_page_size - used - sizeof(recv_t::data));
-    recv_t::data_t *d= new (alloc(sizeof(recv_t::data) + l))
-      recv_t::data_t(body, l);
-    if (prev)
-      prev->append(d);
-    else
-      recv->data= d;
-    prev= d;
-
-    body+= l;
-    len-= l;
-    if (!len)
-      break;
-  }
-}
-
 /** Trim old log records for a page.
 @param start_lsn oldest log sequence number to preserve
 @return whether all the log for the page was trimmed */
 inline bool page_recv_t::trim(lsn_t start_lsn)
 {
-  if (log_sys.is_physical())
-  {
-    while (log.head)
-    {
-      if (log.head->lsn >= start_lsn) return false;
-      last_offset= 1; /* the next record must not be same_page */
-      log_rec_t *next= log.head->next;
-      recv_sys.free(log.head);
-      log.head= next;
-    }
-    log.tail= nullptr;
-    return true;
-  }
-
   while (log.head)
   {
     if (log.head->lsn >= start_lsn) return false;
+    last_offset= 1; /* the next record must not be same_page */
     log_rec_t *next= log.head->next;
-    static_cast<const recv_t*>(log.head)->free();
+    recv_sys.free(log.head);
     log.head= next;
   }
   log.tail= nullptr;
@@ -2207,21 +1451,10 @@ inline bool page_recv_t::trim(lsn_t start_lsn)
 inline void page_recv_t::recs_t::clear()
 {
   ut_ad(mutex_own(&recv_sys.mutex));
-  if (log_sys.is_physical())
-  {
-    for (const log_rec_t *l= head; l; )
-    {
-      const log_rec_t *next= l->next;
-      recv_sys.free(l);
-      l= next;
-    }
-    head= tail= nullptr;
-    return;
-  }
   for (const log_rec_t *l= head; l; )
   {
     const log_rec_t *next= l->next;
-    static_cast<const recv_t*>(l)->free();
+    recv_sys.free(l);
     l= next;
   }
   head= tail= nullptr;
@@ -2302,8 +1535,13 @@ append:
 @param apply           whether to apply file-level log records
 @return whether FILE_CHECKPOINT record was seen the first time,
 or corruption was noticed */
-inline bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t store, bool apply)
+bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t store, bool apply)
 {
+  ut_ad(log_mutex_own());
+  ut_ad(mutex_own(&mutex));
+  ut_ad(parse_start_lsn);
+  ut_ad(log_sys.is_physical());
+
   const byte *const end= buf + len;
 loop:
   const byte *const log= buf + recovered_offset;
@@ -2731,30 +1969,6 @@ same_page:
   goto loop;
 }
 
-
-/*********************************************************************//**
-Copies the log record body from recv to buf. */
-static ATTRIBUTE_COLD
-void
-recv_data_copy_to_buf(
-/*==================*/
-	byte*	buf,	/*!< in: buffer of length at least recv->len */
-	const recv_t& recv)	/*!< in: log record */
-{
-	const recv_t::data_t* recv_data = recv.data;
-	ulint len = recv.len;
-
-	do {
-		ulint offset = page_offset(recv_data + 1);
-		const ulint chunk_limit = (srv_page_size - offset);
-		const ulint l = std::min(len, chunk_limit);
-		memcpy(buf, reinterpret_cast<const byte*>(recv_data + 1), l);
-		recv_data = recv_data->next;
-		buf += l;
-		len -= l;
-	} while (len);
-}
-
 /** Apply the hashed log records to the page, if the page lsn is less than the
 lsn of a log record.
 @param[in,out]	block		buffer pool page
@@ -2767,9 +1981,6 @@ static void recv_recover_page(buf_block_t* block, mtr_t& mtr,
 			      fil_space_t* space = NULL,
 			      mlog_init_t::init* init = NULL)
 {
-	page_t*		page;
-	page_zip_des_t* page_zip;
-
 	ut_ad(mutex_own(&recv_sys.mutex));
 	ut_ad(recv_sys.apply_log_recs);
 	ut_ad(recv_needed_recovery);
@@ -2778,6 +1989,7 @@ static void recv_recover_page(buf_block_t* block, mtr_t& mtr,
 	ut_ad(block->page.id == p->first);
 	ut_ad(!p->second.is_being_processed());
 	ut_ad(!space || space->id == block->page.id.space());
+	ut_ad(log_sys.is_physical());
 
 	if (UNIV_UNLIKELY(srv_print_verbose_log == 2)) {
 		ib::info() << "Applying log to page " << block->page.id;
@@ -2791,12 +2003,9 @@ static void recv_recover_page(buf_block_t* block, mtr_t& mtr,
 
 	mutex_exit(&recv_sys.mutex);
 
-	page = block->frame;
-	page_zip = buf_block_get_page_zip(block);
-
 	byte *frame = UNIV_LIKELY_NULL(block->page.zip.data)
 		? block->page.zip.data
-		: page;
+		: block->frame;
 	const lsn_t page_lsn = init
 		? 0
 		: mach_read_from_8(frame + FIL_PAGE_LSN);
@@ -2804,174 +2013,113 @@ static void recv_recover_page(buf_block_t* block, mtr_t& mtr,
 	lsn_t start_lsn = 0, end_lsn = 0;
 	ut_d(lsn_t recv_start_lsn = 0);
 	const lsn_t init_lsn = init ? init->lsn : 0;
-	const bool is_physical = log_sys.is_physical();
 
-	for (const log_rec_t* l : p->second.log) {
+	for (const log_rec_t* recv : p->second.log) {
+		const log_phys_t* l = static_cast<const log_phys_t*>(recv);
 		ut_ad(l->lsn);
 		ut_ad(end_lsn <= l->lsn);
 		end_lsn = l->lsn;
 		ut_ad(end_lsn <= log_sys.log.scanned_lsn);
 
-		const recv_t* recv = static_cast<const recv_t*>(l);
-		ut_ad(recv->start_lsn);
-		ut_ad(recv_start_lsn <= recv->start_lsn);
-		ut_d(recv_start_lsn = recv->start_lsn);
+		ut_ad(l->start_lsn);
+		ut_ad(recv_start_lsn <= l->start_lsn);
+		ut_d(recv_start_lsn = l->start_lsn);
 
-		if (recv->start_lsn < page_lsn) {
+		if (l->start_lsn < page_lsn) {
 			/* This record has already been applied. */
 			DBUG_PRINT("ib_log", ("apply skip %u:%u LSN " LSN_PF
 					      " < " LSN_PF,
 					      block->page.id.space(),
 					      block->page.id.page_no(),
-					      recv->start_lsn, page_lsn));
+					      l->start_lsn, page_lsn));
 			continue;
 		}
 
-		if (recv->start_lsn < init_lsn) {
-			DBUG_PRINT("ib_log", ("init skip %s %u:%u LSN " LSN_PF
+		if (l->start_lsn < init_lsn) {
+			DBUG_PRINT("ib_log", ("init skip %u:%u LSN " LSN_PF
 					      " < " LSN_PF,
-					      is_physical
-					      ? "?"
-					      : get_mlog_string(recv->type),
 					      block->page.id.space(),
 					      block->page.id.page_no(),
-					      recv->start_lsn, init_lsn));
+					      l->start_lsn, init_lsn));
 			continue;
 		}
 
-		if (is_physical) {
-			const log_phys_t *f= static_cast<const log_phys_t*>(l);
 
-			if (UNIV_UNLIKELY(srv_print_verbose_log == 2)) {
-				ib::info() << "apply " << f->start_lsn
-					   << ": " << block->page.id;
-			}
+		if (UNIV_UNLIKELY(srv_print_verbose_log == 2)) {
+			ib::info() << "apply " << l->start_lsn
+				   << ": " << block->page.id;
+		}
 
-			DBUG_PRINT("ib_log", ("apply " LSN_PF ": %u:%u",
-					      f->start_lsn,
-					      block->page.id.space(),
-					      block->page.id.page_no()));
+		DBUG_PRINT("ib_log", ("apply " LSN_PF ": %u:%u",
+				      l->start_lsn,
+				      block->page.id.space(),
+				      block->page.id.page_no()));
 
-			log_phys_t::apply_status a= f->apply(
-				*block, p->second.last_offset);
+		log_phys_t::apply_status a= l->apply(*block,
+						     p->second.last_offset);
+
+		switch (a) {
+		case log_phys_t::APPLIED_NO:
+			ut_ad(!mtr.has_modifications());
+			free_page = true;
+			start_lsn = 0;
+			continue;
+		case log_phys_t::APPLIED_YES:
+			goto set_start_lsn;
+		case log_phys_t::APPLIED_TO_FSP_HEADER:
+		case log_phys_t::APPLIED_TO_ENCRYPTION:
+			break;
+		}
 
+		if (fil_space_t* s = space
+		    ? space
+		    : fil_space_acquire(block->page.id.space())) {
 			switch (a) {
-			case log_phys_t::APPLIED_NO:
-				ut_ad(!mtr.has_modifications());
-				free_page = true;
-				start_lsn = 0;
-				continue;
-			case log_phys_t::APPLIED_YES:
-				goto set_start_lsn;
 			case log_phys_t::APPLIED_TO_FSP_HEADER:
-			case log_phys_t::APPLIED_TO_ENCRYPTION:
+				s->flags = mach_read_from_4(
+					FSP_HEADER_OFFSET
+					+ FSP_SPACE_FLAGS + frame);
+				s->size_in_header = mach_read_from_4(
+					FSP_HEADER_OFFSET + FSP_SIZE
+					+ frame);
+				s->free_limit = mach_read_from_4(
+					FSP_HEADER_OFFSET
+					+ FSP_FREE_LIMIT + frame);
+				s->free_len = mach_read_from_4(
+					FSP_HEADER_OFFSET + FSP_FREE
+					+ FLST_LEN + frame);
 				break;
-			}
-
-			if (fil_space_t* s = space
-			    ? space
-			    : fil_space_acquire(block->page.id.space())) {
-				switch (a) {
-				case log_phys_t::APPLIED_TO_FSP_HEADER:
-					s->flags = mach_read_from_4(
-						FSP_HEADER_OFFSET
-						+ FSP_SPACE_FLAGS + frame);
-					s->size_in_header = mach_read_from_4(
-						FSP_HEADER_OFFSET + FSP_SIZE
-						+ frame);
-					s->free_limit = mach_read_from_4(
-						FSP_HEADER_OFFSET
-						+ FSP_FREE_LIMIT + frame);
-					s->free_len = mach_read_from_4(
-						FSP_HEADER_OFFSET + FSP_FREE
-						+ FLST_LEN + frame);
+			default:
+				byte* b= frame
+					+ fsp_header_get_encryption_offset(
+						block->zip_size())
+					+ FSP_HEADER_OFFSET;
+				if (memcmp(b, CRYPT_MAGIC, MAGIC_SZ)) {
 					break;
-				default:
-					byte* b= frame
-						+ fsp_header_get_encryption_offset(
-							block->zip_size())
-						+ FSP_HEADER_OFFSET;
-					if (memcmp(b, CRYPT_MAGIC, MAGIC_SZ)) {
-						break;
-					}
-					b += MAGIC_SZ;
-					if (*b != CRYPT_SCHEME_UNENCRYPTED
-					    && *b != CRYPT_SCHEME_1) {
-						break;
-					}
-					if (b[1] != MY_AES_BLOCK_SIZE) {
-						break;
-					}
-					if (b[2 + MY_AES_BLOCK_SIZE + 4 + 4]
-					    > FIL_ENCRYPTION_OFF) {
-						break;
-					}
-					fil_crypt_parse(s, b);
 				}
-
-				if (s != space) {
-					s->release();
+				b += MAGIC_SZ;
+				if (*b != CRYPT_SCHEME_UNENCRYPTED
+				    && *b != CRYPT_SCHEME_1) {
+					break;
 				}
-			}
-		} else {
-			if (recv->type == MLOG_INIT_FREE_PAGE) {
-				/* This does not really modify the page. */
-				ut_ad(!mtr.has_modifications());
-				free_page = true;
-				start_lsn = 0;
-				continue;
-			}
-
-			if (UNIV_UNLIKELY(srv_print_verbose_log == 2)) {
-				ib::info() << "apply " << recv->start_lsn
-					   << ":" << recv->type
-					   << " len " << recv->len
-					   << " page " << block->page.id;
-			}
-
-			DBUG_LOG("ib_log", "apply " << recv->start_lsn << ": "
-				 << get_mlog_string(recv->type)
-				 << " len " << recv->len
-				 << " page " << block->page.id);
-
-			ulint data_offset = page_offset(recv->data + 1);
-			byte* buf;
-			const byte* recs;
-
-			if (srv_page_size - data_offset < recv->len) {
-				/* We have to copy the record body to
-				a separate buffer */
-				recs = buf = static_cast<byte*>
-					(ut_malloc_nokey(recv->len));
-				recv_data_copy_to_buf(buf, *recv);
-			} else {
-				buf = NULL;
-				recs = reinterpret_cast<const byte*>
-					(recv->data + 1);
+				if (b[1] != MY_AES_BLOCK_SIZE) {
+					break;
+				}
+				if (b[2 + MY_AES_BLOCK_SIZE + 4 + 4]
+				    > FIL_ENCRYPTION_OFF) {
+					break;
+				}
+				fil_crypt_parse(s, b);
 			}
 
-			recv_parse_or_apply_log_rec_body(
-				recv->type, recs, recs + recv->len,
-				block->page.id, true, block, &mtr);
-
-			ut_free(buf);
-
-			end_lsn = recv->start_lsn + recv->len;
-			mach_write_to_8(FIL_PAGE_LSN + page, end_lsn);
-			mach_write_to_8(srv_page_size
-					- FIL_PAGE_END_LSN_OLD_CHKSUM
-					+ page, end_lsn);
-
-			if (UNIV_LIKELY_NULL(page_zip)) {
-				memcpy_aligned<8>(FIL_PAGE_LSN
-						  + page_zip->data,
-						  FIL_PAGE_LSN + page, 8);
+			if (s != space) {
+				s->release();
 			}
 		}
 
 set_start_lsn:
 		if (!start_lsn) {
-			start_lsn = recv->start_lsn;
+			start_lsn = l->start_lsn;
 		}
 	}
 
@@ -3318,158 +2466,6 @@ done:
 	mutex_exit(&recv_sys.mutex);
 }
 
-/** Tries to parse a single log record.
-@param[out]	type		log record type
-@param[in]	ptr		pointer to a buffer
-@param[in]	end_ptr		end of the buffer
-@param[out]	space_id	tablespace identifier
-@param[out]	page_no		page number
-@param[in]	apply		whether to apply MLOG_FILE_* records
-@param[out]	body		start of log record body
-@return length of the record, or 0 if the record was not complete */
-static
-ulint
-recv_parse_log_rec(
-	mlog_id_t*	type,
-	const byte*	ptr,
-	const byte*	end_ptr,
-	ulint*		space,
-	ulint*		page_no,
-	bool		apply,
-	const byte**	body)
-{
-	const byte*	new_ptr;
-
-	*body = NULL;
-
-	UNIV_MEM_INVALID(type, sizeof *type);
-	UNIV_MEM_INVALID(space, sizeof *space);
-	UNIV_MEM_INVALID(page_no, sizeof *page_no);
-	UNIV_MEM_INVALID(body, sizeof *body);
-
-	if (ptr == end_ptr) {
-
-		return(0);
-	}
-
-	switch (*ptr) {
-	case MLOG_MULTI_REC_END:
-	case MLOG_DUMMY_RECORD:
-		*type = static_cast<mlog_id_t>(*ptr);
-		return(1);
-	case MLOG_CHECKPOINT:
-		if (end_ptr < ptr + SIZE_OF_MLOG_CHECKPOINT) {
-			return(0);
-		}
-		*type = static_cast<mlog_id_t>(*ptr);
-		return(SIZE_OF_MLOG_CHECKPOINT);
-	case MLOG_MULTI_REC_END | MLOG_SINGLE_REC_FLAG:
-	case MLOG_DUMMY_RECORD | MLOG_SINGLE_REC_FLAG:
-	case MLOG_CHECKPOINT | MLOG_SINGLE_REC_FLAG:
-		ib::error() << "Incorrect log record type "
-			<< ib::hex(unsigned(*ptr));
-		recv_sys.found_corrupt_log = true;
-		return(0);
-	}
-
-	new_ptr = mlog_parse_initial_log_record(ptr, end_ptr, type, space,
-						page_no);
-	*body = new_ptr;
-
-	if (UNIV_UNLIKELY(!new_ptr)) {
-
-		return(0);
-	}
-
-	const byte*	old_ptr = new_ptr;
-	new_ptr = recv_parse_or_apply_log_rec_body(
-		*type, new_ptr, end_ptr, page_id_t(*space, *page_no), apply,
-		NULL, NULL);
-
-	if (UNIV_UNLIKELY(new_ptr == NULL)) {
-		return(0);
-	}
-
-	if (*page_no == 0 && *type == MLOG_4BYTES
-	    && apply
-	    && mach_read_from_2(old_ptr) == FSP_HEADER_OFFSET + FSP_SIZE) {
-		old_ptr += 2;
-
-		ulint size = mach_parse_compressed(&old_ptr, end_ptr);
-
-		recv_spaces_t::iterator it = recv_spaces.find(*space);
-
-		ut_ad(!recv_sys.mlog_checkpoint_lsn
-		      || *space == TRX_SYS_SPACE
-		      || srv_is_undo_tablespace(*space)
-		      || it != recv_spaces.end());
-
-		if (it != recv_spaces.end() && !it->second.space) {
-			it->second.size = size;
-		}
-
-		fil_space_set_recv_size(*space, size);
-	}
-
-	return ulint(new_ptr - ptr);
-}
-
-/** Prints diagnostic info of corrupt log.
-@param[in]	ptr	pointer to corrupt log record
-@param[in]	type	type of the log record (could be garbage)
-@param[in]	space	tablespace ID (could be garbage)
-@param[in]	page_no	page number (could be garbage)
-@return whether processing should continue */
-static
-bool
-recv_report_corrupt_log(
-	const byte*	ptr,
-	int		type,
-	ulint		space,
-	ulint		page_no)
-{
-	ib::error() <<
-		"############### CORRUPT LOG RECORD FOUND ##################";
-
-	const ulint ptr_offset = ulint(ptr - recv_sys.buf);
-
-	ib::info() << "Log record type " << type << ", page " << space << ":"
-		<< page_no << ". Log parsing proceeded successfully up to "
-		<< recv_sys.recovered_lsn << ". Previous log record type "
-		<< recv_previous_parsed_rec_type << ", is multi "
-		<< recv_previous_parsed_rec_is_multi << " Recv offset "
-		<< ptr_offset << ", prev "
-		<< recv_previous_parsed_rec_offset;
-
-	ut_ad(ptr <= recv_sys.buf + recv_sys.len);
-
-	const ulint	limit	= 100;
-	const ulint	prev_offset = std::min(recv_previous_parsed_rec_offset,
-					       ptr_offset);
-	const ulint	before = std::min(prev_offset, limit);
-	const ulint	after = std::min(recv_sys.len - ptr_offset, limit);
-
-	ib::info() << "Hex dump starting " << before << " bytes before and"
-		" ending " << after << " bytes after the corrupted record:";
-
-	const byte* start = recv_sys.buf + prev_offset - before;
-
-	ut_print_buf(stderr, start, ulint(ptr - start) + after);
-	putc('\n', stderr);
-
-	if (!srv_force_recovery) {
-		ib::info() << "Set innodb_force_recovery to ignore this error.";
-		return(false);
-	}
-
-	ib::warn() << "The log file may have been corrupt and it is possible"
-		" that the log scan did not proceed far enough in recovery!"
-		" Please run CHECK TABLE on your InnoDB tables to check"
-		" that they are ok! If mysqld crashes after this recovery; "
-		<< FORCE_RECOVERY_MSG;
-	return(true);
-}
-
 /** Check whether the number of read redo log blocks exceeds the maximum.
 Store last_stored_lsn if the recovery is not in the last phase.
 @param[in,out] store    whether to store page operations
@@ -3487,331 +2483,6 @@ inline bool recv_sys_t::is_memory_exhausted(store_t *store)
   return true;
 }
 
-/** Parse log records from a buffer and optionally store them to a
-hash table to wait merging to file pages.
-@param[in]	checkpoint_lsn		the LSN of the latest checkpoint
-@param[in]	store			whether to store page operations
-@param[in]	apply			whether to apply the records
-@return whether MLOG_CHECKPOINT or FILE_CHECKPOINT record
-was seen the first time, or corruption was noticed */
-bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t *store, bool apply)
-{
-	ut_ad(log_mutex_own());
-	ut_ad(mutex_own(&recv_sys.mutex));
-	ut_ad(recv_sys.parse_start_lsn != 0);
-
-	if (log_sys.is_physical()) {
-		return recv_sys.parse(checkpoint_lsn, *store, apply);
-	}
-
-	bool		single_rec;
-	ulint		len;
-	lsn_t		new_recovered_lsn;
-	lsn_t		old_lsn;
-	mlog_id_t	type;
-	ulint		space;
-	ulint		page_no;
-	const byte*	body;
-	const bool	last_phase = (*store == STORE_IF_EXISTS);
-
-loop:
-	const byte* ptr = recv_sys.buf + recv_sys.recovered_offset;
-	const byte* end_ptr = recv_sys.buf + recv_sys.len;
-
-	if (ptr == end_ptr) {
-
-		return(false);
-	}
-
-	/* Check for memory overflow and ignore the parsing of remaining
-	redo log records if InnoDB ran out of memory */
-	if (recv_sys.is_memory_exhausted(store) && last_phase) {
-		return false;
-	}
-
-	switch (*ptr) {
-	case MLOG_CHECKPOINT:
-	case MLOG_DUMMY_RECORD:
-		single_rec = true;
-		break;
-	default:
-		single_rec = !!(*ptr & MLOG_SINGLE_REC_FLAG);
-	}
-
-	if (single_rec) {
-		/* The mtr did not modify multiple pages */
-
-		old_lsn = recv_sys.recovered_lsn;
-
-		/* Try to parse a log record, fetching its type, space id,
-		page no, and a pointer to the body of the log record */
-
-		len = recv_parse_log_rec(&type, ptr, end_ptr, &space,
-					 &page_no, apply, &body);
-
-		if (recv_sys.found_corrupt_log) {
-			recv_report_corrupt_log(ptr, type, space, page_no);
-			return(true);
-		}
-
-		if (recv_sys.found_corrupt_fs) {
-			return(true);
-		}
-
-		if (len == 0) {
-			return(false);
-		}
-
-		new_recovered_lsn = recv_calc_lsn_on_data_add(old_lsn, len);
-
-		if (new_recovered_lsn > recv_sys.scanned_lsn) {
-			/* The log record filled a log block, and we require
-			that also the next log block should have been scanned
-			in */
-
-			return(false);
-		}
-
-		recv_previous_parsed_rec_type = type;
-		recv_previous_parsed_rec_offset = recv_sys.recovered_offset;
-		recv_previous_parsed_rec_is_multi = 0;
-
-		recv_sys.recovered_offset += len;
-		recv_sys.recovered_lsn = new_recovered_lsn;
-
-		switch (type) {
-			lsn_t	lsn;
-		case MLOG_DUMMY_RECORD:
-			/* Do nothing */
-			break;
-		case MLOG_CHECKPOINT:
-			compile_time_assert(SIZE_OF_MLOG_CHECKPOINT == 1 + 8);
-			lsn = mach_read_from_8(ptr + 1);
-
-			if (UNIV_UNLIKELY(srv_print_verbose_log == 2)) {
-				fprintf(stderr,
-					"MLOG_CHECKPOINT(" LSN_PF ") %s at "
-					LSN_PF "\n", lsn,
-					lsn != checkpoint_lsn ? "ignored"
-					: recv_sys.mlog_checkpoint_lsn
-					? "reread" : "read",
-					recv_sys.recovered_lsn);
-			}
-
-			DBUG_PRINT("ib_log",
-				   ("MLOG_CHECKPOINT(" LSN_PF ") %s at "
-				    LSN_PF,
-				    lsn,
-				    lsn != checkpoint_lsn ? "ignored"
-				    : recv_sys.mlog_checkpoint_lsn
-				    ? "reread" : "read",
-				    recv_sys.recovered_lsn));
-
-			if (lsn == checkpoint_lsn) {
-				if (recv_sys.mlog_checkpoint_lsn) {
-					ut_ad(recv_sys.mlog_checkpoint_lsn
-					      <= recv_sys.recovered_lsn);
-					break;
-				}
-				recv_sys.mlog_checkpoint_lsn
-					= recv_sys.recovered_lsn;
-				return(true);
-			}
-			break;
-		default:
-			switch (*store) {
-			case STORE_NO:
-				break;
-			case STORE_IF_EXISTS:
-				if (!fil_space_get_size(space)) {
-					break;
-				}
-				/* fall through */
-			case STORE_YES:
-				recv_sys.add(
-					type, page_id_t(space, page_no), body,
-					ptr + len, old_lsn,
-					recv_sys.recovered_lsn);
-			}
-			/* fall through */
-		case MLOG_FILE_NAME:
-		case MLOG_FILE_DELETE:
-		case MLOG_FILE_CREATE2:
-		case MLOG_FILE_RENAME2:
-		case MLOG_TRUNCATE:
-			/* These were already handled by
-			recv_parse_log_rec() and
-			recv_parse_or_apply_log_rec_body(). */
-			DBUG_PRINT("ib_log",
-				("scan " LSN_PF ": log rec %s"
-				" len " ULINTPF
-				" page " ULINTPF ":" ULINTPF,
-				old_lsn, get_mlog_string(type),
-				len, space, page_no));
-		}
-	} else {
-		/* Check that all the records associated with the single mtr
-		are included within the buffer */
-
-		ulint	total_len	= 0;
-		ulint	n_recs		= 0;
-		bool	only_mlog_file	= true;
-		ulint	mlog_rec_len	= 0;
-
-		for (;;) {
-			len = recv_parse_log_rec(
-				&type, ptr, end_ptr, &space, &page_no,
-				false, &body);
-
-			if (recv_sys.found_corrupt_log) {
-corrupted_log:
-				recv_report_corrupt_log(
-					ptr, type, space, page_no);
-				return(true);
-			}
-
-			if (ptr == end_ptr) {
-			} else if (type == MLOG_CHECKPOINT
-				   || (*ptr & MLOG_SINGLE_REC_FLAG)) {
-				recv_sys.found_corrupt_log = true;
-				goto corrupted_log;
-			}
-
-			if (recv_sys.found_corrupt_fs) {
-				return(true);
-			}
-
-			if (len == 0) {
-				return(false);
-			}
-
-			recv_previous_parsed_rec_type = type;
-			recv_previous_parsed_rec_offset
-				= recv_sys.recovered_offset + total_len;
-			recv_previous_parsed_rec_is_multi = 1;
-
-			/* MLOG_FILE_NAME redo log records doesn't make changes
-			to persistent data. If only MLOG_FILE_NAME redo
-			log record exists then reset the parsing buffer pointer
-			by changing recovered_lsn and recovered_offset. */
-			if (type != MLOG_FILE_NAME && only_mlog_file == true) {
-				only_mlog_file = false;
-			}
-
-			if (only_mlog_file) {
-				new_recovered_lsn = recv_calc_lsn_on_data_add(
-					recv_sys.recovered_lsn, len);
-				mlog_rec_len += len;
-				recv_sys.recovered_offset += len;
-				recv_sys.recovered_lsn = new_recovered_lsn;
-			}
-
-			total_len += len;
-			n_recs++;
-
-			ptr += len;
-
-			if (type == MLOG_MULTI_REC_END) {
-				DBUG_PRINT("ib_log",
-					   ("scan " LSN_PF
-					    ": multi-log end"
-					    " total_len " ULINTPF
-					    " n=" ULINTPF,
-					    recv_sys.recovered_lsn,
-					    total_len, n_recs));
-				total_len -= mlog_rec_len;
-				break;
-			}
-
-			DBUG_PRINT("ib_log",
-				   ("scan " LSN_PF ": multi-log rec %s"
-				    " len " ULINTPF
-				    " page " ULINTPF ":" ULINTPF,
-				    recv_sys.recovered_lsn,
-				    get_mlog_string(type), len, space, page_no));
-		}
-
-		new_recovered_lsn = recv_calc_lsn_on_data_add(
-			recv_sys.recovered_lsn, total_len);
-
-		if (new_recovered_lsn > recv_sys.scanned_lsn) {
-			/* The log record filled a log block, and we require
-			that also the next log block should have been scanned
-			in */
-
-			return(false);
-		}
-
-		/* Add all the records to the hash table */
-
-		ptr = recv_sys.buf + recv_sys.recovered_offset;
-
-		for (;;) {
-			old_lsn = recv_sys.recovered_lsn;
-			/* This will apply MLOG_FILE_ records. We
-			had to skip them in the first scan, because we
-			did not know if the mini-transaction was
-			completely recovered (until MLOG_MULTI_REC_END). */
-			len = recv_parse_log_rec(
-				&type, ptr, end_ptr, &space, &page_no,
-				apply, &body);
-
-			if (recv_sys.found_corrupt_log
-			    && !recv_report_corrupt_log(
-				    ptr, type, space, page_no)) {
-				return(true);
-			}
-
-			if (recv_sys.found_corrupt_fs) {
-				return(true);
-			}
-
-			ut_a(len != 0);
-			ut_a(!(*ptr & MLOG_SINGLE_REC_FLAG));
-
-			recv_sys.recovered_offset += len;
-			recv_sys.recovered_lsn
-				= recv_calc_lsn_on_data_add(old_lsn, len);
-
-			switch (type) {
-			case MLOG_MULTI_REC_END:
-				/* Found the end mark for the records */
-				goto loop;
-			case MLOG_FILE_NAME:
-			case MLOG_FILE_DELETE:
-			case MLOG_FILE_CREATE2:
-			case MLOG_FILE_RENAME2:
-			case MLOG_TRUNCATE:
-				/* These were already handled by
-				recv_parse_log_rec() and
-				recv_parse_or_apply_log_rec_body(). */
-				break;
-			default:
-				switch (*store) {
-				case STORE_NO:
-					break;
-				case STORE_IF_EXISTS:
-					if (!fil_space_get_size(space)) {
-						break;
-					}
-					/* fall through */
-				case STORE_YES:
-					recv_sys.add(
-						type,
-						page_id_t(space, page_no),
-						body, ptr + len,
-						old_lsn,
-						new_recovered_lsn);
-				}
-			}
-
-			ptr += len;
-		}
-	}
-
-	goto loop;
-}
-
 /** Adds data from a new log block to the parsing buffer of recv_sys if
 recv_sys.parse_start_lsn is non-zero.
 @param[in]	log_block	log block to add
@@ -3923,12 +2594,11 @@ static bool recv_scan_log_recs(
 	ut_ad(start_lsn % OS_FILE_LOG_BLOCK_SIZE == 0);
 	ut_ad(end_lsn % OS_FILE_LOG_BLOCK_SIZE == 0);
 	ut_ad(end_lsn >= start_lsn + OS_FILE_LOG_BLOCK_SIZE);
+	ut_ad(log_sys.is_physical());
 
 	const byte* const	log_end = log_block
 		+ ulint(end_lsn - start_lsn);
-	const ulint sizeof_checkpoint= log_sys.is_physical()
-		? SIZE_OF_FILE_CHECKPOINT
-		: SIZE_OF_MLOG_CHECKPOINT;
+	constexpr ulint sizeof_checkpoint= SIZE_OF_FILE_CHECKPOINT;
 
 	do {
 		ut_ad(!finished);
@@ -3978,13 +2648,9 @@ static bool recv_scan_log_recs(
 		if (data_len == LOG_BLOCK_HDR_SIZE + sizeof_checkpoint
 		    && scanned_lsn == checkpoint_lsn + sizeof_checkpoint
 		    && log_block[LOG_BLOCK_HDR_SIZE]
-		    == (log_sys.is_physical()
-			? FILE_CHECKPOINT | (SIZE_OF_FILE_CHECKPOINT - 2)
-			: MLOG_CHECKPOINT)
+		    == (FILE_CHECKPOINT | (SIZE_OF_FILE_CHECKPOINT - 2))
 		    && checkpoint_lsn == mach_read_from_8(
-			    (log_sys.is_physical()
-			     ? LOG_BLOCK_HDR_SIZE + 1 + 2
-			     : LOG_BLOCK_HDR_SIZE + 1)
+			    (LOG_BLOCK_HDR_SIZE + 1 + 2)
 			    + log_block)) {
 			/* The redo log is logically empty. */
 			ut_ad(recv_sys.mlog_checkpoint_lsn == 0
@@ -4067,8 +2733,7 @@ static bool recv_scan_log_recs(
 
 	if (more_data && !recv_sys.found_corrupt_log) {
 		/* Try to parse more log records */
-
-		if (recv_parse_log_recs(checkpoint_lsn, store, apply)) {
+		if (recv_sys.parse(checkpoint_lsn, *store, apply)) {
 			ut_ad(recv_sys.found_corrupt_log
 			      || recv_sys.found_corrupt_fs
 			      || recv_sys.mlog_checkpoint_lsn
@@ -4126,9 +2791,6 @@ recv_group_scan_log_recs(
 	recv_sys.scanned_lsn = *contiguous_lsn;
 	recv_sys.recovered_lsn = *contiguous_lsn;
 	recv_sys.scanned_checkpoint_no = 0;
-	recv_previous_parsed_rec_type = MLOG_SINGLE_REC_FLAG;
-	recv_previous_parsed_rec_offset	= 0;
-	recv_previous_parsed_rec_is_multi = 0;
 	ut_ad(recv_max_page_lsn == 0);
 	ut_ad(last_phase || !recv_writer_thread_active);
 	mutex_exit(&recv_sys.mutex);
@@ -4320,15 +2982,9 @@ recv_init_crash_recovery_spaces(bool rescan, bool& missing_tablespace)
 			are some redo log records for it. */
 			fil_names_dirty(rs.second.space);
 		} else if (rs.second.name == "") {
-			ib::error() << (log_sys.is_physical()
-					? "Missing FILE_CREATE, FILE_DELETE"
-					" or FILE_MODIFY"
-					" before FILE_CHECKPOINT"
-					" for tablespace "
-					: "Missing MLOG_FILE_NAME"
-					" or MLOG_FILE_DELETE"
-					" before MLOG_CHECKPOINT"
-					" for tablespace ") << rs.first;
+			ib::error() << "Missing FILE_CREATE, FILE_DELETE"
+				" or FILE_MODIFY before FILE_CHECKPOINT"
+				" for tablespace " << rs.first;
 			recv_sys.found_corrupt_log = true;
 			return(DB_CORRUPTION);
 		} else {
@@ -4357,7 +3013,7 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 {
 	ulint		max_cp_field;
 	lsn_t		checkpoint_lsn;
-	bool		rescan;
+	bool		rescan = false;
 	ib_uint64_t	checkpoint_no;
 	lsn_t		contiguous_lsn;
 	byte*		buf;
@@ -4432,7 +3088,14 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 		return(DB_ERROR);
 	}
 
-	/* Look for MLOG_CHECKPOINT or FILE_CHECKPOINT. */
+	size_t sizeof_checkpoint;
+
+	if (!log_sys.is_physical()) {
+		sizeof_checkpoint = 9/* size of MLOG_CHECKPOINT */;
+		goto completed;
+	}
+
+	/* Look for FILE_CHECKPOINT. */
 	recv_group_scan_log_recs(checkpoint_lsn, &contiguous_lsn, false);
 	/* The first scan should not have stored or applied any records. */
 	ut_ad(recv_sys.pages.empty());
@@ -4454,9 +3117,7 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 		if (!srv_read_only_mode && scan_lsn != checkpoint_lsn) {
 			log_mutex_exit();
 			ib::error err;
-			err << (log_sys.is_physical()
-				? "Missing FILE_CHECKPOINT"
-				: "Missing MLOG_CHECKPOINT");
+			err << "Missing FILE_CHECKPOINT";
 			if (end_lsn) {
 				err << " at " << end_lsn;
 			}
@@ -4466,7 +3127,6 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 		}
 
 		log_sys.log.scanned_lsn = checkpoint_lsn;
-		rescan = false;
 	} else {
 		contiguous_lsn = checkpoint_lsn;
 		rescan = recv_group_scan_log_recs(
@@ -4482,10 +3142,9 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 	/* NOTE: we always do a 'recovery' at startup, but only if
 	there is something wrong we will print a message to the
 	user about recovery: */
-	const ulint sizeof_checkpoint= log_sys.is_physical()
-		? SIZE_OF_FILE_CHECKPOINT
-		: SIZE_OF_MLOG_CHECKPOINT;
+	sizeof_checkpoint= SIZE_OF_FILE_CHECKPOINT;
 
+completed:
 	if (flush_lsn == checkpoint_lsn + sizeof_checkpoint
 	    && recv_sys.mlog_checkpoint_lsn == checkpoint_lsn) {
 		/* The redo log is logically empty. */
@@ -4501,8 +3160,8 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 				" tablespace file header, " << flush_lsn << ".";
 		}
 
-		if (!recv_needed_recovery) {
-
+		if (log_sys.log.is_physical()
+		    && !recv_needed_recovery) {
 			ib::info() << "The log sequence number " << flush_lsn
 				<< " in the system tablespace does not match"
 				" the log sequence number " << checkpoint_lsn
@@ -4593,8 +3252,9 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 		ut_ad(!rescan || recv_sys.pages.empty());
 	}
 
-	if (log_sys.log.scanned_lsn < checkpoint_lsn
-	    || log_sys.log.scanned_lsn < recv_max_page_lsn) {
+	if (log_sys.is_physical()
+	    && (log_sys.log.scanned_lsn < checkpoint_lsn
+		|| log_sys.log.scanned_lsn < recv_max_page_lsn)) {
 
 		ib::error() << "We scanned the log up to "
 			<< log_sys.log.scanned_lsn
@@ -4631,7 +3291,7 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 	log_sys.last_checkpoint_lsn = checkpoint_lsn;
 
 	if (!srv_read_only_mode && srv_operation == SRV_OPERATION_NORMAL) {
-		/* Write a MLOG_CHECKPOINT or FILE_CHECKPOINT first,
+		/* Write a FILE_CHECKPOINT marker as the first thing,
 		before generating any other redo log. This ensures
 		that subsequent crash recovery will be possible even
 		if the server were killed soon after this. */
@@ -4724,177 +3384,3 @@ recv_dblwr_t::find_page(ulint space_id, ulint page_no)
 
   return result;
 }
-
-#ifndef DBUG_OFF
-/** Return string name of the redo log record type.
-@param[in]	type	record log record enum
-@return string name of record log record */
-static const char* get_mlog_string(mlog_id_t type)
-{
-	switch (type) {
-	case MLOG_SINGLE_REC_FLAG:
-		return("MLOG_SINGLE_REC_FLAG");
-
-	case MLOG_1BYTE:
-		return("MLOG_1BYTE");
-
-	case MLOG_2BYTES:
-		return("MLOG_2BYTES");
-
-	case MLOG_4BYTES:
-		return("MLOG_4BYTES");
-
-	case MLOG_8BYTES:
-		return("MLOG_8BYTES");
-
-	case MLOG_REC_INSERT:
-		return("MLOG_REC_INSERT");
-
-	case MLOG_REC_CLUST_DELETE_MARK:
-		return("MLOG_REC_CLUST_DELETE_MARK");
-
-	case MLOG_REC_SEC_DELETE_MARK:
-		return("MLOG_REC_SEC_DELETE_MARK");
-
-	case MLOG_REC_UPDATE_IN_PLACE:
-		return("MLOG_REC_UPDATE_IN_PLACE");
-
-	case MLOG_REC_DELETE:
-		return("MLOG_REC_DELETE");
-
-	case MLOG_LIST_END_DELETE:
-		return("MLOG_LIST_END_DELETE");
-
-	case MLOG_LIST_START_DELETE:
-		return("MLOG_LIST_START_DELETE");
-
-	case MLOG_LIST_END_COPY_CREATED:
-		return("MLOG_LIST_END_COPY_CREATED");
-
-	case MLOG_PAGE_REORGANIZE:
-		return("MLOG_PAGE_REORGANIZE");
-
-	case MLOG_PAGE_CREATE:
-		return("MLOG_PAGE_CREATE");
-
-	case MLOG_UNDO_INSERT:
-		return("MLOG_UNDO_INSERT");
-
-	case MLOG_UNDO_ERASE_END:
-		return("MLOG_UNDO_ERASE_END");
-
-	case MLOG_UNDO_INIT:
-		return("MLOG_UNDO_INIT");
-
-	case MLOG_UNDO_HDR_REUSE:
-		return("MLOG_UNDO_HDR_REUSE");
-
-	case MLOG_UNDO_HDR_CREATE:
-		return("MLOG_UNDO_HDR_CREATE");
-
-	case MLOG_REC_MIN_MARK:
-		return("MLOG_REC_MIN_MARK");
-
-	case MLOG_IBUF_BITMAP_INIT:
-		return("MLOG_IBUF_BITMAP_INIT");
-
-	case MLOG_WRITE_STRING:
-		return("MLOG_WRITE_STRING");
-
-	case MLOG_MULTI_REC_END:
-		return("MLOG_MULTI_REC_END");
-
-	case MLOG_DUMMY_RECORD:
-		return("MLOG_DUMMY_RECORD");
-
-	case MLOG_FILE_DELETE:
-		return("MLOG_FILE_DELETE");
-
-	case MLOG_COMP_REC_MIN_MARK:
-		return("MLOG_COMP_REC_MIN_MARK");
-
-	case MLOG_COMP_PAGE_CREATE:
-		return("MLOG_COMP_PAGE_CREATE");
-
-	case MLOG_COMP_REC_INSERT:
-		return("MLOG_COMP_REC_INSERT");
-
-	case MLOG_COMP_REC_CLUST_DELETE_MARK:
-		return("MLOG_COMP_REC_CLUST_DELETE_MARK");
-
-	case MLOG_COMP_REC_UPDATE_IN_PLACE:
-		return("MLOG_COMP_REC_UPDATE_IN_PLACE");
-
-	case MLOG_COMP_REC_DELETE:
-		return("MLOG_COMP_REC_DELETE");
-
-	case MLOG_COMP_LIST_END_DELETE:
-		return("MLOG_COMP_LIST_END_DELETE");
-
-	case MLOG_COMP_LIST_START_DELETE:
-		return("MLOG_COMP_LIST_START_DELETE");
-
-	case MLOG_COMP_LIST_END_COPY_CREATED:
-		return("MLOG_COMP_LIST_END_COPY_CREATED");
-
-	case MLOG_COMP_PAGE_REORGANIZE:
-		return("MLOG_COMP_PAGE_REORGANIZE");
-
-	case MLOG_FILE_CREATE2:
-		return("MLOG_FILE_CREATE2");
-
-	case MLOG_ZIP_WRITE_NODE_PTR:
-		return("MLOG_ZIP_WRITE_NODE_PTR");
-
-	case MLOG_ZIP_WRITE_BLOB_PTR:
-		return("MLOG_ZIP_WRITE_BLOB_PTR");
-
-	case MLOG_ZIP_WRITE_HEADER:
-		return("MLOG_ZIP_WRITE_HEADER");
-
-	case MLOG_ZIP_PAGE_COMPRESS:
-		return("MLOG_ZIP_PAGE_COMPRESS");
-
-	case MLOG_ZIP_PAGE_COMPRESS_NO_DATA:
-		return("MLOG_ZIP_PAGE_COMPRESS_NO_DATA");
-
-	case MLOG_ZIP_PAGE_REORGANIZE:
-		return("MLOG_ZIP_PAGE_REORGANIZE");
-
-	case MLOG_ZIP_WRITE_TRX_ID:
-		return("MLOG_ZIP_WRITE_TRX_ID");
-
-	case MLOG_FILE_RENAME2:
-		return("MLOG_FILE_RENAME2");
-
-	case MLOG_FILE_NAME:
-		return("MLOG_FILE_NAME");
-
-	case MLOG_CHECKPOINT:
-		return("MLOG_CHECKPOINT");
-
-	case MLOG_PAGE_CREATE_RTREE:
-		return("MLOG_PAGE_CREATE_RTREE");
-
-	case MLOG_COMP_PAGE_CREATE_RTREE:
-		return("MLOG_COMP_PAGE_CREATE_RTREE");
-
-	case MLOG_INIT_FILE_PAGE2:
-		return("MLOG_INIT_FILE_PAGE2");
-
-	case MLOG_TRUNCATE:
-		return("MLOG_TRUNCATE");
-
-	case MLOG_MEMSET:
-		return("MLOG_MEMSET");
-
-	case MLOG_INIT_FREE_PAGE:
-		return("MLOG_INIT_FREE_PAGE");
-
-	case MLOG_FILE_WRITE_CRYPT_DATA:
-		return("MLOG_FILE_WRITE_CRYPT_DATA");
-	}
-	DBUG_ASSERT(0);
-	return(NULL);
-}
-#endif /* !DBUG_OFF */
diff --git a/storage/innobase/mach/mach0data.cc b/storage/innobase/mach/mach0data.cc
deleted file mode 100644
index 85533908d16..00000000000
--- a/storage/innobase/mach/mach0data.cc
+++ /dev/null
@@ -1,129 +0,0 @@
-/*****************************************************************************
-
-Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-
-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
-Foundation; version 2 of the License.
-
-This program is distributed in the hope that it will be useful, but WITHOUT
-ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
-
-You should have received a copy of the GNU General Public License along with
-this program; if not, write to the Free Software Foundation, Inc.,
-51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA
-
-*****************************************************************************/
-
-/******************************************************************//**
-@file mach/mach0data.cc
-Utilities for converting data from the database file
-to the machine format.
-
-Created 11/28/1995 Heikki Tuuri
-***********************************************************************/
-
-#include "mach0data.h"
-
-/** Read a 32-bit integer in a compressed form.
-@param[in,out]	ptr	pointer to memory where to read;
-advanced by the number of bytes consumed, or set NULL if out of space
-@param[in]	end_ptr	end of the buffer
-@return unsigned value */
-ib_uint32_t
-mach_parse_compressed(
-	const byte**	ptr,
-	const byte*	end_ptr)
-{
-	ulint	val;
-
-	if (*ptr >= end_ptr) {
-		*ptr = NULL;
-		return(0);
-	}
-
-	val = mach_read_from_1(*ptr);
-
-	if (val < 0x80) {
-		/* 0nnnnnnn (7 bits) */
-		++*ptr;
-		return(static_cast<ib_uint32_t>(val));
-	}
-
-	/* Workaround GCC bug
-	https://gcc.gnu.org/bugzilla/show_bug.cgi?id=77673:
-	the compiler moves mach_read_from_4 right to the beginning of the
-	function, causing and out-of-bounds read if we are reading a short
-	integer close to the end of buffer. */
-#if defined(__GNUC__) && (__GNUC__ >= 5) && !defined(__clang__)
-#define DEPLOY_FENCE
-#endif
-
-#ifdef DEPLOY_FENCE
-	__atomic_thread_fence(__ATOMIC_ACQUIRE);
-#endif
-
-	if (val < 0xC0) {
-		/* 10nnnnnn nnnnnnnn (14 bits) */
-		if (end_ptr >= *ptr + 2) {
-			val = mach_read_from_2(*ptr) & 0x3FFF;
-			ut_ad(val > 0x7F);
-			*ptr += 2;
-			return(static_cast<ib_uint32_t>(val));
-		}
-		*ptr = NULL;
-		return(0);
-	}
-
-#ifdef DEPLOY_FENCE
-	__atomic_thread_fence(__ATOMIC_ACQUIRE);
-#endif
-
-	if (val < 0xE0) {
-		/* 110nnnnn nnnnnnnn nnnnnnnn (21 bits) */
-		if (end_ptr >= *ptr + 3) {
-			val = mach_read_from_3(*ptr) & 0x1FFFFF;
-			ut_ad(val > 0x3FFF);
-			*ptr += 3;
-			return(static_cast<ib_uint32_t>(val));
-		}
-		*ptr = NULL;
-		return(0);
-	}
-
-#ifdef DEPLOY_FENCE
-	__atomic_thread_fence(__ATOMIC_ACQUIRE);
-#endif
-
-	if (val < 0xF0) {
-		/* 1110nnnn nnnnnnnn nnnnnnnn nnnnnnnn (28 bits) */
-		if (end_ptr >= *ptr + 4) {
-			val = mach_read_from_4(*ptr) & 0xFFFFFFF;
-			ut_ad(val > 0x1FFFFF);
-			*ptr += 4;
-			return(static_cast<ib_uint32_t>(val));
-		}
-		*ptr = NULL;
-		return(0);
-	}
-
-#ifdef DEPLOY_FENCE
-	__atomic_thread_fence(__ATOMIC_ACQUIRE);
-#endif
-
-#undef DEPLOY_FENCE
-
-	ut_ad(val == 0xF0);
-
-	/* 11110000 nnnnnnnn nnnnnnnn nnnnnnnn nnnnnnnn (32 bits) */
-	if (end_ptr >= *ptr + 5) {
-		val = mach_read_from_4(*ptr + 1);
-		ut_ad(val > 0xFFFFFFF);
-		*ptr += 5;
-		return(static_cast<ib_uint32_t>(val));
-	}
-
-	*ptr = NULL;
-	return(0);
-}
diff --git a/storage/innobase/mtr/mtr0log.cc b/storage/innobase/mtr/mtr0log.cc
deleted file mode 100644
index fb363f012ce..00000000000
--- a/storage/innobase/mtr/mtr0log.cc
+++ /dev/null
@@ -1,360 +0,0 @@
-/*****************************************************************************
-
-Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, 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
-Foundation; version 2 of the License.
-
-This program is distributed in the hope that it will be useful, but WITHOUT
-ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
-
-You should have received a copy of the GNU General Public License along with
-this program; if not, write to the Free Software Foundation, Inc.,
-51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA
-
-*****************************************************************************/
-
-/**************************************************//**
-@file mtr/mtr0log.cc
-Mini-transaction log routines
-
-Created 12/7/1995 Heikki Tuuri
-*******************************************************/
-
-#include "mtr0log.h"
-#include "buf0buf.h"
-#include "dict0mem.h"
-#include "log0recv.h"
-#include "page0page.h"
-
-/********************************************************//**
-Parses an initial log record written by mlog_write_initial_log_record_low().
-@return parsed record end, NULL if not a complete record */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-mlog_parse_initial_log_record(
-/*==========================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	mlog_id_t*	type,	/*!< out: log record type: MLOG_1BYTE, ... */
-	ulint*		space,	/*!< out: space id */
-	ulint*		page_no)/*!< out: page number */
-{
-	if (end_ptr < ptr + 1) {
-
-		return(NULL);
-	}
-
-	*type = mlog_id_t(*ptr & ~MLOG_SINGLE_REC_FLAG);
-	if (UNIV_UNLIKELY(*type > MLOG_BIGGEST_TYPE
-			  && !EXTRA_CHECK_MLOG_NUMBER(*type))) {
-		recv_sys.found_corrupt_log = true;
-		return NULL;
-	}
-
-	ptr++;
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	*space = mach_parse_compressed(&ptr, end_ptr);
-
-	if (ptr != NULL) {
-		*page_no = mach_parse_compressed(&ptr, end_ptr);
-	}
-
-	return(const_cast<byte*>(ptr));
-}
-
-/********************************************************//**
-Parses a log record written by mtr_t::write(), mtr_t::memset().
-@return parsed record end, NULL if not a complete record or a corrupt record */
-const byte*
-mlog_parse_nbytes(
-/*==============*/
-	mlog_id_t	type,	/*!< in: log record type: MLOG_1BYTE, ... */
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	byte*		page,	/*!< in: page where to apply the log
-				record, or NULL */
-	void*		page_zip)/*!< in/out: compressed page, or NULL */
-{
-	ulint		offset;
-	ulint		val;
-	ib_uint64_t	dval;
-
-	ut_ad(type <= MLOG_8BYTES || type == MLOG_MEMSET);
-	ut_a(!page || !page_zip
-	     || type == MLOG_MEMSET
-	     || !fil_page_index_page_check(page));
-	if (end_ptr < ptr + 2) {
-		return NULL;
-	}
-
-	offset = mach_read_from_2(ptr);
-	ptr += 2;
-
-	if (UNIV_UNLIKELY(offset >= srv_page_size)) {
-		goto corrupt;
-	}
-
-	switch (type) {
-	case MLOG_MEMSET:
-		if (end_ptr < ptr + 3) {
-			return NULL;
-		}
-		val = mach_read_from_2(ptr);
-		ptr += 2;
-		if (UNIV_UNLIKELY(offset + val > srv_page_size)) {
-			goto corrupt;
-		}
-		if (page) {
-			memset(page + offset, *ptr, val);
-			if (page_zip) {
-				ut_ad(offset + val <= PAGE_DATA
-				      || !fil_page_index_page_check(page));
-				memset(static_cast<page_zip_des_t*>(page_zip)
-				       ->data + offset, *ptr, val);
-			}
-		}
-		return const_cast<byte*>(++ptr);
-	case MLOG_8BYTES:
-		dval = mach_u64_parse_compressed(&ptr, end_ptr);
-
-		if (ptr == NULL) {
-			return NULL;
-		}
-
-		if (page) {
-			if (page_zip) {
-				mach_write_to_8
-					(((page_zip_des_t*) page_zip)->data
-					 + offset, dval);
-			}
-			mach_write_to_8(page + offset, dval);
-		}
-
-		return const_cast<byte*>(ptr);
-	default:
-		val = mach_parse_compressed(&ptr, end_ptr);
-	}
-
-	if (ptr == NULL) {
-		return NULL;
-	}
-
-	switch (type) {
-	case MLOG_1BYTE:
-		if (val > 0xFFUL) {
-			goto corrupt;
-		}
-		if (page) {
-			if (page_zip) {
-				mach_write_to_1
-					(((page_zip_des_t*) page_zip)->data
-					 + offset, val);
-			}
-			mach_write_to_1(page + offset, val);
-		}
-		break;
-	case MLOG_2BYTES:
-		if (val > 0xFFFFUL) {
-			goto corrupt;
-		}
-		if (page) {
-			if (page_zip) {
-				mach_write_to_2
-					(((page_zip_des_t*) page_zip)->data
-					 + offset, val);
-			}
-			mach_write_to_2(page + offset, val);
-		}
-
-		break;
-	case MLOG_4BYTES:
-		if (page) {
-			if (page_zip) {
-				mach_write_to_4
-					(((page_zip_des_t*) page_zip)->data
-					 + offset, val);
-			}
-			mach_write_to_4(page + offset, val);
-		}
-		break;
-	default:
-	corrupt:
-		recv_sys.found_corrupt_log = true;
-		ptr = NULL;
-	}
-
-	return const_cast<byte*>(ptr);
-}
-
-/********************************************************//**
-Parses a log record written by mtr_t::memcpy().
-@return parsed record end, NULL if not a complete record */
-const byte*
-mlog_parse_string(
-/*==============*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	byte*		page,	/*!< in: page where to apply the log record,
-				or NULL */
-	void*		page_zip)/*!< in/out: compressed page, or NULL */
-{
-	ulint	offset;
-	ulint	len;
-
-	ut_a(!page || !page_zip
-	     || (fil_page_get_type(page) != FIL_PAGE_INDEX
-		 && fil_page_get_type(page) != FIL_PAGE_RTREE));
-
-	if (end_ptr < ptr + 4) {
-
-		return(NULL);
-	}
-
-	offset = mach_read_from_2(ptr);
-	ptr += 2;
-	len = mach_read_from_2(ptr);
-	ptr += 2;
-
-	if (offset >= srv_page_size || len + offset > srv_page_size) {
-		recv_sys.found_corrupt_log = TRUE;
-
-		return(NULL);
-	}
-
-	if (end_ptr < ptr + len) {
-
-		return(NULL);
-	}
-
-	if (page) {
-		if (page_zip) {
-			memcpy(((page_zip_des_t*) page_zip)->data
-				+ offset, ptr, len);
-		}
-		memcpy(page + offset, ptr, len);
-	}
-
-	return(ptr + len);
-}
-
-/********************************************************//**
-Parses a log record written by mlog_open_and_write_index.
-@return parsed record end, NULL if not a complete record */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-mlog_parse_index(
-/*=============*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	bool		comp,	/*!< in: TRUE=compact row format */
-	dict_index_t**	index)	/*!< out, own: dummy index */
-{
-	ulint		i, n, n_uniq;
-	dict_table_t*	table;
-	dict_index_t*	ind;
-	ulint		n_core_fields = 0;
-
-	if (comp) {
-		if (end_ptr < ptr + 4) {
-			return(NULL);
-		}
-		n = mach_read_from_2(ptr);
-		ptr += 2;
-		if (n & 0x8000) { /* record after instant ADD COLUMN */
-			n &= 0x7FFF;
-
-			n_core_fields = mach_read_from_2(ptr);
-
-			if (!n_core_fields || n_core_fields > n) {
-				recv_sys.found_corrupt_log = TRUE;
-				return(NULL);
-			}
-
-			ptr += 2;
-
-			if (end_ptr < ptr + 2) {
-				return(NULL);
-			}
-		}
-
-		n_uniq = mach_read_from_2(ptr);
-		ptr += 2;
-		ut_ad(n_uniq <= n);
-		if (end_ptr < ptr + n * 2) {
-			return(NULL);
-		}
-	} else {
-		n = n_uniq = 1;
-	}
-	table = dict_mem_table_create("LOG_DUMMY", NULL, n, 0,
-				      comp ? DICT_TF_COMPACT : 0, 0);
-	ind = dict_mem_index_create(table, "LOG_DUMMY", 0, n);
-	ind->n_uniq = (unsigned int) n_uniq;
-	if (n_uniq != n) {
-		ut_a(n_uniq + DATA_ROLL_PTR <= n);
-		ind->type = DICT_CLUSTERED;
-	}
-	if (comp) {
-		for (i = 0; i < n; i++) {
-			ulint	len = mach_read_from_2(ptr);
-			ptr += 2;
-			/* The high-order bit of len is the NOT NULL flag;
-			the rest is 0 or 0x7fff for variable-length fields,
-			and 1..0x7ffe for fixed-length fields. */
-			dict_mem_table_add_col(
-				table, NULL, NULL,
-				((len + 1) & 0x7fff) <= 1
-				? DATA_BINARY : DATA_FIXBINARY,
-				len & 0x8000 ? DATA_NOT_NULL : 0,
-				len & 0x7fff);
-
-			dict_index_add_col(ind, table,
-					   dict_table_get_nth_col(table, i),
-					   0);
-		}
-		dict_table_add_system_columns(table, table->heap);
-		if (n_uniq != n) {
-			/* Identify DB_TRX_ID and DB_ROLL_PTR in the index. */
-			ut_a(DATA_TRX_ID_LEN
-			     == dict_index_get_nth_col(ind, DATA_TRX_ID - 1
-						       + n_uniq)->len);
-			ut_a(DATA_ROLL_PTR_LEN
-			     == dict_index_get_nth_col(ind, DATA_ROLL_PTR - 1
-						       + n_uniq)->len);
-			ind->fields[DATA_TRX_ID - 1 + n_uniq].col
-				= &table->cols[n + DATA_TRX_ID];
-			ind->fields[DATA_ROLL_PTR - 1 + n_uniq].col
-				= &table->cols[n + DATA_ROLL_PTR];
-		}
-
-		ut_ad(table->n_cols == table->n_def);
-
-		if (n_core_fields) {
-			for (i = n_core_fields; i < n; i++) {
-				ind->fields[i].col->def_val.len
-					= UNIV_SQL_NULL;
-			}
-			ind->n_core_fields = n_core_fields;
-			ind->n_core_null_bytes = UT_BITS_IN_BYTES(
-				ind->get_n_nullable(n_core_fields));
-		} else {
-			ind->n_core_null_bytes = UT_BITS_IN_BYTES(
-				unsigned(ind->n_nullable));
-			ind->n_core_fields = ind->n_fields;
-		}
-	}
-	/* avoid ut_ad(index->cached) in dict_index_get_n_unique_in_tree */
-	ind->cached = TRUE;
-	ut_d(ind->is_dummy = true);
-	*index = ind;
-	return(ptr);
-}
diff --git a/storage/innobase/page/page0cur.cc b/storage/innobase/page/page0cur.cc
index d6d908a3163..321c5a850cf 100644
--- a/storage/innobase/page/page0cur.cc
+++ b/storage/innobase/page/page0cur.cc
@@ -784,226 +784,18 @@ page_cur_open_on_rnd_user_rec(
 				       ut_rnd_interval(n_recs) + 1);
 }
 
-static void rec_set_heap_no(rec_t *rec, ulint heap_no, bool compact)
-{
-  rec_set_bit_field_2(rec, heap_no,
-                      compact ? REC_NEW_HEAP_NO : REC_OLD_HEAP_NO,
-                      REC_HEAP_NO_MASK, REC_HEAP_NO_SHIFT);
-}
-
-static rec_t*
-page_cur_parse_insert_rec_zip(
-	page_cur_t*	cursor,	/*!< in/out: page cursor */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	const rec_t*	rec,	/*!< in: pointer to a physical record */
-	offset_t*	offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr);	/*!< in/out: mini-transaction */
-/***********************************************************//**
-Parses a log record of a record insert on a page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_cur_parse_insert_rec(
-/*======================*/
-	bool		is_short,/*!< in: true if short inserts */
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr)	/*!< in: mtr or NULL */
-{
-	ulint	origin_offset		= 0; /* remove warning */
-	ulint	end_seg_len;
-	ulint	mismatch_index		= 0; /* remove warning */
-	page_t*	page;
-	rec_t*	cursor_rec;
-	byte	buf1[1024];
-	byte*	buf;
-	const byte*	ptr2		= ptr;
-	ulint		info_and_status_bits = 0; /* remove warning */
-	page_cur_t	cursor;
-	mem_heap_t*	heap		= NULL;
-	offset_t	offsets_[REC_OFFS_NORMAL_SIZE];
-	offset_t*	offsets		= offsets_;
-	rec_offs_init(offsets_);
-
-	page = block ? buf_block_get_frame(block) : NULL;
-
-	if (is_short) {
-		cursor_rec = page_rec_get_prev(page_get_supremum_rec(page));
-	} else {
-		ulint	offset;
-
-		/* Read the cursor rec offset as a 2-byte ulint */
-
-		if (UNIV_UNLIKELY(end_ptr < ptr + 2)) {
-
-			return(NULL);
-		}
-
-		offset = mach_read_from_2(ptr);
-		ptr += 2;
-
-		cursor_rec = page + offset;
-
-		if (offset >= srv_page_size) {
-
-			recv_sys.found_corrupt_log = TRUE;
-
-			return(NULL);
-		}
-	}
-
-	end_seg_len = mach_parse_compressed(&ptr, end_ptr);
-
-	if (ptr == NULL) {
-
-		return(NULL);
-	}
-
-	if (end_seg_len >= srv_page_size << 1) {
-		recv_sys.found_corrupt_log = TRUE;
-
-		return(NULL);
-	}
-
-	if (end_seg_len & 0x1UL) {
-		/* Read the info bits */
-
-		if (end_ptr < ptr + 1) {
-
-			return(NULL);
-		}
-
-		info_and_status_bits = mach_read_from_1(ptr);
-		ptr++;
-
-		origin_offset = mach_parse_compressed(&ptr, end_ptr);
-
-		if (ptr == NULL) {
-
-			return(NULL);
-		}
-
-		ut_a(origin_offset < srv_page_size);
-
-		mismatch_index = mach_parse_compressed(&ptr, end_ptr);
-
-		if (ptr == NULL) {
-
-			return(NULL);
-		}
-
-		ut_a(mismatch_index < srv_page_size);
-	}
-
-	if (end_ptr < ptr + (end_seg_len >> 1)) {
-
-		return(NULL);
-	}
-
-	if (!block) {
-
-		return(const_cast<byte*>(ptr + (end_seg_len >> 1)));
-	}
-
-	ut_ad(!!page_is_comp(page) == dict_table_is_comp(index->table));
-	ut_ad(!buf_block_get_page_zip(block) || page_is_comp(page));
-
-	/* Read from the log the inserted index record end segment which
-	differs from the cursor record */
-
-	const bool is_leaf = page_is_leaf(page);
-
-	offsets = rec_get_offsets(cursor_rec, index, offsets, is_leaf,
-				  ULINT_UNDEFINED, &heap);
-
-	if (!(end_seg_len & 0x1UL)) {
-		info_and_status_bits = rec_get_info_and_status_bits(
-			cursor_rec, page_is_comp(page));
-		origin_offset = rec_offs_extra_size(offsets);
-		mismatch_index = rec_offs_size(offsets) - (end_seg_len >> 1);
-	}
-
-	end_seg_len >>= 1;
-
-	if (mismatch_index + end_seg_len < sizeof buf1) {
-		buf = buf1;
-	} else {
-		buf = static_cast<byte*>(
-			ut_malloc_nokey(mismatch_index + end_seg_len));
-	}
-
-	/* Build the inserted record to buf */
-
-        if (UNIV_UNLIKELY(mismatch_index >= srv_page_size)) {
-
-		ib::fatal() << "is_short " << is_short << ", "
-			<< "info_and_status_bits " << info_and_status_bits
-			<< ", offset " << page_offset(cursor_rec) << ","
-			" o_offset " << origin_offset << ", mismatch index "
-			<< mismatch_index << ", end_seg_len " << end_seg_len
-			<< " parsed len " << (ptr - ptr2);
-	}
-
-	memcpy(buf, rec_get_start(cursor_rec, offsets), mismatch_index);
-	memcpy(buf + mismatch_index, ptr, end_seg_len);
-	rec_set_heap_no(buf + origin_offset, PAGE_HEAP_NO_USER_LOW,
-			page_is_comp(page));
-
-	if (page_is_comp(page)) {
-		rec_set_info_and_status_bits(buf + origin_offset,
-					     info_and_status_bits);
-	} else {
-		rec_set_bit_field_1(buf + origin_offset, info_and_status_bits,
-				    REC_OLD_INFO_BITS,
-				    REC_INFO_BITS_MASK, REC_INFO_BITS_SHIFT);
-	}
-
-	page_cur_position(cursor_rec, block, &cursor);
-
-	offsets = rec_get_offsets(buf + origin_offset, index, offsets,
-				  is_leaf, ULINT_UNDEFINED, &heap);
-	/* The redo log record should only have been written
-	after the write was successful. */
-	if (block->page.zip.data) {
-		if (!page_cur_parse_insert_rec_zip(&cursor, index,
-						   buf + origin_offset,
-						   offsets, mtr)) {
-			ut_error;
-		}
-	} else if (!page_cur_insert_rec_low(&cursor, index,
-					    buf + origin_offset,
-					    offsets, mtr)) {
-		ut_error;
-	}
-
-	if (buf != buf1) {
-
-		ut_free(buf);
-	}
-
-	if (UNIV_LIKELY_NULL(heap)) {
-		mem_heap_free(heap);
-	}
-
-	return(const_cast<byte*>(ptr + end_seg_len));
-}
-
 /**
 Set the owned records field of the record pointed to by a directory slot.
-@tparam compressed  whether to update any ROW_FORMAT=COMPRESSED page as well
 @param[in,out]  block    file page
 @param[in]      slot     sparse directory slot
 @param[in,out]  n        number of records owned by the directory slot
 @param[in,out]  mtr      mini-transaction */
-template<bool compressed>
 static void page_dir_slot_set_n_owned(buf_block_t *block,
                                       const page_dir_slot_t *slot,
                                       ulint n, mtr_t *mtr)
 {
   rec_t *rec= const_cast<rec_t*>(page_dir_slot_get_rec(slot));
-  page_rec_set_n_owned<compressed>(block, rec, n, page_rec_is_comp(rec), mtr);
+  page_rec_set_n_owned<true>(block, rec, n, page_rec_is_comp(rec), mtr);
 }
 
 /**
@@ -1111,10 +903,10 @@ static void page_dir_balance_slot(buf_block_t *block, ulint s, mtr_t *mtr)
 				    <= PAGE_DIR_SLOT_MAX_N_OWNED);
 		/* Merge the slots. */
 		ulint n_owned = page_dir_slot_get_n_owned(slot);
-		page_dir_slot_set_n_owned<true>(block, slot, 0, mtr);
-		page_dir_slot_set_n_owned<true>(block, up_slot, n_owned
-						+ page_dir_slot_get_n_owned(
-							up_slot), mtr);
+		page_dir_slot_set_n_owned(block, slot, 0, mtr);
+		page_dir_slot_set_n_owned(block, up_slot, n_owned
+					  + page_dir_slot_get_n_owned(up_slot),
+					  mtr);
 		/* Shift the slots */
 		page_dir_slot_t* last_slot = page_dir_get_nth_slot(
 			block->frame, n_slots - 1);
@@ -1165,7 +957,7 @@ static void page_dir_balance_slot(buf_block_t *block, ulint s, mtr_t *mtr)
 
 	mtr->write<2>(*block, slot, page_offset(new_rec));
 func_exit:
-	page_dir_slot_set_n_owned<true>(block, up_slot, up_n_owned - 1, mtr);
+	page_dir_slot_set_n_owned(block, up_slot, up_n_owned - 1, mtr);
 }
 
 /** Allocate space for inserting an index record.
@@ -1235,7 +1027,6 @@ page_cur_insert_rec_low(
   ut_ad(fil_page_index_page_check(block->frame));
   ut_ad(mach_read_from_8(PAGE_HEADER + PAGE_INDEX_ID + block->frame) ==
         index->id ||
-        index->is_dummy ||
         mtr->is_inside_ibuf());
 
   ut_ad(!page_rec_is_supremum(cur->rec));
@@ -1643,9 +1434,7 @@ page_cur_insert_rec_zip(
   ut_ad(fil_page_get_type(cursor->block->frame) == FIL_PAGE_INDEX ||
         fil_page_get_type(cursor->block->frame) == FIL_PAGE_RTREE);
   ut_ad(mach_read_from_8(PAGE_HEADER + PAGE_INDEX_ID + cursor->block->frame) ==
-        index->id ||
-        index->is_dummy ||
-        mtr->is_inside_ibuf());
+        index->id || mtr->is_inside_ibuf());
   ut_ad(!page_get_instant(cursor->block->frame));
   ut_ad(!page_cur_is_after_last(cursor));
 #ifdef UNIV_ZIP_DEBUG
@@ -1977,767 +1766,6 @@ inc_dir:
   return insert_rec;
 }
 
-/** Increment PAGE_N_DIRECTION.
-@param[in,out]	block		ROW_FORMAT=COMPRESSED index page
-@param[in,out]	ptr		the PAGE_DIRECTION_B field
-@param[in]	dir		PAGE_RIGHT or PAGE_LEFT */
-static inline void page_direction_increment(buf_block_t *block, byte *ptr,
-                                            uint dir)
-{
-  ut_ad(ptr == PAGE_HEADER + PAGE_DIRECTION_B + block->frame);
-  ut_ad(dir == PAGE_RIGHT || dir == PAGE_LEFT);
-  block->page.zip.data[PAGE_HEADER + PAGE_DIRECTION_B]= *ptr= dir;
-  mach_write_to_2(PAGE_HEADER + PAGE_N_DIRECTION + block->frame,
-                  1U + page_header_get_field(block->frame, PAGE_N_DIRECTION));
-  memcpy_aligned<2>(PAGE_HEADER + PAGE_N_DIRECTION + block->frame,
-                    PAGE_HEADER + PAGE_N_DIRECTION + block->page.zip.data, 2);
-}
-
-/***********************************************************//**
-Inserts a record next to page cursor on a compressed and uncompressed
-page. Returns pointer to inserted record if succeed, i.e.,
-enough space available, NULL otherwise.
-The cursor stays at the same position.
-
-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().
-
-@return pointer to record if succeed, NULL otherwise */
-static rec_t*
-page_cur_parse_insert_rec_zip(
-	page_cur_t*	cursor,	/*!< in/out: page cursor */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	const rec_t*	rec,	/*!< in: pointer to a physical record */
-	offset_t*	offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction */
-{
-	byte*		insert_buf;
-	ulint		rec_size;
-	page_t*		page;		/*!< the relevant page */
-	rec_t*		insert_rec;	/*!< inserted record */
-	ulint		heap_no;	/*!< heap number of the inserted
-					record */
-	page_zip_des_t*	page_zip;
-
-	ut_ad(!log_sys.is_physical());
-
-	page_zip = page_cur_get_page_zip(cursor);
-	ut_ad(page_zip);
-	ut_ad(rec_offs_validate(rec, index, offsets));
-
-	page = page_cur_get_page(cursor);
-	ut_ad(dict_table_is_comp(index->table));
-	ut_ad(page_is_comp(page));
-	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->is_inside_ibuf());
-	ut_ad(!page_get_instant(page));
-	ut_ad(!page_cur_is_after_last(cursor));
-#ifdef UNIV_ZIP_DEBUG
-	ut_a(page_zip_validate(page_zip, page, index));
-#endif /* UNIV_ZIP_DEBUG */
-
-	/* 1. Get the size of the physical record in the page */
-	rec_size = rec_offs_size(offsets);
-
-#ifdef UNIV_DEBUG_VALGRIND
-	{
-		const void*	rec_start
-			= rec - rec_offs_extra_size(offsets);
-		ulint		extra_size
-			= rec_offs_extra_size(offsets)
-			- (rec_offs_comp(offsets)
-			   ? REC_N_NEW_EXTRA_BYTES
-			   : REC_N_OLD_EXTRA_BYTES);
-
-		/* All data bytes of the record must be valid. */
-		UNIV_MEM_ASSERT_RW(rec, rec_offs_data_size(offsets));
-		/* The variable-length header must be valid. */
-		UNIV_MEM_ASSERT_RW(rec_start, extra_size);
-	}
-#endif /* UNIV_DEBUG_VALGRIND */
-
-	const bool reorg_before_insert = page_has_garbage(page)
-		&& rec_size > page_get_max_insert_size(page, 1)
-		&& rec_size <= page_get_max_insert_size_after_reorganize(
-			page, 1);
-	constexpr uint16_t page_free_f = PAGE_FREE + PAGE_HEADER;
-	byte* const page_free = my_assume_aligned<4>(page_free_f + page);
-	uint16_t free_rec;
-
-	/* 2. Try to find suitable space from page memory management */
-	if (!page_zip_available(page_zip, dict_index_is_clust(index),
-				rec_size, 1)
-	    || reorg_before_insert) {
-		/* The values can change dynamically. */
-		ulint	level		= page_zip_level;
-#ifdef UNIV_DEBUG
-		rec_t*	cursor_rec	= page_cur_get_rec(cursor);
-#endif /* UNIV_DEBUG */
-
-		/* If we are not writing compressed page images, we
-		must reorganize the page before attempting the
-		insert. */
-		if (recv_recovery_is_on() && !log_sys.is_physical()) {
-			/* Insert into the uncompressed page only.
-			The page reorganization or creation that we
-			would attempt outside crash recovery would
-			have been covered by a previous redo log record. */
-		} else if (page_is_empty(page)) {
-			ut_ad(page_cur_is_before_first(cursor));
-
-			/* This is an empty page. Recreate it to
-			get rid of the modification log. */
-			page_create_zip(page_cur_get_block(cursor), index,
-					page_header_get_field(page, PAGE_LEVEL),
-					0, mtr);
-			ut_ad(!page_header_get_ptr(page, PAGE_FREE));
-
-			if (page_zip_available(
-				    page_zip, dict_index_is_clust(index),
-				    rec_size, 1)) {
-				free_rec = 0;
-				goto use_heap;
-			}
-
-			/* The cursor should remain on the page infimum. */
-			return(NULL);
-		} else if (!page_zip->m_nonempty && !page_has_garbage(page)) {
-			/* The page has been freshly compressed, so
-			reorganizing it will not help. */
-		} else {
-			ulint pos = page_rec_get_n_recs_before(cursor->rec);
-
-			if (!page_zip_reorganize(page_cur_get_block(cursor),
-						 index, level, mtr, true)) {
-				ut_ad(cursor->rec == cursor_rec);
-				return NULL;
-			}
-
-			if (pos) {
-				cursor->rec = page_rec_get_nth(page, pos);
-			} else {
-				ut_ad(cursor->rec == page_get_infimum_rec(
-					      page));
-			}
-
-			ut_ad(!page_header_get_ptr(page, PAGE_FREE));
-
-			if (page_zip_available(
-				    page_zip, dict_index_is_clust(index),
-				    rec_size, 1)) {
-				/* After reorganizing, there is space
-				available. */
-				free_rec = 0;
-				goto use_heap;
-			}
-		}
-
-		/* 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, 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
-		been the case or had the redo logging of compressed
-		pages been enabled during runtime then we'd have seen
-		a MLOG_ZIP_PAGE_COMPRESS redo record. Therefore, we
-		know that we don't need to reorganize the page. We,
-		however, do need to recompress the page. That will
-		happen when the next redo record is read which must
-		be of type MLOG_ZIP_PAGE_COMPRESS_NO_DATA and it must
-		contain a valid compression level value.
-		This implies that during recovery from this point till
-		the next redo is applied the uncompressed and
-		compressed versions are not identical and
-		page_zip_validate will fail but that is OK because
-		we call page_zip_validate only after processing
-		all changes to a page under a single mtr during
-		recovery. */
-		if (insert_rec == NULL) {
-			/* Out of space.
-			This should never occur during crash recovery,
-			because the MLOG_COMP_REC_INSERT should only
-			be logged after a successful operation. */
-			ut_ad(!recv_recovery_is_on());
-			ut_ad(!index->is_dummy);
-		} else if (recv_recovery_is_on() && !log_sys.is_physical()) {
-			/* This should be followed by
-			MLOG_ZIP_PAGE_COMPRESS_NO_DATA,
-			which should succeed. */
-			rec_offs_make_valid(insert_rec, index,
-					    page_is_leaf(page), offsets);
-		} else {
-			ulint	pos = page_rec_get_n_recs_before(insert_rec);
-			ut_ad(pos > 0);
-
-			/* We are writing entire page images to the
-			log.  Reduce the redo log volume by
-			reorganizing the page at the same time. */
-			if (page_zip_reorganize(cursor->block, index,
-						level, mtr)) {
-				/* The page was reorganized: Seek to pos. */
-				if (pos > 1) {
-					cursor->rec = page_rec_get_nth(
-						page, pos - 1);
-				} else {
-					cursor->rec = page + PAGE_NEW_INFIMUM;
-				}
-
-				insert_rec = page + rec_get_next_offs(
-					cursor->rec, TRUE);
-				rec_offs_make_valid(
-					insert_rec, index,
-					page_is_leaf(page), offsets);
-				return insert_rec;
-			}
-
-			/* Theoretically, we could try one last resort
-			of btr_page_reorganize_low() followed by
-			page_zip_available(), but that would be very
-			unlikely to succeed. (If the full reorganized
-			page failed to compress, why would it succeed
-			to compress the page, plus log the insert of
-			this record?) */
-
-			/* Out of space: restore the page */
-			if (!page_zip_decompress(page_zip, page, FALSE)) {
-				ut_error; /* Memory corrupted? */
-			}
-			ut_ad(page_validate(page, index));
-			insert_rec = NULL;
-		}
-
-		return(insert_rec);
-	}
-
-	free_rec = mach_read_from_2(page_free);
-	if (free_rec) {
-		/* Try to allocate from the head of the free list. */
-		lint	extra_size_diff;
-		offset_t	foffsets_[REC_OFFS_NORMAL_SIZE];
-		offset_t*	foffsets	= foffsets_;
-		mem_heap_t*	heap		= NULL;
-
-		rec_offs_init(foffsets_);
-
-		foffsets = rec_get_offsets(page + free_rec, index, foffsets,
-					   page_is_leaf(page),
-					   ULINT_UNDEFINED, &heap);
-		if (rec_offs_size(foffsets) < rec_size) {
-too_small:
-			if (UNIV_LIKELY_NULL(heap)) {
-				mem_heap_free(heap);
-			}
-
-			free_rec = 0;
-			goto use_heap;
-		}
-
-		insert_buf = page + free_rec - rec_offs_extra_size(foffsets);
-
-		/* On compressed pages, do not relocate records from
-		the free list.  If extra_size would grow, use the heap. */
-		extra_size_diff = lint(rec_offs_extra_size(offsets)
-				       - rec_offs_extra_size(foffsets));
-
-		if (UNIV_UNLIKELY(extra_size_diff < 0)) {
-			/* Add an offset to the extra_size. */
-			if (rec_offs_size(foffsets)
-			    < rec_size - ulint(extra_size_diff)) {
-
-				goto too_small;
-			}
-
-			insert_buf -= extra_size_diff;
-		} else if (UNIV_UNLIKELY(extra_size_diff)) {
-			/* Do not allow extra_size to grow */
-
-			goto too_small;
-		}
-
-		heap_no = rec_get_heap_no_new(page + free_rec);
-		int16_t next_rec = mach_read_from_2(page + free_rec - REC_NEXT);
-		/* We assume that int16_t is safe to use here.
-		With innodb_page_size=64k it would be unsafe,
-		but that cannot be used with ROW_FORMAT=COMPRESSED. */
-		static_assert(UNIV_ZIP_SIZE_SHIFT_MAX == 14, "compatibility");
-		if (next_rec) {
-			next_rec += free_rec;
-			ut_ad(int{PAGE_NEW_SUPREMUM_END + REC_N_NEW_EXTRA_BYTES}
-			      <= next_rec);
-			ut_ad(static_cast<uint16_t>(next_rec) < srv_page_size);
-		}
-		mtr->write<2>(*cursor->block, page_free,
-			      static_cast<uint16_t>(next_rec));
-		byte* garbage = my_assume_aligned<2>(page_free + 2);
-		ut_ad(mach_read_from_2(garbage) >= rec_size);
-		mtr->write<2>(*cursor->block, garbage,
-			      mach_read_from_2(garbage) - rec_size);
-		compile_time_assert(PAGE_GARBAGE == PAGE_FREE + 2);
-		compile_time_assert(!((PAGE_HEADER + PAGE_FREE) % 4));
-		memcpy_aligned<4>(&page_zip->data[page_free_f], page_free, 4);
-		/* TODO: group with PAGE_LAST_INSERT */
-
-		if (!page_is_leaf(page)) {
-			/* Zero out the node pointer of free_rec,
-			in case it will not be overwritten by
-			insert_rec. */
-
-			ut_ad(rec_size > REC_NODE_PTR_SIZE);
-
-			if (rec_offs_extra_size(foffsets)
-			    + rec_offs_data_size(foffsets) > rec_size) {
-
-				memset(rec_get_end(page + free_rec, foffsets)
-				       - REC_NODE_PTR_SIZE, 0,
-				       REC_NODE_PTR_SIZE);
-			}
-		} else if (dict_index_is_clust(index)) {
-			/* Zero out the DB_TRX_ID and DB_ROLL_PTR
-			columns of free_rec, in case it will not be
-			overwritten by insert_rec. */
-
-			ulint	trx_id_offs;
-			ulint	len;
-
-			trx_id_offs = rec_get_nth_field_offs(
-				foffsets, index->db_trx_id(), &len);
-			ut_ad(len == DATA_TRX_ID_LEN);
-
-			if (DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN + trx_id_offs
-			    + rec_offs_extra_size(foffsets) > rec_size) {
-				/* We will have to zero out the
-				DB_TRX_ID and DB_ROLL_PTR, because
-				they will not be fully overwritten by
-				insert_rec. */
-
-				memset(page + free_rec + trx_id_offs, 0,
-				       DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
-			}
-
-			ut_ad(free_rec + trx_id_offs + DATA_TRX_ID_LEN
-			      == rec_get_nth_field(free_rec, foffsets,
-						   index->db_roll_ptr(), &len));
-			ut_ad(len == DATA_ROLL_PTR_LEN);
-		}
-
-		if (UNIV_LIKELY_NULL(heap)) {
-			mem_heap_free(heap);
-		}
-	} else {
-use_heap:
-		ut_ad(!free_rec);
-		insert_buf = page_mem_alloc_heap<true>(cursor->block, rec_size,
-						       &heap_no, mtr);
-
-		if (UNIV_UNLIKELY(insert_buf == NULL)) {
-			return(NULL);
-		}
-
-		page_zip_dir_add_slot(cursor->block, index, mtr);
-	}
-
-	/* 3. Create the record */
-	insert_rec = rec_copy(insert_buf, rec, offsets);
-	rec_offs_make_valid(insert_rec, index, page_is_leaf(page), offsets);
-
-	/* 4. Insert the record in the linked list of records */
-	ut_ad(cursor->rec != insert_rec);
-
-	/* next record after current before the insertion */
-	const rec_t* next_rec = page_rec_get_next_low(cursor->rec, TRUE);
-	ut_ad(rec_get_status(cursor->rec) <= REC_STATUS_INFIMUM);
-	ut_ad(rec_get_status(insert_rec) < REC_STATUS_INFIMUM);
-	ut_ad(rec_get_status(next_rec) != REC_STATUS_INFIMUM);
-
-	mach_write_to_2(insert_rec - REC_NEXT, static_cast<uint16_t>
-			(next_rec - insert_rec));
-	mach_write_to_2(cursor->rec - REC_NEXT, static_cast<uint16_t>
-			(insert_rec - cursor->rec));
-	byte* n_recs = my_assume_aligned<2>(PAGE_N_RECS + PAGE_HEADER + page);
-	mtr->write<2>(*cursor->block, n_recs, 1U + mach_read_from_2(n_recs));
-	memcpy_aligned<2>(&page_zip->data[PAGE_N_RECS + PAGE_HEADER], n_recs,
-			  2);
-
-	/* 5. Set the n_owned field in the inserted record to zero,
-	and set the heap_no field */
-	rec_set_bit_field_1(insert_rec, 0, REC_NEW_N_OWNED,
-			    REC_N_OWNED_MASK, REC_N_OWNED_SHIFT);
-	rec_set_bit_field_2(insert_rec, heap_no, REC_NEW_HEAP_NO,
-			    REC_HEAP_NO_MASK, REC_HEAP_NO_SHIFT);
-
-	UNIV_MEM_ASSERT_RW(rec_get_start(insert_rec, offsets),
-			   rec_offs_size(offsets));
-
-	page_zip_dir_insert(cursor, free_rec, insert_rec, mtr);
-
-	/* 6. Update the last insertion info in page header */
-	byte* last_insert = my_assume_aligned<4>(PAGE_LAST_INSERT + PAGE_HEADER
-						 + page);
-	const uint16_t last_insert_rec = mach_read_from_2(last_insert);
-	ut_ad(!last_insert_rec
-	      || rec_get_node_ptr_flag(page + last_insert_rec)
-	      == rec_get_node_ptr_flag(insert_rec));
-	/* FIXME: combine with PAGE_DIRECTION changes */
-	mtr->write<2>(*cursor->block, last_insert, page_offset(insert_rec));
-	memcpy_aligned<4>(&page_zip->data[PAGE_LAST_INSERT + PAGE_HEADER],
-			  last_insert, 2);
-
-	if (!index->is_spatial()) {
-		byte* ptr = PAGE_HEADER + PAGE_DIRECTION_B + page;
-		if (UNIV_UNLIKELY(!last_insert_rec)) {
-no_direction:
-			page_zip->data[PAGE_HEADER + PAGE_DIRECTION_B] = *ptr
-				= PAGE_NO_DIRECTION;
-			memset_aligned<2>(PAGE_HEADER + PAGE_N_DIRECTION + page,
-					  0, 2);
-			memset_aligned<2>(PAGE_HEADER + PAGE_N_DIRECTION
-					  + page_zip->data, 0, 2);
-		} else if (page + last_insert_rec == cursor->rec
-			   && page_ptr_get_direction(ptr) != PAGE_LEFT) {
-			page_direction_increment(cursor->block, ptr,
-						 PAGE_RIGHT);
-		} else if (page_ptr_get_direction(ptr) != PAGE_RIGHT
-			   && page_rec_get_next(insert_rec)
-			   == page + last_insert_rec) {
-			page_direction_increment(cursor->block, ptr,
-						 PAGE_LEFT);
-		} else {
-			goto no_direction;
-		}
-	}
-
-	/* 7. It remains to update the owner record. */
-	ulint	n_owned;
-
-	while (!(n_owned = rec_get_n_owned_new(next_rec))) {
-		next_rec = page_rec_get_next_low(next_rec, true);
-	}
-
-	rec_set_bit_field_1(const_cast<rec_t*>(next_rec), n_owned + 1,
-			    REC_NEW_N_OWNED,
-			    REC_N_OWNED_MASK, REC_N_OWNED_SHIFT);
-
-	/* 8. Now we have incremented the n_owned field of the owner
-	record. If the number exceeds PAGE_DIR_SLOT_MAX_N_OWNED,
-	we have to split the corresponding directory slot in two. */
-	if (UNIV_UNLIKELY(n_owned == PAGE_DIR_SLOT_MAX_N_OWNED)) {
-		page_dir_split_slot<true>(page_cur_get_block(cursor),
-					  page_dir_find_owner_slot(next_rec),
-					  mtr);
-	}
-
-	page_zip_write_rec(cursor->block, insert_rec, index, offsets, 1, mtr);
-	return insert_rec;
-}
-
-/**********************************************************//**
-Parses a log record of copying a record list end to a new created page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_parse_copy_rec_list_to_created_page(
-/*=====================================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr)	/*!< in: mtr or NULL */
-{
-	ulint		log_data_len;
-
-	ut_ad(index->is_dummy);
-
-	if (ptr + 4 > end_ptr) {
-
-		return(NULL);
-	}
-
-	log_data_len = mach_read_from_4(ptr);
-	ptr += 4;
-
-	const byte* rec_end = ptr + log_data_len;
-
-	if (rec_end > end_ptr) {
-
-		return(NULL);
-	}
-
-	if (!block) {
-
-		return(rec_end);
-	}
-
-	ut_ad(fil_page_index_page_check(block->frame));
-	/* This function is never invoked on the clustered index root page,
-	except in the redo log apply of
-	page_copy_rec_list_end_to_created_page().
-	For other pages, this field must be zero-initialized. */
-	ut_ad(!page_get_instant(block->frame)
-	      || !page_has_siblings(block->frame));
-
-	while (ptr < rec_end) {
-		ptr = page_cur_parse_insert_rec(true, ptr, end_ptr,
-						block, index, mtr);
-	}
-
-	ut_a(ptr == rec_end);
-
-	memset_aligned<2>(PAGE_HEADER + PAGE_LAST_INSERT + block->frame, 0, 2);
-	if (block->page.zip.data) {
-		memset_aligned<2>(PAGE_HEADER + PAGE_LAST_INSERT
-				  + block->page.zip.data, 0, 2);
-	}
-
-	if (index->is_spatial()) {
-		return rec_end;
-	}
-
-	block->frame[PAGE_HEADER + PAGE_DIRECTION_B] &= ~((1U << 3) - 1);
-	block->frame[PAGE_HEADER + PAGE_DIRECTION_B] |= PAGE_NO_DIRECTION;
-	if (block->page.zip.data) {
-		block->page.zip.data[PAGE_HEADER + PAGE_DIRECTION_B]
-			= PAGE_NO_DIRECTION;
-	}
-
-	return(rec_end);
-}
-
-/*************************************************************//**
-Copies records from page to a newly created page, from a given record onward,
-including that record. Infimum and supremum records are not copied.
-
-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(
-/*===================================*/
-	buf_block_t*	block,		/*!< in/out: index page to copy to */
-	rec_t*		rec,		/*!< in: first record to copy */
-	dict_index_t*	index,		/*!< in: record descriptor */
-	mtr_t*		mtr)		/*!< in: mtr */
-{
-	page_dir_slot_t* slot = 0; /* remove warning */
-	page_t*	new_page = block->frame;
-	byte*	heap_top;
-	rec_t*	insert_rec = 0; /* remove warning */
-	rec_t*	prev_rec;
-	ulint	count;
-	ulint	n_recs;
-	ulint	slot_index;
-	ulint	rec_size;
-	mem_heap_t*	heap		= NULL;
-	offset_t	offsets_[REC_OFFS_NORMAL_SIZE];
-	offset_t*	offsets		= offsets_;
-	rec_offs_init(offsets_);
-
-	/* The record was never emitted for ROW_FORMAT=COMPRESSED pages. */
-	ut_ad(!block->page.zip.data);
-	ut_ad(page_dir_get_n_heap(new_page) == PAGE_HEAP_NO_USER_LOW);
-	ut_ad(page_align(rec) != new_page);
-	ut_ad(page_rec_is_comp(rec) == page_is_comp(new_page));
-	ut_ad(fil_page_index_page_check(new_page));
-	/* This function is never invoked on the clustered index root page,
-	except in btr_lift_page_up(). */
-	ut_ad(!page_get_instant(new_page) || !page_has_siblings(new_page));
-
-	if (page_rec_is_infimum(rec)) {
-
-		rec = page_rec_get_next(rec);
-	}
-
-	if (page_rec_is_supremum(rec)) {
-
-		return;
-	}
-
-#ifdef UNIV_DEBUG
-	/* To pass the debug tests we have to set these dummy values
-	in the debug version */
-	mach_write_to_2(PAGE_HEADER + PAGE_N_DIR_SLOTS + new_page,
-			srv_page_size / 2);
-	mach_write_to_2(PAGE_HEADER + PAGE_HEAP_TOP + new_page,
-			srv_page_size - 1);
-#endif
-	prev_rec = page_get_infimum_rec(new_page);
-	if (page_is_comp(new_page)) {
-		heap_top = new_page + PAGE_NEW_SUPREMUM_END;
-	} else {
-		heap_top = new_page + PAGE_OLD_SUPREMUM_END;
-	}
-	count = 0;
-	slot_index = 0;
-	n_recs = 0;
-
-	const bool is_leaf = page_is_leaf(new_page);
-
-	do {
-		offsets = rec_get_offsets(rec, index, offsets, is_leaf,
-					  ULINT_UNDEFINED, &heap);
-		insert_rec = rec_copy(heap_top, rec, offsets);
-
-		const bool comp = page_is_comp(new_page) != 0;
-
-		if (comp) {
-			rec_set_next_offs_new(prev_rec,
-					      page_offset(insert_rec));
-		} else {
-			rec_set_next_offs_old(prev_rec,
-					      page_offset(insert_rec));
-		}
-
-		page_rec_set_n_owned<false>(block, insert_rec, 0, comp, mtr);
-
-		rec_set_heap_no(insert_rec, PAGE_HEAP_NO_USER_LOW + n_recs,
-				page_is_comp(new_page));
-
-		count++;
-		n_recs++;
-
-		if (UNIV_UNLIKELY
-		    (count == (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2)) {
-
-			slot_index++;
-
-			slot = page_dir_get_nth_slot(new_page, slot_index);
-			mach_write_to_2(slot, page_offset(insert_rec));
-			page_dir_slot_set_n_owned<false>(block, slot, count,
-							 mtr);
-
-			count = 0;
-		}
-
-		rec_size = rec_offs_size(offsets);
-
-		ut_ad(heap_top < new_page + srv_page_size);
-
-		heap_top += rec_size;
-
-		rec_offs_make_valid(insert_rec, index, is_leaf, offsets);
-		prev_rec = insert_rec;
-		rec = page_rec_get_next(rec);
-	} while (!page_rec_is_supremum(rec));
-
-	ut_ad(n_recs);
-
-	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. */
-
-		count += (PAGE_DIR_SLOT_MAX_N_OWNED + 1) / 2;
-
-		page_dir_slot_set_n_owned<false>(block, slot, 0, mtr);
-
-		slot_index--;
-	}
-
-	if (UNIV_LIKELY_NULL(heap)) {
-		mem_heap_free(heap);
-	}
-
-	slot = page_dir_get_nth_slot(new_page, 1 + slot_index);
-
-	if (page_is_comp(new_page)) {
-		rec_set_next_offs_new(insert_rec, PAGE_NEW_SUPREMUM);
-		mach_write_to_2(slot, PAGE_NEW_SUPREMUM);
-		rec_set_bit_field_1(new_page + PAGE_NEW_SUPREMUM, count + 1,
-				    REC_NEW_N_OWNED,
-				    REC_N_OWNED_MASK, REC_N_OWNED_SHIFT);
-	} else {
-		rec_set_next_offs_old(insert_rec, PAGE_OLD_SUPREMUM);
-		mach_write_to_2(slot, PAGE_OLD_SUPREMUM);
-		rec_set_bit_field_1(new_page + PAGE_OLD_SUPREMUM, count + 1,
-				    REC_OLD_N_OWNED,
-				    REC_N_OWNED_MASK, REC_N_OWNED_SHIFT);
-	}
-
-	mach_write_to_2(PAGE_HEADER + PAGE_N_DIR_SLOTS + new_page,
-			2 + slot_index);
-	mach_write_to_2(PAGE_HEADER + PAGE_HEAP_TOP + new_page,
-			page_offset(heap_top));
-	mach_write_to_2(PAGE_HEADER + PAGE_N_HEAP + new_page,
-			PAGE_HEAP_NO_USER_LOW + n_recs);
-	mach_write_to_2(PAGE_HEADER + PAGE_N_RECS + new_page, n_recs);
-
-	memset_aligned<2>(PAGE_HEADER + PAGE_LAST_INSERT + new_page, 0, 2);
-	mach_write_to_1(PAGE_HEADER + PAGE_DIRECTION_B + new_page,
-			(mach_read_from_1(PAGE_HEADER + PAGE_DIRECTION_B
-					  + new_page) & ~((1U << 3) - 1))
-			| PAGE_NO_DIRECTION);
-	memset_aligned<2>(PAGE_HEADER + PAGE_N_DIRECTION + new_page, 0, 2);
-}
-
-/***********************************************************//**
-Parses log record of a record delete on a page.
-@return pointer to record end or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_cur_parse_delete_rec(
-/*======================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in: page or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr)	/*!< in/out: mini-transaction,
-				or NULL if block=NULL */
-{
-	ulint		offset;
-	page_cur_t	cursor;
-
-	ut_ad(!block == !mtr);
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	/* Read the cursor rec offset as a 2-byte ulint */
-	offset = mach_read_from_2(ptr);
-	ptr += 2;
-
-	if (UNIV_UNLIKELY(offset >= srv_page_size)) {
-		recv_sys.found_corrupt_log = true;
-		return NULL;
-	}
-
-	if (block) {
-		page_t*		page		= buf_block_get_frame(block);
-		mem_heap_t*	heap		= NULL;
-		offset_t	offsets_[REC_OFFS_NORMAL_SIZE];
-		rec_t*		rec		= page + offset;
-		rec_offs_init(offsets_);
-
-		page_cur_position(rec, block, &cursor);
-		ut_ad(!buf_block_get_page_zip(block) || page_is_comp(page));
-
-		page_cur_delete_rec(&cursor, index,
-				    rec_get_offsets(rec, index, offsets_,
-						    page_rec_is_leaf(rec),
-						    ULINT_UNDEFINED, &heap),
-				    mtr);
-		if (UNIV_LIKELY_NULL(heap)) {
-			mem_heap_free(heap);
-		}
-	}
-
-	return(ptr);
-}
-
 /** Prepend a record to the PAGE_FREE list.
 @param[in,out]  block   index page
 @param[in,out]  rec     record being deleted
@@ -2811,7 +1839,6 @@ page_cur_delete_rec(
 	ut_ad(fil_page_index_page_check(block->frame));
 	ut_ad(mach_read_from_8(PAGE_HEADER + PAGE_INDEX_ID + block->frame)
 	      == index->id
-	      || index->is_dummy
 	      || mtr->is_inside_ibuf());
 	ut_ad(mtr->is_named_space(index->table->space));
 
@@ -2819,11 +1846,6 @@ page_cur_delete_rec(
 	ut_ad(page_rec_is_user_rec(current_rec));
 
 	if (page_get_n_recs(block->frame) == 1
-	    /* Empty the page, unless we are applying the redo log
-	    during crash recovery. During normal operation, the
-	    page_create_empty() gets logged as one of MLOG_PAGE_CREATE,
-	    MLOG_COMP_PAGE_CREATE, MLOG_ZIP_PAGE_COMPRESS. */
-	    && !recv_recovery_is_on() && !log_sys.is_physical()
 	    && !rec_is_alter_metadata(current_rec, *index)) {
 		/* Empty the page. */
 		ut_ad(page_is_leaf(block->frame));
diff --git a/storage/innobase/page/page0page.cc b/storage/innobase/page/page0page.cc
index 7b7479906cf..66b4aa55f1a 100644
--- a/storage/innobase/page/page0page.cc
+++ b/storage/innobase/page/page0page.cc
@@ -826,62 +826,6 @@ zip_reorganize:
 	return(ret);
 }
 
-/**********************************************************//**
-Parses a log record of a record list end or start deletion.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_parse_delete_rec_list(
-/*=======================*/
-	mlog_id_t	type,	/*!< in: MLOG_LIST_END_DELETE,
-				MLOG_LIST_START_DELETE,
-				MLOG_COMP_LIST_END_DELETE or
-				MLOG_COMP_LIST_START_DELETE */
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	buf_block_t*	block,	/*!< in/out: buffer block or NULL */
-	dict_index_t*	index,	/*!< in: record descriptor */
-	mtr_t*		mtr)	/*!< in: mtr or NULL */
-{
-	page_t*	page;
-	ulint	offset;
-
-	ut_ad(type == MLOG_LIST_END_DELETE
-	      || type == MLOG_LIST_START_DELETE
-	      || type == MLOG_COMP_LIST_END_DELETE
-	      || type == MLOG_COMP_LIST_START_DELETE);
-
-	/* Read the record offset as a 2-byte ulint */
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	offset = mach_read_from_2(ptr);
-	ptr += 2;
-
-	if (!block) {
-
-		return(ptr);
-	}
-
-	page = buf_block_get_frame(block);
-
-	ut_ad(!!page_is_comp(page) == dict_table_is_comp(index->table));
-
-	if (type == MLOG_LIST_END_DELETE
-	    || type == MLOG_COMP_LIST_END_DELETE) {
-		page_delete_rec_list_end(page + offset, block, index,
-					 ULINT_UNDEFINED, ULINT_UNDEFINED,
-					 mtr);
-	} else {
-		page_delete_rec_list_start(page + offset, block, index, mtr);
-	}
-
-	return(ptr);
-}
-
 /*************************************************************//**
 Deletes records from a page from a given record onward, including that record.
 The infimum and supremum records are not deleted. */
diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index c6739f067f4..fa72090d651 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -3847,67 +3847,6 @@ void page_zip_write_rec(buf_block_t *block, const byte *rec,
 #endif /* UNIV_ZIP_DEBUG */
 }
 
-/***********************************************************//**
-Parses a log record of writing a BLOB pointer of a record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_blob_ptr(
-/*==========================*/
-	const byte*	ptr,	/*!< in: redo log buffer */
-	const byte*	end_ptr,/*!< in: redo log buffer end */
-	page_t*		page,	/*!< in/out: uncompressed page */
-	page_zip_des_t*	page_zip)/*!< in/out: compressed page */
-{
-	ulint	offset;
-	ulint	z_offset;
-
-	ut_ad(ptr != NULL);
-	ut_ad(end_ptr != NULL);
-	ut_ad(!page == !page_zip);
-
-	if (UNIV_UNLIKELY
-	    (end_ptr < ptr + (2 + 2 + BTR_EXTERN_FIELD_REF_SIZE))) {
-
-		return(NULL);
-	}
-
-	offset = mach_read_from_2(ptr);
-	z_offset = mach_read_from_2(ptr + 2);
-
-	if (offset < PAGE_ZIP_START
-	    || offset >= srv_page_size
-	    || z_offset >= srv_page_size) {
-corrupt:
-		recv_sys.found_corrupt_log = TRUE;
-
-		return(NULL);
-	}
-
-	if (page) {
-
-		if (!page_zip || !page_is_leaf(page)) {
-
-			goto corrupt;
-		}
-
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-
-		memcpy(page + offset,
-		       ptr + 4, BTR_EXTERN_FIELD_REF_SIZE);
-		memcpy(page_zip->data + z_offset,
-		       ptr + 4, BTR_EXTERN_FIELD_REF_SIZE);
-
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-	}
-
-	return(ptr + (2 + 2 + BTR_EXTERN_FIELD_REF_SIZE));
-}
-
 /**********************************************************************//**
 Write a BLOB pointer of a record on the leaf page of a clustered index.
 The information must already have been updated on the uncompressed page. */
@@ -3973,82 +3912,6 @@ page_zip_write_blob_ptr(
 #endif /* UNIV_ZIP_DEBUG */
 }
 
-/***********************************************************//**
-Parses a log record of writing the node pointer of a record.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_node_ptr(
-/*==========================*/
-	const byte*	ptr,	/*!< in: redo log buffer */
-	const byte*	end_ptr,/*!< in: redo log buffer end */
-	page_t*		page,	/*!< in/out: uncompressed page */
-	page_zip_des_t*	page_zip)/*!< in/out: compressed page */
-{
-	ulint	offset;
-	ulint	z_offset;
-
-	ut_ad(ptr != NULL);
-	ut_ad(end_ptr!= NULL);
-	ut_ad(!page == !page_zip);
-
-	if (UNIV_UNLIKELY(end_ptr < ptr + (2 + 2 + REC_NODE_PTR_SIZE))) {
-
-		return(NULL);
-	}
-
-	offset = mach_read_from_2(ptr);
-	z_offset = mach_read_from_2(ptr + 2);
-
-	if (offset < PAGE_ZIP_START
-	    || offset >= srv_page_size
-	    || z_offset >= srv_page_size) {
-corrupt:
-		recv_sys.found_corrupt_log = TRUE;
-
-		return(NULL);
-	}
-
-	if (page) {
-		byte*	storage_end;
-		byte*	field;
-		byte*	storage;
-		ulint	heap_no;
-
-		if (!page_zip || page_is_leaf(page)) {
-
-			goto corrupt;
-		}
-
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-
-		field = page + offset;
-		storage = page_zip->data + z_offset;
-
-		storage_end = page_zip_dir_start(page_zip);
-
-		heap_no = 1 + ulint(storage_end - storage) / REC_NODE_PTR_SIZE;
-
-		if (UNIV_UNLIKELY((storage_end - storage) % REC_NODE_PTR_SIZE)
-		    || UNIV_UNLIKELY(heap_no < PAGE_HEAP_NO_USER_LOW)
-		    || UNIV_UNLIKELY(heap_no >= page_dir_get_n_heap(page))) {
-
-			goto corrupt;
-		}
-
-		memcpy(field, ptr + 4, REC_NODE_PTR_SIZE);
-		memcpy(storage, ptr + 4, REC_NODE_PTR_SIZE);
-
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-	}
-
-	return(ptr + (2 + 2 + REC_NODE_PTR_SIZE));
-}
-
 /**********************************************************************//**
 Write the node pointer of a record on a non-leaf compressed page. */
 void
@@ -4194,67 +4057,6 @@ write:
 	UNIV_MEM_ASSERT_RW(page_zip->data, page_zip_get_size(page_zip));
 }
 
-/** Parse a MLOG_ZIP_WRITE_TRX_ID record.
-@param[in]	ptr		redo log buffer
-@param[in]	end_ptr		end of redo log buffer
-@param[in,out]	page		uncompressed page
-@param[in,out]	page_zip	compressed page
-@return end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_trx_id(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	page_t*		page,
-	page_zip_des_t*	page_zip)
-{
-	const byte* const end = 2 + 2 + DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN
-		+ ptr;
-
-	if (UNIV_UNLIKELY(end_ptr < end)) {
-		return(NULL);
-	}
-
-	uint offset = mach_read_from_2(ptr);
-	uint z_offset = mach_read_from_2(ptr + 2);
-
-	if (offset < PAGE_ZIP_START
-	    || offset >= srv_page_size
-	    || z_offset >= srv_page_size) {
-corrupt:
-		recv_sys.found_corrupt_log = TRUE;
-
-		return(NULL);
-	}
-
-	if (page) {
-		if (!page_zip || !page_is_leaf(page)) {
-			goto corrupt;
-		}
-
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-
-		byte* field = page + offset;
-		byte* storage = page_zip->data + z_offset;
-
-		if (storage >= page_zip_dir_start(page_zip)) {
-			goto corrupt;
-		}
-
-		memcpy(field, ptr + 4, DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
-		memcpy(storage, ptr + 4, DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
-
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-	}
-
-	return end;
-}
-
 /**********************************************************************//**
 Clear an area on the uncompressed and compressed page.
 Do not clear the data payload, as that would grow the modification log. */
@@ -4589,69 +4391,10 @@ void page_zip_dir_delete(buf_block_t *block, byte *rec,
   page_zip_clear_rec(block, rec, index, offsets, mtr);
 }
 
-/***********************************************************//**
-Parses a log record of writing to the header of a page.
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte*
-page_zip_parse_write_header(
-/*========================*/
-	const byte*	ptr,	/*!< in: redo log buffer */
-	const byte*	end_ptr,/*!< in: redo log buffer end */
-	page_t*		page,	/*!< in/out: uncompressed page */
-	page_zip_des_t*	page_zip)/*!< in/out: compressed page */
-{
-	ulint	offset;
-	ulint	len;
-
-	ut_ad(ptr != NULL);
-	ut_ad(end_ptr!= NULL);
-	ut_ad(!page == !page_zip);
-
-	if (UNIV_UNLIKELY(end_ptr < ptr + (1 + 1))) {
-
-		return(NULL);
-	}
-
-	offset = (ulint) *ptr++;
-	len = (ulint) *ptr++;
-
-	if (len == 0 || offset + len >= PAGE_DATA) {
-corrupt:
-		recv_sys.found_corrupt_log = TRUE;
-
-		return(NULL);
-	}
-
-	if (end_ptr < ptr + len) {
-
-		return(NULL);
-	}
-
-	if (page) {
-		if (!page_zip) {
-
-			goto corrupt;
-		}
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-
-		memcpy(page + offset, ptr, len);
-		memcpy(page_zip->data + offset, ptr, len);
-
-#ifdef UNIV_ZIP_DEBUG
-		ut_a(page_zip_validate(page_zip, page, NULL));
-#endif /* UNIV_ZIP_DEBUG */
-	}
-
-	return(ptr + len);
-}
-
 /**********************************************************************//**
 Reorganize and compress a page.  This is a low-level operation for
 compressed pages, to be used when page_zip_compress() fails.
-On success, a redo log entry MLOG_ZIP_PAGE_COMPRESS will be written.
+On success, redo log will be written.
 The function btr_page_reorganize() should be preferred whenever possible.
 IMPORTANT: if page_zip_reorganize() is invoked on a leaf page of a
 non-clustered index, the caller must update the insert buffer free
@@ -4861,72 +4604,6 @@ page_zip_copy_recs(
 #endif /* UNIV_ZIP_DEBUG */
 	page_zip_compress_write_log(block, index, mtr);
 }
-
-/** Parse and optionally apply MLOG_ZIP_PAGE_COMPRESS.
-@param[in]	ptr	log record
-@param[in]	end_ptr	end of log
-@param[in,out]	block	ROW_FORMAT=COMPRESSED block, or NULL for parsing only
-@return	end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-const byte* page_zip_parse_compress(const byte* ptr, const byte* end_ptr,
-				    buf_block_t* block)
-{
-	ulint	size;
-	ulint	trailer_size;
-
-	ut_ad(ptr != NULL);
-	ut_ad(end_ptr!= NULL);
-
-	if (UNIV_UNLIKELY(ptr + (2 + 2) > end_ptr)) {
-
-		return(NULL);
-	}
-
-	size = mach_read_from_2(ptr);
-	ptr += 2;
-	trailer_size = mach_read_from_2(ptr);
-	ptr += 2;
-
-	if (UNIV_UNLIKELY(ptr + 8 + size + trailer_size > end_ptr)) {
-
-		return(NULL);
-	}
-
-	if (block) {
-		ut_ad(buf_block_get_state(block) == BUF_BLOCK_FILE_PAGE);
-		page_zip_des_t* page_zip = buf_block_get_page_zip(block);
-		if (!page_zip || page_zip_get_size(page_zip) < size
-		    || block->page.id.page_no() < 3) {
-corrupt:
-			recv_sys.found_corrupt_log = TRUE;
-
-			return(NULL);
-		}
-
-		memset(page_zip->data, 0, page_zip_get_size(page_zip));
-		mach_write_to_4(FIL_PAGE_OFFSET
-				+ page_zip->data, block->page.id.page_no());
-		mach_write_to_4(FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID
-				+ page_zip->data, block->page.id.space());
-		memcpy(page_zip->data + FIL_PAGE_PREV, ptr, 4);
-		memcpy(page_zip->data + FIL_PAGE_NEXT, ptr + 4, 4);
-		memcpy(page_zip->data + FIL_PAGE_TYPE, ptr + 8, size);
-		memset(page_zip->data + FIL_PAGE_TYPE + size, 0,
-		       page_zip_get_size(page_zip) - trailer_size
-		       - (FIL_PAGE_TYPE + size));
-		memcpy(page_zip->data + page_zip_get_size(page_zip)
-		       - trailer_size, ptr + 8 + size, trailer_size);
-
-		if (UNIV_UNLIKELY(!page_zip_decompress(page_zip, block->frame,
-						       TRUE))) {
-
-			goto corrupt;
-		}
-	}
-
-	return(const_cast<byte*>(ptr) + 8 + size + trailer_size);
-}
 #endif /* !UNIV_INNOCHECKSUM */
 
 /** Calculate the compressed page checksum.
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index 0a3656b2c10..b72c55befb4 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -166,7 +166,7 @@ rec_get_n_extern_new(
 	ulint		i;
 
 	ut_ad(dict_table_is_comp(index->table));
-	ut_ad(!index->table->supports_instant() || index->is_dummy);
+	ut_ad(!index->table->supports_instant());
 	ut_ad(!index->is_instant());
 	ut_ad(rec_get_status(rec) == REC_STATUS_ORDINARY
 	      || rec_get_status(rec) == REC_STATUS_INSTANT);
@@ -283,8 +283,8 @@ rec_init_offsets_comp_ordinary(
 	ut_d(ulint n_null= 0);
 
 	if (mblob) {
-		ut_ad(index->is_dummy || index->table->instant);
-		ut_ad(index->is_dummy || index->is_instant());
+		ut_ad(index->table->instant);
+		ut_ad(index->is_instant());
 		ut_ad(rec_offs_n_fields(offsets)
 		      <= ulint(index->n_fields) + 1);
 		ut_ad(!def_val);
@@ -908,8 +908,8 @@ rec_get_offsets_func(
 		memcpy(&offsets[INDEX_OFFSET], &index, sizeof index);
 #endif /* UNIV_DEBUG */
 		ut_ad(leaf);
-		ut_ad(index->is_dummy || index->table->instant);
-		ut_ad(index->is_dummy || index->is_instant());
+		ut_ad(index->table->instant);
+		ut_ad(index->is_instant());
 		ut_ad(rec_offs_n_fields(offsets)
 		      <= ulint(index->n_fields) + 1);
 		rec_init_offsets_comp_ordinary<true>(rec, index, offsets,
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 97094abc8a3..f17c3eb582b 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -585,92 +585,6 @@ row_upd_changes_disowned_external(
 	return(false);
 }
 
-/*********************************************************************//**
-Parses the log data written by row_upd_index_write_log.
-@return log data end or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-row_upd_index_parse(
-/*================*/
-	const byte*	ptr,	/*!< in: buffer */
-	const byte*	end_ptr,/*!< in: buffer end */
-	mem_heap_t*	heap,	/*!< in: memory heap where update vector is
-				built */
-	upd_t**		update_out)/*!< out: update vector */
-{
-	upd_t*		update;
-	upd_field_t*	upd_field;
-	dfield_t*	new_val;
-	ulint		len;
-	ulint		n_fields;
-	ulint		info_bits;
-	ulint		i;
-
-	if (end_ptr < ptr + 1) {
-
-		return(NULL);
-	}
-
-	info_bits = mach_read_from_1(ptr);
-	ptr++;
-	n_fields = mach_parse_compressed(&ptr, end_ptr);
-
-	if (ptr == NULL) {
-
-		return(NULL);
-	}
-
-	update = upd_create(n_fields, heap);
-	update->info_bits = info_bits;
-
-	for (i = 0; i < n_fields; i++) {
-		ulint	field_no;
-		upd_field = upd_get_nth_field(update, i);
-		new_val = &(upd_field->new_val);
-
-		field_no = mach_parse_compressed(&ptr, end_ptr);
-
-		if (ptr == NULL) {
-
-			return(NULL);
-		}
-
-		/* Check if this is a virtual column, mark the prtype
-		if that is the case */
-		if (field_no >= REC_MAX_N_FIELDS) {
-			new_val->type.prtype |= DATA_VIRTUAL;
-			field_no -= REC_MAX_N_FIELDS;
-		}
-
-		upd_field->field_no = field_no;
-
-		len = mach_parse_compressed(&ptr, end_ptr);
-
-		if (ptr == NULL) {
-
-			return(NULL);
-		}
-
-		if (len != UNIV_SQL_NULL) {
-
-			if (end_ptr < ptr + len) {
-
-				return(NULL);
-			}
-
-			dfield_set_data(new_val,
-					mem_heap_dup(heap, ptr, len), len);
-			ptr += len;
-		} else {
-			dfield_set_null(new_val);
-		}
-	}
-
-	*update_out = update;
-
-	return(const_cast<byte*>(ptr));
-}
-
 /***************************************************************//**
 Builds an update vector from those fields which in a secondary index entry
 differ from a record that has the equal ordering fields. NOTE: we compare
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index d215b07e3a3..50be3196fef 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -54,50 +54,6 @@ const dtuple_t trx_undo_metadata = {
 
 /*=========== UNDO LOG RECORD CREATION AND DECODING ====================*/
 
-/** Parse MLOG_UNDO_INSERT.
-@param[in]	ptr	log record
-@param[in]	end_ptr	end of log record buffer
-@param[in,out] page	page or NULL
-@return	end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_add_undo_rec(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	page_t*		page)
-{
-	ulint	len;
-
-	if (end_ptr < ptr + 2) {
-
-		return(NULL);
-	}
-
-	len = mach_read_from_2(ptr);
-	ptr += 2;
-
-	if (end_ptr < ptr + len) {
-
-		return(NULL);
-	}
-
-	if (page) {
-		ulint first_free = mach_read_from_2(page + TRX_UNDO_PAGE_HDR
-						    + TRX_UNDO_PAGE_FREE);
-		byte* rec = page + first_free;
-
-		mach_write_to_2(rec, first_free + 4 + len);
-		mach_write_to_2(rec + 2 + len, first_free);
-
-		mach_write_to_2(page + TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE,
-				first_free + 4 + len);
-		memcpy(rec + 2, ptr, len);
-	}
-
-	return(const_cast<byte*>(ptr + len));
-}
-
 /** Calculate the free space left for extending an undo log record.
 @param[in]	undo_block	undo log page
 @param[in]	ptr		current end of the undo page
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index 985cbeba2f9..9d1856a0b27 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -292,90 +292,6 @@ trx_undo_get_first_rec(const fil_space_t &space, uint32_t page_no,
 
 /*============== UNDO LOG FILE COPY CREATION AND FREEING ==================*/
 
-/** Parse MLOG_UNDO_INIT.
-@param[in]	ptr	log record
-@param[in]	end_ptr	end of log record buffer
-@param[in,out]	page	page or NULL
-@return	end of log record
-@retval	NULL	if the log record is incomplete */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_page_init(const byte* ptr, const byte* end_ptr, page_t* page)
-{
-	if (end_ptr <= ptr) {
-		return NULL;
-	}
-
-	const ulint type = *ptr++;
-
-	if (type > TRX_UNDO_UPDATE) {
-		recv_sys.found_corrupt_log = true;
-	} else if (page) {
-		/* Starting with MDEV-12288 in MariaDB 10.3.1, we use
-		type=0 for the combined insert/update undo log
-		pages. MariaDB 10.2 would use TRX_UNDO_INSERT or
-		TRX_UNDO_UPDATE. */
-		mach_write_to_2(FIL_PAGE_TYPE + page, FIL_PAGE_UNDO_LOG);
-		mach_write_to_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE + page,
-				type);
-		mach_write_to_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_START + page,
-				TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
-		mach_write_to_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE + page,
-				TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
-	}
-
-	return(const_cast<byte*>(ptr));
-}
-
-/** Parse MLOG_UNDO_HDR_REUSE for crash-upgrade from MariaDB 10.2.
-@param[in]	ptr	redo log record
-@param[in]	end_ptr	end of log buffer
-@param[in,out]	page	undo log page or NULL
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_page_header_reuse(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	page_t*		undo_page)
-{
-	trx_id_t	trx_id = mach_u64_parse_compressed(&ptr, end_ptr);
-
-	if (!ptr || !undo_page) {
-		return(const_cast<byte*>(ptr));
-	}
-
-	compile_time_assert(TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE
-			    + TRX_UNDO_LOG_XA_HDR_SIZE
-			    < UNIV_PAGE_SIZE_MIN - 100);
-
-	const ulint new_free = TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE
-		+ TRX_UNDO_LOG_OLD_HDR_SIZE;
-
-	/* Insert undo data is not needed after commit: we may free all
-	the space on the page */
-
-	ut_ad(mach_read_from_2(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE
-			       + undo_page)
-	      == TRX_UNDO_INSERT);
-
-	byte*	page_hdr = undo_page + TRX_UNDO_PAGE_HDR;
-	mach_write_to_2(page_hdr + TRX_UNDO_PAGE_START, new_free);
-	mach_write_to_2(page_hdr + TRX_UNDO_PAGE_FREE, new_free);
-	mach_write_to_2(TRX_UNDO_SEG_HDR + TRX_UNDO_STATE + undo_page,
-			TRX_UNDO_ACTIVE);
-
-	byte* log_hdr = undo_page + TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE;
-
-	mach_write_to_8(log_hdr + TRX_UNDO_TRX_ID, trx_id);
-	mach_write_to_2(log_hdr + TRX_UNDO_LOG_START, new_free);
-
-	mach_write_to_1(log_hdr + TRX_UNDO_XID_EXISTS, FALSE);
-	mach_write_to_1(log_hdr + TRX_UNDO_DICT_TRANS, FALSE);
-
-	return(const_cast<byte*>(ptr));
-}
-
 /** Initialize the fields in an undo log segment page.
 @param[in,out]	undo_block	undo log segment page
 @param[in,out]	mtr		mini-transaction */
@@ -609,30 +525,6 @@ trx_undo_read_xid(const trx_ulogf_t* log_hdr, XID* xid)
 	memcpy(xid->data, log_hdr + TRX_UNDO_XA_XID, XIDDATASIZE);
 }
 
-/** Parse the redo log entry of an undo log page header create.
-@param[in]	ptr	redo log record
-@param[in]	end_ptr	end of log buffer
-@param[in,out]	block	page frame or NULL
-@param[in,out]	mtr	mini-transaction or NULL
-@return end of log record or NULL */
-ATTRIBUTE_COLD /* only used when crash-upgrading */
-byte*
-trx_undo_parse_page_header(
-	const byte*	ptr,
-	const byte*	end_ptr,
-	buf_block_t*	block,
-	mtr_t*		mtr)
-{
-	trx_id_t	trx_id = mach_u64_parse_compressed(&ptr, end_ptr);
-
-	if (ptr && block) {
-		trx_undo_header_create(block, trx_id, mtr);
-
-	}
-
-	return const_cast<byte*>(ptr);
-}
-
 /** Allocate an undo log page.
 @param[in,out]	undo	undo log
 @param[in,out]	mtr	mini-transaction that does not hold any page latch