MDEV-18519: Assertion failure in btr_page_reorganize_low()

Even after commit 0b47c126
there are a few ib::fatal() calls in non-debug code
that can be replaced easily.

btr_page_reorganize_low(): On size invariant violation, return
an error code instead of crashing.

btr_check_blob_fil_page_type(): On an invalid page type, report
an error but do not crash.

btr_copy_blob_prefix(): Truncate the output if a page type is invalid.

dict_load_foreign_cols(): On an error, return DB_CORRUPTION instead
of crashing.

fil_space_decrypt_full_crc32(), fil_space_decrypt_for_non_full_checksum():
On error, return DB_DECRYPTION_FAILED instead of crashing.

fil_set_max_space_id_if_bigger(): Replace ib::fatal() with an
equivalent ut_a() assertion.

storage/innobase/btr/btr0btr.cc

@@ -1305,10 +1305,12 @@ static dberr_t btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
     page_get_max_insert_size_after_reorganize(block->page.frame, 1);
 
   if (UNIV_UNLIKELY(data_size1 != data_size2 || max1 != max2))
-    ib::fatal() << "Page old data size " << data_size1
-                << " new data size " << data_size2
-                << ", page old max ins size " << max1
-                << " new max ins size " << max2;
+  {
+    sql_print_error("InnoDB: Page old data size %u new data size %u"
+                    ", page old max ins size %zu new max ins size %zu",
+                    data_size1, data_size2, max1, max2);
+    return DB_CORRUPTION;
+  }
 
   /* Restore the cursor position. */
   if (pos)

storage/innobase/btr/btr0cur.cc

@@ -71,6 +71,7 @@ Created 10/16/1994 Heikki Tuuri
 #ifdef WITH_WSREP
 #include "mysql/service_wsrep.h"
 #endif /* WITH_WSREP */
