MDEV-22929 MariaBackup option to report and/or continue when corruption is encounteredbb-10.2-MDEV-22929-log_corrupted_pages

The new option --log-innodb-page-corruption is introduced.

When this option is set, backup is not interrupted if innodb corrupted
page is detected. Instead it logs all found corrupted pages in
innodb_corrupted_pages file in backup directory and finishes with error.

For incremental backup corrupted pages are also copied to .delta file,
because we can't do LSN check for such pages during backup,
innodb_corrupted_pages will also be created in incremental backup
directory.

During --prepare, corrupted pages list is read from the file just after
redo log is applied, and each page from the list is checked if it is allocated
in it's tablespace or not. If it is not allocated, then it is zeroed out,
flushed to the tablespace and removed from the list. If all pages are removed
from the list, then --prepare is finished successfully and
innodb_corrupted_pages file is removed from backup directory. Otherwise
--prepare is finished with error message and innodb_corrupted_pages contains
the list of the pages, which are detected as corrupted during backup, and are
allocated in their tablespaces, what means backup directory contains corrupted
innodb pages, and backup can not be considered as consistent.

For incremental --prepare corrupted pages from .delta files are applied
to the base backup, innodb_corrupted_pages is read from both base in
incremental directories, and the same action is proceded for corrupted
pages list as for full --prepare. innodb_corrupted_pages file is
modified or removed only in base directory.

If DDL happens during backup, it is also processed at the end of backup
to have correct tablespace names in innodb_corrupted_pages.

15 files changed, 1196 insertions, 116 deletions

diff --git a/extra/mariabackup/backup_copy.cc b/extra/mariabackup/backup_copy.cc
index 0ba220364e2..d3fa3605c21 100644
--- a/extra/mariabackup/backup_copy.cc
+++ b/extra/mariabackup/backup_copy.cc
@@ -867,21 +867,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);
 
@@ -905,7 +898,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;
@@ -914,7 +906,6 @@ backup_file_vprintf(const char *filename, const char *fmt, va_list ap)
 	return(true);
 
 error:
-	free(buf);
 	if (dstfile != NULL) {
 		ds_close(dstfile);
 	}
@@ -922,8 +913,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, ...)
@@ -1446,7 +1450,7 @@ out:
 	return(ret);
 }
 
