MDEV-20931 ALTER...IMPORT can crash the server

Main idea: don't log-and-crash but propogate error to the upper layers of stack
to handle it and show to a user.

mysql-test/suite/innodb/r/import_corrupted.result

+call mtr.add_suppression("Table `test`.`t2` should have 2 indexes but the tablespace has 1 indexes");
+call mtr.add_suppression("Index for table 't2' is corrupt; try to repair it");
+call mtr.add_suppression("Trying to read page number 23 in space .*, space name test/t2, which is outside the tablespace bounds. Byte offset 0, len 16384");
+CREATE TABLE t1 (
+id INT AUTO_INCREMENT PRIMARY KEY,
+not_id INT,
+data CHAR(255),
+data2 BLOB
+) ENGINE=INNODB;
+ALTER TABLE t1 MODIFY not_id INT UNIQUE KEY;
+connect  purge_control,localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+DELETE FROM t1 WHERE id % 2 = 1;
+FLUSH TABLES t1 FOR EXPORT;
+UNLOCK TABLES;
+connection purge_control;
+COMMIT;
+connection default;
+DROP TABLE t1;
+CREATE TABLE t2 (
+id INT AUTO_INCREMENT PRIMARY KEY,
+not_id INT UNIQUE KEY,
+data CHAR(255),
+data2 BLOB
+) ENGINE=INNODB;
+ALTER TABLE t2 DISCARD TABLESPACE;
+ALTER TABLE t2 IMPORT TABLESPACE;
+ERROR HY000: Index for table 't2' is corrupt; try to repair it
+DROP TABLE t2;

mysql-test/suite/innodb/t/import_corrupted.test

+--source include/have_innodb.inc
+
+call mtr.add_suppression("Table `test`.`t2` should have 2 indexes but the tablespace has 1 indexes");
+call mtr.add_suppression("Index for table 't2' is corrupt; try to repair it");
+call mtr.add_suppression("Trying to read page number 23 in space .*, space name test/t2, which is outside the tablespace bounds. Byte offset 0, len 16384");
+
+let MYSQLD_DATADIR = `SELECT @@datadir`;
+
+CREATE TABLE t1 (
+  id INT AUTO_INCREMENT PRIMARY KEY,
+  not_id INT,
+  data CHAR(255),
+  data2 BLOB
+) ENGINE=INNODB;
+
+--disable_query_log
+--let i = 0
+while ($i != 1000) {
+  eval INSERT INTO t1 VALUES (DEFAULT, $i, REPEAT('b', 255), REPEAT('a', 5000));
+  --inc $i
+}
+--enable_query_log
+
+ALTER TABLE t1 MODIFY not_id INT UNIQUE KEY;
+
+connect (purge_control,localhost,root,,);
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+
+DELETE FROM t1 WHERE id % 2 = 1;
+
+FLUSH TABLES t1 FOR EXPORT;
+
+--copy_file $MYSQLD_DATADIR/test/t1.ibd $MYSQLD_DATADIR/test/tmp.ibd
+--copy_file $MYSQLD_DATADIR/test/t1.cfg $MYSQLD_DATADIR/test/tmp.cfg
+
+perl;
+use strict;
+die unless open(FILE, "+<$ENV{MYSQLD_DATADIR}/test/tmp.ibd");
+die unless truncate(FILE, 16384*23);
+close(FILE);
+EOF
+
+UNLOCK TABLES;
+connection purge_control;
+COMMIT;
+connection default;
+DROP TABLE t1;
+
+CREATE TABLE t2 (
+  id INT AUTO_INCREMENT PRIMARY KEY,
+  not_id INT UNIQUE KEY,
+  data CHAR(255),
+  data2 BLOB
+) ENGINE=INNODB;
+
+ALTER TABLE t2 DISCARD TABLESPACE;
+
+--copy_file $MYSQLD_DATADIR/test/tmp.ibd $MYSQLD_DATADIR/test/t2.ibd
+--copy_file $MYSQLD_DATADIR/test/tmp.cfg $MYSQLD_DATADIR/test/t2.cfg
+
+--error ER_NOT_KEYFILE
+ALTER TABLE t2 IMPORT TABLESPACE;
+
+DROP TABLE t2;
+
+--remove_file $MYSQLD_DATADIR/test/tmp.ibd
+--remove_file $MYSQLD_DATADIR/test/t2.ibd

