Merge 10.4 into 10.5

123 files changed, 2481 insertions, 450 deletions

diff --git a/extra/mariabackup/backup_copy.cc b/extra/mariabackup/backup_copy.cc
index 711f4aa666f..1972bceacb7 100644
--- a/extra/mariabackup/backup_copy.cc
+++ b/extra/mariabackup/backup_copy.cc
@@ -863,21 +863,14 @@ datafile_rsync_backup(const char *filepath, bool save_to_list, FILE *f)
 	return(true);
 }
 
-
-static
-bool
-backup_file_vprintf(const char *filename, const char *fmt, va_list ap)
+bool backup_file_print_buf(const char *filename, const char *buf, int buf_len)
 {
 	ds_file_t	*dstfile	= NULL;
 	MY_STAT		 stat;			/* unused for now */
-	char		*buf		= 0;
-	int		 buf_len;
 	const char	*action;
 
 	memset(&stat, 0, sizeof(stat));
 
-	buf_len = vasprintf(&buf, fmt, ap);
-
 	stat.st_size = buf_len;
 	stat.st_mtime = my_time(0);
 
@@ -901,7 +894,6 @@ backup_file_vprintf(const char *filename, const char *fmt, va_list ap)
 
 	/* close */
 	msg("        ...done");
-	free(buf);
 
 	if (ds_close(dstfile)) {
 		goto error_close;
@@ -910,7 +902,6 @@ backup_file_vprintf(const char *filename, const char *fmt, va_list ap)
 	return(true);
 
 error:
-	free(buf);
 	if (dstfile != NULL) {
 		ds_close(dstfile);
 	}
@@ -918,8 +909,21 @@ error:
 error_close:
 	msg("Error: backup file failed.");
 	return(false); /*ERROR*/
-}
 
+	return true;
+};
+
+static
+bool
+backup_file_vprintf(const char *filename, const char *fmt, va_list ap)
+{
+	char		*buf		= 0;
+	int		 buf_len;
+	buf_len = vasprintf(&buf, fmt, ap);
+	bool result = backup_file_print_buf(filename, buf, buf_len);
+	free(buf);
+	return result;
+}
 
 bool
 backup_file_printf(const char *filename, const char *fmt, ...)
@@ -1441,7 +1445,7 @@ out:
 	return(ret);
 }
 
