MDEV-22929 MariaBackup option to report and/or continue when corruption is encountered

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.

10 files changed, 489 insertions, 127 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..3d48cb3e108 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
@@ -372,16 +373,15 @@ static bool page_is_corrupted(const byte *page, ulint page_no,
 	return buf_page_is_corrupted(true, page, cursor->page_size, space);
 }
 
-/************************************************************************
-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;
@@ -455,20 +455,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..d40610a6ca7 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 */
@@ -89,17 +90,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 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;