Merge with 5.1

Fixes to get Aria handler tests to work.
Fixes LP#697597 "HANDLER + Aria asserts in maria-5.3-handler" 

36 files changed, 338 insertions, 153 deletions

diff --git a/storage/ibmdb2i/db2i_ioBuffers.h b/storage/ibmdb2i/db2i_ioBuffers.h
index 350d854f055..8fb815ba3be 100644
--- a/storage/ibmdb2i/db2i_ioBuffers.h
+++ b/storage/ibmdb2i/db2i_ioBuffers.h
@@ -290,7 +290,7 @@ class IOAsyncReadBuffer : public IOReadBuffer
       Return a pointer to the next row in the table, where "next" is defined
       by the orientation.
       
-      @param orientaton
+      @param orientation
       @param[out] rrn The relative record number of the row returned. Not reliable
                       if NULL is returned by this function.
       
diff --git a/storage/innodb_plugin/trx/trx0i_s.c b/storage/innodb_plugin/trx/trx0i_s.c
index 5bc8302d0c0..c0b9a73a68c 100644
--- a/storage/innodb_plugin/trx/trx0i_s.c
+++ b/storage/innodb_plugin/trx/trx0i_s.c
@@ -157,10 +157,6 @@ struct trx_i_s_cache_struct {
 	ullint		last_read;	/*!< last time the cache was read;
 					measured in microseconds since
 					epoch */
-	mutex_t		last_read_mutex;/*!< mutex protecting the
-					last_read member - it is updated
-					inside a shared lock of the
-					rw_lock member */
 	i_s_table_cache_t innodb_trx;	/*!< innodb_trx table */
 	i_s_table_cache_t innodb_locks;	/*!< innodb_locks table */
 	i_s_table_cache_t innodb_lock_waits;/*!< innodb_lock_waits table */
@@ -1101,13 +1097,6 @@ can_cache_be_updated(
 {
 	ullint	now;
 
-	/* Here we read cache->last_read without acquiring its mutex
-	because last_read is only updated when a shared rw lock on the
-	whole cache is being held (see trx_i_s_cache_end_read()) and
-	we are currently holding an exclusive rw lock on the cache.
-	So it is not possible for last_read to be updated while we are
-	reading it. */
-
 #ifdef UNIV_SYNC_DEBUG
 	ut_a(rw_lock_own(&cache->rw_lock, RW_LOCK_EX));
 #endif
@@ -1205,6 +1194,12 @@ trx_i_s_possibly_fetch_data_into_cache(
 /*===================================*/
 	trx_i_s_cache_t*	cache)	/*!< in/out: cache */
 {
+	ullint	now;
+
+#ifdef UNIV_SYNC_DEBUG
+	ut_a(rw_lock_own(&cache->rw_lock, RW_LOCK_EX));
+#endif
+
 	if (!can_cache_be_updated(cache)) {
 
 		return(1);
@@ -1217,6 +1212,10 @@ trx_i_s_possibly_fetch_data_into_cache(
 
 	mutex_exit(&kernel_mutex);
 
+	/* update cache last read time */
+	now = ut_time_us(NULL);
+	cache->last_read = now;
+
 	return(0);
 }
 
@@ -1247,16 +1246,12 @@ trx_i_s_cache_init(
 	release kernel_mutex
 	release trx_i_s_cache_t::rw_lock
 	acquire trx_i_s_cache_t::rw_lock, S
-	acquire trx_i_s_cache_t::last_read_mutex
-	release trx_i_s_cache_t::last_read_mutex
 	release trx_i_s_cache_t::rw_lock */
 
 	rw_lock_create(&cache->rw_lock, SYNC_TRX_I_S_RWLOCK);
 
 	cache->last_read = 0;
 
-	mutex_create(&cache->last_read_mutex, SYNC_TRX_I_S_LAST_READ);
-
 	table_cache_init(&cache->innodb_trx, sizeof(i_s_trx_row_t));
 	table_cache_init(&cache->innodb_locks, sizeof(i_s_locks_row_t));
 	table_cache_init(&cache->innodb_lock_waits,
@@ -1307,18 +1302,10 @@ trx_i_s_cache_end_read(
 /*===================*/
 	trx_i_s_cache_t*	cache)	/*!< in: cache */
 {
-	ullint	now;
-
 #ifdef UNIV_SYNC_DEBUG
 	ut_a(rw_lock_own(&cache->rw_lock, RW_LOCK_SHARED));
 #endif
 
-	/* update cache last read time */
-	now = ut_time_us(NULL);
-	mutex_enter(&cache->last_read_mutex);
-	cache->last_read = now;
-	mutex_exit(&cache->last_read_mutex);
-
 	rw_lock_s_unlock(&cache->rw_lock);
 }
 
diff --git a/storage/innodb_plugin/trx/trx0sys.c b/storage/innodb_plugin/trx/trx0sys.c
index 6eb356947cc..48963bc7e45 100644
--- a/storage/innodb_plugin/trx/trx0sys.c
+++ b/storage/innodb_plugin/trx/trx0sys.c
@@ -1343,7 +1343,7 @@ trx_sys_print_mysql_binlog_offset_from_page(
 
 
 /* THESE ARE COPIED FROM NON-HOTBACKUP PART OF THE INNODB SOURCE TREE
-   (This code duplicaton should be fixed at some point!)
+   (This code duplication should be fixed at some point!)
 */
 
 #define	TRX_SYS_SPACE	0	/* the SYSTEM tablespace */
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 5ea8f1c9a40..6f01d1e9b5e 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -821,7 +821,7 @@ int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
                 HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
                 HA_DUPLICATE_POS | HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
                 HA_FILE_BASED | HA_CAN_GEOMETRY | CANNOT_ROLLBACK_FLAG |
-                HA_CAN_BIT_FIELD | HA_CAN_RTREEKEYS |
+                HA_CAN_BIT_FIELD | HA_CAN_RTREEKEYS | HA_CAN_VIRTUAL_COLUMNS |
                 HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT),
 can_enable_indexes(1), bulk_insert_single_undo(BULK_INSERT_NONE)
 {}
@@ -1071,8 +1071,6 @@ int ha_maria::close(void)
 
 int ha_maria::write_row(uchar * buf)
 {
-  ha_statistic_increment(&SSV::ha_write_count);
-
   /* If we have a timestamp column, update it to the current time */
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
     table->timestamp_field->set_time();
@@ -2072,6 +2070,7 @@ bool ha_maria::check_and_repair(THD *thd)
   DBUG_ENTER("ha_maria::check_and_repair");
 
   check_opt.init();
+  check_opt.flags= T_MEDIUM | T_AUTO_REPAIR;
 
   error= 1;
   if ((file->s->state.changed &
@@ -2092,7 +2091,6 @@ bool ha_maria::check_and_repair(THD *thd)
     DBUG_RETURN(error);
 
   error= 0;
-  check_opt.flags= T_MEDIUM | T_AUTO_REPAIR;
   // Don't use quick if deleted rows
   if (!file->state->del && (maria_recover_options & HA_RECOVER_QUICK))
     check_opt.flags |= T_QUICK;
@@ -2143,7 +2141,6 @@ bool ha_maria::is_crashed() const
 int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 {
   CHECK_UNTIL_WE_FULLY_IMPLEMENTED_VERSIONING("UPDATE in WRITE CONCURRENT");
-  ha_statistic_increment(&SSV::ha_update_count);
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
     table->timestamp_field->set_time();
   return maria_update(file, old_data, new_data);
@@ -2153,7 +2150,6 @@ int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 int ha_maria::delete_row(const uchar * buf)
 {
   CHECK_UNTIL_WE_FULLY_IMPLEMENTED_VERSIONING("DELETE in WRITE CONCURRENT");
-  ha_statistic_increment(&SSV::ha_delete_count);
   return maria_delete(file, buf);
 }
 
@@ -2177,7 +2173,6 @@ int ha_maria::index_read_map(uchar * buf, const uchar * key,
 			     enum ha_rkey_function find_flag)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_key_count);
   int error= maria_rkey(file, buf, active_index, key, keypart_map, find_flag);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2188,7 +2183,6 @@ int ha_maria::index_read_idx_map(uchar * buf, uint index, const uchar * key,
 				 key_part_map keypart_map,
 				 enum ha_rkey_function find_flag)
 {
-  ha_statistic_increment(&SSV::ha_read_key_count);
   int error= maria_rkey(file, buf, index, key, keypart_map, find_flag);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2200,7 +2194,6 @@ int ha_maria::index_read_last_map(uchar * buf, const uchar * key,
 {
   DBUG_ENTER("ha_maria::index_read_last_map");
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_key_count);
   int error= maria_rkey(file, buf, active_index, key, keypart_map,
                         HA_READ_PREFIX_LAST);
   table->status= error ? STATUS_NOT_FOUND : 0;
@@ -2211,7 +2204,6 @@ int ha_maria::index_read_last_map(uchar * buf, const uchar * key,
 int ha_maria::index_next(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_next_count);
   int error= maria_rnext(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2221,7 +2213,6 @@ int ha_maria::index_next(uchar * buf)
 int ha_maria::index_prev(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_prev_count);
   int error= maria_rprev(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2231,7 +2222,6 @@ int ha_maria::index_prev(uchar * buf)
 int ha_maria::index_first(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_first_count);
   int error= maria_rfirst(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2241,7 +2231,6 @@ int ha_maria::index_first(uchar * buf)
 int ha_maria::index_last(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_last_count);
   int error= maria_rlast(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2254,7 +2243,6 @@ int ha_maria::index_next_same(uchar * buf,
 {
   int error;
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_next_count);
   /*
     TODO: Delete this loop in Maria 1.5 as versioning will ensure this never
     happens
@@ -2306,7 +2294,6 @@ int ha_maria::rnd_end()
 
 int ha_maria::rnd_next(uchar *buf)
 {
-  ha_statistic_increment(&SSV::ha_read_rnd_next_count);
   int error= maria_scan(file, buf);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2328,7 +2315,6 @@ int ha_maria::restart_rnd_next(uchar *buf)
 
 int ha_maria::rnd_pos(uchar *buf, uchar *pos)
 {
-  ha_statistic_increment(&SSV::ha_read_rnd_count);
   int error= maria_rrnd(file, buf, my_get_ptr(pos, ref_length));
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -3353,6 +3339,9 @@ static int ha_maria_init(void *p)
   maria_hton->panic= maria_hton_panic;
   maria_hton->commit= maria_commit;
   maria_hton->rollback= maria_rollback;
+#ifdef MARIA_CANNOT_ROLLBACK
+  maria_hton->commit= 0;
+#endif
   maria_hton->flush_logs= maria_flush_logs;
   maria_hton->show_status= maria_show_status;
   /* TODO: decide if we support Maria being used for log tables */
diff --git a/storage/maria/ha_maria.h b/storage/maria/ha_maria.h
index 0433134275c..5e7b068d01c 100644
--- a/storage/maria/ha_maria.h
+++ b/storage/maria/ha_maria.h
@@ -157,7 +157,6 @@ public:
   int assign_to_keycache(THD * thd, HA_CHECK_OPT * check_opt);
   int preload_keys(THD * thd, HA_CHECK_OPT * check_opt);
   bool check_if_incompatible_data(HA_CREATE_INFO * info, uint table_changes);
-  bool check_if_supported_virtual_columns(void) { return TRUE;}
 #ifdef HAVE_REPLICATION
   int dump(THD * thd, int fd);
   int net_read_dump(NET * net);
diff --git a/storage/maria/ma_bitmap.c b/storage/maria/ma_bitmap.c
index 7e68437de2b..bc83f6ea0e9 100644
--- a/storage/maria/ma_bitmap.c
+++ b/storage/maria/ma_bitmap.c
@@ -399,7 +399,8 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
       become false, wake them up.
     */
     DBUG_PRINT("info", ("bitmap flusher waking up others"));
-    pthread_cond_broadcast(&bitmap->bitmap_cond);
+    if (bitmap->flush_all_requested)
+      pthread_cond_broadcast(&bitmap->bitmap_cond);
   }
   pthread_mutex_unlock(&bitmap->bitmap_lock);
   DBUG_RETURN(res);
@@ -465,7 +466,8 @@ void _ma_bitmap_unlock(MARIA_SHARE *share)
   bitmap->flush_all_requested--;
   bitmap->non_flushable= 0;
   pthread_mutex_unlock(&bitmap->bitmap_lock);
-  pthread_cond_broadcast(&bitmap->bitmap_cond);
+  if (bitmap->flush_all_requested > 0)
+    pthread_cond_broadcast(&bitmap->bitmap_cond);
   DBUG_VOID_RETURN;
 }
 
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index fd02e2ac0ec..a2faeb6e7fa 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -2506,7 +2506,7 @@ static my_bool free_full_page_range(MARIA_HA *info, pgcache_page_no_t page,
   }
   if (delete_count &&
       pagecache_delete_pages(share->pagecache, &info->dfile,
-                             page, delete_count, PAGECACHE_LOCK_WRITE, 0))
+                             page, delete_count, PAGECACHE_LOCK_WRITE, 1))
     res= 1;
 
   if (share->now_transactional)
@@ -2816,7 +2816,6 @@ static my_bool write_block_record(MARIA_HA *info,
   DBUG_PRINT("info", ("Used head length on page: %u  header_length: %u",
                       head_length,
                       (uint) (flag & ROW_FLAG_TRANSID ? TRANSID_SIZE : 0)));
-  DBUG_ASSERT(data <= end_of_data);
   if (head_length < share->base.min_block_length)
   {
     /* Extend row to be of size min_block_length */
@@ -2825,6 +2824,7 @@ static my_bool write_block_record(MARIA_HA *info,
     data+= diff_length;
     head_length= share->base.min_block_length;
   }
+  DBUG_ASSERT(data <= end_of_data);
   /*
     If this is a redo entry (ie, undo_lsn != LSN_ERROR) then we should have
     written exactly head_length bytes (same as original record).
@@ -3492,7 +3492,9 @@ static my_bool allocate_and_write_block_record(MARIA_HA *info,
 
   /* page will be pinned & locked by get_head_or_tail_page */
   if (get_head_or_tail_page(info, blocks->block, info->buff,
-                            row->space_on_head_page, HEAD_PAGE,
+                            max(row->space_on_head_page,
+                                info->s->base.min_block_length),
+                            HEAD_PAGE,
                             PAGECACHE_LOCK_WRITE, &row_pos))
     goto err;
   row->lastpos= ma_recordpos(blocks->block->page, row_pos.rownr);
@@ -4179,6 +4181,13 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
                                 log_data, NULL))
         DBUG_RETURN(1);
     }
+    /*
+      Mark that this page must be written to disk by page cache, even
+      if we could call pagecache_delete() on it.
+      This is needed to ensure that repair finds the empty page on disk
+      and not old data.
+    */
+    pagecache_set_write_on_delete_by_link(page_link.link);
     DBUG_ASSERT(empty_space >= share->bitmap.sizes[0]);
   }
 
@@ -4932,7 +4941,7 @@ int _ma_read_block_record2(MARIA_HA *info, uchar *record,
       goto err;
   }
 #ifdef EXTRA_DEBUG
-  if (share->calc_checksum)
+  if (share->calc_checksum && !info->in_check_table)
   {
     /* Esnure that row checksum is correct */
     DBUG_ASSERT(((share->calc_checksum)(info, record) & 255) ==
@@ -6485,7 +6494,13 @@ err:
    @param  info            Maria handler
    @param  header          Header (without FILEID)
 
-   @note It marks the pages free in the bitmap
+   Mark the pages free in the bitmap.
+
+   We have to check against _ma_redo_not_needed_for_page()
+   to guard against the case where we first clear a block and after
+   that insert new data into the blocks.  If we would unconditionally
+   clear the bitmap here, future changes would be ignored for the page
+   if it's not in the dirty list (ie, it would be flushed).
 
    @return Operation status
      @retval 0      OK
@@ -6494,19 +6509,25 @@ err:
 
 uint _ma_apply_redo_free_blocks(MARIA_HA *info,
                                 LSN lsn __attribute__((unused)),
+                                LSN redo_lsn,
                                 const uchar *header)
 {
   MARIA_SHARE *share= info->s;
   uint ranges;
+  uint16 sid;
   DBUG_ENTER("_ma_apply_redo_free_blocks");
 
   share->state.changed|= (STATE_CHANGED | STATE_NOT_ZEROFILLED |
                           STATE_NOT_MOVABLE);
 
+  sid= fileid_korr(header);
+  header+= FILEID_STORE_SIZE;
   ranges= pagerange_korr(header);
   header+= PAGERANGE_STORE_SIZE;
   DBUG_ASSERT(ranges > 0);
 
+  /** @todo leave bitmap lock to the bitmap code... */
+  pthread_mutex_lock(&share->bitmap.bitmap_lock);
   while (ranges--)
   {
     my_bool res;
@@ -6523,18 +6544,22 @@ uint _ma_apply_redo_free_blocks(MARIA_HA *info,
 
     DBUG_PRINT("info", ("page: %lu  pages: %u", (long) page, page_range));
 
-    /** @todo leave bitmap lock to the bitmap code... */
-    pthread_mutex_lock(&share->bitmap.bitmap_lock);
-    res= _ma_bitmap_reset_full_page_bits(info, &share->bitmap, start_page,
-                                         page_range);
-    pthread_mutex_unlock(&share->bitmap.bitmap_lock);
-    if (res)
+    for ( ; page_range-- ; start_page++)
     {
-      _ma_mark_file_crashed(share);
-      DBUG_ASSERT(0);
-      DBUG_RETURN(res);
+      if (_ma_redo_not_needed_for_page(sid, redo_lsn, start_page, FALSE))
+        continue;
+      res= _ma_bitmap_reset_full_page_bits(info, &share->bitmap, start_page,
+                                           1);
+      if (res)
+      {
+        pthread_mutex_unlock(&share->bitmap.bitmap_lock);
+        _ma_mark_file_crashed(share);
+        DBUG_ASSERT(0);
+        DBUG_RETURN(res);
+      }
     }
   }
+  pthread_mutex_unlock(&share->bitmap.bitmap_lock);
   DBUG_RETURN(0);
 }
 
@@ -6687,21 +6712,23 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
       uint      page_range;
       pgcache_page_no_t page, start_page;
       uchar     *buff;
+      uint	data_on_page= data_size;
 
       start_page= page= page_korr(header);
       header+= PAGE_STORE_SIZE;
       page_range= pagerange_korr(header);
       header+= PAGERANGE_STORE_SIZE;
 
-      for (i= page_range; i-- > 0 ; page++)
+      for (i= page_range; i-- > 0 ; page++, data+= data_on_page)
       {
         MARIA_PINNED_PAGE page_link;
         enum pagecache_page_lock unlock_method;
         enum pagecache_page_pin unpin_method;
-        uint length;
 
         set_if_smaller(first_page2, page);
         set_if_bigger(last_page2, page);
+        if (i == 0 && sub_ranges == 0)
+          data_on_page= data_size - empty_space; /* data on last page */
         if (_ma_redo_not_needed_for_page(sid, redo_lsn, page, FALSE))
           continue;
 
@@ -6764,7 +6791,7 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
                                        PAGECACHE_LOCK_WRITE_UNLOCK,
                                        PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
                                        LSN_IMPOSSIBLE, 0, FALSE);
-              continue;
+              goto fix_bitmap;
             }
             DBUG_ASSERT((found_page_type == (uchar) BLOB_PAGE) ||
                         (found_page_type == (uchar) UNALLOCATED_PAGE));
@@ -6780,33 +6807,32 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
         lsn_store(buff, lsn);
         buff[PAGE_TYPE_OFFSET]= BLOB_PAGE;
 
-        length= data_size;
-        if (i == 0 && sub_ranges == 0)
+        if (data_on_page != data_size)
         {
           /*
             Last page may be only partly filled. We zero the rest, like
             write_full_pages() does.
           */
-          length-= empty_space;
           bzero(buff + share->block_size - PAGE_SUFFIX_SIZE - empty_space,
                 empty_space);
         }
-        memcpy(buff+ PAGE_TYPE_OFFSET + 1, data, length);
-        data+= length;
+        memcpy(buff+ PAGE_TYPE_OFFSET + 1, data, data_on_page);
         if (pagecache_write(share->pagecache,
                             &info->dfile, page, 0,
                             buff, PAGECACHE_PLAIN_PAGE,
                             unlock_method, unpin_method,
                             PAGECACHE_WRITE_DELAY, 0, LSN_IMPOSSIBLE))
           goto err;
-      }
+
+    fix_bitmap:
       /** @todo leave bitmap lock to the bitmap code... */
-      pthread_mutex_lock(&share->bitmap.bitmap_lock);
-      res= _ma_bitmap_set_full_page_bits(info, &share->bitmap, start_page,
-                                         page_range);
-      pthread_mutex_unlock(&share->bitmap.bitmap_lock);
-      if (res)
-        goto err;
+        pthread_mutex_lock(&share->bitmap.bitmap_lock);
+        res= _ma_bitmap_set_full_page_bits(info, &share->bitmap, page,
+                                           1);
+        pthread_mutex_unlock(&share->bitmap.bitmap_lock);
+        if (res)
+          goto err;
+      }
     }
   }
   *first_page= first_page2;
diff --git a/storage/maria/ma_blockrec.h b/storage/maria/ma_blockrec.h
index a5858880dd0..e63af8eed5b 100644
--- a/storage/maria/ma_blockrec.h
+++ b/storage/maria/ma_blockrec.h
@@ -235,7 +235,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
 uint _ma_apply_redo_purge_row_head_or_tail(MARIA_HA *info, LSN lsn,
                                            uint page_type,
                                            const uchar *header);
-uint _ma_apply_redo_free_blocks(MARIA_HA *info, LSN lsn,
+uint _ma_apply_redo_free_blocks(MARIA_HA *info, LSN lsn, LSN rec_lsn,
                                 const uchar *header);
 uint _ma_apply_redo_free_head_or_tail(MARIA_HA *info, LSN lsn,
                                       const uchar *header);
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index 159569359f6..b12f2b894e8 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -100,6 +100,9 @@ static my_bool _ma_flush_table_files_before_swap(HA_CHECK *param,
 static TrID max_trid_in_system(void);
 static void _ma_check_print_not_visible_error(HA_CHECK *param, TrID used_trid);
 void retry_if_quick(MARIA_SORT_PARAM *param, int error);
+static void print_bitmap_description(MARIA_SHARE *share,
+                                     pgcache_page_no_t page,
+                                     uchar *buff);
 
 
 /* Initialize check param with default values */
@@ -1842,6 +1845,8 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
       }
       param->used+= block_size;
       param->link_used+= block_size;
+      if (param->verbose > 2)
+        print_bitmap_description(share, page, bitmap_buff);
       continue;
     }
     /* Skip pages marked as empty in bitmap */
@@ -2034,6 +2039,8 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
   bzero((char*) param->tmp_key_crc,
         share->base.keys * sizeof(param->tmp_key_crc[0]));
 
+  info->in_check_table= 1;       /* Don't assert on checksum errors */
+
   switch (share->data_file_type) {
   case BLOCK_RECORD:
     error= check_block_record(param, info, extend, record);
@@ -2049,6 +2056,8 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
     break;
   } /* switch */
 
+  info->in_check_table= 0;
+
   if (error)
     goto err;
 
@@ -2177,12 +2186,17 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
            llstr(param->del_length, llbuff2));
     printf("Empty space:  %12s    Linkdata:     %10s\n",
            llstr(param->empty, llbuff),llstr(param->link_used, llbuff2));
-    if (param->lost)
-      printf("Lost space:   %12s", llstr(param->lost, llbuff));
-    if (param->max_found_trid)
+    if (share->data_file_type == BLOCK_RECORD)
     {
-      printf("Max trans. id: %11s\n",
-             llstr(param->max_found_trid, llbuff));
+      printf("Full pages:   %12s    Tail count: %12s\n",
+             llstr(param->full_page_count, llbuff),
+             llstr(param->tail_count, llbuff2));
+      printf("Lost space:   %12s\n", llstr(param->lost, llbuff));
+      if (param->max_found_trid)
+      {
+        printf("Max trans. id: %11s\n",
+               llstr(param->max_found_trid, llbuff));
+      }
     }
   }
   my_free(record,MYF(0));
@@ -2616,6 +2630,7 @@ int maria_repair(HA_CHECK *param, register MARIA_HA *info,
 
   maria_lock_memory(param);			/* Everything is alloced */
 
+  sort_param.sort_info->info->in_check_table= 1;
   /* Re-create all keys, which are set in key_map. */
   while (!(error=sort_get_next_record(&sort_param)))
   {
@@ -2783,6 +2798,7 @@ err:
   VOID(end_io_cache(&sort_info.new_info->rec_cache));
   info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED);
   sort_info.new_info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED);
+  sort_param.sort_info->info->in_check_table= 0;
   /* this below could fail, shouldn't we detect error? */
   if (got_error)
   {
@@ -3133,7 +3149,8 @@ static int sort_one_index(HA_CHECK *param, MARIA_HA *info,
   key.keyinfo= keyinfo;
 
   if (!(buff= (uchar*) my_alloca((uint) keyinfo->block_length +
-                                 keyinfo->maxlength)))
+                                 keyinfo->maxlength +
+                                 MARIA_INDEX_OVERHEAD_SIZE)))
   {
     _ma_check_print_error(param,"Not enough memory for key block");
     DBUG_RETURN(-1);
@@ -3232,6 +3249,7 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
   uint block_size= share->block_size;
   my_bool zero_lsn= (share->base.born_transactional &&
                      !(param->testflag & T_ZEROFILL_KEEP_LSN));
+  int error= 1;
   DBUG_ENTER("maria_zerofill_index");
 
   if (!(param->testflag & T_SILENT))
@@ -3256,7 +3274,7 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
       _ma_check_print_error(param,
                             "Page %9s: Got error %d when reading index file",
                             llstr(pos, llbuff), my_errno);
-      DBUG_RETURN(1);
+      goto end;
     }
     if (zero_lsn)
       bzero(buff, LSN_SIZE);
@@ -3264,7 +3282,7 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
     if (share->base.born_transactional)
     {
       uint keynr= _ma_get_keynr(share, buff);
-      if (keynr != MARIA_DELETE_KEY_NR)
+      if (keynr < share->base.keys)
       {
         MARIA_PAGE page;
         DBUG_ASSERT(keynr < share->base.keys);
@@ -3276,7 +3294,7 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
                                 "Page %9s: Got error %d when reading index "
                                 "file",
                                 llstr(pos, llbuff), my_errno);
-          DBUG_RETURN(1);
+          goto end;
         }
       }
     }
@@ -3290,10 +3308,13 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
                              PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
                              LSN_IMPOSSIBLE, 1, FALSE);
   }
+  error= 0;                                     /* ok */
+
+end:
   if (flush_pagecache_blocks(share->pagecache, &share->kfile,
                              FLUSH_FORCE_WRITE))
     DBUG_RETURN(1);
-  DBUG_RETURN(0);
+  DBUG_RETURN(error);
 }
 
 
@@ -4753,7 +4774,7 @@ static int sort_get_next_record(MARIA_SORT_PARAM *sort_param)
         DBUG_RETURN(-1);
       }
       /* Retry only if wrong record, not if disk error */
-      if (flag != HA_ERR_WRONG_IN_RECORD)
+      if (flag != HA_ERR_WRONG_IN_RECORD && flag != HA_ERR_WRONG_CRC)
       {
         retry_if_quick(sort_param, flag);
         DBUG_RETURN(flag);
@@ -6443,6 +6464,9 @@ static void change_data_file_descriptor(MARIA_HA *info, File new_file)
 
 static void unuse_data_file_descriptor(MARIA_HA *info)
 {
+  (void) flush_pagecache_blocks(info->s->pagecache,
+                                &info->s->bitmap.file,
+                                FLUSH_IGNORE_CHANGED);
   info->dfile.file= info->s->bitmap.file.file= -1;
   _ma_bitmap_reset_cache(info->s);
 }
@@ -6798,3 +6822,46 @@ void retry_if_quick(MARIA_SORT_PARAM *sort_param, int error)
     param->testflag|=T_RETRY_WITHOUT_QUICK;
   }
 }
