Cleanup recent mariabackup validation patches.

- Refactor code to isolate page validation in page_is_corrupted() function.

- Introduce --extended-validation parameter(default OFF) for mariabackup
--backup to enable decryption of encrypted uncompressed pages during
backup.

- mariabackup would still always check checksum on encrypted data,
it is needed to detect  partially written pages.

3 files changed, 93 insertions, 73 deletions

diff --git a/extra/mariabackup/fil_cur.cc b/extra/mariabackup/fil_cur.cc
index 637acfd2e22..b4393f7ffdd 100644
--- a/extra/mariabackup/fil_cur.cc
+++ b/extra/mariabackup/fil_cur.cc
@@ -265,6 +265,87 @@ xb_fil_cur_open(
 	return(XB_FIL_CUR_SUCCESS);
 }
 
+static bool page_is_corrupted(byte *page, ulint page_no, xb_fil_cur_t *cursor, fil_space_t *space)
+{
+	byte	tmp_frame[UNIV_PAGE_SIZE_MAX];
+	byte	tmp_page[UNIV_PAGE_SIZE_MAX];
+
+	ulint page_type = mach_read_from_2(page + FIL_PAGE_TYPE);
+
+	/* We ignore the doublewrite buffer pages.*/
+	if (cursor->space_id == TRX_SYS_SPACE
+		&& page_no >= FSP_EXTENT_SIZE
+		&& page_no < FSP_EXTENT_SIZE * 3) {
+			return false;
+	}
+
+	/* Validate page number. */
+	if (mach_read_from_4(page + FIL_PAGE_OFFSET) != page_no
+		&& space->id != TRX_SYS_SPACE) {
+		/* On pages that are not all zero, the
+		page number must match.
+
+		There may be a mismatch on tablespace ID,
+		because files may be renamed during backup.
+		We disable the page number check
+		on the system tablespace, because it may consist
+		of multiple files, and here we count the pages
+		from the start of each file.)
+
+		The first 38 and last 8 bytes are never encrypted. */
+		const ulint* p = reinterpret_cast<ulint*>(page);
+		const ulint* const end = reinterpret_cast<ulint*>(
+			page + cursor->page_size);
+		do {
+			if (*p++) {
+				return true;
+			}
+		} while (p != end);
+
+		/* Whole zero page is valid. */
+		return false;
+	}
+
+	/* Validate encrypted pages. */
+	if (mach_read_from_4(page + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION) && 
+	(space->crypt_data && space->crypt_data->type!= CRYPT_SCHEME_UNENCRYPTED)) {
+
+		if (!fil_space_verify_crypt_checksum(page, cursor->zip_size))
+			return true;
+
+		/* Compressed encrypted need to be unencryped and uncompressed for verification. */
+		if (page_type != FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED && !opt_extended_validation)
+			return false;
+
+		memcpy(tmp_page, page, cursor->page_size);
+
+		bool decrypted = false;
+		if (!fil_space_decrypt(space, tmp_frame,tmp_page, &decrypted)) {
+			return true;
+		}
+
+		if (page_type != FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED) {
+			return buf_page_is_corrupted(true, tmp_page, cursor->zip_size, space);
+		}
+	}
+
+	if (page_type == FIL_PAGE_PAGE_COMPRESSED) {
+		memcpy(tmp_page, page, cursor->page_size);
+	}
+
+	if (page_type == FIL_PAGE_PAGE_COMPRESSED || page_type == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED) {
+		ulint decomp = fil_page_decompress(tmp_frame, tmp_page);
+		page_type = mach_read_from_2(tmp_page + FIL_PAGE_TYPE);
+
+		return (!decomp
+			|| (decomp != srv_page_size && cursor->zip_size)
+			|| page_type == FIL_PAGE_PAGE_COMPRESSED
+			|| page_type == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED 
+			|| buf_page_is_corrupted(true, tmp_page, cursor->zip_size, space));
+	}
+
+	return buf_page_is_corrupted(true, page, cursor->zip_size, space);
+}
 /************************************************************************
 Reads and verifies the next block of pages from the source
 file. Positions the cursor after the last read non-corrupted page.
@@ -284,8 +365,6 @@ xb_fil_cur_read(
 	xb_fil_cur_result_t	ret;
 	ib_int64_t		offset;
 	ib_int64_t		to_read;
-	byte			tmp_frame[UNIV_PAGE_SIZE_MAX];
-	byte			tmp_page[UNIV_PAGE_SIZE_MAX];
 
 	cursor->read_filter->get_next_batch(&cursor->read_filter_ctxt,
 					    &offset, &to_read);
@@ -347,78 +426,8 @@ read_retry:
 	for (page = cursor->buf, i = 0; i < npages;
 	     page += cursor->page_size, i++) {
 		ulint page_no = cursor->buf_page_no + i;
-		ulint page_type = mach_read_from_2(page + FIL_PAGE_TYPE);
-
-		if (cursor->space_id == TRX_SYS_SPACE
-		    && page_no >= FSP_EXTENT_SIZE
-		    && page_no < FSP_EXTENT_SIZE * 3) {
-			/* We ignore the doublewrite buffer pages */
-                } else if (mach_read_from_4(page + FIL_PAGE_OFFSET) != page_no
-			   && space->id != TRX_SYS_SPACE) {
-			/* On pages that are not all zero, the
-			page number must match.
-
-			There may be a mismatch on tablespace ID,
-			because files may be renamed during backup.
-			We disable the page number check
-			on the system tablespace, because it may consist
-			of multiple files, and here we count the pages
-			from the start of each file.)
-
-			The first 38 and last 8 bytes are never encrypted. */
-			const ulint* p = reinterpret_cast<ulint*>(page);
-			const ulint* const end = reinterpret_cast<ulint*>(
-				page + cursor->page_size);
-			do {
-				if (*p++) {
-					goto corrupted;
-				}
-			} while (p != end);
-		} else if (mach_read_from_4(
-				   page
-				   + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION)
-			   && space->crypt_data
-			   && space->crypt_data->type
-			   != CRYPT_SCHEME_UNENCRYPTED
-			   && fil_space_verify_crypt_checksum(
-				   page, cursor->zip_size)) {
-			bool decrypted = false;
-
-			memcpy(tmp_page, page, cursor->page_size);
-
-			if (!fil_space_decrypt(space, tmp_frame,
-					       tmp_page, &decrypted)) {
-				goto corrupted;
-			}
-
-			if (page_type == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED) {
-				goto page_decomp;
-			}
-
-			if (buf_page_is_corrupted(
-				true, tmp_page, cursor->zip_size, space)) {
-				goto corrupted;
-			}
-		} else if (page_type == FIL_PAGE_PAGE_COMPRESSED) {
-			memcpy(tmp_page, page, cursor->page_size);
-page_decomp:
-			ulint decomp = fil_page_decompress(tmp_frame, tmp_page);
-			page_type = mach_read_from_2(tmp_page + FIL_PAGE_TYPE);
-
-			if (!decomp
-			    || (decomp != srv_page_size && cursor->zip_size)
-			    || page_type == FIL_PAGE_PAGE_COMPRESSED
-			    || page_type == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED
-			    || buf_page_is_corrupted(true, tmp_page,
-						     cursor->zip_size,
-						     space)) {
-				goto corrupted;
-			}
 
-		} else if (page_type == FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED
-			   || buf_page_is_corrupted(true, page,
-						    cursor->zip_size, space)) {
-corrupted:
+		if (page_is_corrupted(page, page_no, cursor, space)){
 			retry_count--;
 
 			if (retry_count == 0) {
diff --git a/extra/mariabackup/xtrabackup.cc b/extra/mariabackup/xtrabackup.cc
index be022f7afa2..d2b241a28f5 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -206,6 +206,7 @@ char*	log_ignored_opt				= NULL;
 
 extern my_bool opt_use_ssl;
 my_bool opt_ssl_verify_server_cert;
+my_bool opt_extended_validation;
 
 /* === metadata of backup === */
 #define XTRABACKUP_METADATA_FILENAME "xtrabackup_checkpoints"
@@ -510,6 +511,7 @@ enum options_xtrabackup
   OPT_XTRA_DATABASES_FILE,
   OPT_XTRA_CREATE_IB_LOGFILE,
   OPT_XTRA_PARALLEL,
+  OPT_XTRA_EXTENDED_VALIDATION,
   OPT_XTRA_STREAM,
   OPT_XTRA_COMPRESS,
   OPT_XTRA_COMPRESS_THREADS,
@@ -976,6 +978,14 @@ struct my_option xb_server_options[] =
    (G_PTR*) &xtrabackup_parallel, (G_PTR*) &xtrabackup_parallel, 0, GET_INT,
    REQUIRED_ARG, 1, 1, INT_MAX, 0, 0, 0},
 
+  {"extended_validation", OPT_XTRA_EXTENDED_VALIDATION,
+   "Enable extended validation for Innodb data pages during backup phase."
+   "Will slow down backup considerably, in case encryption is used.",
+   (G_PTR*)&opt_extended_validation,
+   (G_PTR*)&opt_extended_validation,
+   0, GET_BOOL, NO_ARG, FALSE, 0, 0, 0, 0, 0},
+
+
    {"log", OPT_LOG, "Ignored option for MySQL option compatibility",
    (G_PTR*) &log_ignored_opt, (G_PTR*) &log_ignored_opt, 0,
    GET_STR, OPT_ARG, 0, 0, 0, 0, 0, 0},
diff --git a/extra/mariabackup/xtrabackup.h b/extra/mariabackup/xtrabackup.h
index 9af7c861dd7..2af5396fa06 100644
--- a/extra/mariabackup/xtrabackup.h
+++ b/extra/mariabackup/xtrabackup.h
@@ -128,6 +128,7 @@ extern my_bool		opt_noversioncheck;
 extern my_bool		opt_no_backup_locks;
 extern my_bool		opt_decompress;
 extern my_bool		opt_remove_original;
+extern my_bool		opt_extended_validation;
 
 extern char		*opt_incremental_history_name;
 extern char		*opt_incremental_history_uuid;