MDEV-14545 Backup fails due to MLOG_INDEX_LOAD record

Problem:
=======
  Mariabackup exits during prepare phase if it encounters
MLOG_INDEX_LOAD redo log record. MLOG_INDEX_LOAD record
informs Mariabackup that the backup cannot be completed based
on the redo log scan, because some information is purposely
omitted due to bulk index creation in ALTER TABLE.

Solution:
========
Detect the MLOG_INDEX_LOAD redo record during backup phase and
exit the mariabackup with the proper error message.

5 files changed, 268 insertions, 50 deletions

diff --git a/extra/mariabackup/xtrabackup.cc b/extra/mariabackup/xtrabackup.cc
index acc66238838..78003cd64af 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -449,6 +449,43 @@ void mdl_lock_all()
 	}
 	datafiles_iter_free(it);
 }
+
+/** Check if the space id belongs to the table which name should
+be skipped based on the --tables, --tables-file and --table-exclude
+options.
+@param[in]	space_id	space id to check
+@return true if the space id belongs to skip table/database list. */
+static bool backup_includes(space_id_t space_id)
+{
+	datafiles_iter_t *it = datafiles_iter_new(fil_system);
+	if (!it)
+		return true;
+
+	while (fil_node_t *node = datafiles_iter_next(it)){
+		if (space_id == 0
+		    || (node->space->id == space_id
+			&& !check_if_skip_table(node->space->name))) {
+
+			msg("mariabackup: Unsupported redo log detected "
+			"and it belongs to %s\n",
+			space_id ? node->name: "the InnoDB system tablespace");
+
+			msg("mariabackup: ALTER TABLE or OPTIMIZE TABLE "
+			"was being executed during the backup.\n");
+
+			if (!opt_lock_ddl_per_table) {
+				msg("mariabackup: Use --lock-ddl-per-table "
+				"parameter to lock all the table before "
+				"backup operation.\n");
+			}
+
+			return false;
+		}
+	}
+
+	return true;
+}
+
 /* ======== Date copying thread context ======== */
 
 typedef struct {
@@ -2341,8 +2378,8 @@ lsn_t
 xtrabackup_copy_log(copy_logfile copy, lsn_t start_lsn, lsn_t end_lsn)
 {
 	lsn_t	scanned_lsn	= start_lsn;
-
 	const byte* log_block = log_sys->buf;
+	bool more_data = false;
 
 	for (ulint scanned_checkpoint = 0;
 	     scanned_lsn < end_lsn;
@@ -2357,8 +2394,15 @@ xtrabackup_copy_log(copy_logfile copy, lsn_t start_lsn, lsn_t end_lsn)
 		}
 
 		scanned_checkpoint = checkpoint;
+
 		ulint	data_len = log_block_get_data_len(log_block);
 
+		more_data = recv_sys_add_to_parsing_buf(
+				log_block,
+				scanned_lsn + data_len);
+
+		recv_sys->scanned_lsn = scanned_lsn + data_len;
+
 		if (data_len == OS_FILE_LOG_BLOCK_SIZE) {
 			/* We got a full log block. */
 			scanned_lsn += data_len;
@@ -2374,6 +2418,15 @@ xtrabackup_copy_log(copy_logfile copy, lsn_t start_lsn, lsn_t end_lsn)
 		}
 	}
 
+	if (more_data && recv_parse_log_recs(0, STORE_NO, false)) {
+
+		msg("mariabackup: copying the log failed \n");
+
+		return(0);
+	}
+
+	recv_sys_justify_left_parsing_buf();
+
 	log_sys->log.scanned_lsn = scanned_lsn;
 
 	end_lsn = copy == COPY_LAST
@@ -2407,9 +2460,12 @@ xtrabackup_copy_logfile(copy_logfile copy)
 	lsn_t	start_lsn;
 	lsn_t	end_lsn;
 
+	recv_sys->parse_start_lsn = log_copy_scanned_lsn;
+	recv_sys->scanned_lsn = log_copy_scanned_lsn;
+	recv_sys->recovered_lsn = log_copy_scanned_lsn;
+
 	start_lsn = ut_uint64_align_down(log_copy_scanned_lsn,
 					 OS_FILE_LOG_BLOCK_SIZE);
-
 	/* When copying the first or last part of the log, retry a few
 	times to ensure that all log up to the last checkpoint will be
 	read. */
@@ -3569,8 +3625,6 @@ xtrabackup_backup_func()
 		    "or RENAME TABLE during the backup, inconsistent backup will be "
 		    "produced.\n");
 