-void backup_fix_ddl(void);
+void backup_fix_ddl(CorruptedPages &);
 
 lsn_t get_current_lsn(MYSQL *connection)
 {
@@ -1471,7 +1475,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) {
@@ -1506,7 +1510,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 8f1978e967a..d92e1b2d1cc 100644
--- a/extra/mariabackup/encryption_plugin.cc
+++ b/extra/mariabackup/encryption_plugin.cc
@@ -2,7 +2,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 b229a37d934..843bceae3c6 100644
--- a/extra/mariabackup/fil_cur.cc
+++ b/extra/mariabackup/fil_cur.cc
@@ -36,6 +36,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include "read_filt.h"
 #include "xtrabackup.h"
 #include "xb0xb.h"
+#include "backup_debug.h"
 
 /* Size of read buffer in pages (640 pages = 10M for 16K sized pages) */
 #define XB_FIL_CUR_PAGES 640
@@ -381,8 +382,8 @@ 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_t*	cursor,	/*!< in/out: source file cursor */
+	CorruptedPages &corrupted_pages){
 	byte*			page;
 	ulint			i;
 	ulint			npages;
@@ -455,20 +456,40 @@ read_retry:
 			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);
 				buf_page_print(page, cursor->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 "
+				    ULINTPF ", 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,
+			{
+				ulint corrupted_page_no = 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 d4a7c0d5b39..be7d0e170d2 100644
--- a/extra/mariabackup/fil_cur.h
+++ b/extra/mariabackup/fil_cur.h
@@ -28,6 +28,7 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include <my_dir.h>
 #include "read_filt.h"
 #include "srv0start.h"
+#include "xtrabackup.h"
 
 struct xb_fil_cur_t {
 	pfs_os_file_t	file;		/*!< source file handle */
@@ -98,7 +99,8 @@ 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_t*	cursor,
+	CorruptedPages &corrupted_pages);	/*!< in/out: source file cursor */
 
 /************************************************************************
 Close the source file cursor opened with xb_fil_cur_open() and its
diff --git a/extra/mariabackup/write_filt.cc b/extra/mariabackup/write_filt.cc
index d72c11978a9..ae4c2251b00 100644
--- a/extra/mariabackup/write_filt.cc
+++ b/extra/mariabackup/write_filt.cc
@@ -26,13 +26,12 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1335  USA
 #include "common.h"
 #include "write_filt.h"
 #include "fil_cur.h"
-#include "xtrabackup.h"
 #include <os0proc.h>
 
 /************************************************************************
 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 +44,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 +64,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 +99,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);
 }
@@ -117,15 +118,16 @@ wf_incremental_process(xb_write_filt_ctxt_t *ctxt, ds_file_t *dstfile)
 	byte				*page;
 	const ulint			page_size
 		= cursor->page_size.physical();
-	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) {
@@ -163,7 +165,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.physical();
-	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);
@@ -185,7 +187,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);
 	os_mem_free_large(cp->delta_buf, cp->delta_buf_size);
 }
 
@@ -195,7 +197,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 364d1242b29..20a8a562fa5 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -76,6 +76,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 <crc_glue.h>
 #include <log.h>
 #include <derror.h>
+#include "backup_debug.h"
+
+#define MB_CORRUPTED_PAGES_FILE "innodb_corrupted_pages"
 
 int sys_var_init();
 
@@ -304,6 +308,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;
@@ -363,6 +368,207 @@ 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,
+                                      ulint 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,
+                              ulint 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, ulint 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<ulint>::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(),
+                                 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
+    {
+      ulint 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();
+
+void CorruptedPages::zero_out_free_pages()
+{
+  container_t non_free_pages;
+  byte* buf= static_cast<byte*>(ut_malloc_nokey(2 * UNIV_PAGE_SIZE));
+  byte* zero_page = static_cast<byte*>(ut_align(buf, UNIV_PAGE_SIZE));
+  memset(zero_page, 0, UNIV_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);
+    mutex_enter(&fil_system->mutex);
+    fil_space_t *space = fil_space_get_by_name(space_name.c_str());
+    mutex_exit(&fil_system->mutex);
+    if (!space)
+      die("Can't find space object for space name %s to check corrupted page",
+          space_name.c_str());
+    for (std::set<ulint>::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 " ULINTPF
+            " of tablespace %s can not be fixed",
+            *page_it, space_name.c_str());
+      }
+      else
+      {
+        const page_id_t page_id(space->id, *page_it);
+        dberr_t err= fil_io(IORequestWrite, true, page_id, univ_page_size, 0,
+                            univ_page_size.physical(), zero_page, NULL);
+        if (err != DB_SUCCESS)
+          die("Can't zero out corrupted page " ULINTPF " of tablespace %s",
+              *page_it, space_name.c_str());
+        msg("Corrupted page " ULINTPF
+            " of tablespace %s was successfuly fixed.",
+            *page_it, space_name.c_str());
+      }
+    }
+  }
+  m_spaces.swap(non_free_pages);
+  ut_a(!pthread_mutex_unlock(&m_mutex));
+  ut_free(buf);
+}
+
 /* Simple datasink creation tracking...add datasinks in the reverse order you
 want them destroyed. */
 #define XTRABACKUP_MAX_DATASINKS	10
@@ -376,11 +582,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 set_size);
 static dberr_t enumerate_ibd_files(process_single_tablespace_func_t callback);
 
-
 /* ======== Datafiles iterator ======== */
 struct datafiles_iter_t {
 	fil_system_t	*system;
@@ -732,6 +939,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 ======== */
@@ -837,7 +1045,8 @@ enum options_xtrabackup
   OPT_LOCK_DDL_PER_TABLE,
   OPT_ROCKSDB_DATADIR,
   OPT_BACKUP_ROCKSDB,
-  OPT_XTRA_CHECK_PRIVILEGES
+  OPT_XTRA_CHECK_PRIVILEGES,
+  OPT_XB_IGNORE_INNODB_PAGE_CORRUPTION
 };
 
 struct my_option xb_client_options[]= {
@@ -1234,6 +1443,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
@@ -1519,7 +1739,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 set_size)
 {
   if(dbname && tablename && !is_remote)
   {
@@ -2549,7 +2770,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;
@@ -2610,7 +2832,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;
 	}
@@ -2630,7 +2853,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;
 		}
@@ -2916,6 +3140,21 @@ static os_thread_ret_t 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
@@ -2928,35 +3167,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.*/
@@ -2969,6 +3188,7 @@ 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
@@ -2980,11 +3200,12 @@ 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);
@@ -3120,15 +3341,22 @@ 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] set_size true if we need to set tablespace size in pages explixitly.
+If this parameter is set, the size and free pages limit will not be read
+from page 0.
+*/
+static void xb_load_single_table_tablespace(const char *dirname,
+                                            const char *filname,
+                                            bool is_remote, bool set_size)
 {
 	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) {
@@ -3176,13 +3404,12 @@ xb_load_single_table_tablespace(
 	bool is_empty_file = file->exists() && file->is_empty_file();
 
 	if (err == DB_SUCCESS && file->space_id() != SRV_TMP_SPACE_ID) {
-		os_offset_t	node_size = os_file_get_size(file->handle());
-		os_offset_t	n_pages;
-
-		ut_a(node_size != (os_offset_t) -1);
-
-		n_pages = node_size / page_size_t(file->flags()).physical();
-
+		os_offset_t	n_pages = 0;
+		if (set_size) {
+			os_offset_t	node_size = os_file_get_size(file->handle());
+			ut_a(node_size != (os_offset_t) -1);
+			n_pages = node_size / page_size_t(file->flags()).physical();
+		}
 		space = fil_space_create(
 			name, file->space_id(), file->flags(),
 			FIL_TYPE_TABLESPACE, NULL/* TODO: crypt_data */);
@@ -3210,6 +3437,27 @@ xb_load_single_table_tablespace(
 	ut_free(name);
 }
 
+static void xb_load_single_table_tablespace(const std::string &space_name,
+                                            bool set_size)
+{
+  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, set_size);
+}
+
 /** 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 */
@@ -3251,7 +3499,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);
 			}
 		}
 
