Merge 10.2 into 10.3

29 files changed, 374 insertions, 434 deletions

diff --git a/extra/mariabackup/xtrabackup.cc b/extra/mariabackup/xtrabackup.cc
index 001cce29933..338cf22a7de 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -2748,8 +2748,13 @@ static bool xtrabackup_copy_logfile(bool last = false)
 			my_sleep(1000);
 		}
 
-		start_lsn = (lsn == start_lsn)
-			? 0 : xtrabackup_copy_log(start_lsn, lsn, last);
+		if (lsn == start_lsn) {
+			start_lsn = 0;
+		} else {
+			mutex_enter(&recv_sys->mutex);
+			start_lsn = xtrabackup_copy_log(start_lsn, lsn, last);
+			mutex_exit(&recv_sys->mutex);
+		}
 
 		log_mutex_exit();
 
diff --git a/mysql-test/main/cte_nonrecursive.result b/mysql-test/main/cte_nonrecursive.result
index 9bce4f0a9bd..b846ec2d8ac 100644
--- a/mysql-test/main/cte_nonrecursive.result
+++ b/mysql-test/main/cte_nonrecursive.result
@@ -1659,3 +1659,17 @@ a
 2
 drop view v1;
 drop table t1,t2;
+#
+# MDEV-19112: CTE usage when information_schema is set as default db
+#
+with t as (select 1 as t ) select * from t;
+t
+1
+use information_schema;
+with t as (select 1 as t) select * from t;
+t
+1
+with columns as (select 1 as t) select * from columns;
+t
+1
+use test;
diff --git a/mysql-test/main/cte_nonrecursive.test b/mysql-test/main/cte_nonrecursive.test
index 920c27a70f9..c0c5c22bed0 100644
--- a/mysql-test/main/cte_nonrecursive.test
+++ b/mysql-test/main/cte_nonrecursive.test
@@ -1168,3 +1168,15 @@ select * from v1;
 
 drop view v1;
 drop table t1,t2;
+
+--echo #
+--echo # MDEV-19112: CTE usage when information_schema is set as default db
+--echo #
+
+with t as (select 1 as t ) select * from t;
+
+use information_schema;
+with t as (select 1 as t) select * from t;
+with columns as (select 1 as t) select * from columns;
+
+use test;
diff --git a/mysql-test/suite/innodb/t/drop_table_background.test b/mysql-test/suite/innodb/t/drop_table_background.test
index 8d82bea9675..20101dada84 100644
--- a/mysql-test/suite/innodb/t/drop_table_background.test
+++ b/mysql-test/suite/innodb/t/drop_table_background.test
@@ -3,6 +3,8 @@
 # Embedded server does not support restarting
 --source include/not_embedded.inc
 
+let $MYSQLD_DATADIR=`select @@datadir`;
+
 CREATE TABLE t(c0 SERIAL, c1 INT, c2 INT, c3 INT, c4 INT,
 KEY(c1), KEY(c2), KEY(c2,c1),
 KEY(c3), KEY(c3,c1), KEY(c3,c2), KEY(c3,c2,c1),
@@ -32,7 +34,9 @@ CREATE TABLE target (PRIMARY KEY(c1)) ENGINE=InnoDB SELECT * FROM t;
 --error ER_NO_SUCH_TABLE
 SELECT * from target;
 DROP TABLE t;
---source include/restart_mysqld.inc
+--source include/shutdown_mysqld.inc
+--remove_files_wildcard $MYSQLD_DATADIR/test #sql-*.ibd
+--source include/start_mysqld.inc
 CREATE TABLE t (a INT) ENGINE=InnoDB;
 DROP TABLE t;
 --error ER_BAD_TABLE_ERROR
diff --git a/mysql-test/suite/plugins/r/feedback_plugin_install.result b/mysql-test/suite/plugins/r/feedback_plugin_install.result
index ee74435e05a..c7f7a5c79f3 100644
--- a/mysql-test/suite/plugins/r/feedback_plugin_install.result
+++ b/mysql-test/suite/plugins/r/feedback_plugin_install.result
@@ -6,7 +6,7 @@ select * from information_schema.feedback where variable_name like 'feed%'
        and variable_name not like  '%_uid'
        and variable_name not like '%debug%';
 VARIABLE_NAME	VARIABLE_VALUE
-FEEDBACK used	1
+FEEDBACK used	2
 FEEDBACK version	1.1
 FEEDBACK_HTTP_PROXY	
 FEEDBACK_SEND_RETRY_WAIT	60
diff --git a/mysql-test/suite/plugins/r/feedback_plugin_load.result b/mysql-test/suite/plugins/r/feedback_plugin_load.result
index 8770ce19f49..f96b4d9b71f 100644
--- a/mysql-test/suite/plugins/r/feedback_plugin_load.result
+++ b/mysql-test/suite/plugins/r/feedback_plugin_load.result
@@ -4,7 +4,7 @@ ACTIVE
 SELECT variable_value INTO @feedback_used FROM information_schema.feedback where variable_name = 'FEEDBACK used';
 SELECT variable_value = @feedback_used + 1 FROM information_schema.feedback where variable_name = 'FEEDBACK used';
 variable_value = @feedback_used + 1
-1
+0
 select * from information_schema.feedback where variable_name like 'feed%'
        and variable_name not like '%_uid' and variable_name not like 'FEEDBACK used'
        and variable_name not like '%debug%';
diff --git a/mysql-test/suite/plugins/r/feedback_plugin_send.result b/mysql-test/suite/plugins/r/feedback_plugin_send.result
index 935ea11d67b..35db81fa6c1 100644
--- a/mysql-test/suite/plugins/r/feedback_plugin_send.result
+++ b/mysql-test/suite/plugins/r/feedback_plugin_send.result
@@ -4,7 +4,7 @@ ACTIVE
 SELECT variable_value INTO @feedback_used FROM information_schema.feedback where variable_name = 'FEEDBACK used';
 SELECT variable_value = @feedback_used + 1 FROM information_schema.feedback where variable_name = 'FEEDBACK used';
 variable_value = @feedback_used + 1
-1
+0
 select * from information_schema.feedback where variable_name like 'feed%'
        and variable_name not like '%_uid' and variable_name not like 'FEEDBACK used'
        and variable_name not like '%debug%';
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index cf9a15dec68..ddb8fee6f70 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2016, Oracle and/or its affiliates.
-   Copyright (c) 2010, 2016, MariaDB
+   Copyright (c) 2010, 2019, MariaDB
 
    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
@@ -3447,6 +3447,29 @@ open_and_process_table(THD *thd, LEX *lex, TABLE_LIST *tables,
         goto end;
     }
   }
+
+  if (!tables->derived && is_infoschema_db(&tables->db))
+  {
+    /*
+      Check whether the information schema contains a table
+      whose name is tables->schema_table_name
+    */
+    ST_SCHEMA_TABLE *schema_table;
+    schema_table= find_schema_table(thd, &tables->schema_table_name);
+    if (!schema_table ||
+        (schema_table->hidden &&
+         ((sql_command_flags[lex->sql_command] & CF_STATUS_COMMAND) == 0 ||
+          /*
+            this check is used for show columns|keys from I_S hidden table
+          */
+          lex->sql_command == SQLCOM_SHOW_FIELDS ||
+          lex->sql_command == SQLCOM_SHOW_KEYS)))
+    {
+      my_error(ER_UNKNOWN_TABLE, MYF(0),
+               tables->schema_table_name.str, INFORMATION_SCHEMA_NAME.str);
+      DBUG_RETURN(1);
+    }
+  }
   /*
     If this TABLE_LIST object is a placeholder for an information_schema
     table, create a temporary table to represent the information_schema
diff --git a/sql/sql_cte.cc b/sql/sql_cte.cc
index 01caeda1325..247d7e5a3d1 100644
--- a/sql/sql_cte.cc
+++ b/sql/sql_cte.cc
@@ -1109,6 +1109,7 @@ bool TABLE_LIST::set_as_with_table(THD *thd, With_element *with_elem)
     table= 0;
   }
   with= with_elem;
+  schema_table= NULL;
   if (!with_elem->is_referenced() || with_elem->is_recursive)
   {
     derived= with_elem->spec;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 7f4cc5adaf2..33a1fa8a291 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -8304,7 +8304,6 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
   ptr->derived=	    table->sel;
   if (!ptr->derived && is_infoschema_db(&ptr->db))
   {
-    ST_SCHEMA_TABLE *schema_table;
     if (ptr->updating &&
         /* Special cases which are processed by commands itself */
         lex->sql_command != SQLCOM_CHECK &&