-
-
 	/* initialize components */
         if(innodb_init_param()) {
 fail:
@@ -3842,6 +3896,14 @@ reread_log_header:
 				 &io_watching_thread_id);
 	}
 
+	/* Populate fil_system with tablespaces to copy */
+	err = xb_load_tablespaces();
+	if (err != DB_SUCCESS) {
+		msg("mariabackup: error: xb_load_tablespaces() failed with"
+		    " error %s.\n", ut_strerr(err));
+		goto fail;
+	}
+
 	/* copy log file by current position */
 	log_copy_scanned_lsn = checkpoint_lsn_start;
 	if (xtrabackup_copy_logfile(COPY_FIRST))
@@ -3851,14 +3913,6 @@ reread_log_header:
 	log_copying_running = true;
 	os_thread_create(log_copying_thread, NULL, &log_copying_thread_id);
 
-	/* Populate fil_system with tablespaces to copy */
-	err = xb_load_tablespaces();
-	if (err != DB_SUCCESS) {
-		msg("mariabackup: error: xb_load_tablespaces() failed with"
-		    " error code %u\n", err);
-		goto fail;
-	}
-
 	/* FLUSH CHANGED_PAGE_BITMAPS call */
 	if (!flush_changed_page_bitmaps()) {
 		goto fail;
@@ -5072,6 +5126,7 @@ handle_options(int argc, char **argv, char ***argv_client, char ***argv_server)
 	srv_operation = SRV_OPERATION_RESTORE;
 
 	files_charset_info = &my_charset_utf8_general_ci;
+	check_if_backup_includes = backup_includes;
 
 	setup_error_messages();
 	sys_var_init();
diff --git a/mysql-test/suite/mariabackup/unsupported_redo.result b/mysql-test/suite/mariabackup/unsupported_redo.result
new file mode 100644
index 00000000000..be563682697
--- /dev/null
+++ b/mysql-test/suite/mariabackup/unsupported_redo.result
@@ -0,0 +1,46 @@
+call mtr.add_suppression("InnoDB: New log files created");
+call mtr.add_suppression("InnoDB: Operating system error number .* in a file operation");
+call mtr.add_suppression("InnoDB: The error means the system cannot find the path specified");
+call mtr.add_suppression("InnoDB: If you are installing InnoDB, remember that you must create directories yourself, InnoDB does not create them");
+call mtr.add_suppression("InnoDB: Ignoring tablespace for `test`\\.`t21` because it could not be opened");
+call mtr.add_suppression("InnoDB: Cannot open datafile for read-only: ");
+call mtr.add_suppression("Table .* in the InnoDB data dictionary has tablespace id .*, but tablespace with that id or name does not exist");
+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+alter table t1 FORCE, algorithm=inplace;
+# Fails during full backup
+DROP TABLE t1;
+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+INSERT INTO t1(a) select 1 union select 2 union select 3;
+# Create full backup , modify table, then fails during creation of
+#		incremental/differential backup
+alter table t1 force, algorithm=inplace;
+drop table t1;
+CREATE TABLE t1(i INT) ENGINE INNODB;
+INSERT INTO t1 VALUES(1);
+CREATE TABLE t21(i INT) ENGINE INNODB;
+INSERT INTO t21 VALUES(1);
+CREATE TABLE t2(i int) ENGINE INNODB;
+ALTER TABLE t21 FORCE, ALGORITHM=INPLACE;
+# Create partial backup (excluding table t21), Ignore the
+# unsupported redo log for the table t21.
+t1.frm
+t1.ibd
+t2.frm
+t2.ibd
+# After partial restore operation, t21 files will be missing but t21
+# table information will be present in dictionary. It will
+# restrict creating the table t21 in the future test. To avoid
+# that, take the copy of t21 files and drop the table later.
+# Prepare the full backup
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SHOW TABLES;
+Tables_in_test
+t1
+t2
+DROP TABLE t1;
+DROP TABLE t2;
+# Move the t21 files into data directory
+DROP TABLE t21;
diff --git a/mysql-test/suite/mariabackup/unsupported_redo.test b/mysql-test/suite/mariabackup/unsupported_redo.test
new file mode 100644
index 00000000000..0e3cd9f6703
--- /dev/null
+++ b/mysql-test/suite/mariabackup/unsupported_redo.test
@@ -0,0 +1,92 @@
+--source include/have_innodb.inc
+call mtr.add_suppression("InnoDB: New log files created");
+call mtr.add_suppression("InnoDB: Operating system error number .* in a file operation");
+call mtr.add_suppression("InnoDB: The error means the system cannot find the path specified");
+call mtr.add_suppression("InnoDB: If you are installing InnoDB, remember that you must create directories yourself, InnoDB does not create them");
+call mtr.add_suppression("InnoDB: Ignoring tablespace for `test`\\.`t21` because it could not be opened");
+call mtr.add_suppression("InnoDB: Cannot open datafile for read-only: ");
+call mtr.add_suppression("Table .* in the InnoDB data dictionary has tablespace id .*, but tablespace with that id or name does not exist");
+
+let $basedir=$MYSQLTEST_VARDIR/tmp/backup;
+let $incremental_dir=$MYSQLTEST_VARDIR/tmp/backup_inc1;
+
+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+alter table t1 FORCE, algorithm=inplace;
+
+# Below mariabackup operation may complete successfully if checkpoint happens
+# after the alter table command.
+
+echo # Fails during full backup;
+--disable_result_log
+--error 1
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$basedir;
+--enable_result_log
+
+DROP TABLE t1;
+rmdir $basedir;
+
+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+
+INSERT INTO t1(a) select 1 union select 2 union select 3;
+
+--echo # Create full backup , modify table, then fails during creation of
+--echo #		incremental/differential backup
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$basedir;
+--enable_result_log
+
+alter table t1 force, algorithm=inplace;
+
+--disable_result_log
+--error 1
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$incremental_dir --incremental-basedir=$basedir;
+--enable_result_log
+
+drop table t1;
+rmdir $basedir;
+rmdir $incremental_dir;
+
+CREATE TABLE t1(i INT) ENGINE INNODB;
+INSERT INTO t1 VALUES(1);
+CREATE TABLE t21(i INT) ENGINE INNODB;
+INSERT INTO t21 VALUES(1);
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/bk;
+let old_datadir=$MYSQLTEST_VARDIR/tmp/old_data;
+--mkdir $old_datadir
+
+CREATE TABLE t2(i int) ENGINE INNODB;
+ALTER TABLE t21 FORCE, ALGORITHM=INPLACE;
+
+--echo # Create partial backup (excluding table t21), Ignore the
+--echo # unsupported redo log for the table t21.
+
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup "--tables-exclude=test.t21" --target-dir=$targetdir;
+--enable_result_log
+--list_files $targetdir/test
+
+--echo # After partial restore operation, t21 files will be missing but t21
+--echo # table information will be present in dictionary. It will
+--echo # restrict creating the table t21 in the future test. To avoid
+--echo # that, take the copy of t21 files and drop the table later.
+--copy_file $MYSQLD_DATADIR/test/t21.frm $old_datadir/t21.frm
+
+--echo # Prepare the full backup
+--disable_result_log
+exec $XTRABACKUP --prepare --target-dir=$targetdir;
+--source include/restart_and_restore.inc
+--enable_result_log
+
+SHOW TABLES;
+
+DROP TABLE t1;
+DROP TABLE t2;
+
+--echo # Move the t21 files into data directory
+--copy_file $old_datadir/t21.frm $MYSQLD_DATADIR/test/t21.frm
+
+DROP TABLE t21;
+rmdir $targetdir;
+rmdir $old_datadir;
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 29784a59d95..065326ead88 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -122,6 +122,41 @@ recv_sys_var_init(void);
 void
 recv_apply_hashed_log_recs(bool last_batch);
 
+/** Whether to store redo log records to the hash table */
+enum store_t {
+	/** Do not store redo log records. */
+	STORE_NO,
+	/** Store redo log records. */
+	STORE_YES,
+	/** Store redo log records if the tablespace exists. */
+	STORE_IF_EXISTS
+};
+
+
+/** Adds data from a new log block to the parsing buffer of recv_sys if
+recv_sys->parse_start_lsn is non-zero.
+@param[in]	log_block	log block to add
+@param[in]	scanned_lsn	lsn of how far we were able to find
+				data in this log block
+@return true if more data added */
+bool recv_sys_add_to_parsing_buf(const byte* log_block, lsn_t scanned_lsn);
+
+/** Parse log records from a buffer and optionally store them to a
+hash table to wait merging to file pages.
+@param[in]	checkpoint_lsn	the LSN of the latest checkpoint
+@param[in]	store		whether to store page operations
+@param[in]	apply		whether to apply the records
+@return whether MLOG_CHECKPOINT record was seen the first time,
+or corruption was noticed */
+bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply);
+
+/** Moves the parsing buffer data left to the buffer start. */
+void recv_sys_justify_left_parsing_buf();
+
+/** Backup function checks whether the space id belongs to
+the skip table list given in the mariabackup option. */
+extern bool(*check_if_backup_includes)(ulint space_id);
+
 /** Block of log record data */
 struct recv_data_t{
 	recv_data_t*	next;	/*!< pointer to the next block or NULL */
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index e55c01aca3a..b41ea4dabee 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -158,6 +158,10 @@ typedef std::map<
 
 static recv_spaces_t	recv_spaces;
 
+/** Backup function checks whether the space id belongs to
+the skip table list given in the mariabackup option. */
+bool(*check_if_backup_includes)(ulint space_id);
+
 /** Process a file name from a MLOG_FILE_* record.
 @param[in,out]	name		file name
 @param[in]	len		length of the file name
@@ -173,6 +177,14 @@ fil_name_process(
 	ulint	space_id,
 	bool	deleted)
 {
+	if (srv_operation == SRV_OPERATION_BACKUP) {
+		return true;
+	}
+
+	ut_ad(srv_operation == SRV_OPERATION_NORMAL
+	      || srv_operation == SRV_OPERATION_RESTORE
+	      || srv_operation == SRV_OPERATION_RESTORE_EXPORT);
+
 	bool	processed = true;
 
 	/* We will also insert space=NULL into the map, so that
@@ -557,7 +569,6 @@ recv_sys_init()
 
 	recv_sys->addr_hash = hash_create(size / 512);
 	recv_sys->progress_time = ut_time();
-
 	recv_max_page_lsn = 0;
 
 	/* Call the constructor for recv_sys_t::dblwr member */
@@ -2297,30 +2308,14 @@ recv_report_corrupt_log(
 	return(true);
 }
 
-/** Whether to store redo log records to the hash table */
-enum store_t {
-	/** Do not store redo log records. */
-	STORE_NO,
-	/** Store redo log records. */
-	STORE_YES,
-	/** Store redo log records if the tablespace exists. */
-	STORE_IF_EXISTS
-};
-
 /** Parse log records from a buffer and optionally store them to a
 hash table to wait merging to file pages.
 @param[in]	checkpoint_lsn	the LSN of the latest checkpoint
 @param[in]	store		whether to store page operations
 @param[in]	apply		whether to apply the records
-@param[out]	err		DB_SUCCESS or error code
 @return whether MLOG_CHECKPOINT record was seen the first time,
 or corruption was noticed */
-static MY_ATTRIBUTE((warn_unused_result))
-bool
-recv_parse_log_recs(
-	lsn_t		checkpoint_lsn,
-	store_t		store,
-	bool		apply)
+bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 {
 	byte*		ptr;
 	byte*		end_ptr;
@@ -2460,11 +2455,14 @@ loop:
 			}
 			/* fall through */
 		case MLOG_INDEX_LOAD:
-			/* Mariabackup FIXME: Report an error
-			when encountering MLOG_INDEX_LOAD on
-			--prepare or already on --backup. */
-			ut_a(type != MLOG_INDEX_LOAD
-			     || srv_operation == SRV_OPERATION_NORMAL);
+			if (type == MLOG_INDEX_LOAD) {
+				if (check_if_backup_includes
+				    && !check_if_backup_includes(space)) {
+					ut_ad(srv_operation
+					      == SRV_OPERATION_BACKUP);
+					return true;
+				}
+			}
 			/* fall through */
 		case MLOG_FILE_NAME:
 		case MLOG_FILE_DELETE:
@@ -2652,17 +2650,13 @@ loop:
 	goto loop;
 }
 
