Merge in new handler and handlersocket code into 5.3 main

45 files changed, 398 insertions, 142 deletions

diff --git a/storage/federated/ha_federated.cc b/storage/federated/ha_federated.cc
index fcc178b09a6..34509e04799 100644
--- a/storage/federated/ha_federated.cc
+++ b/storage/federated/ha_federated.cc
@@ -2453,7 +2453,6 @@ int ha_federated::index_init(uint keynr, bool sorted)
 {
   DBUG_ENTER("ha_federated::index_init");
   DBUG_PRINT("info", ("table: '%s'  key: %u", table->s->table_name.str, keynr));
-  active_index= keynr;
   DBUG_RETURN(0);
 }
 
@@ -2589,7 +2588,6 @@ int ha_federated::index_end(void)
 {
   DBUG_ENTER("ha_federated::index_end");
   free_result();
-  active_index= MAX_KEY;
   DBUG_RETURN(0);
 }
 
diff --git a/storage/heap/ha_heap.cc b/storage/heap/ha_heap.cc
index 2d2c23a2ed6..6bb6936de9d 100644
--- a/storage/heap/ha_heap.cc
+++ b/storage/heap/ha_heap.cc
@@ -183,6 +183,19 @@ void ha_heap::set_keys_for_scanning(void)
 }
 
 
+int ha_heap::can_continue_handler_scan()
+{
+  int error= 0;
+  if ((file->key_version != file->s->key_version && inited == INDEX) ||
+      (file->file_version != file->s->file_version && inited == RND))
+  {
+    /* Data changed, not safe to do index or rnd scan */
+    error= HA_ERR_RECORD_CHANGED;
+  }
+  return error;
+}
+
+
 void ha_heap::update_key_stats()
 {
   for (uint i= 0; i < table->s->keys; i++)
diff --git a/storage/heap/ha_heap.h b/storage/heap/ha_heap.h
index 22722129f4c..49aa91a6caa 100644
--- a/storage/heap/ha_heap.h
+++ b/storage/heap/ha_heap.h
@@ -1,4 +1,4 @@
-/* Copyright (C) 2000-2006 MySQL AB
+/* Copyright (C) 2000-2006 MySQL AB, 2009-2011 Monty Program Ab
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -52,6 +52,7 @@ public:
   {
     return (HA_FAST_KEY_READ | HA_NO_BLOBS | HA_NULL_IN_KEY |
             HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
+            HA_CAN_SQL_HANDLER |
             HA_REC_NOT_IN_SEQ | HA_CAN_INSERT_DELAYED | HA_NO_TRANSACTIONS |
             HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT);
   }
@@ -93,6 +94,7 @@ public:
   int rnd_next(uchar *buf);
   int rnd_pos(uchar * buf, uchar *pos);
   void position(const uchar *record);
+  int can_continue_handler_scan();
   int info(uint);
   int extra(enum ha_extra_function operation);
   int reset();
diff --git a/storage/heap/hp_clear.c b/storage/heap/hp_clear.c
index babfcbd6f41..b999f1ff5e8 100644
--- a/storage/heap/hp_clear.c
+++ b/storage/heap/hp_clear.c
@@ -40,6 +40,8 @@ void hp_clear(HP_SHARE *info)
   info->blength=1;
   info->changed=0;
   info->del_link=0;
+  info->key_version++;
+  info->file_version++;
   DBUG_VOID_RETURN;
 }
 
diff --git a/storage/heap/hp_delete.c b/storage/heap/hp_delete.c
index 9e9e28da335..1af9cf26f8c 100644
--- a/storage/heap/hp_delete.c
+++ b/storage/heap/hp_delete.c
@@ -47,6 +47,7 @@ int heap_delete(HP_INFO *info, const uchar *record)
   share->del_link=pos;
   pos[share->reclength]=0;		/* Record deleted */
   share->deleted++;
+  share->key_version++;
   info->current_hash_ptr=0;
 #if !defined(DBUG_OFF) && defined(EXTRA_HEAP_DEBUG)
   DBUG_EXECUTE("check_heap",heap_check_heap(info, 0););
diff --git a/storage/heap/hp_rfirst.c b/storage/heap/hp_rfirst.c
index d0d2ec9b506..8e562983b02 100644
--- a/storage/heap/hp_rfirst.c
+++ b/storage/heap/hp_rfirst.c
@@ -24,6 +24,8 @@ int heap_rfirst(HP_INFO *info, uchar *record, int inx)
   
   DBUG_ENTER("heap_rfirst");
   info->lastinx= inx;
+  info->key_version= info->s->key_version;
+
   if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
   {
     uchar *pos;
@@ -57,15 +59,8 @@ int heap_rfirst(HP_INFO *info, uchar *record, int inx)
   }
   else
   {
-    if (!(info->s->records))
-    {
-      my_errno=HA_ERR_END_OF_FILE;
-      DBUG_RETURN(my_errno);
-    }
-    DBUG_ASSERT(0); /* TODO fix it */
-    info->current_record=0;
-    info->current_hash_ptr=0;
-    info->update=HA_STATE_PREV_FOUND;
-    DBUG_RETURN(heap_rnext(info,record));
+    /* We can't scan a non existing key value with hash index */
+    my_errno= HA_ERR_WRONG_COMMAND;
+    DBUG_RETURN(my_errno);
   }
 }
