Fix Bug #59048 truncate table or create index could leave index->page

to be FIL_NULL

rb://545 approved by Sunny Bains

storage/innobase/dict/dict0load.c

@@ -1323,7 +1323,10 @@ ulint
 dict_load_indexes(
 /*==============*/
 	dict_table_t*	table,	/*!< in/out: table */
-	mem_heap_t*	heap)	/*!< in: memory heap for temporary storage */
+	mem_heap_t*	heap,	/*!< in: memory heap for temporary storage */
+	dict_err_ignore_t ignore_err)
+				/*!< in: error to be ignored when
+				loading the index definition */
 {
 	dict_table_t*	sys_indexes;
 	dict_index_t*	sys_index;
@@ -1406,10 +1409,22 @@ dict_load_indexes(
 				"InnoDB: but the index tree has been freed!\n",
 				index->name, table->name);
 
+			if (ignore_err & DICT_ERR_IGNORE_INDEX_ROOT) {
+				/* 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;
+				fprintf(stderr,
+					"InnoDB: Index is corrupt but forcing"
+					" load into data dictionary\n");
+			} else {
 corrupted:
-			dict_mem_index_free(index);
-			error = DB_CORRUPTION;
-			goto func_exit;
+				dict_mem_index_free(index);
+				error = DB_CORRUPTION;
+				goto func_exit;
+			}
 		} else if (!dict_index_is_clust(index)
 			   && NULL == dict_table_get_first_index(table)) {
 
@@ -1618,7 +1633,10 @@ dict_load_table(
 /*============*/
 	const char*	name,	/*!< in: table name in the
 				databasename/tablename format */
-	ibool		cached)	/*!< in: TRUE=add to cache, FALSE=do not */
+	ibool		cached,	/*!< in: TRUE=add to cache, FALSE=do not */
+	dict_err_ignore_t ignore_err)
+				/*!< in: error to be ignored when loading
+				table and its indexes' definition */
 {
 	dict_table_t*	table;
 	dict_table_t*	sys_tables;
@@ -1733,7 +1751,7 @@ err_exit:
 
 	mem_heap_empty(heap);
 
-	err = dict_load_indexes(table, heap);
+	err = dict_load_indexes(table, heap, ignore_err);
 
 	/* Initialize table foreign_child value. Its value could be
 	changed when dict_load_foreigns() is called below */
@@ -1869,7 +1887,7 @@ dict_load_table_on_id(
 	field = rec_get_nth_field_old(rec, 1, &len);
 	/* Load the table definition to memory */
 	table = dict_load_table(mem_heap_strdupl(heap, (char*) field, len),
-				TRUE);
+				TRUE, DICT_ERR_IGNORE_NONE);
 
 	btr_pcur_close(&pcur);
 	mtr_commit(&mtr);
@@ -1894,7 +1912,7 @@ dict_load_sys_table(
 
 	heap = mem_heap_create(1000);
 
-	dict_load_indexes(table, heap);
+	dict_load_indexes(table, heap, DICT_ERR_IGNORE_NONE);
 
 	mem_heap_free(heap);
 }

storage/innobase/include/dict0dict.h

@@ -441,6 +441,18 @@ function.
 @return	table, NULL if not found */
 UNIV_INLINE
 dict_table_t*
+dict_table_get_low_ignore_err(
+/*===========================*/
+	const char*	table_name,	/*!< in: table name */
+	dict_err_ignore_t
+			ignore_err);	/*!< in: error to be ignored when
+					loading a table definition */
+/**********************************************************************//**
+Gets a table; loads it to the dictionary cache if necessary. A low-level
+function.
+@return	table, NULL if not found */
+UNIV_INLINE
+dict_table_t*
 dict_table_get_low(
 /*===============*/
 	const char*	table_name);	/*!< in: table name */

storage/innobase/include/dict0dict.ic

@@ -827,6 +827,34 @@ dict_table_check_if_in_cache_low(
 	return(table);
 }
 
+/**********************************************************************//**
+load a table into dictionary cache, ignore any error specified during load; 
+@return	table, NULL if not found */
+UNIV_INLINE
+dict_table_t*
+dict_table_get_low_ignore_err(
+/*==========================*/
+	const char*	table_name,	/*!< in: table name */
+	dict_err_ignore_t
+			ignore_err)	/*!< in: error to be ignored when
+					loading a table definition */
+{
+	dict_table_t*	table;
+
+	ut_ad(table_name);
+	ut_ad(mutex_own(&(dict_sys->mutex)));
+
+	table = dict_table_check_if_in_cache_low(table_name);
+
+	if (table == NULL) {
+		table = dict_load_table(table_name, TRUE, ignore_err);
+	}
+
+	ut_ad(!table || table->cached);
+
+	return(table);
+}
+
 /**********************************************************************//**
 Gets a table; loads it to the dictionary cache if necessary. A low-level
 function.
@@ -845,7 +873,7 @@ dict_table_get_low(
 	table = dict_table_check_if_in_cache_low(table_name);
 
 	if (table == NULL) {
-		table = dict_load_table(table_name, TRUE);
+		table = dict_load_table(table_name, TRUE, DICT_ERR_IGNORE_NONE);
 	}
 
 	ut_ad(!table || table->cached);

storage/innobase/include/dict0load.h

@@ -170,7 +170,10 @@ dict_load_table(
 /*============*/
 	const char*	name,	/*!< in: table name in the
 				databasename/tablename format */
-	ibool		cached);/*!< in: TRUE=add to cache, FALSE=do not */
+	ibool		cached,	/*!< in: TRUE=add to cache, FALSE=do not */
+	dict_err_ignore_t ignore_err);
+				/*!< in: error to be ignored when loading
+				table and its indexes' definition */
 /***********************************************************************//**
 Loads a table object based on the table id.
 @return	table; NULL if table does not exist */