@@ -8316,20 +8315,8 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
                INFORMATION_SCHEMA_NAME.str);
       DBUG_RETURN(0);
     }
+    ST_SCHEMA_TABLE *schema_table;
     schema_table= find_schema_table(thd, &ptr->table_name);
-    if (unlikely(!schema_table) ||
-        (schema_table->hidden && 
-         ((sql_command_flags[lex->sql_command] & CF_STATUS_COMMAND) == 0 || 
-          /*
-            this check is used for show columns|keys from I_S hidden table
-          */
-          lex->sql_command == SQLCOM_SHOW_FIELDS ||
-          lex->sql_command == SQLCOM_SHOW_KEYS)))
-    {
-      my_error(ER_UNKNOWN_TABLE, MYF(0),
-               ptr->table_name.str, INFORMATION_SCHEMA_NAME.str);
-      DBUG_RETURN(0);
-    }
     ptr->schema_table_name= ptr->table_name;
     ptr->schema_table= schema_table;
   }
diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index cf027053cf1..10e8dfc6904 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -273,11 +273,6 @@ btr_root_get(
 	And block the segment list access by others.*/
 	buf_block_t* root = btr_root_block_get(index, RW_SX_LATCH,
 					       mtr);
-
-	if (root && root->page.encrypted == true) {
-		root = NULL;
-	}
-
 	return(root ? buf_block_get_frame(root) : NULL);
 }
 
@@ -1081,17 +1076,17 @@ void btr_page_get_father(dict_index_t* index, buf_block_t* block, mtr_t* mtr,
 	mem_heap_free(heap);
 }
 
+/** PAGE_INDEX_ID value for freed index B-trees */
+static const index_id_t	BTR_FREED_INDEX_ID = 0;
+
 /** Free a B-tree root page. btr_free_but_not_root() must already
 have been called.
 In a persistent tablespace, the caller must invoke fsp_init_file_page()
 before mtr.commit().
-@param[in,out]	block	index root page
-@param[in,out]	mtr	mini-transaction */
-static
-void
-btr_free_root(
-	buf_block_t*	block,
-	mtr_t*		mtr)
+@param[in,out]	block		index root page
+@param[in,out]	mtr		mini-transaction
+@param[in]	invalidate	whether to invalidate PAGE_INDEX_ID */
+static void btr_free_root(buf_block_t* block, mtr_t* mtr, bool invalidate)
 {
 	fseg_header_t*	header;
 
@@ -1105,31 +1100,18 @@ btr_free_root(
 #ifdef UNIV_BTR_DEBUG
 	ut_a(btr_root_fseg_validate(header, block->page.id.space()));
 #endif /* UNIV_BTR_DEBUG */
+	if (invalidate) {
+		btr_page_set_index_id(
+			buf_block_get_frame(block),
+			buf_block_get_page_zip(block),
+			BTR_FREED_INDEX_ID, mtr);
+	}
 
 	while (!fseg_free_step(header, true, mtr)) {
 		/* Free the entire segment in small steps. */
 	}
 }
 
-/** PAGE_INDEX_ID value for freed index B-trees */
-static const index_id_t	BTR_FREED_INDEX_ID = 0;
-
-/** Invalidate an index root page so that btr_free_root_check()
-will not find it.
-@param[in,out]	block	index root page
-@param[in,out]	mtr	mini-transaction */
-static
-void
-btr_free_root_invalidate(
-	buf_block_t*	block,
-	mtr_t*		mtr)
-{
-	btr_page_set_index_id(
-		buf_block_get_frame(block),
-		buf_block_get_page_zip(block),
-		BTR_FREED_INDEX_ID, mtr);
-}
-
 /** Prepare to free a B-tree.
 @param[in]	page_id		page id
 @param[in]	page_size	page size
@@ -1247,11 +1229,8 @@ btr_create(
 				 PAGE_HEADER + PAGE_BTR_SEG_LEAF, mtr)) {
 			/* Not enough space for new segment, free root
 			segment before return. */
-			btr_free_root(block, mtr);
-			if (!index->table->is_temporary()) {
-				btr_free_root_invalidate(block, mtr);
-			}
-
+			btr_free_root(block, mtr,
+				      !index->table->is_temporary());
 			return(FIL_NULL);
 		}
 