diff --git a/storage/heap/hp_rkey.c b/storage/heap/hp_rkey.c
index 27d1114770e..c2edf63f6f2 100644
--- a/storage/heap/hp_rkey.c
+++ b/storage/heap/hp_rkey.c
@@ -30,6 +30,7 @@ int heap_rkey(HP_INFO *info, uchar *record, int inx, const uchar *key,
   }
   info->lastinx= inx;
   info->current_record= (ulong) ~0L;		/* For heap_rrnd() */
+  info->key_version= info->s->key_version;
 
   if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
   {
diff --git a/storage/heap/hp_rlast.c b/storage/heap/hp_rlast.c
index 45ad7c21f49..0710401e5a5 100644
--- a/storage/heap/hp_rlast.c
+++ b/storage/heap/hp_rlast.c
@@ -25,6 +25,7 @@ int heap_rlast(HP_INFO *info, uchar *record, int inx)
 
   DBUG_ENTER("heap_rlast");
   info->lastinx= inx;
+  info->key_version= info->s->key_version;
   if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
   {
     uchar *pos;
@@ -47,9 +48,8 @@ int heap_rlast(HP_INFO *info, uchar *record, int inx)
   }
   else
   {
-    info->current_ptr=0;
-    info->current_hash_ptr=0;
-    info->update=HA_STATE_NEXT_FOUND;
-    DBUG_RETURN(heap_rprev(info,record));
+    /* We can't scan a non existing key value with hash index */
+    my_errno= HA_ERR_WRONG_COMMAND;
+    DBUG_RETURN(my_errno);
   }
 }
diff --git a/storage/heap/hp_rnext.c b/storage/heap/hp_rnext.c
index 3d715f4e6d3..7a654850e0e 100644
--- a/storage/heap/hp_rnext.c
+++ b/storage/heap/hp_rnext.c
@@ -70,6 +70,7 @@ int heap_rnext(HP_INFO *info, uchar *record)
       custom_arg.keyseg = keyinfo->seg;
       custom_arg.key_length = info->lastkey_len;
       custom_arg.search_flag = SEARCH_SAME | SEARCH_FIND;
+      info->last_find_flag= HA_READ_KEY_OR_NEXT;
       pos = tree_search_key(&keyinfo->rb_tree, info->lastkey, info->parents, 
                            &info->last_pos, info->last_find_flag, &custom_arg);
     }
diff --git a/storage/heap/hp_rprev.c b/storage/heap/hp_rprev.c
index 63bfffffba9..1d71c20eef4 100644
--- a/storage/heap/hp_rprev.c
+++ b/storage/heap/hp_rprev.c
@@ -41,6 +41,7 @@ int heap_rprev(HP_INFO *info, uchar *record)
       custom_arg.keyseg = keyinfo->seg;
       custom_arg.key_length = keyinfo->length;
       custom_arg.search_flag = SEARCH_SAME;
+      info->last_find_flag= HA_READ_KEY_OR_PREV;
       pos = tree_search_key(&keyinfo->rb_tree, info->lastkey, info->parents, 
                            &info->last_pos, info->last_find_flag, &custom_arg);
     }
diff --git a/storage/heap/hp_scan.c b/storage/heap/hp_scan.c
index e8913e92c86..397dd8b54d4 100644
--- a/storage/heap/hp_scan.c
+++ b/storage/heap/hp_scan.c
@@ -31,6 +31,8 @@ int heap_scan_init(register HP_INFO *info)
   info->current_record= (ulong) ~0L;		/* No current record */
   info->update=0;
   info->next_block=0;