storage/innobase/include/dict0mem.h

@@ -361,6 +361,8 @@ struct dict_index_struct{
 				/*!< TRUE if this index is marked to be
 				dropped in ha_innobase::prepare_drop_index(),
 				otherwise FALSE */
+	unsigned	corrupted:1;
+				/*!< TRUE if the index object is corrupted */
 	dict_field_t*	fields;	/*!< array of field descriptions */
 #ifndef UNIV_HOTBACKUP
 	UT_LIST_NODE_T(dict_index_t)
@@ -494,6 +496,8 @@ struct dict_table_struct{
 				to the dictionary cache */
 	unsigned	n_def:10;/*!< number of columns defined so far */
 	unsigned	n_cols:10;/*!< number of columns */
+	unsigned	corrupted:1;
+				/*!< TRUE if table is corrupted */
 	dict_col_t*	cols;	/*!< array of column descriptions */
 	const char*	col_names;
 				/*!< Column names packed in a character string

storage/innobase/include/dict0types.h

@@ -43,4 +43,18 @@ typedef struct tab_node_struct		tab_node_t;
 typedef ib_id_t		table_id_t;
 typedef ib_id_t		index_id_t;
 
+/** Error to ignore when we load table dictionary into memory. However,
+the table and index will be marked as "corrupted", and caller will
+be responsible to deal with corrupted table or index.
+Note: please define the IGNORE_ERR_* as bits, so their value can
+be or-ed together */
+enum dict_err_ignore {
+        DICT_ERR_IGNORE_NONE = 0,        /*!< no error to ignore */
+        DICT_ERR_IGNORE_INDEX_ROOT = 1, /*!< ignore error if index root
+					page is FIL_NUL or incorrect value */
+        DICT_ERR_IGNORE_ALL = 0xFFFF	/*!< ignore all errors */
+};
+
+typedef enum dict_err_ignore		dict_err_ignore_t;
+
 #endif

storage/innobase/row/row0mysql.c

@@ -3132,7 +3132,7 @@ row_drop_table_for_mysql(
 	ut_ad(rw_lock_own(&dict_operation_lock, RW_LOCK_EX));
 #endif /* UNIV_SYNC_DEBUG */
 
-	table = dict_table_get_low(name);
+	table = dict_table_get_low_ignore_err(name, DICT_ERR_IGNORE_INDEX_ROOT);
 
 	if (!table) {
 		err = DB_TABLE_NOT_FOUND;
@@ -3367,7 +3367,7 @@ check_next_foreign:
 
 		dict_table_remove_from_cache(table);
 
-		if (dict_load_table(name, TRUE) != NULL) {
+		if (dict_load_table(name, TRUE, DICT_ERR_IGNORE_NONE) != NULL) {
 			ut_print_timestamp(stderr);
 			fputs("  InnoDB: Error: not able to remove table ",
 			      stderr);
@@ -3513,7 +3513,7 @@ row_mysql_drop_temp_tables(void)
 		btr_pcur_store_position(&pcur, &mtr);
 		btr_pcur_commit_specify_mtr(&pcur, &mtr);
 
-		table = dict_load_table(table_name, TRUE);
+		table = dict_load_table(table_name, TRUE, DICT_ERR_IGNORE_NONE);
 
 		if (table) {
 			row_drop_table_for_mysql(table_name, trx, FALSE);