+#include "log.h"
 
 /** Buffered B-tree operation types, introduced as part of delete buffering. */
 enum btr_op_t {
@@ -7196,29 +7197,29 @@ btr_store_big_rec_extern_fields(
 }
 
 /** Check the FIL_PAGE_TYPE on an uncompressed BLOB page.
-@param[in]      block   uncompressed BLOB page
-@param[in]      read    true=read, false=purge */
-static void btr_check_blob_fil_page_type(const buf_block_t& block, bool read)
+@param block   uncompressed BLOB page
+@param op      operation
+@return whether the type is invalid */
+static bool btr_check_blob_fil_page_type(const buf_block_t& block,
+                                         const char *op)
 {
   uint16_t type= fil_page_get_type(block.page.frame);
 
-  if (UNIV_LIKELY(type == FIL_PAGE_TYPE_BLOB))
-    return;
-  /* FIXME: take the tablespace as a parameter */
-  if (fil_space_t *space= fil_space_t::get(block.page.id().space()))
+  if (UNIV_LIKELY(type == FIL_PAGE_TYPE_BLOB));
+  else if (fil_space_t *space= fil_space_t::get(block.page.id().space()))
   {
     /* Old versions of InnoDB did not initialize FIL_PAGE_TYPE on BLOB
     pages.  Do not print anything about the type mismatch when reading
     a BLOB page that may be from old versions. */
-    if (space->full_crc32() || DICT_TF_HAS_ATOMIC_BLOBS(space->flags))
-    {
-      ib::fatal() << "FIL_PAGE_TYPE=" << type
-		  << (read ? " on BLOB read file " : " on BLOB purge file ")
-		  << space->chain.start->name
-		  << " page " << block.page.id().page_no();
-    }
+    bool fail= space->full_crc32() || DICT_TF_HAS_ATOMIC_BLOBS(space->flags);
+    if (fail)
+      sql_print_error("InnoDB: FIL_PAGE_TYPE=%u on BLOB %s file %s page %u",
+                      type, op, space->chain.start->name,
+                      block.page.id().page_no());
     space->release();
+    return fail;
   }
+  return false;
 }
 
 /*******************************************************************//**
@@ -7365,7 +7366,7 @@ btr_free_externally_stored_field(
 			}
 		} else {
 			ut_ad(!block->page.zip.data);
-			btr_check_blob_fil_page_type(*ext_block, false);
+			btr_check_blob_fil_page_type(*ext_block, "purge");
 
 			const uint32_t next_page_no = mach_read_from_4(
 				page + FIL_PAGE_DATA
@@ -7499,14 +7500,12 @@ btr_copy_blob_prefix(
 		mtr_start(&mtr);
 
 		block = buf_page_get(id, 0, RW_S_LATCH, &mtr);
-		if (!block) {
+		if (!block || btr_check_blob_fil_page_type(*block, "read")) {
 			mtr.commit();
 			return copied_len;
 		}
 		page = buf_block_get_frame(block);
 
-		btr_check_blob_fil_page_type(*block, true);
-
 		blob_header = page + offset;
 		part_len = btr_blob_get_part_len(blob_header);
 		copy_len = ut_min(part_len, len - copied_len);

storage/innobase/dict/dict0load.cc

@@ -2657,7 +2657,6 @@ static dberr_t dict_load_foreign_cols(dict_foreign_t *foreign, trx_id_t trx_id)
 		goto func_exit;
 	}
 	for (ulint i = 0; i < foreign->n_fields; i++) {
-retry:
 		ut_a(btr_pcur_is_on_user_rec(&pcur));
 
 		const rec_t* rec = btr_pcur_get_rec(&pcur);
@@ -2690,9 +2689,7 @@ static dberr_t dict_load_foreign_cols(dict_foreign_t *foreign, trx_id_t trx_id)
 
 		if (rec_get_deleted_flag(rec, 0)) {
 			ut_ad(id);
-next:
-			btr_pcur_move_to_next_user_rec(&pcur, &mtr);
-			goto retry;
+			goto next;
 		}
 
 		field = rec_get_nth_field_old(
@@ -2718,7 +2715,7 @@ static dberr_t dict_load_foreign_cols(dict_foreign_t *foreign, trx_id_t trx_id)
 				rec, DICT_FLD__SYS_FOREIGN_COLS__REF_COL_NAME,
 				&ref_col_name_len);
 
-			ib::fatal	sout;
+			ib::error	sout;
 
 			sout << "Unable to load column names for foreign"
 				" key '" << foreign->id
@@ -2733,6 +2730,9 @@ static dberr_t dict_load_foreign_cols(dict_foreign_t *foreign, trx_id_t trx_id)
 			sout << "', REF_COL_NAME='";
 			sout.write(ref_col_name, ref_col_name_len);
 			sout << "')";
+
+			err = DB_CORRUPTION;
+			break;
 		}
 
 		field = rec_get_nth_field_old(
@@ -2750,6 +2750,7 @@ static dberr_t dict_load_foreign_cols(dict_foreign_t *foreign, trx_id_t trx_id)
 		foreign->referenced_col_names[i] = mem_heap_strdupl(
 			foreign->heap, (char*) field, len);
 
+next:
 		btr_pcur_move_to_next_user_rec(&pcur, &mtr);
 	}
 func_exit:

storage/innobase/fil/fil0crypt.cc

@@ -665,15 +665,7 @@ static dberr_t fil_space_decrypt_full_crc32(
 					   (uint) space, offset, lsn);
 
 	if (rc != MY_AES_OK || dstlen != srclen) {
-		if (rc == -1) {
-			return DB_DECRYPTION_FAILED;
-		}
-
-		ib::fatal() << "Unable to decrypt data-block "
-			    << " src: " << src << "srclen: "
-			    << srclen << " buf: " << dst << "buflen: "
-			    << dstlen << " return-code: " << rc
-			    << " Can't continue!";
+		return DB_DECRYPTION_FAILED;
 	}
 
 	/* Copy only checksum part in the trailer */
@@ -735,18 +727,7 @@ static dberr_t fil_space_decrypt_for_non_full_checksum(
 					   space, offset, lsn);
 
 	if (! ((rc == MY_AES_OK) && ((ulint) dstlen == srclen))) {
-
-		if (rc == -1) {
-			return DB_DECRYPTION_FAILED;
-		}
-
-		ib::fatal() << "Unable to decrypt data-block "
-			    << " src: " << static_cast<const void*>(src)
-			    << "srclen: "
-			    << srclen << " buf: "
-			    << static_cast<const void*>(dst) << "buflen: "
-			    << dstlen << " return-code: " << rc
-			    << " Can't continue!";
+		return DB_DECRYPTION_FAILED;
 	}
 
 	/* For compressed tables we do not store the FIL header because

storage/innobase/fil/fil0fil.cc

@@ -1393,9 +1393,7 @@ fil_set_max_space_id_if_bigger(
 /*===========================*/
 	ulint	max_id)	/*!< in: maximum known id */
 {
-	if (max_id >= SRV_SPACE_ID_UPPER_BOUND) {
-		ib::fatal() << "Max tablespace id is too high, " << max_id;
-	}
+	ut_a(max_id < SRV_SPACE_ID_UPPER_BOUND);
 
 	mysql_mutex_lock(&fil_system.mutex);