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 @@ -692,6 +692,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         ut_a((ibool)!!page_is_comp(buf_block_get_frame(block))
  21              == dict_table_is_comp(index->table));
  22  #ifdef UNIV_BTR_DEBUG
  23 @@ -978,6 +984,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 @@ -1437,6 +1449,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 @@ -1460,6 +1479,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 @@ -1493,6 +1518,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 @@ -590,6 +620,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 @@ -664,6 +707,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 @@ -857,6 +910,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 @@ -977,6 +1041,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 @@ -1190,6 +1262,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 @@ -2922,6 +3000,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 @@ -3630,6 +3713,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 @@ -102,6 +102,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 @@ -397,6 +403,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 @@ -1151,6 +1152,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 @@ -1945,6 +1951,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 @@ -2524,6 +2537,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 @@ -3198,6 +3218,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 @@ -3836,6 +3857,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 @@ -3877,6 +3899,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 @@ -3893,6 +3932,7 @@
 410                                 }
 411                         }
 412                 }
 413 +               } /**/
 414
 415                 if (recv_recovery_is_on()) {
 416                         /* Pages must be uncompressed for crash recovery. */
 417 @@ -3902,8 +3942,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 @@ -1293,7 +1294,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 @@ -4367,6 +4368,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 @@ -4514,6 +4521,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 @@ -4706,6 +4719,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 @@ -5685,4 +5705,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 @@ -5277,6 +5280,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 @@ -5291,6 +5322,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 @@ -5791,3 +5824,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 @@ -369,6 +369,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 @@ -787,6 +793,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 @@ -1866,6 +1878,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 @@ -1979,6 +1996,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 @@ -2072,6 +2094,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 @@ -2098,7 +2125,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 @@ -3304,6 +3331,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 @@ -3551,6 +3583,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 @@ -3635,6 +3672,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 @@ -4011,6 +4011,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         /* Create buffers for packing the fields of a record. Why
 801         table->reclength did not work here? Obviously, because char
 802         fields when packed actually became 1 byte longer, when we also
 803 @@ -4038,6 +4044,19 @@
 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_buff);
 810 +
 811 +               DBUG_RETURN(HA_ERR_CRASHED_ON_USAGE);
 812 +       }
 813 +
 814 +       share->ib_table = ib_table;
 815 +
 816 +
 817 +
 818 +
 819 +
 820         if (NULL == ib_table) {
 821                 if (is_part && retries < 10) {
 822                         ++retries;
 823 @@ -5187,6 +5206,10 @@
 824
 825         ha_statistic_increment(&SSV::ha_write_count);
 826
 827 +       if (share->ib_table->is_corrupt) {
 828 +               DBUG_RETURN(HA_ERR_CRASHED);
 829 +       }
 830 +
 831         if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
 832                 table->timestamp_field->set_time();
 833
 834 @@ -5403,6 +5426,10 @@
 835  func_exit:
 836         innobase_active_small();
 837
 838 +       if (share->ib_table->is_corrupt) {
 839 +               DBUG_RETURN(HA_ERR_CRASHED);
 840 +       }
 841 +
 842         DBUG_RETURN(error_result);
 843  }
 844
 845 @@ -5580,6 +5607,10 @@
 846
 847         ha_statistic_increment(&SSV::ha_update_count);
 848
 849 +       if (share->ib_table->is_corrupt) {
 850 +               DBUG_RETURN(HA_ERR_CRASHED);
 851 +       }
 852 +
 853         if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
 854                 table->timestamp_field->set_time();
 855
 856 @@ -5668,6 +5699,10 @@
 857
 858         innobase_active_small();
 859
 860 +       if (share->ib_table->is_corrupt) {
 861 +               DBUG_RETURN(HA_ERR_CRASHED);
 862 +       }
 863 +
 864         DBUG_RETURN(error);
 865  }
 866
 867 @@ -5689,6 +5724,10 @@
 868
 869         ha_statistic_increment(&SSV::ha_delete_count);
 870
 871 +       if (share->ib_table->is_corrupt) {
 872 +               DBUG_RETURN(HA_ERR_CRASHED);
 873 +       }
 874 +
 875         if (!prebuilt->upd_node) {
 876                 row_get_prebuilt_update_vector(prebuilt);
 877         }
 878 @@ -5715,6 +5754,10 @@
 879
 880         innobase_active_small();
 881
 882 +       if (share->ib_table->is_corrupt) {
 883 +               DBUG_RETURN(HA_ERR_CRASHED);
 884 +       }
 885 +
 886         DBUG_RETURN(error);
 887  }
 888
 889 @@ -5954,6 +5997,10 @@
 890
 891         ha_statistic_increment(&SSV::ha_read_key_count);
 892
 893 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 894 +               DBUG_RETURN(HA_ERR_CRASHED);
 895 +       }
 896 +
 897         index = prebuilt->index;
 898
 899         if (UNIV_UNLIKELY(index == NULL) || dict_index_is_corrupted(index)) {
 900 @@ -6022,6 +6069,10 @@
 901                 ret = DB_UNSUPPORTED;
 902         }
 903
 904 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 905 +               DBUG_RETURN(HA_ERR_CRASHED);
 906 +       }
 907 +
 908         switch (ret) {
 909         case DB_SUCCESS:
 910                 error = 0;
 911 @@ -6137,6 +6188,10 @@
 912  {
 913         DBUG_ENTER("change_active_index");
 914
 915 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 916 +               DBUG_RETURN(HA_ERR_CRASHED);
 917 +       }
 918 +
 919         ut_ad(user_thd == ha_thd());
 920         ut_a(prebuilt->trx == thd_to_trx(user_thd));
 921
 922 @@ -6250,6 +6305,10 @@
 923
 924         DBUG_ENTER("general_fetch");
 925
 926 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 927 +               DBUG_RETURN(HA_ERR_CRASHED);
 928 +       }
 929 +
 930         ut_a(prebuilt->trx == thd_to_trx(user_thd));
 931
 932         innodb_srv_conc_enter_innodb(prebuilt->trx);
 933 @@ -6259,6 +6318,10 @@
 934
 935         innodb_srv_conc_exit_innodb(prebuilt->trx);
 936
 937 +       if (srv_pass_corrupt_table <= 1 && share->ib_table->is_corrupt) {
 938 +               DBUG_RETURN(HA_ERR_CRASHED);
 939 +       }
 940 +
 941         switch (ret) {
 942         case DB_SUCCESS:
 943                 error = 0;
 944 @@ -7525,10 +7588,18 @@
 945
 946         update_thd(ha_thd());
 947
 948 +       if (share->ib_table->is_corrupt) {
 949 +               DBUG_RETURN(HA_ERR_CRASHED);
 950 +       }
 951 +
 952         /* Truncate the table in InnoDB */
 953
 954         error = row_truncate_table_for_mysql(prebuilt->table, prebuilt->trx);
 955
 956 +       if (share->ib_table->is_corrupt) {
 957 +               DBUG_RETURN(HA_ERR_CRASHED);
 958 +       }
 959 +
 960         error = convert_error_code_to_mysql(error, prebuilt->table->flags,
 961                                             NULL);
 962
 963 @@ -8040,6 +8111,16 @@
 964         return(ranges + (double) rows / (double) total_rows * time_for_scan);
 965  }
 966
 967 +UNIV_INTERN
 968 +bool
 969 +ha_innobase::is_corrupt() const
 970 +{
 971 +       if (share->ib_table)
 972 +               return ((bool)share->ib_table->is_corrupt);
 973 +       else
 974 +               return (FALSE);
 975 +}
 976 +
 977  /*********************************************************************//**
 978  Calculates the key number used inside MySQL for an Innobase index. We will
 979  first check the "index translation table" for a match of the index to get
 980 @@ -8217,7 +8298,7 @@
 981         ib_table = prebuilt->table;
 982
 983         if (flag & HA_STATUS_TIME) {
 984 -               if (called_from_analyze || innobase_stats_on_metadata) {
 985 +               if ((called_from_analyze || innobase_stats_on_metadata) && !share->ib_table->is_corrupt) {
 986                         /* In sql_show we call with this flag: update
 987                         then statistics so that they are up-to-date */
 988
 989 @@ -8517,10 +8598,18 @@
 990         THD*            thd,            /*!< in: connection thread handle */
 991         HA_CHECK_OPT*   check_opt)      /*!< in: currently ignored */
 992  {
 993 +       if (share->ib_table->is_corrupt) {
 994 +               return(HA_ADMIN_CORRUPT);
 995 +       }
 996 +
 997         /* Simply call ::info() with all the flags */
 998         info_low(HA_STATUS_TIME | HA_STATUS_CONST | HA_STATUS_VARIABLE,
 999                  true /* called from analyze */);
1000
1001 +       if (share->ib_table->is_corrupt) {
1002 +               return(HA_ADMIN_CORRUPT);
1003 +       }
1004 +
1005         return(0);
1006  }
1007
1008 @@ -8756,6 +8845,10 @@
1009                 my_error(ER_QUERY_INTERRUPTED, MYF(0));
1010         }
1011
1012 +       if (share->ib_table->is_corrupt) {
1013 +               return(HA_ADMIN_CORRUPT);
1014 +       }
1015 +
1016         DBUG_RETURN(is_ok ? HA_ADMIN_OK : HA_ADMIN_CORRUPT);
1017  }
1018
1019 @@ -9526,6 +9619,10 @@
1020
1021         update_thd(thd);
1022
1023 +       if (share->ib_table->is_corrupt) {
1024 +               DBUG_RETURN(HA_ERR_CRASHED);
1025 +       }
1026 +
1027         if (prebuilt->table->ibd_file_missing && !thd_tablespace_op(thd)) {
1028                 ut_print_timestamp(stderr);
1029                 fprintf(stderr,
1030 @@ -11990,6 +12087,26 @@
1031    "dump file (if present). Disabled by default.",
1032    NULL, NULL, FALSE);
1033
1034 +const char *corrupt_table_action_names[]=
1035 +{
1036 +  "assert", /* 0 */
1037 +  "warn", /* 1 */
1038 +  "salvage", /* 2 */
1039 +  NullS
1040 +};
1041 +TYPELIB corrupt_table_action_typelib=
1042 +{
1043 +  array_elements(corrupt_table_action_names) - 1, "corrupt_table_action_typelib",
1044 +  corrupt_table_action_names, NULL
1045 +};
1046 +static MYSQL_SYSVAR_ENUM(corrupt_table_action, srv_pass_corrupt_table,
1047 +  PLUGIN_VAR_RQCMDARG,
1048 +  "Warn corruptions of user tables as 'corrupt table' instead of not crashing itself, "
1049 +  "when used with file_per_table. "
1050 +  "All file io for the datafile after detected as corrupt are disabled, "
1051 +  "except for the deletion.",
1052 +  NULL, NULL, 0, &corrupt_table_action_typelib);
1053 +
1054  static struct st_mysql_sys_var* innobase_system_variables[]= {
1055    MYSQL_SYSVAR(additional_mem_pool_size),
1056    MYSQL_SYSVAR(autoextend_increment),
1057 @@ -12083,6 +12200,7 @@
1058  #ifdef UNIV_DEBUG
1059    MYSQL_SYSVAR(flush_checkpoint_debug),
1060  #endif
1061 +  MYSQL_SYSVAR(corrupt_table_action),
1062    NULL
1063  };
1064
1065 --- a/storage/innobase/handler/ha_innodb.h
1066 +++ b/storage/innobase/handler/ha_innodb.h
1067 @@ -52,6 +52,7 @@
1068         innodb_idx_translate_t  idx_trans_tbl;  /*!< index translation
1069                                                 table between MySQL and
1070                                                 Innodb */
1071 +       dict_table_t*           ib_table;
1072  } INNOBASE_SHARE;
1073
1074
1075 @@ -135,6 +136,7 @@
1076         int close(void);
1077         double scan_time();
1078         double read_time(uint index, uint ranges, ha_rows rows);
1079 +       bool is_corrupt() const;
1080
1081         int write_row(uchar * buf);
1082         int update_row(const uchar * old_data, uchar * new_data);
1083 --- a/storage/innobase/include/btr0btr.ic
1084 +++ b/storage/innobase/include/btr0btr.ic
1085 @@ -28,7 +28,7 @@
1086  #include "mtr0mtr.h"
1087  #include "mtr0log.h"
1088  #include "page0zip.h"
1089 -
1090 +#include "srv0srv.h"
1091  #define BTR_MAX_NODE_LEVEL     50      /*!< Maximum B-tree page level
1092                                         (not really a hard limit).
1093                                         Used in debug assertions
1094 @@ -59,7 +59,9 @@
1095         block = buf_page_get_gen(space, zip_size, page_no, mode,
1096                                  NULL, BUF_GET, file, line, mtr);
1097
1098 -       if (mode != RW_NO_LATCH) {
1099 +       ut_a(srv_pass_corrupt_table || block);
1100 +
1101 +       if (block && mode != RW_NO_LATCH) {
1102
1103                 buf_block_dbg_add_level(
1104                         block, index != NULL && dict_index_is_ibuf(index)
1105 --- a/storage/innobase/include/buf0buf.h
1106 +++ b/storage/innobase/include/buf0buf.h
1107 @@ -1003,7 +1003,7 @@
1108         const buf_block_t*      block)  /*!< in: pointer to the control block */
1109         __attribute__((pure));
1110  #else /* UNIV_DEBUG */
1111 -# define buf_block_get_frame(block) (block)->frame
1112 +# define buf_block_get_frame(block) (block ? (block)->frame : 0)
1113  #endif /* UNIV_DEBUG */
1114  /*********************************************************************//**
1115  Gets the space id of a block.
1116 @@ -1450,6 +1450,7 @@
1117                                         0 if the block was never accessed
1118                                         in the buffer pool */
1119         /* @} */
1120 +       ibool           is_corrupt;
1121  # if defined UNIV_DEBUG_FILE_ACCESSES || defined UNIV_DEBUG
1122         ibool           file_page_was_freed;
1123                                         /*!< this is set to TRUE when fsp
1124 --- a/storage/innobase/include/buf0buf.ic
1125 +++ b/storage/innobase/include/buf0buf.ic
1126 @@ -34,7 +34,7 @@
1127  #include "buf0flu.h"
1128  #include "buf0lru.h"
1129  #include "buf0rea.h"
1130 -
1131 +#include "srv0srv.h"
1132  /*********************************************************************//**
1133  Gets the current size of buffer buf_pool in bytes.
1134  @return size in bytes */
1135 @@ -637,6 +637,12 @@
1136  /*================*/
1137         const buf_block_t*      block)  /*!< in: pointer to the control block */
1138  {
1139 +       ut_a(srv_pass_corrupt_table || block);
1140 +
1141 +       if (srv_pass_corrupt_table && !block) {
1142 +               return(0);
1143 +       }
1144 +
1145         ut_ad(block);
1146
1147         switch (buf_block_get_state(block)) {
1148 --- a/storage/innobase/include/dict0dict.h
1149 +++ b/storage/innobase/include/dict0dict.h
1150 @@ -1326,6 +1326,15 @@
1151  /*========================*/
1152         ulint           space_id);      /*!< in: space ID */
1153
1154 +/*************************************************************************
1155 +set is_corrupt flag by space_id*/
1156 +
1157 +void
1158 +dict_table_set_corrupt_by_space(
1159 +/*============================*/
1160 +       ulint   space_id,
1161 +       ibool   need_mutex);
1162 +
1163  #ifndef UNIV_NONINL
1164  #include "dict0dict.ic"
1165  #endif
1166 --- a/storage/innobase/include/dict0mem.h
1167 +++ b/storage/innobase/include/dict0mem.h
1168 @@ -670,6 +670,7 @@
1169                                 the AUTOINC lock on this table. */
1170                                 /* @} */
1171         /*----------------------*/
1172 +       ibool           is_corrupt;
1173  #endif /* !UNIV_HOTBACKUP */
1174
1175  #ifdef UNIV_DEBUG
1176 --- a/storage/innobase/include/fil0fil.h
1177 +++ b/storage/innobase/include/fil0fil.h
1178 @@ -754,6 +754,19 @@
1179  fil_system_hash_nodes(void);
1180  /*========================*/
1181
1182 +/*************************************************************************
1183 +functions to access is_corrupt flag of fil_space_t*/
1184 +
1185 +ibool
1186 +fil_space_is_corrupt(
1187 +/*=================*/
1188 +       ulint   space_id);
1189 +
1190 +void
1191 +fil_space_set_corrupt(
1192 +/*==================*/
1193 +       ulint   space_id);
1194 +
1195  typedef        struct fil_space_struct fil_space_t;
1196
1197  #endif
1198 --- a/storage/innobase/include/fut0fut.ic
1199 +++ b/storage/innobase/include/fut0fut.ic
1200 @@ -23,6 +23,7 @@
1201  Created 12/13/1995 Heikki Tuuri
1202  ***********************************************************************/
1203
1204 +#include "srv0srv.h"
1205  #include "sync0rw.h"
1206  #include "buf0buf.h"
1207
1208 @@ -48,6 +49,12 @@
1209         ut_ad((rw_latch == RW_S_LATCH) || (rw_latch == RW_X_LATCH));
1210
1211         block = buf_page_get(space, zip_size, addr.page, rw_latch, mtr);
1212 +
1213 +       if (srv_pass_corrupt_table && !block) {
1214 +               return(0);
1215 +       }
1216 +       ut_a(block);
1217 +
1218         ptr = buf_block_get_frame(block) + addr.boffset;
1219
1220         buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
1221 --- a/storage/innobase/include/page0page.h
1222 +++ b/storage/innobase/include/page0page.h
1223 @@ -497,7 +497,7 @@
1224  page_is_leaf(
1225  /*=========*/
1226         const page_t*   page)   /*!< in: page */
1227 -       __attribute__((nonnull, pure));
1228 +       __attribute__((pure));
1229  /************************************************************//**
1230  Gets the pointer to the next record on the page.
1231  @return        pointer to next record */
1232 --- a/storage/innobase/include/page0page.ic
1233 +++ b/storage/innobase/include/page0page.ic
1234 @@ -274,6 +274,9 @@
1235  /*=========*/
1236         const page_t*   page)   /*!< in: page */
1237  {
1238 +       if (!page) {
1239 +               return(FALSE);
1240 +       }
1241         return(!*(const uint16*) (page + (PAGE_HEADER + PAGE_LEVEL)));
1242  }
1243
1244 --- a/storage/innobase/include/page0zip.h
1245 +++ b/storage/innobase/include/page0zip.h
1246 @@ -114,7 +114,7 @@
1247         const page_t*   page,   /*!< in: uncompressed page */
1248         dict_index_t*   index,  /*!< in: index of the B-tree node */
1249         mtr_t*          mtr)    /*!< in: mini-transaction, or NULL */
1250 -       __attribute__((nonnull(1,2,3)));
1251 +       __attribute__((nonnull(1,3)));
1252
1253  /**********************************************************************//**
1254  Decompress a page.  This function should tolerate errors on the compressed
1255 --- a/storage/innobase/include/srv0srv.h
1256 +++ b/storage/innobase/include/srv0srv.h
1257 @@ -245,6 +245,7 @@
1258  extern ulint   srv_adaptive_flushing_method;
1259
1260  extern ulint   srv_expand_import;
1261 +extern ulint   srv_pass_corrupt_table;
1262
1263  extern ulint   srv_dict_size_limit;
1264  /*-------------------------------------------*/
1265 --- a/storage/innobase/page/page0zip.c
1266 +++ b/storage/innobase/page/page0zip.c
1267 @@ -1195,6 +1195,10 @@
1268         FILE*           logfile = NULL;
1269  #endif
1270
1271 +       if (!page) {
1272 +               return(FALSE);
1273 +       }
1274 +
1275         ut_a(page_is_comp(page));
1276         ut_a(fil_page_get_type(page) == FIL_PAGE_INDEX);
1277         ut_ad(page_simple_validate_new((page_t*) page));
1278 --- a/storage/innobase/row/row0ins.c
1279 +++ b/storage/innobase/row/row0ins.c
1280 @@ -1341,6 +1341,12 @@
1281                 const rec_t*            rec = btr_pcur_get_rec(&pcur);
1282                 const buf_block_t*      block = btr_pcur_get_block(&pcur);
1283
1284 +               if (srv_pass_corrupt_table && !block) {
1285 +                       err = DB_CORRUPTION;
1286 +                       break;
1287 +               }
1288 +               ut_a(block);
1289 +
1290                 if (page_rec_is_infimum(rec)) {
1291
1292                         continue;
1293 --- a/storage/innobase/row/row0merge.c
1294 +++ b/storage/innobase/row/row0merge.c
1295 @@ -1245,6 +1245,13 @@
1296
1297                 if (UNIV_LIKELY(has_next)) {
1298                         rec = btr_pcur_get_rec(&pcur);
1299 +
1300 +                       if (srv_pass_corrupt_table && !rec) {
1301 +                               err = DB_CORRUPTION;
1302 +                               goto err_exit;
1303 +                       }
1304 +                       ut_a(rec);
1305 +
1306                         offsets = rec_get_offsets(rec, clust_index, NULL,
1307                                                   ULINT_UNDEFINED, &row_heap);
1308
1309 --- a/storage/innobase/row/row0sel.c
1310 +++ b/storage/innobase/row/row0sel.c
1311 @@ -3912,6 +3912,13 @@
1312         /* PHASE 4: Look for matching records in a loop */
1313
1314         rec = btr_pcur_get_rec(pcur);
1315 +
1316 +       if (srv_pass_corrupt_table && !rec) {
1317 +               err = DB_CORRUPTION;
1318 +               goto lock_wait_or_error;
1319 +       }
1320 +       ut_a(rec);
1321 +
1322         ut_ad(!!page_rec_is_comp(rec) == comp);
1323  #ifdef UNIV_SEARCH_DEBUG
1324         /*
1325 @@ -3989,7 +3996,13 @@
1326         if (UNIV_UNLIKELY(next_offs >= UNIV_PAGE_SIZE - PAGE_DIR)) {
1327
1328  wrong_offs:
1329 -               if (srv_force_recovery == 0 || moves_up == FALSE) {
1330 +               if (srv_pass_corrupt_table && !trx_sys_sys_space(index->table->space)) {
1331 +                       index->table->is_corrupt = TRUE;
1332 +                       fil_space_set_corrupt(index->table->space);
1333 +               }
1334 +
1335 +               if ((srv_force_recovery == 0 || moves_up == FALSE)
1336 +                   && srv_pass_corrupt_table <= 1) {
1337                         ut_print_timestamp(stderr);
1338                         buf_page_print(page_align(rec), 0);
1339                         fprintf(stderr,
1340 @@ -4040,7 +4053,8 @@
1341
1342         offsets = rec_get_offsets(rec, index, offsets, ULINT_UNDEFINED, &heap);
1343
1344 -       if (UNIV_UNLIKELY(srv_force_recovery > 0)) {
1345 +       if (UNIV_UNLIKELY(srv_force_recovery > 0)
1346 +           || (srv_pass_corrupt_table == 2 && index->table->is_corrupt)) {
1347                 if (!rec_validate(rec, offsets)
1348                     || !btr_index_rec_validate(rec, index, FALSE)) {
1349                         fprintf(stderr,
1350 --- a/storage/innobase/srv/srv0srv.c
1351 +++ b/storage/innobase/srv/srv0srv.c
1352 @@ -435,6 +435,7 @@
1353  UNIV_INTERN ulint      srv_adaptive_flushing_method = 0; /* 0: native  1: estimate  2: keep_average */
1354
1355  UNIV_INTERN ulint      srv_expand_import = 0; /* 0:disable 1:enable */
1356 +UNIV_INTERN ulint      srv_pass_corrupt_table = 0; /* 0:disable 1:enable */
1357
1358  UNIV_INTERN ulint      srv_dict_size_limit = 0;
1359  /*-------------------------------------------*/
1360 --- a/storage/innobase/srv/srv0start.c
1361 +++ b/storage/innobase/srv/srv0start.c
1362 @@ -2155,6 +2155,13 @@
1363
1364         os_fast_mutex_free(&srv_os_test_mutex);
1365
1366 +       if (!srv_file_per_table_original_value
1367 +           && srv_pass_corrupt_table) {
1368 +               fprintf(stderr, "InnoDB: Warning:"
1369 +                       " The option innodb_file_per_table is disabled,"
1370 +                       " so using the option innodb_pass_corrupt_table doesn't make sense.\n");
1371 +       }
1372 +
1373         if (srv_print_verbose_log) {
1374                 ut_print_timestamp(stderr);
1375                 fprintf(stderr,