Merge 10.3 into 10.4

35 files changed, 331 insertions, 203 deletions

diff --git a/storage/connect/inihandl.cpp b/storage/connect/inihandl.cpp
index 8e79aeac7ef..dacab3c485c 100644
--- a/storage/connect/inihandl.cpp
+++ b/storage/connect/inihandl.cpp
@@ -194,7 +194,7 @@ static void PROFILE_Save( FILE *file, PROFILESECTION *section )
     }
 
     for (key = section->key; key; key = key->next)
-      if (key->name && key->name[0]) {
+      if (key->name[0]) {
         fprintf(file, "%s", SVP(key->name));
 
         if (key->value)
diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index d6eff63b360..0c563954703 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -3355,25 +3355,22 @@ btr_parse_set_min_rec_mark(
 	return(ptr + 2);
 }
 
-/****************************************************************//**
-Sets a record as the predefined minimum record. */
-void
-btr_set_min_rec_mark(
-/*=================*/
-	rec_t*	rec,	/*!< in: record */
-	mtr_t*	mtr)	/*!< in: mtr */
+/** Sets a record as the predefined minimum record. */
+void btr_set_min_rec_mark(rec_t* rec, mtr_t* mtr)
 {
-	ulint	info_bits;
+	const bool comp = page_rec_is_comp(rec);
 
-	if (page_rec_is_comp(rec)) {
-		info_bits = rec_get_info_bits(rec, TRUE);
+	ut_ad(rec == page_rec_get_next_const(page_get_infimum_rec(
+						     page_align(rec))));
+	ut_ad(!(rec_get_info_bits(page_rec_get_next(rec), comp)
+		& REC_INFO_MIN_REC_FLAG));
 
+	size_t info_bits = rec_get_info_bits(rec, comp);
+	if (comp) {
 		rec_set_info_bits_new(rec, info_bits | REC_INFO_MIN_REC_FLAG);
 
 		btr_set_min_rec_mark_log(rec, MLOG_COMP_REC_MIN_MARK, mtr);
 	} else {
-		info_bits = rec_get_info_bits(rec, FALSE);
-
 		rec_set_info_bits_old(rec, info_bits | REC_INFO_MIN_REC_FLAG);
 
 		btr_set_min_rec_mark_log(rec, MLOG_REC_MIN_MARK, mtr);
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 2731211db72..9d734e4a237 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -5911,6 +5911,9 @@ btr_cur_pessimistic_delete(
 #endif /* UNIV_ZIP_DEBUG */
 	}
 
+	rec_t* next_rec = NULL;
+	bool min_mark_next_rec = false;
+
 	if (page_is_leaf(page)) {
 		const bool is_metadata = rec_is_metadata(
 			rec, page_rec_is_comp(rec));
@@ -5988,20 +5991,14 @@ discard_page:
 			goto return_after_reservations;
 		}
 
-		rec_t*	next_rec = page_rec_get_next(rec);
+		next_rec = page_rec_get_next(rec);
 
 		if (!page_has_prev(page)) {
-
 			/* If we delete the leftmost node pointer on a
 			non-leaf level, we must mark the new leftmost node
 			pointer as the predefined minimum record */
 
-			/* This will make page_zip_validate() fail until
-			page_cur_delete_rec() completes.  This is harmless,
-			because everything will take place within a single
-			mini-transaction and because writing to the redo log
-			is an atomic operation (performed by mtr_commit()). */
-			btr_set_min_rec_mark(next_rec, mtr);
+			min_mark_next_rec = true;
 		} else if (dict_index_is_spatial(index)) {
 			/* For rtree, if delete the leftmost node pointer,
 			we need to update parent page. */
@@ -6069,6 +6066,11 @@ discard_page:
 				block->zip_size(), mtr);
 		page_cur_delete_rec(btr_cur_get_page_cur(cursor), index,
 				    offsets, mtr);
+
+		if (min_mark_next_rec) {
+			btr_set_min_rec_mark(next_rec, mtr);
+		}
+
 #ifdef UNIV_ZIP_DEBUG
 		ut_a(!page_zip || page_zip_validate(page_zip, page, index));
 #endif /* UNIV_ZIP_DEBUG */
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index aa681e67201..fa6a33662ad 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -592,12 +592,6 @@ decrypt_failed:
 				    << " in file " << space->chain.start->name
 				    << " looks corrupted; key_version="
 				    << key_version;
-			/* Mark page encrypted in case it should be. */
-			if (space->crypt_data->type
-			    != CRYPT_SCHEME_UNENCRYPTED) {
-				bpage->encrypted = true;
-			}
-
 			return false;
 		}
 
@@ -608,8 +602,7 @@ decrypt_failed:
 		ut_d(fil_page_type_validate(space, dst_frame));
 
 		/* decrypt using crypt_buf to dst_frame */
-		if (!fil_space_decrypt(space, slot->crypt_buf,
-				       dst_frame, &bpage->encrypted)) {
+		if (!fil_space_decrypt(space, slot->crypt_buf, dst_frame)) {
 			slot->release();
 			goto decrypt_failed;
 		}
@@ -1579,7 +1572,6 @@ buf_block_init(
 	block->page.buf_fix_count = 0;
 	block->page.io_fix = BUF_IO_NONE;
 	block->page.flush_observer = NULL;
-	block->page.encrypted = false;
 	block->page.real_size = 0;
 	block->page.write_size = 0;
 	block->modify_clock = 0;
@@ -4056,7 +4048,6 @@ err_exit:
 	if (encrypted) {
 		ib::info() << "Row compressed page could be encrypted"
 			" with key_version " << key_version;
-		block->page.encrypted = true;
 	}
 
 	if (space) {
@@ -5264,7 +5255,6 @@ buf_page_init_low(
 	bpage->newest_modification = 0;
 	bpage->oldest_modification = 0;
 	bpage->write_size = 0;
-	bpage->encrypted = false;
 	bpage->real_size = 0;
 	bpage->slot = NULL;
 
@@ -5853,17 +5843,19 @@ buf_page_monitor(
 }
 
 /** Mark a table corrupted.
-Also remove the bpage from LRU list.
-@param[in]	bpage	Corrupted page. */
-static void buf_mark_space_corrupt(buf_page_t* bpage, const fil_space_t* space)
+@param[in]	bpage	corrupted page
+@param[in]	space	tablespace of the corrupted page */
+ATTRIBUTE_COLD
+static void buf_mark_space_corrupt(buf_page_t* bpage, const fil_space_t& space)
 {
 	/* If block is not encrypted find the table with specified
 	space id, and mark it corrupted. Encrypted tables
 	are marked unusable later e.g. in ::open(). */
-	if (!bpage->encrypted) {
-		dict_set_corrupted_by_space(space);
+	if (!space.crypt_data
+	    || space.crypt_data->type == CRYPT_SCHEME_UNENCRYPTED) {
+		dict_set_corrupted_by_space(&space);
 	} else {
-		dict_set_encrypted_by_space(space);
+		dict_set_encrypted_by_space(&space);
 	}
 }
 
@@ -5902,7 +5894,7 @@ buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
 	mutex_exit(buf_page_get_mutex(bpage));
 
 	if (!srv_force_recovery) {
-		buf_mark_space_corrupt(bpage, space);
+		buf_mark_space_corrupt(bpage, *space);
 	}
 
 	/* After this point bpage can't be referenced. */
@@ -5955,7 +5947,6 @@ static dberr_t buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 	byte* dst_frame = (bpage->zip.data) ? bpage->zip.data :
 		((buf_block_t*) bpage)->frame;
 	dberr_t err = DB_SUCCESS;
-	bool corrupted = false;
 	uint key_version = buf_page_get_key_version(dst_frame, space->flags);
 
 	/* In buf_decrypt_after_read we have either decrypted the page if
@@ -5964,43 +5955,28 @@ static dberr_t buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 	not decrypted and it could be either encrypted and corrupted
 	or corrupted or good page. If we decrypted, there page could
 	still be corrupted if used key does not match. */
-	const bool still_encrypted = (!space->full_crc32() && key_version)
+	const bool seems_encrypted = (!space->full_crc32() && key_version)
 		&& space->crypt_data
-		&& space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED
-		&& !bpage->encrypted
-		&& fil_space_verify_crypt_checksum(dst_frame,
-						   bpage->zip_size());
+		&& space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED;
 	ut_ad(space->purpose != FIL_TYPE_TEMPORARY || space->full_crc32());
 
-	if (!still_encrypted) {
-		/* 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())
-		    && (key_version || space->is_compressed()
-			|| space->purpose == FIL_TYPE_TEMPORARY)) {
-			corrupted = buf_page_full_crc32_is_corrupted(
-					space->id, dst_frame,
-					space->is_compressed());
-		} else {
-			corrupted = buf_page_is_corrupted(
-				true, dst_frame, space->flags);
-		}
-
-		if (!corrupted) {
-			bpage->encrypted = false;
-		} else {
+	/* 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())
+	    && (key_version || space->is_compressed()
+		|| space->purpose == FIL_TYPE_TEMPORARY)) {
+		if (buf_page_full_crc32_is_corrupted(
+			    space->id, dst_frame, space->is_compressed())) {
 			err = DB_PAGE_CORRUPTED;
 		}
+
+	} else if (buf_page_is_corrupted(true, dst_frame, space->flags)) {
+		err = DB_PAGE_CORRUPTED;
 	}
 
-	/* Pages that we think are unencrypted but do not match the checksum
-	checks could be corrupted or encrypted or both. */
-	if (corrupted && !bpage->encrypted) {
-		/* An error will be reported by
-		buf_page_io_complete(). */
-	} else if (still_encrypted || (bpage->encrypted && corrupted)) {
-		bpage->encrypted = true;
+	if (seems_encrypted && err == DB_PAGE_CORRUPTED
+	    && bpage->id.page_no() != 0) {
 		err = DB_DECRYPTION_FAILED;
 
 		ib::error()
@@ -6061,7 +6037,6 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 	if (io_type == BUF_IO_READ) {
 		ulint	read_page_no = 0;
 		ulint	read_space_id = 0;
-		uint	key_version = 0;
 		byte*	frame = bpage->zip.data
 			? bpage->zip.data
 			: reinterpret_cast<buf_block_t*>(bpage)->frame;
@@ -6101,7 +6076,6 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 		read_page_no = mach_read_from_4(frame + FIL_PAGE_OFFSET);
 		read_space_id = mach_read_from_4(
 			frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
-		key_version = buf_page_get_key_version(frame, space->flags);
 
 		if (bpage->id.space() == TRX_SYS_SPACE
 		    && buf_dblwr_page_inside(bpage->id.page_no())) {
@@ -6213,24 +6187,9 @@ database_corrupted:
 			|| !is_predefined_tablespace(bpage->id.space()))
 		    && fil_page_get_type(frame) == FIL_PAGE_INDEX
 		    && page_is_leaf(frame)) {
-
-			if (bpage->encrypted) {
-				ib::warn()
-					<< "Table in tablespace "
-					<< bpage->id.space()
-					<< " encrypted. However key "
-					"management plugin or used "
-					<< "key_version " << key_version
-					<< " is not found or"
-					" used encryption algorithm or method does not match."
-					" Can't continue opening the table.";
-			} else {
-
-				ibuf_merge_or_delete_for_page(
-					(buf_block_t*) bpage, bpage->id,
-					bpage->zip_size(), true);
-			}
-
+			ibuf_merge_or_delete_for_page(
+				reinterpret_cast<buf_block_t*>(bpage),
+				bpage->id, bpage->zip_size(), true);
 		}
 
 		space->release_for_io();
diff --git a/storage/innobase/buf/buf0flu.cc b/storage/innobase/buf/buf0flu.cc
index 65febb14a76..d215f92f80f 100644
--- a/storage/innobase/buf/buf0flu.cc
+++ b/storage/innobase/buf/buf0flu.cc
@@ -835,7 +835,7 @@ buf_flush_init_for_writing(
 
 	if (use_full_checksum) {
 		mach_write_to_4(page + srv_page_size - FIL_PAGE_FCRC32_END_LSN,
-				(ulint) newest_lsn);
+				static_cast<uint32_t>(newest_lsn));
 	} else {
 		mach_write_to_8(page + srv_page_size - FIL_PAGE_END_LSN_OLD_CHKSUM,
 				newest_lsn);
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 38d1b9778a5..d81e16b63c1 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -1014,7 +1014,6 @@ Decrypt a page.
 @param[in]	space			Tablespace
 @param[in]	tmp_frame		Temporary buffer used for decrypting
 @param[in,out]	src_frame		Page to decrypt
-@param[out]	decrypted		true if page was decrypted
 @return decrypted page, or original not encrypted page if decryption is
 not needed.*/
 UNIV_INTERN
@@ -1022,13 +1021,11 @@ byte*
 fil_space_decrypt(
 	const fil_space_t* space,
 	byte*		tmp_frame,
-	byte*		src_frame,
-	bool*		decrypted)
+	byte*		src_frame)
 {
 	dberr_t err = DB_SUCCESS;
 	byte* res = NULL;
 	const ulint physical_size = space->physical_size();
-	*decrypted = false;
 
 	ut_ad(space->crypt_data != NULL && space->crypt_data->is_encrypted());
 	ut_ad(space->pending_io());
@@ -1040,7 +1037,6 @@ fil_space_decrypt(
 
 	if (err == DB_SUCCESS) {
 		if (encrypted) {
-			*decrypted = true;
 			/* Copy the decrypted page back to page buffer, not
 			really any other options. */
 			memcpy(src_frame, tmp_frame, physical_size);
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 138a420040d..13d606bcc34 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -5835,7 +5835,6 @@ add_all_virtual:
 		buf_block_t* root = btr_root_block_get(index, RW_X_LATCH,
 						       &mtr);
 		DBUG_ASSERT(root);
-		DBUG_ASSERT(!root->page.encrypted);
 		if (fil_page_get_type(root->frame) != FIL_PAGE_TYPE_INSTANT) {
 			DBUG_ASSERT(!"wrong page type");
 			err = DB_CORRUPTION;
@@ -5925,8 +5924,7 @@ empty_table:
 	mtr.start();
 	index->set_modified(mtr);
 	if (buf_block_t* root = btr_root_block_get(index, RW_SX_LATCH, &mtr)) {
-		if (root->page.encrypted
-		    || fil_page_get_type(root->frame) != FIL_PAGE_INDEX) {
+		if (fil_page_get_type(root->frame) != FIL_PAGE_INDEX) {
 			DBUG_ASSERT(!"wrong page type");
 			goto err_exit;
 		}
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index c21a26a898d..09050f0c197 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -336,9 +336,8 @@ ibuf_header_page_get(
 		page_id_t(IBUF_SPACE_ID, FSP_IBUF_HEADER_PAGE_NO),
 		0, RW_X_LATCH, mtr);
 
-	if (!block->page.encrypted) {
+	if (block) {
 		buf_block_dbg_add_level(block, SYNC_IBUF_HEADER);
-
 		page = buf_block_get_frame(block);
 	}
 
diff --git a/storage/innobase/include/btr0btr.h b/storage/innobase/include/btr0btr.h
index 1ec2eee9a77..e0fa2e734f8 100644
--- a/storage/innobase/include/btr0btr.h
+++ b/storage/innobase/include/btr0btr.h
@@ -515,14 +515,10 @@ btr_insert_on_non_leaf_level_func(
 	mtr_t*		mtr);	/*!< in: mtr */
 #define btr_insert_on_non_leaf_level(f,i,l,t,m)			\
 	btr_insert_on_non_leaf_level_func(f,i,l,t,__FILE__,__LINE__,m)
-/****************************************************************//**
-Sets a record as the predefined minimum record. */
-void
-btr_set_min_rec_mark(
-/*=================*/
-	rec_t*	rec,	/*!< in/out: record */
-	mtr_t*	mtr)	/*!< in: mtr */
-	MY_ATTRIBUTE((nonnull));
+
+/** Sets a record as the predefined minimum record. */
+void btr_set_min_rec_mark(rec_t* rec, mtr_t* mtr) MY_ATTRIBUTE((nonnull));
+
 /** Seek to the parent page of a B-tree page.
 @param[in,out]	index	b-tree
 @param[in]	block	child page
diff --git a/storage/innobase/include/buf0buf.h b/storage/innobase/include/buf0buf.h
index 647cc3d4274..93a27028ec4 100644
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -1491,8 +1491,6 @@ public:
 					if written again we check is TRIM
 					operation needed. */
 
-	bool            encrypted;	/*!< page is still encrypted */
-
 	ulint           real_size;	/*!< Real size of the page
 					Normal pages == srv_page_size
 					page compressed pages, payload
diff --git a/storage/innobase/include/buf0flu.h b/storage/innobase/include/buf0flu.h
index 23d14797510..e022dd55215 100644
--- a/storage/innobase/include/buf0flu.h
+++ b/storage/innobase/include/buf0flu.h
@@ -174,16 +174,6 @@ void
 buf_flush_wait_flushed(
 	lsn_t		new_oldest);
 
-/******************************************************************//**
-Waits until a flush batch of the given type ends. This is called by
-a thread that only wants to wait for a flush to end but doesn't do
-any flushing itself. */
-void
-buf_flush_wait_batch_end_wait_only(
-/*===============================*/
-	buf_pool_t*	buf_pool,	/*!< in: buffer pool instance */
-	buf_flush_t	type);		/*!< in: BUF_FLUSH_LRU
-					or BUF_FLUSH_LIST */
 /********************************************************************//**
 This function should be called at a mini-transaction commit, if a page was
 modified in it. Puts the block to the list of modified blocks, if it not
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 79924edceb4..fd0d3e12601 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -376,7 +376,6 @@ Decrypt a page
 @param[in]	space			Tablespace
 @param[in]	tmp_frame		Temporary buffer used for decrypting
 @param[in,out]	src_frame		Page to decrypt
-@param[out]	decrypted		true if page was decrypted
 @return decrypted page, or original not encrypted page if decryption is
 not needed.*/
 UNIV_INTERN
@@ -384,8 +383,7 @@ byte*
 fil_space_decrypt(
 	const fil_space_t* space,
 	byte*		tmp_frame,
-	byte*		src_frame,
-	bool*		decrypted)
+	byte*		src_frame)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /**
diff --git a/storage/innobase/include/page0page.h b/storage/innobase/include/page0page.h
index a6dd19cc147..0fc397cd288 100644
--- a/storage/innobase/include/page0page.h
+++ b/storage/innobase/include/page0page.h
@@ -1292,15 +1292,12 @@ ibool
 page_simple_validate_new(
 /*=====================*/
 	const page_t*	page);	/*!< in: index page in ROW_FORMAT!=REDUNDANT */
-/***************************************************************//**
-This function checks the consistency of an index page.
-@return TRUE if ok */
-ibool
-page_validate(
-/*==========*/
-	const page_t*	page,	/*!< in: index page */
-	dict_index_t*	index);	/*!< in: data dictionary index containing
-				the page record type definition */
+/** Check the consistency of an index page.
+@param[in]	page	index page
+@param[in]	index	B-tree or R-tree index
+@return	whether the page is valid */
+bool page_validate(const page_t* page, const dict_index_t* index)
+	MY_ATTRIBUTE((nonnull));
 /***************************************************************//**
 Looks in the page record list for a record with the given heap number.
 @return record, NULL if not found */
diff --git a/storage/innobase/include/page0page.ic b/storage/innobase/include/page0page.ic
index 5ab50b7bb8e..3507ea8830c 100644
--- a/storage/innobase/include/page0page.ic
+++ b/storage/innobase/include/page0page.ic
@@ -651,6 +651,10 @@ page_rec_get_next_low(
 		return(NULL);
 	}
 
+	ut_ad(page_rec_is_infimum(rec)
+	      || !(rec_get_info_bits(page + offs, comp)
+		   & REC_INFO_MIN_REC_FLAG));
+
 	return(page + offs);
 }
 
diff --git a/storage/innobase/include/univ.i b/storage/innobase/include/univ.i
index 31380f09d0f..644a9484af4 100644
--- a/storage/innobase/include/univ.i
+++ b/storage/innobase/include/univ.i
@@ -93,7 +93,6 @@ support cross-platform development and expose comonly used SQL names. */
 #include <sys/stat.h>
 
 #ifndef _WIN32
-# include <sys/mman.h> /* mmap() for os0proc.cc */
 # include <sched.h>
 # include "my_config.h"
 #endif
diff --git a/storage/innobase/include/ut0new.h b/storage/innobase/include/ut0new.h
index 6732a12ecf1..a190b872549 100644
--- a/storage/innobase/include/ut0new.h
+++ b/storage/innobase/include/ut0new.h
@@ -129,9 +129,6 @@ InnoDB:
 #include <string.h> /* strlen(), strrchr(), strncmp() */
 
 #include "my_global.h" /* needed for headers from mysql/psi/ */
-#if !defined(DBUG_OFF) && defined(HAVE_MADVISE)
-#include <sys/mman.h>
-#endif
 
 /* JAN: TODO: missing 5.7 header */
 #ifdef HAVE_MYSQL_MEMORY_H
@@ -252,7 +249,7 @@ static inline void ut_allocate_trace_dontdump(void *ptr, size_t	bytes,
 
 #if defined(DBUG_OFF) && defined(HAVE_MADVISE) && defined(MADV_DONTDUMP)
 	if (dontdump && madvise(ptr, bytes, MADV_DONTDUMP)) {
-		ib::warn() << "Failed to set memory to DONTDUMP: "
+		ib::warn() << "Failed to set memory to " DONTDUMP_STR ": "
 			   << strerror(errno)
 			   << " ptr " << ptr
 			   << " size " << bytes;
@@ -270,7 +267,7 @@ static inline void ut_allocate_trace_dontdump(void *ptr, size_t	bytes,
 static inline void ut_dodump(void* ptr, size_t m_size)
 {
 	if (ptr && madvise(ptr, m_size, MADV_DODUMP)) {
-		ib::warn() << "Failed to set memory to DODUMP: "
+		ib::warn() << "Failed to set memory to " DODUMP_STR ": "
 			   << strerror(errno)
 			   << " ptr " << ptr
 			   << " size " << m_size;
diff --git a/storage/innobase/page/page0cur.cc b/storage/innobase/page/page0cur.cc
index c2bdaee8c13..f777e7b8378 100644
--- a/storage/innobase/page/page0cur.cc
+++ b/storage/innobase/page/page0cur.cc
@@ -2420,10 +2420,6 @@ page_cur_delete_rec(
 	if (cur_n_owned <= PAGE_DIR_SLOT_MIN_N_OWNED) {
 		page_dir_balance_slot(page, page_zip, cur_slot_no);
 	}
-
-#ifdef UNIV_ZIP_DEBUG
-	ut_a(!page_zip || page_zip_validate(page_zip, page, index));
-#endif /* UNIV_ZIP_DEBUG */
 }
 
 #ifdef UNIV_COMPILE_TEST_FUNCS
diff --git a/storage/innobase/page/page0page.cc b/storage/innobase/page/page0page.cc
index fa73b86f03f..a40372a2ed9 100644
--- a/storage/innobase/page/page0page.cc
+++ b/storage/innobase/page/page0page.cc
@@ -2353,19 +2353,16 @@ func_exit:
 	return(ret);
 }
 
-/***************************************************************//**
-This function checks the consistency of an index page.
-@return TRUE if ok */
-ibool
-page_validate(
-/*==========*/
-	const page_t*	page,	/*!< in: index page */
-	dict_index_t*	index)	/*!< in: data dictionary index containing
-				the page record type definition */
+/** Check the consistency of an index page.
+@param[in]	page	index page
+@param[in]	index	B-tree or R-tree index
+@return	whether the page is valid */
+bool page_validate(const page_t* page, const dict_index_t* index)
 {
 	const page_dir_slot_t*	slot;
 	const rec_t*		rec;
 	const rec_t*		old_rec		= NULL;
+	const rec_t*		first_rec	= NULL;
 	ulint			offs;
 	ulint			n_slots;
 	ibool			ret		= TRUE;
@@ -2489,6 +2486,41 @@ wrong_page_type:
 			goto next_rec;
 		}
 
+		if (rec == first_rec) {
+			if ((rec_get_info_bits(rec, page_is_comp(page))
+			     & REC_INFO_MIN_REC_FLAG)) {
+				if (page_has_prev(page)) {
+					ib::error() << "REC_INFO_MIN_REC_FLAG "
+						"is set in on non-left page";
+					ret = false;
+				} else if (!page_is_leaf(page)) {
+					/* leftmost node pointer page */
+				} else if (!index->is_instant()) {
+					ib::error() << "REC_INFO_MIN_REC_FLAG "
+						"is set in a leaf-page record";
+					ret = false;
+				} else if (!rec_get_deleted_flag(
+						   rec, page_is_comp(page))
+					   != !index->table->instant) {
+					ib::error() << (index->table->instant
+							? "Metadata record "
+							"is not delete-marked"
+							: "Metadata record "
+							"is delete-marked");
+					ret = false;
+				}
+			} else if (!page_has_prev(page)
+				   && index->is_instant()) {
+				ib::error() << "Metadata record is missing";
+				ret = false;
+			}
+		} else if (rec_get_info_bits(rec, page_is_comp(page))
+			   & REC_INFO_MIN_REC_FLAG) {
+			ib::error() << "REC_INFO_MIN_REC_FLAG record is not "
+				       "first in page";
+			ret = false;
+		}
+
 		/* Check that the records are in the ascending order */
 		if (count >= PAGE_HEAP_NO_USER_LOW
 		    && !page_rec_is_supremum(rec)) {
@@ -2595,6 +2627,11 @@ next_rec:
 		old_rec = rec;
 		rec = page_rec_get_next_const(rec);
 
+		if (page_rec_is_infimum(old_rec)
+		    && page_rec_is_user_rec(rec)) {
+			first_rec = rec;
+		}
+
 		/* set old_offsets to offsets; recycle offsets */
 		{
 			ulint* offs = old_offsets;
diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index 243f44ad118..d0318b51f4a 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -4315,10 +4315,6 @@ page_zip_clear_rec(
 	} else {
 		ut_ad(!rec_offs_any_extern(offsets));
 	}
-
-#ifdef UNIV_ZIP_DEBUG
-	ut_a(page_zip_validate(page_zip, page, index));
-#endif /* UNIV_ZIP_DEBUG */
 }
 
 /**********************************************************************//**
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 148d223a364..a45d9f59caf 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -2022,29 +2022,13 @@ dberr_t PageConverter::operator()(buf_block_t* block) UNIV_NOTHROW
 
 	ulint		page_type;
 
-	dberr_t err = update_page(block, page_type);
-	if (err != DB_SUCCESS) return err;
+	if (dberr_t err = update_page(block, page_type)) {
+		return err;
+	}
 
 	const bool full_crc32 = fil_space_t::full_crc32(get_space_flags());
-	const bool page_compressed = fil_space_t::is_compressed(get_space_flags());
 
 	if (!block->page.zip.data) {
-		if (full_crc32
-		    && (block->page.encrypted || page_compressed)
-		    && block->page.id.page_no() > 0) {
-			byte* page = block->frame;
-			mach_write_to_8(page + FIL_PAGE_LSN, m_current_lsn);
-
-			if (!page_compressed) {
-				mach_write_to_4(
-					page + (srv_page_size
-						- FIL_PAGE_FCRC32_END_LSN),
-					(ulint) m_current_lsn);
-			}
-
-			return err;
-		}
-
 		buf_flush_init_for_writing(
 			NULL, block->frame, NULL, m_current_lsn, full_crc32);
 	} else if (fil_page_type_is_index(page_type)) {
@@ -3484,10 +3468,6 @@ not_encrypted:
 				goto page_corrupted;
 			}
 
-			if (encrypted) {
-				block->page.encrypted = true;
-			}
-
 			if ((err = callback(block)) != DB_SUCCESS) {
 				goto func_exit;
 			} else if (!updated) {
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index 2431afcd1e9..e89f306b8cf 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -52,9 +52,6 @@
 #ifdef HAVE_SYS_VADVISE_H
 #include <sys/vadvise.h>
 #endif
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 
 /* Functions defined in this file */
 
diff --git a/storage/maria/ma_extra.c b/storage/maria/ma_extra.c
index 12cb8f838ef..b464cf4f94e 100644
--- a/storage/maria/ma_extra.c
+++ b/storage/maria/ma_extra.c
@@ -14,9 +14,6 @@
    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */
 
 #include "maria_def.h"
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 #include "ma_blockrec.h"
 
 static void maria_extra_keyflag(MARIA_HA *info,
diff --git a/storage/maria/ma_packrec.c b/storage/maria/ma_packrec.c
index e2c1e353616..2e7add8f61d 100644
--- a/storage/maria/ma_packrec.c
+++ b/storage/maria/ma_packrec.c
@@ -1528,9 +1528,6 @@ static uint max_bit(register uint value)
 /*****************************************************************************
 	Some redefined functions to handle files when we are using memmap
 *****************************************************************************/
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 
 #ifdef HAVE_MMAP
 
diff --git a/storage/maria/maria_chk.c b/storage/maria/maria_chk.c
index 54f7c21539b..2f130de1c7a 100644
--- a/storage/maria/maria_chk.c
+++ b/storage/maria/maria_chk.c
@@ -23,9 +23,6 @@
 #include <my_getopt.h>
 #include <my_check_opt.h>
 #include <my_handler_errors.h>
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 
 static uint decode_bits;
 static char **default_argv;
diff --git a/storage/myisam/mi_check.c b/storage/myisam/mi_check.c
index 22eb97b24f3..f377028be52 100644
--- a/storage/myisam/mi_check.c
+++ b/storage/myisam/mi_check.c
@@ -46,9 +46,6 @@
 #ifdef HAVE_SYS_VADVISE_H
 #include <sys/vadvise.h>
 #endif
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 #include "rt_index.h"
 #include <mysqld_error.h>
 
diff --git a/storage/myisam/mi_extra.c b/storage/myisam/mi_extra.c
index 6c7144a949d..d1bf903bd46 100644
--- a/storage/myisam/mi_extra.c
+++ b/storage/myisam/mi_extra.c
@@ -15,9 +15,6 @@
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335  USA */
 
 #include "myisamdef.h"
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 
 static void mi_extra_keyflag(MI_INFO *info, enum ha_extra_function function);
 
diff --git a/storage/myisam/mi_packrec.c b/storage/myisam/mi_packrec.c
index 72ce17b6a78..b28eaf21090 100644
--- a/storage/myisam/mi_packrec.c
+++ b/storage/myisam/mi_packrec.c
@@ -1479,9 +1479,6 @@ static uint max_bit(register uint value)
 /*****************************************************************************
 	Some redefined functions to handle files when we are using memmap
 *****************************************************************************/
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 
 #ifdef HAVE_MMAP
 
diff --git a/storage/myisam/myisamchk.c b/storage/myisam/myisamchk.c
index a3021ed72a7..f616a9f9b1a 100644
--- a/storage/myisam/myisamchk.c
+++ b/storage/myisam/myisamchk.c
@@ -21,9 +21,6 @@
 #include <stdarg.h>
 #include <my_getopt.h>
 #include <my_bit.h>
-#ifdef HAVE_SYS_MMAN_H
-#include <sys/mman.h>
-#endif
 
 static uint decode_bits;
 static char **default_argv;
diff --git a/storage/sequence/mysql-test/sequence/group_by.result b/storage/sequence/mysql-test/sequence/group_by.result
index 2941e85c1f4..7902fe6a348 100644
--- a/storage/sequence/mysql-test/sequence/group_by.result
+++ b/storage/sequence/mysql-test/sequence/group_by.result
@@ -97,3 +97,9 @@ count(NULL)
 select count(NULL) from seq_1_to_3;
 count(NULL)
 0
+#
+# MDEV-20753: Sequence with limit 0 crashes server
+#
+select count(NULL) from seq_1_to_3 limit 0;
+count(NULL)
+# End of 10.3 tests
diff --git a/storage/sequence/mysql-test/sequence/group_by.test b/storage/sequence/mysql-test/sequence/group_by.test
index b3f4ab58771..18e44cd2ab1 100644
--- a/storage/sequence/mysql-test/sequence/group_by.test
+++ b/storage/sequence/mysql-test/sequence/group_by.test
@@ -49,3 +49,10 @@ explain select count(*) from seq_1_to_15_step_2 group by mod(seq,2);
 create temporary table t1 select * from seq_1_to_3;
 select count(NULL) from t1;
 select count(NULL) from seq_1_to_3;
+
+--echo #
+--echo # MDEV-20753: Sequence with limit 0 crashes server
+--echo #
+select count(NULL) from seq_1_to_3 limit 0;
+
+--echo # End of 10.3 tests
diff --git a/storage/spider/mysql-test/spider/bugfix/include/select_by_null_deinit.inc b/storage/spider/mysql-test/spider/bugfix/include/select_by_null_deinit.inc
new file mode 100644
index 00000000000..76b7582abfe
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/include/select_by_null_deinit.inc
@@ -0,0 +1,11 @@
+--let $MASTER_1_COMMENT_2_1= $MASTER_1_COMMENT_2_1_BACKUP
+--let $CHILD2_1_DROP_TABLES= $CHILD2_1_DROP_TABLES_BACKUP
+--let $CHILD2_1_CREATE_TABLES= $CHILD2_1_CREATE_TABLES_BACKUP
+--let $CHILD2_1_SELECT_TABLES= $CHILD2_1_SELECT_TABLES_BACKUP
+--disable_warnings
+--disable_query_log
+--disable_result_log
+--source ../t/test_deinit.inc
+--enable_result_log
+--enable_query_log
+--enable_warnings
diff --git a/storage/spider/mysql-test/spider/bugfix/include/select_by_null_init.inc b/storage/spider/mysql-test/spider/bugfix/include/select_by_null_init.inc
new file mode 100644
index 00000000000..d07ba2a9fee
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/include/select_by_null_init.inc
@@ -0,0 +1,24 @@
+--disable_warnings
+--disable_query_log
+--disable_result_log
+--source ../t/test_init.inc
+--enable_result_log
+--enable_query_log
+--enable_warnings
+--let $MASTER_1_COMMENT_2_1_BACKUP= $MASTER_1_COMMENT_2_1
+let $MASTER_1_COMMENT_2_1=
+  COMMENT='table "tbl_a", srv "s_2_1"';
+--let $CHILD2_1_DROP_TABLES_BACKUP= $CHILD2_1_DROP_TABLES
+let $CHILD2_1_DROP_TABLES=
+  DROP TABLE IF EXISTS tbl_a;
+--let $CHILD2_1_CREATE_TABLES_BACKUP= $CHILD2_1_CREATE_TABLES
+let $CHILD2_1_CREATE_TABLES=
+  CREATE TABLE tbl_a (
+    pkey int NOT NULL,
+    PRIMARY KEY (pkey)
+  ) $CHILD2_1_ENGINE $CHILD2_1_CHARSET;
+--let $CHILD2_1_SELECT_TABLES_BACKUP= $CHILD2_1_SELECT_TABLES
+let $CHILD2_1_SELECT_TABLES=
+  SELECT pkey FROM tbl_a ORDER BY pkey;
+let $CHILD2_1_SELECT_ARGUMENT1=
+  SELECT argument FROM mysql.general_log WHERE argument LIKE '%select %';
diff --git a/storage/spider/mysql-test/spider/bugfix/r/select_by_null.result b/storage/spider/mysql-test/spider/bugfix/r/select_by_null.result
new file mode 100644
index 00000000000..a7fa1b3b91f
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/r/select_by_null.result
@@ -0,0 +1,90 @@
+for master_1
+for child2
+child2_1
+child2_2
+child2_3
+for child3
+
+this test is for MDEV-16279
+
+drop and create databases
+connection master_1;
+CREATE DATABASE auto_test_local;
+USE auto_test_local;
+connection child2_1;
+SET @old_log_output = @@global.log_output;
+SET GLOBAL log_output = 'TABLE,FILE';
+CREATE DATABASE auto_test_remote;
+USE auto_test_remote;
+
+create table and insert
+connection child2_1;
+CHILD2_1_CREATE_TABLES
+TRUNCATE TABLE mysql.general_log;
+connection master_1;
+CREATE TABLE tbl_a (
+pkey int NOT NULL,
+PRIMARY KEY (pkey)
+) MASTER_1_ENGINE MASTER_1_CHARSET MASTER_1_COMMENT_2_1
+INSERT INTO tbl_a (pkey) VALUES (0),(1),(2),(3),(4),(5),(6),(7),(8),(9);
+INSERT INTO tbl_a (pkey) VALUES (10),(11),(12),(13),(14),(15),(16),(17),(18),(19);
+INSERT INTO tbl_a (pkey) VALUES (20),(21),(22),(23),(24),(25),(26),(27),(28),(29);
+FLUSH TABLES;
+
+select test 1
+connection child2_1;
+TRUNCATE TABLE mysql.general_log;
+connection master_1;
+SELECT pkey FROM tbl_a WHERE NULL;
+pkey
+connection child2_1;
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%select %';
+argument
+SELECT argument FROM mysql.general_log WHERE argument LIKE '%select %'
+SELECT pkey FROM tbl_a ORDER BY pkey;
+pkey
+0
+1
+2
+3
+4
+5
+6
+7
+8
+9
+10
+11
+12
+13
+14
+15
+16
+17
+18
+19
+20
+21
+22
+23
+24
+25
+26
+27
+28
+29
+
+deinit
+connection master_1;
+DROP DATABASE IF EXISTS auto_test_local;
+connection child2_1;
+DROP DATABASE IF EXISTS auto_test_remote;
+SET GLOBAL log_output = @old_log_output;
+for master_1
+for child2
+child2_1
+child2_2
+child2_3
+for child3
+
+end of test
diff --git a/storage/spider/mysql-test/spider/bugfix/t/select_by_null.cnf b/storage/spider/mysql-test/spider/bugfix/t/select_by_null.cnf
new file mode 100644
index 00000000000..05dfd8a0bce
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/t/select_by_null.cnf
@@ -0,0 +1,3 @@
+!include include/default_mysqld.cnf
+!include ../my_1_1.cnf
+!include ../my_2_1.cnf
diff --git a/storage/spider/mysql-test/spider/bugfix/t/select_by_null.test b/storage/spider/mysql-test/spider/bugfix/t/select_by_null.test
new file mode 100644
index 00000000000..1f540b2a15f
--- /dev/null
+++ b/storage/spider/mysql-test/spider/bugfix/t/select_by_null.test
@@ -0,0 +1,72 @@
+--source ../include/select_by_null_init.inc
+--echo
+--echo this test is for MDEV-16279
+--echo
+--echo drop and create databases
+
+--connection master_1
+--disable_warnings
+CREATE DATABASE auto_test_local;
+USE auto_test_local;
+
+--connection child2_1
+SET @old_log_output = @@global.log_output;
+SET GLOBAL log_output = 'TABLE,FILE';
+CREATE DATABASE auto_test_remote;
+USE auto_test_remote;
+--enable_warnings
+
+--echo
+--echo create table and insert
+
+--connection child2_1
+--disable_query_log
+echo CHILD2_1_CREATE_TABLES;
+eval $CHILD2_1_CREATE_TABLES;
+--enable_query_log
+TRUNCATE TABLE mysql.general_log;
+
+--connection master_1
+--disable_query_log
+echo CREATE TABLE tbl_a (
+    pkey int NOT NULL,
+    PRIMARY KEY (pkey)
+) MASTER_1_ENGINE MASTER_1_CHARSET MASTER_1_COMMENT_2_1;
+eval CREATE TABLE tbl_a (
+    pkey int NOT NULL,
+    PRIMARY KEY (pkey)
+) $MASTER_1_ENGINE $MASTER_1_CHARSET $MASTER_1_COMMENT_2_1;
+--enable_query_log
+INSERT INTO tbl_a (pkey) VALUES (0),(1),(2),(3),(4),(5),(6),(7),(8),(9);
+INSERT INTO tbl_a (pkey) VALUES (10),(11),(12),(13),(14),(15),(16),(17),(18),(19);
+INSERT INTO tbl_a (pkey) VALUES (20),(21),(22),(23),(24),(25),(26),(27),(28),(29);
+FLUSH TABLES;
+
+--echo
+--echo select test 1
+
+--connection child2_1
+TRUNCATE TABLE mysql.general_log;
+
+--connection master_1
+SELECT pkey FROM tbl_a WHERE NULL;
+
+--connection child2_1
+eval $CHILD2_1_SELECT_ARGUMENT1;
+eval $CHILD2_1_SELECT_TABLES;
+
+--echo
+--echo deinit
+--disable_warnings
+
+--connection master_1
+DROP DATABASE IF EXISTS auto_test_local;
+
+--connection child2_1
+DROP DATABASE IF EXISTS auto_test_remote;
+SET GLOBAL log_output = @old_log_output;
+
+--enable_warnings
+--source ../include/select_by_null_deinit.inc
+--echo
+--echo end of test