@@ -3308,7 +3556,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);
 				}
 			}
 
@@ -4080,6 +4328,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);
@@ -4401,6 +4650,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;
 		os_thread_create(data_copy_thread_func, data_threads + i,
 				 &data_threads[i].id);
 	}
@@ -4421,7 +4671,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();
@@ -4438,6 +4688,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;
 	}
@@ -4465,7 +4718,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);
 }
 
 
@@ -4487,7 +4746,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;
@@ -4510,6 +4769,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;
 		}
 
@@ -4530,15 +4790,21 @@ void backup_fix_ddl(void)
 				/* table was renamed, but we need a full copy
 				of it because of optimized DDL. We emulate a drop/create.*/
 				dropped_tables.insert(name);
+				if (opt_log_innodb_page_corruption)
+					corrupted_pages.drop_space(id);
 				new_tables.insert(new_name);
 			} else {
 				/* Renamed, and no optimized DDL*/
 				renamed_tables[name] = new_name;
+				if (opt_log_innodb_page_corruption)
+					corrupted_pages.rename_space(id, new_name);
 			}
 		} else if (has_optimized_ddl) {
 			/* Table was recreated, or optimized DDL ran.
 			In both cases we need a full copy in the backup.*/
 			new_tables.insert(name);
+			if (opt_log_innodb_page_corruption)
+				corrupted_pages.drop_space(id);
 		}
 	}
 
@@ -4558,6 +4824,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);
 		}
 	}
 
@@ -4600,23 +4868,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(fil_system);
@@ -4629,7 +4881,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);
@@ -5538,6 +5791,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 */
@@ -5711,6 +5965,30 @@ static bool xtrabackup_prepare_func(char** argv)
 		goto error_cleanup;
 	}
 
+        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 (xtrabackup_rollback_xa)
         {
           /* Please do not merge MDEV-21168 fix in 10.5+ */
@@ -5839,7 +6117,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 2dbdd442f95..15d53d00db4 100644
--- a/extra/mariabackup/xtrabackup.h
+++ b/extra/mariabackup/xtrabackup.h
@@ -35,6 +35,32 @@ struct xb_delta_info_t
 	ulint		space_id;
 };
 
+class CorruptedPages
+{
+public:
+  CorruptedPages();
+  ~CorruptedPages();
+  void add_page(const char *file_name, ulint space_id, ulint page_no);
+  bool contains(ulint space_id, ulint 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, ulint page_no,
+                        bool convert_space_name);
+  struct space_info_t {
+    std::string space_name;
+    std::set<ulint> 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;
 
@@ -110,6 +136,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/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 49e952eefea..1ee6038f072 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..13e373b2f70
--- /dev/null
+++ b/mysql-test/suite/mariabackup/log_page_corruption.result
@@ -0,0 +1,141 @@
+########
+# 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
+# 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);
+# 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
+# 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
+# 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 server
+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