+  info->key_version= info->s->key_version;
+  info->file_version= info->s->file_version;
   DBUG_RETURN(0);
 }
 
diff --git a/storage/heap/hp_update.c b/storage/heap/hp_update.c
index 7f469af3c96..ab831382325 100644
--- a/storage/heap/hp_update.c
+++ b/storage/heap/hp_update.c
@@ -21,7 +21,7 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
 {
   HP_KEYDEF *keydef, *end, *p_lastinx;
   uchar *pos;
-  my_bool auto_key_changed= 0;
+  my_bool auto_key_changed= 0, key_changed= 0;
   HP_SHARE *share= info->s;
   DBUG_ENTER("heap_update");
 
@@ -54,6 +54,8 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
 #endif
   if (auto_key_changed)
     heap_update_auto_increment(info, heap_new);
+  if (key_changed)
+    share->key_version++;
   DBUG_RETURN(0);
 
  err:
diff --git a/storage/heap/hp_write.c b/storage/heap/hp_write.c
index 4e8fa7e3580..bf27503de9b 100644
--- a/storage/heap/hp_write.c
+++ b/storage/heap/hp_write.c
@@ -56,6 +56,7 @@ int heap_write(HP_INFO *info, const uchar *record)
   pos[share->reclength]=1;		/* Mark record as not deleted */
   if (++share->records == share->blength)
     share->blength+= share->blength;
+  info->s->key_version++;
   info->current_ptr=pos;
   info->current_hash_ptr=0;
   info->update|=HA_STATE_AKTIV;
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/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 406def92808..2cf981d5076 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -228,7 +228,7 @@ static MYSQL_SYSVAR_ULONGLONG(pagecache_buffer_size, pagecache_buffer_size,
        "The size of the buffer used for index blocks for Aria tables. "
        "Increase this to get better index handling (for all reads and "
        "multiple writes) to as much as you can afford.", 0, 0,
-       KEY_CACHE_SIZE, 0, ~(ulong) 0, 1);
+       KEY_CACHE_SIZE, 8192*16L, ~(ulong) 0, 1);
 
 static MYSQL_SYSVAR_ULONG(pagecache_division_limit, pagecache_division_limit,
        PLUGIN_VAR_RQCMDARG,
@@ -826,7 +826,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)
 {}
@@ -1078,8 +1078,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();
@@ -2086,6 +2084,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 &
@@ -2106,7 +2105,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;
@@ -2157,7 +2155,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);
@@ -2167,7 +2164,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);
 }
 
@@ -2191,7 +2187,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;
@@ -2202,7 +2197,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;
@@ -2214,7 +2208,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;
@@ -2225,7 +2218,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;
@@ -2235,7 +2227,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;
@@ -2245,7 +2236,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;
@@ -2255,7 +2245,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;
@@ -2268,7 +2257,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
@@ -2320,7 +2308,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;
@@ -2342,7 +2329,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;
@@ -3367,6 +3353,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 b7bca730051..4877f5c6d71 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 */
@@ -878,6 +881,7 @@ static int chk_index(HA_CHECK *param, MARIA_HA *info, MARIA_KEYDEF *keyinfo,
   }
 
   info->last_key.keyinfo= tmp_key.keyinfo= keyinfo;
+  info->lastinx= ~0;                            /* Safety */
   tmp_key.data= tmp_key_buff;
   for ( ;; )
   {
@@ -1133,6 +1137,7 @@ static int check_keys_in_record(HA_CHECK *param, MARIA_HA *info, int extend,
       {
         (*keyinfo->make_key)(info, &key, keynr, info->lastkey_buff, record,
                              start_recpos, 0);
+        info->last_key.keyinfo= key.keyinfo;
         if (extend)
         {
           /* We don't need to lock the key tree here as we don't allow
@@ -1844,6 +1849,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 */
@@ -2036,6 +2043,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);
@@ -2057,6 +2066,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;
 
@@ -2185,12 +2196,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));
@@ -2631,6 +2647,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)))
   {
@@ -2798,6 +2815,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)
   {
@@ -3248,6 +3266,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))
@@ -3272,7 +3291,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);
@@ -3280,7 +3299,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);
@@ -3292,7 +3311,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;
         }
       }
     }
@@ -3306,10 +3325,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);
 }
 
 
@@ -4769,7 +4791,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);
@@ -6463,6 +6485,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);
 }
