Merge 10.3 into 10.4

22 files changed, 402 insertions, 399 deletions

diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 94fef62099d..d261465066e 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -75,6 +75,8 @@ Created 11/5/1995 Heikki Tuuri
 #include "lzo/lzo1x.h"
 #endif
 
+using st_::span;
+
 #ifdef HAVE_LIBNUMA
 #include <numa.h>
 #include <numaif.h>
@@ -457,7 +459,7 @@ buf_pool_register_chunk(
 @return true if temporary tablespace decrypted, false if not */
 static bool buf_tmp_page_decrypt(byte* tmp_frame, byte* src_frame)
 {
-	if (buf_page_is_zeroes(src_frame, srv_page_size)) {
+	if (buf_is_zeroes(span<const byte>(src_frame, srv_page_size))) {
 		return true;
 	}
 
@@ -964,20 +966,14 @@ static void buf_page_check_lsn(bool check_lsn, const byte* read_buf)
 #endif /* !UNIV_INNOCHECKSUM */
 }
 
-/** Check if a page is all zeroes.
-@param[in]	read_buf	database page
-@param[in]	page_size	page frame size
-@return whether the page is all zeroes */
-bool buf_page_is_zeroes(const void* read_buf, size_t page_size)
+
+/** Check if a buffer is all zeroes.
+@param[in]	buf	data to check
+@return whether the buffer is all zeroes */
+bool buf_is_zeroes(span<const byte> buf)
 {
-	const ulint* b = reinterpret_cast<const ulint*>(read_buf);
-	const ulint* const e = b + page_size / sizeof *b;
-	do {
-		if (*b++) {
-			return false;
-		}
-	} while (b != e);
-	return true;
+  ut_ad(buf.size() <= sizeof field_ref_zero);
+  return memcmp(buf.data(), field_ref_zero, buf.size()) == 0;
 }
 
 /** Check if a page is corrupt.
@@ -1006,7 +1002,7 @@ buf_page_is_corrupted(
 		uint crc32 = mach_read_from_4(end);
 
 		if (!crc32 && size == srv_page_size
-		    && buf_page_is_zeroes(read_buf, size)) {
+		    && buf_is_zeroes(span<const byte>(read_buf, size))) {
 			return false;
 		}
 
@@ -1095,8 +1091,9 @@ buf_page_is_corrupted(
 	compile_time_assert(!(FIL_PAGE_LSN % 8));
 
 	/* A page filled with NUL bytes is considered not corrupted.
-	The FIL_PAGE_FILE_FLUSH_LSN field may be written nonzero for
-	the first page of each file of the system tablespace.
+	Before MariaDB Server 10.1.25 (MDEV-12113) or 10.2.2 (or MySQL 5.7),
+	the FIL_PAGE_FILE_FLUSH_LSN field may have been written nonzero
+	for the first page of each file of the system tablespace.
 	We want to ignore it for the system tablespace, but because
 	we do not know the expected tablespace here, we ignore the
 	field for all data files, except for
@@ -4209,7 +4206,45 @@ buf_wait_for_read(
 	}
 }
 
-/** This is the general function used to get access to a database page.
+/** Lock the page with the given latch type.
+@param[in,out]	block		block to be locked
+@param[in]	rw_latch	RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH
+@param[in]	mtr		mini-transaction
+@param[in]	file		file name
+@param[in]	line		line where called
+@return pointer to locked block */
+static buf_block_t* buf_page_mtr_lock(buf_block_t *block,
+                                      ulint rw_latch,
+                                      mtr_t* mtr,
+                                      const char *file,
+                                      unsigned line)
+{
+  mtr_memo_type_t fix_type;
+  switch (rw_latch)
+  {
+  case RW_NO_LATCH:
+    fix_type= MTR_MEMO_BUF_FIX;
+    break;
+  case RW_S_LATCH:
+    rw_lock_s_lock_inline(&block->lock, 0, file, line);
+    fix_type= MTR_MEMO_PAGE_S_FIX;
+    break;
+  case RW_SX_LATCH:
+    rw_lock_sx_lock_inline(&block->lock, 0, file, line);
+    fix_type= MTR_MEMO_PAGE_SX_FIX;
+    break;
+  default:
+    ut_ad(rw_latch == RW_X_LATCH);
+    rw_lock_x_lock_inline(&block->lock, 0, file, line);
+    fix_type= MTR_MEMO_PAGE_X_FIX;
+    break;
+  }
+
+  mtr_memo_push(mtr, block, fix_type);
+  return block;
+}
+
+/** This is the low level function used to get access to a database page.
 @param[in]	page_id		page id
 @param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
 @param[in]	rw_latch	RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH
@@ -4222,7 +4257,7 @@ BUF_PEEK_IF_IN_POOL, BUF_GET_NO_LATCH, or BUF_GET_IF_IN_POOL_OR_WATCH
 @param[out]	err		DB_SUCCESS or error code
 @return pointer to the block or NULL */
 buf_block_t*
-buf_page_get_gen(
+buf_page_get_low(
 	const page_id_t		page_id,
 	ulint			zip_size,
 	ulint			rw_latch,
@@ -4868,35 +4903,7 @@ evict_from_pool:
 		return NULL;
 	}
 
-	mtr_memo_type_t	fix_type;
-
-	switch (rw_latch) {
-	case RW_NO_LATCH:
-
-		fix_type = MTR_MEMO_BUF_FIX;
-		break;
-
-	case RW_S_LATCH:
-		rw_lock_s_lock_inline(&fix_block->lock, 0, file, line);
-
-		fix_type = MTR_MEMO_PAGE_S_FIX;
-		break;
-
-	case RW_SX_LATCH:
-		rw_lock_sx_lock_inline(&fix_block->lock, 0, file, line);
-
-		fix_type = MTR_MEMO_PAGE_SX_FIX;
-		break;
-
-	default:
-		ut_ad(rw_latch == RW_X_LATCH);
-		rw_lock_x_lock_inline(&fix_block->lock, 0, file, line);
-
-		fix_type = MTR_MEMO_PAGE_X_FIX;
-		break;
-	}
-
-	mtr_memo_push(mtr, fix_block, fix_type);
+	fix_block = buf_page_mtr_lock(fix_block, rw_latch, mtr, file, line);
 
 	if (mode != BUF_PEEK_IF_IN_POOL && !access_time) {
 		/* In the case of a first access, try to apply linear
@@ -4911,6 +4918,43 @@ evict_from_pool:
 	return(fix_block);
 }
 
+/** This is the general function used to get access to a database page.
+It does page initialization and applies the buffered redo logs.
+@param[in]	page_id		page id
+@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@param[in]	rw_latch	RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH
+@param[in]	guess		guessed block or NULL
+@param[in]	mode		BUF_GET, BUF_GET_IF_IN_POOL,
+BUF_PEEK_IF_IN_POOL, BUF_GET_NO_LATCH, or BUF_GET_IF_IN_POOL_OR_WATCH
+@param[in]	file		file name
+@param[in]	line		line where called
+@param[in]	mtr		mini-transaction
+@param[out]	err		DB_SUCCESS or error code
+@return pointer to the block or NULL */
+buf_block_t*
+buf_page_get_gen(
+	const page_id_t		page_id,
+	ulint			zip_size,
+	ulint			rw_latch,
+	buf_block_t*		guess,
+	ulint			mode,
+	const char*		file,
+	unsigned		line,
+	mtr_t*			mtr,
+	dberr_t*		err)
+{
+  if (buf_block_t *block= recv_recovery_create_page(page_id))
+  {
+    block->fix();
+    ut_ad(rw_lock_s_lock_nowait(block->debug_latch, file, line));
+    block= buf_page_mtr_lock(block, rw_latch, mtr, file, line);
+    return block;
+  }
+
+  return buf_page_get_low(page_id, zip_size, rw_latch,
+                          guess, mode, file, line, mtr, err);
+}
+
 /********************************************************************//**
 This is the general function used to get optimistic access to a database
 page.
@@ -5927,7 +5971,8 @@ static dberr_t buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 	/* If traditional checksums match, we assume that page is
 	not anymore encrypted. */
 	if (space->full_crc32()
-	    && !buf_page_is_zeroes(dst_frame, space->physical_size())
+	    && !buf_is_zeroes(span<const byte>(dst_frame,
+					       space->physical_size()))
 	    && (key_version || space->is_compressed()
 		|| space->purpose == FIL_TYPE_TEMPORARY)) {
 		if (buf_page_full_crc32_is_corrupted(
diff --git a/storage/innobase/buf/buf0dblwr.cc b/storage/innobase/buf/buf0dblwr.cc
index 00d6c1f9ded..f9b5b30c364 100644
--- a/storage/innobase/buf/buf0dblwr.cc
+++ b/storage/innobase/buf/buf0dblwr.cc
@@ -34,6 +34,8 @@ Created 2011/12/19
 #include "fil0crypt.h"
 #include "fil0pagecompress.h"
 
+using st_::span;
+
 /** The doublewrite buffer */
 buf_dblwr_t*	buf_dblwr = NULL;
 
@@ -577,7 +579,7 @@ buf_dblwr_process()
 
 		const ulint physical_size = space->physical_size();
 		const ulint zip_size = space->zip_size();
-		ut_ad(!buf_page_is_zeroes(page, physical_size));
+		ut_ad(!buf_is_zeroes(span<const byte>(page, physical_size)));
 
 		/* We want to ensure that for partial reads the
 		unread portion of the page is NUL. */
@@ -600,8 +602,8 @@ buf_dblwr_process()
 				<< "error: " << ut_strerr(err);
 		}
 
-		const bool is_all_zero = buf_page_is_zeroes(
-			read_buf, physical_size);
+		const bool is_all_zero = buf_is_zeroes(
+			span<const byte>(read_buf, physical_size));
 		const bool expect_encrypted = space->crypt_data
 			&& space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED;
 		bool is_corrupted = false;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 7c06171d413..c8c9b92f727 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -198,9 +198,6 @@ static char* innodb_large_prefix;
 stopword table to be used */
 static char*	innobase_server_stopword_table;
 
-/* Below we have boolean-valued start-up parameters, and their default
-values */
-
 static my_bool	innobase_use_atomic_writes;
 static my_bool	innobase_use_checksums;
 static my_bool	innobase_locks_unsafe_for_binlog;
@@ -224,6 +221,9 @@ my_bool innodb_evict_tables_on_commit_debug;
 extern my_bool srv_scrub_force_testing;
 #endif
 
+/** File format constraint for ALTER TABLE */
+ulong innodb_instant_alter_column_allowed;
+
 /** Note we cannot use rec_format_enum because we do not allow
 COMPRESSED row format for innodb_default_row_format option. */
 enum default_row_format_enum {
@@ -457,6 +457,22 @@ static TYPELIB innodb_change_buffering_typelib = {
 	NULL
 };
 
+/** Allowed values of innodb_instant_alter_column_allowed */
+const char* innodb_instant_alter_column_allowed_names[] = {
+	"never", /* compatible with MariaDB 5.5 to 10.2 */
+	"add_last",/* allow instant ADD COLUMN ... LAST */
+	"add_drop_reorder", /* allow instant ADD anywhere & DROP & reorder */
+	NullS
+};
+
+/** Enumeration of innodb_instant_alter_column_allowed */
+static TYPELIB innodb_instant_alter_column_allowed_typelib = {
+	array_elements(innodb_instant_alter_column_allowed_names) - 1,
+	"innodb_instant_alter_column_allowed_typelib",
+	innodb_instant_alter_column_allowed_names,
+	NULL
+};
+
 /** Retrieve the FTS Relevance Ranking result for doc with doc_id
 of m_prebuilt->fts_doc_id
 @param[in,out]	fts_hdl	FTS handler
@@ -18899,6 +18915,12 @@ static MYSQL_SYSVAR_BOOL(stats_include_delete_marked,
   "Include delete marked records when calculating persistent statistics",
   NULL, NULL, FALSE);
 
+static MYSQL_SYSVAR_ENUM(instant_alter_column_allowed,
+			 innodb_instant_alter_column_allowed,
+  PLUGIN_VAR_RQCMDARG,
+  "File format constraint for ALTER TABLE", NULL, NULL, 2/*add_drop_reorder*/,
+  &innodb_instant_alter_column_allowed_typelib);
+
 static MYSQL_SYSVAR_ULONG(io_capacity, srv_io_capacity,
   PLUGIN_VAR_RQCMDARG,
   "Number of IOPs the server can do. Tunes the background IO rate",
@@ -19015,6 +19037,7 @@ static MYSQL_SYSVAR_ENUM(flush_method, innodb_flush_method,
 static MYSQL_SYSVAR_STR(file_format, innodb_file_format,
   PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
   "Deprecated parameter with no effect.", NULL, NULL, NULL);
+
 static MYSQL_SYSVAR_STR(large_prefix, innodb_large_prefix,
   PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
   "Deprecated parameter with no effect.", NULL, NULL, NULL);
@@ -20181,6 +20204,7 @@ static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(random_read_ahead),
   MYSQL_SYSVAR(read_ahead_threshold),
   MYSQL_SYSVAR(read_only),
+  MYSQL_SYSVAR(instant_alter_column_allowed),
   MYSQL_SYSVAR(io_capacity),
   MYSQL_SYSVAR(io_capacity_max),
   MYSQL_SYSVAR(page_cleaners),
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 4200e87fa7f..9b5b4d6e566 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -59,6 +59,8 @@ Smart ALTER TABLE
 #include "span.h"
 
 using st_::span;
+/** File format constraint for ALTER TABLE */
+extern ulong innodb_instant_alter_column_allowed;
 
 static const char *MSG_UNSUPPORTED_ALTER_ONLINE_ON_VIRTUAL_COLUMN=
 			"INPLACE ADD or DROP of virtual columns cannot be "
@@ -1998,6 +2000,37 @@ ha_innobase::check_if_supported_inplace_alter(
 
 	const char* reason_rebuild = NULL;
 
+	switch (innodb_instant_alter_column_allowed) {
+	case 0: /* never */
+		if ((ha_alter_info->handler_flags
+		     & (ALTER_ADD_STORED_BASE_COLUMN
+			| ALTER_STORED_COLUMN_ORDER
+			| ALTER_DROP_STORED_COLUMN))
+		    || m_prebuilt->table->is_instant()) {
+			reason_rebuild =
+				"innodb_instant_alter_column_allowed=never";
+innodb_instant_alter_column_allowed_reason:
+			if (ha_alter_info->handler_flags
+			    & ALTER_RECREATE_TABLE) {
+				reason_rebuild = NULL;
+			} else {
+				ha_alter_info->handler_flags
+					|= ALTER_RECREATE_TABLE;
+				ha_alter_info->unsupported_reason
+					= reason_rebuild;
+			}
+		}
+		break;
+	case 1: /* add_last */
+		if ((ha_alter_info->handler_flags
+		     & (ALTER_STORED_COLUMN_ORDER | ALTER_DROP_STORED_COLUMN))
+		    || m_prebuilt->table->instant) {
+			reason_rebuild = "innodb_instant_atler_column_allowed="
+				"add_last";
+			goto innodb_instant_alter_column_allowed_reason;
+		}
+	}
+
 	switch (ha_alter_info->handler_flags & ~INNOBASE_INPLACE_IGNORE) {
 	case ALTER_OPTIONS:
 		if (alter_options_need_rebuild(ha_alter_info, table)) {
@@ -2488,6 +2521,7 @@ cannot_create_many_fulltext_index:
 	}
 
 	if (need_rebuild || fts_need_rebuild) {
+		ha_alter_info->handler_flags |= ALTER_RECREATE_TABLE;
 		DBUG_RETURN(online
 			    ? HA_ALTER_INPLACE_COPY_NO_LOCK
 			    : HA_ALTER_INPLACE_COPY_LOCK);
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index 642ba94d9dc..a0fe5306e65 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, 2019, MariaDB Corporation.
+Copyright (c) 2016, 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
@@ -28,6 +28,8 @@ Created 7/19/1997 Heikki Tuuri
 #include "sync0sync.h"
 #include "btr0sea.h"
 
+using st_::span;
+
 #if defined UNIV_DEBUG || defined UNIV_IBUF_DEBUG
 my_bool	srv_ibuf_disable_background_merge;
 #endif /* UNIV_DEBUG || UNIV_IBUF_DEBUG */
@@ -4896,7 +4898,8 @@ dberr_t ibuf_check_bitmap_on_import(const trx_t* trx, fil_space_t* space)
 		bitmap_page = ibuf_bitmap_get_map_page(
 			page_id_t(space->id, page_no), zip_size, &mtr);
 
-		if (buf_page_is_zeroes(bitmap_page, physical_size)) {
+		if (buf_is_zeroes(span<const byte>(bitmap_page,
+						   physical_size))) {
 			/* This means we got all-zero page instead of
 			ibuf bitmap page. The subsequent page should be
 			all-zero pages. */
@@ -4908,7 +4911,9 @@ dberr_t ibuf_check_bitmap_on_import(const trx_t* trx, fil_space_t* space)
 					page_id_t(space->id, curr_page),
 					zip_size, RW_S_LATCH, &mtr);
 	                        page_t*	page = buf_block_get_frame(block);
-				ut_ad(buf_page_is_zeroes(page, physical_size));
+				ut_ad(buf_is_zeroes(span<const byte>(
+							    page,
+							    physical_size)));
 			}
 #endif /* UNIV_DEBUG */
 			ibuf_exit(&mtr);
diff --git a/storage/innobase/include/btr0types.h b/storage/innobase/include/btr0types.h
index 5c3147f56ea..83c374e2561 100644
--- a/storage/innobase/include/btr0types.h
+++ b/storage/innobase/include/btr0types.h
@@ -56,10 +56,4 @@ in the index record. */
 #define BTR_EXTERN_LOCAL_STORED_MAX_SIZE	\
 	(BTR_EXTERN_FIELD_REF_SIZE * 2)
 
-/** A field reference full of zero, for use in assertions and checks,
-and dummy default values of instantly dropped columns.
-Initially, BLOB field references are set to zero, in
-dtuple_convert_big_rec(). */
-extern const byte field_ref_zero[UNIV_PAGE_SIZE_MAX];
-
 #endif
diff --git a/storage/innobase/include/buf0buf.h b/storage/innobase/include/buf0buf.h
index a6119e38dfd..93cf3742c88 100644
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -33,6 +33,7 @@ Created 11/5/1995 Heikki Tuuri
 #include "fil0fil.h"
 #include "mtr0types.h"
 #include "buf0types.h"
+#include "span.h"
 #ifndef UNIV_INNOCHECKSUM
 #include "hash0hash.h"
 #include "ut0byte.h"
@@ -431,6 +432,7 @@ the same set of mutexes or latches.
 buf_page_t* buf_page_get_zip(const page_id_t page_id, ulint zip_size);
 
 /** This is the general function used to get access to a database page.
+It does page initialization and applies the buffered redo logs.
 @param[in]	page_id		page id
 @param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
 @param[in]	rw_latch	RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH
@@ -454,6 +456,30 @@ buf_page_get_gen(
 	mtr_t*			mtr,
 	dberr_t*		err);
 
+/** Low level function used to get access to a database page.
+@param[in]	page_id		page id
+@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@param[in]	rw_latch	RW_S_LATCH, RW_X_LATCH, RW_NO_LATCH
+@param[in]	guess		guessed block or NULL
+@param[in]	mode		BUF_GET, BUF_GET_IF_IN_POOL,
+BUF_PEEK_IF_IN_POOL, BUF_GET_NO_LATCH, or BUF_GET_IF_IN_POOL_OR_WATCH
+@param[in]	file		file name
+@param[in]	line		line where called
+@param[in]	mtr		mini-transaction
+@param[out]	err		DB_SUCCESS or error code
+@return pointer to the block or NULL */
+buf_block_t*
+buf_page_get_low(
+	const page_id_t		page_id,
+	ulint			zip_size,
+	ulint			rw_latch,
+	buf_block_t*		guess,
+	ulint			mode,
+	const char*		file,
+	unsigned		line,
+	mtr_t*			mtr,
+	dberr_t*		err);
+
 /** Initialize a page in the buffer pool. The page is usually not read
 from a file even if it cannot be found in the buffer buf_pool. This is one
 of the functions which perform to a block a state transition NOT_USED =>
@@ -616,11 +642,10 @@ buf_block_buf_fix_inc_func(
 # endif /* UNIV_DEBUG */
 #endif /* !UNIV_INNOCHECKSUM */
 
-/** Check if a page is all zeroes.
-@param[in]	read_buf	database page
-@param[in]	page_size	page frame size
-@return whether the page is all zeroes */
-bool buf_page_is_zeroes(const void* read_buf, size_t page_size);
+/** Check if a buffer is all zeroes.
+@param[in]	buf	data to check
+@return whether the buffer is all zeroes */
+bool buf_is_zeroes(st_::span<const byte> buf);
 
 /** Checks if the page is in crc32 checksum format.
 @param[in]	read_buf		database page
diff --git a/storage/innobase/include/buf0types.h b/storage/innobase/include/buf0types.h
index 969edfe2374..5532a524782 100644
--- a/storage/innobase/include/buf0types.h
+++ b/storage/innobase/include/buf0types.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 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
@@ -211,6 +211,12 @@ private:
                 const page_id_t        page_id);
 };
 
+/** A field reference full of zero, for use in assertions and checks,
+and dummy default values of instantly dropped columns.
+Initially, BLOB field references are set to zero, in
+dtuple_convert_big_rec(). */
+extern const byte field_ref_zero[UNIV_PAGE_SIZE_MAX];
+
 #ifndef UNIV_INNOCHECKSUM
 
 #include "ut0mutex.h"
diff --git a/storage/innobase/include/data0data.h b/storage/innobase/include/data0data.h
index 5e53dce8429..04ddf5b0a42 100644
--- a/storage/innobase/include/data0data.h
+++ b/storage/innobase/include/data0data.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 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
@@ -32,6 +32,7 @@ Created 5/30/1994 Heikki Tuuri
 #include "mem0mem.h"
 #include "dict0types.h"
 #include "btr0types.h"
+#include <vector>
 
 #include <ostream>
 
@@ -526,9 +527,6 @@ struct dtuple_t {
 	dfield_t*	fields;		/*!< fields */
 	ulint		n_v_fields;	/*!< number of virtual fields */
 	dfield_t*	v_fields;	/*!< fields on virtual column */
-	UT_LIST_NODE_T(dtuple_t) tuple_list;
-					/*!< data tuples can be linked into a
-					list using this field */
 #ifdef UNIV_DEBUG
 	ulint		magic_n;	/*!< magic number, used in
 					debug assertions */
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 76f87dd6e4c..a24b2a43fda 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -971,7 +971,7 @@ public:
 
 	/** Return the next fil_space_t from key rotation list.
 	Once started, the caller must keep calling this until it returns NULL.
-	fil_space_acquire() and fil_space_release() are invoked here which
+	fil_space_acquire() and fil_space_t::release() are invoked here, which
 	blocks a concurrent operation from dropping the tablespace.
 	@param[in]      prev_space      Previous tablespace or NULL to start
 					from beginning of fil_system->rotation
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 19f213db58b..21ddd2b0388 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -336,4 +336,22 @@ times! */
 roll-forward */
 #define RECV_SCAN_SIZE		(4U << srv_page_size_shift)
 
+/** This is a low level function for the recovery system
+to create a page which has buffered intialized redo log records.
+@param[in]	page_id	page to be created using redo logs
+@return whether the page creation successfully */
+buf_block_t* recv_recovery_create_page_low(const page_id_t page_id);
+
+/** Recovery system creates a page which has buffered intialized
+redo log records.
+@param[in]	page_id	page to be created using redo logs
+@return block which contains page was initialized */
+inline buf_block_t* recv_recovery_create_page(const page_id_t page_id)
+{
+  if (UNIV_LIKELY(!recv_recovery_on))
+    return NULL;
+
+  return recv_recovery_create_page_low(page_id);
+}
+
 #endif
diff --git a/storage/innobase/include/page0zip.h b/storage/innobase/include/page0zip.h
index 1637ea00330..2f11a1c77f3 100644
--- a/storage/innobase/include/page0zip.h
+++ b/storage/innobase/include/page0zip.h
@@ -497,7 +497,7 @@ page_zip_calc_checksum(
 @param data    ROW_FORMAT=COMPRESSED page
 @param size    size of the page, in bytes
 @return whether the stored checksum matches innodb_checksum_algorithm */
-bool page_zip_verify_checksum(const void *data, size_t size);
+bool page_zip_verify_checksum(const byte *data, size_t size);
 
 #ifndef UNIV_INNOCHECKSUM
 /**********************************************************************//**
diff --git a/storage/innobase/include/row0ins.h b/storage/innobase/include/row0ins.h
index 9b6aac9c548..95f4388902d 100644
--- a/storage/innobase/include/row0ins.h
+++ b/storage/innobase/include/row0ins.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
@@ -31,6 +31,7 @@ Created 4/20/1996 Heikki Tuuri
 #include "que0types.h"
 #include "trx0types.h"
 #include "row0types.h"
+#include <vector>
 
 /***************************************************************//**
 Checks if foreign key constraint fails for an index entry. Sets shared locks
@@ -52,15 +53,7 @@ row_ins_check_foreign_constraint(
 	dtuple_t*	entry,	/*!< in: index entry for index */
 	que_thr_t*	thr)	/*!< in: query thread */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
-/*********************************************************************//**
-Creates an insert node struct.
-@return own: insert node struct */
-ins_node_t*
-ins_node_create(
-/*============*/
-	ulint		ins_type,	/*!< in: INS_VALUES, ... */
-	dict_table_t*	table,		/*!< in: table where to insert */
-	mem_heap_t*	heap);		/*!< in: mem heap where created */
+
 /*********************************************************************//**
 Sets a new row to insert for an INS_DIRECT node. This function is only used
 if we have constructed the row separately, which is a rare case; this
@@ -158,10 +151,31 @@ row_ins_step(
 /*=========*/
 	que_thr_t*	thr);	/*!< in: query thread */
 
-/* Insert node structure */
+/* Insert node types */
+#define INS_SEARCHED	0	/* INSERT INTO ... SELECT ... */
+#define INS_VALUES	1	/* INSERT INTO ... VALUES ... */
+#define INS_DIRECT	2	/* this is for internal use in dict0crea:
+				insert the row directly */
+
+/* Node execution states */
+#define	INS_NODE_SET_IX_LOCK	1	/* we should set an IX lock on table */
+#define INS_NODE_ALLOC_ROW_ID	2	/* row id should be allocated */
+#define	INS_NODE_INSERT_ENTRIES 3	/* index entries should be built and
+					inserted */
 
-struct ins_node_t{
-	que_common_t	common;	/*!< node type: QUE_NODE_INSERT */
+/** Insert node structure */
+struct ins_node_t
+{
+	explicit ins_node_t(ulint ins_type, dict_table_t *table) :
+		common(QUE_NODE_INSERT, NULL),
+		ins_type(ins_type),
+		row(NULL), table(table), select(NULL), values_list(NULL),
+		state(INS_NODE_SET_IX_LOCK), index(NULL),
+		entry_list(), entry(entry_list.end()),
+		trx_id(0), entry_sys_heap(mem_heap_create(128))
+	{
+	}
+	que_common_t common;	 /*!< node type: QUE_NODE_INSERT */
 	ulint		ins_type;/* INS_VALUES, INS_SEARCHED, or INS_DIRECT */
 	dtuple_t*	row;	/*!< row to insert */
 	dict_table_t*	table;	/*!< table where to insert */
@@ -171,11 +185,12 @@ struct ins_node_t{
 	ulint		state;	/*!< node execution state */
 	dict_index_t*	index;	/*!< NULL, or the next index where the index
 				entry should be inserted */
-	dtuple_t*	entry;	/*!< NULL, or entry to insert in the index;
+	std::vector<dtuple_t*>
+			entry_list;/* list of entries, one for each index */
+	std::vector<dtuple_t*>::iterator
+			entry;	/*!< NULL, or entry to insert in the index;
 				after a successful insert of the entry,
 				this should be reset to NULL */
-	UT_LIST_BASE_NODE_T(dtuple_t)
-			entry_list;/* list of entries, one for each index */
 	/** buffer for the system columns */
 	byte		sys_buf[DATA_ROW_ID_LEN
 				+ DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN];
@@ -188,20 +203,18 @@ struct ins_node_t{
 				entry_list and sys fields are stored here;
 				if this is NULL, entry list should be created
 				and buffers for sys fields in row allocated */
-	ulint		magic_n;
 };
 
-#define	INS_NODE_MAGIC_N	15849075
+/** Create an insert object.
+@param ins_type     INS_VALUES, ...
+@param table        table where to insert
+@param heap         memory heap
+@return the created object */
+inline ins_node_t *ins_node_create(ulint ins_type, dict_table_t *table,
+                                   mem_heap_t *heap)
+{
+  return new (mem_heap_alloc(heap, sizeof(ins_node_t)))
+    ins_node_t(ins_type, table);
+}
 
-/* Insert node types */
-#define INS_SEARCHED	0	/* INSERT INTO ... SELECT ... */
-#define INS_VALUES	1	/* INSERT INTO ... VALUES ... */
-#define INS_DIRECT	2	/* this is for internal use in dict0crea:
-				insert the row directly */
-
-/* Node execution states */
-#define	INS_NODE_SET_IX_LOCK	1	/* we should set an IX lock on table */
-#define INS_NODE_ALLOC_ROW_ID	2	/* row id should be allocated */
-#define	INS_NODE_INSERT_ENTRIES 3	/* index entries should be built and
-					inserted */
 #endif
diff --git a/storage/innobase/include/row0types.h b/storage/innobase/include/row0types.h
index 5f1e46c6a4d..048b161b884 100644
--- a/storage/innobase/include/row0types.h
+++ b/storage/innobase/include/row0types.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2012, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, MariaDB Corporation.
+Copyright (c) 2018, 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
@@ -24,8 +24,8 @@ Row operation global types
 Created 12/27/1996 Heikki Tuuri
 *******************************************************/
 
-#ifndef row0types_h
-#define row0types_h
+#pragma once
+#include "buf0types.h"
 
 struct plan_t;
 
@@ -146,5 +146,3 @@ public:
 		return first_use;
 	}
 };
-
-#endif
diff --git a/storage/innobase/include/span.h b/storage/innobase/include/span.h
deleted file mode 100644
index faeb41029b8..00000000000
--- a/storage/innobase/include/span.h
+++ /dev/null
@@ -1,145 +0,0 @@
-/*****************************************************************************
-
-Copyright (c) 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
-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
-
-*****************************************************************************/
-
-#pragma once
-
-#include <cstddef>
-#include <iterator>
-
-namespace st_ {
-
-template <class ElementType> class span {
-public:
-	typedef ElementType element_type;
-	typedef ElementType value_type;
-	typedef size_t index_type;
-	typedef ptrdiff_t difference_type;
-	typedef element_type* pointer;
-	typedef const element_type* const_pointer;
-	typedef element_type& reference;
-	typedef const element_type& const_reference;
-	typedef pointer iterator;
-	typedef const pointer const_iterator;
-	typedef std::reverse_iterator<iterator> reverse_iterator;
-	typedef std::reverse_iterator<const_iterator> const_reverse_iterator;
-
-	span() : data_(NULL), size_(0) {}
-
-	span(pointer ptr, index_type count) : data_(ptr), size_(count) {}
-
-	span(pointer first, pointer last) : data_(first), size_(last - first) {}
-
-	template <size_t N> span(element_type (&arr)[N]) : data_(arr), size_(N)
-	{
-	}
-
-	template <class Container>
-	span(Container& cont) : data_(cont.begin()), size_(cont.size())
-	{
-	}
-
-	template <class Container>
-	span(const Container& cont) : data_(cont.begin()), size_(cont.size())
-	{
-	}
-
-	span(const span& other) : data_(other.data_), size_(other.size_) {}
-
-	~span(){};
-
-	span& operator=(const span& other)
-	{
-		data_ = other.data_;
-		size_ = other.size_;
-		return *this;
-	}
-
-	template <size_t Count> span<element_type> first() const
-	{
-		assert(!empty());
-		return span(data_, 1);
-	}
-	template <size_t Count> span<element_type> last() const
-	{
-		assert(!empty());
-		return span(data_ + size() - 1, 1);
-	}
-
-	span<element_type> first(index_type count) const
-	{
-		assert(!empty());
-		return span(data_, 1);
-	}
-	span<element_type> last(index_type count) const
-	{
-		assert(!empty());
-		return span(data_ + size() - 1, 1);
-	}
-	span<element_type> subspan(index_type offset, index_type count) const
-	{
-		assert(!empty());
-		assert(size() >= offset + count);
-		return span(data_ + offset, count);
-	}
-
-	index_type size() const { return size_; }
-	index_type size_bytes() const { return size_ * sizeof(ElementType); }
-	bool empty() const __attribute__((warn_unused_result))
-	{
-		return size_ == 0;
-	}
-
-	reference operator[](index_type idx) const
-	{
-		assert(size() > idx);
-		return data_[idx];
-	}
-	reference front() const
-	{
-		assert(!empty());
-		return data_[0];
-	}
-	reference back() const
-	{
-		assert(!empty());
-		return data_[size() - 1];
-	}
-	pointer data() const
-	{
-		assert(!empty());
-		return data_;
-	}
-
-	iterator begin() const { return data_; }
-	iterator end() const { return data_ + size_; }
-	reverse_iterator rbegin() const
-	{
-		return std::reverse_iterator<iterator>(std::advance(end(), -1));
-	}
-	reverse_iterator rend() const
-	{
-		return std::reverse_iterator<iterator>(
-		    std::advance(begin(), -1));
-	}
-
-private:
-	pointer data_;
-	index_type size_;
-};
-
-} // namespace st_
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 106d7e0d49d..1c96a43715b 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -308,7 +308,7 @@ public:
 			if (!i.second.created) {
 				continue;
 			}
-			if (buf_block_t* block = buf_page_get_gen(
+			if (buf_block_t* block = buf_page_get_low(
 				    i.first, 0, RW_X_LATCH, NULL,
 				    BUF_GET_IF_IN_POOL, __FILE__, __LINE__,
 				    &mtr, NULL)) {
@@ -2177,6 +2177,96 @@ static void recv_read_in_area(const page_id_t page_id)
 	mutex_enter(&recv_sys.mutex);
 }
 
+/** This is another low level function for the recovery system
+to create a page which has buffered page intialization redo log records.
+@param[in]	page_id		page to be created using redo logs
+@param[in,out]	recv_addr	Hashed redo logs for the given page id
+@return whether the page creation successfully */
+static buf_block_t* recv_recovery_create_page_low(const page_id_t page_id,
+                                                  recv_addr_t* recv_addr)
+{
+  mtr_t mtr;
+  mlog_init_t::init &i= mlog_init.last(page_id);
+  const lsn_t end_lsn= UT_LIST_GET_LAST(recv_addr->rec_list)->end_lsn;
+
+  if (end_lsn < i.lsn)
+  {
+    DBUG_LOG("ib_log", "skip log for page "
+             << page_id
+             << " LSN " << end_lsn
+             << " < " << i.lsn);
+    recv_addr->state= RECV_PROCESSED;
+ignore:
+    ut_a(recv_sys.n_addrs);
+    recv_sys.n_addrs--;
+    return NULL;
+  }
+
+  fil_space_t *space= fil_space_acquire_for_io(recv_addr->space);
+  if (!space)
+  {
+    recv_addr->state= RECV_PROCESSED;
+    goto ignore;
+  }
+
+  if (space->enable_lsn)
+  {
+init_fail:
+    space->release_for_io();
+    recv_addr->state= RECV_NOT_PROCESSED;
+    return NULL;
+  }
+
+  /* Determine if a tablespace could be for an internal table
+  for FULLTEXT INDEX. For those tables, no MLOG_INDEX_LOAD record
+  used to be written when redo logging was disabled. Hence, we
+  cannot optimize away page reads, because all the redo
+  log records for initializing and modifying the page in the
+  past could be older than the page in the data file.
+
+  The check is too broad, causing all
+  tables whose names start with FTS_ to skip the optimization. */
+
+  if (strstr(space->name, "/FTS_"))
+    goto init_fail;
+
+  mtr.start();
+  mtr.set_log_mode(MTR_LOG_NONE);
+  buf_block_t *block= buf_page_create(page_id, space->zip_size(), &mtr);
+  if (recv_addr->state == RECV_PROCESSED)
+    /* The page happened to exist in the buffer pool, or it was
+    just being read in. Before buf_page_get_with_no_latch() returned,
+    all changes must have been applied to the page already. */
+    mtr.commit();
+  else
+  {
+    i.created= true;
+    buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
+    mtr.x_latch_at_savepoint(0, block);
+    recv_recover_page(block, mtr, recv_addr, &i);
+    ut_ad(mtr.has_committed());
+  }
+
+  space->release_for_io();
+  return block;
+}
+
+/** This is a low level function for the recovery system
+to create a page which has buffered intialized redo log records.
+@param[in]      page_id page to be created using redo logs
+@return whether the page creation successfully */
+buf_block_t* recv_recovery_create_page_low(const page_id_t page_id)
+{
+  buf_block_t* block= nullptr;
+  mutex_enter(&recv_sys.mutex);
+  recv_addr_t* recv_addr= recv_get_fil_addr_struct(page_id.space(),
+                                                   page_id.page_no());
+  if (recv_addr && recv_addr->state == RECV_WILL_NOT_READ)
+    block= recv_recovery_create_page_low(page_id, recv_addr);
+  mutex_exit(&recv_sys.mutex);
+  return block;
+}
+
 /** Apply the hash table of stored log records to persistent data pages.
 @param[in]	last_batch	whether the change buffer merge will be
 				performed as part of the operation */
@@ -2272,7 +2362,7 @@ ignore:
 apply:
 				mtr.start();
 				mtr.set_log_mode(MTR_LOG_NONE);
-				if (buf_block_t* block = buf_page_get_gen(
+				if (buf_block_t* block = buf_page_get_low(
 					    page_id, 0, RW_X_LATCH, NULL,
 					    BUF_GET_IF_IN_POOL,
 					    __FILE__, __LINE__, &mtr, NULL)) {
@@ -2285,79 +2375,9 @@ apply:
 					mtr.commit();
 					recv_read_in_area(page_id);
 				}
-			} else {
-				mlog_init_t::init& i = mlog_init.last(page_id);
-				const lsn_t end_lsn = UT_LIST_GET_LAST(
-					recv_addr->rec_list)->end_lsn;
-
-				if (end_lsn < i.lsn) {
-					DBUG_LOG("ib_log", "skip log for page "
-						 << page_id
-						 << " LSN " << end_lsn
-						 << " < " << i.lsn);
-skip:
-					recv_addr->state = RECV_PROCESSED;
-					goto ignore;
-				}
-
-				fil_space_t* space = fil_space_acquire_for_io(
-					recv_addr->space);
-				if (!space) {
-					goto skip;
-				}
-
-				if (space->enable_lsn) {
-do_read:
-					space->release_for_io();
-					recv_addr->state = RECV_NOT_PROCESSED;
-					goto apply;
-				}
-
-				/* Determine if a tablespace could be
-				for an internal table for FULLTEXT INDEX.
-				For those tables, no MLOG_INDEX_LOAD record
-				used to be written when redo logging was
-				disabled. Hence, we cannot optimize
-				away page reads when crash-upgrading
-				from MariaDB versions before 10.4,
-				because all the redo log records for
-				initializing and modifying the page in
-				the past could be older than the page
-				in the data file.
-
-				The check is too broad, causing all
-				tables whose names start with FTS_ to
-				skip the optimization. */
-				if ((log_sys.log.format
-				     & ~log_t::FORMAT_ENCRYPTED)
-				    != log_t::FORMAT_10_4
-				    && strstr(space->name, "/FTS_")) {
-					goto do_read;
-				}
-
-				mtr.start();
-				mtr.set_log_mode(MTR_LOG_NONE);
-				buf_block_t* block = buf_page_create(
-					page_id, space->zip_size(), &mtr);
-				if (recv_addr->state == RECV_PROCESSED) {
-					/* The page happened to exist
-					in the buffer pool, or it was
-					just being read in. Before
-					buf_page_get_with_no_latch()
-					returned, all changes must have
-					been applied to the page already. */
-					mtr.commit();
-				} else {
-					i.created = true;
-					buf_block_dbg_add_level(
-						block, SYNC_NO_ORDER_CHECK);
-					mtr.x_latch_at_savepoint(0, block);
-					recv_recover_page(block, mtr,
-							  recv_addr, &i);
-					ut_ad(mtr.has_committed());
-				}
-
-				space->release_for_io();
+			} else if (!recv_recovery_create_page_low(
+					page_id, recv_addr)) {
+				goto apply;
 			}
 		}
 	}
diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index ade9350dfa0..727e6c21653 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -31,14 +31,17 @@ Created June 2005 by Marko Makela
 #include "buf0checksum.h"
 #include "ut0crc32.h"
 #include "zlib.h"
+#include "span.h"
 
-#ifndef UNIV_INNOCHECKSUM
+using st_::span;
 
 /** A BLOB field reference full of zero, for use in assertions and tests.
 Initially, BLOB field references are set to zero, in
 dtuple_convert_big_rec(). */
+alignas(UNIV_PAGE_SIZE_MIN)
 const byte field_ref_zero[UNIV_PAGE_SIZE_MAX] = { 0, };
 
+#ifndef UNIV_INNOCHECKSUM
 #include "mtr0log.h"
 #include "dict0dict.h"
 #include "btr0cur.h"
@@ -4992,7 +4995,7 @@ page_zip_calc_checksum(
 @param data    ROW_FORMAT=COMPRESSED page
 @param size    size of the page, in bytes
 @return whether the stored checksum matches innodb_checksum_algorithm */
-bool page_zip_verify_checksum(const void *data, size_t size)
+bool page_zip_verify_checksum(const byte *data, size_t size)
 {
 	const srv_checksum_algorithm_t	curr_algo =
 		static_cast<srv_checksum_algorithm_t>(srv_checksum_algorithm);
@@ -5001,17 +5004,12 @@ bool page_zip_verify_checksum(const void *data, size_t size)
 		return true;
 	}
 
-	for (size_t i = 0; i < size; i++) {
-		if (static_cast<const byte*>(data)[i] != 0) {
-			goto not_all_zeroes;
-		}
+	if (buf_is_zeroes(span<const byte>(data, size))) {
+		return true;
 	}
 
-	return true;
-
-not_all_zeroes:
 	const uint32_t stored = mach_read_from_4(
-		static_cast<const byte*>(data) + FIL_PAGE_SPACE_OR_CHKSUM);
+		data + FIL_PAGE_SPACE_OR_CHKSUM);
 
 	uint32_t calc = page_zip_calc_checksum(data, size, curr_algo);
 
diff --git a/storage/innobase/que/que0que.cc b/storage/innobase/que/que0que.cc
index 6e4bce86df8..3f4810dcc0e 100644
--- a/storage/innobase/que/que0que.cc
+++ b/storage/innobase/que/que0que.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 2018, 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
@@ -460,6 +460,8 @@ que_graph_free_recursive(
 		que_graph_free_recursive(ins->select);
 		ins->select = NULL;
 
+		ins->~ins_node_t();
+
 		if (ins->entry_sys_heap != NULL) {
 			mem_heap_free(ins->entry_sys_heap);
 			ins->entry_sys_heap = NULL;
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index ba48103137a..5bc69c282e1 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -50,6 +50,8 @@ Created 2012-02-08 by Sunny Bains.
 #include <my_aes.h>
 #endif
 
+using st_::span;
+
 /** The size of the buffer to use for IO.
 @param n physical page size
 @return number of pages */
@@ -3454,7 +3456,8 @@ fil_iterate(
 			byte*	src = readptr + i * size;
 			const ulint page_no = page_get_page_no(src);
 			if (!page_no && block->page.id.page_no()) {
-				if (!buf_page_is_zeroes(src, size)) {
+				if (!buf_is_zeroes(span<const byte>(src,
+								    size))) {
 					goto page_corrupted;
 				}
 				/* Proceed to the next page,
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 735bd4517a1..1bde1199906 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, 2018, MariaDB Corporation.
+Copyright (c) 2016, 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
@@ -58,39 +58,6 @@ check.
 If you make a change in this module make sure that no codepath is
 introduced where a call to log_free_check() is bypassed. */
 
-/*********************************************************************//**
-Creates an insert node struct.
-@return own: insert node struct */
-ins_node_t*
-ins_node_create(
-/*============*/
-	ulint		ins_type,	/*!< in: INS_VALUES, ... */
-	dict_table_t*	table,		/*!< in: table where to insert */
-	mem_heap_t*	heap)		/*!< in: mem heap where created */
-{
-	ins_node_t*	node;
-
-	node = static_cast<ins_node_t*>(
-		mem_heap_zalloc(heap, sizeof(ins_node_t)));
-
-	if (!node) {
-		return(NULL);
-	}
-
-	node->common.type = QUE_NODE_INSERT;
-
-	node->ins_type = ins_type;
-
-	node->state = INS_NODE_SET_IX_LOCK;
-	node->table = table;
-
-	node->entry_sys_heap = mem_heap_create(128);
-
-	node->magic_n = INS_NODE_MAGIC_N;
-
-	return(node);
-}
-
 /***********************************************************//**
 Creates an entry template for each index of a table. */
 static
@@ -104,12 +71,12 @@ ins_node_create_entry_list(
 
 	ut_ad(node->entry_sys_heap);
 
-	UT_LIST_INIT(node->entry_list, &dtuple_t::tuple_list);
-
 	/* We will include all indexes (include those corrupted
-	secondary indexes) in the entry list. Filteration of
+	secondary indexes) in the entry list. Filtration of
 	these corrupted index will be done in row_ins() */
 
+	node->entry_list.reserve(UT_LIST_GET_LEN(node->table->indexes));
+
 	for (index = dict_table_get_first_index(node->table);
 	     index != 0;
 	     index = dict_table_get_next_index(index)) {
@@ -118,7 +85,7 @@ ins_node_create_entry_list(
 			node->row, NULL, index, node->entry_sys_heap,
 			ROW_BUILD_FOR_INSERT);
 
-		UT_LIST_ADD_LAST(node->entry_list, entry);
+		node->entry_list.push_back(entry);
 	}
 }
 
@@ -186,7 +153,8 @@ ins_node_set_new_row(
 {
 	node->state = INS_NODE_SET_IX_LOCK;
 	node->index = NULL;
-	node->entry = NULL;
+	node->entry_list.clear();
+	node->entry = node->entry_list.end();
 
 	node->row = row;
 
@@ -3516,16 +3484,16 @@ row_ins_index_entry_step(
 
 	ut_ad(dtuple_check_typed(node->row));
 
-	err = row_ins_index_entry_set_vals(node->index, node->entry,
+	err = row_ins_index_entry_set_vals(node->index, *node->entry,
 					   node->row);
 
 	if (err != DB_SUCCESS) {
 		DBUG_RETURN(err);
 	}
 
-	ut_ad(dtuple_check_typed(node->entry));
+	ut_ad(dtuple_check_typed(*node->entry));
 
-	err = row_ins_index_entry(node->index, node->entry, thr);
+	err = row_ins_index_entry(node->index, *node->entry, thr);
 
 	DEBUG_SYNC_C_IF_THD(thr_get_trx(thr)->mysql_thd,
 			    "after_row_ins_index_entry_step");
@@ -3643,7 +3611,8 @@ row_ins(
 		row_ins_alloc_row_id_step(node);
 
 		node->index = dict_table_get_first_index(node->table);
-		node->entry = UT_LIST_GET_FIRST(node->entry_list);
+		ut_ad(node->entry_list.empty() == false);
+		node->entry = node->entry_list.begin();
 
 		if (node->ins_type == INS_SEARCHED) {
 
@@ -3669,20 +3638,16 @@ row_ins(
 		}
 
 		node->index = dict_table_get_next_index(node->index);
-		node->entry = UT_LIST_GET_NEXT(tuple_list, node->entry);
-
-		DBUG_EXECUTE_IF(
-			"row_ins_skip_sec",
-			node->index = NULL; node->entry = NULL; break;);
+		++node->entry;
 
 		/* Skip corrupted secondary index and its entry */
 		while (node->index && node->index->is_corrupted()) {
 			node->index = dict_table_get_next_index(node->index);
-			node->entry = UT_LIST_GET_NEXT(tuple_list, node->entry);
+			++node->entry;
 		}
 	}
 
-	ut_ad(node->entry == NULL);
+	ut_ad(node->entry == node->entry_list.end());
 
 	node->state = INS_NODE_ALLOC_ROW_ID;
 
@@ -3738,14 +3703,14 @@ row_ins_step(
 		DBUG_ASSERT(node->table->get_ref_count() > 0);
 		DBUG_ASSERT(node->ins_type == INS_DIRECT);
 		/* No-rollback tables can consist only of a single index. */
-		DBUG_ASSERT(UT_LIST_GET_LEN(node->entry_list) == 1);
+		DBUG_ASSERT(node->entry_list.size() == 1);
 		DBUG_ASSERT(UT_LIST_GET_LEN(node->table->indexes) == 1);
 		/* There should be no possibility for interruption and
 		restarting here. In theory, we could allow resumption
 		from the INS_NODE_INSERT_ENTRIES state here. */
 		DBUG_ASSERT(node->state == INS_NODE_SET_IX_LOCK);
 		node->index = dict_table_get_first_index(node->table);
-		node->entry = UT_LIST_GET_FIRST(node->entry_list);
+		node->entry = node->entry_list.begin();
 		node->state = INS_NODE_INSERT_ENTRIES;
 		goto do_insert;
 	}
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 6e6c1abb3a8..65e34b317c3 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1098,7 +1098,7 @@ row_get_prebuilt_insert_row(
 		may need to rebuild the row insert template. */
 
 		if (prebuilt->trx_id == table->def_trx_id
-		    && UT_LIST_GET_LEN(prebuilt->ins_node->entry_list)
+		    && prebuilt->ins_node->entry_list.size()
 		    == UT_LIST_GET_LEN(table->indexes)) {
 
 			return(prebuilt->ins_node->row);
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index 1ff18a2a7eb..e612a9f577b 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -43,10 +43,8 @@ Created 3/26/1996 Heikki Tuuri
 const dtuple_t trx_undo_metadata = {
 	/* This also works for REC_INFO_METADATA_ALTER, because the
 	delete-mark (REC_INFO_DELETED_FLAG) is ignored when searching. */
-	REC_INFO_METADATA_ADD,
-	0, 0,
-	NULL, 0, NULL,
-	UT_LIST_NODE_T(dtuple_t)()
+	REC_INFO_METADATA_ADD, 0, 0,
+	NULL, 0, NULL
 #ifdef UNIV_DEBUG
 	, DATA_TUPLE_MAGIC_N
 #endif /* UNIV_DEBUG */