storage/innobase/btr/btr0btr.cc

@@ -3141,7 +3141,7 @@ btr_page_split_and_insert(
 @param[in,out]	page		page to remove
 @param[in]	index		index tree
 @param[in,out]	mtr		mini-transaction */
-void
+dberr_t
 btr_level_list_remove_func(
 	ulint			space,
 	const page_size_t&	page_size,
@@ -3184,6 +3184,10 @@ btr_level_list_remove_func(
 				page_id_t(space, next_page_no), page_size,
 				RW_X_LATCH, index, mtr);
 
+		if (!next_block) {
+			return DB_ERROR;
+		}
+
 		page_t*		next_page
 			= buf_block_get_frame(next_block);
 #ifdef UNIV_BTR_DEBUG
@@ -3196,6 +3200,8 @@ btr_level_list_remove_func(
 				  buf_block_get_page_zip(next_block),
 				  prev_page_no, mtr);
 	}
+
+	return DB_SUCCESS;
 }
 
 /****************************************************************//**
@@ -3675,7 +3681,10 @@ btr_compress(
 		btr_search_drop_page_hash_index(block);
 
 		/* Remove the page from the level list */
-		btr_level_list_remove(space, page_size, page, index, mtr);
+		if (DB_SUCCESS != btr_level_list_remove(space, page_size,
+							page, index, mtr)) {
+			goto err_exit;
+		}
 
 		if (dict_index_is_spatial(index)) {
 			rec_t*  my_rec = father_cursor.page_cur.rec;
@@ -3807,7 +3816,11 @@ btr_compress(
 #endif /* UNIV_BTR_DEBUG */
 
 		/* Remove the page from the level list */
-		btr_level_list_remove(space, page_size, (page_t*)page, index, mtr);
+		if (DB_SUCCESS != btr_level_list_remove(space, page_size,
+							(page_t*)page,
+							index, mtr)) {
+			goto err_exit;
+		}
 
 		ut_ad(btr_node_ptr_get_child_page_no(
 			btr_cur_get_rec(&father_cursor), offsets)
@@ -4186,7 +4199,8 @@ btr_discard_page(
 	}
 
 	/* Remove the page from the level list */
-	btr_level_list_remove(space, page_size, page, index, mtr);
+	ut_a(DB_SUCCESS == btr_level_list_remove(space, page_size, page,
+						 index, mtr));
 
 #ifdef UNIV_ZIP_DEBUG
 	{

storage/innobase/btr/btr0cur.cc

@@ -344,10 +344,12 @@ btr_cur_latch_leaves(
 				page_size, RW_X_LATCH, cursor->index, mtr);
 			latch_leaves.blocks[2] = get_block;
 #ifdef UNIV_BTR_DEBUG
+			if (get_block) {
 				ut_a(page_is_comp(get_block->frame)
 				     == page_is_comp(page));
 				ut_a(btr_page_get_prev(get_block->frame)
 				     == page_get_page_no(page));
+			}
 #endif /* UNIV_BTR_DEBUG */
 			if (spatial) {
 				cursor->rtr_info->tree_blocks[

storage/innobase/btr/btr0defragment.cc

@@ -487,7 +487,8 @@ btr_defragment_merge_pages(
 		lock_update_merge_left(to_block, orig_pred,
 				       from_block);
 		btr_search_drop_page_hash_index(from_block);
-		btr_level_list_remove(space, page_size, (page_t*)from_page, index, mtr);
+		ut_a(DB_SUCCESS == btr_level_list_remove(space, page_size,
+					   (page_t*)from_page, index, mtr));
 		btr_page_get_father(index, from_block, mtr, &parent);
 		btr_cur_node_ptr_delete(&parent, mtr);
 		/* btr_blob_dbg_remove(from_page, index,

storage/innobase/buf/buf0buf.cc

@@ -4360,6 +4360,10 @@ buf_page_get_low(
 				return (NULL);
 			}
 
+			if (local_err == DB_IO_ERROR) {
+				return NULL;
+			}
+
 			ib::fatal() << "Unable to read page " << page_id
 				<< " into the buffer pool after "
 				<< BUF_PAGE_READ_MAX_RETRIES

storage/innobase/buf/buf0rea.cc

@@ -201,7 +201,8 @@ buf_read_page_low(
 			}
 			return(0);
 		} else if (IORequest::ignore_missing(type)
-			   || *err == DB_TABLESPACE_DELETED) {
+			   || *err == DB_TABLESPACE_DELETED
+			   || *err == DB_IO_ERROR) {
 			buf_read_page_handle_error(bpage);
 			return(0);
 		}

storage/innobase/fil/fil0fil.cc

@@ -4794,27 +4794,30 @@ fil_node_complete_io(fil_node_t* node, const IORequest& type)
 	}
 }
 
-/** Report information about an invalid page access. */
-static
-void
-fil_report_invalid_page_access(
-	ulint		block_offset,	/*!< in: block offset */
-	ulint		space_id,	/*!< in: space id */
-	const char*	space_name,	/*!< in: space name */
-	ulint		byte_offset,	/*!< in: byte offset */
-	ulint		len,		/*!< in: I/O length */
-	bool		is_read)	/*!< in: I/O type */
+/** Compose error message about an invalid page access.
+@param[in]	block_offset	block offset
+@param[in]	space_id	space id
+@param[in]	space_name	space name
+@param[in]	byte_offset	byte offset
+@param[in]	len		I/O length
+@param[in]	is_read		I/O type
+@return	std::string with error message */
+static std::string fil_invalid_page_access_msg(size_t block_offset,
+                                               size_t space_id,
+                                               const char *space_name,
+                                               size_t byte_offset, size_t len,
+                                               bool is_read)
 {
-	ib::fatal()
-		<< "Trying to " << (is_read ? "read" : "write")
-		<< " page number " << block_offset << " in"
-		" space " << space_id << ", space name " << space_name << ","
-		" which is outside the tablespace bounds. Byte offset "
-		<< byte_offset << ", len " << len <<
-		(space_id == 0 && !srv_was_started
+  std::stringstream ss;
+  ss << "Trying to " << (is_read ? "read" : "write") << " page number "
+     << block_offset << " in space " << space_id << ", space name "
+     << space_name << ", which is outside the tablespace bounds. Byte offset "
+     << byte_offset << ", len " << len
+     << (space_id == 0 && !srv_was_started
              ? "Please check that the configuration matches"
                " the InnoDB system tablespace location (ibdata files)"
              : "");
+  return ss.str();
 }
 
 /** Reads or writes data. This operation could be asynchronous (aio).
@@ -4951,7 +4954,17 @@ fil_io(
 				return(DB_ERROR);
 			}
 
-			fil_report_invalid_page_access(
+			if (space->purpose == FIL_TYPE_IMPORT) {
+				mutex_exit(&fil_system->mutex);
+				ib::error() << fil_invalid_page_access_msg(
+					page_id.page_no(), page_id.space(),
+					space->name, byte_offset, len,
+					req_type.is_read());
+
+				return DB_IO_ERROR;
+			}
+
+			ib::fatal() << fil_invalid_page_access_msg(
 				page_id.page_no(), page_id.space(),
 				space->name, byte_offset, len,
 				req_type.is_read());
@@ -5032,7 +5045,7 @@ fil_io(
 			return(DB_ERROR);
 		}
 
-		fil_report_invalid_page_access(
+		ib::fatal() << fil_invalid_page_access_msg(
 			page_id.page_no(), page_id.space(),
 			space->name, byte_offset, len, req_type.is_read());
 	}

storage/innobase/include/btr0btr.h

@@ -731,7 +731,7 @@ btr_validate_index(
 /*************************************************************//**
 Removes a page from the level list of pages. */
 UNIV_INTERN
-void
+MY_ATTRIBUTE((warn_unused_result)) dberr_t
 btr_level_list_remove_func(
 /*=======================*/
 	ulint			space,	/*!< in: space where removed */