@@ -6818,3 +6843,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 a2d40b1c828..2d1ed6d3405 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..13c3a24f362 100644
--- a/storage/maria/ma_ft_boolean_search.c
+++ b/storage/maria/ma_ft_boolean_search.c
@@ -356,7 +356,8 @@ 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;
+    info->lastinx= ~0;                            /* Safety */
     key.data= ftbw->word;
     key.data_length= ftbw->len;
     key.ref_length= 0;
@@ -380,7 +381,8 @@ 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;
+    info->lastinx= ~0;                            /* Safety */
     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 89d41589b83..c679fba7a12 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..42a42b65b8d 100644
--- a/storage/maria/ma_pagecache.c
+++ b/storage/maria/ma_pagecache.c
@@ -1,4 +1,4 @@
-/* Copyright (C) 2000-2008 MySQL AB
+/* Copyright (C) 2000-2008 MySQL AB, 2008-2011 Monty Program Ab
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -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
@@ -759,6 +760,8 @@ ulong init_pagecache(PAGECACHE *pagecache, size_t use_mem,
   {
     if (blocks < 8)
     {
+      my_message(ENOMEM, "Not enough memory to allocate 8 pagecache pages",
+                 MYF(0));
       my_errno= ENOMEM;
       goto err;
     }
@@ -1215,7 +1218,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 +3476,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 +3529,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 */
@@ -4187,6 +4216,7 @@ static void free_block(PAGECACHE *pagecache, PAGECACHE_BLOCK_LINK *block)
   DBUG_ASSERT(block->rlocks == 0);
   DBUG_ASSERT(block->rlocks_queue == 0);
   DBUG_ASSERT(block->pins == 0);
+  DBUG_ASSERT((block->status & ~(PCBLOCK_ERROR | PCBLOCK_READ | PCBLOCK_IN_FLUSH | PCBLOCK_CHANGED | PCBLOCK_REASSIGNED | PCBLOCK_DEL_WRITE)) == 0);
   block->status= 0;
 #ifndef DBUG_OFF
   block->type= PAGECACHE_EMPTY_PAGE;
@@ -4515,6 +4545,7 @@ static int flush_pagecache_blocks_int(PAGECACHE *pagecache,
           KEYCACHE_DBUG_ASSERT(count<= pagecache->blocks_used);
         }
       }