@@ -1424,8 +1403,7 @@ btr_free_if_exists(
 
 	btr_free_but_not_root(root, mtr->get_log_mode());
 	mtr->set_named_space_id(page_id.space());
-	btr_free_root(root, mtr);
-	btr_free_root_invalidate(root, mtr);
+	btr_free_root(root, mtr, true);
 }
 
 /** Free an index tree in a temporary tablespace or during TRUNCATE TABLE.
@@ -1445,7 +1423,7 @@ btr_free(
 
 	if (block) {
 		btr_free_but_not_root(block, MTR_LOG_NO_REDO);
-		btr_free_root(block, &mtr);
+		btr_free_root(block, &mtr, false);
 	}
 	mtr.commit();
 }
@@ -5423,8 +5401,8 @@ btr_validate_index(
 
 	page_t*	root = btr_root_get(index, &mtr);
 
-	if (root == NULL && !index->is_readable()) {
-		err = DB_DECRYPTION_FAILED;
+	if (!root) {
+		err = DB_CORRUPTION;
 		mtr_commit(&mtr);
 		return err;
 	}
diff --git a/storage/innobase/btr/btr0bulk.cc b/storage/innobase/btr/btr0bulk.cc
index bd79a805a7d..9702ae9b4ad 100644
--- a/storage/innobase/btr/btr0bulk.cc
+++ b/storage/innobase/btr/btr0bulk.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2014, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -662,11 +662,11 @@ PageBulk::latch()
 	/* In case the block is S-latched by page_cleaner. */
 	if (!buf_page_optimistic_get(RW_X_LATCH, m_block, m_modify_clock,
 				     __FILE__, __LINE__, &m_mtr)) {
-		m_block = buf_page_get_gen(
-			page_id_t(m_index->table->space_id, m_page_no),
-			page_size_t(m_index->table->space->flags),
-			RW_X_LATCH, m_block, BUF_GET_IF_IN_POOL,
-			__FILE__, __LINE__, &m_mtr, &m_err);
+		m_block = buf_page_get_gen(page_id_t(m_index->table->space_id,
+						     m_page_no),
+					   univ_page_size, RW_X_LATCH,
+					   m_block, BUF_GET_IF_IN_POOL,
+					   __FILE__, __LINE__, &m_mtr, &m_err);
 
 		if (m_err != DB_SUCCESS) {
 			return (m_err);
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index e54dd7023e6..2d5c0ddc3ca 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -4263,7 +4263,8 @@ buf_page_get_gen(
 		Skip the assertion on space_page_size. */
 		break;
 	case BUF_PEEK_IF_IN_POOL:
-		/* In this mode, the caller may pass a dummy page size,
+	case BUF_GET_IF_IN_POOL:
+		/* The caller may pass a dummy page size,
 		because it does not really matter. */
 		break;
 	default:
@@ -4272,7 +4273,6 @@ buf_page_get_gen(
 		ut_ad(rw_latch == RW_NO_LATCH);
 		/* fall through */
 	case BUF_GET:
-	case BUF_GET_IF_IN_POOL:
 	case BUF_GET_IF_IN_POOL_OR_WATCH:
 	case BUF_GET_POSSIBLY_FREED:
 		bool			found;
@@ -6101,9 +6101,7 @@ database_corrupted:
 				page_not_corrupt: bpage = bpage; );
 
 		if (recv_recovery_is_on()) {
-			/* Pages must be uncompressed for crash recovery. */
-			ut_a(uncompressed);
-			recv_recover_page(TRUE, (buf_block_t*) bpage);
+			recv_recover_page(bpage);
 		}
 
 		/* If space is being truncated then avoid ibuf operation.
@@ -6123,7 +6121,7 @@ database_corrupted:
 					<< " encrypted. However key "
 					"management plugin or used "
 					<< "key_version " << key_version
-					<< "is not found or"
+					<< " is not found or"
 					" used encryption algorithm or method does not match."
 					" Can't continue opening the table.";
 			} else {
diff --git a/storage/innobase/buf/buf0rea.cc b/storage/innobase/buf/buf0rea.cc
index 3edb6c6ee98..18006563f0e 100644
--- a/storage/innobase/buf/buf0rea.cc
+++ b/storage/innobase/buf/buf0rea.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2018, MariaDB Corporation.
+Copyright (c) 2015, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -186,13 +186,13 @@ buf_read_page_low(
 		thd_wait_end(NULL);
 	}
 
-	if (*err != DB_SUCCESS) {
+	if (UNIV_UNLIKELY(*err != DB_SUCCESS)) {
 		if (*err == DB_TABLESPACE_TRUNCATED) {
 			/* Remove the page which is outside the
 			truncated tablespace bounds when recovering
 			from a crash happened during a truncation */
 			buf_read_page_handle_error(bpage);
-			if (recv_recovery_on) {
+			if (recv_recovery_is_on()) {
 				mutex_enter(&recv_sys->mutex);
 				ut_ad(recv_sys->n_addrs > 0);
 				recv_sys->n_addrs--;
diff --git a/storage/innobase/dict/dict0load.cc b/storage/innobase/dict/dict0load.cc
index 283ad22ba55..6cb62f7e256 100644
--- a/storage/innobase/dict/dict0load.cc
+++ b/storage/innobase/dict/dict0load.cc
@@ -1422,7 +1422,7 @@ next:
 		look to see if it is already in the tablespace cache. */
 		if (const fil_space_t* space
 		    = fil_space_for_table_exists_in_mem(
-			    space_id, table_name.m_name, false, flags)) {
+			    space_id, table_name.m_name, flags)) {
 			/* Recovery can open a datafile that does not
 			match SYS_DATAFILES.  If they don't match, update
 			SYS_DATAFILES. */
@@ -2797,11 +2797,16 @@ dict_load_tablespace(
 
 	/* The tablespace may already be open. */
 	table->space = fil_space_for_table_exists_in_mem(
-		table->space_id, table->name.m_name, false, table->flags);
+		table->space_id, table->name.m_name, table->flags);
 	if (table->space) {
 		return;
 	}
 
+	if (ignore_err == DICT_ERR_IGNORE_DROP) {
+		table->file_unreadable = true;
+		return;
+	}
+
 	if (!(ignore_err & DICT_ERR_IGNORE_RECOVER_LOCK)) {
 		ib::error() << "Failed to find tablespace for table "
 			<< table->name << " in the cache. Attempting"
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index cb74ba8d687..4da8a4c4d60 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -1271,9 +1271,7 @@ fil_space_free(
 			rw_lock_x_unlock(&space->latch);
 		}
 
-		bool	need_mutex = !recv_recovery_on;
-
-		if (need_mutex) {
+		if (!recv_recovery_is_on()) {
 			log_mutex_enter();
 		}
 
@@ -1284,7 +1282,7 @@ fil_space_free(
 			UT_LIST_REMOVE(fil_system.named_spaces, space);
 		}
 
-		if (need_mutex) {
+		if (!recv_recovery_is_on()) {
 			log_mutex_exit();
 		}
 
@@ -1343,9 +1341,8 @@ fil_space_create(
 	UT_LIST_INIT(space->chain, &fil_node_t::chain);
 
 	if ((purpose == FIL_TYPE_TABLESPACE || purpose == FIL_TYPE_IMPORT)
-	    && !recv_recovery_on
+	    && !recv_recovery_is_on()
 	    && id > fil_system.max_assigned_id) {
-
 		if (!fil_system.space_id_reuse_warned) {
 			fil_system.space_id_reuse_warned = true;
 
@@ -2905,7 +2902,7 @@ fil_rename_tablespace(
 	ut_ad(strchr(old_file_name, OS_PATH_SEPARATOR) != NULL);
 	ut_ad(strchr(new_file_name, OS_PATH_SEPARATOR) != NULL);
 
-	if (!recv_recovery_on) {
+	if (!recv_recovery_is_on()) {
 		fil_name_write_rename(id, old_file_name, new_file_name);
 		log_mutex_enter();
 	}
@@ -2927,7 +2924,7 @@ fil_rename_tablespace(
 		node->name = new_file_name;
 	}
 
-	if (!recv_recovery_on) {
+	if (!recv_recovery_is_on()) {
 		log_mutex_exit();
 	}
 
@@ -3949,9 +3946,6 @@ memory cache. Note that if we have not done a crash recovery at the database
 startup, there may be many tablespaces which are not yet in the memory cache.
 @param[in]	id		Tablespace ID
 @param[in]	name		Tablespace name used in fil_space_create().
-@param[in]	print_error_if_does_not_exist
-				Print detailed error information to the
-error log if a matching tablespace is not found from memory.
 @param[in]	table_flags	table flags
 @return the tablespace
 @retval	NULL	if no matching tablespace exists in the memory cache */
@@ -3959,7 +3953,6 @@ fil_space_t*
 fil_space_for_table_exists_in_mem(
 	ulint		id,
 	const char*	name,
-	bool		print_error_if_does_not_exist,
 	ulint		table_flags)
 {
 	const ulint	expected_flags = dict_tf_to_fsp_flags(table_flags);
@@ -3977,7 +3970,8 @@ fil_space_for_table_exists_in_mem(
 				<< ", but the tablespace"
 				" with that id has name " << space->name << "."
 				" Have you deleted or moved .ibd files?";
-			goto error_exit;
+			ib::info() << TROUBLESHOOT_DATADICT_MSG;
+			goto func_exit;
 		}
 
 		/* Adjust the flags that are in FSP_FLAGS_MEM_MASK.
@@ -3991,17 +3985,6 @@ fil_space_for_table_exists_in_mem(
 		return space;
 	}
 
-	if (print_error_if_does_not_exist) {
-		ib::error() << "Table " << name
-			    << " in the InnoDB data dictionary"
-			" has tablespace id " << id
-			    << ", but tablespace with that id"
-			" or name does not exist. Have"
-			" you deleted or moved .ibd files?";
-error_exit:
-		ib::info() << TROUBLESHOOT_DATADICT_MSG;
-	}
-
 func_exit:
 	mutex_exit(&fil_system.mutex);
 	return NULL;
diff --git a/storage/innobase/fsp/fsp0fsp.cc b/storage/innobase/fsp/fsp0fsp.cc
index f4a0c9e3461..d300212a56c 100644
--- a/storage/innobase/fsp/fsp0fsp.cc
+++ b/storage/innobase/fsp/fsp0fsp.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -574,13 +574,9 @@ xdes_get_offset(
 	       * FSP_EXTENT_SIZE);
 }
 
-/***********************************************************//**
-Inits a file page whose prior contents should be ignored. */
-static
-void
-fsp_init_file_page_low(
-/*===================*/
-	buf_block_t*	block)	/*!< in: pointer to a page */
+/** Initialize a file page whose prior contents should be ignored.
+@param[in,out]	block	buffer pool block */
+void fsp_apply_init_file_page(buf_block_t* block)
 {
 	page_t*		page	= buf_block_get_frame(block);
 
@@ -638,14 +634,10 @@ fsp_space_modify_check(
 /** Initialize a file page.
 @param[in,out]	block	file page
 @param[in,out]	mtr	mini-transaction */
-static
-void
-fsp_init_file_page(buf_block_t* block, mtr_t* mtr)
+static void fsp_init_file_page(buf_block_t* block, mtr_t* mtr)
 {
-	fsp_init_file_page_low(block);
-
-	mlog_write_initial_log_record(buf_block_get_frame(block),
-				      MLOG_INIT_FILE_PAGE2, mtr);
+	fsp_apply_init_file_page(block);
+	mlog_write_initial_log_record(block->frame, MLOG_INIT_FILE_PAGE2, mtr);
 }
 
 #ifdef UNIV_DEBUG
@@ -661,26 +653,6 @@ fsp_init_file_page(const fil_space_t* space, buf_block_t* block, mtr_t* mtr)
 # define fsp_init_file_page(space, block, mtr) fsp_init_file_page(block, mtr)
 #endif
 
-/***********************************************************//**
-Parses a redo log record of a file page init.
-@return end of log record or NULL */
-byte*
-fsp_parse_init_file_page(
-/*=====================*/
-	byte*		ptr,	/*!< in: buffer */
-	byte*		end_ptr MY_ATTRIBUTE((unused)), /*!< in: buffer end */
-	buf_block_t*	block)	/*!< in: block or NULL */
-{
-	ut_ad(ptr != NULL);
-	ut_ad(end_ptr != NULL);
-
-	if (block) {
-		fsp_init_file_page_low(block);
-	}
-
-	return(ptr);
-}
-
 /**********************************************************************//**
 Writes the space id and flags to a tablespace header.  The flags contain
 row type, physical/compressed page size, and logical/uncompressed page
diff --git a/storage/innobase/include/dict0types.h b/storage/innobase/include/dict0types.h
index e3b536b83ee..80b918b2df0 100644
--- a/storage/innobase/include/dict0types.h
+++ b/storage/innobase/include/dict0types.h
@@ -78,7 +78,10 @@ enum dict_err_ignore_t {
 					Silently load a missing
 					tablespace, and do not load
 					incomplete index definitions. */
-	DICT_ERR_IGNORE_ALL = 0xFFFF	/*!< ignore all errors */
+	/** ignore all errors above */
+	DICT_ERR_IGNORE_ALL = 15,
+	/** prepare to drop the table; do not attempt to load tablespace */
+	DICT_ERR_IGNORE_DROP = 31
 };
 
 /** Quiescing states for flushing tables to disk. */
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 0a5c1a5cfc5..8fedd962290 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2018, MariaDB Corporation.
+Copyright (c) 2013, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -1034,9 +1034,6 @@ memory cache. Note that if we have not done a crash recovery at the database
 startup, there may be many tablespaces which are not yet in the memory cache.
 @param[in]	id		Tablespace ID
 @param[in]	name		Tablespace name used in fil_space_create().
-@param[in]	print_error_if_does_not_exist
-				Print detailed error information to the
-error log if a matching tablespace is not found from memory.
 @param[in]	table_flags	table flags
 @return the tablespace
 @retval	NULL	if no matching tablespace exists in the memory cache */
@@ -1044,7 +1041,6 @@ fil_space_t*
 fil_space_for_table_exists_in_mem(
 	ulint		id,
 	const char*	name,
-	bool		print_error_if_does_not_exist,
 	ulint		table_flags);
 
 /** Try to extend a tablespace if it is smaller than the specified size.
@@ -1250,16 +1246,12 @@ fil_names_write_if_was_clean(
 	return(was_clean);
 }
 
-extern volatile bool	recv_recovery_on;
-
 /** During crash recovery, open a tablespace if it had not been opened
 yet, to get valid size and flags.
 @param[in,out]	space	tablespace */
-inline
-void
-fil_space_open_if_needed(
-	fil_space_t*	space)
+inline void fil_space_open_if_needed(fil_space_t* space)
 {
+	ut_d(extern volatile bool recv_recovery_on);
 	ut_ad(recv_recovery_on);
 
 	if (space->size == 0) {
@@ -1267,10 +1259,7 @@ fil_space_open_if_needed(
 		until the files are opened for the first time.
 		fil_space_get_size() will open the file
 		and adjust the size and flags. */
-#ifdef UNIV_DEBUG
-		ulint		size	=
-#endif /* UNIV_DEBUG */
-			fil_space_get_size(space->id);
+		ut_d(ulint size	=) fil_space_get_size(space->id);
 		ut_ad(size == space->size);
 	}
 }
diff --git a/storage/innobase/include/fsp0fsp.h b/storage/innobase/include/fsp0fsp.h
index 76c7762fac3..cdbcbd2c82e 100644
--- a/storage/innobase/include/fsp0fsp.h
+++ b/storage/innobase/include/fsp0fsp.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2018, MariaDB Corporation.
+Copyright (c) 2013, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -627,15 +627,10 @@ fsp_descr_page(
 	const page_id_t		page_id,
 	const page_size_t&	page_size);
 
-/***********************************************************//**
-Parses a redo log record of a file page init.
-@return end of log record or NULL */
-byte*
-fsp_parse_init_file_page(
-/*=====================*/
-	byte*		ptr,	/*!< in: buffer */
-	byte*		end_ptr, /*!< in: buffer end */
-	buf_block_t*	block);	/*!< in: block or NULL */
+/** Initialize a file page whose prior contents should be ignored.
+@param[in,out]	block	buffer pool block */
+void fsp_apply_init_file_page(buf_block_t* block);
+
 #ifdef UNIV_BTR_PRINT
 /*******************************************************************//**
 Writes info of a segment. */
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 89485b7f31d..7107e35810b 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -40,7 +40,7 @@ Created 9/20/1997 Heikki Tuuri
 extern bool	recv_writer_thread_active;
 
 /** @return whether recovery is currently running. */
-#define recv_recovery_is_on() recv_recovery_on
+#define recv_recovery_is_on() UNIV_UNLIKELY(recv_recovery_on)
 
 /** Find the latest checkpoint in the log header.
 @param[out]	max_field	LOG_CHECKPOINT_1 or LOG_CHECKPOINT_2
@@ -49,12 +49,9 @@ dberr_t
 recv_find_max_checkpoint(ulint* max_field)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
-/** Apply the hashed log records to the page, if the page lsn is less than the
-lsn of a log record.
-@param just_read_in	whether the page recently arrived to the I/O handler
-@param block		the page in the buffer pool */
-void
-recv_recover_page(bool just_read_in, buf_block_t* block);
+/** Apply any buffered redo log to a page that was just read from a data file.
+@param[in,out]	bpage	buffer pool page */
+ATTRIBUTE_COLD void recv_recover_page(buf_page_t* bpage);
 
 /** Start recovering from a redo log checkpoint.
 @see recv_recovery_from_checkpoint_finish
diff --git a/storage/innobase/include/page0zip.h b/storage/innobase/include/page0zip.h
index bb9bb049c22..c9ce4efd3dc 100644
--- a/storage/innobase/include/page0zip.h
+++ b/storage/innobase/include/page0zip.h
@@ -2,7 +2,7 @@
 
 Copyright (c) 2005, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -496,16 +496,14 @@ page_zip_copy_recs(
 	dict_index_t*		index,		/*!< in: index of the B-tree */
 	mtr_t*			mtr);		/*!< in: mini-transaction */
 
-/**********************************************************************//**
-Parses a log record of compressing an index page.
-@return end of log record or NULL */
-byte*
-page_zip_parse_compress(
-/*====================*/
-	byte*		ptr,		/*!< in: buffer */
-	byte*		end_ptr,	/*!< in: buffer end */
-	page_t*		page,		/*!< out: uncompressed page */
-	page_zip_des_t*	page_zip);	/*!< out: compressed page */
+/** 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 */
+byte* page_zip_parse_compress(const byte* ptr, const byte* end_ptr,
+			      buf_block_t* block);
 
 #endif /* !UNIV_INNOCHECKSUM */
 
diff --git a/storage/innobase/include/row0merge.h b/storage/innobase/include/row0merge.h
index 7d49f0ee346..1bc3bc060dc 100644
--- a/storage/innobase/include/row0merge.h
+++ b/storage/innobase/include/row0merge.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2005, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2017, MariaDB Corporation.
+Copyright (c) 2015, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -292,6 +292,12 @@ row_merge_drop_table(
 	dict_table_t*	table)		/*!< in: table instance to drop */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
+/** Write an MLOG_INDEX_LOAD record to indicate in the redo-log
+that redo-logging of individual index pages was disabled, and
+the flushing of such pages to the data files was completed.
+@param[in]	index	an index tree on which redo logging was disabled */
+void row_merge_write_redo(const dict_index_t* index);
+
 /** Build indexes on a table by reading a clustered index, creating a temporary
 file containing index entries, merge sorting these index entries and inserting
 sorted index entries to indexes.
diff --git a/storage/innobase/log/log0log.cc b/storage/innobase/log/log0log.cc
index 9d94f6e83b5..8c17f1383df 100644
--- a/storage/innobase/log/log0log.cc
+++ b/storage/innobase/log/log0log.cc
@@ -2,7 +2,7 @@
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2009, Google Inc.
-Copyright (c) 2014, 2018, MariaDB Corporation.
+Copyright (c) 2014, 2019, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -1193,14 +1193,11 @@ synchronization objects!
 this lsn
 @return false if there was a flush batch of the same type running,
 which means that we could not start this flush batch */
-static
-bool
-log_preflush_pool_modified_pages(
-	lsn_t			new_oldest)
+static bool log_preflush_pool_modified_pages(lsn_t new_oldest)
 {
 	bool	success;
 
-	if (recv_recovery_on) {
+	if (recv_recovery_is_on()) {
 		/* If the recovery is running, we must first apply all
 		log records to their respective file pages to get the
 		right modify lsn values to these pages: otherwise, there
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 39ebb6885d5..60732ed8fef 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -675,7 +675,7 @@ DECLARE_THREAD(recv_writer_thread)(
 
 		mutex_enter(&recv_sys->writer_mutex);
 
-		if (!recv_recovery_on) {
+		if (!recv_recovery_is_on()) {
 			mutex_exit(&recv_sys->writer_mutex);
 			break;
 		}
@@ -770,7 +770,7 @@ recv_sys_debug_free(void)
 
 	/* wake page cleaner up to progress */
 	if (!srv_read_only_mode) {
-		ut_ad(!recv_recovery_on);
+		ut_ad(!recv_recovery_is_on());
 		ut_ad(!recv_writer_thread_active);
 		os_event_reset(buf_flush_event);
 		os_event_set(recv_sys->flush_start);
@@ -1615,7 +1615,7 @@ parse_log:
 		break;
 	case MLOG_INIT_FILE_PAGE2:
 		/* Allow anything in page_type when creating a page. */
-		ptr = fsp_parse_init_file_page(ptr, end_ptr, block);
+		if (block) fsp_apply_init_file_page(block);
 		break;
 	case MLOG_WRITE_STRING:
 		ptr = mlog_parse_string(ptr, end_ptr, page, page_zip);
@@ -1637,7 +1637,7 @@ parse_log:
 		break;
 	case MLOG_ZIP_PAGE_COMPRESS:
 		/* Allow anything in page_type when creating a page. */
-		ptr = page_zip_parse_compress(ptr, end_ptr, page, page_zip);
+		ptr = page_zip_parse_compress(ptr, end_ptr, block);
 		break;
 	case MLOG_ZIP_PAGE_COMPRESS_NO_DATA:
 		if (NULL != (ptr = mlog_parse_index(
@@ -1719,6 +1719,8 @@ recv_get_fil_addr_struct(
 	ulint	space,	/*!< in: space id */
 	ulint	page_no)/*!< in: page number */
 {
+	ut_ad(mutex_own(&recv_sys->mutex));
+
 	recv_addr_t*	recv_addr;
 
 	for (recv_addr = static_cast<recv_addr_t*>(
@@ -1792,10 +1794,6 @@ recv_add_to_hash_table(
 		HASH_INSERT(recv_addr_t, addr_hash, recv_sys->addr_hash,
 			    recv_fold(space, page_no), recv_addr);
 		recv_sys->n_addrs++;
-#if 0
-		fprintf(stderr, "Inserting log rec for space %lu, page %lu\n",
-			space, page_no);
-#endif
 	}
 
 	UT_LIST_ADD_LAST(recv_addr->rec_list, recv);
@@ -1864,48 +1862,20 @@ recv_data_copy_to_buf(
 
 /** Apply the hashed log records to the page, if the page lsn is less than the
 lsn of a log record.
-@param just_read_in	whether the page recently arrived to the I/O handler
-@param block		the page in the buffer pool */
-void
-recv_recover_page(bool just_read_in, buf_block_t* block)
+@param[in,out]	block		buffer pool page
+@param[in,out]	mtr		mini-transaction
+@param[in,out]	recv_addr	recovery address */
+static void recv_recover_page(buf_block_t* block, mtr_t& mtr,
+			      recv_addr_t* recv_addr)
 {
 	page_t*		page;
 	page_zip_des_t*	page_zip;
-	recv_addr_t*	recv_addr;
-	recv_t*		recv;
-	byte*		buf;
-	lsn_t		start_lsn;
-	lsn_t		end_lsn;
-	lsn_t		page_lsn;
-	lsn_t		page_newest_lsn;
-	ibool		modification_to_page;
-	mtr_t		mtr;
-
-	mutex_enter(&(recv_sys->mutex));
-
-	if (recv_sys->apply_log_recs == FALSE) {
-
-		/* Log records should not be applied now */
-
-		mutex_exit(&(recv_sys->mutex));
-
-		return;
-	}
-
-	recv_addr = recv_get_fil_addr_struct(block->page.id.space(),
-					     block->page.id.page_no());
-
-	if ((recv_addr == NULL)
-	    || (recv_addr->state == RECV_BEING_PROCESSED)
-	    || (recv_addr->state == RECV_PROCESSED)) {
-		ut_ad(recv_addr == NULL || recv_needed_recovery);
-
-		mutex_exit(&(recv_sys->mutex));
-
-		return;
-	}
 
+	ut_ad(mutex_own(&recv_sys->mutex));
+	ut_ad(recv_sys->apply_log_recs);
 	ut_ad(recv_needed_recovery);
+	ut_ad(recv_addr->state != RECV_BEING_PROCESSED);
+	ut_ad(recv_addr->state != RECV_PROCESSED);
 
 	if (UNIV_UNLIKELY(srv_print_verbose_log == 2)) {
 		fprintf(stderr, "Applying log to page %u:%u\n",
@@ -1915,95 +1885,53 @@ recv_recover_page(bool just_read_in, buf_block_t* block)
 	DBUG_LOG("ib_log", "Applying log to page " << block->page.id);
 
 	recv_addr->state = RECV_BEING_PROCESSED;
-
-	mutex_exit(&(recv_sys->mutex));
-
-	mtr_start(&mtr);
-	mtr_set_log_mode(&mtr, MTR_LOG_NONE);
+	mutex_exit(&recv_sys->mutex);
 
 	page = block->frame;
 	page_zip = buf_block_get_page_zip(block);
 
-	if (just_read_in) {
-		/* Move the ownership of the x-latch on the page to
-		this OS thread, so that we can acquire a second
-		x-latch on it.  This is needed for the operations to
-		the page to pass the debug checks. */
-
-		rw_lock_x_lock_move_ownership(&block->lock);
+	/* The page may have been modified in the buffer pool.
+	FIL_PAGE_LSN would only be updated right before flushing. */
+	lsn_t page_lsn = buf_page_get_newest_modification(&block->page);
+	if (!page_lsn) {
+		page_lsn = mach_read_from_8(page + FIL_PAGE_LSN);
 	}
 
-	ibool	success = buf_page_get_known_nowait(
-		RW_X_LATCH, block, BUF_KEEP_OLD,
-		__FILE__, __LINE__, &mtr);
-	ut_a(success);
-
-	buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
-
-	/* Read the newest modification lsn from the page */
-	page_lsn = mach_read_from_8(page + FIL_PAGE_LSN);
-
-	/* It may be that the page has been modified in the buffer
-	pool: read the newest modification lsn there */
+	lsn_t start_lsn = 0, end_lsn = 0;
+	fil_space_t* space;
 
-	page_newest_lsn = buf_page_get_newest_modification(&block->page);
-
-	if (page_newest_lsn) {
-
-		page_lsn = page_newest_lsn;
+	if (srv_is_tablespace_truncated(recv_addr->space)) {
+		/* The table will be truncated after applying
+		normal redo log records. */
+		goto skip_log;
 	}
 
-	modification_to_page = FALSE;
-	start_lsn = end_lsn = 0;
-
-	recv = UT_LIST_GET_FIRST(recv_addr->rec_list);
-	fil_space_t* space = fil_space_acquire(block->page.id.space());
+	space = fil_space_acquire(recv_addr->space);
+	if (!space) {
+		goto skip_log;
+	}
 
-	while (recv) {
+	for (recv_t* recv = UT_LIST_GET_FIRST(recv_addr->rec_list);
+	     recv; recv = UT_LIST_GET_NEXT(rec_list, recv)) {
+		ut_ad(recv->start_lsn);
 		end_lsn = recv->end_lsn;
-
 		ut_ad(end_lsn <= log_sys.log.scanned_lsn);
 
-		if (recv->len > RECV_DATA_BLOCK_SIZE) {
-			/* We have to copy the record body to a separate
-			buffer */
-
-			buf = static_cast<byte*>(ut_malloc_nokey(recv->len));
-
-			recv_data_copy_to_buf(buf, recv);
+		if (recv->start_lsn < page_lsn) {
+			/* Ignore this record, because there are later changes
+			for this page. */
+		} else if (srv_was_tablespace_truncated(space)
+			   && recv->start_lsn
+			   < truncate_t::get_truncated_tablespace_init_lsn(
+				   recv_addr->space)) {
+			/* If per-table tablespace was truncated and
+			there exist REDO records before truncate that
+			are to be applied as part of recovery
+			(checkpoint didn't happen since truncate was
+			done) skip such records using lsn check as
+			they may not stand valid post truncate. */
 		} else {
-			buf = ((byte*)(recv->data)) + sizeof(recv_data_t);
-		}
-
-		/* If per-table tablespace was truncated and there exist REDO
-		records before truncate that are to be applied as part of
-		recovery (checkpoint didn't happen since truncate was done)
-		skip such records using lsn check as they may not stand valid
-		post truncate.
-		LSN at start of truncate is recorded and any redo record
-		with LSN less than recorded LSN is skipped.
-		Note: We can't skip complete recv_addr as same page may have
-		valid REDO records post truncate those needs to be applied. */
-
-		/* Ignore applying the redo logs for tablespace that is
-		truncated. Post recovery there is fixup action that will
-		restore the tablespace back to normal state.
-		Applying redo at this stage can result in error given that
-		redo will have action recorded on page before tablespace
-		was re-inited and that would lead to an error while applying
-		such action. */
-		if (recv->start_lsn >= page_lsn
-		    && !srv_is_tablespace_truncated(space->id)
-		    && !(srv_was_tablespace_truncated(space)
-			 && recv->start_lsn
-			 < truncate_t::get_truncated_tablespace_init_lsn(
-				 space->id))) {
-
-			lsn_t	end_lsn;
-
-			if (!modification_to_page) {
-
-				modification_to_page = TRUE;
+			if (!start_lsn) {
 				start_lsn = recv->start_lsn;
 			}
 
@@ -2019,11 +1947,23 @@ recv_recover_page(bool just_read_in, buf_block_t* block)
 				 << " len " << recv->len
 				 << " page " << block->page.id);
 
+			byte* buf;
+
+			if (recv->len > RECV_DATA_BLOCK_SIZE) {
+				/* We have to copy the record body to
+				a separate buffer */
+				buf = static_cast<byte*>
+					(ut_malloc_nokey(recv->len));
+				recv_data_copy_to_buf(buf, recv);
+			} else {
+				buf = reinterpret_cast<byte*>(recv->data)
+					+ sizeof *recv->data;
+			}
+
 			recv_parse_or_apply_log_rec_body(
 				recv->type, buf, buf + recv->len,
 				block->page.id.space(),
-				block->page.id.page_no(),
-				true, block, &mtr);
+				block->page.id.page_no(), true, block, &mtr);
 
 			end_lsn = recv->start_lsn + recv->len;
 			mach_write_to_8(FIL_PAGE_LSN + page, end_lsn);
@@ -2032,32 +1972,26 @@ recv_recover_page(bool just_read_in, buf_block_t* block)
 					+ page, end_lsn);
 
 			if (page_zip) {
-				mach_write_to_8(FIL_PAGE_LSN
-						+ page_zip->data, end_lsn);
+				mach_write_to_8(FIL_PAGE_LSN + page_zip->data,
+						end_lsn);
 			}
-		}
 
-		if (recv->len > RECV_DATA_BLOCK_SIZE) {
-			ut_free(buf);
+			if (recv->len > RECV_DATA_BLOCK_SIZE) {
+				ut_free(buf);
+			}
 		}
-
-		recv = UT_LIST_GET_NEXT(rec_list, recv);
 	}
 
 	space->release();
 
+skip_log:
 #ifdef UNIV_ZIP_DEBUG
-	if (fil_page_index_page_check(page)) {
-		page_zip_des_t*	page_zip = buf_block_get_page_zip(block);
-
-		ut_a(!page_zip
-		     || page_zip_validate_low(page_zip, page, NULL, FALSE));
-	}
+	ut_ad(!fil_page_index_page_check(page)
+	      || !page_zip
+	      || page_zip_validate_low(page_zip, page, NULL, FALSE));
 #endif /* UNIV_ZIP_DEBUG */
 
-	if (modification_to_page) {
-		ut_a(block);
-
+	if (start_lsn) {
 		log_flush_order_mutex_enter();
 		buf_flush_recv_note_modification(block, start_lsn, end_lsn);
 		log_flush_order_mutex_exit();
@@ -2067,8 +2001,7 @@ recv_recover_page(bool just_read_in, buf_block_t* block)
 	lsn values of page */
 
 	mtr.discard_modifications();
-
-	mtr_commit(&mtr);
+	mtr.commit();
 
 	ib_time_t time = ut_time();
 
@@ -2078,6 +2011,7 @@ recv_recover_page(bool just_read_in, buf_block_t* block)
 		recv_max_page_lsn = page_lsn;
 	}
 
+	ut_ad(recv_addr->state == RECV_BEING_PROCESSED);
 	recv_addr->state = RECV_PROCESSED;
 
 	ut_a(recv_sys->n_addrs > 0);
@@ -2088,52 +2022,76 @@ recv_recover_page(bool just_read_in, buf_block_t* block)
 				INNODB_EXTEND_TIMEOUT_INTERVAL, "To recover: " ULINTPF " pages from log", n);
 		}
 	}
+}
 
+/** Apply any buffered redo log to a page that was just read from a data file.
+@param[in,out]	bpage	buffer pool page */
+void recv_recover_page(buf_page_t* bpage)
+{
+	mtr_t mtr;
+	mtr.start();
+	mtr.set_log_mode(MTR_LOG_NONE);
+
+	ut_ad(buf_page_get_state(bpage) == BUF_BLOCK_FILE_PAGE);
+	buf_block_t* block = reinterpret_cast<buf_block_t*>(bpage);
+
+	/* Move the ownership of the x-latch on the page to
+	this OS thread, so that we can acquire a second
+	x-latch on it.  This is needed for the operations to
+	the page to pass the debug checks. */
+	rw_lock_x_lock_move_ownership(&block->lock);
+	buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
+	ibool	success = buf_page_get_known_nowait(
+		RW_X_LATCH, block, BUF_KEEP_OLD,
+		__FILE__, __LINE__, &mtr);
+	ut_a(success);
+
+	mutex_enter(&recv_sys->mutex);
+	if (!recv_sys->apply_log_recs) {
+	} else if (recv_addr_t* recv_addr = recv_get_fil_addr_struct(
+			   bpage->id.space(), bpage->id.page_no())) {
+		switch (recv_addr->state) {
+		case RECV_BEING_PROCESSED:
+		case RECV_PROCESSED:
+			break;
+		default:
+			recv_recover_page(block, mtr, recv_addr);
+			goto func_exit;
+		}
+	}
+
+	mtr.commit();
+func_exit:
 	mutex_exit(&recv_sys->mutex);
+	ut_ad(mtr.has_committed());
 }
 
 /** Reads in pages which have hashed log records, from an area around a given
 page number.
-@param[in]	page_id	page id
-@return number of pages found */
-static ulint recv_read_in_area(const page_id_t page_id)
+@param[in]	page_id	page id */
+static void recv_read_in_area(const page_id_t page_id)
 {
-	recv_addr_t* recv_addr;
 	ulint	page_nos[RECV_READ_AHEAD_AREA];
-	ulint	low_limit;
-	ulint	n;
-
-	low_limit = page_id.page_no()
+	ulint	page_no = page_id.page_no()
 		- (page_id.page_no() % RECV_READ_AHEAD_AREA);
+	ulint*	p = page_nos;
 
-	n = 0;
-
-	for (ulint page_no = low_limit;
-	     page_no < low_limit + RECV_READ_AHEAD_AREA;
-	     page_no++) {
-
-		recv_addr = recv_get_fil_addr_struct(page_id.space(), page_no);
-
-		const page_id_t	cur_page_id(page_id.space(), page_no);
-
-		if (recv_addr && !buf_page_peek(cur_page_id)) {
-
-			mutex_enter(&(recv_sys->mutex));
-
-			if (recv_addr->state == RECV_NOT_PROCESSED) {
-				recv_addr->state = RECV_BEING_READ;
-
-				page_nos[n] = page_no;
-
-				n++;
-			}
-
-			mutex_exit(&(recv_sys->mutex));
+	for (const ulint up_limit = page_no + RECV_READ_AHEAD_AREA;
+	     page_no < up_limit; page_no++) {
+		recv_addr_t* recv_addr = recv_get_fil_addr_struct(
+			page_id.space(), page_no);
+		if (recv_addr
+		    && recv_addr->state == RECV_NOT_PROCESSED
+		    && !buf_page_peek(page_id_t(page_id.space(), page_no))) {
+			recv_addr->state = RECV_BEING_READ;
+			*p++ = page_no;
 		}
 	}
 
-	buf_read_recv_pages(FALSE, page_id.space(), page_nos, n);
-	return(n);
+	mutex_exit(&recv_sys->mutex);
+	buf_read_recv_pages(FALSE, page_id.space(), page_nos,
+			    ulint(p - page_nos));
+	mutex_enter(&recv_sys->mutex);
 }
 
 /** Apply the hash table of stored log records to persistent data pages.
@@ -2186,59 +2144,55 @@ void recv_apply_hashed_log_recs(bool last_batch)
 		}
 	}
 
+	mtr_t mtr;
+
 	for (ulint i = 0; i < hash_get_n_cells(recv_sys->addr_hash); i++) {
 		for (recv_addr_t* recv_addr = static_cast<recv_addr_t*>(
 			     HASH_GET_FIRST(recv_sys->addr_hash, i));
 		     recv_addr;
 		     recv_addr = static_cast<recv_addr_t*>(
 				HASH_GET_NEXT(addr_hash, recv_addr))) {
-
-			if (srv_is_tablespace_truncated(recv_addr->space)) {
-				/* Avoid applying REDO log for the tablespace
-				that is schedule for TRUNCATE. */
+			if (!UT_LIST_GET_LEN(recv_addr->rec_list)) {
+ignore:
 				ut_a(recv_sys->n_addrs);
-				recv_addr->state = RECV_DISCARDED;
 				recv_sys->n_addrs--;
 				continue;
 			}
 
-			if (recv_addr->state == RECV_DISCARDED
-			    || !UT_LIST_GET_LEN(recv_addr->rec_list)) {
-				ut_a(recv_sys->n_addrs);
-				recv_sys->n_addrs--;
+			switch (recv_addr->state) {
+			case RECV_BEING_READ:
+			case RECV_BEING_PROCESSED:
+			case RECV_PROCESSED:
 				continue;
+			case RECV_DISCARDED:
+				goto ignore;
+			case RECV_NOT_PROCESSED:
+				break;
 			}
 
-			const page_id_t		page_id(recv_addr->space,
-							recv_addr->page_no);
-			bool			found;
-			const page_size_t&	page_size
-				= fil_space_get_page_size(recv_addr->space,
-							  &found);
-
-			ut_ad(found);
-
-			if (recv_addr->state == RECV_NOT_PROCESSED) {
-				mutex_exit(&recv_sys->mutex);
-
-				if (buf_page_peek(page_id)) {
-					mtr_t	mtr;
-					mtr.start();
-
-					buf_block_t* block = buf_page_get(
-						page_id, page_size,
-						RW_X_LATCH, &mtr);
-
-					buf_block_dbg_add_level(
-						block, SYNC_NO_ORDER_CHECK);
-
-					recv_recover_page(FALSE, block);
-					mtr.commit();
-				} else {
-					recv_read_in_area(page_id);
-				}
+			if (srv_is_tablespace_truncated(recv_addr->space)) {
+				/* Avoid applying REDO log for the tablespace
+				that is schedule for TRUNCATE. */
+				recv_addr->state = RECV_DISCARDED;
+				goto ignore;
+			}
 
-				mutex_enter(&recv_sys->mutex);
+			const page_id_t page_id(recv_addr->space,
+						recv_addr->page_no);
+
+			mtr.start();
+			mtr.set_log_mode(MTR_LOG_NONE);
+			if (buf_block_t* block = buf_page_get_gen(
+				    page_id, univ_page_size, RW_X_LATCH,
+				    NULL, BUF_GET_IF_IN_POOL,
+				    __FILE__, __LINE__, &mtr, NULL)) {
+				buf_block_dbg_add_level(
+					block, SYNC_NO_ORDER_CHECK);
+				recv_recover_page(block, mtr, recv_addr);
+				ut_ad(mtr.has_committed());
+			} else {
+				mtr.commit();
+				recv_read_in_area(page_id);
 			}
 		}
 	}
@@ -2485,6 +2439,15 @@ recv_report_corrupt_log(
 	return(true);
 }
 
+/** Report a MLOG_INDEX_LOAD operation.
+@param[in]	space_id	tablespace identifier */
+ATTRIBUTE_COLD static void recv_mlog_index_load(ulint space_id)
+{
+	if (log_optimized_ddl_op) {
+		log_optimized_ddl_op(space_id);
+	}
+}
+
 /** 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
@@ -2506,6 +2469,7 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 	byte*		body;
 
 	ut_ad(log_mutex_own());
+	ut_ad(mutex_own(&recv_sys->mutex));
 	ut_ad(recv_sys->parse_start_lsn != 0);
 loop:
 	ptr = recv_sys->buf + recv_sys->recovered_offset;
@@ -2640,9 +2604,7 @@ loop:
 			/* fall through */
 		case MLOG_INDEX_LOAD:
 			if (type == MLOG_INDEX_LOAD) {
-				if (log_optimized_ddl_op) {
-					log_optimized_ddl_op(space);
-				}
+				recv_mlog_index_load(space);
 			}
 			/* fall through */
 		case MLOG_FILE_NAME:
@@ -2796,10 +2758,7 @@ corrupted_log:
 				break;
 #endif /* UNIV_LOG_LSN_DEBUG */
 			case MLOG_INDEX_LOAD:
-				/* Mariabackup FIXME: Report an error
-				when encountering MLOG_INDEX_LOAD on
-				--prepare or already on --backup. */
-				ut_a(srv_operation == SRV_OPERATION_NORMAL);
+				recv_mlog_index_load(space);
 				break;
 			case MLOG_FILE_NAME:
 			case MLOG_FILE_DELETE:
@@ -3077,6 +3036,8 @@ recv_scan_log_recs(
 
 	*group_scanned_lsn = scanned_lsn;
 
+	mutex_enter(&recv_sys->mutex);
+
 	if (more_data && !recv_sys->found_corrupt_log) {
 		/* Try to parse more log records */
 
@@ -3086,7 +3047,8 @@ recv_scan_log_recs(
 			      || recv_sys->found_corrupt_fs
 			      || recv_sys->mlog_checkpoint_lsn
 			      == recv_sys->recovered_lsn);
-			return(true);
+			finished = true;
+			goto func_exit;
 		}
 
 		if (*store_to_hash != STORE_NO
@@ -3107,6 +3069,8 @@ recv_scan_log_recs(
 		}
 	}
 
+func_exit:
+	mutex_exit(&recv_sys->mutex);
 	return(finished);
 }
 
diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index 8b07b2b953d..ac0ff39f4f9 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -4917,23 +4917,20 @@ page_zip_copy_recs(
 	page_zip_compress_write_log(page_zip, page, index, mtr);
 }
 
-/**********************************************************************//**
-Parses a log record of compressing an index page.
-@return end of log record or NULL */
-byte*
-page_zip_parse_compress(
-/*====================*/
-	byte*		ptr,	/*!< in: buffer */
-	byte*		end_ptr,/*!< in: buffer end */
-	page_t*		page,	/*!< out: uncompressed page */
-	page_zip_des_t*	page_zip)/*!< out: compressed page */
+/** 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 */
+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);
-	ut_ad(!page == !page_zip);
 
 	if (UNIV_UNLIKELY(ptr + (2 + 2) > end_ptr)) {
 
@@ -4950,14 +4947,22 @@ page_zip_parse_compress(
 		return(NULL);
 	}
 
-	if (page) {
-		if (!page_zip || page_zip_get_size(page_zip) < size) {
+	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);
@@ -4967,14 +4972,14 @@ corrupt:
 		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, page,
+		if (UNIV_UNLIKELY(!page_zip_decompress(page_zip, block->frame,
 						       TRUE))) {
 
 			goto corrupt;
 		}
 	}
 
-	return(ptr + 8 + size + trailer_size);
+	return(const_cast<byte*>(ptr) + 8 + size + trailer_size);
 }
 #endif /* !UNIV_INNOCHECKSUM */
 
diff --git a/storage/innobase/row/row0ftsort.cc b/storage/innobase/row/row0ftsort.cc
index b9f01f4547e..8afee045493 100644
--- a/storage/innobase/row/row0ftsort.cc
+++ b/storage/innobase/row/row0ftsort.cc
@@ -1806,5 +1806,9 @@ exit:
 		ib::info() << "InnoDB_FTS: inserted " << count << " records";
 	}
 
+	if (psort_info[0].psort_common->trx->get_flush_observer()) {
+		row_merge_write_redo(aux_index);
+	}
+
 	return(error);
 }
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 101287c204e..1f4ffdf1e8c 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -4519,17 +4519,14 @@ row_merge_drop_table(
 that redo-logging of individual index pages was disabled, and
 the flushing of such pages to the data files was completed.
 @param[in]	index	an index tree on which redo logging was disabled */
-static
-void
-row_merge_write_redo(
-	const dict_index_t*	index)
+void row_merge_write_redo(const dict_index_t* index)
 {
-	mtr_t	mtr;
-	byte*	log_ptr;
-
 	ut_ad(!index->table->is_temporary());
+	ut_ad(!(index->type & (DICT_SPATIAL | DICT_FTS)));
+
+	mtr_t mtr;
 	mtr.start();
-	log_ptr = mlog_open(&mtr, 11 + 8);
+	byte* log_ptr = mlog_open(&mtr, 11 + 8);
 	log_ptr = mlog_write_initial_log_record_low(
 		MLOG_INDEX_LOAD,
 		index->table->space_id, index->page, log_ptr, &mtr);
@@ -5070,7 +5067,10 @@ func_exit:
 				     = dict_table_get_first_index(new_table);
 			     index != NULL;
 			     index = dict_table_get_next_index(index)) {
-				row_merge_write_redo(index);
+				if (!(index->type
+				      & (DICT_FTS | DICT_SPATIAL))) {
+					row_merge_write_redo(index);
+				}
 			}
 		}
 	}
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 0400cce24ef..237e3f7d749 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -2771,7 +2771,7 @@ row_mysql_drop_garbage_tables()
 			btr_pcur_commit_specify_mtr(&pcur, &mtr);
 
 			if (dict_load_table(table_name, true,
-					    DICT_ERR_IGNORE_ALL)) {
+					    DICT_ERR_IGNORE_DROP)) {
 				row_drop_table_for_mysql(table_name, trx,
 							 SQLCOM_DROP_TABLE);
 				trx_commit_for_mysql(trx);