innodb_pass_corrupt_table.patch

   1 # name       : innodb_pass_corrupt_table.patch
   2 # introduced : 11 or before
   3 # maintainer : Yasufumi
   4 #
   5 #!!! notice !!!
   6 # Any small change to this file in the main branch
   7 # should be done or reviewed by the maintainer!
   8 --- a/storage/innobase/btr/btr0btr.c
   9 +++ b/storage/innobase/btr/btr0btr.c
  10 @@ -713,6 +713,12 @@
  11
  12         block = btr_block_get(space, zip_size, root_page_no, RW_X_LATCH,
  13                               index, mtr);
  14 +
  15 +       if (srv_pass_corrupt_table && !block) {
  16 +               return(0);
  17 +       }
  18 +       ut_a(block);
  19 +
  20         btr_assert_not_corrupted(block, index);
  21  #ifdef UNIV_BTR_DEBUG
  22         if (!dict_index_is_ibuf(index)) {
  23 @@ -998,6 +1004,12 @@
  24
  25         root = btr_root_get(index, &mtr);
  26
  27 +       if (srv_pass_corrupt_table && !root) {
  28 +               mtr_commit(&mtr);
  29 +               return(0);
  30 +       }
  31 +       ut_a(root);
  32 +
  33         if (flag == BTR_N_LEAF_PAGES) {
  34                 seg_header = root + PAGE_HEADER + PAGE_BTR_SEG_LEAF;
  35
  36 @@ -1457,6 +1469,13 @@
  37
  38         root = btr_page_get(space, zip_size, root_page_no, RW_X_LATCH,
  39                             NULL, &mtr);
  40 +
  41 +       if (srv_pass_corrupt_table && !root) {
  42 +               mtr_commit(&mtr);
  43 +               return;
  44 +       }
  45 +       ut_a(root);
  46 +
  47  #ifdef UNIV_BTR_DEBUG
  48         ut_a(btr_root_fseg_validate(FIL_PAGE_DATA + PAGE_BTR_SEG_LEAF
  49                                     + root, space));
  50 @@ -1480,6 +1499,12 @@
  51
  52         root = btr_page_get(space, zip_size, root_page_no, RW_X_LATCH,
  53                             NULL, &mtr);
  54 +
  55 +       if (srv_pass_corrupt_table && !root) {
  56 +               mtr_commit(&mtr);
  57 +               return;
  58 +       }
  59 +       ut_a(root);
  60  #ifdef UNIV_BTR_DEBUG
  61         ut_a(btr_root_fseg_validate(FIL_PAGE_DATA + PAGE_BTR_SEG_TOP
  62                                     + root, space));
  63 @@ -1513,6 +1538,11 @@
  64         block = btr_block_get(space, zip_size, root_page_no, RW_X_LATCH,
  65                               NULL, mtr);
  66
  67 +       if (srv_pass_corrupt_table && !block) {
  68 +               return;
  69 +       }
  70 +       ut_a(block);
  71 +
  72         btr_search_drop_page_hash_index(block);
  73
  74         header = buf_block_get_frame(block) + PAGE_HEADER + PAGE_BTR_SEG_TOP;
  75 --- a/storage/innobase/btr/btr0cur.c
  76 +++ b/storage/innobase/btr/btr0cur.c
  77 @@ -251,6 +251,11 @@
  78                 mode = latch_mode == BTR_SEARCH_LEAF ? RW_S_LATCH : RW_X_LATCH;
  79                 get_block = btr_block_get(
  80                         space, zip_size, page_no, mode, cursor->index, mtr);
  81 +
  82 +               if (srv_pass_corrupt_table && !get_block) {
  83 +                       return;
  84 +               }
  85 +               ut_a(get_block);
  86  #ifdef UNIV_BTR_DEBUG
  87                 ut_a(page_is_comp(get_block->frame) == page_is_comp(page));
  88  #endif /* UNIV_BTR_DEBUG */
  89 @@ -264,6 +269,11 @@
  90                         get_block = btr_block_get(
  91                                 space, zip_size, left_page_no,
  92                                 RW_X_LATCH, cursor->index, mtr);
  93 +
  94 +                       if (srv_pass_corrupt_table && !get_block) {
  95 +                               return;
  96 +                       }
  97 +                       ut_a(get_block);
  98  #ifdef UNIV_BTR_DEBUG
  99                         ut_a(page_is_comp(get_block->frame)
 100                              == page_is_comp(page));
 101 @@ -276,6 +286,11 @@
 102                 get_block = btr_block_get(
 103                         space, zip_size, page_no,
 104                         RW_X_LATCH, cursor->index, mtr);
 105 +
 106 +               if (srv_pass_corrupt_table && !get_block) {
 107 +                       return;
 108 +               }
 109 +               ut_a(get_block);
 110  #ifdef UNIV_BTR_DEBUG
 111                 ut_a(page_is_comp(get_block->frame) == page_is_comp(page));
 112  #endif /* UNIV_BTR_DEBUG */
 113 @@ -287,6 +302,11 @@
 114                         get_block = btr_block_get(
 115                                 space, zip_size, right_page_no,
 116                                 RW_X_LATCH, cursor->index, mtr);
 117 +
 118 +                       if (srv_pass_corrupt_table && !get_block) {
 119 +                               return;
 120 +                       }
 121 +                       ut_a(get_block);
 122  #ifdef UNIV_BTR_DEBUG
 123                         ut_a(page_is_comp(get_block->frame)
 124                              == page_is_comp(page));
 125 @@ -309,6 +329,11 @@
 126                                 space, zip_size,
 127                                 left_page_no, mode, cursor->index, mtr);
 128                         cursor->left_block = get_block;
 129 +
 130 +                       if (srv_pass_corrupt_table && !get_block) {
 131 +                               return;
 132 +                       }
 133 +                       ut_a(get_block);
 134  #ifdef UNIV_BTR_DEBUG
 135                         ut_a(page_is_comp(get_block->frame)
 136                              == page_is_comp(page));
 137 @@ -320,6 +345,11 @@
 138
 139                 get_block = btr_block_get(
 140                         space, zip_size, page_no, mode, cursor->index, mtr);
 141 +
 142 +               if (srv_pass_corrupt_table && !get_block) {
 143 +                       return;
 144 +               }
 145 +               ut_a(get_block);
 146  #ifdef UNIV_BTR_DEBUG
 147                 ut_a(page_is_comp(get_block->frame) == page_is_comp(page));
 148  #endif /* UNIV_BTR_DEBUG */
 149 @@ -595,6 +625,19 @@
 150                 file, line, mtr);
 151
 152         if (block == NULL) {
 153 +               if (srv_pass_corrupt_table
 154 +                   && buf_mode != BUF_GET_IF_IN_POOL
 155 +                   && buf_mode != BUF_GET_IF_IN_POOL_OR_WATCH) {
 156 +                       page_cursor->block = 0;
 157 +                       page_cursor->rec = 0;
 158 +                       if (estimate) {
 159 +                               cursor->path_arr->nth_rec = ULINT_UNDEFINED;
 160 +                       }
 161 +                       goto func_exit;
 162 +               }
 163 +               ut_a(buf_mode == BUF_GET_IF_IN_POOL
 164 +                    || buf_mode == BUF_GET_IF_IN_POOL_OR_WATCH);
 165 +
 166                 /* This must be a search to perform an insert/delete
 167                 mark/ delete; try using the insert/delete buffer */
 168
 169 @@ -669,6 +712,16 @@
 170         block->check_index_page_at_flush = TRUE;
 171         page = buf_block_get_frame(block);
 172
 173 +       if (srv_pass_corrupt_table && !page) {
 174 +               page_cursor->block = 0;
 175 +               page_cursor->rec = 0;
 176 +               if (estimate) {
 177 +                       cursor->path_arr->nth_rec = ULINT_UNDEFINED;
 178 +               }
 179 +               goto func_exit;
 180 +       }
 181 +       ut_a(page);
 182 +
 183         if (rw_latch != RW_NO_LATCH) {
 184  #ifdef UNIV_ZIP_DEBUG
 185                 const page_zip_des_t*   page_zip
 186 @@ -862,6 +915,17 @@
 187                                          RW_NO_LATCH, NULL, BUF_GET,
 188                                          file, line, mtr);
 189                 page = buf_block_get_frame(block);
 190 +
 191 +               if (srv_pass_corrupt_table && !page) {
 192 +                       page_cursor->block = 0;
 193 +                       page_cursor->rec = 0;
 194 +                       if (estimate) {
 195 +                               cursor->path_arr->nth_rec = ULINT_UNDEFINED;
 196 +                       }
 197 +                       break;
 198 +               }
 199 +               ut_a(page);
 200 +
 201                 ut_ad(index->id == btr_page_get_index_id(page));
 202
 203                 block->check_index_page_at_flush = TRUE;
 204 @@ -982,6 +1046,14 @@
 205                                          RW_NO_LATCH, NULL, BUF_GET,
 206                                          file, line, mtr);
 207                 page = buf_block_get_frame(block);
 208 +
 209 +               if (srv_pass_corrupt_table && !page) {
 210 +                       page_cursor->block = 0;
 211 +                       page_cursor->rec = 0;
 212 +                       break;
 213 +               }
 214 +               ut_a(page);
 215 +
 216                 ut_ad(index->id == btr_page_get_index_id(page));
 217
 218                 if (height == ULINT_UNDEFINED) {
 219 @@ -1195,6 +1267,12 @@
 220         *big_rec = NULL;
 221
 222         block = btr_cur_get_block(cursor);
 223 +
 224 +       if (srv_pass_corrupt_table && !block) {
 225 +               return(DB_CORRUPTION);
 226 +       }
 227 +       ut_a(block);
 228 +
 229         page = buf_block_get_frame(block);
 230         index = cursor->index;
 231         zip_size = buf_block_get_zip_size(block);
 232 @@ -2927,6 +3005,11 @@
 233
 234         block = btr_cur_get_block(cursor);
 235
 236 +       if (srv_pass_corrupt_table && !block) {
 237 +               return(DB_CORRUPTION);
 238 +       }
 239 +       ut_a(block);
 240 +
 241         ut_ad(page_is_leaf(buf_block_get_frame(block)));
 242
 243         rec = btr_cur_get_rec(cursor);
 244 @@ -3635,6 +3718,11 @@
 245
 246                 page = btr_cur_get_page(&cursor);
 247
 248 +               if (srv_pass_corrupt_table && !page) {
 249 +                       break;
 250 +               }
 251 +               ut_a(page);
 252 +
 253                 rec = page_rec_get_next(page_get_infimum_rec(page));
 254
 255                 if (!page_rec_is_supremum(rec)) {
 256 --- a/storage/innobase/btr/btr0pcur.c
 257 +++ b/storage/innobase/btr/btr0pcur.c
 258 @@ -32,7 +32,7 @@
 259  #include "ut0byte.h"
 260  #include "rem0cmp.h"
 261  #include "trx0trx.h"
 262 -
 263 +#include "srv0srv.h"
 264  /**************************************************************//**
 265  Allocates memory for a persistent cursor object and initializes the cursor.
 266  @return        own: persistent cursor */
 267 @@ -114,6 +114,12 @@
 268         ut_ad(cursor->latch_mode != BTR_NO_LATCHES);
 269
 270         block = btr_pcur_get_block(cursor);
 271 +
 272 +       if (srv_pass_corrupt_table && !block) {
 273 +               return;
 274 +       }
 275 +       ut_a(block);
 276 +
 277         index = btr_cur_get_index(btr_pcur_get_btr_cur(cursor));
 278
 279         page_cursor = btr_pcur_get_page_cur(cursor);
 280 @@ -409,6 +415,15 @@
 281                                    cursor->latch_mode,
 282                                    btr_pcur_get_btr_cur(cursor)->index, mtr);
 283         next_page = buf_block_get_frame(next_block);
 284 +
 285 +       if (srv_pass_corrupt_table && !next_page) {
 286 +               btr_leaf_page_release(btr_pcur_get_block(cursor),
 287 +                                     cursor->latch_mode, mtr);
 288 +               btr_pcur_get_page_cur(cursor)->block = 0;
 289 +               btr_pcur_get_page_cur(cursor)->rec = 0;
 290 +               return;
 291 +       }
 292 +       ut_a(next_page);
 293  #ifdef UNIV_BTR_DEBUG
 294         ut_a(page_is_comp(next_page) == page_is_comp(page));
 295         ut_a(btr_page_get_prev(next_page, mtr)
 296 --- a/storage/innobase/btr/btr0sea.c
 297 +++ b/storage/innobase/btr/btr0sea.c
 298 @@ -42,7 +42,7 @@
 299  #include "btr0pcur.h"
 300  #include "btr0btr.h"
 301  #include "ha0ha.h"
 302 -
 303 +#include "srv0srv.h"
 304  /** Flag: has the search system been enabled?
 305  Protected by btr_search_latch. */
 306  UNIV_INTERN char               btr_search_enabled      = TRUE;
 307 @@ -609,6 +609,11 @@
 308
 309         block = btr_cur_get_block(cursor);
 310
 311 +       if (srv_pass_corrupt_table && !block) {
 312 +               return;
 313 +       }
 314 +       ut_a(block);
 315 +
 316         /* NOTE that the following two function calls do NOT protect
 317         info or block->n_fields etc. with any semaphore, to save CPU time!
 318         We cannot assume the fields are consistent when we return from
 319 --- a/storage/innobase/buf/buf0buf.c
 320 +++ b/storage/innobase/buf/buf0buf.c
 321 @@ -52,6 +52,7 @@
 322  #include "log0recv.h"
 323  #include "page0zip.h"
 324  #include "trx0trx.h"
 325 +#include "srv0start.h"
 326
 327  /* prototypes for new functions added to ha_innodb.cc */
 328  trx_t* innobase_get_trx();
 329 @@ -1150,6 +1151,11 @@
 330                         ready = buf_flush_ready_for_replace(&block->page);
 331                         mutex_exit(&block->mutex);
 332
 333 +                       if (block->page.is_corrupt) {
 334 +                               /* corrupt page may remain, it can be skipped */
 335 +                               break;
 336 +                       }
 337 +
 338                         if (!ready) {
 339
 340                                 return(block);
 341 @@ -1946,6 +1952,13 @@
 342                 return(NULL);
 343         }
 344
 345 +       if (srv_pass_corrupt_table <= 1) {
 346 +               if (bpage->is_corrupt) {
 347 +                       rw_lock_s_unlock(&buf_pool->page_hash_latch);
 348 +                       return(NULL);
 349 +               }
 350 +       }
 351 +
 352         block_mutex = buf_page_get_mutex_enter(bpage);
 353
 354         rw_lock_s_unlock(&buf_pool->page_hash_latch);
 355 @@ -2525,6 +2538,13 @@
 356                 return(NULL);
 357         }
 358
 359 +       if (srv_pass_corrupt_table <= 1) {
 360 +               if (block->page.is_corrupt) {
 361 +                       mutex_exit(block_mutex);
 362 +                       return(NULL);
 363 +               }
 364 +       }
 365 +
 366         switch (buf_block_get_state(block)) {
 367                 buf_page_t*     bpage;
 368                 ibool           success;
 369 @@ -3199,6 +3219,7 @@
 370         bpage->newest_modification = 0;
 371         bpage->oldest_modification = 0;
 372         HASH_INVALIDATE(bpage, hash);
 373 +       bpage->is_corrupt = FALSE;
 374  #if defined UNIV_DEBUG_FILE_ACCESSES || defined UNIV_DEBUG
 375         bpage->file_page_was_freed = FALSE;
 376  #endif /* UNIV_DEBUG_FILE_ACCESSES || UNIV_DEBUG */
 377 @@ -3837,6 +3858,7 @@
 378                                 (ulong) bpage->offset);
 379                 }
 380
 381 +               if (!srv_pass_corrupt_table || !bpage->is_corrupt) {
 382                 /* From version 3.23.38 up we store the page checksum
 383                 to the 4 first bytes of the page end lsn field */
 384
 385 @@ -3878,6 +3900,23 @@
 386                               REFMAN "forcing-innodb-recovery.html\n"
 387                               "InnoDB: about forcing recovery.\n", stderr);
 388
 389 +                       if (srv_pass_corrupt_table && !trx_sys_sys_space(bpage->space)
 390 +                           && bpage->space < SRV_LOG_SPACE_FIRST_ID) {
 391 +                               trx_t*  trx;
 392 +
 393 +                               fprintf(stderr,
 394 +                                       "InnoDB: space %u will be treated as corrupt.\n",
 395 +                                       bpage->space);
 396 +                               fil_space_set_corrupt(bpage->space);
 397 +
 398 +                               trx = innobase_get_trx();
 399 +                               if (trx && trx->dict_operation_lock_mode == RW_X_LATCH) {
 400 +                                       dict_table_set_corrupt_by_space(bpage->space, FALSE);
 401 +                               } else {
 402 +                                       dict_table_set_corrupt_by_space(bpage->space, TRUE);
 403 +                               }
 404 +                               bpage->is_corrupt = TRUE;
 405 +                       } else
 406                         if (srv_force_recovery < SRV_FORCE_IGNORE_CORRUPT) {
 407                                 /* If page space id is larger than TRX_SYS_SPACE
 408                                 (0), we will attempt to mark the corresponding
 409 @@ -3894,6 +3933,7 @@
 410                                 }
 411                         }
 412                 }
 413 +               } /**/
 414
 415                 if (recv_recovery_is_on()) {
 416                         /* Pages must be uncompressed for crash recovery. */
 417 @@ -3903,8 +3943,11 @@
 418
 419                 if (uncompressed && !recv_no_ibuf_operations) {
 420                         ibuf_merge_or_delete_for_page(
 421 +                               /* Delete possible entries, if bpage is_corrupt */
 422 +                               (srv_pass_corrupt_table && bpage->is_corrupt) ? NULL :
 423                                 (buf_block_t*) bpage, bpage->space,
 424                                 bpage->offset, buf_page_get_zip_size(bpage),
 425 +                               (srv_pass_corrupt_table && bpage->is_corrupt) ? FALSE :
 426                                 TRUE);
 427                 }
 428         }
 429 --- a/storage/innobase/buf/buf0rea.c
 430 +++ b/storage/innobase/buf/buf0rea.c
 431 @@ -195,7 +195,14 @@
 432                               ((buf_block_t*) bpage)->frame, bpage, trx);
 433         }
 434         thd_wait_end(NULL);
 435 +
 436 +       if (srv_pass_corrupt_table) {
 437 +               if (*err != DB_SUCCESS) {
 438 +                       bpage->is_corrupt = TRUE;
 439 +               }
 440 +       } else {
 441         ut_a(*err == DB_SUCCESS);
 442 +       }
 443
 444         if (sync) {
 445                 /* The i/o is already completed when we arrive from
 446 --- a/storage/innobase/dict/dict0dict.c
 447 +++ b/storage/innobase/dict/dict0dict.c
 448 @@ -55,6 +55,7 @@
 449  #include "m_ctype.h" /* my_isspace() */
 450  #include "ha_prototypes.h" /* innobase_strcasecmp(), innobase_casedn_str()*/
 451  #include "row0upd.h"
 452 +#include "srv0start.h" /* SRV_LOG_SPACE_FIRST_ID */
 453
 454  #include <ctype.h>
 455
 456 @@ -750,7 +751,7 @@
 457
 458         mutex_exit(&(dict_sys->mutex));
 459
 460 -       if (table != NULL) {
 461 +       if (table != NULL && !table->is_corrupt) {
 462                 /* If table->ibd_file_missing == TRUE, this will
 463                 print an error message and return without doing
 464                 anything. */
 465 @@ -1298,7 +1299,7 @@
 466                     + dict_sys->size) > srv_dict_size_limit ) {
 467                 prev_table = UT_LIST_GET_PREV(table_LRU, table);
 468
 469 -               if (table == self || table->n_mysql_handles_opened)
 470 +               if (table == self || table->n_mysql_handles_opened || table->is_corrupt)
 471                         goto next_loop;
 472
 473                 cached_foreign_tables = 0;
 474 @@ -4377,6 +4378,12 @@
 475         heap = mem_heap_create(1000);
 476
 477         while (index) {
 478 +               if (table->is_corrupt) {
 479 +                       ut_a(srv_pass_corrupt_table);
 480 +                       mem_heap_free(heap);
 481 +                       return(FALSE);
 482 +               }
 483 +
 484                 size = btr_get_size(index, BTR_TOTAL_SIZE);
 485
 486                 index->stat_index_size = size;
 487 @@ -4524,6 +4531,12 @@
 488         heap = mem_heap_create(1000);
 489
 490         while (index) {
 491 +               if (table->is_corrupt) {
 492 +                       ut_a(srv_pass_corrupt_table);
 493 +                       mem_heap_free(heap);
 494 +                       return;
 495 +               }
 496 +
 497  /*===========================================*/
 498  {
 499         dict_table_t*   sys_stats;
 500 @@ -4716,6 +4729,13 @@
 501                      || (srv_force_recovery < SRV_FORCE_NO_LOG_REDO
 502                          && dict_index_is_clust(index)))) {
 503                         ulint   size;
 504 +
 505 +                       if (table->is_corrupt) {
 506 +                               ut_a(srv_pass_corrupt_table);
 507 +                               dict_table_stats_unlock(table, RW_X_LATCH);
 508 +                               return;
 509 +                       }
 510 +
 511                         size = btr_get_size(index, BTR_TOTAL_SIZE);
 512
 513                         index->stat_index_size = size;
 514 @@ -5695,4 +5715,42 @@
 515
 516         index->type |= DICT_CORRUPT;
 517  }
 518 +
 519 +/*************************************************************************
 520 +set is_corrupt flag by space_id*/
 521 +
 522 +void
 523 +dict_table_set_corrupt_by_space(
 524 +/*============================*/
 525 +       ulint   space_id,
 526 +       ibool   need_mutex)
 527 +{
 528 +       dict_table_t*   table;
 529 +       ibool           found = FALSE;
 530 +
 531 +       ut_a(!trx_sys_sys_space(space_id) && space_id < SRV_LOG_SPACE_FIRST_ID);
 532 +
 533 +       if (need_mutex)
 534 +               mutex_enter(&(dict_sys->mutex));
 535 +
 536 +       table = UT_LIST_GET_FIRST(dict_sys->table_LRU);
 537 +
 538 +       while (table) {
 539 +               if (table->space == space_id) {
 540 +                       table->is_corrupt = TRUE;
 541 +                       found = TRUE;
 542 +               }
 543 +
 544 +               table = UT_LIST_GET_NEXT(table_LRU, table);
 545 +       }
 546 +
 547 +       if (need_mutex)
 548 +               mutex_exit(&(dict_sys->mutex));
 549 +
 550 +       if (!found) {
 551 +               fprintf(stderr, "InnoDB: space to be marked as "
 552 +                       "crashed was not found for id %lu.\n",
 553 +                       (ulong) space_id);
 554 +       }
 555 +}
 556  #endif /* !UNIV_HOTBACKUP */
 557 --- a/storage/innobase/dict/dict0mem.c
 558 +++ b/storage/innobase/dict/dict0mem.c
 559 @@ -96,6 +96,8 @@
 560         /* The number of transactions that are either waiting on the
 561         AUTOINC lock or have been granted the lock. */
 562         table->n_waiting_or_granted_auto_inc_locks = 0;
 563 +
 564 +       table->is_corrupt = FALSE;
 565  #endif /* !UNIV_HOTBACKUP */
 566
 567         ut_d(table->magic_n = DICT_TABLE_MAGIC_N);
 568 --- a/storage/innobase/fil/fil0fil.c
 569 +++ b/storage/innobase/fil/fil0fil.c
 570 @@ -235,6 +235,7 @@
 571                                 file we have written to */
 572         ibool           is_in_unflushed_spaces; /*!< TRUE if this space is
 573                                 currently in unflushed_spaces */
 574 +       ibool           is_corrupt;
 575         UT_LIST_NODE_T(fil_space_t) space_list;
 576                                 /*!< list of all spaces */
 577         ulint           magic_n;/*!< FIL_SPACE_MAGIC_N */
 578 @@ -1294,6 +1295,8 @@
 579                     ut_fold_string(name), space);
 580         space->is_in_unflushed_spaces = FALSE;
 581
 582 +       space->is_corrupt = FALSE;
 583 +
 584         UT_LIST_ADD_LAST(space_list, fil_system->space_list, space);
 585
 586         mutex_exit(&fil_system->mutex);
 587 @@ -5285,6 +5288,34 @@
 588         ut_a(byte_offset % OS_FILE_LOG_BLOCK_SIZE == 0);
 589         ut_a((len % OS_FILE_LOG_BLOCK_SIZE) == 0);
 590
 591 +       if (srv_pass_corrupt_table == 1 && space->is_corrupt) {
 592 +               /* should ignore i/o for the crashed space */
 593 +               mutex_enter(&fil_system->mutex);
 594 +               fil_node_complete_io(node, fil_system, type);
 595 +               mutex_exit(&fil_system->mutex);
 596 +               if (mode == OS_AIO_NORMAL) {
 597 +                       ut_a(space->purpose == FIL_TABLESPACE);
 598 +                       buf_page_io_complete(message);
 599 +               }
 600 +               if (type == OS_FILE_READ) {
 601 +                       return(DB_TABLESPACE_DELETED);
 602 +               } else {
 603 +                       return(DB_SUCCESS);
 604 +               }
 605 +       } else {
 606 +               if (srv_pass_corrupt_table > 1 && space->is_corrupt) {
 607 +                       /* should ignore write i/o for the crashed space */
 608 +                       if (type == OS_FILE_WRITE) {
 609 +                               mutex_enter(&fil_system->mutex);
 610 +                               fil_node_complete_io(node, fil_system, type);
 611 +                               mutex_exit(&fil_system->mutex);
 612 +                               if (mode == OS_AIO_NORMAL) {
 613 +                                       ut_a(space->purpose == FIL_TABLESPACE);
 614 +                                       buf_page_io_complete(message);
 615 +                               }
 616 +                               return(DB_SUCCESS);
 617 +                       }
 618 +               }
 619  #ifdef UNIV_HOTBACKUP
 620         /* In ibbackup do normal i/o, not aio */
 621         if (type == OS_FILE_READ) {
 622 @@ -5299,6 +5330,8 @@
 623         ret = os_aio(type, mode | wake_later, node->name, node->handle, buf,
 624                      offset_low, offset_high, len, node, message, trx);
 625  #endif
 626 +       } /**/
 627 +
 628         ut_a(ret);
 629
 630         if (mode == OS_AIO_SYNC) {
 631 @@ -5799,3 +5832,46 @@
 632                 return 0;
 633         }
 634  }
 635 +
 636 +/*************************************************************************
 637 +functions to access is_corrupt flag of fil_space_t*/
 638 +
 639 +ibool
 640 +fil_space_is_corrupt(
 641 +/*=================*/
 642 +       ulint   space_id)
 643 +{
 644 +       fil_space_t*    space;
 645 +       ibool           ret = FALSE;
 646 +
 647 +       mutex_enter(&fil_system->mutex);
 648 +
 649 +       space = fil_space_get_by_id(space_id);
 650 +
 651 +       if (space && space->is_corrupt) {
 652 +               ret = TRUE;
 653 +       }
 654 +
 655 +       mutex_exit(&fil_system->mutex);
 656 +
 657 +       return(ret);
 658 +}
 659 +
 660 +void
 661 +fil_space_set_corrupt(
 662 +/*==================*/
 663 +       ulint   space_id)
 664 +{
 665 +       fil_space_t*    space;
 666 +
 667 +       mutex_enter(&fil_system->mutex);
 668 +
 669 +       space = fil_space_get_by_id(space_id);
 670 +
 671 +       if (space) {
 672 +               space->is_corrupt = TRUE;
 673 +       }
 674 +
 675 +       mutex_exit(&fil_system->mutex);
 676 +}
 677 +
 678 --- a/storage/innobase/fsp/fsp0fsp.c
 679 +++ b/storage/innobase/fsp/fsp0fsp.c
 680 @@ -308,6 +308,12 @@
 681         ut_ad(id || !zip_size);
 682
 683         block = buf_page_get(id, zip_size, 0, RW_X_LATCH, mtr);
 684 +
 685 +       if (srv_pass_corrupt_table && !block) {
 686 +               return(0);
 687 +       }
 688 +       ut_a(block);
 689 +
 690         header = FSP_HEADER_OFFSET + buf_block_get_frame(block);
 691         buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 692
 693 @@ -726,6 +732,12 @@
 694         fsp_header_t*   sp_header;
 695
 696         block = buf_page_get(space, zip_size, 0, RW_X_LATCH, mtr);
 697 +
 698 +       if (srv_pass_corrupt_table && !block) {
 699 +               return(0);
 700 +       }
 701 +       ut_a(block);
 702 +
 703         buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 704
 705         sp_header = FSP_HEADER_OFFSET + buf_block_get_frame(block);
 706 @@ -1805,6 +1817,11 @@
 707  {
 708         fseg_inode_t*   inode;
 709
 710 +       if (srv_pass_corrupt_table && !page) {
 711 +               return(ULINT_UNDEFINED);
 712 +       }
 713 +       ut_a(page);
 714 +
 715         for (; i < FSP_SEG_INODES_PER_PAGE(zip_size); i++) {
 716
 717                 inode = fsp_seg_inode_page_get_nth_inode(
 718 @@ -1918,6 +1935,11 @@
 719
 720         page = buf_block_get_frame(block);
 721
 722 +       if (srv_pass_corrupt_table && !page) {
 723 +               return(0);
 724 +       }
 725 +       ut_a(page);
 726 +
 727         n = fsp_seg_inode_page_find_free(page, 0, zip_size, mtr);
 728
 729         ut_a(n != ULINT_UNDEFINED);
 730 @@ -2011,6 +2033,11 @@
 731
 732         inode = fut_get_ptr(space, zip_size, inode_addr, RW_X_LATCH, mtr);
 733
 734 +       if (srv_pass_corrupt_table && !inode) {
 735 +               return(0);
 736 +       }
 737 +       ut_a(inode);
 738 +
 739         if (UNIV_UNLIKELY(!mach_read_from_8(inode + FSEG_ID))) {
 740
 741                 inode = NULL;
 742 @@ -2037,7 +2064,7 @@
 743  {
 744         fseg_inode_t*   inode
 745                 = fseg_inode_try_get(header, space, zip_size, mtr);
 746 -       ut_a(inode);
 747 +       ut_a(srv_pass_corrupt_table || inode);
 748         return(inode);
 749  }
 750
 751 @@ -3243,6 +3270,11 @@
 752
 753         descr = xdes_get_descriptor(space, zip_size, page, mtr);
 754
 755 +       if (srv_pass_corrupt_table && !descr) {
 756 +               /* The page may be corrupt. pass it. */
 757 +               return;
 758 +       }
 759 +
 760         ut_a(descr);
 761         if (xdes_get_bit(descr, XDES_FREE_BIT, page % FSP_EXTENT_SIZE, mtr)) {
 762                 fputs("InnoDB: Dump of the tablespace extent descriptor: ",
 763 @@ -3490,6 +3522,11 @@
 764
 765         descr = xdes_get_descriptor(space, zip_size, header_page, mtr);
 766
 767 +       if (srv_pass_corrupt_table && !descr) {
 768 +               /* The page may be corrupt. pass it. */
 769 +               return(TRUE);
 770 +       }
 771 +
 772         /* Check that the header resides on a page which has not been
 773         freed yet */
 774
 775 @@ -3574,6 +3611,12 @@
 776
 777         inode = fseg_inode_get(header, space, zip_size, mtr);
 778
 779 +       if (srv_pass_corrupt_table && !inode) {
 780 +               /* ignore the corruption */
 781 +               return(TRUE);
 782 +       }
 783 +       ut_a(inode);
 784 +
 785         descr = fseg_get_first_extent(inode, space, zip_size, mtr);
 786
 787         if (descr != NULL) {
 788 --- a/storage/innobase/handler/ha_innodb.cc
 789 +++ b/storage/innobase/handler/ha_innodb.cc
 790 @@ -4024,6 +4024,12 @@
 791                 DBUG_RETURN(1);
 792         }
 793
 794 +       if (srv_pass_corrupt_table <= 1 && share->ib_table && share->ib_table->is_corrupt) {
 795 +               free_share(share);
 796 +
 797 +               DBUG_RETURN(HA_ERR_CRASHED_ON_USAGE);
 798 +       }
 799 +
 800         /* Will be allocated if it is needed in ::update_row() */
 801         upd_buf = NULL;
 802         upd_buf_size = 0;
 803 @@ -4043,6 +4049,17 @@
 804         /* Get pointer to a table object in InnoDB dictionary cache */
 805         ib_table = dict_table_get(norm_name, TRUE);
 806
 807 +       if (srv_pass_corrupt_table <= 1 && ib_table && ib_table->is_corrupt) {
 808 +               free_share(share);
 809 +               my_free(upd_buf);
 810 +               upd_buf = NULL;
 811 +               upd_buf_size = 0;
 812 +
 813 +               DBUG_RETURN(HA_ERR_CRASHED_ON_USAGE);
 814 +       }
 815 +
 816 +       share->ib_table = ib_table;
 817 +
 818         if (NULL == ib_table) {
 819                 if (is_part && retries < 10) {
 820                         /* MySQL partition engine hard codes the file name
 821 @@ -5263,6 +5280,10 @@
 822
 823         ha_statistic_increment(&SSV::ha_write_count);
 824
 825 +       if (share->ib_table->is_corrupt) {
 826 +               DBUG_RETURN(HA_ERR_CRASHED);
 827 +       }
 828 +
 829         if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
 830                 table->timestamp_field->set_time();
 831
 832 @@ -5479,6 +5500,10 @@
 833  func_exit:
 834         innobase_active_small();
 835
 836 +       if (share->ib_table->is_corrupt) {
 837 +               DBUG_RETURN(HA_ERR_CRASHED);
 838 +       }
 839 +
 840         DBUG_RETURN(error_result);
 841  }
 842
 843 @@ -5673,6 +5698,10 @@
 844
 845         ha_statistic_increment(&SSV::ha_update_count);
 846
 847 +       if (share->ib_table->is_corrupt) {
 848 +               DBUG_RETURN(HA_ERR_CRASHED);
 849 +       }
 850 +
 851         if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
 852                 table->timestamp_field->set_time();
 853
 854 @@ -5760,6 +5789,10 @@
 855
 856         innobase_active_small();
 857
 858 +       if (share->ib_table->is_corrupt) {
 859 +               DBUG_RETURN(HA_ERR_CRASHED);
 860 +       }
 861 +
 862         DBUG_RETURN(error);
 863  }
 864
 865 @@ -5781,6 +5814,10 @@
 866
 867         ha_statistic_increment(&SSV::ha_delete_count);
 868
 869 +       if (share->ib_table->is_corrupt) {
 870 +               DBUG_RETURN(HA_ERR_CRASHED);
 871 +       }
 872 +
 873         if (!prebuilt->upd_node) {
 874                 row_get_prebuilt_update_vector(prebuilt);
 875         }
 876 @@ -5807,6 +5844,10 @@
 877
 878         innobase_active_small();
 879
 880 +       if (share->ib_table->is_corrupt) {
 881 +               DBUG_RETURN(HA_ERR_CRASHED);
 882 +       }
 883 +
 884         DBUG_RETURN(error);
 885  }
 886
 887 @@ -6046,6 +6087,10 @@
 888
 889         ha_statistic_increment(&SSV::ha_read_key_count);
 890
 891 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 892 +               DBUG_RETURN(HA_ERR_CRASHED);
 893 +       }
 894 +
 895         index = prebuilt->index;
 896
 897         if (UNIV_UNLIKELY(index == NULL) || dict_index_is_corrupted(index)) {
 898 @@ -6113,6 +6158,10 @@
 899                 ret = DB_UNSUPPORTED;
 900         }
 901
 902 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 903 +               DBUG_RETURN(HA_ERR_CRASHED);
 904 +       }
 905 +
 906         switch (ret) {
 907         case DB_SUCCESS:
 908                 error = 0;
 909 @@ -6227,6 +6276,10 @@
 910  {
 911         DBUG_ENTER("change_active_index");
 912
 913 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 914 +               DBUG_RETURN(HA_ERR_CRASHED);
 915 +       }
 916 +
 917         ut_ad(user_thd == ha_thd());
 918         ut_a(prebuilt->trx == thd_to_trx(user_thd));
 919
 920 @@ -6340,6 +6393,10 @@
 921
 922         DBUG_ENTER("general_fetch");
 923
 924 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 925 +               DBUG_RETURN(HA_ERR_CRASHED);
 926 +       }
 927 +
 928         ut_a(prebuilt->trx == thd_to_trx(user_thd));
 929
 930         innodb_srv_conc_enter_innodb(prebuilt->trx);
 931 @@ -6349,6 +6406,10 @@
 932
 933         innodb_srv_conc_exit_innodb(prebuilt->trx);
 934
 935 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 936 +               DBUG_RETURN(HA_ERR_CRASHED);
 937 +       }
 938 +
 939         switch (ret) {
 940         case DB_SUCCESS:
 941                 error = 0;
 942 @@ -7615,10 +7676,18 @@
 943
 944         update_thd(ha_thd());
 945
 946 +       if (share->ib_table->is_corrupt) {
 947 +               DBUG_RETURN(HA_ERR_CRASHED);
 948 +       }
 949 +
 950         /* Truncate the table in InnoDB */
 951
 952         error = row_truncate_table_for_mysql(prebuilt->table, prebuilt->trx);
 953
 954 +       if (share->ib_table->is_corrupt) {
 955 +               DBUG_RETURN(HA_ERR_CRASHED);
 956 +       }
 957 +
 958         error = convert_error_code_to_mysql(error, prebuilt->table->flags,
 959                                             NULL);
 960
 961 @@ -8124,6 +8193,16 @@
 962         return(ranges + (double) rows / (double) total_rows * time_for_scan);
 963  }
 964
 965 +UNIV_INTERN
 966 +bool
 967 +ha_innobase::is_corrupt() const
 968 +{
 969 +       if (share->ib_table)
 970 +               return ((bool)share->ib_table->is_corrupt);
 971 +       else
 972 +               return (FALSE);
 973 +}
 974 +
 975  /*********************************************************************//**
 976  Calculates the key number used inside MySQL for an Innobase index. We will
 977  first check the "index translation table" for a match of the index to get
 978 @@ -8301,7 +8380,7 @@
 979         ib_table = prebuilt->table;
 980
 981         if (flag & HA_STATUS_TIME) {
 982 -               if (called_from_analyze || innobase_stats_on_metadata) {
 983 +               if ((called_from_analyze || innobase_stats_on_metadata) && !share->ib_table->is_corrupt) {
 984                         /* In sql_show we call with this flag: update
 985                         then statistics so that they are up-to-date */
 986
 987 @@ -8601,10 +8680,18 @@
 988         THD*            thd,            /*!< in: connection thread handle */
 989         HA_CHECK_OPT*   check_opt)      /*!< in: currently ignored */
 990  {
 991 +       if (share->ib_table->is_corrupt) {
 992 +               return(HA_ADMIN_CORRUPT);
 993 +       }
 994 +
 995         /* Simply call ::info() with all the flags */
 996         info_low(HA_STATUS_TIME | HA_STATUS_CONST | HA_STATUS_VARIABLE,
 997                  true /* called from analyze */);
 998
 999 +       if (share->ib_table->is_corrupt) {
1000 +               return(HA_ADMIN_CORRUPT);
1001 +       }
1002 +
1003         return(0);
1004  }
1005
1006 @@ -8840,6 +8927,10 @@
1007                 my_error(ER_QUERY_INTERRUPTED, MYF(0));
1008         }
1009
1010 +       if (share->ib_table->is_corrupt) {
1011 +               return(HA_ADMIN_CORRUPT);
1012 +       }
1013 +
1014         DBUG_RETURN(is_ok ? HA_ADMIN_OK : HA_ADMIN_CORRUPT);
1015  }
1016
1017 @@ -9610,6 +9701,10 @@
1018
1019         update_thd(thd);
1020
1021 +       if (share->ib_table->is_corrupt) {
1022 +               DBUG_RETURN(HA_ERR_CRASHED);
1023 +       }
1024 +
1025         if (prebuilt->table->ibd_file_missing && !thd_tablespace_op(thd)) {
1026                 ut_print_timestamp(stderr);
1027                 fprintf(stderr,
1028 @@ -12073,6 +12168,26 @@
1029    "dump file (if present). Disabled by default.",
1030    NULL, NULL, FALSE);
1031
1032 +const char *corrupt_table_action_names[]=
1033 +{
1034 +  "assert", /* 0 */
1035 +  "warn", /* 1 */
1036 +  "salvage", /* 2 */
1037 +  NullS
1038 +};
1039 +TYPELIB corrupt_table_action_typelib=
1040 +{
1041 +  array_elements(corrupt_table_action_names) - 1, "corrupt_table_action_typelib",
1042 +  corrupt_table_action_names, NULL
1043 +};
1044 +static MYSQL_SYSVAR_ENUM(corrupt_table_action, srv_pass_corrupt_table,
1045 +  PLUGIN_VAR_RQCMDARG,
1046 +  "Warn corruptions of user tables as 'corrupt table' instead of not crashing itself, "
1047 +  "when used with file_per_table. "
1048 +  "All file io for the datafile after detected as corrupt are disabled, "
1049 +  "except for the deletion.",
1050 +  NULL, NULL, 0, &corrupt_table_action_typelib);
1051 +
1052  static struct st_mysql_sys_var* innobase_system_variables[]= {
1053    MYSQL_SYSVAR(additional_mem_pool_size),
1054    MYSQL_SYSVAR(autoextend_increment),
1055 @@ -12166,6 +12281,7 @@
1056  #ifdef UNIV_DEBUG
1057    MYSQL_SYSVAR(trx_rseg_n_slots_debug),
1058  #endif /* UNIV_DEBUG */
1059 +  MYSQL_SYSVAR(corrupt_table_action),
1060    NULL
1061  };
1062
1063 --- a/storage/innobase/handler/ha_innodb.h
1064 +++ b/storage/innobase/handler/ha_innodb.h
1065 @@ -52,6 +52,7 @@
1066         innodb_idx_translate_t  idx_trans_tbl;  /*!< index translation
1067                                                 table between MySQL and
1068                                                 Innodb */
1069 +       dict_table_t*           ib_table;
1070  } INNOBASE_SHARE;
1071
1072
1073 @@ -136,6 +137,7 @@
1074         int close(void);
1075         double scan_time();
1076         double read_time(uint index, uint ranges, ha_rows rows);
1077 +       bool is_corrupt() const;
1078
1079         int write_row(uchar * buf);
1080         int update_row(const uchar * old_data, uchar * new_data);
1081 --- a/storage/innobase/include/btr0btr.ic
1082 +++ b/storage/innobase/include/btr0btr.ic
1083 @@ -28,7 +28,7 @@
1084  #include "mtr0mtr.h"
1085  #include "mtr0log.h"
1086  #include "page0zip.h"
1087 -
1088 +#include "srv0srv.h"
1089  #define BTR_MAX_NODE_LEVEL     50      /*!< Maximum B-tree page level
1090                                         (not really a hard limit).
1091                                         Used in debug assertions
1092 @@ -59,7 +59,9 @@
1093         block = buf_page_get_gen(space, zip_size, page_no, mode,
1094                                  NULL, BUF_GET, file, line, mtr);
1095
1096 -       if (mode != RW_NO_LATCH) {
1097 +       ut_a(srv_pass_corrupt_table || block);
1098 +
1099 +       if (block && mode != RW_NO_LATCH) {
1100
1101                 buf_block_dbg_add_level(
1102                         block, index != NULL && dict_index_is_ibuf(index)
1103 --- a/storage/innobase/include/buf0buf.h
1104 +++ b/storage/innobase/include/buf0buf.h
1105 @@ -1023,7 +1023,7 @@
1106         const buf_block_t*      block)  /*!< in: pointer to the control block */
1107         __attribute__((pure));
1108  #else /* UNIV_DEBUG */
1109 -# define buf_block_get_frame(block) (block)->frame
1110 +# define buf_block_get_frame(block) (block ? (block)->frame : 0)
1111  #endif /* UNIV_DEBUG */
1112  /*********************************************************************//**
1113  Gets the space id of a block.
1114 @@ -1470,6 +1470,7 @@
1115                                         0 if the block was never accessed
1116                                         in the buffer pool */
1117         /* @} */
1118 +       ibool           is_corrupt;
1119  # if defined UNIV_DEBUG_FILE_ACCESSES || defined UNIV_DEBUG
1120         ibool           file_page_was_freed;
1121                                         /*!< this is set to TRUE when fsp
1122 --- a/storage/innobase/include/buf0buf.ic
1123 +++ b/storage/innobase/include/buf0buf.ic
1124 @@ -34,7 +34,7 @@
1125  #include "buf0flu.h"
1126  #include "buf0lru.h"
1127  #include "buf0rea.h"
1128 -
1129 +#include "srv0srv.h"
1130  /*********************************************************************//**
1131  Gets the current size of buffer buf_pool in bytes.
1132  @return size in bytes */
1133 @@ -681,6 +681,12 @@
1134  /*================*/
1135         const buf_block_t*      block)  /*!< in: pointer to the control block */
1136  {
1137 +       ut_a(srv_pass_corrupt_table || block);
1138 +
1139 +       if (srv_pass_corrupt_table && !block) {
1140 +               return(0);
1141 +       }
1142 +
1143         ut_ad(block);
1144
1145         switch (buf_block_get_state(block)) {
1146 --- a/storage/innobase/include/dict0dict.h
1147 +++ b/storage/innobase/include/dict0dict.h
1148 @@ -1326,6 +1326,15 @@
1149  /*========================*/
1150         ulint           space_id);      /*!< in: space ID */
1151
1152 +/*************************************************************************
1153 +set is_corrupt flag by space_id*/
1154 +
1155 +void
1156 +dict_table_set_corrupt_by_space(
1157 +/*============================*/
1158 +       ulint   space_id,
1159 +       ibool   need_mutex);
1160 +
1161  #ifndef UNIV_NONINL
1162  #include "dict0dict.ic"
1163  #endif
1164 --- a/storage/innobase/include/dict0mem.h
1165 +++ b/storage/innobase/include/dict0mem.h
1166 @@ -670,6 +670,7 @@
1167                                 the AUTOINC lock on this table. */
1168                                 /* @} */
1169         /*----------------------*/
1170 +       ibool           is_corrupt;
1171  #endif /* !UNIV_HOTBACKUP */
1172
1173  #ifdef UNIV_DEBUG
1174 --- a/storage/innobase/include/fil0fil.h
1175 +++ b/storage/innobase/include/fil0fil.h
1176 @@ -759,6 +759,19 @@
1177  fil_system_hash_nodes(void);
1178  /*========================*/
1179
1180 +/*************************************************************************
1181 +functions to access is_corrupt flag of fil_space_t*/
1182 +
1183 +ibool
1184 +fil_space_is_corrupt(
1185 +/*=================*/
1186 +       ulint   space_id);
1187 +
1188 +void
1189 +fil_space_set_corrupt(
1190 +/*==================*/
1191 +       ulint   space_id);
1192 +
1193  typedef        struct fil_space_struct fil_space_t;
1194
1195  #endif
1196 --- a/storage/innobase/include/fut0fut.ic
1197 +++ b/storage/innobase/include/fut0fut.ic
1198 @@ -23,6 +23,7 @@
1199  Created 12/13/1995 Heikki Tuuri
1200  ***********************************************************************/
1201
1202 +#include "srv0srv.h"
1203  #include "sync0rw.h"
1204  #include "buf0buf.h"
1205
1206 @@ -48,6 +49,12 @@
1207         ut_ad((rw_latch == RW_S_LATCH) || (rw_latch == RW_X_LATCH));
1208
1209         block = buf_page_get(space, zip_size, addr.page, rw_latch, mtr);
1210 +
1211 +       if (srv_pass_corrupt_table && !block) {
1212 +               return(0);
1213 +       }
1214 +       ut_a(block);
1215 +
1216         ptr = buf_block_get_frame(block) + addr.boffset;
1217
1218         buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
1219 --- a/storage/innobase/include/page0page.h
1220 +++ b/storage/innobase/include/page0page.h
1221 @@ -497,7 +497,7 @@
1222  page_is_leaf(
1223  /*=========*/
1224         const page_t*   page)   /*!< in: page */
1225 -       __attribute__((nonnull, pure));
1226 +       __attribute__((pure));
1227  /************************************************************//**
1228  Gets the pointer to the next record on the page.
1229  @return        pointer to next record */
1230 --- a/storage/innobase/include/page0page.ic
1231 +++ b/storage/innobase/include/page0page.ic
1232 @@ -274,6 +274,9 @@
1233  /*=========*/
1234         const page_t*   page)   /*!< in: page */
1235  {
1236 +       if (!page) {
1237 +               return(FALSE);
1238 +       }
1239         return(!*(const uint16*) (page + (PAGE_HEADER + PAGE_LEVEL)));
1240  }
1241
1242 --- a/storage/innobase/include/page0zip.h
1243 +++ b/storage/innobase/include/page0zip.h
1244 @@ -114,7 +114,7 @@
1245         const page_t*   page,   /*!< in: uncompressed page */
1246         dict_index_t*   index,  /*!< in: index of the B-tree node */
1247         mtr_t*          mtr)    /*!< in: mini-transaction, or NULL */
1248 -       __attribute__((nonnull(1,2,3)));
1249 +       __attribute__((nonnull(1,3)));
1250
1251  /**********************************************************************//**
1252  Decompress a page.  This function should tolerate errors on the compressed
1253 --- a/storage/innobase/include/srv0srv.h
1254 +++ b/storage/innobase/include/srv0srv.h
1255 @@ -245,6 +245,7 @@
1256  extern ulint   srv_adaptive_flushing_method;
1257
1258  extern ulint   srv_expand_import;
1259 +extern ulint   srv_pass_corrupt_table;
1260
1261  extern ulint   srv_dict_size_limit;
1262  /*-------------------------------------------*/
1263 --- a/storage/innobase/page/page0zip.c
1264 +++ b/storage/innobase/page/page0zip.c
1265 @@ -1195,6 +1195,10 @@
1266         FILE*           logfile = NULL;
1267  #endif
1268
1269 +       if (!page) {
1270 +               return(FALSE);
1271 +       }
1272 +
1273         ut_a(page_is_comp(page));
1274         ut_a(fil_page_get_type(page) == FIL_PAGE_INDEX);
1275         ut_ad(page_simple_validate_new((page_t*) page));
1276 --- a/storage/innobase/row/row0ins.c
1277 +++ b/storage/innobase/row/row0ins.c
1278 @@ -1341,6 +1341,12 @@
1279                 const rec_t*            rec = btr_pcur_get_rec(&pcur);
1280                 const buf_block_t*      block = btr_pcur_get_block(&pcur);
1281
1282 +               if (srv_pass_corrupt_table && !block) {
1283 +                       err = DB_CORRUPTION;
1284 +                       break;
1285 +               }
1286 +               ut_a(block);
1287 +
1288                 if (page_rec_is_infimum(rec)) {
1289
1290                         continue;
1291 --- a/storage/innobase/row/row0merge.c
1292 +++ b/storage/innobase/row/row0merge.c
1293 @@ -1245,6 +1245,13 @@
1294
1295                 if (UNIV_LIKELY(has_next)) {
1296                         rec = btr_pcur_get_rec(&pcur);
1297 +
1298 +                       if (srv_pass_corrupt_table && !rec) {
1299 +                               err = DB_CORRUPTION;
1300 +                               goto err_exit;
1301 +                       }
1302 +                       ut_a(rec);
1303 +
1304                         offsets = rec_get_offsets(rec, clust_index, NULL,
1305                                                   ULINT_UNDEFINED, &row_heap);
1306
1307 --- a/storage/innobase/row/row0sel.c
1308 +++ b/storage/innobase/row/row0sel.c
1309 @@ -3919,6 +3919,13 @@
1310         /* PHASE 4: Look for matching records in a loop */
1311
1312         rec = btr_pcur_get_rec(pcur);
1313 +
1314 +       if (srv_pass_corrupt_table && !rec) {
1315 +               err = DB_CORRUPTION;
1316 +               goto lock_wait_or_error;
1317 +       }
1318 +       ut_a(rec);
1319 +
1320         ut_ad(!!page_rec_is_comp(rec) == comp);
1321  #ifdef UNIV_SEARCH_DEBUG
1322         /*
1323 @@ -3996,7 +4003,13 @@
1324         if (UNIV_UNLIKELY(next_offs >= UNIV_PAGE_SIZE - PAGE_DIR)) {
1325
1326  wrong_offs:
1327 -               if (srv_force_recovery == 0 || moves_up == FALSE) {
1328 +               if (srv_pass_corrupt_table && !trx_sys_sys_space(index->table->space)) {
1329 +                       index->table->is_corrupt = TRUE;
1330 +                       fil_space_set_corrupt(index->table->space);
1331 +               }
1332 +
1333 +               if ((srv_force_recovery == 0 || moves_up == FALSE)
1334 +                   && srv_pass_corrupt_table <= 1) {
1335                         ut_print_timestamp(stderr);
1336                         buf_page_print(page_align(rec), 0);
1337                         fprintf(stderr,
1338 @@ -4047,7 +4060,8 @@
1339
1340         offsets = rec_get_offsets(rec, index, offsets, ULINT_UNDEFINED, &heap);
1341
1342 -       if (UNIV_UNLIKELY(srv_force_recovery > 0)) {
1343 +       if (UNIV_UNLIKELY(srv_force_recovery > 0)
1344 +           || (srv_pass_corrupt_table == 2 && index->table->is_corrupt)) {
1345                 if (!rec_validate(rec, offsets)
1346                     || !btr_index_rec_validate(rec, index, FALSE)) {
1347                         fprintf(stderr,
1348 --- a/storage/innobase/srv/srv0srv.c
1349 +++ b/storage/innobase/srv/srv0srv.c
1350 @@ -435,6 +435,7 @@
1351  UNIV_INTERN ulint      srv_adaptive_flushing_method = 0; /* 0: native  1: estimate  2: keep_average */
1352
1353  UNIV_INTERN ulint      srv_expand_import = 0; /* 0:disable 1:enable */
1354 +UNIV_INTERN ulint      srv_pass_corrupt_table = 0; /* 0:disable 1:enable */
1355
1356  UNIV_INTERN ulint      srv_dict_size_limit = 0;
1357  /*-------------------------------------------*/
1358 --- a/storage/innobase/srv/srv0start.c
1359 +++ b/storage/innobase/srv/srv0start.c
1360 @@ -2175,6 +2175,13 @@
1361
1362         os_fast_mutex_free(&srv_os_test_mutex);
1363
1364 +       if (!srv_file_per_table_original_value
1365 +           && srv_pass_corrupt_table) {
1366 +               fprintf(stderr, "InnoDB: Warning:"
1367 +                       " The option innodb_file_per_table is disabled,"
1368 +                       " so using the option innodb_pass_corrupt_table doesn't make sense.\n");
1369 +       }
1370 +
1371         if (srv_print_verbose_log) {
1372                 ut_print_timestamp(stderr);
1373                 fprintf(stderr,