+      count++;    /* Allocate one extra for easy end-of-buffer test */
       /* Allocate a new buffer only if its bigger than the one we have */
       if (count > FLUSH_CACHE &&
           !(cache=
@@ -4552,22 +4583,24 @@ restart:
         DBUG_ASSERT(filter_res == FLUSH_FILTER_OK);
       }
       {
+        DBUG_ASSERT(!(block->status & PCBLOCK_IN_FLUSH));
         /*
-           Mark the block with BLOCK_IN_FLUSH in order not to let
-           other threads to use it for new pages and interfere with
-           our sequence of flushing dirty file pages
+          We care only for the blocks for which flushing was not
+          initiated by other threads as a result of page swapping
         */
-        block->status|= PCBLOCK_IN_FLUSH;
-
         if (! (block->status & PCBLOCK_IN_SWITCH))
         {
-	  /*
-	    We care only for the blocks for which flushing was not
-	    initiated by other threads as a result of page swapping
+          /*
+            Mark the block with BLOCK_IN_FLUSH in order not to let
+            other threads to use it for new pages and interfere with
+            our sequence of flushing dirty file pages
           */
+          block->status|= PCBLOCK_IN_FLUSH;
+
           reg_requests(pagecache, block, 1);
           if (type != FLUSH_IGNORE_CHANGED)
           {
+            *pos++= block;
 	    /* It's not a temporary file */
             if (pos == end)
             {
@@ -4587,7 +4620,6 @@ restart:
               */
               goto restart;
             }
-            *pos++= block;
           }
           else
           {
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 415f20cc631..4b0dde1860a 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 2fd8bf4e603..e5ee84b61c7 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;
   ICP_RESULT icp_res= ICP_MATCH;
+  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
   {
@@ -118,7 +133,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 (error || icp_res != ICP_MATCH)
   {
diff --git a/storage/maria/ma_rsame.c b/storage/maria/ma_rsame.c
index 4bdbfd526ba..a5953cd09cb 100644
--- a/storage/maria/ma_rsame.c
+++ b/storage/maria/ma_rsame.c
@@ -19,7 +19,7 @@
   Find current row with read on position or read on key
 
   @notes
-  If inx >= 0 find record using key
+  If inx >= 0 find record using key else re-read row on last position
 
   @warning
   This function is not row version safe.
@@ -29,6 +29,7 @@
   @retval 0                      Ok
   @retval HA_ERR_KEY_NOT_FOUND   Row is deleted
   @retval HA_ERR_END_OF_FILE     End of file
+  @retval HA_ERR_WRONG_INDEX	 Wrong inx argument
 */
 
 
@@ -36,10 +37,10 @@ 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) < 0)
   {
     DBUG_PRINT("error", ("wrong index usage"));
-    DBUG_RETURN(my_errno=HA_ERR_WRONG_INDEX);
+    DBUG_RETURN(my_errno);
   }
   if (info->cur_row.lastpos == HA_OFFSET_ERROR ||
       info->update & HA_STATE_DELETED)
@@ -55,8 +56,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_index.c b/storage/maria/ma_rt_index.c
index 62474dbbad8..bd1b9038ccf 100644
--- a/storage/maria/ma_rt_index.c
+++ b/storage/maria/ma_rt_index.c
@@ -134,7 +134,6 @@ static int maria_rtree_find_req(MARIA_HA *info, MARIA_KEYDEF *keyinfo,
         tmp_key.data_length= key_data_length;
 
         info->cur_row.lastpos= _ma_row_pos_from_key(&tmp_key);
-        info->last_key.keyinfo= keyinfo;
         info->last_key.data_length= key_data_length;
         info->last_key.ref_length=  share->base.rec_reflength;
         info->last_key.flag= 0;
diff --git a/storage/maria/ma_search.c b/storage/maria/ma_search.c
index 4ac6dfeb15f..df74fed842a 100644
--- a/storage/maria/ma_search.c
+++ b/storage/maria/ma_search.c
@@ -38,12 +38,18 @@ 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);
   }
-  if (info->opt_flag & WRITE_CACHE_USED && flush_io_cache(&info->rec_cache))
+  if ((info->opt_flag & WRITE_CACHE_USED) && flush_io_cache(&info->rec_cache))
+  {
+    if (unlikely(!my_errno))
+      my_errno= HA_ERR_INTERNAL_ERROR;          /* Impossible */
     return(-1);
+  }
   return(inx);
 } /* _ma_check_index */
 
@@ -95,6 +101,7 @@ int _ma_search(register MARIA_HA *info, MARIA_KEY *key, uint32 nextflag,
 
    @note
      Position to row is stored in info->lastpos
+     Last used key is stored in info->last_key
 
    @return
    @retval  0   ok (key found)
@@ -120,6 +127,7 @@ static int _ma_search_no_save(register MARIA_HA *info, MARIA_KEY *key,
                       (ulong) (pos / info->s->block_size),
                       nextflag, (ulong) info->cur_row.lastpos));
   DBUG_EXECUTE("key", _ma_print_key(DBUG_FILE, key););
+  DBUG_ASSERT(info->last_key.keyinfo == key->keyinfo);
 
   if (pos == HA_OFFSET_ERROR)
   {
@@ -184,7 +192,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];
@@ -1702,7 +1709,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 */
   {
@@ -1784,8 +1791,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 */
 
@@ -1836,8 +1841,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 c7d277486c4..6f92ca1559c 100644
--- a/storage/maria/ma_unique.c
+++ b/storage/maria/ma_unique.c
@@ -34,6 +34,7 @@ my_bool _ma_check_unique(MARIA_HA *info, MARIA_UNIQUEDEF *def, uchar *record,
   MARIA_KEYDEF *keyinfo= &info->s->keyinfo[def->key];
   uchar *key_buff= info->lastkey_buff2;
   MARIA_KEY key;
+  int error= 0;
   DBUG_ENTER("_ma_check_unique");
   DBUG_PRINT("enter",("unique_hash: %lu", (ulong) unique_hash));
 
@@ -44,13 +45,19 @@ 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;
 
+  /* Setup that unique key is active key */
+  info->last_key.keyinfo= keyinfo;
+
+  /* any key pointer in data is destroyed */
+  info->lastinx= ~0;
+
   DBUG_ASSERT(key.data_length == MARIA_UNIQUE_HASH_LENGTH);
   if (_ma_search(info, &key, SEARCH_FIND | SEARCH_SAVE_BUFF,
                  info->s->state.key_root[def->key]))
   {
     info->page_changed=1;			/* Can't optimize read next */
     info->cur_row.lastpos= lastpos;
-    DBUG_RETURN(0);				/* No matching rows */
+    goto end;
   }
 
   for (;;)
@@ -64,7 +71,8 @@ my_bool _ma_check_unique(MARIA_HA *info, MARIA_UNIQUEDEF *def, uchar *record,
       info->page_changed= 1;			/* Can't optimize read next */
       info->cur_row.lastpos= lastpos;
       DBUG_PRINT("info",("Found duplicate"));
-      DBUG_RETURN(1);				/* Found identical  */
+      error= 1;                                 /* Found identical  */
+      goto end;
     }
     DBUG_ASSERT(info->last_key.data_length == MARIA_UNIQUE_HASH_LENGTH);
     if (_ma_search_next(info, &info->last_key, SEARCH_BIGGER,
@@ -73,9 +81,12 @@ my_bool _ma_check_unique(MARIA_HA *info, MARIA_UNIQUEDEF *def, uchar *record,
     {
       info->page_changed= 1;			/* Can't optimize read next */
       info->cur_row.lastpos= lastpos;
-      DBUG_RETURN(0);				/* end of tree */
+      break;                                    /* end of tree */
     }
   }
+
+end:
+  DBUG_RETURN(error);
 }
 
 
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index beb286c7d4b..dd190e4e4ec 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -574,6 +574,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..7a630e274e3 100644
--- a/storage/maria/maria_read_log.c
+++ b/storage/maria/maria_read_log.c
@@ -192,14 +192,18 @@ static struct my_option my_long_options[] =
   {"display-only", 'd', "display brief info read from records' header",
    &opt_display_only, &opt_display_only, 0, GET_BOOL,
    NO_ARG,0, 0, 0, 0, 0, 0},
+  { "end-lsn", 'e', "Stop applying at this lsn. If end-lsn is used, UNDO:s "
+    "will not be applied", &opt_end_lsn, &opt_end_lsn,
+    0, GET_ULL, REQUIRED_ARG, 0, 0, ~(longlong) 0, 0, 0, 0 },
   {"aria-log-dir-path", 'l',
     "Path to the directory where to store transactional log",
     (uchar **) &maria_data_root, (uchar **) &maria_data_root, 0,
     GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
-  { "page-buffer-size", 'P', "",
+  { "page-buffer-size", 'P',
+    "The size of the buffer used for index blocks for Maria tables",
     &opt_page_buffer_size, &opt_page_buffer_size, 0,
     GET_ULONG, REQUIRED_ARG, (long) USE_BUFFER_INIT,
-    (long) USE_BUFFER_INIT, (long) ~(ulong) 0, (long) MALLOC_OVERHEAD,
+    1024L*1024L, (long) ~(ulong) 0, (long) MALLOC_OVERHEAD,
     (long) IO_SIZE, 0},
   { "start-from-lsn", 'o', "Start reading log from this lsn",
     &opt_start_from_lsn, &opt_start_from_lsn,
@@ -207,15 +211,12 @@ static struct my_option my_long_options[] =
   {"start-from-checkpoint", 'C', "Start applying from last checkpoint",
    &opt_start_from_checkpoint, &opt_start_from_checkpoint, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
-  { "end-lsn", 'e', "Stop applying at this lsn. If end-lsn is used, UNDO:s "
-    "will not be applied", &opt_end_lsn, &opt_end_lsn,
-    0, GET_ULL, REQUIRED_ARG, 0, 0, ~(longlong) 0, 0, 0, 0 },
   {"silent", 's', "Print less information during apply/undo phase",
    &opt_silent, &opt_silent, 0,
    GET_BOOL, NO_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},
+  {"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},
   {"tmpdir", 't', "Path for temporary files. Multiple paths can be specified, "
    "separated by "
 #if defined( __WIN__) || defined(__NETWARE__)
@@ -227,6 +228,9 @@ static struct my_option my_long_options[] =
   {"undo", 'u', "Apply UNDO records to tables. (disable with --disable-undo)",
    (uchar **) &opt_apply_undo, (uchar **) &opt_apply_undo, 0,
    GET_BOOL, NO_ARG, 1, 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},
   {"version", 'V', "Print version and exit.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
@@ -245,7 +249,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 +270,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 +290,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 +319,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 +327,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 b32cce9e43c..5755c275d2e 100644
--- a/storage/myisam/ha_myisam.cc
+++ b/storage/myisam/ha_myisam.cc
@@ -552,6 +552,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/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index 39621852c4c..f0a62d10617 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -1480,7 +1480,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/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 */