+
+/* Print information about bitmap page */
+
+static void print_bitmap_description(MARIA_SHARE *share,
+                                     pgcache_page_no_t page,
+                                     uchar *bitmap_data)
+{
+  uchar *pos, *end;
+  MARIA_FILE_BITMAP *bitmap= &share->bitmap;
+  uint count=0, dot_printed= 0;
+  char buff[80], last[80];
+
+  printf("Bitmap page %lu\n", (ulong) page);
+  page++;
+  last[0]=0;
+  for (pos= bitmap_data, end= pos+ bitmap->used_size ; pos < end ; pos+= 6)
+  {
+    ulonglong bits= uint6korr(pos);    /* 6 bytes = 6*8/3= 16 patterns */
+    uint i;
+
+    for (i= 0; i < 16 ; i++, bits>>= 3)
+    {
+      if (count > 60)
+      {
+        buff[count]= 0;
+        if (strcmp(buff, last))
+        {
+          memcpy(last, buff, count+1);
+          printf("%8lu: %s\n", (ulong) page - count, buff);
+          dot_printed= 0;
+        }
+        else if (!(dot_printed++))
+          printf("...\n");
+        count= 0;
+      }
+      buff[count++]= '0' + (uint) (bits & 7);
+      page++;
+    }
+  }
+  buff[count]= 0;
+  printf("%8lu: %s\n", (ulong) page - count, buff);
+  fputs("\n", stdout);
+}
diff --git a/storage/maria/ma_extra.c b/storage/maria/ma_extra.c
index 7a30b613ea5..d39abc1a68e 100644
--- a/storage/maria/ma_extra.c
+++ b/storage/maria/ma_extra.c
@@ -50,7 +50,7 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
 
   switch (function) {
   case HA_EXTRA_RESET_STATE:		/* Reset state (don't free buffers) */
-    info->lastinx= 0;			/* Use first index as def */
+    info->lastinx= ~0;			/* Detect index changes */
     info->last_search_keypage= info->cur_row.lastpos= HA_OFFSET_ERROR;
     info->page_changed= 1;
 					/* Next/prev gives first/last */
@@ -553,7 +553,7 @@ int maria_reset(MARIA_HA *info)
 #endif
   info->opt_flag&= ~(KEY_READ_USED | REMEMBER_OLD_POS);
   info->quick_mode= 0;
-  info->lastinx= 0;			/* Use first index as def */
+  info->lastinx= ~0;			/* detect index changes */
   info->last_search_keypage= info->cur_row.lastpos= HA_OFFSET_ERROR;
   info->page_changed= 1;
   info->update= ((info->update & HA_STATE_CHANGED) | HA_STATE_NEXT_FOUND |
diff --git a/storage/maria/ma_ft_boolean_search.c b/storage/maria/ma_ft_boolean_search.c
index 0783f679843..1b153f1a76f 100644
--- a/storage/maria/ma_ft_boolean_search.c
+++ b/storage/maria/ma_ft_boolean_search.c
@@ -356,7 +356,7 @@ static int _ft2_search(FTB *ftb, FTB_WORD *ftbw, my_bool init_search)
   {
     ftbw->key_root=info->s->state.key_root[ftb->keynr];
     ftbw->keyinfo=info->s->keyinfo+ftb->keynr;
-    key.keyinfo= ftbw->keyinfo;
+    info->last_key.keyinfo= key.keyinfo= ftbw->keyinfo;
     key.data= ftbw->word;
     key.data_length= ftbw->len;
     key.ref_length= 0;
@@ -380,7 +380,7 @@ static int _ft2_search(FTB *ftb, FTB_WORD *ftbw, my_bool init_search)
                    max_docid);
     }
 
-    key.keyinfo= ftbw->keyinfo;
+    info->last_key.keyinfo= key.keyinfo= ftbw->keyinfo;
     key.data= lastkey_buf;
     key.data_length= USE_WHOLE_KEY;
     key.ref_length= 0;
diff --git a/storage/maria/ma_open.c b/storage/maria/ma_open.c
index ae3a3918eee..9d3e75cf4dc 100644
--- a/storage/maria/ma_open.c
+++ b/storage/maria/ma_open.c
@@ -130,6 +130,8 @@ static MARIA_HA *maria_clone_internal(MARIA_SHARE *share, const char *name,
 
   info.s=share;
   info.cur_row.lastpos= HA_OFFSET_ERROR;
+  /* Impossible first index to force initialization in _ma_check_index() */
+  info.lastinx= ~0;
   info.update= (short) (HA_STATE_NEXT_FOUND+HA_STATE_PREV_FOUND);
   info.opt_flag=READ_CHECK_USED;
   info.this_unique= (ulong) info.dfile.file; /* Uniq number in process */
diff --git a/storage/maria/ma_pagecache.c b/storage/maria/ma_pagecache.c
index 8105c6f752d..9a253fbf90c 100644
--- a/storage/maria/ma_pagecache.c
+++ b/storage/maria/ma_pagecache.c
@@ -158,6 +158,7 @@ struct st_pagecache_hash_link
 #define PCBLOCK_IN_FLUSH   16 /* block is in flush operation                 */
 #define PCBLOCK_CHANGED    32 /* block buffer contains a dirty page          */
 #define PCBLOCK_DIRECT_W   64 /* possible direct write to the block          */
+#define PCBLOCK_DEL_WRITE 128 /* should be written on delete                 */
 
 /* page status, returned by find_block */
 #define PAGE_READ               0
@@ -1215,7 +1216,7 @@ static void link_to_file_list(PAGECACHE *pagecache,
   link_changed(block, &pagecache->file_blocks[FILE_HASH(*file)]);
   if (block->status & PCBLOCK_CHANGED)
   {
-    block->status&= ~PCBLOCK_CHANGED;
+    block->status&= ~(PCBLOCK_CHANGED | PCBLOCK_DEL_WRITE);
     block->rec_lsn= LSN_MAX;
     pagecache->blocks_changed--;
     pagecache->global_blocks_changed--;
@@ -3473,6 +3474,31 @@ no_key_cache:					/* Key cache is not used */
 
 
 /*
+  @brief Set/reset flag that page always should be flushed on delete
+
+  @param pagecache      pointer to a page cache data structure
+  @param link           direct link to page (returned by read or write)
+  @param write          write on delete flag value
+
+*/
+
+void pagecache_set_write_on_delete_by_link(PAGECACHE_BLOCK_LINK *block)
+{
+  DBUG_ENTER("pagecache_set_write_on_delete_by_link");
+  DBUG_PRINT("enter", ("fd: %d block 0x%lx  %d -> TRUE",
+                       block->hash_link->file.file,
+                       (ulong) block,
+                       (int) block->status & PCBLOCK_DEL_WRITE));
+  DBUG_ASSERT(block->pins); /* should be pinned */
+  DBUG_ASSERT(block->wlocks); /* should be write locked */
+
+  block->status|= PCBLOCK_DEL_WRITE;
+
+  DBUG_VOID_RETURN;
+}
+
+
+/*
   @brief Delete page from the buffer (common part for link and file/page)
 
   @param pagecache      pointer to a page cache data structure
@@ -3501,6 +3527,7 @@ static my_bool pagecache_delete_internal(PAGECACHE *pagecache,
   }
   if (block->status & PCBLOCK_CHANGED)
   {
+    flush= (flush || (block->status & PCBLOCK_DEL_WRITE));
     if (flush)
     {
       /* The block contains a dirty page - push it out of the cache */
diff --git a/storage/maria/ma_pagecache.h b/storage/maria/ma_pagecache.h
index 821728ef374..475729bd46d 100644
--- a/storage/maria/ma_pagecache.h
+++ b/storage/maria/ma_pagecache.h
@@ -251,6 +251,7 @@ extern void pagecache_unpin(PAGECACHE *pagecache,
 extern void pagecache_unpin_by_link(PAGECACHE *pagecache,
                                     PAGECACHE_BLOCK_LINK *link,
                                     LSN lsn);
+extern void pagecache_set_write_on_delete_by_link(PAGECACHE_BLOCK_LINK *block);
 
 
 /* Results of flush operation (bit field in fact) */
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index 7a7286e26f9..d2c7bf3d381 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -28,6 +28,7 @@
 #include "trnman.h"
 #include "ma_key_recover.h"
 #include "ma_recovery_util.h"
+#include "hash.h"
 
 struct st_trn_for_recovery /* used only in the REDO phase */
 {
@@ -58,6 +59,7 @@ static ulonglong now; /**< for tracking execution time of phases */
 static int (*save_error_handler_hook)(uint, const char *,myf);
 static uint recovery_warnings; /**< count of warnings */
 static uint recovery_found_crashed_tables;
+HASH tables_to_redo;                          /* For maria_read_log */
 
 #define prototype_redo_exec_hook(R)                                          \
   static int exec_REDO_LOGREC_ ## R(const TRANSLOG_HEADER_BUFFER *rec)
@@ -184,6 +186,21 @@ static void print_preamble()
 }
 
 
+static my_bool table_is_part_of_recovery_set(LEX_STRING *file_name)
+{
+  uint offset =0;
+  if (!tables_to_redo.records)
+    return 1;                                   /* Default, recover table */
+
+  /* Skip base directory */
+  if (file_name->str[0] == '.' &&
+      (file_name->str[1] == '/' || file_name->str[1] == '\\'))
+    offset= 2;
+  /* Only recover if table is in hash */
+  return my_hash_search(&tables_to_redo, (uchar*) file_name->str + offset,
+                        file_name->length - offset) != 0;
+}
+
 /**
    @brief Recovers from the last checkpoint.
 
@@ -1643,8 +1660,8 @@ prototype_redo_exec_hook(REDO_FREE_BLOCKS)
   }
 
   buff= log_record_buffer.str;
-  if (_ma_apply_redo_free_blocks(info, current_group_end_lsn,
-                                 buff + FILEID_STORE_SIZE))
+  if (_ma_apply_redo_free_blocks(info, current_group_end_lsn, rec->lsn,
+                                 buff))
     goto end;
   error= 0;
 end:
@@ -3015,10 +3032,11 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
     page= page_korr(rec->header + FILEID_STORE_SIZE);
     llstr(page, llbuf);
     break;
+  case LOGREC_REDO_FREE_BLOCKS:
     /*
-      For REDO_FREE_BLOCKS, no need to look at dirty pages list: it does not
-      read data pages, only reads/modifies bitmap page(s) which is cheap.
+      We are checking against the dirty pages in _ma_apply_redo_free_blocks()
     */
+    break;
   default:
     break;
   }
@@ -3036,6 +3054,12 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
   share= info->s;
   tprint(tracef, ", '%s'", share->open_file_name.str);
   DBUG_ASSERT(in_redo_phase);
+  if (!table_is_part_of_recovery_set(&share->open_file_name))
+  {
+    tprint(tracef, ", skipped by user\n");
+    return NULL;
+  }
+
   if (cmp_translog_addr(rec->lsn, share->lsn_of_file_id) <= 0)
   {
     /*
@@ -3069,7 +3093,6 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
       REDO_INSERT_ROW_BLOBS will consult list by itself, as it covers several
       pages.
     */
-    tprint(tracef, " page %s", llbuf);
     if (_ma_redo_not_needed_for_page(sid, rec->lsn, page,
                                      index_page_redo_entry))
       return NULL;
@@ -3106,6 +3129,13 @@ static MARIA_HA *get_MARIA_HA_from_UNDO_record(const
   }
   share= info->s;
   tprint(tracef, ", '%s'", share->open_file_name.str);
+
+  if (!table_is_part_of_recovery_set(&share->open_file_name))
+  {
+    tprint(tracef, ", skipped by user\n");
+    return NULL;
+  }
+
   if (cmp_translog_addr(rec->lsn, share->lsn_of_file_id) <= 0)
   {
     tprint(tracef, ", table's LOGREC_FILE_ID has LSN (%lu,0x%lx) more recent"
diff --git a/storage/maria/ma_recovery.h b/storage/maria/ma_recovery.h
index 0bfcdd17d39..5b22c4fd9b2 100644
--- a/storage/maria/ma_recovery.h
+++ b/storage/maria/ma_recovery.h
@@ -30,4 +30,6 @@ int maria_apply_log(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply,
                     FILE *trace_file,
                     my_bool execute_undo_phase, my_bool skip_DDLs,
                     my_bool take_checkpoints, uint *warnings_count);
+/* Table of tables to recover */
+extern HASH tables_to_redo;
 C_MODE_END
diff --git a/storage/maria/ma_recovery_util.c b/storage/maria/ma_recovery_util.c
index 19e61daf4ef..ed9d8efc81b 100644
--- a/storage/maria/ma_recovery_util.c
+++ b/storage/maria/ma_recovery_util.c
@@ -129,16 +129,20 @@ my_bool _ma_redo_not_needed_for_page(uint16 shortid, LSN lsn,
       Next 2 bytes: table's short id
       Next 5 bytes: page number
     */
+    char llbuf[22];
     uint64 file_and_page_id=
       (((uint64)((index << 16) | shortid)) << 40) | page;
     struct st_dirty_page *dirty_page= (struct st_dirty_page *)
       hash_search(&all_dirty_pages,
                   (uchar *)&file_and_page_id, sizeof(file_and_page_id));
-    DBUG_PRINT("info", ("in dirty pages list: %d", dirty_page != NULL));
+    DBUG_PRINT("info", ("page %lld in dirty pages list: %d",
+                        (ulonglong) page,
+                        dirty_page != NULL));
     if ((dirty_page == NULL) ||
         cmp_translog_addr(lsn, dirty_page->rec_lsn) < 0)
     {
-      tprint(tracef, ", ignoring because of dirty_pages list\n");
+      tprint(tracef, ", ignoring page %s because of dirty_pages list\n",
+             llstr((ulonglong) page, llbuf));
       return TRUE;
     }
   }
diff --git a/storage/maria/ma_rkey.c b/storage/maria/ma_rkey.c
index 24b275d0ba6..e488be0ec86 100644
--- a/storage/maria/ma_rkey.c
+++ b/storage/maria/ma_rkey.c
@@ -44,7 +44,7 @@ int maria_rkey(MARIA_HA *info, uchar *buf, int inx, const uchar *key_data,
 
   info->update&= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
   info->last_key_func= search_flag;
-  keyinfo= share->keyinfo + inx;
+  keyinfo= info->last_key.keyinfo;
 
   key_buff= info->lastkey_buff+info->s->base.max_key_length;
 
diff --git a/storage/maria/ma_rnext.c b/storage/maria/ma_rnext.c
index bdba5ff3a17..570ea6a0c1d 100644
--- a/storage/maria/ma_rnext.c
+++ b/storage/maria/ma_rnext.c
@@ -31,6 +31,7 @@ int maria_rnext(MARIA_HA *info, uchar *buf, int inx)
   MARIA_SHARE *share= info->s;
   MARIA_KEYDEF *keyinfo;
   int icp_res= 1;
+  uint update_mask= HA_STATE_NEXT_FOUND;
   DBUG_ENTER("maria_rnext");
 
   if ((inx = _ma_check_index(info,inx)) < 0)
@@ -62,6 +63,20 @@ int maria_rnext(MARIA_HA *info, uchar *buf, int inx)
       error= _ma_search_first(info, keyinfo, share->state.key_root[inx]);
       break;
     }
+    /*
+      "search first" failed. This means we have no pivot for
+      "search next", or in other words MI_INFO::lastkey is
+      likely uninitialized.
+
+      Normally SQL layer would never request "search next" if
+      "search first" failed. But HANDLER may do anything.
+
+      As mi_rnext() without preceeding mi_rkey()/mi_rfirst()
+      equals to mi_rfirst(), we must restore original state
+      as if failing mi_rfirst() was not called.
+    */
+    if (error)
+      update_mask|= HA_STATE_PREV_FOUND;
   }
   else
   {
@@ -106,7 +121,7 @@ int maria_rnext(MARIA_HA *info, uchar *buf, int inx)
 
 	/* Don't clear if database-changed */
   info->update&= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
-  info->update|= HA_STATE_NEXT_FOUND;
+  info->update|= update_mask;
   
   if (icp_res == 2)
     my_errno=HA_ERR_END_OF_FILE; /* got beyond the end of scanned range */
diff --git a/storage/maria/ma_rsame.c b/storage/maria/ma_rsame.c
index 4bdbfd526ba..4bd218a01df 100644
--- a/storage/maria/ma_rsame.c
+++ b/storage/maria/ma_rsame.c
@@ -36,7 +36,7 @@ int maria_rsame(MARIA_HA *info, uchar *record, int inx)
 {
   DBUG_ENTER("maria_rsame");
 
-  if (inx != -1 && ! maria_is_key_active(info->s->state.key_map, inx))
+  if (inx >= 0 && !_ma_check_index(info, inx))
   {
     DBUG_PRINT("error", ("wrong index usage"));
     DBUG_RETURN(my_errno=HA_ERR_WRONG_INDEX);
@@ -55,8 +55,7 @@ int maria_rsame(MARIA_HA *info, uchar *record, int inx)
 
   if (inx >= 0)
   {
-    MARIA_KEYDEF *keyinfo= info->s->keyinfo + inx;
-    info->lastinx= inx;
+    MARIA_KEYDEF *keyinfo= info->last_key.keyinfo;
     (*keyinfo->make_key)(info, &info->last_key, (uint) inx,
                          info->lastkey_buff, record,
                          info->cur_row.lastpos,
diff --git a/storage/maria/ma_rt_split.c b/storage/maria/ma_rt_split.c
index 856edc60490..6f32a60c073 100644
--- a/storage/maria/ma_rt_split.c
+++ b/storage/maria/ma_rt_split.c
@@ -544,8 +544,7 @@ int maria_rtree_split_page(const MARIA_KEY *key, MARIA_PAGE *page,
   }
   DBUG_PRINT("rtree", ("split new block: %lu", (ulong) *new_page_offs));
 
-  my_afree(new_page);
-
+  my_afree(new_page_buff);
 split_err:
   my_afree(coord_buf);
   DBUG_RETURN(err_code);
diff --git a/storage/maria/ma_search.c b/storage/maria/ma_search.c
index 9f1e8e2554b..61ac1c6c23b 100644
--- a/storage/maria/ma_search.c
+++ b/storage/maria/ma_search.c
@@ -38,6 +38,8 @@ int _ma_check_index(MARIA_HA *info, int inx)
   if (info->lastinx != inx)             /* Index changed */
   {
     info->lastinx = inx;
+    info->last_key.keyinfo= info->s->keyinfo + inx;
+    info->last_key.flag= 0;
     info->page_changed=1;
     info->update= ((info->update & (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED)) |
                    HA_STATE_NEXT_FOUND | HA_STATE_PREV_FOUND);
@@ -180,7 +182,6 @@ static int _ma_search_no_save(register MARIA_HA *info, MARIA_KEY *key,
     }
   }
 
-  info->last_key.keyinfo= keyinfo;
   if ((nextflag & (SEARCH_SMALLER | SEARCH_LAST)) && flag != 0)
   {
     uint not_used[2];
@@ -1696,7 +1697,7 @@ int _ma_search_next(register MARIA_HA *info, MARIA_KEY *key,
   }
 
   tmp_key.data=   lastkey;
-  info->last_key.keyinfo= tmp_key.keyinfo= keyinfo;
+  tmp_key.keyinfo= keyinfo;
 
   if (nextflag & SEARCH_BIGGER)                                 /* Next key */
   {
@@ -1778,8 +1779,6 @@ int _ma_search_first(MARIA_HA *info, MARIA_KEYDEF *keyinfo,
     first_pos= page.buff + share->keypage_header + page.node;
   } while ((pos= _ma_kpos(page.node, first_pos)) != HA_OFFSET_ERROR);
 
-  info->last_key.keyinfo= keyinfo;
-
   if (!(*keyinfo->get_key)(&info->last_key, page.flag, page.node, &first_pos))
     DBUG_RETURN(-1);                            /* Crashed */
 
@@ -1830,8 +1829,6 @@ int _ma_search_last(MARIA_HA *info, MARIA_KEYDEF *keyinfo,
     end_of_page= page.buff + page.size;
   } while ((pos= _ma_kpos(page.node, end_of_page)) != HA_OFFSET_ERROR);
 
-  info->last_key.keyinfo= keyinfo;
-
   if (!_ma_get_last_key(&info->last_key, &page, end_of_page))
     DBUG_RETURN(-1);
   info->cur_row.lastpos= _ma_row_pos_from_key(&info->last_key);
diff --git a/storage/maria/ma_sort.c b/storage/maria/ma_sort.c
index f7f79f90cf0..a36bb980145 100644
--- a/storage/maria/ma_sort.c
+++ b/storage/maria/ma_sort.c
@@ -275,12 +275,13 @@ static ha_rows find_all_keys(MARIA_SORT_PARAM *info, uint keys,
   idx=error=0;
   sort_keys[0]= (uchar*) (sort_keys+keys);
 
+  info->sort_info->info->in_check_table= 1;
   while (!(error=(*info->key_read)(info,sort_keys[idx])))
   {
     if (info->real_key_length > info->key_length)
     {
       if (write_key(info,sort_keys[idx],tempfile_for_exceptions))
-        DBUG_RETURN(HA_POS_ERROR);		/* purecov: inspected */
+        goto err;                             /* purecov: inspected */
       continue;
     }
 
@@ -289,7 +290,7 @@ static ha_rows find_all_keys(MARIA_SORT_PARAM *info, uint keys,
       if (info->write_keys(info,sort_keys,idx-1,
                            (BUFFPEK *)alloc_dynamic(buffpek),
                            tempfile))
-      DBUG_RETURN(HA_POS_ERROR);		/* purecov: inspected */
+        goto err;                             /* purecov: inspected */
 
       sort_keys[0]=(uchar*) (sort_keys+keys);
       memcpy(sort_keys[0],sort_keys[idx-1],(size_t) info->key_length);
@@ -298,18 +299,23 @@ static ha_rows find_all_keys(MARIA_SORT_PARAM *info, uint keys,
     sort_keys[idx]=sort_keys[idx-1]+info->key_length;
   }
   if (error > 0)
-    DBUG_RETURN(HA_POS_ERROR);		/* Aborted by get_key */ /* purecov: inspected */
+    goto err;                             /* purecov: inspected */
   if (buffpek->elements)
   {
     if (info->write_keys(info,sort_keys,idx,(BUFFPEK *)alloc_dynamic(buffpek),
                          tempfile))
-      DBUG_RETURN(HA_POS_ERROR);		/* purecov: inspected */
+      goto err;                         /* purecov: inspected */      
     *maxbuffer=buffpek->elements-1;
   }
   else
     *maxbuffer=0;
 
+  info->sort_info->info->in_check_table= 0;
   DBUG_RETURN((*maxbuffer)*(keys-1)+idx);
+
+err:
+  info->sort_info->info->in_check_table= 0;   /* purecov: inspected */
+  DBUG_RETURN(HA_POS_ERROR);                  /* purecov: inspected */
 } /* find_all_keys */
 
 
diff --git a/storage/maria/ma_unique.c b/storage/maria/ma_unique.c
index a90578c2162..6090c624f36 100644
--- a/storage/maria/ma_unique.c
+++ b/storage/maria/ma_unique.c
@@ -43,6 +43,7 @@ my_bool _ma_check_unique(MARIA_HA *info, MARIA_UNIQUEDEF *def, uchar *record,
 
   /* The above changed info->lastkey_buff2. Inform maria_rnext_same(). */
   info->update&= ~HA_STATE_RNEXT_SAME;
+  info->last_key.keyinfo= keyinfo;
 
   DBUG_ASSERT(key.data_length == MARIA_UNIQUE_HASH_LENGTH);
   if (_ma_search(info, &key, SEARCH_FIND, info->s->state.key_root[def->key]))
diff --git a/storage/maria/ma_write.c b/storage/maria/ma_write.c
index 02eeec754ee..49f86fe291b 100644
--- a/storage/maria/ma_write.c
+++ b/storage/maria/ma_write.c
@@ -22,8 +22,6 @@
 #include "ma_key_recover.h"
 #include "ma_blockrec.h"
 
-#define MAX_POINTER_LENGTH 8
-
 	/* Functions declared in this file */
 
 static int w_search(MARIA_HA *info, uint32 comp_flag,
@@ -802,7 +800,7 @@ int _ma_insert(register MARIA_HA *info, MARIA_KEY *key,
 #endif
   if (t_length > 0)
   {
-    if (t_length >= keyinfo->maxlength*2+MAX_POINTER_LENGTH)
+    if (t_length >= keyinfo->maxlength*2+MARIA_INDEX_OVERHEAD_SIZE)
     {
       my_errno=HA_ERR_CRASHED;
       DBUG_RETURN(-1);
@@ -811,7 +809,7 @@ int _ma_insert(register MARIA_HA *info, MARIA_KEY *key,
   }
   else
   {
-    if (-t_length >= keyinfo->maxlength*2+MAX_POINTER_LENGTH)
+    if (-t_length >= keyinfo->maxlength*2+MARIA_INDEX_OVERHEAD_SIZE)
     {
       my_errno=HA_ERR_CRASHED;
       DBUG_RETURN(-1);
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index c39aeaabaa6..4195be4a53a 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -152,11 +152,13 @@ typedef struct st_maria_state_info
 #define MARIA_COLUMNDEF_SIZE	(2*7+1+1+4)
 #define MARIA_BASE_INFO_SIZE	(MY_UUID_SIZE + 5*8 + 6*4 + 11*2 + 6 + 5*2 + 1 + 16)
 #define MARIA_INDEX_BLOCK_MARGIN 16	/* Safety margin for .MYI tables */
+#define MARIA_MAX_POINTER_LENGTH 7	/* Node pointer */
 /* Internal management bytes needed to store 2 transid/key on an index page */
 #define MARIA_MAX_PACK_TRANSID_SIZE   (TRANSID_SIZE+1)
 #define MARIA_TRANSID_PACK_OFFSET     (256- TRANSID_SIZE - 1)
 #define MARIA_MIN_TRANSID_PACK_OFFSET (MARIA_TRANSID_PACK_OFFSET-TRANSID_SIZE)
-#define MARIA_INDEX_OVERHEAD_SIZE     (MARIA_MAX_PACK_TRANSID_SIZE * 2)
+#define MARIA_INDEX_OVERHEAD_SIZE     (MARIA_MAX_PACK_TRANSID_SIZE * 2 + \
+                                       MARIA_MAX_POINTER_LENGTH)
 #define MARIA_DELETE_KEY_NR  255	/* keynr for deleted blocks */
 
 /*
@@ -571,6 +573,7 @@ struct st_maria_handler
   my_bool was_locked;			/* Was locked in panic */
   my_bool append_insert_at_end;		/* Set if concurrent insert */
   my_bool quick_mode;
+  my_bool in_check_table;                /* We are running check tables */
   /* Marker if key_del_changed */
   /* If info->keyread_buff can't be used for rnext */
   my_bool page_changed;
diff --git a/storage/maria/maria_read_log.c b/storage/maria/maria_read_log.c
index de45eb0bcb6..3657e2db332 100644
--- a/storage/maria/maria_read_log.c
+++ b/storage/maria/maria_read_log.c
@@ -213,6 +213,9 @@ static struct my_option my_long_options[] =
   {"silent", 's', "Print less information during apply/undo phase",
    &opt_silent, &opt_silent, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"tables-to-redo", 'T',
+   "List of tables sepearated with , that we should apply REDO on. Use this if you only want to recover some tables",
+   0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"verbose", 'v', "Print more information during apply/undo phase",
    &maria_recovery_verbose, &maria_recovery_verbose, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
@@ -245,7 +248,7 @@ static void print_version(void)
 static void usage(void)
 {
   print_version();
-  puts("Copyright (C) 2007 MySQL AB");
+  puts("Copyright (C) 2007 MySQL AB, 2009-2011 Monty Program Ab");
   puts("This software comes with ABSOLUTELY NO WARRANTY. This is free software,");
   puts("and you are welcome to modify and redistribute it under the GPL license\n");
 
@@ -266,10 +269,18 @@ static void usage(void)
 
 #include <help_end.h>
 
+static uchar* my_hash_get_string(const uchar *record, size_t *length,
+                                my_bool first __attribute__ ((unused)))
+{
+  *length= (size_t) (strcend((const char*) record,',')- (const char*) record);
+  return (uchar*) record;
+}
+
+
 static my_bool
 get_one_option(int optid __attribute__((unused)),
                const struct my_option *opt __attribute__((unused)),
-               char *argument __attribute__((unused)))
+               char *argument)
 {
   switch (optid) {
   case '?':
@@ -278,6 +289,23 @@ get_one_option(int optid __attribute__((unused)),
   case 'V':
     print_version();
     exit(0);
+  case 'T':
+  {
+    char *pos;
+    if (!my_hash_inited(&tables_to_redo))
+    {
+      my_hash_init2(&tables_to_redo, 16, &my_charset_bin,
+                    16, 0, 0, my_hash_get_string, 0, HASH_UNIQUE);
+    }
+    do
+    {
+      pos= strcend(argument, ',');
+      if (pos != argument)                      /* Skip empty strings */
+        my_hash_insert(&tables_to_redo, (uchar*) argument);
+      argument= pos+1;
+    } while (*(pos++));
+    break;
+  }
 #ifndef DBUG_OFF
   case '#':
     DBUG_SET_INITIAL(argument ? argument : default_dbug_option);
@@ -290,6 +318,7 @@ get_one_option(int optid __attribute__((unused)),
 static void get_options(int *argc,char ***argv)
 {
   int ho_error;
+  my_bool need_help= 0;
 
   if ((ho_error=handle_options(argc, argv, my_long_options, get_one_option)))
     exit(ho_error);
@@ -297,8 +326,23 @@ static void get_options(int *argc,char ***argv)
   if (!opt_apply)
     opt_apply_undo= FALSE;
 
-  if (((opt_display_only + opt_apply) != 1) || (*argc > 0))
+  if (*argc > 0)
+  {
+    need_help= 1;
+    fprintf(stderr, "Too many arguments given\n");
+  }
+  if ((opt_display_only + opt_apply) != 1)
+  {
+    need_help= 1;
+    fprintf(stderr,
+            "You must use one and only one of the options 'display-only' or "
+            "'apply'\n");
+  }
+
+  if (need_help)
   {
+    fflush(stderr);
+    need_help =1;
     usage();
     exit(1);
   }
diff --git a/storage/myisam/ha_myisam.cc b/storage/myisam/ha_myisam.cc
index 7ba36f3b937..b2fad816eba 100644
--- a/storage/myisam/ha_myisam.cc
+++ b/storage/myisam/ha_myisam.cc
@@ -545,6 +545,7 @@ ha_myisam::ha_myisam(handlerton *hton, TABLE_SHARE *table_arg)
   :handler(hton, table_arg), file(0),
   int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
                   HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
+                  HA_CAN_VIRTUAL_COLUMNS |
                   HA_DUPLICATE_POS | HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
                   HA_FILE_BASED | HA_CAN_GEOMETRY | HA_NO_TRANSACTIONS |
                   HA_CAN_INSERT_DELAYED | HA_CAN_BIT_FIELD | HA_CAN_RTREEKEYS |
diff --git a/storage/myisam/ha_myisam.h b/storage/myisam/ha_myisam.h
index 210dc7bc745..b9700e0ef90 100644
--- a/storage/myisam/ha_myisam.h
+++ b/storage/myisam/ha_myisam.h
@@ -141,7 +141,6 @@ class ha_myisam: public handler
   int assign_to_keycache(THD* thd, HA_CHECK_OPT* check_opt);
   int preload_keys(THD* thd, HA_CHECK_OPT* check_opt);
   bool check_if_incompatible_data(HA_CREATE_INFO *info, uint table_changes);
-  bool check_if_supported_virtual_columns(void) { return TRUE;}
 #ifdef HAVE_REPLICATION
   int dump(THD* thd, int fd);
   int net_read_dump(NET* net);
diff --git a/storage/myisammrg/ha_myisammrg.cc b/storage/myisammrg/ha_myisammrg.cc
index c61d65572cb..ff2d01e09c0 100644
--- a/storage/myisammrg/ha_myisammrg.cc
+++ b/storage/myisammrg/ha_myisammrg.cc
@@ -1057,9 +1057,9 @@ THR_LOCK_DATA **ha_myisammrg::store_lock(THD *thd,
 
   /*
     When MERGE table is open, but not yet attached, other threads
-    could flush it, which means call mysql_lock_abort_for_thread()
+    could flush it, which means calling mysql_lock_abort_for_thread()
     on this threads TABLE. 'children_attached' is FALSE in this
-    situaton. Since the table is not locked, return no lock data.
+    situation. Since the table is not locked, return no lock data.
   */
   if (!this->file->children_attached)
     goto end; /* purecov: tested */
diff --git a/storage/ndb/include/ndbapi/NdbError.hpp b/storage/ndb/include/ndbapi/NdbError.hpp
index aa27caf78f9..b752e578bc1 100644
--- a/storage/ndb/include/ndbapi/NdbError.hpp
+++ b/storage/ndb/include/ndbapi/NdbError.hpp
@@ -66,7 +66,7 @@ struct NdbError {
     
     /**
      * The error code indicates a permanent error.<br>
-     * (Includes classificatons: NdbError::PermanentError, 
+     * (Includes classifications: NdbError::PermanentError, 
      *  NdbError::ApplicationError, NdbError::NoDataFound,
      *  NdbError::ConstraintViolation, NdbError::SchemaError,
      *  NdbError::UserDefinedError, NdbError::InternalError, and, 
diff --git a/storage/ndb/include/util/File.hpp b/storage/ndb/include/util/File.hpp
index b9d348683ec..bbddc24583a 100644
--- a/storage/ndb/include/util/File.hpp
+++ b/storage/ndb/include/util/File.hpp
@@ -31,7 +31,7 @@ public:
    * Returns time for last contents modification of a file.
    *
    * @param aFileName a filename to check.
-   * @return the time for last contents modificaton of the file.
+   * @return the time for last contents modification of the file.
    */
   static time_t mtime(const char* aFileName);      
 
diff --git a/storage/sphinx/Makefile.am b/storage/sphinx/Makefile.am
index ce4324ae5cb..67d01d08484 100644
--- a/storage/sphinx/Makefile.am
+++ b/storage/sphinx/Makefile.am
@@ -49,6 +49,6 @@ libsphinx_la_CXXFLAGS =	$(AM_CXXFLAGS)
 libsphinx_la_CFLAGS =	$(AM_CFLAGS)
 libsphinx_la_SOURCES=	ha_sphinx.cc
 
-EXTRA_DIST =		CMakeLists.txt
+EXTRA_DIST =		CMakeLists.txt plug.in
 # Don't update the files from bitkeeper
 %::SCCS/s.%
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index b630cecf86c..42b818c41e1 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -1477,7 +1477,7 @@ UNIV_INTERN
 ha_innobase::ha_innobase(handlerton *hton, TABLE_SHARE *table_arg)
   :handler(hton, table_arg),
   int_table_flags(HA_REC_NOT_IN_SEQ |
-		  HA_NULL_IN_KEY |
+		  HA_NULL_IN_KEY | HA_CAN_VIRTUAL_COLUMNS |
 		  HA_CAN_INDEX_BLOBS |
 		  HA_CAN_SQL_HANDLER |
 		  HA_PRIMARY_KEY_REQUIRED_FOR_POSITION |
diff --git a/storage/xtradb/trx/trx0i_s.c b/storage/xtradb/trx/trx0i_s.c
index 5bc8302d0c0..c0b9a73a68c 100644
--- a/storage/xtradb/trx/trx0i_s.c
+++ b/storage/xtradb/trx/trx0i_s.c
@@ -157,10 +157,6 @@ struct trx_i_s_cache_struct {
 	ullint		last_read;	/*!< last time the cache was read;
 					measured in microseconds since
 					epoch */
-	mutex_t		last_read_mutex;/*!< mutex protecting the
-					last_read member - it is updated
-					inside a shared lock of the
-					rw_lock member */
 	i_s_table_cache_t innodb_trx;	/*!< innodb_trx table */
 	i_s_table_cache_t innodb_locks;	/*!< innodb_locks table */
 	i_s_table_cache_t innodb_lock_waits;/*!< innodb_lock_waits table */
@@ -1101,13 +1097,6 @@ can_cache_be_updated(
 {
 	ullint	now;
 
-	/* Here we read cache->last_read without acquiring its mutex
-	because last_read is only updated when a shared rw lock on the
-	whole cache is being held (see trx_i_s_cache_end_read()) and
-	we are currently holding an exclusive rw lock on the cache.
-	So it is not possible for last_read to be updated while we are
-	reading it. */
-
 #ifdef UNIV_SYNC_DEBUG
 	ut_a(rw_lock_own(&cache->rw_lock, RW_LOCK_EX));
 #endif
@@ -1205,6 +1194,12 @@ trx_i_s_possibly_fetch_data_into_cache(
 /*===================================*/
 	trx_i_s_cache_t*	cache)	/*!< in/out: cache */
 {
+	ullint	now;
+
+#ifdef UNIV_SYNC_DEBUG
+	ut_a(rw_lock_own(&cache->rw_lock, RW_LOCK_EX));
+#endif
+
 	if (!can_cache_be_updated(cache)) {
 
 		return(1);
@@ -1217,6 +1212,10 @@ trx_i_s_possibly_fetch_data_into_cache(
 
 	mutex_exit(&kernel_mutex);
 
+	/* update cache last read time */
+	now = ut_time_us(NULL);
+	cache->last_read = now;
+
 	return(0);
 }
 
@@ -1247,16 +1246,12 @@ trx_i_s_cache_init(
 	release kernel_mutex
 	release trx_i_s_cache_t::rw_lock
 	acquire trx_i_s_cache_t::rw_lock, S
-	acquire trx_i_s_cache_t::last_read_mutex
-	release trx_i_s_cache_t::last_read_mutex
 	release trx_i_s_cache_t::rw_lock */
 
 	rw_lock_create(&cache->rw_lock, SYNC_TRX_I_S_RWLOCK);
 
 	cache->last_read = 0;
 
-	mutex_create(&cache->last_read_mutex, SYNC_TRX_I_S_LAST_READ);
-
 	table_cache_init(&cache->innodb_trx, sizeof(i_s_trx_row_t));
 	table_cache_init(&cache->innodb_locks, sizeof(i_s_locks_row_t));
 	table_cache_init(&cache->innodb_lock_waits,
@@ -1307,18 +1302,10 @@ trx_i_s_cache_end_read(
 /*===================*/
 	trx_i_s_cache_t*	cache)	/*!< in: cache */
 {
-	ullint	now;
-
 #ifdef UNIV_SYNC_DEBUG
 	ut_a(rw_lock_own(&cache->rw_lock, RW_LOCK_SHARED));
 #endif
 
-	/* update cache last read time */
-	now = ut_time_us(NULL);
-	mutex_enter(&cache->last_read_mutex);
-	cache->last_read = now;
-	mutex_exit(&cache->last_read_mutex);
-
 	rw_lock_s_unlock(&cache->rw_lock);
 }
 
diff --git a/storage/xtradb/trx/trx0sys.c b/storage/xtradb/trx/trx0sys.c
index 566f7c95793..a9fb5bb38d8 100644
--- a/storage/xtradb/trx/trx0sys.c
+++ b/storage/xtradb/trx/trx0sys.c
@@ -1667,7 +1667,7 @@ trx_sys_print_mysql_binlog_offset_from_page(
 
 
 /* THESE ARE COPIED FROM NON-HOTBACKUP PART OF THE INNODB SOURCE TREE
-   (This code duplicaton should be fixed at some point!)
+   (This code duplication should be fixed at some point!)
 */
 
 #define	TRX_SYS_SPACE	0	/* the SYSTEM tablespace */