-/*******************************************************//**
-Adds data from a new log block to the parsing buffer of recv_sys if
+/** Adds data from a new log block to the parsing buffer of recv_sys if
 recv_sys->parse_start_lsn is non-zero.
+@param[in]	log_block	log block to add
+@param[in]	scanned_lsn	lsn of how far we were able to find
+				data in this log block
 @return true if more data added */
-static
-bool
-recv_sys_add_to_parsing_buf(
-/*========================*/
-	const byte*	log_block,	/*!< in: log block */
-	lsn_t		scanned_lsn)	/*!< in: lsn of how far we were able
-					to find data in this log block */
+bool recv_sys_add_to_parsing_buf(const byte* log_block, lsn_t scanned_lsn)
 {
 	ulint	more_len;
 	ulint	data_len;
@@ -2727,12 +2721,8 @@ recv_sys_add_to_parsing_buf(
 	return(true);
 }
 
-/*******************************************************//**
-Moves the parsing buffer data left to the buffer start. */
-static
-void
-recv_sys_justify_left_parsing_buf(void)
-/*===================================*/
+/** Moves the parsing buffer data left to the buffer start. */
+void recv_sys_justify_left_parsing_buf()
 {
 	ut_memmove(recv_sys->buf, recv_sys->buf + recv_sys->recovered_offset,
 		   recv_sys->len - recv_sys->recovered_offset);