Fix bug #11830883, SUPPORT "CORRUPTED" BIT FOR INNODB TABLES AND INDEXES.

Also addressed issues in bug #11745133, where we could mark a table
corrupted instead of crashing the server when found a corrupted buffer/page
if the table created with innodb_file_per_table on.

2 files changed, 261 insertions, 11 deletions

diff --git a/storage/innobase/dict/dict0dict.c b/storage/innobase/dict/dict0dict.c
index ead86e37380..dfb733cb36c 100644
--- a/storage/innobase/dict/dict0dict.c
+++ b/storage/innobase/dict/dict0dict.c
@@ -54,6 +54,7 @@ UNIV_INTERN dict_index_t*	dict_ind_compact;
 #include "row0merge.h"
 #include "m_ctype.h" /* my_isspace() */
 #include "ha_prototypes.h" /* innobase_strcasecmp(), innobase_casedn_str()*/
+#include "row0upd.h"
 
 #include <ctype.h>
 
@@ -611,8 +612,7 @@ dict_table_get_on_id(
 {
 	dict_table_t*	table;
 
-	if (table_id <= DICT_FIELDS_ID
-	    || trx->dict_operation_lock_mode == RW_X_LATCH) {
+	if (trx->dict_operation_lock_mode == RW_X_LATCH) {
 
 		/* Note: An X latch implies that the transaction
 		already owns the dictionary mutex. */
@@ -5046,4 +5046,179 @@ dict_close(void)
 		rw_lock_free(&dict_table_stats_latches[i]);
 	}
 }
+
+/**********************************************************************//**
+Find a table in dict_sys->table_LRU list with specified space id
+@return table if found, NULL if not */
+static
+dict_table_t*
+dict_find_table_by_space(
+/*=====================*/
+	ulint	space_id)		/*!< in: space ID */
+{
+	dict_table_t*   table;
+	ulint		num_item;
+	ulint		count = 0;
+
+	ut_ad(space_id > 0);
+
+	table = UT_LIST_GET_FIRST(dict_sys->table_LRU);
+	num_item =  UT_LIST_GET_LEN(dict_sys->table_LRU);
+
+	/* This function intentionally does not acquire mutex as it is used
+	by error handling code in deep call stack as last means to avoid
+	killing the server, so it worth to risk some consequencies for
+	the action. */
+	while (table && count < num_item) {
+		if (table->space == space_id) {
+			return(table);
+		}
+
+		table = UT_LIST_GET_NEXT(table_LRU, table);
+		count++;
+	}
+
+	return(NULL);
+}
+
+/**********************************************************************//**
+Flags a table with specified space_id corrupted in the data dictionary
+cache
+@return TRUE if successful */
+UNIV_INTERN
+ibool
+dict_set_corrupted_by_space(
+/*========================*/
+	ulint	space_id)		/*!< in: space ID */
+{
+	dict_table_t*   table;
+
+	table = dict_find_table_by_space(space_id);
+
+	if (!table) {
+		return(FALSE);
+	}
+
+	/* mark the table->corrupted bit only, since the caller
+	could be too deep in the stack for SYS_INDEXES update */
+	table->corrupted = TRUE;
+
+	return(TRUE);
+}
+
+/**********************************************************************//**
+Flags an index corrupted both in the data dictionary cache
+and in the SYS_INDEXES */
+UNIV_INTERN
+void
+dict_set_corrupted(
+/*===============*/
+	dict_index_t*	index)		/*!< in/out: index */
+{
+	mem_heap_t*	heap;
+	mtr_t		mtr;
+	dict_index_t*	sys_index;
+	dtuple_t*	tuple;
+	dfield_t*	dfield;
+	byte*		buf;
+	const char*	status;
+	btr_cur_t	cursor;
+
+	ut_ad(index);
+	ut_ad(mutex_own(&dict_sys->mutex));
+	ut_ad(!dict_table_is_comp(dict_sys->sys_tables));
+	ut_ad(!dict_table_is_comp(dict_sys->sys_indexes));
+
+#ifdef UNIV_SYNC_DEBUG
+        ut_ad(sync_thread_levels_empty_except_dict());
+#endif
+
+	/* Mark the table as corrupted only if the clustered index
+	is corrupted */
+	if (dict_index_is_clust(index)) {
+		index->table->corrupted = TRUE;
+	}
+
+	if (UNIV_UNLIKELY(dict_index_is_corrupted(index))) {
+		/* The index was already flagged corrupted. */
+		ut_ad(index->table->corrupted);
+		return;
+	}
+
+	heap = mem_heap_create(sizeof(dtuple_t) + 2 * (sizeof(dfield_t)
+			       + sizeof(que_fork_t) + sizeof(upd_node_t)
+			       + sizeof(upd_t) + 12));
+	mtr_start(&mtr);
+	index->type |= DICT_CORRUPT;
+
+	sys_index = UT_LIST_GET_FIRST(dict_sys->sys_indexes->indexes);
+
+	/* Find the index row in SYS_INDEXES */
+	tuple = dtuple_create(heap, 2);
+
+	dfield = dtuple_get_nth_field(tuple, 0);
+	buf = mem_heap_alloc(heap, 8);
+	mach_write_to_8(buf, index->table->id);
+	dfield_set_data(dfield, buf, 8);
+
+	dfield = dtuple_get_nth_field(tuple, 1);
+	buf = mem_heap_alloc(heap, 8);
+	mach_write_to_8(buf, index->id);
+	dfield_set_data(dfield, buf, 8);
+
+	dict_index_copy_types(tuple, sys_index, 2);
+
+	btr_cur_search_to_nth_level(sys_index, 0, tuple, PAGE_CUR_GE,
+				    BTR_MODIFY_LEAF,
+				    &cursor, 0, __FILE__, __LINE__, &mtr);
+
+	if (cursor.up_match == dtuple_get_n_fields(tuple)) {
+		/* UPDATE SYS_INDEXES SET TYPE=index->type
+		WHERE TABLE_ID=index->table->id AND INDEX_ID=index->id */
+		ulint	len;
+		byte*	field	= rec_get_nth_field_old(
+			btr_cur_get_rec(&cursor),
+			DICT_SYS_INDEXES_TYPE_FIELD, &len);
+		if (len != 4) {
+			goto fail;
+		}
+		mlog_write_ulint(field, index->type, MLOG_4BYTES, &mtr);
+		status = "  InnoDB: Flagged corruption of ";
+	} else {
+fail:
+		status = "  InnoDB: Unable to flag corruption of ";
+	}
+
+	mtr_commit(&mtr);
+	mem_heap_free(heap);
+
+	ut_print_timestamp(stderr);
+	fputs(status, stderr);
+	dict_index_name_print(stderr, NULL, index);
+	putc('\n', stderr);
+}
+
+/**********************************************************************//**
+Flags an index corrupted in the data dictionary cache only. This
+is used mostly to mark a corrupted index when index's own dictionary
+is corrupted, and we force to load such index for repair purpose */
+UNIV_INTERN
+void
+dict_set_corrupted_index_cache_only(
+/*================================*/
+	dict_index_t*	index)		/*!< in/out: index */
+{
+	ut_ad(index);
+	ut_ad(mutex_own(&dict_sys->mutex));
+	ut_ad(!dict_table_is_comp(dict_sys->sys_tables));
+	ut_ad(!dict_table_is_comp(dict_sys->sys_indexes));
+
+	/* Mark the table as corrupted only if the clustered index
+	is corrupted */
+	if (dict_index_is_clust(index)) {
+		index->table->corrupted = TRUE;
+	}
+
+	index->type |= DICT_CORRUPT;
+}
 #endif /* !UNIV_HOTBACKUP */
diff --git a/storage/innobase/dict/dict0load.c b/storage/innobase/dict/dict0load.c
index ab1fb16361e..60590aa6638 100644
--- a/storage/innobase/dict/dict0load.c
+++ b/storage/innobase/dict/dict0load.c
@@ -52,6 +52,11 @@ static const char* SYSTEM_TABLE_NAME[] = {
 	"SYS_FOREIGN",
 	"SYS_FOREIGN_COLS"
 };
+
+/* If this flag is TRUE, then we will load the cluster index's (and tables')
+metadata even if it is marked as "corrupted". */
+UNIV_INTERN my_bool     srv_load_corrupted = FALSE;
+
 /****************************************************************//**
 Compare the name of an index column.
 @return	TRUE if the i'th column of index is 'name'. */
@@ -1324,6 +1329,9 @@ err_len:
 		goto err_len;
 	}
 	type = mach_read_from_4(field);
+	if (UNIV_UNLIKELY(type & (~0 << DICT_IT_BITS))) {
+		return("unknown SYS_INDEXES.TYPE bits");
+	}
 
 	field = rec_get_nth_field_old(rec, 7/*SPACE*/, &len);
 	if (UNIV_UNLIKELY(len != 4)) {
@@ -1423,16 +1431,47 @@ dict_load_indexes(
 			goto next_rec;
 		} else if (err_msg) {
 			fprintf(stderr, "InnoDB: %s\n", err_msg);
+			if (ignore_err & DICT_ERR_IGNORE_CORRUPT) {
+				goto next_rec;
+			}
 			error = DB_CORRUPTION;
 			goto func_exit;
 		}
 
 		ut_ad(index);
 
+		/* Check whether the index is corrupted */
+		if (dict_index_is_corrupted(index)) {
+			ut_print_timestamp(stderr);
+			fputs("  InnoDB: ", stderr);
+			dict_index_name_print(stderr, NULL, index);
+			fputs(" is corrupted\n", stderr);
+
+			if (!srv_load_corrupted
+			    && !(ignore_err & DICT_ERR_IGNORE_CORRUPT)
+			    && dict_index_is_clust(index)) {
+				dict_mem_index_free(index);
+
+				error = DB_INDEX_CORRUPT;
+				goto func_exit;
+			} else {
+				/* We will load the index if
+				1) srv_load_corrupted is TRUE
+				2) ignore_err is set with
+				DICT_ERR_IGNORE_CORRUPT
+				3) if the index corrupted is a secondary
+				index */
+				ut_print_timestamp(stderr);
+				fputs("  InnoDB: load corrupted index ", stderr);
+				dict_index_name_print(stderr, NULL, index);
+				putc('\n', stderr);
+			}
+		}
+
 		/* We check for unsupported types first, so that the
 		subsequent checks are relevant for the supported types. */
-		if (index->type & ~(DICT_CLUSTERED | DICT_UNIQUE)) {
-
+		if (index->type & ~(DICT_CLUSTERED | DICT_UNIQUE
+				    | DICT_CORRUPT)) {
 			fprintf(stderr,
 				"InnoDB: Error: unknown type %lu"
 				" of index %s of table %s\n",
@@ -1453,9 +1492,13 @@ dict_load_indexes(
 				/* If caller can tolerate this error,
 				we will continue to load the index and
 				let caller deal with this error. However
-				mark the index and table corrupted */
-				index->corrupted = TRUE;
-				table->corrupted = TRUE;
+				mark the index and table corrupted. We
+				only need to mark such in the index
+				dictionary cache for such metadata corruption,
+				since we would always be able to set it
+				when loading the dictionary cache */
+				dict_set_corrupted_index_cache_only(index);
+
 				fprintf(stderr,
 					"InnoDB: Index is corrupt but forcing"
 					" load into data dictionary\n");
@@ -1495,9 +1538,10 @@ corrupted:
 					index->name, table->name);
 
 				/* If the force recovery flag is set, and
-				if the failed index is not the primary index, we
-				will continue and open other indexes */
-				if (srv_force_recovery
+				if the failed index is not the clustered index,
+				we will continue and open other indexes */
+				if ((srv_force_recovery
+				     || srv_load_corrupted)
 				    && !dict_index_is_clust(index)) {
 					error = DB_SUCCESS;
 					goto next_rec;
@@ -1812,6 +1856,30 @@ err_exit:
 
 	err = dict_load_indexes(table, heap, ignore_err);
 
+	if (err == DB_INDEX_CORRUPT) {
+		/* Refuse to load the table if the table has a corrupted
+		cluster index */
+		if (!srv_load_corrupted) {
+			fprintf(stderr, "InnoDB: Error: Load table ");
+			ut_print_name(stderr, NULL, TRUE, table->name);
+			fprintf(stderr, " failed, the table has corrupted"
+					" clustered indexes. Turn on"
+					" 'innodb_force_load_corrupted'"
+					" to drop it\n");
+
+			dict_table_remove_from_cache(table);
+			table = NULL;
+			goto func_exit;
+		} else {
+			dict_index_t*	clust_index;
+			clust_index = dict_table_get_first_index(table);
+
+			if (dict_index_is_corrupted(clust_index)) {
+				table->corrupted = TRUE;
+			}
+		}
+	}
+
 	/* Initialize table foreign_child value. Its value could be
 	changed when dict_load_foreigns() is called below */
 	table->fk_max_recusive_level = 0;
@@ -1838,9 +1906,15 @@ err_exit:
 		index = dict_table_get_first_index(table);
 
 		if (!srv_force_recovery || !index
-		     || !dict_index_is_clust(index)) {
+		    || !dict_index_is_clust(index)) {
 			dict_table_remove_from_cache(table);
 			table = NULL;
+		} else if (dict_index_is_corrupted(index)) {
+
+			/* It is possible we force to load a corrupted
+			clustered index if srv_load_corrupted is set.
+			Mark the table as corrupted in this case */
+			table->corrupted = TRUE;
 		}
 	}
 #if 0
@@ -1867,6 +1941,7 @@ err_exit:
 		mutex_exit(&dict_foreign_err_mutex);
 	}
 #endif /* 0 */
+func_exit:
 	mem_heap_free(heap);
 
 	return(table);