-void backup_fix_ddl(void);
+void backup_fix_ddl(CorruptedPages &);
 
 lsn_t get_current_lsn(MYSQL *connection)
 {
@@ -1466,7 +1470,7 @@ lsn_t get_current_lsn(MYSQL *connection)
 lsn_t server_lsn_after_lock;
 extern void backup_wait_for_lsn(lsn_t lsn);
 /** Start --backup */
-bool backup_start()
+bool backup_start(CorruptedPages &corrupted_pages)
 {
 	if (!opt_no_lock) {
 		if (opt_safe_slave_backup) {
@@ -1501,7 +1505,7 @@ bool backup_start()
 
 	msg("Waiting for log copy thread to read lsn %llu", (ulonglong)server_lsn_after_lock);
 	backup_wait_for_lsn(server_lsn_after_lock);
-	backup_fix_ddl();
+	backup_fix_ddl(corrupted_pages);
 
 	// There is no need to stop slave thread before coping non-Innodb data when
 	// --no-lock option is used because --no-lock option requires that no DDL or
diff --git a/extra/mariabackup/backup_copy.h b/extra/mariabackup/backup_copy.h
index 7c886719f37..62b2b1bc232 100644
--- a/extra/mariabackup/backup_copy.h
+++ b/extra/mariabackup/backup_copy.h
@@ -33,7 +33,7 @@ copy_file(ds_ctxt_t *datasink,
 	  uint thread_n);
 
 /** Start --backup */
-bool backup_start();
+bool backup_start(CorruptedPages &corrupted_pages);
 /** Release resources after backup_start() */
 void backup_release();
 /** Finish after backup_start() and backup_release() */
@@ -51,5 +51,6 @@ directory_exists(const char *dir, bool create);
 
 lsn_t
 get_current_lsn(MYSQL *connection);
+bool backup_file_print_buf(const char *filename, const char *buf, int buf_len);
 
 #endif
diff --git a/extra/mariabackup/backup_debug.h b/extra/mariabackup/backup_debug.h
new file mode 100644
index 00000000000..cefbc287361
--- /dev/null
+++ b/extra/mariabackup/backup_debug.h
@@ -0,0 +1,32 @@
+#pragma once
+#include "my_dbug.h"
+#ifndef DBUG_OFF
+extern char *dbug_mariabackup_get_val(const char *event, const char *key);
+/*
+In debug mode,  execute SQL statement that was passed via environment.
+To use this facility, you need to
+
+1. Add code DBUG_EXECUTE_MARIABACKUP_EVENT("my_event_name", key););
+  to the code. key is usually a table name
+2. Set environment variable my_event_name_$key SQL statement you want to execute
+   when event occurs, in DBUG_EXECUTE_IF from above.
+   In mtr , you can set environment via 'let' statement (do not use $ as the first char
+   for the variable)
+3. start mariabackup with --dbug=+d,debug_mariabackup_events
+*/
+extern void dbug_mariabackup_event(
+	const char *event,const char *key);
+#define DBUG_MARIABACKUP_EVENT(A, B) \
+	DBUG_EXECUTE_IF("mariabackup_events", \
+		dbug_mariabackup_event(A,B););
+#define DBUG_EXECUTE_FOR_KEY(EVENT, KEY, CODE) \
+	DBUG_EXECUTE_IF("mariabackup_inject_code", {\
+		char *dbug_val = dbug_mariabackup_get_val(EVENT, KEY); \
+		if (dbug_val && *dbug_val) CODE \
+	})
+#else
+#define DBUG_MARIABACKUP_EVENT(A,B)
+#define DBUG_MARIABACKUP_EVENT_LOCK(A,B)
+#define DBUG_EXECUTE_FOR_KEY(EVENT, KEY, CODE)
+#endif
+
diff --git a/extra/mariabackup/encryption_plugin.cc b/extra/mariabackup/encryption_plugin.cc
index a3242078293..dbaa67e1324 100644
--- a/extra/mariabackup/encryption_plugin.cc
+++ b/extra/mariabackup/encryption_plugin.cc
@@ -18,7 +18,6 @@
 #include <mysql.h>
 #include <xtrabackup.h>
 #include <encryption_plugin.h>
-#include <backup_copy.h>
 #include <sql_plugin.h>
 #include <sstream>
 #include <vector>
diff --git a/extra/mariabackup/fil_cur.cc b/extra/mariabackup/fil_cur.cc
index 4f9e493b347..1364b337bec 100644
--- a/extra/mariabackup/fil_cur.cc
+++ b/extra/mariabackup/fil_cur.cc
@@ -35,6 +35,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include "common.h"
 #include "read_filt.h"
 #include "xtrabackup.h"
+#include "backup_debug.h"
 
 /* Size of read buffer in pages (640 pages = 10M for 16K sized pages) */
 #define XB_FIL_CUR_PAGES 640
@@ -351,19 +352,18 @@ static bool page_is_corrupted(const byte *page, ulint page_no,
 	return buf_page_is_corrupted(true, page, space->flags);
 }
 
-/************************************************************************
-Reads and verifies the next block of pages from the source
+/** Reads and verifies the next block of pages from the source
 file. Positions the cursor after the last read non-corrupted page.
-
+@param[in,out] cursor source file cursor
+@param[out] corrupted_pages adds corrupted pages if
+opt_log_innodb_page_corruption is set
 @return XB_FIL_CUR_SUCCESS if some have been read successfully, XB_FIL_CUR_EOF
 if there are no more pages to read and XB_FIL_CUR_ERROR on error. */
-xb_fil_cur_result_t
-xb_fil_cur_read(
-/*============*/
-	xb_fil_cur_t*	cursor)	/*!< in/out: source file cursor */
+xb_fil_cur_result_t xb_fil_cur_read(xb_fil_cur_t*	cursor,
+                                    CorruptedPages &corrupted_pages)
 {
 	byte*			page;
-	ulint			i;
+	unsigned			i;
 	ulint			npages;
 	ulint			retry_count;
 	xb_fil_cur_result_t	ret;
@@ -417,7 +417,7 @@ read_retry:
 	cursor->buf_read = 0;
 	cursor->buf_npages = 0;
 	cursor->buf_offset = offset;
-	cursor->buf_page_no = (ulint)(offset / page_size);
+	cursor->buf_page_no = static_cast<unsigned>(offset / page_size);
 
 	if (os_file_read(IORequestRead, cursor->file, cursor->buf, offset,
 			  (ulint) to_read) != DB_SUCCESS) {
@@ -428,26 +428,47 @@ read_retry:
 	partially written pages */
 	for (page = cursor->buf, i = 0; i < npages;
 	     page += page_size, i++) {
-		ulint page_no = cursor->buf_page_no + i;
+		unsigned page_no = cursor->buf_page_no + i;
 
 		if (page_is_corrupted(page, page_no, cursor, space)){
 			retry_count--;
 
 			if (retry_count == 0) {
+				const char *ignore_corruption_warn = opt_log_innodb_page_corruption ?
+					" WARNING!!! The corruption is ignored due to"
+					" log-innodb-page-corruption option, the backup can contain"
+					" corrupted data." : "";
 				msg(cursor->thread_n,
 				    "Error: failed to read page after "
 				    "10 retries. File %s seems to be "
-				    "corrupted.", cursor->abs_path);
-				ret = XB_FIL_CUR_ERROR;
+				    "corrupted.%s", cursor->abs_path, ignore_corruption_warn);
 				ut_print_buf(stderr, page, page_size);
-				break;
+				if (opt_log_innodb_page_corruption) {
+					corrupted_pages.add_page(cursor->node->name, cursor->node->space->id,
+						page_no);
+					retry_count = 1;
+				}
+				else {
+					ret = XB_FIL_CUR_ERROR;
+					break;
+				}
+			}
+			else {
+				msg(cursor->thread_n, "Database page corruption detected at page "
+				    UINT32PF ", retrying...",
+				    page_no);
+				os_thread_sleep(100000);
+				goto read_retry;
 			}
-			msg(cursor->thread_n, "Database page corruption detected at page "
-			    ULINTPF ", retrying...", 
-			    page_no);
-			os_thread_sleep(100000);
-			goto read_retry;
 		}
+		DBUG_EXECUTE_FOR_KEY("add_corrupted_page_for", cursor->node->space->name,
+			{
+				unsigned corrupted_page_no =
+					static_cast<unsigned>(strtoul(dbug_val, NULL, 10));
+				if (page_no == corrupted_page_no)
+					corrupted_pages.add_page(cursor->node->name, cursor->node->space->id,
+						corrupted_page_no);
+			});
 		cursor->buf_read += page_size;
 		cursor->buf_npages++;
 	}
diff --git a/extra/mariabackup/fil_cur.h b/extra/mariabackup/fil_cur.h
index 70e4888ba63..0027b7768e9 100644
--- a/extra/mariabackup/fil_cur.h
+++ b/extra/mariabackup/fil_cur.h
@@ -29,6 +29,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include "read_filt.h"
 #include "srv0start.h"
 #include "srv0srv.h"
+#include "xtrabackup.h"
 
 struct xb_fil_cur_t {
 	pfs_os_file_t	file;		/*!< source file handle */
@@ -52,7 +53,7 @@ struct xb_fil_cur_t {
 					last cursor read */
 	ib_int64_t	buf_offset;	/*!< file offset of the first page in
 					buffer */
-	ulint		buf_page_no;	/*!< number of the first page in
+	unsigned		buf_page_no;	/*!< number of the first page in
 					buffer */
 	uint		thread_n;	/*!< thread number for diagnostics */
 	ulint		space_id;	/*!< ID of tablespace */
@@ -88,17 +89,15 @@ xb_fil_cur_open(
 	uint		thread_n,	/*!< thread number for diagnostics */
 	ulonglong max_file_size = ULLONG_MAX);
 
-/************************************************************************
-Reads and verifies the next block of pages from the source
+/** Reads and verifies the next block of pages from the source
 file. Positions the cursor after the last read non-corrupted page.
-
+@param[in,out] cursor source file cursor
+@param[out] corrupted_pages adds corrupted pages if
+opt_log_innodb_page_corruption is set
 @return XB_FIL_CUR_SUCCESS if some have been read successfully, XB_FIL_CUR_EOF
 if there are no more pages to read and XB_FIL_CUR_ERROR on error. */
-xb_fil_cur_result_t
-xb_fil_cur_read(
-/*============*/
-	xb_fil_cur_t*	cursor);	/*!< in/out: source file cursor */
-
+xb_fil_cur_result_t xb_fil_cur_read(xb_fil_cur_t *cursor,
+                                    CorruptedPages &corrupted_pages);
 /************************************************************************
 Close the source file cursor opened with xb_fil_cur_open() and its
 associated read filter. */
diff --git a/extra/mariabackup/write_filt.cc b/extra/mariabackup/write_filt.cc
index d8910699d16..8339286e1df 100644
--- a/extra/mariabackup/write_filt.cc
+++ b/extra/mariabackup/write_filt.cc
@@ -32,7 +32,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 /************************************************************************
 Write-through page write filter. */
 static my_bool wf_wt_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
-			  xb_fil_cur_t *cursor);
+			  xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages);
 static my_bool wf_wt_process(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile);
 
 xb_write_filt_t wf_write_through = {
@@ -45,7 +45,7 @@ xb_write_filt_t wf_write_through = {
 /************************************************************************
 Incremental page write filter. */
 static my_bool wf_incremental_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
-				   xb_fil_cur_t *cursor);
+				   xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages);
 static my_bool wf_incremental_process(xb_write_filt_ctxt_t *ctxt,
 				      ds_file_t *dstfile);
 static my_bool wf_incremental_finalize(xb_write_filt_ctxt_t *ctxt,
@@ -65,11 +65,11 @@ Initialize incremental page write filter.
 @return TRUE on success, FALSE on error. */
 static my_bool
 wf_incremental_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
-		    xb_fil_cur_t *cursor)
+		    xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages)
 {
 	char				meta_name[FN_REFLEN];
 	xb_wf_incremental_ctxt_t	*cp =
-		&(ctxt->u.wf_incremental_ctxt);
+		&(ctxt->wf_incremental_ctxt);
 
 	ctxt->cursor = cursor;
 
@@ -100,7 +100,9 @@ wf_incremental_init(xb_write_filt_ctxt_t *ctxt, char *dst_name,
 	strcat(dst_name, ".delta");
 
 	mach_write_to_4(cp->delta_buf, 0x78747261UL); /*"xtra"*/
+
 	cp->npages = 1;
+	cp->corrupted_pages = corrupted_pages;
 
 	return(TRUE);
 }
@@ -112,19 +114,20 @@ Run the next batch of pages through incremental page write filter.
 static my_bool
 wf_incremental_process(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile)
 {
-	ulint				i;
+	unsigned				i;
 	xb_fil_cur_t			*cursor = ctxt->cursor;
 	byte				*page;
 	const ulint			page_size = cursor->page_size;
-	xb_wf_incremental_ctxt_t	*cp = &(ctxt->u.wf_incremental_ctxt);
+	xb_wf_incremental_ctxt_t	*cp = &(ctxt->wf_incremental_ctxt);
 
 	for (i = 0, page = cursor->buf; i < cursor->buf_npages;
 	     i++, page += page_size) {
 
-		if (incremental_lsn >= mach_read_from_8(page + FIL_PAGE_LSN)) {
-
+		if ((!cp->corrupted_pages ||
+				!cp->corrupted_pages->contains(cursor->node->space->id,
+					cursor->buf_page_no + i)) &&
+				incremental_lsn >= mach_read_from_8(page + FIL_PAGE_LSN))
 			continue;
-		}
 
 		/* updated page */
 		if (cp->npages == page_size / 4) {
@@ -161,7 +164,7 @@ wf_incremental_finalize(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile)
 {
 	xb_fil_cur_t			*cursor = ctxt->cursor;
 	const ulint			page_size = cursor->page_size;
-	xb_wf_incremental_ctxt_t	*cp = &(ctxt->u.wf_incremental_ctxt);
+	xb_wf_incremental_ctxt_t	*cp = &(ctxt->wf_incremental_ctxt);
 
 	if (cp->npages != page_size / 4) {
 		mach_write_to_4(cp->delta_buf + cp->npages * 4, 0xFFFFFFFFUL);
@@ -183,7 +186,7 @@ Free the incremental page write filter's buffer. */
 static void
 wf_incremental_deinit(xb_write_filt_ctxt_t *ctxt)
 {
-	xb_wf_incremental_ctxt_t	*cp = &(ctxt->u.wf_incremental_ctxt);
+	xb_wf_incremental_ctxt_t	*cp = &(ctxt->wf_incremental_ctxt);
 	my_large_free(cp->delta_buf, cp->delta_buf_size);
 }
 
@@ -193,7 +196,7 @@ Initialize the write-through page write filter.
 @return TRUE on success, FALSE on error. */
 static my_bool
 wf_wt_init(xb_write_filt_ctxt_t *ctxt, char *dst_name __attribute__((unused)),
-	   xb_fil_cur_t *cursor)
+	   xb_fil_cur_t *cursor, CorruptedPages *)
 {
 	ctxt->cursor = cursor;
 
diff --git a/extra/mariabackup/write_filt.h b/extra/mariabackup/write_filt.h
index febf25f2a8a..6c3ef24291f 100644
--- a/extra/mariabackup/write_filt.h
+++ b/extra/mariabackup/write_filt.h
@@ -27,26 +27,26 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 
 #include "fil_cur.h"
 #include "datasink.h"
+#include "xtrabackup.h"
 
 /* Incremental page filter context */
 typedef struct {
 	ulint		delta_buf_size;
 	byte		*delta_buf;
 	ulint		 npages;
+	CorruptedPages *corrupted_pages;
 } xb_wf_incremental_ctxt_t;
 
 /* Page filter context used as an opaque structure by callers */
 typedef struct {
 	xb_fil_cur_t	*cursor;
-	union {
-		xb_wf_incremental_ctxt_t	wf_incremental_ctxt;
-	} u;
+	xb_wf_incremental_ctxt_t	wf_incremental_ctxt;
 } xb_write_filt_ctxt_t;
 
 
 typedef struct {
 	my_bool	(*init)(xb_write_filt_ctxt_t *ctxt, char *dst_name,
-			xb_fil_cur_t *cursor);
+			xb_fil_cur_t *cursor, CorruptedPages *corrupted_pages);
 	my_bool	(*process)(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile);
 	my_bool	(*finalize)(xb_write_filt_ctxt_t *, ds_file_t *dstfile);
 	void (*deinit)(xb_write_filt_ctxt_t *);
diff --git a/extra/mariabackup/xtrabackup.cc b/extra/mariabackup/xtrabackup.cc
index be865b0fea0..d44e212d58a 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -77,6 +77,7 @@ Street, Fifth Floor, Boston, MA 02110-1335 USA
 #include <list>
 #include <sstream>
 #include <set>
+#include <fstream>
 #include <mysql.h>
 
 #define G_PTR uchar*
@@ -104,6 +105,9 @@ Street, Fifth Floor, Boston, MA 02110-1335 USA
 #include <log.h>
 #include <derror.h>
 #include <thr_timer.h>
+#include "backup_debug.h"
+
+#define MB_CORRUPTED_PAGES_FILE "innodb_corrupted_pages"
 
 int sys_var_init();
 
@@ -287,6 +291,7 @@ my_bool opt_noversioncheck = FALSE;
 my_bool opt_no_backup_locks = FALSE;
 my_bool opt_decompress = FALSE;
 my_bool opt_remove_original;
+my_bool opt_log_innodb_page_corruption;
 
 my_bool opt_lock_ddl_per_table = FALSE;
 static my_bool opt_check_privileges;
@@ -349,6 +354,212 @@ struct ddl_tracker_t {
 
 static ddl_tracker_t ddl_tracker;
 
+// Convert non-null terminated filename to space name
+std::string filename_to_spacename(const byte *filename, size_t len);
+
+CorruptedPages::CorruptedPages() { ut_a(!pthread_mutex_init(&m_mutex, NULL)); }
+
+CorruptedPages::~CorruptedPages() { ut_a(!pthread_mutex_destroy(&m_mutex)); }
+
+void CorruptedPages::add_page_no_lock(const char *space_name, ulint space_id,
+                                      unsigned page_no,
+                                      bool convert_space_name)
+{
+  space_info_t  &space_info = m_spaces[space_id];
+  if (space_info.space_name.empty())
+    space_info.space_name=
+        convert_space_name
+            ? filename_to_spacename(reinterpret_cast<const byte *>(space_name),
+                                    strlen(space_name))
+            : space_name;
+  (void)space_info.pages.insert(page_no);
+}
+
+void CorruptedPages::add_page(const char *file_name, ulint space_id,
+                              unsigned page_no)
+{
+  ut_a(!pthread_mutex_lock(&m_mutex));
+  add_page_no_lock(file_name, space_id, page_no, true);
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+}
+
+bool CorruptedPages::contains(ulint space_id, unsigned page_no) const
+{
+  bool result = false;
+  ut_a(!pthread_mutex_lock(&m_mutex));
+  container_t::const_iterator space_it= m_spaces.find(space_id);
+  if (space_it != m_spaces.end())
+    result = space_it->second.pages.count(page_no);
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+  return result;
+}
+
+void CorruptedPages::drop_space(ulint space_id)
+{
+  ut_a(!pthread_mutex_lock(&m_mutex));
+  m_spaces.erase(space_id);
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+}
+
+void CorruptedPages::rename_space(ulint space_id, const std::string &new_name)
+{
+  ut_a(!pthread_mutex_lock(&m_mutex));
+  container_t::iterator space_it = m_spaces.find(space_id);
+  if (space_it != m_spaces.end())
+    space_it->second.space_name = new_name;
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+}
+
+bool CorruptedPages::print_to_file(const char *filename) const
+{
+  std::ostringstream out;
+  ut_a(!pthread_mutex_lock(&m_mutex));
+  if (!m_spaces.size())
+  {
+    ut_a(!pthread_mutex_unlock(&m_mutex));
+    return true;
+  }
+  for (container_t::const_iterator space_it=
+           m_spaces.begin();
+       space_it != m_spaces.end(); ++space_it)
+  {
+    out << space_it->second.space_name << " " << space_it->first << "\n";
+    bool first_page_no= true;
+    for (std::set<unsigned>::const_iterator page_it=
+             space_it->second.pages.begin();
+         page_it != space_it->second.pages.end(); ++page_it)
+      if (first_page_no)
+      {
+        out << *page_it;
+        first_page_no= false;
+      }
+      else
+        out << " " << *page_it;
+    out << "\n";
+  }
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+  if (xtrabackup_backup)
+    return backup_file_print_buf(filename, out.str().c_str(),
+                                 static_cast<int>(out.str().size()));
+  std::ofstream outfile;
+  outfile.open(filename);
+  if (!outfile.is_open())
+    die("Can't open %s, error number: %d, error message: %s", filename, errno,
+        strerror(errno));
+  outfile << out.str();
+  return true;
+}
+
+void CorruptedPages::read_from_file(const char *file_name)
+{
+  MY_STAT mystat;
+  if (!my_stat(file_name, &mystat, MYF(0)))
+    return;
+  std::ifstream infile;
+  infile.open(file_name);
+  if (!infile.is_open())
+    die("Can't open %s, error number: %d, error message: %s", file_name, errno,
+        strerror(errno));
+  std::string line;
+  std::string space_name;
+  ulint space_id;
+  ulint line_number= 0;
+  while (std::getline(infile, line))
+  {
+    ++line_number;
+    std::istringstream iss(line);
+    if (line_number & 1) {
+      if (!(iss >> space_name))
+        die("Can't parse space name from corrupted pages file at "
+            "line " ULINTPF,
+            line_number);
+      if (!(iss >> space_id))
+        die("Can't parse space id from corrupted pages file at line " ULINTPF,
+            line_number);
+    }
+    else
+    {
+      std::istringstream iss(line);
+      unsigned page_no;
+      while ((iss >> page_no))
+        add_page_no_lock(space_name.c_str(), space_id, page_no, false);
+      if (!iss.eof())
+        die("Corrupted pages file parse error on line number " ULINTPF,
+            line_number);
+    }
+  }
+}
+
+bool CorruptedPages::empty() const
+{
+  ut_a(!pthread_mutex_lock(&m_mutex));
+  bool result= !m_spaces.size();
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+  return result;
+}
+
+static void xb_load_single_table_tablespace(const std::string &space_name,
+                                            bool set_size);
+static void xb_data_files_close();
+static fil_space_t* fil_space_get_by_name(const char* name);
+
+void CorruptedPages::zero_out_free_pages()
+{
+  container_t non_free_pages;
+  byte *zero_page=
+      static_cast<byte *>(aligned_malloc(srv_page_size, srv_page_size));
+  memset(zero_page, 0, srv_page_size);
+
+  ut_a(!pthread_mutex_lock(&m_mutex));
+  for (container_t::const_iterator space_it= m_spaces.begin();
+       space_it != m_spaces.end(); ++space_it)
+  {
+    ulint space_id = space_it->first;
+    const std::string &space_name = space_it->second.space_name;
+    // There is no need to close tablespaces explixitly as they will be closed
+    // in innodb_shutdown().
+    xb_load_single_table_tablespace(space_name, false);
+    fil_space_t *space = fil_space_t::get(space_id);
+    if (!space)
+      die("Can't find space object for space name %s to check corrupted page",
+          space_name.c_str());
+    for (std::set<unsigned>::const_iterator page_it=
+             space_it->second.pages.begin();
+         page_it != space_it->second.pages.end(); ++page_it)
+    {
+      bool is_free= fseg_page_is_free(space, *page_it);
+      if (!is_free) {
+        space_info_t &space_info = non_free_pages[space_id];
+        space_info.pages.insert(*page_it);
+        if (space_info.space_name.empty())
+          space_info.space_name = space_name;
+        msg("Error: corrupted page " UINT32PF
+            " of tablespace %s can not be fixed",
+            *page_it, space_name.c_str());
+      }
+      else
+      {
+        space->reacquire();
+        auto err= space
+                      ->io(IORequest(IORequest::PUNCH_RANGE),
+                           *page_it * srv_page_size, srv_page_size, zero_page)
+                      .err;
+        if (err != DB_SUCCESS)
+          die("Can't zero out corrupted page " UINT32PF " of tablespace %s",
+              *page_it, space_name.c_str());
+        msg("Corrupted page " UINT32PF
+            " of tablespace %s was successfuly fixed.",
+            *page_it, space_name.c_str());
+      }
+    }
+    space->flush();
+    space->release();
+  }
+  m_spaces.swap(non_free_pages);
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+  aligned_free(zero_page);
+}
+
 /* Simple datasink creation tracking...add datasinks in the reverse order you
 want them destroyed. */
 #define XTRABACKUP_MAX_DATASINKS	10
@@ -362,11 +573,12 @@ xtrabackup_add_datasink(ds_ctxt_t *ds)
 	datasinks[actual_datasinks] = ds; actual_datasinks++;
 }
 
-
-typedef void (*process_single_tablespace_func_t)(const char *dirname, const char *filname, bool is_remote);
+typedef void (*process_single_tablespace_func_t)(const char *dirname,
+                                                 const char *filname,
+                                                 bool is_remote,
+                                                 bool skip_node_page0);
 static dberr_t enumerate_ibd_files(process_single_tablespace_func_t callback);
 
-
 /* ======== Datafiles iterator ======== */
 struct datafiles_iter_t {
 	fil_space_t	*space;
@@ -690,6 +902,7 @@ typedef struct {
 	uint			*count;
 	pthread_mutex_t*	count_mutex;
 	os_thread_id_t		id;
+	CorruptedPages *corrupted_pages;
 } data_thread_ctxt_t;
 
 /* ======== for option and variables ======== */
@@ -791,7 +1004,8 @@ enum options_xtrabackup
   OPT_ROCKSDB_DATADIR,
   OPT_BACKUP_ROCKSDB,
   OPT_XTRA_CHECK_PRIVILEGES,
-  OPT_XTRA_MYSQLD_ARGS
+  OPT_XTRA_MYSQLD_ARGS,
+  OPT_XB_IGNORE_INNODB_PAGE_CORRUPTION
 };
 
 struct my_option xb_client_options[]= {
@@ -1182,6 +1396,17 @@ struct my_option xb_client_options[]= {
      " uses old (pre-4.1.1) protocol.",
      &opt_secure_auth, &opt_secure_auth, 0, GET_BOOL, NO_ARG, 1, 0, 0, 0, 0,
      0},
+
+    {"log-innodb-page-corruption", OPT_XB_IGNORE_INNODB_PAGE_CORRUPTION,
+     "Continue backup if innodb corrupted pages are found. The pages are "
+     "logged in " MB_CORRUPTED_PAGES_FILE
+     " and backup is finished with error. "
+     "--prepare will try to fix corrupted pages. If " MB_CORRUPTED_PAGES_FILE
+     " exists after --prepare in base backup directory, backup still contains "
+     "corrupted pages and can not be considered as consistent.",
+     &opt_log_innodb_page_corruption, &opt_log_innodb_page_corruption, 0,
+     GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+
 #define MYSQL_CLIENT
 #include "sslopt-longopts.h"
 #undef MYSQL_CLIENT
@@ -1474,7 +1699,8 @@ debug_sync_point(const char *name)
 
 static std::set<std::string> tables_for_export;
 
-static void append_export_table(const char *dbname, const char *tablename, bool is_remote)
+static void append_export_table(const char *dbname, const char *tablename,
+                                bool is_remote, bool skip_node_page0)
 {
   if(dbname && tablename && !is_remote)
   {
@@ -2473,7 +2699,8 @@ for full backup, pages filter for incremental backup, etc.
 @return FALSE on success and TRUE on error */
 static my_bool xtrabackup_copy_datafile(fil_node_t *node, uint thread_n,
                                         const char *dest_name,
-                                        const xb_write_filt_t &write_filter)
+                                        const xb_write_filt_t &write_filter,
+                                        CorruptedPages &corrupted_pages)
 {
 	char			 dst_name[FN_REFLEN];
 	ds_file_t		*dstfile = NULL;
@@ -2538,7 +2765,8 @@ static my_bool xtrabackup_copy_datafile(fil_node_t *node, uint thread_n,
 	ut_a(write_filter.process != NULL);
 
 	if (write_filter.init != NULL &&
-	    !write_filter.init(&write_filt_ctxt, dst_name, &cursor)) {
+		!write_filter.init(&write_filt_ctxt, dst_name, &cursor,
+			opt_log_innodb_page_corruption ? &corrupted_pages : NULL)) {
 		msg (thread_n, "mariabackup: error: failed to initialize page write filter.");
 		goto error;
 	}
@@ -2558,7 +2786,8 @@ static my_bool xtrabackup_copy_datafile(fil_node_t *node, uint thread_n,
 	}
 
 	/* The main copy loop */
-	while ((res = xb_fil_cur_read(&cursor)) == XB_FIL_CUR_SUCCESS) {
+	while ((res = xb_fil_cur_read(&cursor, corrupted_pages)) ==
+		XB_FIL_CUR_SUCCESS) {
 		if (!write_filter.process(&write_filt_ctxt, dstfile)) {
 			goto error;
 		}
@@ -2836,6 +3065,21 @@ static os_thread_ret_t DECLARE_THREAD(io_watching_thread)(void*)
 }
 
 #ifndef DBUG_OFF
+char *dbug_mariabackup_get_val(const char *event, const char *key)
+{
+	char envvar[FN_REFLEN];
+	if (key) {
+		snprintf(envvar, sizeof(envvar), "%s_%s", event, key);
+		char *slash = strchr(envvar, '/');
+		if (slash)
+			*slash = '_';
+	} else {
+		strncpy(envvar, event, sizeof envvar - 1);
+		envvar[sizeof envvar - 1] = '\0';
+	}
+	return getenv(envvar);
+}
+
 /*
 In debug mode,  execute SQL statement that was passed via environment.
 To use this facility, you need to
@@ -2848,35 +3092,15 @@ To use this facility, you need to
    for the variable)
 3. start mariabackup with --dbug=+d,debug_mariabackup_events
 */
-static void dbug_mariabackup_event(const char *event,const char *key)
+void dbug_mariabackup_event(const char *event,const char *key)
 {
-	char envvar[FN_REFLEN];
-	if (key) {
-		snprintf(envvar, sizeof(envvar), "%s_%s", event, key);
-		char *slash = strchr(envvar, '/');
-		if (slash)
-			*slash = '_';
-	} else {
-		strncpy(envvar, event, sizeof envvar - 1);
-		envvar[sizeof envvar - 1] = '\0';
-	}
-	char *sql = getenv(envvar);
-	if (sql) {
+	char *sql = dbug_mariabackup_get_val(event, key);
+	if (sql && *sql) {
 		msg("dbug_mariabackup_event : executing '%s'", sql);
 		xb_mysql_query(mysql_connection, sql, false, true);
 	}
-
 }
-#define DBUG_MARIABACKUP_EVENT(A, B) DBUG_EXECUTE_IF("mariabackup_events", dbug_mariabackup_event(A,B););
-#define DBUG_MB_INJECT_CODE(EVENT, KEY, CODE) \
-	DBUG_EXECUTE_IF("mariabackup_inject_code", {\
-		char *env = getenv(EVENT); \
-		if (env && !strcmp(env, KEY)) { CODE } \
-	})
-#else
-#define DBUG_MARIABACKUP_EVENT(A,B)
-#define DBUG_MB_INJECT_CODE(EVENT, KEY, CODE)
-#endif
+#endif // DBUG_OFF
 
 /**************************************************************************
 Datafiles copying thread.*/
@@ -2889,6 +3113,7 @@ DECLARE_THREAD(data_copy_thread_func)(
 	data_thread_ctxt_t	*ctxt = (data_thread_ctxt_t *) arg;
 	uint			num = ctxt->num;
 	fil_node_t*		node;
+	ut_ad(ctxt->corrupted_pages);
 
 	/*
 	  Initialize mysys thread-specific memory so we can
@@ -2900,11 +3125,12 @@ DECLARE_THREAD(data_copy_thread_func)(
 
 	while ((node = datafiles_iter_next(ctxt->it)) != NULL) {
 		DBUG_MARIABACKUP_EVENT("before_copy", node->space->name);
-		DBUG_MB_INJECT_CODE("wait_innodb_redo_before_copy", node->space->name,
+		DBUG_EXECUTE_FOR_KEY("wait_innodb_redo_before_copy", node->space->name,
 			backup_wait_for_lsn(get_current_lsn(mysql_connection)););
 		/* copy the datafile */
 		if (xtrabackup_copy_datafile(node, num, NULL,
-			xtrabackup_incremental ? wf_incremental : wf_write_through))
+			xtrabackup_incremental ? wf_incremental : wf_write_through,
+			*ctxt->corrupted_pages))
 			die("failed to copy datafile.");
 
 		DBUG_MARIABACKUP_EVENT("after_copy", node->space->name);
@@ -3041,15 +3267,24 @@ xb_new_datafile(const char *name, bool is_remote)
 }
 
 
-static
-void
-xb_load_single_table_tablespace(
-	const char *dirname,
-	const char *filname,
-	bool is_remote)
+/** Load tablespace.
+
+@param[in] dirname directory name of the tablespace to open
+@param[in] filname file name of the tablespece to open
+@param[in] is_remote true if tablespace file is .isl
+@param[in] skip_node_page0 true if we don't need to read node page 0. Otherwise
+node page0 will be read, and it's size and free pages limit
+will be set from page 0, what is neccessary for checking and fixing corrupted
+pages.
+*/
+static void xb_load_single_table_tablespace(const char *dirname,
+                                            const char *filname,
+                                            bool is_remote,
+                                            bool skip_node_page0)
 {
 	ut_ad(srv_operation == SRV_OPERATION_BACKUP
-	      || srv_operation == SRV_OPERATION_RESTORE_DELTA);
+	      || srv_operation == SRV_OPERATION_RESTORE_DELTA
+	      || srv_operation == SRV_OPERATION_RESTORE);
 	/* Ignore .isl files on XtraBackup recovery. All tablespaces must be
 	local. */
 	if (is_remote && srv_operation == SRV_OPERATION_RESTORE_DELTA) {
@@ -3102,8 +3337,8 @@ xb_load_single_table_tablespace(
 			FIL_TYPE_TABLESPACE, NULL/* TODO: crypt_data */);
 
 		ut_a(space != NULL);
-
-		space->add(file->filepath(), file->detach(), 0, false, false);
+		space->add(file->filepath(),
+			skip_node_page0 ? file->detach() : pfs_os_file_t(), 0, false, false);
 		mutex_enter(&fil_system.mutex);
 		space->read_page0();
 		mutex_exit(&fil_system.mutex);
@@ -3123,6 +3358,28 @@ xb_load_single_table_tablespace(
 	ut_free(name);
 }
 
+static void xb_load_single_table_tablespace(const std::string &space_name,
+                                            bool skip_node_page0)
+{
+  std::string name(space_name);
+  bool is_remote= access((name + ".ibd").c_str(), R_OK) != 0;
+  const char *extension= is_remote ? ".isl" : ".ibd";
+  name.append(extension);
+  char buf[FN_REFLEN];
+  strncpy(buf, name.c_str(), sizeof buf - 1);
+  buf[sizeof buf - 1]= '\0';
+  const char *dbname= buf;
+  char *p= strchr(buf, '/');
+  if (p == 0)
+    die("Unexpected tablespace %s filename %s", space_name.c_str(),
+        name.c_str());
+  ut_a(p);
+  *p= 0;
+  const char *tablename= p + 1;
+  xb_load_single_table_tablespace(dbname, tablename, is_remote,
+                                  skip_node_page0);
+}
+
 /** Scan the database directories under the MySQL datadir, looking for
 .ibd files and determining the space id in each of them.
 @return	DB_SUCCESS or error number */
@@ -3164,7 +3421,7 @@ static dberr_t enumerate_ibd_files(process_single_tablespace_func_t callback)
 			bool is_ibd = !is_isl && ends_with(dbinfo.name,".ibd");
 
 			if (is_isl || is_ibd) {
-				(*callback)(NULL, dbinfo.name, is_isl);
+				(*callback)(NULL, dbinfo.name, is_isl, false);
 			}
 		}
 
@@ -3221,7 +3478,7 @@ static dberr_t enumerate_ibd_files(process_single_tablespace_func_t callback)
 				if (strlen(fileinfo.name) > 4) {
 					bool is_isl= false;
 					if (ends_with(fileinfo.name, ".ibd") || ((is_isl = ends_with(fileinfo.name, ".isl"))))
-						(*callback)(dbinfo.name, fileinfo.name, is_isl);
+						(*callback)(dbinfo.name, fileinfo.name, is_isl, false);
 				}
 			}
 
@@ -3935,6 +4192,7 @@ static bool xtrabackup_backup_func()
 	uint			 i;
 	uint			 count;
 	pthread_mutex_t		 count_mutex;
+	CorruptedPages corrupted_pages;
 	data_thread_ctxt_t 	*data_threads;
 	pthread_mutex_init(&backup_mutex, NULL);
 	pthread_cond_init(&scanned_lsn_cond, NULL);
@@ -4201,6 +4459,7 @@ fail_before_log_copying_thread_start:
 		data_threads[i].num = i+1;
 		data_threads[i].count = &count;
 		data_threads[i].count_mutex = &count_mutex;
+		data_threads[i].corrupted_pages = &corrupted_pages;
 		data_threads[i].id = os_thread_create(data_copy_thread_func,
 						      data_threads + i);
 	}
@@ -4221,7 +4480,7 @@ fail_before_log_copying_thread_start:
 	datafiles_iter_free(it);
 	}
 
-	bool ok = backup_start();
+	bool ok = backup_start(corrupted_pages);
 
 	if (ok) {
 		ok = xtrabackup_backup_low();
@@ -4238,6 +4497,9 @@ fail_before_log_copying_thread_start:
 		}
 	}
 
+	if (opt_log_innodb_page_corruption)
+		ok = corrupted_pages.print_to_file(MB_CORRUPTED_PAGES_FILE);
+
 	if (!ok) {
 		goto fail;
 	}
@@ -4265,7 +4527,13 @@ fail_before_log_copying_thread_start:
 	log_file_op = NULL;
 	pthread_mutex_destroy(&backup_mutex);
 	pthread_cond_destroy(&scanned_lsn_cond);
-	return(true);
+	if (opt_log_innodb_page_corruption && !corrupted_pages.empty()) {
+		msg("Error: corrupted innodb pages are found and logged to "
+			MB_CORRUPTED_PAGES_FILE " file");
+		return false;
+	}
+	else
+		return(true);
 }
 
 
@@ -4287,7 +4555,7 @@ FTWRL.  This ensures consistent backup in presence of DDL.
   It is the responsibility of the prepare phase to deal with .new, .ren, and .del
   files.
 */
-void backup_fix_ddl(void)
+void backup_fix_ddl(CorruptedPages &corrupted_pages)
 {
 	std::set<std::string> new_tables;
 	std::set<std::string> dropped_tables;
@@ -4309,6 +4577,7 @@ void backup_fix_ddl(void)
 
 		if (ddl_tracker.drops.find(id) != ddl_tracker.drops.end()) {
 			dropped_tables.insert(name);
+			corrupted_pages.drop_space(id);
 			continue;
 		}
 
@@ -4320,6 +4589,8 @@ void backup_fix_ddl(void)
 		const std::string new_name = ddl_tracker.id_to_name[id];
 		if (new_name != name) {
 			renamed_tables[name] = new_name;
+			if (opt_log_innodb_page_corruption)
+				corrupted_pages.rename_space(id, new_name);
 		}
 	}
 
@@ -4339,6 +4610,8 @@ void backup_fix_ddl(void)
 		if (ddl_tracker.drops.find(id) == ddl_tracker.drops.end()) {
 			dropped_tables.erase(name);
 			new_tables.insert(name);
+			if (opt_log_innodb_page_corruption)
+				corrupted_pages.drop_space(id);
 		}
 	}
 
@@ -4386,23 +4659,7 @@ void backup_fix_ddl(void)
 		const char *space_name = iter->c_str();
 		if (check_if_skip_table(space_name))
 			continue;
-		std::string name(*iter);
-		bool is_remote = access((name + ".ibd").c_str(), R_OK) != 0;
-		const char *extension = is_remote ? ".isl" : ".ibd";
-		name.append(extension);
-		char buf[FN_REFLEN];
-		strncpy(buf, name.c_str(), sizeof buf - 1);
-		buf[sizeof buf - 1] = '\0';
-		const char *dbname = buf;
-		char *p = strchr(buf, '/');
-		if (p == 0) {
-			msg("Unexpected tablespace %s filename %s", space_name, name.c_str());
-			ut_a(0);
-		}
-		ut_a(p);
-		*p = 0;
-		const char *tablename = p + 1;
-		xb_load_single_table_tablespace(dbname, tablename, is_remote);
+		xb_load_single_table_tablespace(*iter, false);
 	}
 
 	it = datafiles_iter_new();
@@ -4415,7 +4672,8 @@ void backup_fix_ddl(void)
 			continue;
 		std::string dest_name(node->space->name);
 		dest_name.append(".new");
-		xtrabackup_copy_datafile(node, 0, dest_name.c_str(), wf_write_through);
+		xtrabackup_copy_datafile(node, 0, dest_name.c_str(), wf_write_through,
+			corrupted_pages);
 	}
 
 	datafiles_iter_free(it);
@@ -5341,6 +5599,7 @@ static ibool prepare_handle_del_files(const char *datadir, const char *db, const
 @return	whether the operation succeeded */
 static bool xtrabackup_prepare_func(char** argv)
 {
+	CorruptedPages corrupted_pages;
 	char			 metadata_path[FN_REFLEN];
 
 	/* cd to target-dir */
@@ -5506,6 +5765,30 @@ static bool xtrabackup_prepare_func(char** argv)
 
 	ut_ad(!fil_system.freeze_space_list);
 
+        corrupted_pages.read_from_file(MB_CORRUPTED_PAGES_FILE);
+        if (xtrabackup_incremental)
+        {
+          char inc_filename[FN_REFLEN];
+          sprintf(inc_filename, "%s/%s", xtrabackup_incremental_dir,
+                  MB_CORRUPTED_PAGES_FILE);
+          corrupted_pages.read_from_file(inc_filename);
+        }
+        if (!corrupted_pages.empty())
+          corrupted_pages.zero_out_free_pages();
+        if (corrupted_pages.empty())
+        {
+          if (!xtrabackup_incremental && unlink(MB_CORRUPTED_PAGES_FILE) &&
+              errno != ENOENT)
+          {
+            char errbuf[MYSYS_STRERROR_SIZE];
+            my_strerror(errbuf, sizeof(errbuf), errno);
+            die("Error: unlink %s failed: %s", MB_CORRUPTED_PAGES_FILE,
+                errbuf);
+          }
+        }
+        else
+          corrupted_pages.print_to_file(MB_CORRUPTED_PAGES_FILE);
+
 	if (ok) {
 		msg("Last binlog file %s, position %lld",
 		    trx_sys.recovered_binlog_filename,
@@ -5564,7 +5847,7 @@ static bool xtrabackup_prepare_func(char** argv)
 
 error_cleanup:
 	xb_filters_free();
-        return ok && !ib::error::was_logged();
+        return ok && !ib::error::was_logged() && corrupted_pages.empty();
 }
 
 /**************************************************************************
diff --git a/extra/mariabackup/xtrabackup.h b/extra/mariabackup/xtrabackup.h
index f18d79aea55..6376849430c 100644
--- a/extra/mariabackup/xtrabackup.h
+++ b/extra/mariabackup/xtrabackup.h
@@ -25,6 +25,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include "datasink.h"
 #include "xbstream.h"
 #include "changed_page_bitmap.h"
+#include <set>
 
 struct xb_delta_info_t
 {
@@ -36,6 +37,32 @@ struct xb_delta_info_t
 	ulint	space_id;
 };
 
+class CorruptedPages
+{
+public:
+  CorruptedPages();
+  ~CorruptedPages();
+  void add_page(const char *file_name, ulint space_id, unsigned page_no);
+  bool contains(ulint space_id, unsigned page_no) const;
+  void drop_space(ulint space_id);
+  void rename_space(ulint space_id, const std::string &new_name);
+  bool print_to_file(const char *file_name) const;
+  void read_from_file(const char *file_name);
+  bool empty() const;
+  void zero_out_free_pages();
+
+private:
+  void add_page_no_lock(const char *space_name, ulint space_id,
+                        unsigned page_no, bool convert_space_name);
+  struct space_info_t {
+    std::string space_name;
+    std::set<unsigned> pages;
+  };
+  typedef std::map<ulint, space_info_t> container_t;
+  mutable pthread_mutex_t m_mutex;
+  container_t m_spaces;
+};
+
 /* value of the --incremental option */
 extern lsn_t incremental_lsn;
 
@@ -111,6 +138,7 @@ extern my_bool		opt_remove_original;
 extern my_bool		opt_extended_validation;
 extern my_bool		opt_encrypted_backup;
 extern my_bool		opt_lock_ddl_per_table;
+extern my_bool    opt_log_innodb_page_corruption;
 
 extern char		*opt_incremental_history_name;
 extern char		*opt_incremental_history_uuid;
diff --git a/mysql-test/lib/mtr_report.pm b/mysql-test/lib/mtr_report.pm
index 32250802815..473b21441e2 100644
--- a/mysql-test/lib/mtr_report.pm
+++ b/mysql-test/lib/mtr_report.pm
@@ -514,6 +514,10 @@ sub mtr_report_stats ($$$$) {
       # if a test case has to be retried it should have the result MTR_RES_FAILED in jUnit XML
       if ($test->{'result'} eq "MTR_RES_FAILED" || $test->{'retries'} > 0) {
         my $logcontents = $test->{'logfile-failed'} || $test->{'logfile'};
+        # remove any double ] that would end the cdata
+        $logcontents =~ s/]]/\x{fffd}/g;
+        # replace wide characters that aren't allowed in XML 1.0
+        $logcontents =~ s/[\x00-\x08\x0B\x0C\x0E-\x1F]/\x{fffd}/g;
 
         $xml_report .= qq(>\n\t\t\t<failure message="" type="MTR_RES_FAILED">\n<![CDATA[$logcontents]]>\n\t\t\t</failure>\n\t\t</testcase>\n);
       } elsif ($test->{'result'} eq "MTR_RES_SKIPPED" && $test->{'disable'}) {
@@ -530,9 +534,9 @@ sub mtr_report_stats ($$$$) {
     # save to file
     my $xml_file = $::opt_xml_report;
 
-    open XML_FILE, ">", $xml_file or die "Cannot create file $xml_file: $!";
-    print XML_FILE $xml_report;
-    close XML_FILE;
+    open (my $XML_UFILE, '>:encoding(UTF-8)', $xml_file) or die 'Cannot create file $xml_file: $!';
+    print $XML_UFILE $xml_report;
+    close $XML_UFILE or warn "File close failed!";
   }
 
   if (@$extra_warnings)
diff --git a/mysql-test/main/auto_increment_ranges_innodb.result b/mysql-test/main/auto_increment_ranges_innodb.result
index 61eccc6f944..7800099200f 100644
--- a/mysql-test/main/auto_increment_ranges_innodb.result
+++ b/mysql-test/main/auto_increment_ranges_innodb.result
@@ -289,3 +289,51 @@ pk	f
 5	a
 6	<===
 drop table t1;
+#
+# MDEV-21842: auto_increment does not increment with compound primary
+# key on partitioned table
+#
+create or replace table `t` (
+`id` bigint(20) unsigned not null auto_increment,
+`a` int(10) not null ,
+`dt` date not null,
+primary key (`id`, `dt`) ,
+unique key (`a`, `dt`)
+)
+partition by range  columns(`dt`)
+(
+partition `p202002` values less than ('2020-03-01'),
+partition `P202003` values less than ('2020-04-01')
+);
+connect  con1, localhost, root,,;
+connect  con2, localhost, root,,;
+connection con1;
+start transaction;
+insert into t (a, dt) values (1, '2020-02-29');
+connection con2;
+start transaction;
+insert into t (a, dt) values (1, '2020-02-29');
+connection con1;
+insert into t (a, dt) values (2, '2020-02-29');
+select auto_increment from information_schema.tables where table_name='t';
+auto_increment
+4
+commit;
+connection con2;
+ERROR 23000: Duplicate entry '1-2020-02-29' for key 'a'
+connection con1;
+select auto_increment from information_schema.tables where table_name='t';
+auto_increment
+4
+insert into t (a, dt) values (3, '2020-02-29');
+insert into t (a, dt) values (4, '2020-02-29');
+disconnect con1;
+disconnect con2;
+connection default;
+select * from t;
+id	a	dt
+1	1	2020-02-29
+3	2	2020-02-29
+4	3	2020-02-29
+5	4	2020-02-29
+drop table t;
diff --git a/mysql-test/main/auto_increment_ranges_innodb.test b/mysql-test/main/auto_increment_ranges_innodb.test
index 016ca16bd91..92d377eb147 100644
--- a/mysql-test/main/auto_increment_ranges_innodb.test
+++ b/mysql-test/main/auto_increment_ranges_innodb.test
@@ -18,3 +18,62 @@ select * from t1;
 drop table t1;
 --let $datadir=`select @@datadir`
 --remove_file $datadir/test/load.data
+
+--echo #
+--echo # MDEV-21842: auto_increment does not increment with compound primary
+--echo # key on partitioned table
+--echo #
+
+create or replace table `t` (
+  `id` bigint(20) unsigned not null auto_increment,
+  `a` int(10) not null ,
+  `dt` date not null,
+  primary key (`id`, `dt`) ,
+  unique key (`a`, `dt`)
+)
+ partition by range  columns(`dt`)
+(
+ partition `p202002` values less than ('2020-03-01'),
+ partition `P202003` values less than ('2020-04-01')
+);
+
+connect (con1, localhost, root,,);
+connect (con2, localhost, root,,);
+
+--connection con1
+start transaction;
+insert into t (a, dt) values (1, '2020-02-29');
+
+--connection con2
+start transaction;
+let $conn2_id= `SELECT CONNECTION_ID()`;
+send insert into t (a, dt) values (1, '2020-02-29');
+
+--connection con1
+# Ensure that the above insert via conn2 increments next_auto_inc_val
+# before the following insert via conn1 starts.
+let $wait_condition=select 1 from Information_schema.INNODB_TRX
+    where trx_mysql_thread_id = $conn2_id and trx_state = 'LOCK WAIT'
+    and trx_query = "insert into t (a, dt) values (1, '2020-02-29')";
+--source include/wait_condition.inc
+
+insert into t (a, dt) values (2, '2020-02-29');
+select auto_increment from information_schema.tables where table_name='t';
+commit;
+
+--connection con2
+--error ER_DUP_ENTRY
+reap;
+
+--connection con1
+select auto_increment from information_schema.tables where table_name='t';
+insert into t (a, dt) values (3, '2020-02-29');
+insert into t (a, dt) values (4, '2020-02-29');
+
+disconnect con1;
+disconnect con2;
+
+--connection default
+select * from t;
+drop table t;
+
diff --git a/mysql-test/main/derived_cond_pushdown.result b/mysql-test/main/derived_cond_pushdown.result
index 33e8eab968b..6ea4528e641 100644
--- a/mysql-test/main/derived_cond_pushdown.result
+++ b/mysql-test/main/derived_cond_pushdown.result
@@ -10566,6 +10566,47 @@ a
 abc
 DROP VIEW v1;
 DROP TABLE t1;
+#
+# MDEV-19179: pushdown into UNION of aggregation selects whose
+#             corresponding columns have different names
+#
+create table t1 (a int);
+insert into t1 values (3), (7), (1);
+select *
+from (select min(a) as x from t1 union all select max(a) as y from t1) t
+where x>0;
+x
+1
+7
+explain extended select *
+from (select min(a) as x from t1 union all select max(a) as y from t1) t
+where x>0;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
+1	PRIMARY	<derived2>	ALL	NULL	NULL	NULL	NULL	6	100.00	Using where
+2	DERIVED	t1	ALL	NULL	NULL	NULL	NULL	3	100.00	
+3	UNION	t1	ALL	NULL	NULL	NULL	NULL	3	100.00	
+Warnings:
+Note	1003	/* select#1 */ select `t`.`x` AS `x` from (/* select#2 */ select min(`test`.`t1`.`a`) AS `x` from `test`.`t1` having `x` > 0 union all /* select#3 */ select max(`test`.`t1`.`a`) AS `x` from `test`.`t1` having `x` > 0) `t` where `t`.`x` > 0
+prepare stmt from "select *
+from (select min(a) as x from t1 union all select max(a) as y from t1) t
+where x>0";
+execute stmt;
+x
+1
+7
+execute stmt;
+x
+1
+7
+deallocate prepare stmt;
+create view v1(m) as
+select min(a) as x from t1 union all select max(a) as y from t1;
+select * from v1 where m > 0;
+m
+1
+7
+drop view v1;
+drop table t1;
 # End of 10.2 tests
 #
 # MDEV-14579: pushdown conditions into materialized views/derived tables
diff --git a/mysql-test/main/derived_cond_pushdown.test b/mysql-test/main/derived_cond_pushdown.test
index b2f97029ede..7667cd44ed2 100644
--- a/mysql-test/main/derived_cond_pushdown.test
+++ b/mysql-test/main/derived_cond_pushdown.test
@@ -2185,6 +2185,34 @@ SELECT * FROM v1 WHERE IF( a REGEXP 'def', 'foo', a ) IN ('abc', 'foobar');
 DROP VIEW v1;
 DROP TABLE t1;
 
+--echo #
+--echo # MDEV-19179: pushdown into UNION of aggregation selects whose
+--echo #             corresponding columns have different names
+--echo #
+
+create table t1 (a int);
+insert into t1 values (3), (7), (1);
+
+let $q=
+select *
+from (select min(a) as x from t1 union all select max(a) as y from t1) t
+where x>0;
+
+eval $q;
+eval explain extended $q;
+
+eval prepare stmt from "$q";
+execute stmt;
+execute stmt;
+deallocate prepare stmt;
+
+create view v1(m) as
+select min(a) as x from t1 union all select max(a) as y from t1;
+select * from v1 where m > 0;
+
+drop view v1;
+drop table t1;
+
 --echo # End of 10.2 tests
 
 --echo #
diff --git a/mysql-test/main/information_schema.result b/mysql-test/main/information_schema.result
index 9f36ba35a22..627fb9e4569 100644
--- a/mysql-test/main/information_schema.result
+++ b/mysql-test/main/information_schema.result
@@ -2219,8 +2219,6 @@ SCHEMA_NAME
 # End of 10.1 tests
 #
 #
-# Start of 10.2 Test
-#
 # MDEV-14836: Assertion `m_status == DA_ERROR' failed in
 # Diagnostics_area::sql_errno upon query from I_S with LIMIT ROWS EXAMINED
 #
@@ -2305,5 +2303,12 @@ mysql	global_priv	Priv	json_valid(`Priv`)	def	mysql
 test	t	a	`i` > 0	def	test
 drop table t;
 #
+# MDEV-24230 subquery on information_schema fails with error message
+#
+create table t1 (n int);
+create table t2 (n int);
+insert into t1 set n = (select table_rows from information_schema.tables where table_name='t2');
+drop table t1, t2;
+#
 # End of 10.3 tests
 #
diff --git a/mysql-test/main/information_schema.test b/mysql-test/main/information_schema.test
index 9c77606b9d4..ca49091ef76 100644
--- a/mysql-test/main/information_schema.test
+++ b/mysql-test/main/information_schema.test
@@ -1924,8 +1924,6 @@ SELECT SCHEMA_NAME from information_schema.schemata where schema_name=REPEAT('a'
 --echo #
 
 --echo #
---echo # Start of 10.2 Test
---echo #
 --echo # MDEV-14836: Assertion `m_status == DA_ERROR' failed in
 --echo # Diagnostics_area::sql_errno upon query from I_S with LIMIT ROWS EXAMINED
 --echo #
@@ -1999,5 +1997,13 @@ from information_schema.TABLE_CONSTRAINTS tc
 drop table t;
 
 --echo #
+--echo # MDEV-24230 subquery on information_schema fails with error message
+--echo #
+create table t1 (n int);
+create table t2 (n int);
+insert into t1 set n = (select table_rows from information_schema.tables where table_name='t2');
+drop table t1, t2;
+
+--echo #
 --echo # End of 10.3 tests
 --echo #
diff --git a/mysql-test/main/lock_view.test b/mysql-test/main/lock_view.test
index dd8809ab89d..4b1adac5be1 100644
--- a/mysql-test/main/lock_view.test
+++ b/mysql-test/main/lock_view.test
@@ -1,4 +1,5 @@
 source include/not_embedded.inc;
+source include/have_perfschema.inc;
 #
 # LOCK TABLES and privileges on views
 #
diff --git a/mysql-test/main/mysqldump-system.test b/mysql-test/main/mysqldump-system.test
index 9aaa5fff6f9..c1965410167 100644
--- a/mysql-test/main/mysqldump-system.test
+++ b/mysql-test/main/mysqldump-system.test
@@ -3,6 +3,10 @@
 --source include/have_udf.inc
 --source include/platform.inc
 
+if (!$AUTH_SOCKET_SO) {
+  --skip Need auth socket plugin
+}
+
 --echo #
 --echo # MDEV-23630: mysqldump to logically dump system tables
 --echo #
diff --git a/mysql-test/main/partition_innodb.result b/mysql-test/main/partition_innodb.result
index e212dadd06a..67829fd46d9 100644
--- a/mysql-test/main/partition_innodb.result
+++ b/mysql-test/main/partition_innodb.result
@@ -1087,10 +1087,10 @@ INSERT INTO t1 VALUES ();
 SELECT * FROM t1;
 a
 -1
-1
 3
 4
 6
+7
 DROP TABLE t1;
 #
 # End of 10.3 tests
diff --git a/mysql-test/main/sp.result b/mysql-test/main/sp.result
index d5e722ff1bd..17b0af92a40 100644
--- a/mysql-test/main/sp.result
+++ b/mysql-test/main/sp.result
@@ -8448,8 +8448,25 @@ ERROR 22007: Incorrect integer value: 'y' for column ``.``.`a` at row 1
 DROP TABLE t1;
 SET sql_mode=DEFAULT;
 #
-# Start of 10.3 tests
+# MDEV-24220: error when opening a table for the second call of SP
 #
+CREATE TABLE t1 (a INT, b INT);
+INSERT INTO t1 VALUES (1,1),(2,2);
+CREATE VIEW v1 AS SELECT MAX(a) as f FROM t1;
+CREATE PROCEDURE p1()
+BEGIN
+SELECT * FROM v1;
+END $
+CALL p1;
+f
+2
+ALTER TABLE t1 DROP a;
+CALL p1;
+ERROR HY000: View 'test.v1' references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them
+DROP PROCEDURE p1;
+DROP VIEW v1;
+DROP TABLE t1;
+#End of 10.2 tests
 #
 # MDEV-12007 Allow ROW variables as a cursor FETCH target
 #
diff --git a/mysql-test/main/sp.test b/mysql-test/main/sp.test
index c6c00ca8d91..bf3a70b6283 100644
--- a/mysql-test/main/sp.test
+++ b/mysql-test/main/sp.test
@@ -9991,9 +9991,30 @@ DROP TABLE t1;
 SET sql_mode=DEFAULT;
 
 --echo #
---echo # Start of 10.3 tests
+--echo # MDEV-24220: error when opening a table for the second call of SP
 --echo #
 
+CREATE TABLE t1 (a INT, b INT);
+INSERT INTO t1 VALUES (1,1),(2,2);
+CREATE VIEW v1 AS SELECT MAX(a) as f FROM t1;
+--delimiter $
+CREATE PROCEDURE p1()
+BEGIN
+  SELECT * FROM v1;
+END $
+--delimiter ;
+
+CALL p1;
+ALTER TABLE t1 DROP a;
+-- error ER_VIEW_INVALID
+CALL p1;
+
+DROP PROCEDURE p1;
+DROP VIEW v1;
+DROP TABLE t1;
+
+--echo #End of 10.2 tests
+
 --echo #
 --echo # MDEV-12007 Allow ROW variables as a cursor FETCH target
 --echo #
diff --git a/mysql-test/main/subselect4.result b/mysql-test/main/subselect4.result
index 00f61e9433b..44b879ae33b 100644
--- a/mysql-test/main/subselect4.result
+++ b/mysql-test/main/subselect4.result
@@ -2719,6 +2719,41 @@ SET join_cache_level= @save_join_cache_level;
 DROP TABLE t1,t2,t3,t4;
 # End of 10.2 tests
 #
+# MDEV-21265: IN predicate conversion to IN subquery should be allowed for a broader set of datatype comparison
+#
+CREATE TABLE t1(a VARCHAR(50) collate utf8_general_ci, b INT);
+INSERT INTO t1 VALUES ('abc',1), ('def', 2), ('ghi', 3), ('jkl', 4), ('mno', 5);
+CREATE TABLE t2(a VARCHAR(50) collate utf8mb4_general_ci, b INT);
+INSERT INTO t2 VALUES ('abc',1), ('def', 2), ('ghi', 3), ('jkl', 4), ('mno', 5);
+set @save_in_predicate_conversion_threshold= @@in_predicate_conversion_threshold;
+set in_predicate_conversion_threshold=2;
+set names 'utf8mb4';
+#
+# IN predicate to IN subquery is not allowed as materialization is not allowed
+# The character set on the inner side is not equal to or a proper subset of the outer side
+#
+EXPLAIN
+SELECT * FROM t1 WHERE (t1.a,t1.b) IN (('abx',1),('def',2), ('abc', 3));
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	5	Using where
+set names 'utf8';
+#
+# IN predicate to IN subquery is performed as materialization is llowed
+# The character set on the inner side is a proper subset of the outer side
+#
+# this test in 10.5 has only 2 rows in the IN predicate
+EXPLAIN
+SELECT * FROM t2 WHERE (t2.a,t2.b) IN (('abc',1), ('def', 2));
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	PRIMARY	t2	ALL	NULL	NULL	NULL	NULL	5	
+1	PRIMARY	<subquery2>	eq_ref	distinct_key	distinct_key	16	func,func	1	Using where
+2	MATERIALIZED	<derived3>	ALL	NULL	NULL	NULL	NULL	2	
+3	DERIVED	NULL	NULL	NULL	NULL	NULL	NULL	NULL	No tables used
+set names default;
+set @@in_predicate_conversion_threshold= @save_in_predicate_conversion_threshold;
+DROP TABLE t1,t2;
+# End of 10.3 tests
+#
 # MDEV-19134: EXISTS() slower if ORDER BY is defined
 #
 create table t0 (a int);
diff --git a/mysql-test/main/subselect4.test b/mysql-test/main/subselect4.test
index c9b05c4f015..d46bb846a9e 100644
--- a/mysql-test/main/subselect4.test
+++ b/mysql-test/main/subselect4.test
@@ -2240,6 +2240,45 @@ DROP TABLE t1,t2,t3,t4;
 --echo # End of 10.2 tests
 
 --echo #
+--echo # MDEV-21265: IN predicate conversion to IN subquery should be allowed for a broader set of datatype comparison
+--echo #
+
+CREATE TABLE t1(a VARCHAR(50) collate utf8_general_ci, b INT);
+INSERT INTO t1 VALUES ('abc',1), ('def', 2), ('ghi', 3), ('jkl', 4), ('mno', 5);
+
+CREATE TABLE t2(a VARCHAR(50) collate utf8mb4_general_ci, b INT);
+INSERT INTO t2 VALUES ('abc',1), ('def', 2), ('ghi', 3), ('jkl', 4), ('mno', 5);
+
+set @save_in_predicate_conversion_threshold= @@in_predicate_conversion_threshold;
+set in_predicate_conversion_threshold=2;
+
+set names 'utf8mb4';
+--echo #
+--echo # IN predicate to IN subquery is not allowed as materialization is not allowed
+--echo # The character set on the inner side is not equal to or a proper subset of the outer side
+--echo #
+
+EXPLAIN
+SELECT * FROM t1 WHERE (t1.a,t1.b) IN (('abx',1),('def',2), ('abc', 3));
+
+set names 'utf8';
+--echo #
+--echo # IN predicate to IN subquery is performed as materialization is llowed
+--echo # The character set on the inner side is a proper subset of the outer side
+--echo #
+
+--echo # this test in 10.5 has only 2 rows in the IN predicate
+
+EXPLAIN
+SELECT * FROM t2 WHERE (t2.a,t2.b) IN (('abc',1), ('def', 2));
+
+set names default;
+set @@in_predicate_conversion_threshold= @save_in_predicate_conversion_threshold;
+DROP TABLE t1,t2;
+
+--echo # End of 10.3 tests
+
+--echo #
 --echo # MDEV-19134: EXISTS() slower if ORDER BY is defined
 --echo #
 create table t0 (a int);
diff --git a/mysql-test/main/xa.result b/mysql-test/main/xa.result
index 2fe8316b65b..46c0c8551a1 100644
--- a/mysql-test/main/xa.result
+++ b/mysql-test/main/xa.result
@@ -297,7 +297,7 @@ DROP TABLE t1;
 #
 # Bug#12352846 - TRANS_XA_START(THD*):
 #                ASSERTION THD->TRANSACTION.XID_STATE.XID.IS_NULL()
-#                FAILED 
+#                FAILED
 #
 CREATE TABLE t1 (a INT) ENGINE=InnoDB;
 CREATE TABLE t2 (a INT) ENGINE=InnoDB;
@@ -349,6 +349,32 @@ XA END 'xid1';
 XA ROLLBACK 'xid1';
 DROP TABLE t1, t2, t3;
 #
+# MDEV 15532 XA: Assertion `!log->same_pk' failed in
+# row_log_table_apply_delete
+#
+CREATE TABLE t1 (a INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1),(2);
+connect  con1,localhost,root,,test;
+XA START 'xid';
+UPDATE t1 SET a = 5;
+connection default;
+SET innodb_lock_wait_timeout= 2, lock_wait_timeout= 2;
+ALTER TABLE non_existing_table1;
+ERROR 42S02: Table 'test.non_existing_table1' doesn't exist
+ALTER TABLE t1 FORCE;;
+connection con1;
+ALTER TABLE non_existing_table2;
+ERROR XAE07: XAER_RMFAIL: The command cannot be executed when global transaction is in the  ACTIVE state
+DELETE FROM t1 LIMIT 1;
+connection default;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection con1;
+XA END 'xid';
+XA ROLLBACK 'xid';
+DROP TABLE t1;
+disconnect con1;
+connection default;
+#
 # MDEV-21766 - Forbid XID with empty 'gtrid'
 #
 CREATE TABLE t1(a INT) ENGINE=InnoDB;
diff --git a/mysql-test/main/xa.test b/mysql-test/main/xa.test
index 07183feda76..f5a5cc96197 100644
--- a/mysql-test/main/xa.test
+++ b/mysql-test/main/xa.test
@@ -392,7 +392,7 @@ DROP TABLE t1;
 --echo #
 --echo # Bug#12352846 - TRANS_XA_START(THD*):
 --echo #                ASSERTION THD->TRANSACTION.XID_STATE.XID.IS_NULL()
---echo #                FAILED 
+--echo #                FAILED
 --echo #
 
 CREATE TABLE t1 (a INT) ENGINE=InnoDB;
@@ -449,7 +449,7 @@ CREATE TABLE t1 (pk INT PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE t2 (pk INT PRIMARY KEY) ENGINE=InnoDB;
 INSERT INTO t2 VALUES (1),(2);
 CREATE TABLE t3 (i INT) ENGINE=InnoDB;
- 
+
 XA BEGIN 'xid1';
 REPLACE INTO t1 SELECT * FROM t2;
 
@@ -478,6 +478,44 @@ XA END 'xid1';
 XA ROLLBACK 'xid1';
 DROP TABLE t1, t2, t3;
 
+--echo #
+--echo # MDEV 15532 XA: Assertion `!log->same_pk' failed in
+--echo # row_log_table_apply_delete
+--echo #
+
+CREATE TABLE t1 (a INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1),(2);
+
+--connect (con1,localhost,root,,test)
+
+XA START 'xid';
+UPDATE t1 SET a = 5;
+
+--connection default
+SET innodb_lock_wait_timeout= 2, lock_wait_timeout= 2;
+
+--error ER_NO_SUCH_TABLE
+ALTER TABLE non_existing_table1;
+
+--send ALTER TABLE t1 FORCE;
+
+--connection con1
+--error ER_XAER_RMFAIL
+
+ALTER TABLE non_existing_table2;
+DELETE FROM t1 LIMIT 1;
+
+--connection default
+--error ER_LOCK_WAIT_TIMEOUT
+--reap
+
+# Cleanup
+--connection con1
+XA END 'xid';
+XA ROLLBACK 'xid';
+DROP TABLE t1;
+--disconnect con1
+connection default;
 
 --echo #
 --echo # MDEV-21766 - Forbid XID with empty 'gtrid'
diff --git a/mysql-test/suite/binlog/r/binlog_mysqlbinlog_suppress_O_TMPFILE.result b/mysql-test/suite/binlog/r/binlog_mysqlbinlog_suppress_O_TMPFILE.result
new file mode 100644
index 00000000000..0154495f502
--- /dev/null
+++ b/mysql-test/suite/binlog/r/binlog_mysqlbinlog_suppress_O_TMPFILE.result
@@ -0,0 +1,21 @@
+RESET MASTER;
+CREATE TABLE t(f text);
+INSERT INTO t VALUES (repeat('x',4096));
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+SELECT COUNT(*) FROM t;
+COUNT(*)
+512
+FLUSH LOGS;
+DROP TABLE t;
+# 512- Rows must be present
+include/assert.inc [Table t should have 512 rows.]
+DROP TABLE t;
+RESET MASTER;
diff --git a/mysql-test/suite/binlog/t/binlog_mysqlbinlog_suppress_O_TMPFILE.test b/mysql-test/suite/binlog/t/binlog_mysqlbinlog_suppress_O_TMPFILE.test
new file mode 100644
index 00000000000..c97269a8ad8
--- /dev/null
+++ b/mysql-test/suite/binlog/t/binlog_mysqlbinlog_suppress_O_TMPFILE.test
@@ -0,0 +1,51 @@
+# ==== Purpose ====
+#
+# Suppress the following informational note that gets printed to standard
+# error when O_TMPFILE flag is not supported by underlying operating system.
+#
+# Note: ../client/mysqlbinlog: O_TMPFILE is not supported on /tmp (disabling
+# future attempts)
+#
+# Step 1: Generate a binarylog file with a size greater than 1MB.
+# Step 2: Use mysqlbinlog tool to generate sql file and redirect the standard
+#         error to standard output (2>&1)
+# Step 3: Source the generated sql file as inpurt to mysql client, observe no
+#         syntax error is reported.
+#
+# ==== References ====
+#
+# MDEV-23846: O_TMPFILE error in mysqlbinlog stream output breaks restore
+#
+--source include/have_binlog_format_row.inc
+
+RESET MASTER;
+CREATE TABLE t(f text);
+INSERT INTO t VALUES (repeat('x',4096));
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+INSERT INTO t SELECT * FROM t;
+SELECT COUNT(*) FROM t;
+FLUSH LOGS;
+let $MYSQLD_DATADIR= `select @@datadir`;
+
+--exec $MYSQL_BINLOG $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_base64.sql 2>&1
+#
+#  Clear database and restore from binlog
+#
+DROP TABLE t;
+
+--exec $MYSQL test < $MYSQLTEST_VARDIR/tmp/mysqlbinlog_base64.sql
+
+--echo # 512- Rows must be present
+--let $assert_cond= COUNT(*) = 512 FROM t
+--let $assert_text= Table t should have 512 rows.
+--source include/assert.inc
+
+DROP TABLE t;
+RESET MASTER;
diff --git a/mysql-test/suite/galera/disabled.def b/mysql-test/suite/galera/disabled.def
index d940c702d54..046feac5566 100644
--- a/mysql-test/suite/galera/disabled.def
+++ b/mysql-test/suite/galera/disabled.def
@@ -12,14 +12,11 @@
 
 GCF-1081 : MDEV-18283 Galera test failure on galera.GCF-1081
 GCF-939 : MDEV-21520 galera.GCF-939
-MDEV-16509 : MDEV-21523 galera.MDEV-16509
 MDEV-20225 : MDEV-20886 galera.MDEV-20225
 MW-286 : MDEV-18464 Killing thread can cause mutex deadlock if done concurrently with Galera/replication victim kill
 MW-328A : MDEV-22666 galera.MW-328A MTR failed: "Semaphore wait has lasted > 600 seconds" and do not release port 16002
-MW-328A : MDEV-22666?
 MW-328B : MDEV-22666 galera.MW-328A MTR failed: "Semaphore wait has lasted > 600 seconds" and do not release port 16002
 MW-329 : MDEV-19962 Galera test failure on MW-329
-galera.galera_defaults : MDEV-21494 Galera test sporadic failure on galera.galera_defaults
 galera_as_slave_replication_bundle : MDEV-15785 OPTION_GTID_BEGIN is set in Gtid_log_event::do_apply_event()
 galera_bf_abort_group_commit : MDEV-18282 Galera test failure on galera.galera_bf_abort_group_commit
 galera_binlog_stmt_autoinc : MDEV-19959 Galera test failure on galera_binlog_stmt_autoinc	
@@ -44,7 +41,6 @@ galera_var_reject_queries : assertion in inline_mysql_socket_send
 galera_var_replicate_myisam_on : MDEV-24062 Galera test failure on galera_var_replicate_myisam_on
 galera_var_retry_autocommit: MDEV-18181 Galera test failure on galera.galera_var_retry_autocommit
 galera_wan : MDEV-17259 Test failure on galera.galera_wan
-lp1376747-4 : MDEV-21911 Galera test failure on lp1376747-4
 partition : MDEV-19958 Galera test failure on galera.partition
 query_cache: MDEV-15805 Test failure on galera.query_cache
 sql_log_bin : MDEV-21491 galera.sql_log_bin
diff --git a/mysql-test/suite/galera/r/galera#505.result b/mysql-test/suite/galera/r/galera#505.result
index bc7eb3b9ed4..d5e78c570b5 100644
--- a/mysql-test/suite/galera/r/galera#505.result
+++ b/mysql-test/suite/galera/r/galera#505.result
@@ -2,6 +2,7 @@ connection node_2;
 connection node_1;
 connection node_1;
 SET SESSION wsrep_sync_wait=0;
+# Correct Galera library found
 SET SESSION wsrep_sync_wait=DEFAULT;
 SET GLOBAL wsrep_provider_options = 'pc.weight=3';
 SHOW GLOBAL VARIABLES LIKE 'wsrep_provider_options';
diff --git a/mysql-test/suite/galera/r/galera_as_slave_replay.result b/mysql-test/suite/galera/r/galera_as_slave_replay.result
new file mode 100644
index 00000000000..760617be5f7
--- /dev/null
+++ b/mysql-test/suite/galera/r/galera_as_slave_replay.result
@@ -0,0 +1,95 @@
+connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2;
+connection node_2a;
+connection node_1;
+RESET MASTER;
+connection node_2a;
+START SLAVE;
+connection node_1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY, f2 CHAR(1)) engine=innodb;
+INSERT INTO t1 VALUES (1, 'a');
+INSERT INTO t1 VALUES (3, 'a');
+set binlog_format=STATEMENT;
+SET AUTOCOMMIT=ON;
+START TRANSACTION;
+SELECT * FROM t1  FOR UPDATE;
+f1	f2
+1	a
+3	a
+UPDATE t1 SET f2 = 'c' WHERE f1 > 1;
+connection node_2a;
+SET SESSION wsrep_sync_wait = 0;
+connect node_3, 127.0.0.1, root, , test, $NODE_MYPORT_3;
+connection node_3;
+SET SESSION wsrep_sync_wait = 0;
+connection node_2a;
+SET GLOBAL wsrep_provider_options = 'dbug=d,commit_monitor_enter_sync';
+SET GLOBAL debug_dbug = "d,sync.wsrep_apply_cb";
+connection node_3;
+INSERT INTO test.t1 VALUES (2, 'b');
+connection node_1;
+COMMIT;
+connection node_2a;
+SET SESSION wsrep_on = 0;
+SET SESSION wsrep_on = 1;
+SET GLOBAL debug_dbug = "";
+SET DEBUG_SYNC = "now SIGNAL signal.wsrep_apply_cb";
+connection node_2a;
+SET GLOBAL wsrep_provider_options = 'dbug=';
+SET GLOBAL wsrep_provider_options = 'signal=commit_monitor_enter_sync';
+connection node_1;
+SELECT COUNT(*) = 1 FROM t1 WHERE f2 = 'a';
+COUNT(*) = 1
+1
+SELECT COUNT(*) = 1 FROM t1 WHERE f2 = 'c';
+COUNT(*) = 1
+1
+SELECT * FROM t1;
+f1	f2
+1	a
+3	c
+connection node_2a;
+set session wsrep_sync_wait=15;
+set session wsrep_sync_wait=0;
+wsrep_local_replays
+1
+SELECT *  FROM t1;
+f1	f2
+1	a
+2	b
+3	c
+SET DEBUG_SYNC = "RESET";
+#
+# test phase with real abort
+#
+connection node_1;
+set binlog_format=ROW;
+insert into t1 values (4, 'd');
+SET AUTOCOMMIT=ON;
+START TRANSACTION;
+UPDATE t1 SET f2 = 'd' WHERE f1 = 3;
+connection node_2a;
+SET GLOBAL wsrep_provider_options = 'dbug=d,commit_monitor_enter_sync';
+SET GLOBAL debug_dbug = "d,sync.wsrep_apply_cb";
+connection node_3;
+UPDATE test.t1 SET f2 = 'e' WHERE f1 = 3;
+connection node_1;
+COMMIT;
+connection node_2a;
+SET GLOBAL debug_dbug = "";
+SET DEBUG_SYNC = "now SIGNAL signal.wsrep_apply_cb";
+connection node_2a;
+SET GLOBAL wsrep_provider_options = 'dbug=';
+SET GLOBAL wsrep_provider_options = 'signal=commit_monitor_enter_sync';
+SET DEBUG_SYNC = "RESET";
+connection node_2a;
+set session wsrep_sync_wait=15;
+SELECT COUNT(*) = 1 FROM test.t1 WHERE f2 = 'e';
+COUNT(*) = 1
+1
+set session wsrep_sync_wait=0;
+STOP SLAVE;
+RESET SLAVE;
+DROP TABLE t1;
+connection node_1;
+DROP TABLE t1;
+RESET MASTER;
diff --git a/mysql-test/suite/galera/r/galera_defaults.result b/mysql-test/suite/galera/r/galera_defaults.result
index 6dd5258ff6d..9a5c1e54b06 100644
--- a/mysql-test/suite/galera/r/galera_defaults.result
+++ b/mysql-test/suite/galera/r/galera_defaults.result
@@ -1,7 +1,8 @@
 connection node_2;
 connection node_1;
-SELECT COUNT(*) `expect 48` FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES WHERE VARIABLE_NAME LIKE 'wsrep_%';
-expect 48
+# Correct Galera library found
+SELECT COUNT(*) `expect 49` FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES WHERE VARIABLE_NAME LIKE 'wsrep_%';
+expect 49
 49
 SELECT VARIABLE_NAME, VARIABLE_VALUE
 FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES
diff --git a/mysql-test/suite/galera/r/galera_toi_alter_auto_increment.result b/mysql-test/suite/galera/r/galera_toi_alter_auto_increment.result
index a23b0523140..a9923ba4211 100644
--- a/mysql-test/suite/galera/r/galera_toi_alter_auto_increment.result
+++ b/mysql-test/suite/galera/r/galera_toi_alter_auto_increment.result
@@ -1,7 +1,7 @@
 connection node_2;
 connection node_1;
 connection node_1;
-CREATE TABLE ten (f1 INTEGER) ENGINE=InnoDB;
+CREATE TABLE ten (f1 INTEGER NOT NULL PRIMARY KEY) ENGINE=InnoDB;
 INSERT INTO ten VALUES (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
 CREATE TABLE t1 (f1 INTEGER AUTO_INCREMENT PRIMARY KEY, f2 INTEGER) ENGINE=InnoDB;
 INSERT INTO t1 (f2) SELECT 1 FROM ten;
diff --git a/mysql-test/suite/galera/t/MDEV-16509.test b/mysql-test/suite/galera/t/MDEV-16509.test
index da79cdc7d4c..eecc0e5e4e6 100644
--- a/mysql-test/suite/galera/t/MDEV-16509.test
+++ b/mysql-test/suite/galera/t/MDEV-16509.test
@@ -3,6 +3,7 @@
 #
 
 --source include/galera_cluster.inc
+--source include/have_debug.inc
 --source include/have_debug_sync.inc
 
 
diff --git a/mysql-test/suite/galera/t/galera#505.test b/mysql-test/suite/galera/t/galera#505.test
index 785b1411596..67ec1045119 100644
--- a/mysql-test/suite/galera/t/galera#505.test
+++ b/mysql-test/suite/galera/t/galera#505.test
@@ -5,12 +5,9 @@
 --connection node_1
 
 SET SESSION wsrep_sync_wait=0;
---disable_result_log
---disable_query_log
 --let $galera_version=25.3.24
 source ../../wsrep/include/check_galera_version.inc;
---enable_result_log
---enable_query_log
+
 SET SESSION wsrep_sync_wait=DEFAULT;
 
 # Convert "... pc.weight = N; ..." to "N; ..."
diff --git a/mysql-test/suite/galera/t/galera_as_slave_replay.cnf b/mysql-test/suite/galera/t/galera_as_slave_replay.cnf
new file mode 100644
index 00000000000..b1f9d7e9cbd
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_as_slave_replay.cnf
@@ -0,0 +1,11 @@
+!include ../galera_2nodes_as_slave.cnf
+
+[mysqld]
+binlog-format=row
+
+[mysqld.1]
+wsrep_restart_slave=1
+
+[mysqld.2]
+wsrep_restart_slave=1
+
diff --git a/mysql-test/suite/galera/t/galera_as_slave_replay.test b/mysql-test/suite/galera/t/galera_as_slave_replay.test
new file mode 100644
index 00000000000..93f95349e6d
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_as_slave_replay.test
@@ -0,0 +1,200 @@
+#
+# This test tests the operation of transaction replay for async replication slave.
+# If a potentially conflicting galera transaction arrives at
+# just the right time during the commit and has lock conflict with async replication transaction
+# applied by slave SQL thread, then the async replication transaction should either abort
+# or rollback and replay (depending on the nature of lock conflict).
+#
+
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+--source include/galera_have_debug_sync.inc
+
+--connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2
+
+--connection node_2a
+--source include/galera_cluster.inc
+#--source suite/galera/include/galera_have_debug_sync.inc
+
+#
+# node 1 is native MariaDB server operating as async replication master
+#
+--connection node_1
+RESET MASTER;
+
+--connection node_2a
+#
+# count the number of wsrep replay's done in the node
+#
+--let $wsrep_local_replays_old = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_local_replays'`
+
+
+#
+# nodes 2 and 3 form a galera cluster, node 2 operates as slave for native MariaDB naster in node 1
+#
+--disable_query_log
+--eval CHANGE MASTER TO  MASTER_HOST='127.0.0.1', MASTER_USER='root', MASTER_PORT=$NODE_MYPORT_1;
+--enable_query_log
+START SLAVE;
+
+--connection node_1
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY, f2 CHAR(1)) engine=innodb;
+INSERT INTO t1 VALUES (1, 'a');
+INSERT INTO t1 VALUES (3, 'a');
+
+#
+# use statement format replication to cause a false positive conflict with async replication transaction
+# and galera replication. The conflict will be on GAP lock, and slave SQL thread should rollback
+# and replay
+#
+set binlog_format=STATEMENT;
+
+SET AUTOCOMMIT=ON;
+START TRANSACTION;
+
+SELECT * FROM t1  FOR UPDATE;
+UPDATE t1 SET f2 = 'c' WHERE f1 > 1;
+
+--connection node_2a
+# wait for create table and inserts to be replicated from master
+SET SESSION wsrep_sync_wait = 0;
+--let $wait_condition = SELECT COUNT(*) = 2 FROM test.t1;
+--source include/wait_condition.inc
+
+# wait for create table and inserts to be replicated in cluster
+--connect node_3, 127.0.0.1, root, , test, $NODE_MYPORT_3
+--connection node_3
+SET SESSION wsrep_sync_wait = 0;
+--let $wait_condition = SELECT COUNT(*) = 2 FROM test.t1;
+--source include/wait_condition.inc
+
+--connection node_2a
+# Block the future commit of async replication
+--let $galera_sync_point = commit_monitor_enter_sync
+--source include/galera_set_sync_point.inc
+
+# block also the applier before applying begins
+SET GLOBAL debug_dbug = "d,sync.wsrep_apply_cb";
+
+#
+# now inject a conflicting insert from node 3, it will replicate with
+# earlier seqno (than async transaction) and pause before applying in node 2
+#
+--connection node_3
+INSERT INTO test.t1 VALUES (2, 'b');
+
+#
+# send the update from master, this will succeed here, beceuase of async replication.
+# async replication will apply this in node 2 and pause before commit phase,
+--connection node_1
+--error 0
+COMMIT;
+
+# Wait until async slave commit is blocked in node_2
+--connection node_2a
+--source include/galera_wait_sync_point.inc
+
+#
+# release the applier
+# note: have to clear wsrep_apply_cb sync point first, as async replication will go for replay
+# and as this sync point, after BF applier is released to progress
+#
+SET GLOBAL debug_dbug = "";
+SET DEBUG_SYNC = "now SIGNAL signal.wsrep_apply_cb";
+
+# Unblock the async slave commit
+--connection node_2a
+--source include/galera_clear_sync_point.inc
+--source include/galera_signal_sync_point.inc
+
+--connection node_1
+
+SELECT COUNT(*) = 1 FROM t1 WHERE f2 = 'a';
+SELECT COUNT(*) = 1 FROM t1 WHERE f2 = 'c';
+SELECT * FROM t1;
+
+--connection node_2a
+
+# wsrep_local_replays has increased by 1
+set session wsrep_sync_wait=15;
+--let $wsrep_local_replays_new = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_local_replays'`
+set session wsrep_sync_wait=0;
+
+--disable_query_log
+--eval SELECT $wsrep_local_replays_new - $wsrep_local_replays_old = 1 AS wsrep_local_replays;
+--enable_query_log
+
+#
+#  replaying of async transaction should be effective, and row 3 having 'c' in f2
+#
+SELECT *  FROM t1;
+SET DEBUG_SYNC = "RESET";
+
+#********************************************************************************
+#	test phase 2
+#********************************************************************************
+
+--echo #
+--echo # test phase with real abort
+--echo #
+
+--connection node_1
+
+set binlog_format=ROW;
+
+insert into t1 values (4, 'd');
+
+SET AUTOCOMMIT=ON;
+START TRANSACTION;
+
+UPDATE t1 SET f2 = 'd' WHERE f1 = 3;
+
+--connection node_2a
+# wait for the last insert to be replicated from master
+--let $wait_condition = SELECT COUNT(*) = 4 FROM test.t1;
+--source include/wait_condition.inc
+
+# Block the commit
+--let $galera_sync_point = commit_monitor_enter_sync
+--source include/galera_set_sync_point.inc
+
+# block applier
+SET GLOBAL debug_dbug = "d,sync.wsrep_apply_cb";
+
+# Inject a conflicting update from node 3
+--connection node_3
+UPDATE test.t1 SET f2 = 'e' WHERE f1 = 3;
+
+# send the update from master
+--connection node_1
+--error 0
+COMMIT;
+
+--connection node_2a
+
+# release the applier
+SET GLOBAL debug_dbug = "";
+SET DEBUG_SYNC = "now SIGNAL signal.wsrep_apply_cb";
+
+
+# Unblock the async slave commit
+--connection node_2a
+--source include/galera_clear_sync_point.inc
+--source include/galera_signal_sync_point.inc
+SET DEBUG_SYNC = "RESET";
+
+--connection node_2a
+
+set session wsrep_sync_wait=15;
+SELECT COUNT(*) = 1 FROM test.t1 WHERE f2 = 'e';
+set session wsrep_sync_wait=0;
+
+STOP SLAVE;
+RESET SLAVE;
+
+DROP TABLE t1;
+
+--connection node_1
+DROP TABLE t1;
+RESET MASTER;
diff --git a/mysql-test/suite/galera/t/galera_defaults.cnf b/mysql-test/suite/galera/t/galera_defaults.cnf
new file mode 100644
index 00000000000..fb143baf30f
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_defaults.cnf
@@ -0,0 +1,7 @@
+!include ../galera_2nodes.cnf
+
+[mysqld.1]
+wsrep_provider_options='base_port=@mysqld.1.#galera_port;gmcast.segment=1'
+
+[mysqld.2]
+wsrep_provider_options='base_port=@mysqld.2.#galera_port;gmcast.segment=1'
diff --git a/mysql-test/suite/galera/t/galera_defaults.test b/mysql-test/suite/galera/t/galera_defaults.test
index 3f8be268135..28e6f0cce38 100644
--- a/mysql-test/suite/galera/t/galera_defaults.test
+++ b/mysql-test/suite/galera/t/galera_defaults.test
@@ -1,25 +1,24 @@
 #
 # The purpose of this test is to preserve the current state of the following:
 # * SHOW VARIABLES LIKE 'wsrep%'
-# * wsrep_provider_options
 # * The names of the Galera status variables
 #
+# Note that wsrep_provider_options contains paths and other non-deterministic parts
+#
 # This way, if there is any change, inadvertent or not, the test will fail and the
 # developer and QA will be alerted.
 #
 
 --source include/galera_cluster.inc
---source include/have_innodb.inc
+--source include/force_restart.inc
 
 # Make sure that the test is operating on the right version of galera library.
---disable_query_log
---let $galera_version=25.3.20
+--let $galera_version=26.4.6
 source ../wsrep/include/check_galera_version.inc;
---enable_query_log
 
 # Global Variables
 
-SELECT COUNT(*) `expect 48` FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES WHERE VARIABLE_NAME LIKE 'wsrep_%';
+SELECT COUNT(*) `expect 49` FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES WHERE VARIABLE_NAME LIKE 'wsrep_%';
 
 SELECT VARIABLE_NAME, VARIABLE_VALUE
 FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES
diff --git a/mysql-test/suite/galera/t/galera_fk_cascade_delete.test b/mysql-test/suite/galera/t/galera_fk_cascade_delete.test
index 6f0de0a1f4a..a3e0dbcf36f 100644
--- a/mysql-test/suite/galera/t/galera_fk_cascade_delete.test
+++ b/mysql-test/suite/galera/t/galera_fk_cascade_delete.test
@@ -40,11 +40,19 @@ set wsrep_sync_wait=0;
 
 --let $wait_condition = SELECT COUNT(*) = 2 FROM child;
 --source include/wait_condition.inc
+--let $wait_condition = SELECT COUNT(*) = 2 FROM parent;
+--source include/wait_condition.inc
+--let $wait_condition = SELECT COUNT(*) = 2 FROM grandparent;
+--source include/wait_condition.inc
 DELETE FROM grandparent WHERE id = 1;
 
 --connection node_1
 --let $wait_condition = SELECT COUNT(*) = 1 FROM child;
 --source include/wait_condition.inc
+--let $wait_condition = SELECT COUNT(*) = 1 FROM parent;
+--source include/wait_condition.inc
+--let $wait_condition = SELECT COUNT(*) = 1 FROM grandparent;
+--source include/wait_condition.inc
 SELECT COUNT(*), COUNT(*) = 0 FROM parent WHERE grandparent_id = 1;
 SELECT COUNT(*), COUNT(*) = 0 FROM child WHERE parent_id = 1;
 
diff --git a/mysql-test/suite/galera/t/galera_rsu_simple.test b/mysql-test/suite/galera/t/galera_rsu_simple.test
index 5841dbd8006..aa6f25b6db6 100644
--- a/mysql-test/suite/galera/t/galera_rsu_simple.test
+++ b/mysql-test/suite/galera/t/galera_rsu_simple.test
@@ -8,6 +8,9 @@
 CREATE TABLE t1 (f1 INTEGER PRIMARY KEY) Engine=InnoDB;
 
 --connection node_2
+--let $wait_condition = SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 't1'
+--source include/wait_condition.inc
+
 SET SESSION wsrep_OSU_method = "RSU";
 ALTER TABLE t1 ADD COLUMN f2 INTEGER;
 SELECT COUNT(*) = 2 FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 't1';
diff --git a/mysql-test/suite/galera/t/galera_toi_alter_auto_increment.test b/mysql-test/suite/galera/t/galera_toi_alter_auto_increment.test
index 641d2101c80..793e87cb53e 100644
--- a/mysql-test/suite/galera/t/galera_toi_alter_auto_increment.test
+++ b/mysql-test/suite/galera/t/galera_toi_alter_auto_increment.test
@@ -7,7 +7,7 @@
 --source include/have_innodb.inc
 
 --connection node_1
-CREATE TABLE ten (f1 INTEGER) ENGINE=InnoDB;
+CREATE TABLE ten (f1 INTEGER NOT NULL PRIMARY KEY) ENGINE=InnoDB;
 INSERT INTO ten VALUES (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
 
 CREATE TABLE t1 (f1 INTEGER AUTO_INCREMENT PRIMARY KEY, f2 INTEGER) ENGINE=InnoDB;
@@ -83,6 +83,8 @@ SET GLOBAL auto_increment_offset = 1;
 CREATE TABLE t1 (f1 INTEGER AUTO_INCREMENT PRIMARY KEY, f2 INTEGER) ENGINE=InnoDB;
 
 --connection node_2a
+--let $wait_condition = SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 't1'
+--source include/wait_condition.inc
 
 ALTER TABLE t1 AUTO_INCREMENT=100;
 
diff --git a/mysql-test/suite/galera/t/galera_var_ignore_apply_errors.test b/mysql-test/suite/galera/t/galera_var_ignore_apply_errors.test
index 5232d4236e6..0f8efad5163 100644
--- a/mysql-test/suite/galera/t/galera_var_ignore_apply_errors.test
+++ b/mysql-test/suite/galera/t/galera_var_ignore_apply_errors.test
@@ -75,6 +75,9 @@ DELETE FROM t1 WHERE f1 = 1;
 --connection node_1
 SELECT COUNT(*) as expect_0 FROM t1;
 --connection node_2
+--source include/galera_wait_ready.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 2 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_size';
+--source include/wait_condition.inc
 SELECT COUNT(*) as expect_0 FROM t1;
 
 DROP TABLE t1;
@@ -96,6 +99,9 @@ COMMIT;
 --connection node_1
 SELECT COUNT(*) as expect_1 FROM t1;
 --connection node_2
+--source include/galera_wait_ready.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 2 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_size';
+--source include/wait_condition.inc
 SELECT COUNT(*) as expect_1 FROM t1;
 
 DROP TABLE t1;
@@ -120,6 +126,11 @@ DELETE FROM t1;
 
 SELECT COUNT(*) as expect_0 FROM t1;
 --connection node_2
+--source include/galera_wait_ready.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 2 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_size';
+--source include/wait_condition.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
+--source include/wait_condition.inc
 SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
 SELECT COUNT(*) as expect_0 FROM t1;
 DROP TABLE t1;
@@ -154,6 +165,11 @@ SET AUTOCOMMIT=ON;
 
 SELECT COUNT(*) as expect_0 FROM t1;
 --connection node_2
+--source include/galera_wait_ready.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 2 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_size';
+--source include/wait_condition.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
+--source include/wait_condition.inc
 SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
 SELECT COUNT(*) as expect_0 FROM t1;
 DROP TABLE t1;
@@ -183,6 +199,11 @@ DELETE t1, t2 FROM t1 JOIN t2 WHERE t1.f1 = t2.f1;
 SELECT COUNT(*) as expect_0 FROM t1;
 
 --connection node_2
+--source include/galera_wait_ready.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 2 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_size';
+--source include/wait_condition.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
+--source include/wait_condition.inc
 SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
 SELECT COUNT(*) as expect_0 FROM t1;
 DROP TABLE t1,t2;
@@ -212,6 +233,11 @@ SELECT COUNT(*) as expect_0 FROM parent;
 SELECT COUNT(*) as expect_0 FROM child;
 
 --connection node_2
+--source include/galera_wait_ready.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 2 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_size';
+--source include/wait_condition.inc
+--let $wait_condition = SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
+--source include/wait_condition.inc
 SELECT VARIABLE_VALUE = 'Primary' FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_cluster_status';
 SELECT COUNT(*) as expect_0 FROM parent;
 SELECT COUNT(*) as expect_0 FROM child;
diff --git a/mysql-test/suite/galera_3nodes/r/galera_parallel_apply_3nodes.result b/mysql-test/suite/galera_3nodes/r/galera_parallel_apply_3nodes.result
index 2acfe56bede..087f9e7e279 100644
--- a/mysql-test/suite/galera_3nodes/r/galera_parallel_apply_3nodes.result
+++ b/mysql-test/suite/galera_3nodes/r/galera_parallel_apply_3nodes.result
@@ -1,27 +1,35 @@
 connection node_2;
 connection node_1;
+connect node_3, 127.0.0.1, root, , test, $NODE_MYPORT_3;
+connect node_1_ctrl, 127.0.0.1, root, , test, $NODE_MYPORT_1;
 CREATE TABLE t1 (f1 INTEGER PRIMARY KEY) ENGINE=InnoDB;
 INSERT INTO t1 VALUES (1);
 connection node_3;
 SET GLOBAL wsrep_slave_threads = 2;
+connection node_1_ctrl;
+SET SESSION wsrep_sync_wait=0;
 connection node_1;
+SET DEBUG_SYNC = 'wsrep_before_certification SIGNAL before_cert WAIT_FOR continue';
 UPDATE t1 SET f1 = f1 + 10;;
+connection node_1_ctrl;
+SET DEBUG_SYNC = 'now WAIT_FOR before_cert';
+SET GLOBAL debug_dbug = '+d,sync.wsrep_retry_autocommit';
 connection node_2;
 UPDATE t1 SET f1 = f1 + 100;;
+connection node_1_ctrl;
+SET DEBUG_SYNC = 'now WAIT_FOR wsrep_retry_autocommit_reached';
+SET GLOBAL debug_dbug = NULL;
+SET DEBUG_SYNC = 'now SIGNAL wsrep_retry_autocommit_continue';
 connection node_1;
-SELECT COUNT(*) FROM t1;
-COUNT(*)
-1
 connection node_2;
-SELECT COUNT(*) FROM t1;
-COUNT(*)
-1
 connection node_3;
-SELECT COUNT(*) FROM t1;
-COUNT(*)
+SELECT f1 = 111 FROM t1;
+f1 = 111
 1
 SELECT COUNT(*) IN (1, 2) FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'system user' AND STATE LIKE '%committed%';
 COUNT(*) IN (1, 2)
 1
-SET GLOBAL wsrep_slave_threads = 1;;
+SET GLOBAL wsrep_slave_threads = DEFAULT;
 DROP TABLE t1;
+connection node_1;
+SET DEBUG_SYNC= 'RESET';
diff --git a/mysql-test/suite/galera_3nodes/t/galera_parallel_apply_3nodes.test b/mysql-test/suite/galera_3nodes/t/galera_parallel_apply_3nodes.test
index f1168e59193..84629c96c65 100644
--- a/mysql-test/suite/galera_3nodes/t/galera_parallel_apply_3nodes.test
+++ b/mysql-test/suite/galera_3nodes/t/galera_parallel_apply_3nodes.test
@@ -5,46 +5,71 @@
 
 --source include/galera_cluster.inc
 --source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
 
---let $galera_connection_name = node_3
---let $galera_server_number = 3
---source include/galera_connect.inc
+--connect node_3, 127.0.0.1, root, , test, $NODE_MYPORT_3
+--connect node_1_ctrl, 127.0.0.1, root, , test, $NODE_MYPORT_1
 
 CREATE TABLE t1 (f1 INTEGER PRIMARY KEY) ENGINE=InnoDB;
 INSERT INTO t1 VALUES (1);
 
+--let $wsrep_last_committed_before = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME = 'wsrep_last_committed'`
+
 --connection node_3
---let $wsrep_slave_threads_orig = `SELECT @@wsrep_slave_threads`
 SET GLOBAL wsrep_slave_threads = 2;
 
+--connection node_1_ctrl
+SET SESSION wsrep_sync_wait=0;
+
+#
+# We will make the following UPDATE depend on the UPDATE below
+#
 --connection node_1
+SET DEBUG_SYNC = 'wsrep_before_certification SIGNAL before_cert WAIT_FOR continue';
 --send UPDATE t1 SET f1 = f1 + 10;
 
+--connection node_1_ctrl
+SET DEBUG_SYNC = 'now WAIT_FOR before_cert';
+SET GLOBAL debug_dbug = '+d,sync.wsrep_retry_autocommit';
+
 --connection node_2
 --send UPDATE t1 SET f1 = f1 + 100;
 
---connection node_1
 #
-# Note that test is not deterministic. We have following cases possible
-# (1) Both updates are certified locally and then executed by the applier
-# (2) Certification of update in node_1 fails because applier has started
-#     update from node_2
-# (3) Certification of update in node_2 fails because applier has started
-#     update from node_1
+# Let's wait for the first UPDATE the be BF aborted
+#
+--connection node_1_ctrl
+SET DEBUG_SYNC = 'now WAIT_FOR wsrep_retry_autocommit_reached';
+
+#
+# and make sure the second has committed
 #
---error 0,ER_LOCK_DEADLOCK
+--let $wait_condition = SELECT VARIABLE_VALUE > $wsrep_last_committed_before FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME = 'wsrep_last_committed'
+--source include/wait_condition.inc
+
+#
+# now release the first UPDATE.
+#
+SET GLOBAL debug_dbug = NULL;
+SET DEBUG_SYNC = 'now SIGNAL wsrep_retry_autocommit_continue';
+
+#
+# Both UPDATEs should succeed.
+#
+--connection node_1
 --reap
-SELECT COUNT(*) FROM t1;
 
 --connection node_2
---error 0,ER_LOCK_DEADLOCK
 --reap
-SELECT COUNT(*) FROM t1;
 
 --connection node_3
-SELECT COUNT(*) FROM t1;
+SELECT f1 = 111 FROM t1;
 SELECT COUNT(*) IN (1, 2) FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'system user' AND STATE LIKE '%committed%';
 
---eval SET GLOBAL wsrep_slave_threads = $wsrep_slave_threads_orig;
+SET GLOBAL wsrep_slave_threads = DEFAULT;
 
 DROP TABLE t1;
+
+--connection node_1
+SET DEBUG_SYNC= 'RESET';
diff --git a/mysql-test/suite/innodb/r/foreign_key.result b/mysql-test/suite/innodb/r/foreign_key.result
index cd8b2d82a06..399e35b698f 100644
--- a/mysql-test/suite/innodb/r/foreign_key.result
+++ b/mysql-test/suite/innodb/r/foreign_key.result
@@ -409,6 +409,7 @@ CREATE TABLE x AS SELECT * FROM t1;
 ERROR XAE07: XAER_RMFAIL: The command cannot be executed when global transaction is in the  ACTIVE state
 connect  con1,localhost,root,,test;
 SET foreign_key_checks= OFF, innodb_lock_wait_timeout= 1;
+SET lock_wait_timeout=5;
 ALTER TABLE t1 ADD FOREIGN KEY f (a) REFERENCES t1 (pk), LOCK=EXCLUSIVE;
 ERROR HY000: Lock wait timeout exceeded; try restarting transaction
 disconnect con1;
diff --git a/mysql-test/suite/innodb/r/instant_alter_crash.result b/mysql-test/suite/innodb/r/instant_alter_crash.result
index ef0e310492a..a2c388fa103 100644
--- a/mysql-test/suite/innodb/r/instant_alter_crash.result
+++ b/mysql-test/suite/innodb/r/instant_alter_crash.result
@@ -74,7 +74,6 @@ DELETE FROM t1;
 # Kill the server
 disconnect ddl;
 # restart
-SET @saved_frequency= @@GLOBAL.innodb_purge_rseg_truncate_frequency;
 SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
 FOUND 3 /\[Note\] InnoDB: Rolled back recovered transaction / in mysqld.1.err
 SELECT * FROM t1;
@@ -138,6 +137,23 @@ header=0x080008030000 (id=0x000000000000000100)
 UNLOCK TABLES;
 DELETE FROM t2;
 InnoDB		0 transactions not purged
+#
+# MDEV-24323 Crash on recovery after kill during instant ADD COLUMN
+#
+connect ddl, localhost, root;
+CREATE TABLE t3(id INT PRIMARY KEY, c2 INT, v2 INT AS(c2) VIRTUAL, UNIQUE(v2))
+ENGINE=InnoDB;
+INSERT INTO t3 SET id=1,c2=1;
+SET DEBUG_SYNC='innodb_alter_inplace_before_commit SIGNAL ddl WAIT_FOR ever';
+ALTER TABLE t3 ADD COLUMN c3 TEXT NOT NULL DEFAULT 'sic transit gloria mundi';
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR ddl';
+SET GLOBAL innodb_flush_log_at_trx_commit=1;
+SET debug_dbug='+d,dict_sys_mutex_avoid';
+INSERT INTO t1 VALUES(0,0);
+# Kill the server
+disconnect ddl;
+# restart
 SHOW CREATE TABLE t1;
 Table	Create Table
 t1	CREATE TABLE `t1` (
@@ -154,6 +170,14 @@ t2	CREATE TABLE `t2` (
   PRIMARY KEY (`id`),
   UNIQUE KEY `c2` (`c2`)
 ) ENGINE=InnoDB DEFAULT CHARSET=latin1 ROW_FORMAT=REDUNDANT
-DROP TABLE t1,t2;
+SHOW CREATE TABLE t3;
+Table	Create Table
+t3	CREATE TABLE `t3` (
+  `id` int(11) NOT NULL,
+  `c2` int(11) DEFAULT NULL,
+  `v2` int(11) GENERATED ALWAYS AS (`c2`) VIRTUAL,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `v2` (`v2`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+DROP TABLE t1,t2,t3;
 db.opt
-SET GLOBAL innodb_purge_rseg_truncate_frequency=@saved_frequency;
diff --git a/mysql-test/suite/innodb/t/foreign_key.test b/mysql-test/suite/innodb/t/foreign_key.test
index ebde222ee69..94d736e97d4 100644
--- a/mysql-test/suite/innodb/t/foreign_key.test
+++ b/mysql-test/suite/innodb/t/foreign_key.test
@@ -414,6 +414,7 @@ INSERT INTO t1 VALUES (1,2);
 CREATE TABLE x AS SELECT * FROM t1;
 --connect (con1,localhost,root,,test)
 SET foreign_key_checks= OFF, innodb_lock_wait_timeout= 1;
+SET lock_wait_timeout=5;
 --error ER_LOCK_WAIT_TIMEOUT
 ALTER TABLE t1 ADD FOREIGN KEY f (a) REFERENCES t1 (pk), LOCK=EXCLUSIVE;# Cleanup
 --disconnect con1
diff --git a/mysql-test/suite/innodb/t/instant_alter_crash.test b/mysql-test/suite/innodb/t/instant_alter_crash.test
index f9b18aa589f..43db8f619f3 100644
--- a/mysql-test/suite/innodb/t/instant_alter_crash.test
+++ b/mysql-test/suite/innodb/t/instant_alter_crash.test
@@ -91,7 +91,6 @@ DELETE FROM t1;
 disconnect ddl;
 --source include/start_mysqld.inc
 
-SET @saved_frequency= @@GLOBAL.innodb_purge_rseg_truncate_frequency;
 SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
 
 let SEARCH_FILE= $MYSQLTEST_VARDIR/log/mysqld.1.err;
@@ -177,11 +176,32 @@ UNLOCK TABLES;
 DELETE FROM t2;
 --source include/wait_all_purged.inc
 
+--echo #
+--echo # MDEV-24323 Crash on recovery after kill during instant ADD COLUMN
+--echo #
+connect ddl, localhost, root;
+CREATE TABLE t3(id INT PRIMARY KEY, c2 INT, v2 INT AS(c2) VIRTUAL, UNIQUE(v2))
+ENGINE=InnoDB;
+INSERT INTO t3 SET id=1,c2=1;
+
+SET DEBUG_SYNC='innodb_alter_inplace_before_commit SIGNAL ddl WAIT_FOR ever';
+--send
+ALTER TABLE t3 ADD COLUMN c3 TEXT NOT NULL DEFAULT 'sic transit gloria mundi';
+
+connection default;
+SET DEBUG_SYNC='now WAIT_FOR ddl';
+SET GLOBAL innodb_flush_log_at_trx_commit=1;
+SET debug_dbug='+d,dict_sys_mutex_avoid';
+INSERT INTO t1 VALUES(0,0);
+
+--source include/kill_mysqld.inc
+disconnect ddl;
+--source include/start_mysqld.inc
+
 SHOW CREATE TABLE t1;
 SHOW CREATE TABLE t2;
-DROP TABLE t1,t2;
+SHOW CREATE TABLE t3;
+DROP TABLE t1,t2,t3;
 
 --remove_files_wildcard $MYSQLD_DATADIR/test #sql*.frm
 --list_files $MYSQLD_DATADIR/test
-
-SET GLOBAL innodb_purge_rseg_truncate_frequency=@saved_frequency;
diff --git a/mysql-test/suite/maria/create.opt b/mysql-test/suite/maria/create.opt
new file mode 100644
index 00000000000..b1392bfd485
--- /dev/null
+++ b/mysql-test/suite/maria/create.opt
@@ -0,0 +1 @@
+--symbolic-links=1
diff --git a/mysql-test/suite/mariabackup/include/corrupt-page.pl b/mysql-test/suite/mariabackup/include/corrupt-page.pl
new file mode 100644
index 00000000000..d5c75dbde55
--- /dev/null
+++ b/mysql-test/suite/mariabackup/include/corrupt-page.pl
@@ -0,0 +1,146 @@
+use strict;
+use warnings;
+use Fcntl qw(:DEFAULT :seek);
+do "$ENV{MTR_SUITE_DIR}/../innodb/include/crc32.pl";
+
+sub corrupt_space_page_id {
+  my $file_name = shift;
+  my @pages_to_corrupt = @_;
+
+  my $page_size = $ENV{INNODB_PAGE_SIZE};
+
+  sysopen my $ibd_file, $file_name, O_RDWR || die "Cannot open $file_name\n";
+  sysread($ibd_file, $_, 38) || die "Cannot read $file_name\n";
+  my $space = unpack("x[34]N", $_);
+  foreach my $page_no (@pages_to_corrupt) {
+    $space += 10; # generate wrong space id
+    sysseek($ibd_file, $page_size * $page_no, SEEK_SET)
+      || die "Cannot seek $file_name\n";
+
+    my $head = pack("Nx[18]", $page_no + 10); # generate wrong page number
+    my $body = chr(0) x ($page_size - 38 - 8);
+
+    # Calculate innodb_checksum_algorithm=crc32 for the unencrypted page.
+    # The following bytes are excluded:
+    # bytes 0..3 (the checksum is stored there)
+    # bytes 26..37 (encryption key version, post-encryption checksum, tablespace id)
+    # bytes $page_size-8..$page_size-1 (checksum, LSB of FIL_PAGE_LSN)
+    my $polynomial = 0x82f63b78; # CRC-32C
+    my $ck = mycrc32($head, 0, $polynomial) ^ mycrc32($body, 0, $polynomial);
+
+    my $page= pack("N",$ck).$head.pack("NNN",1,$ck,$space).$body.pack("Nx[4]",$ck);
+    die unless syswrite($ibd_file, $page, $page_size) == $page_size;
+  }
+  close $ibd_file;
+}
+
+sub extend_space {
+  my $file_name = shift;
+  my $n_pages = shift;
+
+  my $page_size = $ENV{INNODB_PAGE_SIZE};
+  my $page;
+
+  sysopen my $ibd_file, $file_name, O_RDWR || die "Cannot open $file_name\n";
+  sysread($ibd_file, $page, $page_size)
+    || die "Cannot read $file_name\n";
+  my $size = unpack("N", substr($page, 46, 4));
+  my $packed_new_size = pack("N", $size + $n_pages);
+  substr($page, 46, 4, $packed_new_size);
+
+  my $head = substr($page, 4, 22);
+  my $body = substr($page, 38, $page_size - 38 - 8);
+  my $polynomial = 0x82f63b78; # CRC-32C
+  my $ck = mycrc32($head, 0, $polynomial) ^ mycrc32($body, 0, $polynomial);
+  my $packed_ck = pack("N", $ck);
+  substr($page, 0, 4, $packed_ck);
+  substr($page, $page_size - 8, 4, $packed_ck);
+
+  sysseek($ibd_file, 0, SEEK_SET)
+      || die "Cannot seek $file_name\n";
+  die unless syswrite($ibd_file, $page, $page_size) == $page_size;
+
+  sysseek($ibd_file, 0, SEEK_END)
+      || die "Cannot seek $file_name\n";
+  my $pages_size = $page_size*$n_pages;
+  my $pages = chr(0) x $pages_size;
+  die unless syswrite($ibd_file, $pages, $pages_size) == $pages_size;
+  close $ibd_file;
+  return $size;
+}
+
+sub die_if_page_is_not_zero {
+  my $file_name = shift;
+  my @pages_to_check = @_;
+
+  no locale;
+  my $page_size = $ENV{INNODB_PAGE_SIZE};
+  my $zero_page = chr(0) x $page_size;
+  sysopen my $ibd_file, $file_name, O_RDWR || die "Cannot open $file_name\n";
+  foreach my $page_no_to_check (@pages_to_check) {
+    sysseek($ibd_file, $page_size*$page_no_to_check, SEEK_SET) ||
+      die "Cannot seek $file_name\n";
+    sysread($ibd_file, my $read_page, $page_size) ||
+      die "Cannot read $file_name\n";
+    die "The page $page_no_to_check is not zero-filed in $file_name"
+      if ($read_page cmp $zero_page);
+  }
+  close $ibd_file;
+}
+
+sub print_corrupted_pages_file {
+  my $file_in = shift;
+  my $file_out = shift;
+  open my $fh, '<', $file_in || die $!;
+  my $line_number = 0;
+  my $space = {};
+  my @spaces;
+  while (my $line = <$fh>) {
+    ++$line_number;
+    if ($line_number & 1) {
+      my ($name, $id) = split(/ /, $line);
+      $space->{name} = $name;
+    }
+    else {
+      $space->{pages} = $line;
+      push (@spaces, $space);
+      $space = {};
+    }
+  }
+  close $fh;
+  my @sorted_spaces = sort { $a->{name} cmp $b->{name} } @spaces;
+  open $fh, '>', $file_out || die $!;
+  foreach my $space (@sorted_spaces) {
+    print $fh $space->{name};
+    print $fh "\n";
+    print $fh $space->{pages};
+  }
+  close $fh;
+}
+
+sub append_corrupted_pages {
+  my $file_name = shift;
+  my $space_name = shift;
+  my $pages = shift;
+  open my $fh, '<', $file_name || die $!;
+  my $line_number = 0;
+  my $space_line;
+  while (my $line = <$fh>) {
+    ++$line_number;
+    if ($line_number & 1) {
+      my ($name, $id) = split(/ /, $line);
+      if ($name eq $space_name) {
+        $space_line = $line;
+        last;
+      }
+    }
+  }
+  close $fh;
+  if (not defined $space_line) {
+    die "Can't find requested space $space_name in file $file_name";
+  }
+  open $fh, '>>', $file_name || die $!;
+  print $fh $space_line;
+  print $fh "$pages\n";
+  close $fh;
+}
diff --git a/mysql-test/suite/mariabackup/incremental_ddl_during_backup.test b/mysql-test/suite/mariabackup/incremental_ddl_during_backup.test
index b1ab17a6d8f..ebdb2137523 100644
--- a/mysql-test/suite/mariabackup/incremental_ddl_during_backup.test
+++ b/mysql-test/suite/mariabackup/incremental_ddl_during_backup.test
@@ -22,7 +22,7 @@ INSERT into t1 values(1);
 --let after_copy_test_t2=DROP TABLE test.t2
 --let after_copy_test_t3=CREATE INDEX a_i ON test.t3(i);
 --let before_copy_test_t10=DROP TABLE test.t10
---let wait_innodb_redo_before_copy=test/t10
+--let wait_innodb_redo_before_copy_test_t10 = 1
 
 # mariabackup should crash with assertion if MDEV-24026 is not fixed
 exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$incremental_dir --incremental-basedir=$basedir --dbug=+d,mariabackup_events,mariabackup_inject_code;
diff --git a/mysql-test/suite/mariabackup/log_page_corruption.opt b/mysql-test/suite/mariabackup/log_page_corruption.opt
new file mode 100644
index 00000000000..c44c611ed60
--- /dev/null
+++ b/mysql-test/suite/mariabackup/log_page_corruption.opt
@@ -0,0 +1 @@
+--innodb-checksum-algorithm=crc32
diff --git a/mysql-test/suite/mariabackup/log_page_corruption.result b/mysql-test/suite/mariabackup/log_page_corruption.result
new file mode 100644
index 00000000000..be29ea435b6
--- /dev/null
+++ b/mysql-test/suite/mariabackup/log_page_corruption.result
@@ -0,0 +1,145 @@
+########
+# Test for generating "innodb_corrupted_pages" file during full and
+# incremental backup, including DDL processing
+###
+
+CREATE TABLE t1_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t2_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t3(c INT) ENGINE INNODB;
+CREATE TABLE t5_corrupted_to_rename(c INT) ENGINE INNODB;
+CREATE TABLE t6_corrupted_to_drop(c INT) ENGINE INNODB;
+CREATE TABLE t7_corrupted_to_alter(c INT) ENGINE INNODB;
+CREATE TABLE t1_inc_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t2_inc_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t3_inc(c INT) ENGINE INNODB;
+CREATE TABLE t5_inc_corrupted_to_rename(c INT) ENGINE INNODB;
+CREATE TABLE t6_inc_corrupted_to_drop(c INT) ENGINE INNODB;
+CREATE TABLE t7_inc_corrupted_to_alter(c INT) ENGINE INNODB;
+INSERT INTO t1_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t2_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t3 VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t5_corrupted_to_rename VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t6_corrupted_to_drop VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t7_corrupted_to_alter VALUES (3), (4), (5), (6), (7), (8), (9);
+# Corrupt tables
+# restart
+# Backup must fail due to page corruption
+FOUND 1 /Database page corruption detected.*/ in backup.log
+# "innodb_corrupted_pages" file must not exist
+# Backup must fail, but "innodb_corrupted_pages" file must be created due to --log-innodb-page-corruption option
+FOUND 1 /Database page corruption detected.*/ in backup.log
+--- "innodb_corrupted_pages" file content: ---
+test/t1_corrupted
+6 8 9
+test/t2_corrupted
+7 8 10
+test/t4_corrupted_new
+1
+test/t5_corrupted_to_rename_renamed
+6
+test/t7_corrupted_to_alter
+3
+------
+INSERT INTO t1_inc_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t2_inc_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t3_inc VALUES (3), (4), (5), (6), (7), (8), (9);
+# restart
+# Backup must fail, but "innodb_corrupted_pages" file must be created due to --log-innodb-page-corruption option
+--- "innodb_corrupted_pages" file content: ---
+test/t1_corrupted
+6 8 9
+test/t1_inc_corrupted
+6 8 9
+test/t2_corrupted
+7 8 10
+test/t2_inc_corrupted
+7 8 10
+test/t4_inc_corrupted_new
+1
+test/t5_corrupted_to_rename_renamed
+6
+test/t5_inc_corrupted_to_rename_renamed
+6
+test/t7_inc_corrupted_to_alter
+3
+------
+# Check if corrupted pages were copied to delta files, and non-corrupted pages are not copied.
+DROP TABLE t1_corrupted;
+DROP TABLE t2_corrupted;
+DROP TABLE t4_corrupted_new;
+DROP TABLE t5_corrupted_to_rename_renamed;
+DROP TABLE t7_corrupted_to_alter;
+DROP TABLE t1_inc_corrupted;
+DROP TABLE t2_inc_corrupted;
+DROP TABLE t4_inc_corrupted_new;
+DROP TABLE t5_inc_corrupted_to_rename_renamed;
+DROP TABLE t7_inc_corrupted_to_alter;
+
+########
+# Test for --prepare with "innodb_corrupted_pages" file
+###
+
+# Extend some tablespace and corrupt extended pages for full backup
+# restart
+# Full backup with --log-innodb-page-corruption
+--- "innodb_corrupted_pages" file content: ---
+test/t3
+6 8
+------
+# Extend some tablespace and corrupt extended pages for incremental backup
+# restart
+# Incremental backup --log-innodb-page-corruption
+--- "innodb_corrupted_pages" file content: ---
+test/t3
+6 8
+test/t3_inc
+6 8
+------
+# Full backup prepare
+# "innodb_corrupted_pages" file must not exist after successful prepare
+FOUND 1 /was successfuly fixed.*/ in backup.log
+# Check that fixed pages are zero-filled
+# Incremental backup prepare
+# "innodb_corrupted_pages" file must not exist after successful prepare
+# do not remove "innodb_corrupted_pages" in incremental dir
+FOUND 1 /was successfuly fixed.*/ in backup.log
+# Check that fixed pages are zero-filled
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart
+SELECT * FROM t3;
+c
+3
+4
+5
+6
+7
+8
+9
+SELECT * FROM t3_inc;
+c
+3
+4
+5
+6
+7
+8
+9
+# Test the case when not all corrupted pages are fixed
+
+# Add some fake corrupted pages
+# Full backup prepare
+FOUND 1 /Error: corrupted page.*/ in backup.log
+--- "innodb_corrupted_pages" file content: ---
+test/t3
+3
+------
+# Incremental backup prepare
+FOUND 1 /Error: corrupted page.*/ in backup.log
+--- "innodb_corrupted_pages" file content: ---
+test/t3
+3
+------
+DROP TABLE t3;
+DROP TABLE t3_inc;
diff --git a/mysql-test/suite/mariabackup/log_page_corruption.test b/mysql-test/suite/mariabackup/log_page_corruption.test
new file mode 100644
index 00000000000..e9419687288
--- /dev/null
+++ b/mysql-test/suite/mariabackup/log_page_corruption.test
@@ -0,0 +1,426 @@
+--source include/have_debug.inc
+
+--echo ########
+--echo # Test for generating "innodb_corrupted_pages" file during full and
+--echo # incremental backup, including DDL processing
+--echo ###
+--echo
+
+CREATE TABLE t1_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t2_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t3(c INT) ENGINE INNODB;
+CREATE TABLE t5_corrupted_to_rename(c INT) ENGINE INNODB;
+CREATE TABLE t6_corrupted_to_drop(c INT) ENGINE INNODB;
+CREATE TABLE t7_corrupted_to_alter(c INT) ENGINE INNODB;
+
+CREATE TABLE t1_inc_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t2_inc_corrupted(c INT) ENGINE INNODB;
+CREATE TABLE t3_inc(c INT) ENGINE INNODB;
+CREATE TABLE t5_inc_corrupted_to_rename(c INT) ENGINE INNODB;
+CREATE TABLE t6_inc_corrupted_to_drop(c INT) ENGINE INNODB;
+CREATE TABLE t7_inc_corrupted_to_alter(c INT) ENGINE INNODB;
+
+# Fill tables with several pages
+INSERT INTO t1_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t2_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t3 VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t5_corrupted_to_rename VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t6_corrupted_to_drop VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t7_corrupted_to_alter VALUES (3), (4), (5), (6), (7), (8), (9);
+
+--let MYSQLD_DATADIR=`select @@datadir`
+--let INNODB_PAGE_SIZE=`select @@innodb_page_size`
+
+--source include/shutdown_mysqld.inc
+--echo # Corrupt tables
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+my $schema = "$ENV{MYSQLD_DATADIR}/test";
+
+my $last_page_no = extend_space("$schema/t1_corrupted.ibd", 4);
+corrupt_space_page_id("$schema/t1_corrupted.ibd",
+  $last_page_no, $last_page_no + 2, $last_page_no + 3);
+
+$last_page_no = extend_space("$schema/t2_corrupted.ibd", 5);
+corrupt_space_page_id("$schema/t2_corrupted.ibd",
+  $last_page_no + 1, $last_page_no + 2, $last_page_no + 4);
+
+$last_page_no = extend_space("$schema/t5_corrupted_to_rename.ibd", 1);
+corrupt_space_page_id("$schema/t5_corrupted_to_rename.ibd", $last_page_no);
+
+$last_page_no = extend_space("$schema/t6_corrupted_to_drop.ibd", );
+corrupt_space_page_id("$schema/t6_corrupted_to_drop.ibd", $last_page_no);
+EOF
+--source include/start_mysqld.inc
+
+--let targetdir=$MYSQLTEST_VARDIR/tmp/backup
+--let $backuplog=$MYSQLTEST_VARDIR/tmp/backup.log
+--let corrupted_pages_file = $targetdir/innodb_corrupted_pages
+--let corrupted_pages_file_filt = $MYSQLTEST_VARDIR/tmp/innodb_corrupted_pages_filt
+--let perl_result_file=$MYSQLTEST_VARDIR/tmp/perl_result
+
+--echo # Backup must fail due to page corruption
+--disable_result_log
+--error 1
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$targetdir > $backuplog;
+--enable_result_log
+
+--let SEARCH_PATTERN=Database page corruption detected.*
+--let SEARCH_FILE=$backuplog
+--source include/search_pattern_in_file.inc
+--echo # "innodb_corrupted_pages" file must not exist
+--error 1
+--file_exists $corrupted_pages_file
+--rmdir $targetdir
+
+--let after_load_tablespaces=CREATE TABLE test.t4_corrupted_new ENGINE=INNODB SELECT UUID() from test.seq_1_to_10
+--let add_corrupted_page_for_test_t4_corrupted_new=1
+--let after_copy_test_t5_corrupted_to_rename=RENAME TABLE test.t5_corrupted_to_rename TO test.t5_corrupted_to_rename_renamed
+--let after_copy_test_t6_corrupted_to_drop=DROP TABLE test.t6_corrupted_to_drop
+--let after_copy_test_t7_corrupted_to_alter=ALTER TABLE test.t7_corrupted_to_alter ADD COLUMN (d INT)
+--let add_corrupted_page_for_test_t7_corrupted_to_alter=3
+
+--echo # Backup must fail, but "innodb_corrupted_pages" file must be created due to --log-innodb-page-corruption option
+--disable_result_log
+--error 1
+--exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --log-innodb-page-corruption --target-dir=$targetdir --dbug=+d,mariabackup_events,mariabackup_inject_code  > $backuplog
+--enable_result_log
+
+--let SEARCH_PATTERN=Database page corruption detected.*
+--let SEARCH_FILE=$backuplog
+--source include/search_pattern_in_file.inc
+--echo --- "innodb_corrupted_pages" file content: ---
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+print_corrupted_pages_file($ENV{corrupted_pages_file},
+  $ENV{corrupted_pages_file_filt});
+EOF
+--cat_file $corrupted_pages_file_filt
+--echo ------
+--let after_load_tablespaces=
+--let add_corrupted_page_for_test_t4_corrupted_new=
+--let after_copy_test_t5_corrupted_to_rename=
+--let after_copy_test_t6_corrupted_to_drop=
+--let after_copy_test_t7_corrupted_to_alter=
+--let add_corrupted_page_for_test_t7_corrupted_to_alter=
+# Fill tables for incremental backup with several pages
+INSERT INTO t1_inc_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t2_inc_corrupted VALUES (3), (4), (5), (6), (7), (8), (9);
+INSERT INTO t3_inc VALUES (3), (4), (5), (6), (7), (8), (9);
+
+--source include/shutdown_mysqld.inc
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+my $schema="$ENV{MYSQLD_DATADIR}/test";
+
+open(my $fh, '>', $ENV{perl_result_file}) or die $!;
+
+my $last_page_no = extend_space("$schema/t1_inc_corrupted.ibd", 4);
+corrupt_space_page_id("$schema/t1_inc_corrupted.ibd",
+  $last_page_no, $last_page_no + 2, $last_page_no + 3);
+print $fh "$last_page_no\n";
+
+$last_page_no = extend_space("$schema/t2_inc_corrupted.ibd", 5);
+corrupt_space_page_id("$schema/t2_inc_corrupted.ibd",
+  $last_page_no + 1, $last_page_no + 2, $last_page_no + 4);
+print $fh "$last_page_no\n";
+
+$last_page_no = extend_space("$schema/t5_inc_corrupted_to_rename.ibd", 1);
+corrupt_space_page_id("$schema/t5_inc_corrupted_to_rename.ibd", $last_page_no);
+print $fh "$last_page_no\n";
+
+$last_page_no = extend_space("$schema/t6_inc_corrupted_to_drop.ibd", );
+corrupt_space_page_id("$schema/t6_inc_corrupted_to_drop.ibd", $last_page_no);
+
+close $fh;
+EOF
+--source include/start_mysqld.inc
+
+--let incdir=$MYSQLTEST_VARDIR/tmp/backup_inc
+
+--let after_load_tablespaces=CREATE TABLE test.t4_inc_corrupted_new ENGINE=INNODB SELECT UUID() from test.seq_1_to_10
+--let add_corrupted_page_for_test_t4_inc_corrupted_new=1
+--let after_copy_test_t5_inc_corrupted_to_rename=RENAME TABLE test.t5_inc_corrupted_to_rename TO test.t5_inc_corrupted_to_rename_renamed
+--let after_copy_test_t6_inc_corrupted_to_drop=DROP TABLE test.t6_inc_corrupted_to_drop
+--let after_copy_test_t7_inc_corrupted_to_alter=ALTER TABLE test.t7_inc_corrupted_to_alter ADD COLUMN (d INT)
+--let add_corrupted_page_for_test_t7_inc_corrupted_to_alter=3
+
+--echo # Backup must fail, but "innodb_corrupted_pages" file must be created due to --log-innodb-page-corruption option
+--disable_result_log
+--error 1
+--exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --log-innodb-page-corruption --target-dir=$incdir --incremental-basedir=$targetdir --dbug=+d,mariabackup_events,mariabackup_inject_code  > $backuplog
+--disable_result_log
+
+--let after_load_tablespaces=
+--let add_corrupted_page_for_test_t4_inc_corrupted_new=
+--let after_copy_test_t5_inc_corrupted_to_rename=
+--let after_copy_test_t6_inc_corrupted_to_drop=
+--let after_copy_test_t7_inc_corrupted_to_alter=
+--let add_corrupted_page_for_test_t7_inc_corrupted_to_alter=
+
+--let SEARCH_PATTERN=Database page corruption detected.*
+--let SEARCH_FILE=$backuplog
+--source include/search_pattern_in_file.inc
+--let corrupted_pages_file = $incdir/innodb_corrupted_pages
+--echo --- "innodb_corrupted_pages" file content: ---
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+print_corrupted_pages_file($ENV{corrupted_pages_file},
+  $ENV{corrupted_pages_file_filt});
+EOF
+--cat_file $corrupted_pages_file_filt
+--echo ------
+
+--echo # Check if corrupted pages were copied to delta files, and non-corrupted pages are not copied.
+perl;
+use strict;
+use warnings;
+my $schema = "$ENV{incdir}/test";
+
+open(my $fh, '<', $ENV{perl_result_file}) or die $!;
+
+my $last_page_no = <$fh>;
+die_if_no_pages("$schema/t1_corrupted.ibd.delta",
+  $last_page_no, $last_page_no + 2, $last_page_no + 3);
+
+$last_page_no = <$fh>;
+die_if_no_pages("$schema/t2_corrupted.ibd.delta",
+  $last_page_no + 1, $last_page_no + 2, $last_page_no + 4);
+
+$last_page_no = <$fh>;
+die_if_no_pages("$schema/t5_corrupted_to_rename_renamed.ibd.delta",
+  $last_page_no);
+
+close $fh;
+
+die_if_not_empty("$schema/t3.ibd.delta");
+
+sub read_first_page_from_delta {
+  my $file_name = shift;
+  my $pages_count = shift;
+
+  open my $file, '<:raw', $file_name || die "Cannot open $file_name\n";
+  read $file, my $buffer, $pages_count*4 || die "Cannot read $file_name\n";
+  close $file;
+
+  return unpack("N[$pages_count]", $buffer);
+}
+
+sub die_if_no_pages {
+  my $file_name = shift;
+  my @check_pages = @_;
+  my @read_pages =
+    read_first_page_from_delta($file_name, scalar(@check_pages) + 1);
+  for (my $i = 1; $i < @check_pages + 1; ++$i) {
+    my $check_page_no = $check_pages[$i - 1];
+    die "Corrupted page $check_page_no was not copied to $file_name."
+      if ($i >= @read_pages ||  $read_pages[$i] != $check_page_no);
+  }
+}
+
+sub die_if_not_empty {
+  my $file_name = shift;
+  my ($magic, $full) = read_first_page_from_delta($file_name, 2);
+    die "Delta $file_name must be empty."
+      if ($full != 0xFFFFFFFF);
+}
+EOF
+--rmdir $incdir
+--rmdir $targetdir
+
+DROP TABLE t1_corrupted;
+DROP TABLE t2_corrupted;
+DROP TABLE t4_corrupted_new;
+DROP TABLE t5_corrupted_to_rename_renamed;
+DROP TABLE t7_corrupted_to_alter;
+DROP TABLE t1_inc_corrupted;
+DROP TABLE t2_inc_corrupted;
+DROP TABLE t4_inc_corrupted_new;
+DROP TABLE t5_inc_corrupted_to_rename_renamed;
+DROP TABLE t7_inc_corrupted_to_alter;
+
+--echo
+--echo ########
+--echo # Test for --prepare with "innodb_corrupted_pages" file
+--echo ###
+--echo
+
+--echo # Extend some tablespace and corrupt extended pages for full backup
+--source include/shutdown_mysqld.inc
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+my $schema="$ENV{MYSQLD_DATADIR}/test";
+my $last_page_no = extend_space("$schema/t3.ibd", 3);
+corrupt_space_page_id("$schema/t3.ibd", $last_page_no, $last_page_no + 2);
+open(my $fh, '>', $ENV{perl_result_file}) or die $!;
+print $fh "$last_page_no\n";
+close $fh;
+EOF
+--source include/start_mysqld.inc
+
+--echo # Full backup with --log-innodb-page-corruption
+--disable_result_log
+--error 1
+--exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --log-innodb-page-corruption --target-dir=$targetdir
+--enable_result_log
+--let corrupted_pages_file = $targetdir/innodb_corrupted_pages
+--echo --- "innodb_corrupted_pages" file content: ---
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+print_corrupted_pages_file($ENV{corrupted_pages_file},
+  $ENV{corrupted_pages_file_filt});
+EOF
+--cat_file $corrupted_pages_file_filt
+--echo ------
+
+--echo # Extend some tablespace and corrupt extended pages for incremental backup
+--source include/shutdown_mysqld.inc
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+my $schema="$ENV{MYSQLD_DATADIR}/test";
+my $last_page_no = extend_space("$schema/t3_inc.ibd", 3);
+corrupt_space_page_id("$schema/t3_inc.ibd", $last_page_no, $last_page_no + 2);
+open(my $fh, '>>', $ENV{perl_result_file}) or die $!;
+print $fh "$last_page_no";
+close $fh;
+EOF
+--source include/start_mysqld.inc
+
+--echo # Incremental backup --log-innodb-page-corruption
+--disable_result_log
+--error 1
+--exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --log-innodb-page-corruption --target-dir=$incdir --incremental-basedir=$targetdir --dbug=+d,mariabackup_events,mariabackup_inject_code  > $backuplog
+--disable_result_log
+--let corrupted_pages_file = $incdir/innodb_corrupted_pages
+--echo --- "innodb_corrupted_pages" file content: ---
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+print_corrupted_pages_file($ENV{corrupted_pages_file},
+  $ENV{corrupted_pages_file_filt});
+EOF
+--cat_file $corrupted_pages_file_filt
+--echo ------
+
+--let targetdir2=$targetdir-2
+--let incdir2=$incdir-2
+perl;
+use lib "lib";
+use My::Handles { suppress_init_messages => 1 };
+use My::File::Path;
+copytree($ENV{'targetdir'}, $ENV{'targetdir2'});
+copytree($ENV{'incdir'}, $ENV{'incdir2'});
+EOF
+
+--echo # Full backup prepare
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir > $backuplog;
+--enable_result_log
+
+--echo # "innodb_corrupted_pages" file must not exist after successful prepare
+--error 1
+--file_exists $targetdir/innodb_corrupted_pages
+--let SEARCH_PATTERN=was successfuly fixed.*
+--let SEARCH_FILE=$backuplog
+--source include/search_pattern_in_file.inc
+
+--echo # Check that fixed pages are zero-filled
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+open(my $fh, '<', $ENV{perl_result_file}) or die $!;
+my $last_page_no = <$fh>;
+close $fh;
+my $schema = "$ENV{targetdir}/test";
+die_if_page_is_not_zero("$schema/t3.ibd", $last_page_no, $last_page_no + 2);
+EOF
+
+--echo # Incremental backup prepare
+--disable_result_log
+exec $XTRABACKUP --prepare --target-dir=$targetdir --incremental-dir=$incdir > $backuplog;
+--enable_result_log
+
+--echo # "innodb_corrupted_pages" file must not exist after successful prepare
+--error 1
+--file_exists $targetdir/innodb_corrupted_pages
+--echo # do not remove "innodb_corrupted_pages" in incremental dir
+--file_exists $incdir/innodb_corrupted_pages
+--let SEARCH_PATTERN=was successfuly fixed.*
+--let SEARCH_FILE=$backuplog
+--source include/search_pattern_in_file.inc
+
+--echo # Check that fixed pages are zero-filled
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+open(my $fh, '<', $ENV{perl_result_file}) or die $!;
+my $last_page_no_full = <$fh>;
+my $last_page_no_inc = <$fh>;
+close $fh;
+my $schema = "$ENV{targetdir}/test";
+die_if_page_is_not_zero("$schema/t3.ibd",
+  $last_page_no_full, $last_page_no_full + 2);
+die_if_page_is_not_zero("$schema/t3_inc.ibd",
+  $last_page_no_inc, $last_page_no_inc + 2);
+EOF
+
+--source include/restart_and_restore.inc
+
+SELECT * FROM t3;
+SELECT * FROM t3_inc;
+
+--echo # Test the case when not all corrupted pages are fixed
+--echo
+--echo # Add some fake corrupted pages
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+append_corrupted_pages(
+  "$ENV{targetdir2}/innodb_corrupted_pages", 'test/t3', '3 4');
+append_corrupted_pages(
+  "$ENV{incdir2}/innodb_corrupted_pages", 'test/t3_inc', '4 5');
+EOF
+
+--echo # Full backup prepare
+--disable_result_log
+--error 1
+exec $XTRABACKUP  --prepare --target-dir=$targetdir2 > $backuplog;
+--enable_result_log
+
+--let SEARCH_PATTERN=Error: corrupted page.*
+--let SEARCH_FILE=$backuplog
+--source include/search_pattern_in_file.inc
+--let corrupted_pages_file = $targetdir2/innodb_corrupted_pages
+--echo --- "innodb_corrupted_pages" file content: ---
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+print_corrupted_pages_file($ENV{corrupted_pages_file},
+  $ENV{corrupted_pages_file_filt});
+EOF
+--cat_file $corrupted_pages_file_filt
+--echo ------
+
+--echo # Incremental backup prepare
+--disable_result_log
+--error 1
+exec $XTRABACKUP --prepare --target-dir=$targetdir2 --incremental-dir=$incdir2 > $backuplog;
+--enable_result_log
+
+--let SEARCH_PATTERN=Error: corrupted page.*
+--let SEARCH_FILE=$backuplog
+--source include/search_pattern_in_file.inc
+--let corrupted_pages_file = $targetdir2/innodb_corrupted_pages
+--echo --- "innodb_corrupted_pages" file content: ---
+perl;
+do "$ENV{MTR_SUITE_DIR}/include/corrupt-page.pl";
+print_corrupted_pages_file($ENV{corrupted_pages_file},
+  $ENV{corrupted_pages_file_filt});
+EOF
+--cat_file $corrupted_pages_file_filt
+--echo ------
+
+DROP TABLE t3;
+DROP TABLE t3_inc;
+--remove_file $backuplog
+--remove_file $perl_result_file
+--remove_file $corrupted_pages_file_filt
+--rmdir $targetdir
+--rmdir $targetdir2
+--rmdir $incdir
+--rmdir $incdir2
diff --git a/mysql-test/suite/roles/show_grants.result b/mysql-test/suite/roles/show_grants.result
index 1867133525d..7ae499a9cfc 100644
--- a/mysql-test/suite/roles/show_grants.result
+++ b/mysql-test/suite/roles/show_grants.result
@@ -147,3 +147,18 @@ drop role test_role2;
 delete from mysql.roles_mapping where Role='test_role1';
 delete from mysql.roles_mapping where Role='test_role2';
 flush privileges;
+#
+# MDEV-24289: show grants missing with grant option
+#
+create role anel;
+GRANT SELECT, UPDATE, DELETE, ALTER ON *.* TO 'anel';
+SHOW GRANTS for 'anel';
+Grants for anel
+GRANT SELECT, UPDATE, DELETE, ALTER ON *.* TO `anel`
+create role MariaDB_admin;
+GRANT SELECT, UPDATE, DELETE, ALTER ON *.* TO 'MariaDB_admin' WITH GRANT OPTION;
+SHOW GRANTS for 'MariaDB_admin';
+Grants for MariaDB_admin
+GRANT SELECT, UPDATE, DELETE, ALTER ON *.* TO `MariaDB_admin` WITH GRANT OPTION
+drop role MariaDB_admin;
+drop role anel;
diff --git a/mysql-test/suite/roles/show_grants.test b/mysql-test/suite/roles/show_grants.test
index 9c15d8b8b2b..fc2165ac53b 100644
--- a/mysql-test/suite/roles/show_grants.test
+++ b/mysql-test/suite/roles/show_grants.test
@@ -88,3 +88,16 @@ drop role test_role2;
 delete from mysql.roles_mapping where Role='test_role1';
 delete from mysql.roles_mapping where Role='test_role2';
 flush privileges;
+
+--echo #
+--echo # MDEV-24289: show grants missing with grant option
+--echo #
+create role anel;
+GRANT SELECT, UPDATE, DELETE, ALTER ON *.* TO 'anel';
+SHOW GRANTS for 'anel';
+
+create role MariaDB_admin;
+GRANT SELECT, UPDATE, DELETE, ALTER ON *.* TO 'MariaDB_admin' WITH GRANT OPTION;
+SHOW GRANTS for 'MariaDB_admin';
+drop role MariaDB_admin;
+drop role anel;
diff --git a/mysql-test/suite/wsrep/include/check_galera_version.inc b/mysql-test/suite/wsrep/include/check_galera_version.inc
index 32d01197f94..7a58e657f40 100644
--- a/mysql-test/suite/wsrep/include/check_galera_version.inc
+++ b/mysql-test/suite/wsrep/include/check_galera_version.inc
@@ -21,23 +21,27 @@ SELECT CAST(REGEXP_REPLACE(@GALERA_VERSION,'^(\\d+)\\.(\\d+)\\.(\\d+).*','\\3')
 
 # Actual
 SELECT VARIABLE_VALUE INTO @ACTUAL_GALERA_VERSION FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME LIKE 'wsrep_provider_version';
+SELECT CAST(REGEXP_REPLACE(@ACTUAL_GALERA_VERSION,'^(\\d+)\\.(\\d+).*','\\1') AS UNSIGNED) INTO @ACTUAL_GALERA_MAJOR_VERSION;
 SELECT CAST(REGEXP_REPLACE(@ACTUAL_GALERA_VERSION,'^[\\d\\.]*(\\d+)\\.\\d+.*','\\1') AS UNSIGNED) INTO @ACTUAL_GALERA_MINOR_VERSION;
 SELECT CAST(REGEXP_REPLACE(@ACTUAL_GALERA_VERSION,'^[\\d\\.]*\\.(\\d+).*','\\1') AS UNSIGNED) INTO @ACTUAL_GALERA_RELEASE_VERSION;
+--enable_query_log
 
 # For testing
 #SELECT @GALERA_MAJOR_VERSION;
 #SELECT @GALERA_MINOR_VERSION;
 #SELECT @GALERA_RELEASE_VERSION;
-#SELECT @ACTUAL_GALERA_VERSION;
+#SELECT @ACTUAL_GALERA_MAJOR_VERSION;
 #SELECT @ACTUAL_GALERA_MINOR_VERSION;
 #SELECT @ACTUAL_GALERA_RELEASE_VERSION;
 
-if (!`SELECT (@ACTUAL_GALERA_MINOR_VERSION > @GALERA_MINOR_VERSION) OR
-              (@ACTUAL_GALERA_MINOR_VERSION = @GALERA_MINOR_VERSION AND
+if (!`SELECT (@ACTUAL_GALERA_MAJOR_VERSION >= @GALERA_MAJOR_VERSION AND @ACTUAL_GALERA_MINOR_VERSION > @GALERA_MINOR_VERSION) OR
+              (@ACTUAL_GALERA_MAJOR_VERSION = @GALERA_MAJOR_VERSION AND
+	       @ACTUAL_GALERA_MINOR_VERSION = @GALERA_MINOR_VERSION AND
 	       @ACTUAL_GALERA_RELEASE_VERSION >= @GALERA_RELEASE_VERSION)
    `)
 {
     skip Test requires Galera library version >= $galera_version;
 }
 
---enable_query_log
+--echo # Correct Galera library found
+
diff --git a/mysys/my_mess.c b/mysys/my_mess.c
index 7bc4c038cb6..c9a1aee64b6 100644
--- a/mysys/my_mess.c
+++ b/mysys/my_mess.c
@@ -21,6 +21,8 @@ void my_message_stderr(uint error __attribute__((unused)),
   DBUG_ENTER("my_message_stderr");
   DBUG_PRINT("enter",("message: %s",str));
   (void) fflush(stdout);
+  if (MyFlags & (ME_NOTE | ME_ERROR_LOG_ONLY))
+    DBUG_VOID_RETURN;
   if (MyFlags & ME_BELL)
     (void) fputc('\007', stderr);
   if (my_progname)
diff --git a/plugin/handler_socket/handlersocket/database.cpp b/plugin/handler_socket/handlersocket/database.cpp
index 91b095cb655..937b1177ae4 100644
--- a/plugin/handler_socket/handlersocket/database.cpp
+++ b/plugin/handler_socket/handlersocket/database.cpp
@@ -471,9 +471,7 @@ dbcontext::close_tables_if()
   unlock_tables_if();
   DENA_VERBOSE(100, fprintf(stderr, "HNDSOCK close tables\n"));
   close_thread_tables(thd);
-  #if MYSQL_VERSION_ID >= 50505
-  thd->mdl_context.release_transactional_locks();
-  #endif
+  thd->mdl_context.release_transactional_locks(thd);
   if (!table_vec.empty()) {
     statistic_increment(close_tables_count, &LOCK_status);
     table_vec.clear();
diff --git a/plugin/type_inet/sql_type_inet.h b/plugin/type_inet/sql_type_inet.h
index 8c42431ccaa..03153af6ddc 100644
--- a/plugin/type_inet/sql_type_inet.h
+++ b/plugin/type_inet/sql_type_inet.h
@@ -653,7 +653,7 @@ public:
   }
   bool
   subquery_type_allows_materialization(const Item *inner,
-                                       const Item *outer) const override
+                                       const Item *outer, bool) const override
   {
     /*
       Example:
diff --git a/scripts/wsrep_sst_rsync.sh b/scripts/wsrep_sst_rsync.sh
index d42cf014009..809d3e81e16 100644
--- a/scripts/wsrep_sst_rsync.sh
+++ b/scripts/wsrep_sst_rsync.sh
@@ -401,6 +401,14 @@ then
     MODULE="rsync_sst"
 
     RSYNC_PID="$WSREP_SST_OPT_DATA/$MODULE.pid"
+    # give some time for lingering rsync from previous SST to complete
+    check_round=0
+    while check_pid $RSYNC_PID && [ $check_round -lt 10 ]
+    do
+        wsrep_log_info "lingering rsync daemon found at startup, waiting for it to exit"
+        check_round=$(( check_round + 1 ))
+        sleep 1
+    done
 
     if check_pid $RSYNC_PID
     then
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 7f00a194698..470f59fe15f 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4443,7 +4443,7 @@ int ha_partition::write_row(const uchar * buf)
 
   DBUG_ASSERT(!m_file[part_id]->row_logging);
   error= m_file[part_id]->ha_write_row(buf);
-  if (have_auto_increment && !table->s->next_number_keypart)
+  if (!error && have_auto_increment && !table->s->next_number_keypart)
     set_auto_increment_if_higher(table->next_number_field);
 
 exit:
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 8d0557f4ae4..3c3e394d59f 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -92,7 +92,6 @@ public:
   bool auto_inc_initialized;
   mysql_mutex_t auto_inc_mutex;                /**< protecting auto_inc val */
   ulonglong next_auto_inc_val;                 /**< first non reserved value */
-  ulonglong prev_auto_inc_val;                 /**< stored next_auto_inc_val */
   /**
     Hash of partition names. Initialized in the first ha_partition::open()
     for the table_share. After that it is read-only, i.e. no locking required.
@@ -104,7 +103,6 @@ public:
   Partition_share()
     : auto_inc_initialized(false),
     next_auto_inc_val(0),
-    prev_auto_inc_val(0),
     partition_name_hash_initialized(false),
     partition_names(NULL)
   {
@@ -429,24 +427,6 @@ private:
   MY_BITMAP m_locked_partitions;
   /** Stores shared auto_increment etc. */
   Partition_share *part_share;
-  /** Fix spurious -Werror=overloaded-virtual in GCC 9 */
-  virtual void restore_auto_increment(ulonglong prev_insert_id) override
-  {
-    handler::restore_auto_increment(prev_insert_id);
-  }
-  /** Store and restore next_auto_inc_val over duplicate key errors. */
-  void store_auto_increment() override
-  {
-    DBUG_ASSERT(part_share);
-    part_share->prev_auto_inc_val= part_share->next_auto_inc_val;
-    handler::store_auto_increment();
-  }
-  void restore_auto_increment() override
-  {
-    DBUG_ASSERT(part_share);
-    part_share->next_auto_inc_val= part_share->prev_auto_inc_val;
-    handler::restore_auto_increment();
-  }
   void sum_copy_info(handler *file);
   void sum_copy_infos();
   void reset_copy_info() override;
diff --git a/sql/handler.cc b/sql/handler.cc
index 01825c13da6..314fe7acb8c 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -3514,7 +3514,6 @@ int handler::update_auto_increment()
   THD *thd= table->in_use;
   struct system_variables *variables= &thd->variables;
   int result=0, tmp;
-  enum enum_check_fields save_count_cuted_fields;
   DBUG_ENTER("handler::update_auto_increment");
 
   /*
@@ -3656,10 +3655,10 @@ int handler::update_auto_increment()
                      nr, append ? nb_reserved_values : 0));
 
   /* Store field without warning (Warning will be printed by insert) */
-  save_count_cuted_fields= thd->count_cuted_fields;
-  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
-  tmp= table->next_number_field->store((longlong)nr, TRUE);
-  thd->count_cuted_fields= save_count_cuted_fields;
+  {
+    Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
+    tmp= table->next_number_field->store((longlong)nr, TRUE);
+  }
 
   if (unlikely(tmp))                            // Out of range value in store
   {
diff --git a/sql/handler.h b/sql/handler.h
index 41da458aa2a..4e1e3f0413f 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -3278,9 +3278,6 @@ private:
   */
   Handler_share **ha_share;
 
-  /** Stores next_insert_id for handling duplicate key errors. */
-  ulonglong m_prev_insert_id;
-
 public:
   handler(handlerton *ht_arg, TABLE_SHARE *share_arg)
     :table_share(share_arg), table(0),
@@ -3308,7 +3305,7 @@ public:
     m_psi_numrows(0),
     m_psi_locker(NULL),
     row_logging(0), row_logging_init(0),
-    m_lock_type(F_UNLCK), ha_share(NULL), m_prev_insert_id(0)
+    m_lock_type(F_UNLCK), ha_share(NULL)
   {
     DBUG_PRINT("info",
                ("handler created F_UNLCK %d F_RDLCK %d F_WRLCK %d",
@@ -4028,16 +4025,6 @@ public:
       insert_id_for_cur_row;
   }
 
-  /** Store and restore next_insert_id over duplicate key errors. */
-  virtual void store_auto_increment()
-  {
-    m_prev_insert_id= next_insert_id;
-  }
-  virtual void restore_auto_increment()
-  {
-    restore_auto_increment(m_prev_insert_id);
-  }
-
   virtual void update_create_info(HA_CREATE_INFO *create_info) {}
   int check_old_types();
   virtual int assign_to_keycache(THD* thd, HA_CHECK_OPT* check_opt)
diff --git a/sql/item.cc b/sql/item.cc
index dc90b9fe1b3..2bf53dc885b 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1442,16 +1442,12 @@ int Item::save_in_field_no_warnings(Field *field, bool no_conversions)
   int res;
   TABLE *table= field->table;
   THD *thd= table->in_use;
-  enum_check_fields tmp= thd->count_cuted_fields;
-  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set);
+  Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
   Sql_mode_save sms(thd);
   thd->variables.sql_mode&= ~(MODE_NO_ZERO_IN_DATE | MODE_NO_ZERO_DATE);
   thd->variables.sql_mode|= MODE_INVALID_DATES;
-  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
-
+  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set);
   res= save_in_field(field, no_conversions);
-
-  thd->count_cuted_fields= tmp;
   dbug_tmp_restore_column_map(table->write_set, old_map);
   return res;
 }
diff --git a/sql/item.h b/sql/item.h
index fb480b4c578..677112e448c 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -973,6 +973,13 @@ public:
   void set_name_no_truncate(THD *thd, const char *str, uint length,
                             CHARSET_INFO *cs);
   void init_make_send_field(Send_field *tmp_field, const Type_handler *h);
+  void share_name_with(const Item *item)
+  {
+    name= item->name;
+    common_flags= static_cast<uint8>
+     ((common_flags & ~IS_AUTO_GENERATED_NAME) |
+      (item->common_flags & IS_AUTO_GENERATED_NAME));
+  }
   virtual void cleanup();
   virtual void make_send_field(THD *thd, Send_field *field);
 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 83eb605f463..bfd415344ef 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -321,19 +321,18 @@ static bool convert_const_to_int(THD *thd, Item_field *field_item,
   if ((*item)->const_item() && !(*item)->is_expensive())
   {
     TABLE *table= field->table;
-    sql_mode_t orig_sql_mode= thd->variables.sql_mode;
-    enum_check_fields orig_count_cuted_fields= thd->count_cuted_fields;
+    Sql_mode_save sql_mode(thd);
+    Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
     my_bitmap_map *old_maps[2] = { NULL, NULL };
     ulonglong UNINIT_VAR(orig_field_val); /* original field value if valid */
 
     /* table->read_set may not be set if we come here from a CREATE TABLE */
     if (table && table->read_set)
-      dbug_tmp_use_all_columns(table, old_maps, 
+      dbug_tmp_use_all_columns(table, old_maps,
                                table->read_set, table->write_set);
     /* For comparison purposes allow invalid dates like 2000-01-32 */
-    thd->variables.sql_mode= (orig_sql_mode & ~MODE_NO_ZERO_DATE) | 
+    thd->variables.sql_mode= (thd->variables.sql_mode & ~MODE_NO_ZERO_DATE) |
                              MODE_INVALID_DATES;
-    thd->count_cuted_fields= CHECK_FIELD_IGNORE;
 
     /*
       Store the value of the field/constant because the call to save_in_field
@@ -370,8 +369,6 @@ static bool convert_const_to_int(THD *thd, Item_field *field_item,
       /* orig_field_val must be a valid value that can be restored back. */
       DBUG_ASSERT(!result);
     }
-    thd->variables.sql_mode= orig_sql_mode;
-    thd->count_cuted_fields= orig_count_cuted_fields;
     if (table && table->read_set)
       dbug_tmp_restore_column_maps(table->read_set, table->write_set, old_maps);
   }
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index cbde9599073..7916bcb45a9 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -1520,7 +1520,8 @@ Item_in_subselect::Item_in_subselect(THD *thd, Item * left_exp,
   pushed_cond_guards(NULL), do_not_convert_to_sj(FALSE), is_jtbm_merged(FALSE),
   is_jtbm_const_tab(FALSE), is_flattenable_semijoin(FALSE),
   is_registered_semijoin(FALSE),
-  upper_item(0)
+  upper_item(0),
+  converted_from_in_predicate(FALSE)
 {
   DBUG_ENTER("Item_in_subselect::Item_in_subselect");
   DBUG_PRINT("info", ("in_strategy: %u", (uint)in_strategy));
diff --git a/sql/item_subselect.h b/sql/item_subselect.h
index ec4398b9a76..a346def72f1 100644
--- a/sql/item_subselect.h
+++ b/sql/item_subselect.h
@@ -613,12 +613,18 @@ public:
 
   Item_func_not_all *upper_item; // point on NOT/NOP before ALL/SOME subquery
 
+  /*
+    SET to TRUE if IN subquery is converted from an IN predicate
+  */
+  bool converted_from_in_predicate;
+
   Item_in_subselect(THD *thd_arg, Item * left_expr, st_select_lex *select_lex);
   Item_in_subselect(THD *thd_arg):
     Item_exists_subselect(thd_arg), left_expr_cache(0), first_execution(TRUE),
     in_strategy(SUBS_NOT_TRANSFORMED),
     pushed_cond_guards(NULL), func(NULL), do_not_convert_to_sj(FALSE),
-    is_jtbm_merged(FALSE), is_jtbm_const_tab(FALSE), upper_item(0) {}
+    is_jtbm_merged(FALSE), is_jtbm_const_tab(FALSE), upper_item(0),
+    converted_from_in_predicate(FALSE) {}
   void cleanup() override;
   subs_type substype() override { return IN_SUBS; }
   void reset() override
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index dd7e166059a..aa3e48ee873 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -2988,10 +2988,10 @@ error:
   if (thd->transaction_rollback_request)
   {
     trans_rollback_implicit(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
   else if (! thd->in_multi_stmt_transaction_mode())
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   else
     thd->mdl_context.release_statement_locks();
 
@@ -4039,7 +4039,7 @@ int Xid_log_event::do_commit()
 {
   bool res;
   res= trans_commit(thd); /* Automatically rolls back on error. */
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
   return res;
 }
 #endif
@@ -4088,10 +4088,7 @@ int XA_prepare_log_event::do_commit()
     res= trans_xa_prepare(thd);
   }
   else
-  {
     res= trans_xa_commit(thd);
-    thd->mdl_context.release_transactional_locks();
-  }
 
   return res;
 }
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 4772dc017f9..5f5e5c3ce9a 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -3039,15 +3039,17 @@ void MDL_context::rollback_to_savepoint(const MDL_savepoint &mdl_savepoint)
   implementation of COMMIT (implicit or explicit) and ROLLBACK.
 */
 
-void MDL_context::release_transactional_locks()
+void MDL_context::release_transactional_locks(THD *thd)
 {
   DBUG_ENTER("MDL_context::release_transactional_locks");
+  /* Fail if there are active transactions */
+  DBUG_ASSERT(!(thd->server_status &
+                (SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY)));
   release_locks_stored_before(MDL_STATEMENT, NULL);
   release_locks_stored_before(MDL_TRANSACTION, NULL);
   DBUG_VOID_RETURN;
 }
 
-
 void MDL_context::release_statement_locks()
 {
   DBUG_ENTER("MDL_context::release_transactional_locks");
diff --git a/sql/mdl.h b/sql/mdl.h
index f6b7154fba0..9dbf9aa7f4f 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -921,7 +921,7 @@ public:
   void set_lock_duration(MDL_ticket *mdl_ticket, enum_mdl_duration duration);
 
   void release_statement_locks();
-  void release_transactional_locks();
+  void release_transactional_locks(THD *thd);
   void release_explicit_locks();
   void rollback_to_savepoint(const MDL_savepoint &mdl_savepoint);
 
diff --git a/sql/opt_subselect.cc b/sql/opt_subselect.cc
index 41dcf6442a1..0a87d9ccd2f 100644
--- a/sql/opt_subselect.cc
+++ b/sql/opt_subselect.cc
@@ -872,6 +872,7 @@ bool subquery_types_allow_materialization(THD* thd, Item_in_subselect *in_subs)
 
   bool all_are_fields= TRUE;
   uint32 total_key_length = 0;
+  bool converted_from_in_predicate= in_subs->converted_from_in_predicate;
   for (uint i= 0; i < elements; i++)
   {
     Item *outer= left_exp->element_index(i);
@@ -879,8 +880,11 @@ bool subquery_types_allow_materialization(THD* thd, Item_in_subselect *in_subs)
     all_are_fields &= (outer->real_item()->type() == Item::FIELD_ITEM && 
                        inner->real_item()->type() == Item::FIELD_ITEM);
     total_key_length += inner->max_length;
-    if (!inner->type_handler()->subquery_type_allows_materialization(inner,
-                                                                     outer))
+    if (!inner->
+         type_handler()->
+         subquery_type_allows_materialization(inner,
+                                              outer,
+                                              converted_from_in_predicate))
     {
       trace_transform.add("possible", false);
       trace_transform.add("cause", "types mismatch");
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index af0c4747d2b..bc06188af34 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -445,7 +445,7 @@ rpl_slave_state::truncate_state_table(THD *thd)
       close_thread_tables(thd);
       ha_commit_trans(thd, TRUE);
     }
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
   return err;
 }
@@ -736,7 +736,7 @@ end:
     if (in_transaction)
       thd->mdl_context.release_statement_locks();
     else
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
   }
   thd->lex->restore_backup_query_tables_list(&lex_backup);
   thd->variables.option_bits= thd_saved_option;
@@ -991,7 +991,7 @@ end:
         ha_rollback_trans(thd, FALSE);
     }
     close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     thd->lex->restore_backup_query_tables_list(&lex_backup);
 
     if (err)
diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
index 0726697211b..2319f69d67c 100644
--- a/sql/rpl_injector.cc
+++ b/sql/rpl_injector.cc
@@ -69,34 +69,34 @@ injector::transaction::~transaction()
  */
 int injector::transaction::commit()
 {
-   DBUG_ENTER("injector::transaction::commit()");
-   int error= m_thd->binlog_flush_pending_rows_event(true);
-   /*
-     Cluster replication does not preserve statement or
-     transaction boundaries of the master.  Instead, a new
-     transaction on replication slave is started when a new GCI
-     (global checkpoint identifier) is issued, and is committed
-     when the last event of the check point has been received and
-     processed. This ensures consistency of each cluster in
-     cluster replication, and there is no requirement for stronger
-     consistency: MySQL replication is asynchronous with other
-     engines as well.
-
-     A practical consequence of that is that row level replication
-     stream passed through the injector thread never contains
-     COMMIT events.
-     Here we should preserve the server invariant that there is no
-     outstanding statement transaction when the normal transaction
-     is committed by committing the statement transaction
-     explicitly.
-   */
-   trans_commit_stmt(m_thd);
-   if (!trans_commit(m_thd))
-   {
-     close_thread_tables(m_thd);
-     m_thd->mdl_context.release_transactional_locks();
-   }
-   DBUG_RETURN(error);
+  DBUG_ENTER("injector::transaction::commit()");
+  int error= m_thd->binlog_flush_pending_rows_event(true);
+  /*
+    Cluster replication does not preserve statement or
+    transaction boundaries of the master.  Instead, a new
+    transaction on replication slave is started when a new GCI
+    (global checkpoint identifier) is issued, and is committed
+    when the last event of the check point has been received and
+    processed. This ensures consistency of each cluster in
+    cluster replication, and there is no requirement for stronger
+    consistency: MySQL replication is asynchronous with other
+    engines as well.
+
+    A practical consequence of that is that row level replication
+    stream passed through the injector thread never contains
+    COMMIT events.
+    Here we should preserve the server invariant that there is no
+    outstanding statement transaction when the normal transaction
+    is committed by committing the statement transaction
+    explicitly.
+  */
+  trans_commit_stmt(m_thd);
+  if (!trans_commit(m_thd))
+  {
+    close_thread_tables(m_thd);
+    m_thd->release_transactional_locks();
+  }
+  DBUG_RETURN(error);
 }
 
 
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index b7127d86fcd..8b61a3a708b 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1677,7 +1677,7 @@ end:
   {
     *out_hton= table->s->db_type();
     close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
   return err;
 }
@@ -1704,7 +1704,7 @@ scan_all_gtid_slave_pos_table(THD *thd, int (*cb)(THD *, LEX_CSTRING *, void *),
   {
     my_error(ER_FILE_NOT_FOUND, MYF(0), path, my_errno);
     close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     return 1;
   }
   else
@@ -1717,7 +1717,7 @@ scan_all_gtid_slave_pos_table(THD *thd, int (*cb)(THD *, LEX_CSTRING *, void *),
     err= ha_discover_table_names(thd, &MYSQL_SCHEMA_NAME, dirp, &tl, false);
     my_dirend(dirp);
     close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     if (err)
       return err;
 
@@ -2003,7 +2003,7 @@ end:
     ha_commit_trans(thd, FALSE);
     ha_commit_trans(thd, TRUE);
     close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
 
   return err;
@@ -2292,7 +2292,7 @@ void rpl_group_info::cleanup_context(THD *thd, bool error)
     if (thd->transaction->xid_state.is_explicit_XA())
       xa_trans_force_rollback(thd);
 
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
 
     if (thd == rli->sql_driver_thd)
     {
@@ -2406,10 +2406,10 @@ void rpl_group_info::slave_close_thread_tables(THD *thd)
   if (thd->transaction_rollback_request)
   {
     trans_rollback_implicit(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
   else if (! thd->in_multi_stmt_transaction_mode())
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   else
     thd->mdl_context.release_statement_locks();
 
diff --git a/sql/sp.cc b/sql/sp.cc
index 3737bd11740..abd89dde499 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -1205,8 +1205,6 @@ Sp_handler::sp_create_routine(THD *thd, const sp_head *sp) const
 
   CHARSET_INFO *db_cs= get_default_db_collation(thd, sp->m_db.str);
 
-  enum_check_fields saved_count_cuted_fields;
-
   bool store_failed= FALSE;
   DBUG_ENTER("sp_create_routine");
   DBUG_PRINT("enter", ("type: %s  name: %.*s",
@@ -1240,8 +1238,7 @@ Sp_handler::sp_create_routine(THD *thd, const sp_head *sp) const
   /* Reset sql_mode during data dictionary operations. */
   thd->variables.sql_mode= 0;
 
-  saved_count_cuted_fields= thd->count_cuted_fields;
-  thd->count_cuted_fields= CHECK_FIELD_WARN;
+  Check_level_instant_set check_level_save(thd, CHECK_FIELD_WARN);
 
   if (!(table= open_proc_table_for_update(thd)))
   {
@@ -1500,7 +1497,6 @@ log:
   ret= FALSE;
 
 done:
-  thd->count_cuted_fields= saved_count_cuted_fields;
   thd->variables.sql_mode= saved_mode;
   DBUG_ASSERT(!thd->is_current_stmt_binlog_format_row());
   DBUG_RETURN(ret);
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 9e5b5bee0f2..513e7207b7e 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2390,10 +2390,10 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
       if (thd->transaction_rollback_request)
       {
         trans_rollback_implicit(thd);
-        thd->mdl_context.release_transactional_locks();
+        thd->release_transactional_locks();
       }
       else if (! thd->in_multi_stmt_transaction_mode())
-        thd->mdl_context.release_transactional_locks();
+        thd->release_transactional_locks();
       else
         thd->mdl_context.release_statement_locks();
     }
@@ -3518,10 +3518,10 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
       if (thd->transaction_rollback_request)
       {
         trans_rollback_implicit(thd);
-        thd->mdl_context.release_transactional_locks();
+        thd->release_transactional_locks();
       }
       else if (! thd->in_multi_stmt_transaction_mode())
-        thd->mdl_context.release_transactional_locks();
+        thd->release_transactional_locks();
       else
         thd->mdl_context.release_statement_locks();
     }
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index bad39cfc1d1..25e8aa42c7c 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -9463,6 +9463,8 @@ static bool show_global_privileges(THD *thd, ACL_USER_BASE *acl_entry,
     add_user_parameters(thd, &global, (ACL_USER *)acl_entry,
                         (want_access & GRANT_ACL));
 
+  else if (want_access & GRANT_ACL)
+    global.append(STRING_WITH_LEN(" WITH GRANT OPTION"));
   protocol->prepare_for_resend();
   protocol->store(global.ptr(),global.length(),global.charset());
   if (protocol->write())
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index 2d6f891f56f..7c1f8596ece 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -42,7 +42,7 @@ static bool admin_recreate_table(THD *thd, TABLE_LIST *table_list)
   trans_rollback_stmt(thd);
   trans_rollback(thd);
   close_thread_tables(thd);
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
 
   /*
     table_list->table has been closed and freed. Do not reference
@@ -115,7 +115,7 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
       acquire the exclusive lock to satisfy MDL asserts and avoid
       deadlocks.
     */
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     /*
       Attempt to do full-blown table open in mysql_admin_table() has failed.
       Let us try to open at least a .FRM for this table.
@@ -278,7 +278,7 @@ end:
   }
   /* In case of a temporary table there will be no metadata lock. */
   if (unlikely(error) && has_mdl_lock)
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
 
   DBUG_RETURN(error);
 }
@@ -607,7 +607,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       trans_rollback(thd);
       close_thread_tables(thd);
       table->table= NULL;
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       MDL_REQUEST_INIT(&table->mdl_request, MDL_key::TABLE, table->db.str,
                        table->table_name.str, MDL_SHARED_NO_READ_WRITE,
                        MDL_TRANSACTION);
@@ -668,7 +668,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
         trans_rollback_stmt(thd);
         trans_rollback(thd);
         close_thread_tables(thd);
-        thd->mdl_context.release_transactional_locks();
+        thd->release_transactional_locks();
         DBUG_PRINT("admin", ("simple error, admin next table"));
         continue;
       case -1:           // error, message could be written to net
@@ -735,7 +735,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       trans_commit_stmt(thd);
       trans_commit(thd);
       close_thread_tables(thd);
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       lex->reset_query_tables_list(FALSE);
       /*
         Restore Query_tables_list::sql_command value to make statement
@@ -868,7 +868,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       thd->open_options|= extra_open_options;
       close_thread_tables(thd);
       table->table= NULL;
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       MDL_REQUEST_INIT(&table->mdl_request, MDL_key::TABLE, table->db.str,
                        table->table_name.str, MDL_SHARED_NO_READ_WRITE,
                        MDL_TRANSACTION);
@@ -1099,7 +1099,7 @@ send_result_message:
       trans_commit_stmt(thd);
       trans_commit(thd);
       close_thread_tables(thd);
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       /* Clear references to TABLE and MDL_ticket after releasing them. */
       table->mdl_request.ticket= NULL;
 
@@ -1258,7 +1258,7 @@ send_result_message:
         goto err;
     }
     close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
 
     /*
       If it is CHECK TABLE v1, v2, v3, and v1, v2, v3 are views, we will run
@@ -1296,7 +1296,7 @@ err:
     table->table= 0;
   }
   close_thread_tables(thd);			// Shouldn't be needed
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
   thd->resume_subsequent_commits(suspended_wfc);
   DBUG_RETURN(TRUE);
 }
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index dec5bd91f36..d02d130c084 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -9040,9 +9040,12 @@ void
 close_mysql_tables(THD *thd)
 {
   if (! thd->in_sub_stmt)
+  {
     trans_commit_stmt(thd);
+    trans_commit(thd);
+  }
   close_thread_tables(thd);
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
 }
 
 /*
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index a4fd26cf7d8..a0c23a08b12 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1565,7 +1565,7 @@ void THD::cleanup(void)
     and left the mode a few lines above), there will be outstanding
     metadata locks. Release them.
   */
-  mdl_context.release_transactional_locks();
+  mdl_context.release_transactional_locks(this);
 
   backup_end(this);
   backup_unlock(this);
@@ -4919,7 +4919,7 @@ void destroy_background_thd(MYSQL_THD thd)
 void reset_thd(MYSQL_THD thd)
 {
   close_thread_tables(thd);
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
   thd->free_items();
   free_root(thd->mem_root, MYF(MY_KEEP_PREALLOC));
 }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index e3e3b18472b..1da82edc061 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -4730,6 +4730,13 @@ public:
     locked_tables_mode= mode_arg;
   }
   void leave_locked_tables_mode();
+  /* Relesae transactional locks if there are no active transactions */
+  void release_transactional_locks()
+  {
+    if (!(server_status &
+          (SERVER_STATUS_IN_TRANS | SERVER_STATUS_IN_TRANS_READONLY)))
+      mdl_context.release_transactional_locks(this);
+  }
   int decide_logging_format(TABLE_LIST *tables);
   /*
    In Some cases when decide_logging_format is called it does not have all
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 7586b2831ae..01f67476688 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -248,7 +248,7 @@ int update_portion_of_time(THD *thd, TABLE *table,
   uint dst_fieldno= lcond ? table->s->period.end_fieldno
                           : table->s->period.start_fieldno;
 
-  table->file->store_auto_increment();
+  ulonglong prev_insert_id= table->file->next_insert_id;
   store_record(table, record[1]);
   if (likely(!res))
     res= src->save_in_field(table->field[dst_fieldno], true);
@@ -264,7 +264,7 @@ int update_portion_of_time(THD *thd, TABLE *table,
                                            TRG_ACTION_AFTER, true);
   restore_record(table, record[1]);
   if (res)
-    table->file->restore_auto_increment();
+    table->file->restore_auto_increment(prev_insert_id);
 
   if (likely(!res) && lcond && rcond)
     res= table->period_make_insert(period_conds.end.item,
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index 428e7b1d261..a9155f361b5 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -1401,7 +1401,8 @@ bool pushdown_cond_for_derived(THD *thd, Item *cond, TABLE_LIST *derived)
     DBUG_RETURN(false);
 
   st_select_lex_unit *unit= derived->get_unit();
-  st_select_lex *sl= unit->first_select();
+  st_select_lex *first_sl= unit->first_select();
+  st_select_lex *sl= first_sl;
 
   if (derived->prohibit_cond_pushdown)
     DBUG_RETURN(false);
@@ -1460,6 +1461,20 @@ bool pushdown_cond_for_derived(THD *thd, Item *cond, TABLE_LIST *derived)
     if (!extracted_cond_copy)
       continue;
 
+    /*
+      Rename the columns of all non-first selects of a union to be compatible
+      by names with the columns of the first select. It will allow to use copies
+      of the same expression pushed into having clauses of different selects.
+    */
+    if (sl != first_sl)
+    {
+      DBUG_ASSERT(sl->item_list.elements == first_sl->item_list.elements);
+      List_iterator_fast<Item> it(sl->item_list);
+      List_iterator_fast<Item> nm_it(unit->types);
+      while (Item *item= it++)
+        item->share_name_with(nm_it++);
+    }
+
     /* Collect fields that are used in the GROUP BY of sl */
     if (sl->have_window_funcs())
     {
@@ -1480,7 +1495,7 @@ bool pushdown_cond_for_derived(THD *thd, Item *cond, TABLE_LIST *derived)
                                     &remaining_cond,
                                     &Item::derived_field_transformer_for_where,
                                     (uchar *) sl);
-    
+
     if (!remaining_cond)
       continue;
     /*
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 7833059438e..1e456a724e7 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1769,7 +1769,7 @@ int write_record(THD *thd, TABLE *table, COPY_INFO *info, select_result *sink)
   int error, trg_error= 0;
   char *key=0;
   MY_BITMAP *save_read_set, *save_write_set;
-  table->file->store_auto_increment();
+  ulonglong prev_insert_id= table->file->next_insert_id;
   ulonglong insert_id_for_cur_row= 0;
   ulonglong prev_insert_id_for_cur_row= 0;
   DBUG_ENTER("write_record");
@@ -1918,7 +1918,7 @@ int write_record(THD *thd, TABLE *table, COPY_INFO *info, select_result *sink)
         if (res == VIEW_CHECK_ERROR)
           goto before_trg_err;
 
-        table->file->restore_auto_increment();
+        table->file->restore_auto_increment(prev_insert_id);
         info->touched++;
         if (different_records)
         {
@@ -2110,7 +2110,7 @@ int write_record(THD *thd, TABLE *table, COPY_INFO *info, select_result *sink)
     if (!(thd->variables.old_behavior &
           OLD_MODE_NO_DUP_KEY_WARNINGS_WITH_IGNORE))
       table->file->print_error(error, MYF(ME_WARNING));
-    table->file->restore_auto_increment();
+    table->file->restore_auto_increment(prev_insert_id);
     goto after_trg_or_ignored_err;
   }
 
@@ -2143,7 +2143,7 @@ err:
   table->file->print_error(error,MYF(0));
   
 before_trg_err:
-  table->file->restore_auto_increment();
+  table->file->restore_auto_increment(prev_insert_id);
   if (key)
     my_safe_afree(key, table->s->max_unique_length);
   table->column_bitmaps_set(save_read_set, save_write_set);
@@ -2297,7 +2297,7 @@ public:
     if (table)
     {
       close_thread_tables(&thd);
-      thd.mdl_context.release_transactional_locks();
+      thd.mdl_context.release_transactional_locks(&thd);
     }
     mysql_mutex_destroy(&mutex);
     mysql_cond_destroy(&cond);
@@ -3130,7 +3130,7 @@ pthread_handler_t handle_delayed_insert(void *arg)
     if (thd->mdl_context.clone_ticket(&di->grl_protection) ||
         thd->mdl_context.clone_ticket(&di->table_list.mdl_request))
     {
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       di->handler_thread_initialized= TRUE;
       goto err;
     }
@@ -3356,7 +3356,7 @@ pthread_handler_t handle_delayed_insert(void *arg)
     thd->set_killed(KILL_CONNECTION_HARD);	        // If error
 
     close_thread_tables(thd);			// Free the table
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     mysql_cond_broadcast(&di->cond_client);       // Safety
 
     mysql_mutex_lock(&LOCK_delayed_create);    // Because of delayed_get_table
@@ -4971,6 +4971,7 @@ bool select_create::send_eof()
         WSREP_ERROR("Appending table key for CTAS failed: %s, %d",
                     (wsrep_thd_query(thd)) ?
                     wsrep_thd_query(thd) : "void", rcode);
+        abort_result_set();
         DBUG_RETURN(true);
       }
       /* If commit fails, we should be able to reset the OK status. */
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index a364097931e..f09d7140d3b 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2088,7 +2088,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
         locks.
       */
       trans_rollback_implicit(thd);
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
     }
 
     thd->cleanup_after_query();
@@ -2153,7 +2153,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     ulonglong options= (ulonglong) (uchar) packet[0];
     if (trans_commit_implicit(thd))
       break;
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     if (check_global_access(thd,RELOAD_ACL))
       break;
     general_log_print(thd, command, NullS);
@@ -2188,7 +2188,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     if (trans_commit_implicit(thd))
       break;
     close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     my_ok(thd);
     break;
   }
@@ -3031,7 +3031,7 @@ err:
   /* Close tables and release metadata locks. */
   close_thread_tables(thd);
   DBUG_ASSERT(!thd->locked_tables_mode);
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
   return TRUE;
 }
 
@@ -3776,7 +3776,7 @@ mysql_execute_command(THD *thd)
       /* Commit the normal transaction if one is active. */
       bool commit_failed= trans_commit_implicit(thd);
       /* Release metadata locks acquired in this transaction. */
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       if (commit_failed)
       {
         WSREP_DEBUG("implicit commit failed, MDL released: %lld",
@@ -5087,7 +5087,7 @@ mysql_execute_command(THD *thd)
       res= trans_commit_implicit(thd);
       if (thd->locked_tables_list.unlock_locked_tables(thd))
         res= 1;
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       thd->variables.option_bits&= ~(OPTION_TABLE_LOCK);
       thd->reset_binlog_for_next_statement();
     }
@@ -5104,7 +5104,7 @@ mysql_execute_command(THD *thd)
     if (thd->locked_tables_list.unlock_locked_tables(thd))
       res= 1;
     /* Release transactional metadata locks. */
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     if (res)
       goto error;
 
@@ -5620,7 +5620,7 @@ mysql_execute_command(THD *thd)
     DBUG_PRINT("info", ("Executing SQLCOM_BEGIN  thd: %p", thd));
     if (trans_begin(thd, lex->start_transaction_opt))
     {
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       WSREP_DEBUG("BEGIN failed, MDL released: %lld",
                   (longlong) thd->thread_id);
       WSREP_DEBUG("stmt_da, sql_errno: %d", (thd->get_stmt_da()->is_error()) ? thd->get_stmt_da()->sql_errno() : 0);
@@ -5639,7 +5639,7 @@ mysql_execute_command(THD *thd)
                       (thd->variables.completion_type == 2 &&
                        lex->tx_release != TVL_NO));
     bool commit_failed= trans_commit(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     if (commit_failed)
     {
       WSREP_DEBUG("COMMIT failed, MDL released: %lld",
@@ -5677,7 +5677,7 @@ mysql_execute_command(THD *thd)
                       (thd->variables.completion_type == 2 &&
                        lex->tx_release != TVL_NO));
     bool rollback_failed= trans_rollback(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
 
     if (rollback_failed)
     {
@@ -5864,7 +5864,6 @@ mysql_execute_command(THD *thd)
   case SQLCOM_XA_COMMIT:
   {
     bool commit_failed= trans_xa_commit(thd);
-    thd->mdl_context.release_transactional_locks();
     if (commit_failed)
     {
       WSREP_DEBUG("XA commit failed, MDL released: %lld",
@@ -5882,7 +5881,6 @@ mysql_execute_command(THD *thd)
   case SQLCOM_XA_ROLLBACK:
   {
     bool rollback_failed= trans_xa_rollback(thd);
-    thd->mdl_context.release_transactional_locks();
     if (rollback_failed)
     {
       WSREP_DEBUG("XA rollback failed, MDL released: %lld",
@@ -6093,7 +6091,7 @@ finish:
     */
     THD_STAGE_INFO(thd, stage_rollback_implicit);
     trans_rollback_implicit(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
   else if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END))
   {
@@ -6107,7 +6105,7 @@ finish:
       /* Commit the normal transaction if one is active. */
       trans_commit_implicit(thd);
       thd->get_stmt_da()->set_overwrite_status(false);
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
     }
   }
   else if (! thd->in_sub_stmt && ! thd->in_multi_stmt_transaction_mode())
@@ -6122,7 +6120,7 @@ finish:
       - If in autocommit mode, or outside a transactional context,
       automatically release metadata locks of the current statement.
     */
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
   else if (! thd->in_sub_stmt)
   {
@@ -6147,7 +6145,7 @@ finish:
   {
     WSREP_DEBUG("Forcing release of transactional locks for thd: %lld",
                 (longlong) thd->thread_id);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
 
   /*
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 8f609ac0b7d..7dd025f03e4 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -4289,7 +4289,7 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   if (thd->transaction_rollback_request)
   {
     trans_rollback_implicit(thd);
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
   }
 
   /* Preserve CHANGE MASTER attributes */
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index a2cc265ec5b..808d49d4067 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -24256,8 +24256,7 @@ cmp_buffer_with_ref(THD *thd, TABLE *table, TABLE_REF *tab_ref)
 bool
 cp_buffer_from_ref(THD *thd, TABLE *table, TABLE_REF *ref)
 {
-  enum enum_check_fields save_count_cuted_fields= thd->count_cuted_fields;
-  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
+  Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
   my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set);
   bool result= 0;
 
@@ -24269,7 +24268,6 @@ cp_buffer_from_ref(THD *thd, TABLE *table, TABLE_REF *ref)
       break;
     }
   }
-  thd->count_cuted_fields= save_count_cuted_fields;
   dbug_tmp_restore_column_map(table->write_set, old_map);
   return result;
 }
diff --git a/sql/sql_select.h b/sql/sql_select.h
index bb9ecf88df0..fad008ebcb9 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -1902,18 +1902,11 @@ public:
   {
     enum store_key_result result;
     THD *thd= to_field->table->in_use;
-    enum_check_fields saved_count_cuted_fields= thd->count_cuted_fields;
-    sql_mode_t orig_sql_mode= thd->variables.sql_mode;
+    Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
+    Sql_mode_save sql_mode(thd);
     thd->variables.sql_mode&= ~(MODE_NO_ZERO_IN_DATE | MODE_NO_ZERO_DATE);
     thd->variables.sql_mode|= MODE_INVALID_DATES;
-
-    thd->count_cuted_fields= CHECK_FIELD_IGNORE;
-
     result= copy_inner();
-
-    thd->count_cuted_fields= saved_count_cuted_fields;
-    thd->variables.sql_mode= orig_sql_mode;
-
     return result;
   }
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 0f6200ff237..ca3e2aaa97e 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3671,7 +3671,6 @@ static bool show_status_array(THD *thd, const char *wild,
   char name_buffer[NAME_CHAR_LEN];
   int len;
   SHOW_VAR tmp, *var;
-  enum_check_fields save_count_cuted_fields= thd->count_cuted_fields;
   bool res= FALSE;
   CHARSET_INFO *charset= system_charset_info;
   DBUG_ENTER("show_status_array");
@@ -3794,7 +3793,6 @@ static bool show_status_array(THD *thd, const char *wild,
     }
   }
 end:
-  thd->count_cuted_fields= save_count_cuted_fields;
   DBUG_RETURN(res);
 }
 
@@ -4521,8 +4519,7 @@ fill_schema_table_by_open(THD *thd, MEM_ROOT *mem_root,
                           Open_tables_backup *open_tables_state_backup,
                           bool can_deadlock)
 {
-  Query_arena i_s_arena(mem_root,
-                        Query_arena::STMT_CONVENTIONAL_EXECUTION),
+  Query_arena i_s_arena(mem_root, Query_arena::STMT_CONVENTIONAL_EXECUTION),
               backup_arena, *old_arena;
   LEX *old_lex= thd->lex, temp_lex, *lex;
   LEX_CSTRING db_name, table_name;
@@ -5037,12 +5034,9 @@ end:
 class Warnings_only_error_handler : public Internal_error_handler
 {
 public:
-  bool handle_condition(THD *thd,
-                        uint sql_errno,
-                        const char* sqlstate,
+  bool handle_condition(THD *thd, uint sql_errno, const char* sqlstate,
                         Sql_condition::enum_warning_level *level,
-                        const char* msg,
-                        Sql_condition ** cond_hdl)
+                        const char* msg, Sql_condition ** cond_hdl)
   {
     if (sql_errno == ER_TRG_NO_DEFINER || sql_errno == ER_TRG_NO_CREATION_CTX)
       return true;
@@ -8493,13 +8487,6 @@ static int optimize_schema_tables_memory_usage(TABLE_LIST *table_list)
     DBUG_ASSERT(table->s->keys == 0);
     DBUG_ASSERT(table->s->uniques == 0);
 
-    // XXX HACK HACK HACK: in a stored function, RETURN (SELECT ...)
-    // enables warnings (in THD::sp_eval_expr) for the whole val_xxx/store pair,
-    // while the intention is to warn only for store(). Until this is
-    // fixed let's avoid data truncation warnings in I_S->fill_table()
-    if (thd->count_cuted_fields == CHECK_FIELD_IGNORE)
-    {
-
     uchar *cur= table->field[0]->ptr;
     /* first recinfo could be a NULL bitmap, not an actual Field */
     from_recinfo= to_recinfo= p->start_recinfo + (cur != table->record[0]);
@@ -8533,7 +8520,6 @@ static int optimize_schema_tables_memory_usage(TABLE_LIST *table_list)
       to_recinfo++;
     }
     p->recinfo= to_recinfo;
-    } // XXX end of HACK HACK HACK
 
     // TODO switch from Aria to Memory if all blobs were optimized away?
     if (instantiate_tmp_table(table, p->keyinfo, p->start_recinfo, &p->recinfo,
@@ -8696,6 +8682,7 @@ bool get_schema_tables_result(JOIN *join,
       }
 
       Switch_to_definer_security_ctx backup_ctx(thd, table_list);
+      Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
       if (table_list->schema_table->fill_table(thd, table_list, cond))
       {
         result= 1;
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index 2636299e330..7b600bd45c4 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -2899,7 +2899,6 @@ int read_statistics_for_table(THD *thd, TABLE *table, TABLE_LIST *stat_tables)
   Field **field_ptr;
   KEY *key_info, *key_info_end;
   TABLE_SHARE *table_share= table->s;
-  enum_check_fields old_check_level= thd->count_cuted_fields;
 
   DBUG_ENTER("read_statistics_for_table");
   DEBUG_SYNC(thd, "statistics_mem_alloc_start1");
@@ -2915,7 +2914,7 @@ int read_statistics_for_table(THD *thd, TABLE *table, TABLE_LIST *stat_tables)
   }
 
   /* Don't write warnings for internal field conversions */
-  thd->count_cuted_fields= CHECK_FIELD_IGNORE;
+  Check_level_instant_set check_level_save(thd, CHECK_FIELD_IGNORE);
 
   /* Read statistics from the statistical table table_stats */
   Table_statistics *read_stats= table_share->stats_cb.table_stats;
@@ -2997,7 +2996,6 @@ int read_statistics_for_table(THD *thd, TABLE *table, TABLE_LIST *stat_tables)
     }
   }
 
-  thd->count_cuted_fields= old_check_level;
   table_share->stats_cb.end_stats_load();
   DBUG_RETURN(0);
 }
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c2ba9bcadfb..b8c51f05f77 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -10636,16 +10636,14 @@ do_continue:;
     if (use_inplace)
     {
       table->s->frm_image= &frm;
-      enum_check_fields save_count_cuted_fields= thd->count_cuted_fields;
       /*
         Set the truncated column values of thd as warning
         for alter table.
       */
-      thd->count_cuted_fields = CHECK_FIELD_WARN;
+      Check_level_instant_set check_level_save(thd, CHECK_FIELD_WARN);
       int res= mysql_inplace_alter_table(thd, table_list, table, &altered_table,
                                          &ha_alter_info,
                                          &target_mdl_request, &alter_ctx);
-      thd->count_cuted_fields= save_count_cuted_fields;
       my_free(const_cast<uchar*>(frm.str));
 
       if (res)
diff --git a/sql/sql_tvc.cc b/sql/sql_tvc.cc
index df774a5d8dd..5f732d474f8 100644
--- a/sql/sql_tvc.cc
+++ b/sql/sql_tvc.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2017, MariaDB
+/* Copyright (c) 2017, 2020, 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
@@ -861,7 +861,8 @@ static bool cmp_row_types(Item* item1, Item* item2)
     Item *inner= item1->element_index(i);
     Item *outer= item2->element_index(i);
     if (!inner->type_handler()->subquery_type_allows_materialization(inner,
-                                                                     outer))
+                                                                     outer,
+                                                                     true))
       return true;
   }
   return false;
@@ -941,8 +942,8 @@ Item *Item_func_in::in_predicate_to_in_subs_transformer(THD *thd,
       trace_conv.add("reason", "non-constant element in the IN-list");
       return this;
     }
-    
-    if (cmp_row_types(args[0], args[i]))
+
+    if (cmp_row_types(args[i], args[0]))
     {
       trace_conv.add("done", false);
       trace_conv.add("reason", "type mismatch");
@@ -1027,6 +1028,7 @@ Item *Item_func_in::in_predicate_to_in_subs_transformer(THD *thd,
   if (!(in_subs=
           new (thd->mem_root) Item_in_subselect(thd, args[0], sq_select)))
     goto err;
+  in_subs->converted_from_in_predicate= TRUE;
   sq= in_subs;
   if (negated)
     sq= negate_expression(thd, in_subs);
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index 00b9c71cc37..6c5f72c2ce2 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -20,6 +20,7 @@
 #include "sql_const.h"
 #include "sql_class.h"
 #include "sql_time.h"
+#include "sql_string.h"
 #include "item.h"
 #include "log.h"
 #include "tztime.h"
@@ -7077,7 +7078,8 @@ uint Type_handler_timestamp_common::Item_decimal_precision(const Item *item) con
 
 bool Type_handler_real_result::
        subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const
+                                            const Item *outer,
+                                            bool is_in_predicate) const
 {
   DBUG_ASSERT(inner->cmp_type() == REAL_RESULT);
   return outer->cmp_type() == REAL_RESULT;
@@ -7086,7 +7088,8 @@ bool Type_handler_real_result::
 
 bool Type_handler_int_result::
        subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const
+                                            const Item *outer,
+                                            bool is_in_predicate) const
 {
   DBUG_ASSERT(inner->cmp_type() == INT_RESULT);
   return outer->cmp_type() == INT_RESULT;
@@ -7095,7 +7098,8 @@ bool Type_handler_int_result::
 
 bool Type_handler_decimal_result::
        subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const
+                                            const Item *outer,
+                                            bool is_in_predicate) const
 {
   DBUG_ASSERT(inner->cmp_type() == DECIMAL_RESULT);
   return outer->cmp_type() == DECIMAL_RESULT;
@@ -7104,23 +7108,37 @@ bool Type_handler_decimal_result::
 
 bool Type_handler_string_result::
        subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const
+                                            const Item *outer,
+                                            bool is_in_predicate) const
 {
   DBUG_ASSERT(inner->cmp_type() == STRING_RESULT);
-  return outer->cmp_type() == STRING_RESULT &&
-         outer->collation.collation == inner->collation.collation &&
-         /*
-           Materialization also is unable to work when create_tmp_table() will
-           create a blob column because item->max_length is too big.
-           The following test is copied from varstring_type_handler().
-         */
-         !inner->too_big_for_varchar();
+  if (outer->cmp_type() == STRING_RESULT &&
+      /*
+        Materialization also is unable to work when create_tmp_table() will
+        create a blob column because item->max_length is too big.
+        The following test is copied from varstring_type_handler().
+      */
+      !inner->too_big_for_varchar())
+  {
+    if (outer->collation.collation == inner->collation.collation)
+      return true;
+    if (is_in_predicate)
+    {
+      Charset inner_col(inner->collation.collation);
+      if (inner_col.encoding_allows_reinterpret_as(outer->
+                                                   collation.collation) &&
+          inner_col.eq_collation_specific_names(outer->collation.collation))
+        return true;
+    }
+  }
+  return false;
 }
 
 
 bool Type_handler_temporal_result::
        subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const
+                                            const Item *outer,
+                                            bool is_in_predicate) const
 {
   DBUG_ASSERT(inner->cmp_type() == TIME_RESULT);
   return mysql_timestamp_type() ==
diff --git a/sql/sql_type.h b/sql/sql_type.h
index db4f67e343f..271e74a9762 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -4049,9 +4049,21 @@ public:
                                Item *target_expr, Item *target_value,
                                Item_bool_func2 *source,
                                Item *source_expr, Item *source_const) const= 0;
+
+  /*
+    @brief
+      Check if an IN subquery allows materialization or not
+    @param
+      inner              expression on the inner side of the IN subquery
+      outer              expression on the outer side of the IN subquery
+      is_in_predicate    SET to true if IN subquery was converted from an
+                         IN predicate or we are checking if materialization
+                         strategy can be used for an IN predicate
+  */
   virtual bool
   subquery_type_allows_materialization(const Item *inner,
-                                       const Item *outer) const= 0;
+                                       const Item *outer,
+                                       bool is_in_predicate) const= 0;
   /**
     Make a simple constant replacement item for a constant "src",
     so the new item can futher be used for comparison with "cmp", e.g.:
@@ -4311,8 +4323,8 @@ public:
     DBUG_ASSERT(0);
     return 0;
   }
-  bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const override
+  bool subquery_type_allows_materialization(const Item *, const Item *,
+                                            bool) const override
   {
     DBUG_ASSERT(0);
     return false;
@@ -4714,7 +4726,8 @@ public:
   int stored_field_cmp_to_item(THD *thd, Field *field, Item *item)
                                const override;
   bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer)
+                                            const Item *outer,
+                                            bool is_in_predicate)
                                             const override;
   void make_sort_key_part(uchar *to, Item *item,
                           const SORT_FIELD_ATTR *sort_field,
@@ -4819,7 +4832,9 @@ public:
     return item_val.is_null() ? 0 : my_decimal(field).cmp(item_val.ptr());
   }
   bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const override;
+                                            const Item *outer,
+                                            bool is_in_predicate)
+    const override;
   Field *make_schema_field(MEM_ROOT *root,
                            TABLE *table,
                            const Record_addr &addr,
@@ -5073,7 +5088,9 @@ public:
   const Type_handler *type_handler_for_comparison() const override;
   int stored_field_cmp_to_item(THD *thd, Field *field, Item *item) const override;
   bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const override;
+                                            const Item *outer,
+                                            bool is_in_predicate)
+    const override;
   Field *make_num_distinct_aggregator_field(MEM_ROOT *, const Item *) const override;
   Field *make_table_field(MEM_ROOT *root,
                           const LEX_CSTRING *name,
@@ -5214,7 +5231,9 @@ public:
                                    Item *source_expr, Item *source_const)
     const override;
   bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const override;
+                                            const Item *outer,
+                                            bool is_in_predicate)
+    const override;
   bool Item_func_min_max_fix_attributes(THD *thd, Item_func_min_max *func,
                                         Item **items, uint nitems)
     const override;
@@ -5352,7 +5371,9 @@ public:
                                    Item *source_expr, Item *source_const) const
     override;
   bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const override;
+                                            const Item *outer,
+                                            bool is_in_predicate)
+    const override;
   Item *make_const_item_for_comparison(THD *, Item *src, const Item *cmp) const
     override;
   Item_cache *Item_get_cache(THD *thd, const Item *item) const override;
@@ -6983,8 +7004,8 @@ public:
   {
     return blob_type_handler(item);
   }
-  bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const override
+  bool subquery_type_allows_materialization(const Item *, const Item *, bool)
+    const override
   {
     return false; // Materialization does not work with BLOB columns
   }
@@ -7114,7 +7135,7 @@ public:
   {
     return MYSQL_TYPE_BLOB_COMPRESSED;
   }
-  ulong KEY_pack_flags(uint column_nr) const override
+  ulong KEY_pack_flags(uint) const override
   {
     DBUG_ASSERT(0);
     return 0;
@@ -7125,7 +7146,7 @@ public:
   Field *make_conversion_table_field(MEM_ROOT *root,
                                      TABLE *table, uint metadata,
                                      const Field *target) const override;
-  enum_dynamic_column_type dyncol_type(const Type_all_attributes *attr)
+  enum_dynamic_column_type dyncol_type(const Type_all_attributes *)
                                        const override
   {
     DBUG_ASSERT(0);
diff --git a/sql/sql_type_geom.h b/sql/sql_type_geom.h
index 74290e47afe..a2933114751 100644
--- a/sql/sql_type_geom.h
+++ b/sql/sql_type_geom.h
@@ -67,8 +67,8 @@ public:
            geometry_type() == th->geometry_type();
   }
   bool type_can_have_key_part() const override { return true; }
-  bool subquery_type_allows_materialization(const Item *inner,
-                                            const Item *outer) const override
+  bool subquery_type_allows_materialization(const Item *, const Item *, bool)
+    const override
   {
     return false; // Materialization does not work with GEOMETRY columns
   }
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index 9e51bb43a74..6da1175e709 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -2747,6 +2747,7 @@ bool st_select_lex::cleanup()
     delete join;
     join= 0;
   }
+  leaf_tables.empty();
   for (SELECT_LEX_UNIT *lex_unit= first_inner_unit(); lex_unit ;
        lex_unit= lex_unit->next_unit())
   {
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 64040243df0..d0eff0adada 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -4325,7 +4325,7 @@ static bool fix_autocommit(sys_var *self, THD *thd, enum_var_type type)
     if (trans_commit_stmt(thd) || trans_commit(thd))
     {
       thd->variables.option_bits&= ~OPTION_AUTOCOMMIT;
-      thd->mdl_context.release_transactional_locks();
+      thd->release_transactional_locks();
       WSREP_DEBUG("autocommit, MDL TRX lock released: %lld",
                   (longlong) thd->thread_id);
       return true;
diff --git a/sql/table.cc b/sql/table.cc
index c48a6fed89a..6cd2b1690cf 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -8741,7 +8741,7 @@ int TABLE::period_make_insert(Item *src, Field *dst)
 {
   THD *thd= in_use;
 
-  file->store_auto_increment();
+  ulonglong prev_insert_id= file->next_insert_id;
   store_record(this, record[1]);
   int res= src->save_in_field(dst, true);
 
@@ -8761,7 +8761,7 @@ int TABLE::period_make_insert(Item *src, Field *dst)
 
   restore_record(this, record[1]);
   if (res)
-    file->restore_auto_increment();
+    file->restore_auto_increment(prev_insert_id);
   return res;
 }
 
diff --git a/sql/transaction.cc b/sql/transaction.cc
index 421aa1cfc51..8bd58419ec9 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -151,7 +151,7 @@ bool trans_begin(THD *thd, uint flags)
     Release transactional metadata locks only after the
     transaction has been committed.
   */
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
 
   // The RO/RW options are mutually exclusive.
   DBUG_ASSERT(!((flags & MYSQL_START_TRANS_OPT_READ_ONLY) &&
diff --git a/sql/unireg.cc b/sql/unireg.cc
index ecd5bb8d430..bffe51b6c49 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -1151,7 +1151,6 @@ static bool make_empty_rec(THD *thd, uchar *buff, uint table_options,
   TABLE table;
   TABLE_SHARE share;
   Create_field *field;
-  enum_check_fields old_count_cuted_fields= thd->count_cuted_fields;
   DBUG_ENTER("make_empty_rec");
 
   /* We need a table to generate columns for default values */
@@ -1170,7 +1169,7 @@ static bool make_empty_rec(THD *thd, uchar *buff, uint table_options,
   null_pos= buff;
 
   List_iterator<Create_field> it(create_fields);
-  thd->count_cuted_fields= CHECK_FIELD_WARN;    // To find wrong default values
+  Check_level_instant_set check_level_save(thd, CHECK_FIELD_WARN);
   while ((field=it++))
   {
     Record_addr addr(buff + field->offset + data_offset,
@@ -1217,6 +1216,5 @@ static bool make_empty_rec(THD *thd, uchar *buff, uint table_options,
     *(null_pos + null_count / 8)|= ~(((uchar) 1 << (null_count & 7)) - 1);
 
 err:
-  thd->count_cuted_fields= old_count_cuted_fields;
   DBUG_RETURN(error);
 } /* make_empty_rec */
diff --git a/sql/wsrep_client_service.cc b/sql/wsrep_client_service.cc
index c01c256e872..24e1e198507 100644
--- a/sql/wsrep_client_service.cc
+++ b/sql/wsrep_client_service.cc
@@ -345,7 +345,7 @@ int Wsrep_client_service::bf_rollback()
   {
     m_thd->global_read_lock.unlock_global_read_lock(m_thd);
   }
-  m_thd->mdl_context.release_transactional_locks();
+  m_thd->release_transactional_locks();
   m_thd->mdl_context.release_explicit_locks();
 
   DBUG_RETURN(ret);
diff --git a/sql/wsrep_high_priority_service.cc b/sql/wsrep_high_priority_service.cc
index 5961a9574eb..3116315b9c7 100644
--- a/sql/wsrep_high_priority_service.cc
+++ b/sql/wsrep_high_priority_service.cc
@@ -290,7 +290,7 @@ int Wsrep_high_priority_service::append_fragment_and_commit(
 
   ret= ret || trans_commit(m_thd);
   ret= ret || (m_thd->wsrep_cs().after_applying(), 0);
-  m_thd->mdl_context.release_transactional_locks();
+  m_thd->release_transactional_locks();
 
   free_root(m_thd->mem_root, MYF(MY_KEEP_PREALLOC));
 
@@ -332,7 +332,7 @@ int Wsrep_high_priority_service::commit(const wsrep::ws_handle& ws_handle,
     m_rgi->cleanup_context(thd, 0);
   }
 
-  m_thd->mdl_context.release_transactional_locks();
+  m_thd->release_transactional_locks();
 
   thd_proc_info(thd, "wsrep applier committed");
 
@@ -378,7 +378,7 @@ int Wsrep_high_priority_service::rollback(const wsrep::ws_handle& ws_handle,
      assert(ws_handle == wsrep::ws_handle());
   }
   int ret= (trans_rollback_stmt(m_thd) || trans_rollback(m_thd));
-  m_thd->mdl_context.release_transactional_locks();
+  m_thd->release_transactional_locks();
   m_thd->mdl_context.release_explicit_locks();
 
   free_root(m_thd->mem_root, MYF(MY_KEEP_PREALLOC));
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 1dfeb782bbe..284df936160 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -1267,7 +1267,7 @@ wsrep_append_fk_parent_table(THD* thd, TABLE_LIST* tables, wsrep::key_array* key
   if (!WSREP(thd) || !WSREP_CLIENT(thd)) return;
     TABLE_LIST *table;
 
-    thd->mdl_context.release_transactional_locks();
+    thd->release_transactional_locks();
     uint counter;
     MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
 
diff --git a/sql/wsrep_schema.cc b/sql/wsrep_schema.cc
index 9f568226079..ed6f3ebc881 100644
--- a/sql/wsrep_schema.cc
+++ b/sql/wsrep_schema.cc
@@ -873,7 +873,7 @@ Wsrep_view Wsrep_schema::restore_view(THD* thd, const Wsrep_id& own_id) const {
       close_thread_tables(thd);
     }
   }
-  thd->mdl_context.release_transactional_locks();
+  thd->release_transactional_locks();
 
   thd->variables.wsrep_sync_wait= wsrep_sync_wait_saved;
 
diff --git a/sql/wsrep_server_service.cc b/sql/wsrep_server_service.cc
index 7ba744b4d3c..32b585871d4 100644
--- a/sql/wsrep_server_service.cc
+++ b/sql/wsrep_server_service.cc
@@ -248,7 +248,7 @@ void Wsrep_server_service::log_view(
             WSREP_WARN("Failed to commit transaction for store view");
           }
         }
-        applier->m_thd->mdl_context.release_transactional_locks();
+        applier->m_thd->release_transactional_locks();
       }
 
       /*
diff --git a/sql/wsrep_storage_service.cc b/sql/wsrep_storage_service.cc
index 2ad817fe25a..4885fd9f7e6 100644
--- a/sql/wsrep_storage_service.cc
+++ b/sql/wsrep_storage_service.cc
@@ -176,7 +176,7 @@ int Wsrep_storage_service::commit(const wsrep::ws_handle& ws_handle,
     trans_rollback(m_thd);
   }
   m_thd->wsrep_cs().after_applying();
-  m_thd->mdl_context.release_transactional_locks();
+  m_thd->release_transactional_locks();
   DBUG_RETURN(ret);
 }
 
@@ -191,7 +191,7 @@ int Wsrep_storage_service::rollback(const wsrep::ws_handle& ws_handle,
             ws_handle, ws_meta, false) ||
             trans_rollback(m_thd));
   m_thd->wsrep_cs().after_applying();
-  m_thd->mdl_context.release_transactional_locks();
+  m_thd->release_transactional_locks();
   DBUG_RETURN(ret);
 }
 
diff --git a/sql/xa.cc b/sql/xa.cc
index 15833377fb6..e0defcb92ed 100644
--- a/sql/xa.cc
+++ b/sql/xa.cc
@@ -404,6 +404,7 @@ bool xa_trans_force_rollback(THD *thd)
   xid_cache_delete(thd, &thd->transaction->xid_state);
 
   trans_track_end_trx(thd);
+  thd->mdl_context.release_transactional_locks(thd);
 
   return rc;
 }
@@ -554,11 +555,13 @@ bool trans_xa_prepare(THD *thd)
 
 /**
   Commit and terminate the a XA transaction.
+  Transactional locks are released if transaction ended
 
   @param thd    Current thread
 
   @retval FALSE  Success
   @retval TRUE   Failure
+
 */
 
 bool trans_xa_commit(THD *thd)
@@ -718,6 +721,8 @@ bool trans_xa_commit(THD *thd)
   xid_cache_delete(thd, &xid_state);
 
   trans_track_end_trx(thd);
+  thd->mdl_context.release_transactional_locks(thd);
+
   /* The transaction should be marked as complete in P_S. */
   DBUG_ASSERT(thd->m_transaction_psi == NULL || res);
   DBUG_RETURN(res);
@@ -726,6 +731,7 @@ bool trans_xa_commit(THD *thd)
 
 /**
   Roll back and terminate a XA transaction.
+  Transactional locks are released if transaction ended
 
   @param thd    Current thread
 
@@ -844,6 +850,10 @@ bool trans_xa_detach(THD *thd)
   thd->transaction->all.ha_list= 0;
   thd->transaction->all.no_2pc= 0;
   thd->m_transaction_psi= 0;
+  thd->server_status&= ~(SERVER_STATUS_IN_TRANS |
+                         SERVER_STATUS_IN_TRANS_READONLY);
+  thd->mdl_context.release_transactional_locks(thd);
+
   return false;
 }
 
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 168636cc965..f613162fb04 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -2193,6 +2193,7 @@ dict_index_remove_from_cache_low(
 	if (index->online_log) {
 		ut_ad(index->online_status == ONLINE_INDEX_CREATION);
 		row_log_free(index->online_log);
+		index->online_log = NULL;
 	}
 
 	/* Remove the index from the list of indexes of the table */
diff --git a/storage/innobase/dict/dict0stats.cc b/storage/innobase/dict/dict0stats.cc
index 07d5230b3f7..d6fbb9b7937 100644
--- a/storage/innobase/dict/dict0stats.cc
+++ b/storage/innobase/dict/dict0stats.cc
@@ -1981,7 +1981,7 @@ static index_stats_t dict_stats_analyze_index(dict_index_t* index)
 	since it will be faster and will give better results. */
 
 	if (root_level == 0
-	    || N_SAMPLE_PAGES(index) * n_uniq > index->stat_n_leaf_pages) {
+	    || N_SAMPLE_PAGES(index) * n_uniq > result.n_leaf_pages) {
 
 		if (root_level == 0) {
 			DEBUG_PRINTF("  %s(): just one page,"
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 710a3787e09..363cf50aa90 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -6957,6 +6957,7 @@ error_handling_drop_uncached:
 					if (ok && a == 1) {
 						row_log_free(
 							index->online_log);
+						index->online_log = NULL;
 						ok = false;
 					});
 
@@ -8533,6 +8534,7 @@ innobase_online_rebuild_log_free(
 		      == ONLINE_INDEX_CREATION);
 		clust_index->online_status = ONLINE_INDEX_COMPLETE;
 		row_log_free(clust_index->online_log);
+		clust_index->online_log = NULL;
 		DEBUG_SYNC_C("innodb_online_rebuild_log_free_aborted");
 	}
 
diff --git a/storage/innobase/include/row0log.h b/storage/innobase/include/row0log.h
index 63fd877691c..5ec4b9c1103 100644
--- a/storage/innobase/include/row0log.h
+++ b/storage/innobase/include/row0log.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2011, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 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
@@ -69,7 +69,7 @@ Free the row log for an index that was being created online. */
 void
 row_log_free(
 /*=========*/
-	row_log_t*&	log)	/*!< in,own: row log */
+	row_log_t*	log)	/*!< in,own: row log */
 	MY_ATTRIBUTE((nonnull));
 
 /******************************************************//**
diff --git a/storage/innobase/include/row0log.ic b/storage/innobase/include/row0log.ic
index ba7eb7b025c..44d17bbcdf1 100644
--- a/storage/innobase/include/row0log.ic
+++ b/storage/innobase/include/row0log.ic
@@ -1,6 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2011, 2015, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 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
@@ -38,6 +39,7 @@ row_log_abort_sec(
 	ut_ad(!dict_index_is_clust(index));
 	dict_index_set_online_status(index, ONLINE_INDEX_ABORTED);
 	row_log_free(index->online_log);
+	index->online_log = NULL;
 }
 
 /******************************************************//**
diff --git a/storage/innobase/mtr/mtr0mtr.cc b/storage/innobase/mtr/mtr0mtr.cc
index a6f74bf927f..4c22f05c4ba 100644
--- a/storage/innobase/mtr/mtr0mtr.cc
+++ b/storage/innobase/mtr/mtr0mtr.cc
@@ -893,7 +893,7 @@ inline std::pair<lsn_t,bool> mtr_t::finish_write(ulint len)
 	return std::make_pair(start_lsn, flush);
 }
 
-/** Find out whether a block was X-latched by the mini-transaction */
+/** Find out whether a block was not X-latched by the mini-transaction */
 struct FindBlockX
 {
   const buf_block_t &block;
@@ -903,7 +903,7 @@ struct FindBlockX
   /** @return whether the block was not found x-latched */
   bool operator()(const mtr_memo_slot_t *slot) const
   {
-    return slot->object != &block || slot->type == MTR_MEMO_PAGE_X_FIX;
+    return slot->object != &block || slot->type != MTR_MEMO_PAGE_X_FIX;
   }
 };
 
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index 02429aac23b..4a98ac24185 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -3237,7 +3237,6 @@ row_log_allocate(
 	}
 
 	dict_index_set_online_status(index, ONLINE_INDEX_CREATION);
-	index->online_log = log;
 
 	if (log_tmp_is_encrypted()) {
 		log->crypt_head_size = log->crypt_tail_size = srv_sort_buf_size;
@@ -3252,6 +3251,7 @@ row_log_allocate(
 		}
 	}
 
+	index->online_log = log;
 	/* While we might be holding an exclusive data dictionary lock
 	here, in row_log_abort_sec() we will not always be holding it. Use
 	atomic operations in both cases. */
@@ -3265,7 +3265,7 @@ Free the row log for an index that was being created online. */
 void
 row_log_free(
 /*=========*/
-	row_log_t*&	log)	/*!< in,own: row log */
+	row_log_t*	log)	/*!< in,own: row log */
 {
 	MONITOR_ATOMIC_DEC(MONITOR_ONLINE_CREATE_INDEX);
 
@@ -3285,7 +3285,6 @@ row_log_free(
 
 	mutex_free(&log->mutex);
 	ut_free(log);
-	log = NULL;
 }
 
 /******************************************************//**
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 114d83c8564..0ce136c5906 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -426,6 +426,13 @@ close_table:
 					node->heap);
 			} else {
 				node->ref = &trx_undo_metadata;
+				if (!row_undo_search_clust_to_pcur(node)) {
+					/* An error probably occurred during
+					an insert into the clustered index,
+					after we wrote the undo log record. */
+					goto close_table;
+				}
+				return true;
 			}
 
 			if (!row_undo_search_clust_to_pcur(node)) {
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index f710f54d0c1..e52ca5815f3 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -1313,7 +1313,7 @@ close_table:
 	}
 
 	/* Extract indexed virtual columns from undo log */
-	if (node->table->n_v_cols) {
+	if (node->ref != &trx_undo_metadata && node->table->n_v_cols) {
 		row_upd_replace_vcol(node->row, node->table,
 				     node->update, false, node->undo_row,
 				     (node->cmpl_info & UPD_NODE_NO_ORD_CHANGE)
diff --git a/storage/mroonga/ha_mroonga.hpp b/storage/mroonga/ha_mroonga.hpp
index f129ab74aa2..66767899e21 100644
--- a/storage/mroonga/ha_mroonga.hpp
+++ b/storage/mroonga/ha_mroonga.hpp
@@ -572,11 +572,6 @@ public:
   void set_next_insert_id(ulonglong id);
   void get_auto_increment(ulonglong offset, ulonglong increment, ulonglong nb_desired_values,
                           ulonglong *first_value, ulonglong *nb_reserved_values) mrn_override;
-  /** Fix spurious -Werror=overloaded-virtual in GCC 9 */
-  void restore_auto_increment() mrn_override
-  {
-    handler::restore_auto_increment();
-  }
   void restore_auto_increment(ulonglong prev_insert_id) mrn_override;
   void release_auto_increment() mrn_override;
   int check_for_upgrade(HA_CHECK_OPT *check_opt) mrn_override;
diff --git a/storage/oqgraph/CMakeLists.txt b/storage/oqgraph/CMakeLists.txt
index 68582e2802d..901acc6a8ca 100644
--- a/storage/oqgraph/CMakeLists.txt
+++ b/storage/oqgraph/CMakeLists.txt
@@ -20,6 +20,10 @@ INCLUDE_DIRECTORIES(${Judy_INCLUDE_DIR})
 SET(OQGRAPH_OK 1)
 ENDMACRO()
 
+IF(PLUGIN_OQGRAPH STREQUAL "NO")
+  RETURN()
+ENDIF()
+
 IF(NOT DEFINED OQGRAPH_OK)
   CHECK_OQGRAPH()
   IF (NOT OQGRAPH_OK)
diff --git a/storage/rocksdb/mysql-test/rocksdb/t/checkpoint.test b/storage/rocksdb/mysql-test/rocksdb/t/checkpoint.test
index e5de6246f60..68fe02bbd86 100644
--- a/storage/rocksdb/mysql-test/rocksdb/t/checkpoint.test
+++ b/storage/rocksdb/mysql-test/rocksdb/t/checkpoint.test
@@ -87,7 +87,7 @@ let $checkpoint = $MYSQL_TMP_DIR/already-existing-directory;
 --mkdir $checkpoint
 let $succeeds = 0;
 --source set_checkpoint.inc
---exec rm -rf $checkpoint
+rmdir $checkpoint;
 
 --disable_result_log
 truncate table t1;