Automerge with 5.1

23 files changed, 568 insertions, 206 deletions

diff --git a/storage/federatedx/federatedx_io_mysql.cc b/storage/federatedx/federatedx_io_mysql.cc
index 5245395b060..d6844fab2c6 100644
--- a/storage/federatedx/federatedx_io_mysql.cc
+++ b/storage/federatedx/federatedx_io_mysql.cc
@@ -1,4 +1,4 @@
-/* 
+/*
 Copyright (c) 2007, Antony T Curtis
 All rights reserved.
 
@@ -51,6 +51,12 @@ typedef struct federatedx_savepoint
   uint  flags;
 } SAVEPT;
 
+struct mysql_position
+{
+  MYSQL_RES* result;
+  MYSQL_ROW_OFFSET offset;
+};
+
 
 class federatedx_io_mysql :public federatedx_io
 {
@@ -76,16 +82,16 @@ public:
 
   virtual int error_code();
   virtual const char *error_str();
-  
+
   void reset();
   int commit();
   int rollback();
-  
+
   int savepoint_set(ulong sp);
   ulong savepoint_release(ulong sp);
   ulong savepoint_rollback(ulong sp);
   void savepoint_restrict(ulong sp);
-  
+
   ulong last_savepoint() const;
   ulong actual_savepoint() const;
   bool is_autocommit() const;
@@ -94,7 +100,7 @@ public:
                       uint table_name_length, uint flag);
 
   /* resultset operations */
-  
+
   virtual void free_result(FEDERATEDX_IO_RESULT *io_result);
   virtual unsigned int get_num_fields(FEDERATEDX_IO_RESULT *io_result);
   virtual my_ulonglong get_num_rows(FEDERATEDX_IO_RESULT *io_result);
@@ -104,6 +110,12 @@ public:
                                       unsigned int column);
   virtual bool is_column_null(const FEDERATEDX_IO_ROW *row,
                               unsigned int column) const;
+
+  virtual size_t get_ref_length() const;
+  virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
+                             void *ref);
+  virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
+                            const void *ref);
 };
 
 
@@ -466,14 +478,13 @@ const char *federatedx_io_mysql::error_str()
   return mysql_error(&mysql);
 }
 
-
 FEDERATEDX_IO_RESULT *federatedx_io_mysql::store_result()
 {
   FEDERATEDX_IO_RESULT *result;
   DBUG_ENTER("federatedx_io_mysql::store_result");
-  
+
   result= (FEDERATEDX_IO_RESULT *) mysql_store_result(&mysql);
-  
+
   DBUG_RETURN(result);
 }
 
@@ -590,3 +601,45 @@ error:
   free_result(result);
   return 1;
 }
+
+
+
+size_t federatedx_io_mysql::get_ref_length() const
+{
+  return sizeof(mysql_position);
+}
+
+
+void federatedx_io_mysql::mark_position(FEDERATEDX_IO_RESULT *io_result,
+                                        void *ref)
+{
+  MYSQL_ROWS *tmp= 0;
+  mysql_position& pos= *reinterpret_cast<mysql_position*>(ref);
+  pos.result= (MYSQL_RES *) io_result;
+
+  if (pos.result && pos.result->data)
+  {
+    for (tmp= pos.result->data->data;
+         tmp && (tmp->next != pos.result->data_cursor);
+         tmp= tmp->next)
+    {}
+  }
+
+  pos.offset= tmp;
+}
+
+int federatedx_io_mysql::seek_position(FEDERATEDX_IO_RESULT **io_result,
+                                       const void *ref)
+{
+  const mysql_position& pos= *reinterpret_cast<const mysql_position*>(ref);
+
+  if (!pos.result || !pos.offset)
+    return HA_ERR_END_OF_FILE;
+
+  pos.result->current_row= 0;
+  pos.result->data_cursor= pos.offset;
+  *io_result= (FEDERATEDX_IO_RESULT*) pos.result;
+
+  return 0;
+}
+
diff --git a/storage/federatedx/federatedx_io_null.cc b/storage/federatedx/federatedx_io_null.cc
index cd8fc3eaf85..49f93ab6546 100644
--- a/storage/federatedx/federatedx_io_null.cc
+++ b/storage/federatedx/federatedx_io_null.cc
@@ -96,6 +96,11 @@ public:
                                       unsigned int column);
   virtual bool is_column_null(const FEDERATEDX_IO_ROW *row,
                               unsigned int column) const;
+  virtual size_t get_ref_length() const;
+  virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
+                             void *ref);
+  virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
+                            const void *ref);
 };
 
 
@@ -275,3 +280,20 @@ bool federatedx_io_null::table_metadata(ha_statistics *stats,
 
   return 0;
 }
+
+size_t federatedx_io_null::get_ref_length() const
+{
+  return sizeof(int);
+}
+
+
+void federatedx_io_null::mark_position(FEDERATEDX_IO_RESULT *io_result,
+                                       void *ref)
+{
+}
+
+int federatedx_io_null::seek_position(FEDERATEDX_IO_RESULT **io_result,
+                                      const void *ref)
+{
+  return 0;
+}
diff --git a/storage/federatedx/ha_federatedx.cc b/storage/federatedx/ha_federatedx.cc
index 04ba9984492..8735e5625e6 100644
--- a/storage/federatedx/ha_federatedx.cc
+++ b/storage/federatedx/ha_federatedx.cc
@@ -1717,14 +1717,14 @@ federatedx_txn *ha_federatedx::get_txn(THD *thd, bool no_create)
   return *txnp;
 }
 
-  
+
 int ha_federatedx::disconnect(handlerton *hton, MYSQL_THD thd)
 {
   federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton);
   delete txn;
   return 0;
 }
- 
+
 
 /*
   Used for opening tables. The name will be the name of the file.
@@ -1756,14 +1756,15 @@ int ha_federatedx::open(const char *name, int mode, uint test_if_locked)
     free_share(txn, share);
     DBUG_RETURN(error);
   }
- 
+
+  ref_length= io->get_ref_length();
+
   txn->release(&io);
- 
-  ref_length= (table->s->primary_key != MAX_KEY ?
-               table->key_info[table->s->primary_key].key_length :
-               table->s->reclength);
+
   DBUG_PRINT("info", ("ref_length: %u", ref_length));
 
+  my_init_dynamic_array(&results, sizeof(FEDERATEDX_IO_RESULT*), 4, 4);
+
   reset();
 
   DBUG_RETURN(0);
@@ -1788,8 +1789,9 @@ int ha_federatedx::close(void)
   DBUG_ENTER("ha_federatedx::close");
 
   /* free the result set */
-  if (stored_result)
-    retval= free_result();
+  reset();
+
+  delete_dynamic(&results);
 
   /* Disconnect from mysql */
   if (!thd || !(txn= get_txn(thd, true)))
@@ -1799,7 +1801,7 @@ int ha_federatedx::close(void)
     tmp_txn.release(&io);
 
     DBUG_ASSERT(io == NULL);
-    
+
     if ((error= free_share(&tmp_txn, share)))
       retval= error;
   }
@@ -2525,7 +2527,7 @@ int ha_federatedx::index_read_idx(uchar *buf, uint index, const uchar *key,
                                  uint key_len, enum ha_rkey_function find_flag)
 {
   int retval;
-  FEDERATEDX_IO_RESULT *io_result;
+  FEDERATEDX_IO_RESULT *io_result= 0;
   DBUG_ENTER("ha_federatedx::index_read_idx");
 
   if ((retval= index_read_idx_with_result_set(buf, index, key,
@@ -2601,7 +2603,7 @@ int ha_federatedx::index_read_idx_with_result_set(uchar *buf, uint index,
   if (!(retval= read_next(buf, *result)))
     DBUG_RETURN(retval);
 
-  io->free_result(*result);
+  insert_dynamic(&results, (uchar*) result);
   *result= 0;
   table->status= STATUS_NOT_FOUND;
   DBUG_RETURN(retval);
@@ -2669,10 +2671,7 @@ int ha_federatedx::read_range_first(const key_range *start_key,
     DBUG_RETURN(retval);
 
   if (stored_result)
-  {
-    io->free_result(stored_result);
-    stored_result= 0;
-  }
+    (void) free_result();
 
   if (io->query(sql_query.ptr(), sql_query.length()))
   {
@@ -2773,10 +2772,7 @@ int ha_federatedx::rnd_init(bool scan)
       DBUG_RETURN(error);
 
     if (stored_result)
-    {
-      io->free_result(stored_result);
-      stored_result= 0;
-    }
+      (void) free_result();
 
     if (io->query(share->select_query,
                   strlen(share->select_query)))
@@ -2803,17 +2799,35 @@ int ha_federatedx::rnd_end()
 int ha_federatedx::free_result()
 {
   int error;
-  federatedx_io *tmp_io= 0, **iop;
+  DBUG_ENTER("ha_federatedx::free_result");
   DBUG_ASSERT(stored_result);
-  if (!*(iop= &io) && (error= txn->acquire(share, TRUE, (iop= &tmp_io))))
+  for (uint i= 0; i < results.elements; ++i)
   {
-    DBUG_ASSERT(0);                             // Fail when testing
-    return error;
+    FEDERATEDX_IO_RESULT *result= 0;
+    get_dynamic(&results, (uchar*) &result, i);
+    if (result == stored_result)
+      goto end;
   }
-  (*iop)->free_result(stored_result);
+  if (position_called)
+  {
+    insert_dynamic(&results, (uchar*) &stored_result);
+  }
+  else
+  {
+    federatedx_io *tmp_io= 0, **iop;
+    if (!*(iop= &io) && (error= txn->acquire(share, TRUE, (iop= &tmp_io))))
+    {
+      DBUG_ASSERT(0);                             // Fail when testing
+      insert_dynamic(&results, (uchar*) &stored_result);
+      goto end;
+    }
+    (*iop)->free_result(stored_result);
+    txn->release(&tmp_io);
+  }
+end:
   stored_result= 0;
-  txn->release(&tmp_io);
-  return 0;
+  position_called= FALSE;
+  DBUG_RETURN(0);
 }
 
 int ha_federatedx::index_end(void)
@@ -2862,8 +2876,8 @@ int ha_federatedx::rnd_next(uchar *buf)
 
   SYNOPSIS
     field_in_record_is_null()
-      buf       byte pointer to record 
-      result    mysql result set 
+      buf       byte pointer to record
+      result    mysql result set
 
     DESCRIPTION
      This method is a wrapper method that reads one record from a result
@@ -2896,24 +2910,43 @@ int ha_federatedx::read_next(uchar *buf, FEDERATEDX_IO_RESULT *result)
 }
 
 
-/*
-  store reference to current row so that we can later find it for
-  a re-read, update or delete.
-
-  In case of federatedx, a reference is either a primary key or
-  the whole record.
+/**
+  @brief      Store a reference to current row.
+
+  @details    During a query execution we may have different result sets (RS),
+              e.g. for different ranges. All the RS's used are stored in
+              memory and placed in @c results dynamic array. At the end of
+              execution all stored RS's are freed at once in the
+              @c ha_federated::reset().
+              So, in case of federated, a reference to current row is a
+              stored result address and current data cursor position.
+              As we keep all RS in memory during a query execution,
+              we can get any record using the reference any time until
+              @c ha_federated::reset() is called.
+              TODO: we don't have to store all RS's rows but only those
+              we call @c ha_federated::position() for, so we can free memory
+              where we store other rows in the @c ha_federated::index_end().
+
+  @param[in]  record  record data (unused)
 
-  Called from filesort.cc, sql_select.cc, sql_delete.cc and sql_update.cc.
 */
 
-void ha_federatedx::position(const uchar *record)
+void ha_federatedx::position(const uchar *record __attribute__ ((unused)))
 {
   DBUG_ENTER("ha_federatedx::position");
-  if (table->s->primary_key != MAX_KEY)
-    key_copy(ref, (uchar *)record, table->key_info + table->s->primary_key,
-             ref_length);
-  else
-    memcpy(ref, record, ref_length);
+
+  bzero(ref, ref_length);
+
+  if (!stored_result)
+    DBUG_VOID_RETURN;
+
+  if (txn->acquire(share, TRUE, &io))
+    DBUG_VOID_RETURN;
+
+  io->mark_position(stored_result, ref);
+
+  position_called= TRUE;
+
   DBUG_VOID_RETURN;
 }
 
@@ -2929,23 +2962,23 @@ void ha_federatedx::position(const uchar *record)
 
 int ha_federatedx::rnd_pos(uchar *buf, uchar *pos)
 {
-  int result;
+  int retval;
+  FEDERATEDX_IO_RESULT *result= stored_result;
   DBUG_ENTER("ha_federatedx::rnd_pos");
   ha_statistic_increment(&SSV::ha_read_rnd_count);
-  if (table->s->primary_key != MAX_KEY)
-  {
-    /* We have a primary key, so use index_read_idx to find row */
-    result= index_read_idx(buf, table->s->primary_key, pos,
-                           ref_length, HA_READ_KEY_EXACT);
-  }
-  else
-  {
-    /* otherwise, get the old record ref as obtained in ::position */
-    memcpy(buf, pos, ref_length);
-    result= 0;
-  }
-  table->status= result ? STATUS_NOT_FOUND : 0;
-  DBUG_RETURN(result);
+
+  if ((retval= txn->acquire(share, TRUE, &io)))
+    goto error;
+
+  if ((retval= io->seek_position(&result, pos)))
+    goto error;
+
+  retval= read_next(buf, result);
+  DBUG_RETURN(retval);
+
+error:
+  table->status= STATUS_NOT_FOUND;
+  DBUG_RETURN(retval);
 }
 
 
@@ -2996,15 +3029,20 @@ int ha_federatedx::rnd_pos(uchar *buf, uchar *pos)
 int ha_federatedx::info(uint flag)
 {
   uint error_code;
+  THD *thd= current_thd;
+  federatedx_txn *tmp_txn;
   federatedx_io *tmp_io= 0, **iop= 0;
   DBUG_ENTER("ha_federatedx::info");
 
   error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
   
+  // external_lock may not have been called so txn may not be set
+  tmp_txn= get_txn(thd);
+
   /* we want not to show table status if not needed to do so */
   if (flag & (HA_STATUS_VARIABLE | HA_STATUS_CONST | HA_STATUS_AUTO))
   {
-    if (!*(iop= &io) && (error_code= txn->acquire(share, TRUE, (iop= &tmp_io))))
+    if (!*(iop= &io) && (error_code= tmp_txn->acquire(share, TRUE, (iop= &tmp_io))))
       goto fail;
   }
 
@@ -3029,14 +3067,14 @@ int ha_federatedx::info(uint flag)
     If ::info created it's own transaction, close it. This happens in case
     of show table status;
   */
-  txn->release(&tmp_io);
+  tmp_txn->release(&tmp_io);
 
   DBUG_RETURN(0);
 
 error:
   if (iop && *iop)
   {
-    my_printf_error((*iop)->error_code(), "Got error: %d : %s", MYF(0),
+    my_printf_error((*iop)->error_code(), "Received error: %d : %s", MYF(0),
                     (*iop)->error_code(), (*iop)->error_str());
   }
   else if (remote_error_number != -1 /* error already reported */)
@@ -3045,7 +3083,7 @@ error:
     my_error(error_code, MYF(0), ER(error_code));
   }
 fail:
-  txn->release(&tmp_io);
+  tmp_txn->release(&tmp_io);
   DBUG_RETURN(error_code);
 }
 
@@ -3105,12 +3143,44 @@ int ha_federatedx::extra(ha_extra_function operation)
 
 int ha_federatedx::reset(void)
 {
+  int error = 0;
+
   insert_dup_update= FALSE;
   ignore_duplicates= FALSE;
   replace_duplicates= FALSE;
-  return 0;
-}
+  position_called= FALSE;
+
+  if (stored_result)
+    insert_dynamic(&results, (uchar*) &stored_result);
+  stored_result= 0;
+
+  if (results.elements)
+  {
+    federatedx_txn *tmp_txn;
+    federatedx_io *tmp_io= 0, **iop;
+
+    // external_lock may not have been called so txn may not be set
+    tmp_txn= get_txn(current_thd);
+
+    if (!*(iop= &io) && (error= tmp_txn->acquire(share, TRUE, (iop= &tmp_io))))
+    {
+      DBUG_ASSERT(0);                             // Fail when testing
+      return error;
+    }
+
+    for (uint i= 0; i < results.elements; ++i)
+    {
+      FEDERATEDX_IO_RESULT *result= 0;
+      get_dynamic(&results, (uchar*) &result, i);
+      (*iop)->free_result(result);
+    }
+    tmp_txn->release(&tmp_io);
+    reset_dynamic(&results);
+  }
+
+  return error;
 
+}
 
 /*
   Used to delete all rows in a table. Both for cases of truncate and
@@ -3237,7 +3307,7 @@ static int test_connection(MYSQL_THD thd, federatedx_io *io,
 
   str.length(0);
   str.append(STRING_WITH_LEN("SELECT * FROM "));
-  append_identifier(thd, &str, share->table_name, 
+  append_identifier(thd, &str, share->table_name,
                     share->table_name_length);
   str.append(STRING_WITH_LEN(" WHERE 1=0"));
 
@@ -3288,14 +3358,14 @@ int ha_federatedx::create(const char *name, TABLE *table_arg,
   pthread_mutex_lock(&federatedx_mutex);
   tmp_share.s= get_server(&tmp_share, NULL);
   pthread_mutex_unlock(&federatedx_mutex);
-    
+
   if (tmp_share.s)
   {
     tmp_txn= get_txn(thd);
     if (!(retval= tmp_txn->acquire(&tmp_share, TRUE, &tmp_io)))
     {
       retval= test_connection(thd, tmp_io, &tmp_share);
-      tmp_txn->release(&tmp_io);    
+      tmp_txn->release(&tmp_io);
     }
     free_server(tmp_txn, tmp_share.s);
   }
diff --git a/storage/federatedx/ha_federatedx.h b/storage/federatedx/ha_federatedx.h
index 2fd3c559321..2820f8a6c29 100644
--- a/storage/federatedx/ha_federatedx.h
+++ b/storage/federatedx/ha_federatedx.h
@@ -1,5 +1,5 @@
-/* 
-Copyright (c) 2008, Patrick Galbraith 
+/*
+Copyright (c) 2008, Patrick Galbraith
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without
@@ -43,7 +43,7 @@ class federatedx_io;
 typedef struct st_fedrated_server {
   MEM_ROOT mem_root;
   uint use_count, io_count;
-  
+
   uchar *key;
   uint key_length;
 
@@ -74,10 +74,10 @@ typedef struct st_fedrated_server {
 
 #include <mysql.h>
 
-/* 
+/*
   handler::print_error has a case statement for error numbers.
-  This value is (10000) is far out of range and will envoke the 
-  default: case.  
+  This value is (10000) is far out of range and will envoke the
+  default: case.
   (Current error range is 120-159 from include/my_base.h)
 */
 #define HA_FEDERATEDX_ERROR_WITH_REMOTE_SYSTEM 10000
@@ -158,7 +158,7 @@ public:
   const char * get_database() const { return server->database; }
   ushort       get_port() const     { return server->port; }
   const char * get_socket() const   { return server->socket; }
-  
+
   static bool handles_scheme(const char *scheme);
   static federatedx_io *construct(MEM_ROOT *server_root,
                                   FEDERATEDX_SERVER *server);
@@ -167,7 +167,7 @@ public:
   { return alloc_root(mem_root, size); }
   static void operator delete(void *ptr, size_t size)
   { TRASH(ptr, size); }
-    
+
   virtual int query(const char *buffer, uint length)=0;
   virtual FEDERATEDX_IO_RESULT *store_result()=0;
 
@@ -178,25 +178,25 @@ public:
 
   virtual int error_code()=0;
   virtual const char *error_str()=0;
-  
+
   virtual void reset()=0;
   virtual int commit()=0;
   virtual int rollback()=0;
-  
+
   virtual int savepoint_set(ulong sp)=0;
   virtual ulong savepoint_release(ulong sp)=0;
   virtual ulong savepoint_rollback(ulong sp)=0;
   virtual void savepoint_restrict(ulong sp)=0;
-  
+
   virtual ulong last_savepoint() const=0;
   virtual ulong actual_savepoint() const=0;
   virtual bool is_autocommit() const=0;
 
   virtual bool table_metadata(ha_statistics *stats, const char *table_name,
                               uint table_name_length, uint flag) = 0;
-  
+
   /* resultset operations */
-  
+
   virtual void free_result(FEDERATEDX_IO_RESULT *io_result)=0;
   virtual unsigned int get_num_fields(FEDERATEDX_IO_RESULT *io_result)=0;
   virtual my_ulonglong get_num_rows(FEDERATEDX_IO_RESULT *io_result)=0;
@@ -206,6 +206,13 @@ public:
                                       unsigned int column)=0;
   virtual bool is_column_null(const FEDERATEDX_IO_ROW *row,
                               unsigned int column) const=0;
+
+  virtual size_t get_ref_length() const=0;
+  virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
+                             void *ref)=0;
+  virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
+                            const void *ref)=0;
+
 };
 
 
@@ -215,12 +222,12 @@ class federatedx_txn
   ulong savepoint_level;
   ulong savepoint_stmt;
   ulong savepoint_next;
-  
+
   void release_scan();
 public:
   federatedx_txn();
   ~federatedx_txn();
-  
+
   bool has_connections() const { return txn_list != NULL; }
   bool in_transaction() const { return savepoint_next != 0; }
   int acquire(FEDERATEDX_SHARE *share, bool readonly, federatedx_io **io);
@@ -254,8 +261,12 @@ class ha_federatedx: public handler
   federatedx_txn *txn;
   federatedx_io *io;
   FEDERATEDX_IO_RESULT *stored_result;
+  /**
+      Array of all stored results we get during a query execution.
+  */
+  DYNAMIC_ARRAY results;
+  bool position_called;
   uint fetch_num; // stores the fetch num
-  FEDERATEDX_IO_OFFSET current_position;  // Current position used by ::position()
   int remote_error_number;
   char remote_error_buf[FEDERATEDX_QUERY_BUFFER_SIZE];
   bool ignore_duplicates, replace_duplicates;
@@ -269,7 +280,7 @@ private:
   */
   uint convert_row_to_internal_format(uchar *buf, FEDERATEDX_IO_ROW *row,
                                       FEDERATEDX_IO_RESULT *result);
-  bool create_where_from_key(String *to, KEY *key_info, 
+  bool create_where_from_key(String *to, KEY *key_info,
                              const key_range *start_key,
                              const key_range *end_key,
                              bool records_in_range, bool eq_range);
@@ -348,18 +359,18 @@ public:
     Talk to Kostja about this - how to get the
     number of rows * ...
     disk scan time on other side (block size, size of the row) + network time ...
-    The reason for "records * 1000" is that such a large number forces 
+    The reason for "records * 1000" is that such a large number forces
     this to use indexes "
   */
   double scan_time()
   {
     DBUG_PRINT("info", ("records %lu", (ulong) stats.records));
-    return (double)(stats.records*1000); 
+    return (double)(stats.records*1000);
   }
   /*
     The next method will never be called if you do not implement indexes.
   */
-  double read_time(uint index, uint ranges, ha_rows rows) 
+  double read_time(uint index, uint ranges, ha_rows rows)
   {
     /*
       Per Brian, this number is bugus, but this method must be implemented,
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index b8f31cce755..1475cd52488 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -248,7 +248,7 @@ static MYSQL_THDVAR_ULONG(repair_threads, PLUGIN_VAR_RQCMDARG,
 static MYSQL_THDVAR_ULONG(sort_buffer_size, PLUGIN_VAR_RQCMDARG,
        "The buffer that is allocated when sorting the index when doing a "
        "REPAIR or when creating indexes with CREATE INDEX or ALTER TABLE.",
-       0, 0, 8192*1024, 4, ~0L, 1);
+       0, 0, 128L*1024L*1024L, 4, ~0L, 1);
 
 static MYSQL_THDVAR_ENUM(stats_method, PLUGIN_VAR_RQCMDARG,
        "Specifies how maria index statistics collection code should treat "
@@ -1850,7 +1850,7 @@ int ha_maria::enable_indexes(uint mode)
                         "retrying",
                         my_errno, param.db_name, param.table_name);
       /* This should never fail normally */
-      DBUG_ASSERT(0);
+      DBUG_ASSERT(thd->killed != 0);
       /* Repairing by sort failed. Now try standard repair method. */
       param.testflag &= ~T_REP_BY_SORT;
       error= (repair(thd, &param, 0) != HA_ADMIN_OK);
diff --git a/storage/maria/ma_bitmap.c b/storage/maria/ma_bitmap.c
index 70617160d03..a76b36e1115 100644
--- a/storage/maria/ma_bitmap.c
+++ b/storage/maria/ma_bitmap.c
@@ -365,8 +365,8 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
     */
     if (bitmap->changed)
     {
-      res= write_changed_bitmap(share, bitmap);
       bitmap->changed= FALSE;
+      res= write_changed_bitmap(share, bitmap);
     }
     /*
       We do NOT use FLUSH_KEEP_LAZY because we must be sure that bitmap
@@ -2063,6 +2063,13 @@ my_bool _ma_bitmap_set_full_page_bits(MARIA_HA *info,
   safe_mutex_assert_owner(&info->s->bitmap.bitmap_lock);
 
   bitmap_page= page - page % bitmap->pages_covered;
+  if (page == bitmap_page ||
+      page + page_count >= bitmap_page + bitmap->pages_covered)
+  {
+    DBUG_ASSERT(0);                             /* Wrong in data */
+    DBUG_RETURN(1);
+  }
+
   if (bitmap_page != bitmap->page &&
       _ma_change_bitmap_page(info, bitmap, bitmap_page))
     DBUG_RETURN(1);
@@ -2142,12 +2149,12 @@ void _ma_bitmap_flushable(MARIA_HA *info, int non_flushable_inc)
     DBUG_VOID_RETURN;
 
   bitmap= &share->bitmap;
+  pthread_mutex_lock(&bitmap->bitmap_lock);
+
   if (non_flushable_inc == -1)
   {
-    pthread_mutex_lock(&bitmap->bitmap_lock);
     DBUG_ASSERT((int) bitmap->non_flushable > 0);
     DBUG_ASSERT(info->non_flushable_state == 1);
-    info->non_flushable_state= 0;
     if (--bitmap->non_flushable == 0)
     {
       /*
@@ -2164,11 +2171,11 @@ void _ma_bitmap_flushable(MARIA_HA *info, int non_flushable_inc)
     }
     DBUG_PRINT("info", ("bitmap->non_flushable: %u", bitmap->non_flushable));
     pthread_mutex_unlock(&bitmap->bitmap_lock);
+    info->non_flushable_state= 0;
     DBUG_VOID_RETURN;
   }
   DBUG_ASSERT(non_flushable_inc == 1);
   DBUG_ASSERT(info->non_flushable_state == 0);
-  pthread_mutex_lock(&bitmap->bitmap_lock);
   while (unlikely(bitmap->flush_all_requested))
   {
     /*
@@ -2186,9 +2193,9 @@ void _ma_bitmap_flushable(MARIA_HA *info, int non_flushable_inc)
     pthread_cond_wait(&bitmap->bitmap_cond, &bitmap->bitmap_lock);
   }
   bitmap->non_flushable++;
-  info->non_flushable_state= 1;
   DBUG_PRINT("info", ("bitmap->non_flushable: %u", bitmap->non_flushable));
   pthread_mutex_unlock(&bitmap->bitmap_lock);
+  info->non_flushable_state= 1;
   DBUG_VOID_RETURN;
 }
 
@@ -2217,6 +2224,8 @@ void _ma_bitmap_flushable(MARIA_HA *info, int non_flushable_inc)
   Note that we may have 'filler blocks' that are used to split a block
   in half; These can be recognized by that they have page_count == 0.
 
+  This code also reverse the effect of ma_bitmap_flushable(.., 1);
+
   RETURN
     0  ok
     1  error (Couldn't write or read bitmap page)
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index f62a4e2ce6d..0779564c8af 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -334,12 +334,13 @@ typedef struct st_maria_extent_cursor
     }                                                   \
   } while (0)
 
+
 static my_bool delete_tails(MARIA_HA *info, MARIA_RECORD_POS *tails);
 static my_bool delete_head_or_tail(MARIA_HA *info,
                                    pgcache_page_no_t page, uint record_number,
                                    my_bool head, my_bool from_update);
 #ifndef DBUG_OFF
-static void _ma_print_directory(uchar *buff, uint block_size);
+static void _ma_print_directory(FILE *file, uchar *buff, uint block_size);
 #endif
 static uchar *store_page_range(uchar *to, MARIA_BITMAP_BLOCK *block,
                                uint block_size, ulong length,
@@ -615,7 +616,7 @@ static inline uint end_of_previous_entry(uchar *dir, uchar *end)
 
 #ifndef DBUG_OFF
 
-static void _ma_print_directory(uchar *buff, uint block_size)
+static void _ma_print_directory(FILE *file, uchar *buff, uint block_size)
 {
   uint max_entry= (uint) ((uchar *) buff)[DIR_COUNT_OFFSET], row= 0;
   uint end_of_prev_row= PAGE_HEADER_SIZE;
@@ -624,40 +625,46 @@ static void _ma_print_directory(uchar *buff, uint block_size)
   dir= dir_entry_pos(buff, block_size, max_entry-1);
   end= dir_entry_pos(buff, block_size, 0);
 
-  DBUG_LOCK_FILE;
-  fprintf(DBUG_FILE,"Directory dump (pos:length):\n");
+  DBUG_LOCK_FILE;                               /* If using DBUG_FILE */
+  fprintf(file,"Directory dump (pos:length):\n");
 
   for (row= 1; dir <= end ; end-= DIR_ENTRY_SIZE, row++)
   {
     uint offset= uint2korr(end);
     uint length= uint2korr(end+2);
-    fprintf(DBUG_FILE, "   %4u:%4u", offset, offset ? length : 0);
+    fprintf(file, "   %4u:%4u", offset, offset ? length : 0);
     if (!(row % (80/12)))
-      fputc('\n', DBUG_FILE);
+      fputc('\n', file);
     if (offset)
     {
       DBUG_ASSERT(offset >= end_of_prev_row);
       end_of_prev_row= offset + length;
     }
   }
-  fputc('\n', DBUG_FILE);
-  fflush(DBUG_FILE);
+  fputc('\n', file);
+  fflush(file);
   DBUG_UNLOCK_FILE;
 }
 
 
-static void check_directory(uchar *buff, uint block_size, uint min_row_length)
+static void check_directory(uchar *buff, uint block_size, uint min_row_length,
+                            uint real_empty_size)
 {
   uchar *dir, *end;
   uint max_entry= (uint) buff[DIR_COUNT_OFFSET];
   uint start_of_dir, deleted;
-  uchar free_entry, prev_free_entry;
   uint end_of_prev_row= PAGE_HEADER_SIZE;
+  uint empty_size_on_page;
+  uint empty_size;
+  uchar free_entry, prev_free_entry;
 
   dir= dir_entry_pos(buff, block_size, max_entry-1);
   start_of_dir= (uint) (dir - buff);
   end= dir_entry_pos(buff, block_size, 0);
-  deleted= 0;
+  deleted= empty_size= 0;
+
+  empty_size_on_page= (real_empty_size != (uint) -1 ? real_empty_size :
+                       uint2korr(buff + EMPTY_SPACE_OFFSET));
 
   /* Ensure that all rows are in increasing order and no overlaps */
   for (; dir <= end ; end-= DIR_ENTRY_SIZE)
@@ -668,12 +675,15 @@ static void check_directory(uchar *buff, uint block_size, uint min_row_length)
     {
       DBUG_ASSERT(offset >= end_of_prev_row);
       DBUG_ASSERT(!length || length >= min_row_length);
+      empty_size+= offset - end_of_prev_row;
       end_of_prev_row= offset + length;
     }
     else
       deleted++;
   }
+  empty_size+= start_of_dir - end_of_prev_row;
   DBUG_ASSERT(end_of_prev_row <= start_of_dir);
+  DBUG_ASSERT(empty_size == empty_size_on_page);
 
   /* check free links */
   free_entry= buff[DIR_FREE_OFFSET];
@@ -690,7 +700,7 @@ static void check_directory(uchar *buff, uint block_size, uint min_row_length)
   DBUG_ASSERT(deleted == 0);
 }
 #else
-#define check_directory(A,B,C)
+#define check_directory(A,B,C,D)
 #endif /* DBUG_OFF */
 
 
@@ -792,20 +802,27 @@ static my_bool extend_area_on_page(MARIA_HA *info,
                                    uint *empty_space, uint *ret_offset,
                                    uint *ret_length)
 {
-  uint rec_offset, length;
+  uint rec_offset, length, org_rec_length;
   uint max_entry= (uint) buff[DIR_COUNT_OFFSET];
   DBUG_ENTER("extend_area_on_page");
 
+  /*
+    We can't check for min length here as we may have called
+    extend_directory() to create a new (empty) entry just before
+  */
+  check_directory(buff, block_size, 0, *empty_space);
+
   rec_offset= uint2korr(dir);
   if (rec_offset)
   {
     /* Extending old row;  Mark current space as 'free' */
-    length= uint2korr(dir + 2);
+    length= org_rec_length= uint2korr(dir + 2);
     DBUG_PRINT("info", ("rec_offset: %u  length: %u  request_length: %u  "
                         "empty_space: %u",
-                        rec_offset, length, request_length, *empty_space));
+                        rec_offset, org_rec_length, request_length,
+                        *empty_space));
 
-    *empty_space+= length;
+    *empty_space+= org_rec_length;
   }
   else
   {
@@ -875,6 +892,7 @@ static my_bool extend_area_on_page(MARIA_HA *info,
                                "length: %u  request_length: %u",
                                length, request_length));
           my_errno= HA_ERR_WRONG_IN_RECORD;     /* File crashed */
+          DBUG_ASSERT(0);                       /* For debugging */
           DBUG_RETURN(1);                       /* Error in block */
         }
         *empty_space= length;                   /* All space is here */
@@ -885,7 +903,9 @@ static my_bool extend_area_on_page(MARIA_HA *info,
   int2store(dir + 2, length);
   *ret_offset= rec_offset;
   *ret_length= length;
-  check_directory(buff, block_size, info ? info->s->base.min_block_length : 0);
+
+  check_directory(buff, block_size, info ? info->s->base.min_block_length : 0,
+                  *empty_space - length);
   DBUG_RETURN(0);
 }
 
@@ -1094,7 +1114,7 @@ static uchar *find_free_position(MARIA_HA *info,
     *res_length= length;
 
     check_directory(buff, block_size,
-                    info ? info->s->base.min_block_length : 0);
+                    info ? info->s->base.min_block_length : 0, (uint) -1);
     DBUG_RETURN(dir);
   }
   /* No free places in dir; create a new one */
@@ -1115,7 +1135,8 @@ static uchar *find_free_position(MARIA_HA *info,
   *res_rownr= max_entry;
   *res_length= length;
 
-  check_directory(buff, block_size, info ? info->s->base.min_block_length : 0);
+  check_directory(buff, block_size, info ? info->s->base.min_block_length : 0,
+                  *empty_space);
   DBUG_RETURN(dir);
 }
 
@@ -1195,7 +1216,8 @@ static my_bool extend_directory(MARIA_HA *info, uchar *buff, uint block_size,
   }
 
   check_directory(buff, block_size,
-                  info ? min(info->s->base.min_block_length, length) : 0);
+                  info ? min(info->s->base.min_block_length, length) : 0,
+                  *empty_space);
   DBUG_RETURN(0);
 }
 
@@ -1496,6 +1518,8 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
   {
     /* Move all entries after rownr to end of page */
     uint rownr_length;
+
+    DBUG_ASSERT(extend_block);                  /* Should always be true */
     next_free_pos= end_of_found_block= page_pos=
       block_size - DIR_ENTRY_SIZE * max_entry - PAGE_SUFFIX_SIZE;
     diff= 0;
@@ -1567,13 +1591,13 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
       int2store(dir, offset + diff);            /* correct current pos */
       next_free_pos= offset;
     }
-
     if (page_pos != end_of_found_block)
     {
       uint length= (end_of_found_block - next_free_pos);
       memmove(buff + page_pos - length, buff + next_free_pos, length);
       next_free_pos= page_pos- length;
     }
+
     /* Extend rownr block to cover hole */
     rownr_length= next_free_pos - start_of_found_block;
     int2store(dir+2, rownr_length);
@@ -1596,8 +1620,9 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
     }
     buff[PAGE_TYPE_OFFSET]&= ~(uchar) PAGE_CAN_BE_COMPACTED;
   }
-  check_directory(buff, block_size, min_row_length);
-  DBUG_EXECUTE("directory", _ma_print_directory(buff, block_size););
+  check_directory(buff, block_size, min_row_length,
+                  extend_block ? 0 : (uint) -1);
+  DBUG_EXECUTE("directory", _ma_print_directory(DBUG_FILE, buff, block_size););
   DBUG_VOID_RETURN;
 }
 
@@ -1826,16 +1851,11 @@ static my_bool get_rowpos_in_head_or_tail_page(MARIA_HA *info,
       goto err;
   }
 
+  /*
+    The following dir entry is unused in case of insert / update but
+    not in case of undo_update / undo_delete
+  */
   dir= dir_entry_pos(buff, block_size, rownr);
-#ifdef SANITY_CHECKS
-    /* Tail's should always be unused */
-  if (page_type == TAIL_PAGE && max_entry > rownr &&
-      (dir[0] != 0 || dir[1] != 0))
-  {
-    DBUG_ASSERT(0);
-    goto err;
-  }
-#endif
 
   if (extend_area_on_page(page_type == HEAD_PAGE ? info : 0, buff, dir,
                           rownr, block_size, length,
@@ -2212,6 +2232,8 @@ static void store_extent_info(uchar *to,
   MARIA_BITMAP_BLOCK *block, *end_block;
   uint copy_length;
   my_bool first_found= 0;
+  DBUG_ENTER("store_extent_info");
+  DBUG_PRINT("enter", ("count: %u", count));
 
   for (block= first_block, end_block= first_block+count ;
        block < end_block; block++)
@@ -2231,6 +2253,7 @@ static void store_extent_info(uchar *to,
         page_count|= START_EXTENT_BIT;
       }
       pagerange_store(to + PAGE_STORE_SIZE, page_count);
+      DBUG_DUMP("extent", to, ROW_EXTENT_SIZE);
       to+= ROW_EXTENT_SIZE;
       if (!first_found)
       {
@@ -2245,6 +2268,7 @@ static void store_extent_info(uchar *to,
     data.
   */
   bzero(to, (size_t) (row_extents_second_part + copy_length - to));
+  DBUG_VOID_RETURN;
 }
 
 
@@ -2263,7 +2287,8 @@ static void store_extent_info(uchar *to,
 
    @return
    @retval 0  ok
-   @retval 1  Error (out of memory or disk error changing bitmap)
+   @retval 1  Error (out of memory or disk error changing bitmap) or
+              wrong information in extent information
 */
 
 static my_bool extent_to_bitmap_blocks(MARIA_HA *info,
@@ -2274,7 +2299,7 @@ static my_bool extent_to_bitmap_blocks(MARIA_HA *info,
 {
   MARIA_BITMAP_BLOCK *block, *start_block;
   MARIA_SHARE *share= info->s;
-  uint i;
+  uint i, tail_page;
   DBUG_ENTER("extent_to_bitmap_blocks");
 
   if (allocate_dynamic(&info->bitmap_blocks, extent_count + 2))
@@ -2300,13 +2325,36 @@ static my_bool extent_to_bitmap_blocks(MARIA_HA *info,
       page_count&= ~START_EXTENT_BIT;
       start_block->sub_blocks= (uint) (block - start_block);
       start_block= block;
-
     }
     block->page= page_korr(extent_info);
     block->page_count= page_count;
     block->sub_blocks= 0;
+    if (block->page_count == 0)
+    {
+      /* Extend allocated but not used by write_block_record() */
+      DBUG_ASSERT(block->page == 0);
+      /* This is the last block */
+      blocks->count= i;
+      break;
+    }
+    if ((tail_page= page_count & TAIL_BIT))
+      page_count= 1;
 
-    if (page_count & TAIL_BIT)
+    /* Check if wrong data */
+    if (block->page == 0 || page_count == 0 ||
+        (block->page + page_count) * share->block_size >
+         share->state.state.data_file_length)
+    {
+      DBUG_PRINT("error", ("page: %lu  page_count: %u  tail: %u  length: %ld  data_length: %ld",
+                           (ulong) block->page,
+                           (block->page_count & ~TAIL_BIT),
+                           (uint) test(block->page_count & TAIL_BIT),
+                           (ulong) ((block->page + (page_count & ~TAIL_BIT)) *
+                                    share->block_size),
+                           (ulong) share->state.state.data_file_length));
+      DBUG_RETURN(1);
+    }
+    if (tail_page)
     {
       block->org_bitmap_value= _ma_bitmap_get_page_bits(info, &share->bitmap,
                                                         block->page);
@@ -2318,7 +2366,7 @@ static my_bool extent_to_bitmap_blocks(MARIA_HA *info,
       my_bool res;
       pthread_mutex_lock(&share->bitmap.bitmap_lock);
       res= _ma_bitmap_set_full_page_bits(info, &share->bitmap,
-                                         block->page, block->page_count);
+                                         block->page, page_count);
       pthread_mutex_unlock(&share->bitmap.bitmap_lock);
       if (res)
         DBUG_RETURN(1);
@@ -2740,9 +2788,16 @@ static my_bool write_block_record(MARIA_HA *info,
                    sizeof(char*));
       memcpy(data, tmp_pos, *blob_lengths);
       data+= *blob_lengths;
-      /* Skip over tail page that was to be used to store blob */
-      block++;
-      bitmap_blocks->tail_page_skipped= 1;
+      /*
+        The following is not true when we want to insert data into original
+        place. In this case we don't have any extra blocks allocated
+      */
+      if (likely(undo_lsn == LSN_ERROR))
+      {
+        /* Skip over tail page that was prepared for storing blob */
+        block++;
+        bitmap_blocks->tail_page_skipped= 1;
+      }
     }
     if (head_block->sub_blocks > 1)
     {
@@ -2755,7 +2810,9 @@ static my_bool write_block_record(MARIA_HA *info,
 
   /* Update page directory */
   head_length= (uint) (data - row_pos->data);
-  DBUG_PRINT("info", ("Used head length on page: %u", head_length));
+  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)
   {
@@ -2765,6 +2822,7 @@ static my_bool write_block_record(MARIA_HA *info,
     data+= diff_length;
     head_length= share->base.min_block_length;
   }
+  DBUG_ASSERT(undo_lsn == LSN_ERROR || head_length == row_pos->length);
   int2store(row_pos->dir + 2, head_length);
   /* update empty space at start of block */
   row_pos->empty_space-= head_length;
@@ -2776,7 +2834,8 @@ static my_bool write_block_record(MARIA_HA *info,
     head_block->empty_space= 0;               /* Page is full */
   head_block->used|= BLOCKUSED_USED;
 
-  check_directory(page_buff, share->block_size, share->base.min_block_length);
+  check_directory(page_buff, share->block_size, share->base.min_block_length,
+                  (uint) -1);
 
   /*
      Now we have to write tail pages, as we need to store the position
@@ -2827,11 +2886,13 @@ static my_bool write_block_record(MARIA_HA *info,
         {
           /*
             Set only a bit, to not cause bitmap code to believe a block is full
-            when there is still a lot of entries in it
+            when there is still a lot of entries in it.
           */
           block->used|= BLOCKUSED_USED;
         }
       }
+      DBUG_ASSERT((undo_lsn == LSN_ERROR ||
+                   block == bitmap_blocks->block + bitmap_blocks->count));
       column= save_column;
       block= save_block;
       blob_lengths= save_blob_lengths;
@@ -3225,9 +3286,10 @@ static my_bool write_block_record(MARIA_HA *info,
     else
     {
       uchar log_data[LSN_STORE_SIZE + FILEID_STORE_SIZE +
-                     PAGE_STORE_SIZE + DIRPOS_STORE_SIZE +
+                     PAGE_STORE_SIZE + DIRPOS_STORE_SIZE + 2 +
                      HA_CHECKSUM_STORE_SIZE + 2 + PAGERANGE_STORE_SIZE +
                      ROW_EXTENT_SIZE];
+      uchar *log_pos;
       ha_checksum checksum_delta;
 
       /* LOGREC_UNDO_ROW_INSERT & LOGREC_UNDO_ROW_UPDATE share same header */
@@ -3237,18 +3299,17 @@ static my_bool write_block_record(MARIA_HA *info,
       dirpos_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE +
                    PAGE_STORE_SIZE,
                    row_pos->rownr);
-
-      log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    log_data;
-      log_array[TRANSLOG_INTERNAL_PARTS + 0].length=
-        (LSN_STORE_SIZE + FILEID_STORE_SIZE + PAGE_STORE_SIZE +
-         DIRPOS_STORE_SIZE);
+      log_pos= (log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE +
+                PAGE_STORE_SIZE + DIRPOS_STORE_SIZE);
       store_checksum_in_rec(share, checksum_delta,
                             row->checksum - old_record_checksum,
-                            log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE +
-                            PAGE_STORE_SIZE + DIRPOS_STORE_SIZE,
-                            log_array[TRANSLOG_INTERNAL_PARTS + 0].length);
+                            log_pos, log_pos);
       compile_time_assert(sizeof(ha_checksum) == HA_CHECKSUM_STORE_SIZE);
 
+      log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    log_data;
+      log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos -
+                                                             log_data);
+
       if (!old_record)
       {
         /* Store undo_lsn in case we are aborting the insert */
@@ -3267,16 +3328,18 @@ static my_bool write_block_record(MARIA_HA *info,
       else
       {
         /* Write UNDO log record for the UPDATE */
-        uchar *log_pos= (log_data +
-                         log_array[TRANSLOG_INTERNAL_PARTS + 0].length);
         size_t row_length, extents_length;
-        uint row_parts_count;
+        uint row_parts_count, cur_head_length;
 
         /*
           Write head length and extents of the original row so that we
-          during UNDO can put it back in the original position
+          during UNDO can put it back in the original position.
+          We don't store size for TRANSID, as we don't write this during
+          UNDO.
         */
-        int2store(log_pos, info->cur_row.head_length);
+        cur_head_length= (info->cur_row.head_length -
+                          info->cur_row.header_length);
+        int2store(log_pos, cur_head_length);
         pagerange_store(log_pos + 2, info->cur_row.extents_count);
         log_pos+= 2 + PAGERANGE_STORE_SIZE;
         log_array[TRANSLOG_INTERNAL_PARTS + 0].length+= (2 +
@@ -3589,6 +3652,7 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
   MARIA_PINNED_PAGE page_link;
   uint rownr, org_empty_size, head_length;
   uint block_size= info->s->block_size;
+  uint errpos= 0;
   uchar *dir;
   pgcache_page_no_t page;
   struct st_row_pos_info row_pos;
@@ -3627,11 +3691,21 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
   rownr= ma_recordpos_to_dir_entry(record_pos);
   dir= dir_entry_pos(buff, block_size, rownr);
 
-  if ((org_empty_size + cur_row->head_length) >= new_row->total_length)
+  /*
+    We can't use cur_row->head_length as the block may have been compacted
+    since we read it.
+  */
+  head_length= uint2korr(dir + 2);
+
+  if ((org_empty_size + head_length) >= new_row->total_length)
   {
     uint rec_offset, length;
     MARIA_BITMAP_BLOCK block;
 
+    DBUG_PRINT("info", ("org_empty_size: %u  org_length: %u  new_length: %lu",
+                        org_empty_size, head_length,
+                        new_row->total_length));
+
     /*
       We can fit the new row in the same page as the original head part
       of the row
@@ -3641,7 +3715,10 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
     if (extend_area_on_page(info, buff, dir, rownr, block_size,
                             new_row->total_length, &org_empty_size,
                             &rec_offset, &length))
+    {
+      errpos= 1;
       goto err;
+    }
 
     row_pos.buff= buff;
     row_pos.rownr= rownr;
@@ -3658,9 +3735,15 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
 
     if (*cur_row->tail_positions &&
         delete_tails(info, cur_row->tail_positions))
+    {
+      errpos= 2;
       goto err;
+    }
     if (cur_row->extents_count && free_full_pages(info, cur_row))
+    {
+      errpos= 3;
       goto err;
+    }
     res= write_block_record(info, oldrec, record, new_row, blocks,
                             1, &row_pos, undo_lsn, old_checksum);
     /* We can't update or delete this without re-reading it again */
@@ -3670,14 +3753,23 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
   /* Delete old row */
   if (*cur_row->tail_positions &&
       delete_tails(info, cur_row->tail_positions))
+  {
+    errpos= 4;
     goto err;
+  }
   if (cur_row->extents_count && free_full_pages(info, cur_row))
+  {
+    errpos= 5;
     goto err;
+  }
 
   head_length= uint2korr(dir + 2);
   if (_ma_bitmap_find_new_place(info, new_row, page, head_length +
                                 org_empty_size, blocks))
+  {
+    errpos= 6;
     goto err;
+  }
 
   /*
     Allocate all size in block for record
@@ -3704,10 +3796,14 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
   row_pos.length= head_length;
   if ((res= write_block_record(info, oldrec, record, new_row, blocks, 1,
                                &row_pos, undo_lsn, old_checksum)))
+  {
+    errpos= 7;
     goto err;
+  }
   DBUG_RETURN(0);
 
 err:
+  DBUG_PRINT("error", ("errpos: %d", errpos));
   if (info->non_flushable_state)
     _ma_bitmap_flushable(info, -1);
   _ma_unpin_all_pages_and_finalize_row(info, LSN_IMPOSSIBLE);
@@ -3725,6 +3821,8 @@ err:
 
   This is the main reason we don't make a lot of subfunctions that are
   common between _ma_update_block_record2() and this function.
+
+  Note: If something goes wrong we mark the file crashed
 */
 
 static my_bool _ma_update_at_original_place(MARIA_HA *info,
@@ -3780,6 +3878,10 @@ static my_bool _ma_update_at_original_place(MARIA_HA *info,
 
   if ((org_empty_size + cur_row->head_length) < length_on_head_page)
   {
+    DBUG_PRINT("error",
+               ("org_empty_size: %u  head_length: %u  length_on_page: %u",
+                org_empty_size, (uint) cur_row->head_length,
+                length_on_head_page));
     my_errno= HA_ERR_WRONG_IN_RECORD;
     goto err;
   }
@@ -3799,7 +3901,6 @@ static my_bool _ma_update_at_original_place(MARIA_HA *info,
   row_pos.empty_space= empty_size;
   row_pos.dir= dir;
   row_pos.data= buff + rec_offset;
-  row_pos.length= length_on_head_page;
 
   /* Delete old row */
   if (*cur_row->tail_positions &&
@@ -3829,12 +3930,17 @@ static my_bool _ma_update_at_original_place(MARIA_HA *info,
               max(new_row->total_length, share->base.min_block_length) <=
               length_on_head_page);
 
+  /* Store same amount of data on head page as on original page */
+  row_pos.length= (length_on_head_page - 
+                   (extent_count + 1 - blocks->count) * ROW_EXTENT_SIZE);
+  set_if_bigger(row_pos.length, share->base.min_block_length);
   if ((res= write_block_record(info, oldrec, record, new_row, blocks,
                                1, &row_pos, undo_lsn, old_checksum)))
     goto err;
   DBUG_RETURN(0);
 
 err:
+  _ma_mark_file_crashed(share);
   if (info->non_flushable_state)
     _ma_bitmap_flushable(info, -1);
   _ma_unpin_all_pages_and_finalize_row(info, LSN_IMPOSSIBLE);
@@ -3888,7 +3994,7 @@ static int delete_dir_entry(uchar *buff, uint block_size, uint record_number,
   }
 #endif
 
-  check_directory(buff, block_size, 0);
+  check_directory(buff, block_size, 0, (uint) -1);
   empty_space= uint2korr(buff + EMPTY_SPACE_OFFSET);
   dir= dir_entry_pos(buff, block_size, record_number);
   length= uint2korr(dir + 2);
@@ -3963,7 +4069,7 @@ static int delete_dir_entry(uchar *buff, uint block_size, uint record_number,
 
   *empty_space_res= empty_space;
 
-  check_directory(buff, block_size, 0);
+  check_directory(buff, block_size, 0, empty_space);
   DBUG_RETURN(0);
 }
 
@@ -4165,7 +4271,8 @@ my_bool _ma_delete_block_record(MARIA_HA *info, const uchar *record)
     log_pos= log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE + PAGE_STORE_SIZE;
     dirpos_store(log_pos, record_number);
     log_pos+= DIRPOS_STORE_SIZE;
-    int2store(log_pos, info->cur_row.head_length);
+    int2store(log_pos, info->cur_row.head_length -
+              info->cur_row.header_length);
     log_pos+= 2;
     pagerange_store(log_pos, info->cur_row.extents_count);
     log_pos+= PAGERANGE_STORE_SIZE;
@@ -4543,6 +4650,8 @@ int _ma_read_block_record2(MARIA_HA *info, uchar *record,
   cur_row->head_length= (uint) (end_of_data - data);
   cur_row->full_page_count= cur_row->tail_count= 0;
   cur_row->blob_length= 0;
+  /* Number of bytes in header that we don't need to write during undo */
+  cur_row->header_length= total_header_size[(flag & PRECALC_HEADER_BITMASK)]-1;
 
   if (flag & ROW_FLAG_TRANSID)
   {
@@ -4554,7 +4663,7 @@ int _ma_read_block_record2(MARIA_HA *info, uchar *record,
   }
 
   /* Skip trans header (for now, until we have MVCC csupport) */
-  data+= total_header_size[(flag & PRECALC_HEADER_BITMASK)];
+  data+= cur_row->header_length + 1 ;
   if (flag & ROW_FLAG_NULLS_EXTENDED)
     cur_null_bytes+= data[-1];
 
@@ -4950,7 +5059,10 @@ int _ma_read_block_record(MARIA_HA *info, uchar *record,
   uint offset;
   uint block_size= share->block_size;
   DBUG_ENTER("_ma_read_block_record");
-  DBUG_PRINT("enter", ("rowid: %lu", (long) record_pos));
+  DBUG_PRINT("enter", ("rowid: %lu  page: %lu  rownr: %u",
+                       (ulong) record_pos,
+                       (ulong) ma_recordpos_to_page(record_pos),
+                       ma_recordpos_to_dir_entry(record_pos)));
 
   offset= ma_recordpos_to_dir_entry(record_pos);
 
@@ -6698,7 +6810,7 @@ my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
   pgcache_page_no_t page;
   uint rownr;
   uchar *buff;
-  my_bool res= 1;
+  my_bool res;
   MARIA_PINNED_PAGE page_link;
   MARIA_SHARE *share= info->s;
   ha_checksum checksum;
@@ -6744,11 +6856,16 @@ my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
     goto err;
 
   res= 0;
-err:
+end:
   if (info->non_flushable_state)
     _ma_bitmap_flushable(info, -1);
   _ma_unpin_all_pages_and_finalize_row(info, lsn);
   DBUG_RETURN(res);
+
+err:
+  res= 1;
+  _ma_mark_file_crashed(share);
+  goto end;
 }
 
 
@@ -6971,6 +7088,10 @@ my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
   {
     DBUG_ASSERT(row.checksum == (share->calc_checksum)(info, record));
   }
+  /* Store same amount of data on head page as on original page */
+  row_pos.length= (length_on_head_page - 
+                   (extent_count + 1 - blocks->count) * ROW_EXTENT_SIZE);
+  set_if_bigger(row_pos.length, share->base.min_block_length);
   if (write_block_record(info, (uchar*) 0, record, &row,
                          blocks, blocks->block->org_bitmap_value != 0,
                          &row_pos, undo_lsn, 0))
@@ -6980,6 +7101,7 @@ my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
   DBUG_RETURN(0);
 
 err:
+  _ma_mark_file_crashed(share);
   if (info->non_flushable_state)
     _ma_bitmap_flushable(info, -1);
   _ma_unpin_all_pages_and_finalize_row(info, LSN_IMPOSSIBLE);
@@ -7010,7 +7132,7 @@ my_bool _ma_apply_undo_row_update(MARIA_HA *info, LSN undo_lsn,
   pgcache_page_no_t page;
   ha_checksum checksum_delta;
   uint rownr, field_length_header, extent_count, length_on_head_page;
-  int error= 1;
+  int error;
   DBUG_ENTER("_ma_apply_undo_row_update");
   LINT_INIT(checksum_delta);
 
@@ -7018,6 +7140,7 @@ my_bool _ma_apply_undo_row_update(MARIA_HA *info, LSN undo_lsn,
   header+= PAGE_STORE_SIZE;
   rownr= dirpos_korr(header);
   header+= DIRPOS_STORE_SIZE;
+
   record_pos= ma_recordpos(page, rownr);
   DBUG_PRINT("enter", ("rowid: %lu  page: %lu  rownr: %u",
                        (ulong) record_pos, (ulong) page, rownr));
@@ -7147,9 +7270,14 @@ my_bool _ma_apply_undo_row_update(MARIA_HA *info, LSN undo_lsn,
     goto err;
 
   error= 0;
-err:
+end:
   my_free(current_record, MYF(0));
   DBUG_RETURN(error);
+
+err:
+  error= 1;
+  _ma_mark_file_crashed(share);
+  goto end;
 }
 
 
@@ -7229,3 +7357,25 @@ void maria_ignore_trids(MARIA_HA *info)
     info->trn->min_read_from= ~(TrID) 0;
   }
 }
+
+
+#ifndef DBUG_OFF
+
+/* The following functions are useful to call from debugger */
+
+void _ma_print_block_info(uchar *buff)
+{
+  LSN lsn= lsn_korr(buff);
+
+  printf("LSN: %lu,0x%lx  type: %u  dir_entries: %u  dir_free: %u  empty_space: %u\n",
+         LSN_IN_PARTS(lsn),
+         (uint)buff[PAGE_TYPE_OFFSET],
+         (uint)buff[DIR_COUNT_OFFSET],
+         (uint)buff[DIR_FREE_OFFSET],
+         (uint) uint2korr(buff + EMPTY_SPACE_OFFSET));
+  printf("Start of directory: %lu\n",
+         maria_block_size - PAGE_SUFFIX_SIZE -
+         (uint) buff[DIR_COUNT_OFFSET] * DIR_ENTRY_SIZE);
+  _ma_print_directory(stdout, buff, maria_block_size);
+}
+#endif
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index 28b808cf27e..086cff67dab 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -1930,7 +1930,7 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
         _ma_check_print_error(param,
                               "Page %9s:  Wrong data in bitmap.  Page_type: "
                               "%d  full: %d  empty_space: %u  Bitmap-bits: %d",
-                              llstr(page, llbuff), full_dir, page_type,
+                              llstr(page, llbuff), page_type, full_dir,
                               empty_space, bitmap_pattern);
       if (param->err_count++ > MAXERR || !(param->testflag & T_VERBOSE))
         goto err;
diff --git a/storage/maria/ma_checkpoint.c b/storage/maria/ma_checkpoint.c
index b75267f81f1..cf13cee9452 100644
--- a/storage/maria/ma_checkpoint.c
+++ b/storage/maria/ma_checkpoint.c
@@ -789,7 +789,6 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
         not seen again in the loop.
       */
       share->in_checkpoint= MARIA_CHECKPOINT_LOOKS_AT_ME;
-      /** @todo avoid strlen() */
       total_names_length+= share->open_file_name.length;
     }
   }
diff --git a/storage/maria/ma_key_recover.c b/storage/maria/ma_key_recover.c
index 4dc902f9418..82f8099fe6a 100644
--- a/storage/maria/ma_key_recover.c
+++ b/storage/maria/ma_key_recover.c
@@ -64,8 +64,9 @@ void _ma_unpin_all_pages(MARIA_HA *info, LSN undo_lsn)
       builds.
     */
 #ifdef EXTRA_DEBUG
-    DBUG_ASSERT(!pinned_page->changed ||
-                undo_lsn != LSN_IMPOSSIBLE || !info->s->now_transactional);
+    DBUG_ASSERT((!pinned_page->changed ||
+                 undo_lsn != LSN_IMPOSSIBLE || !info->s->now_transactional) ||
+                (info->s->state.changed & STATE_CRASHED));
 #endif
     pagecache_unlock_by_link(info->s->pagecache, pinned_page->link,
                              pinned_page->unlock, PAGECACHE_UNPIN,
@@ -923,7 +924,7 @@ uint _ma_apply_redo_index(MARIA_HA *info,
       if (length < 0)
         bmove(buff + page_offset, buff + page_offset - length,
               page_length - page_offset + length);
-      else
+      else if (page_length != page_offset)
         bmove_upp(buff + page_length + length, buff + page_length,
                   page_length - page_offset);
       page_length+= length;
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index 58f16658957..a10fac9a15d 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -494,8 +494,10 @@ end:
   }
   else if (!error && max_trid_in_control_file != max_long_trid)
   {
-    /* Set max trid in log file so that one can run maria_chk on the tables */
-    max_trid_in_control_file= trnman_get_max_trid();
+    /*
+      maria_end() will set max trid in log file so that one can run
+      maria_chk on the tables
+    */
     maria_recovery_changed_data= 1;
   }
 
@@ -2395,6 +2397,7 @@ static int run_redo_phase(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply)
   struct st_translog_scanner_data scanner;
   int len;
   uint i;
+  DBUG_ENTER("run_redo_phase");
 
   /* install hooks for execution */
 #define install_redo_exec_hook(R)                                        \
@@ -2459,7 +2462,7 @@ static int run_redo_phase(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply)
   {
     tprint(tracef, "checkpoint address refers to the log end log or "
            "log is empty, nothing to do.\n");
-    return 0;
+    DBUG_RETURN(0);
   }
 
   len= translog_read_record_header(lsn, &rec);
@@ -2467,12 +2470,12 @@ static int run_redo_phase(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply)
   if (len == RECHEADER_READ_ERROR)
   {
     eprint(tracef, "Failed to read header of the first record.");
-    return 1;
+    DBUG_RETURN(1);
   }
   if (translog_scanner_init(lsn, 1, &scanner, 1))
   {
     tprint(tracef, "Scanner init failed\n");
-    return 1;
+    DBUG_RETURN(1);
   }
   for (i= 1;;i++)
   {
@@ -2525,7 +2528,7 @@ static int run_redo_phase(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply)
                    LSN_IN_PARTS(rec2.lsn));
             translog_destroy_scanner(&scanner);
             translog_free_record_header(&rec);
-            return(0);
+            DBUG_RETURN(0);
           }
 
           if (translog_scanner_init(rec2.lsn, 1, &scanner2, 1))
@@ -2623,12 +2626,12 @@ static int run_redo_phase(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply)
     fflush(stderr);
     procent_printed= 1;
   }
-  return 0;
+  DBUG_RETURN(0);
 
 err:
   translog_destroy_scanner(&scanner);
   translog_free_record_header(&rec);
-  return 1;
+  DBUG_RETURN(1);
 }
 
 
@@ -3167,6 +3170,8 @@ static LSN parse_checkpoint_record(LSN lsn)
     return LSN_ERROR;
   next_dirty_page_in_pool= dirty_pages_pool;
   minimum_rec_lsn_of_dirty_pages= LSN_MAX;
+  if (maria_recovery_verbose)
+    tprint(tracef, "Table_id  Is_index       Page_id    Rec_lsn\n");
   for (i= 0; i < nb_dirty_pages ; i++)
   {
     pgcache_page_no_t page_id;
@@ -3183,6 +3188,9 @@ static LSN parse_checkpoint_record(LSN lsn)
     if (new_page((is_index << 16) | table_id,
                  page_id, rec_lsn, next_dirty_page_in_pool++))
       return LSN_ERROR;
+    if (maria_recovery_verbose)
+      tprint(tracef, "%8u  %8u  %12lu    %lu,0x%lx\n", (uint) table_id,
+             (uint) is_index, (ulong) page_id, LSN_IN_PARTS(rec_lsn));
     set_if_smaller(minimum_rec_lsn_of_dirty_pages, rec_lsn);
   }
   /* after that, there will be no insert/delete into the hash */
diff --git a/storage/maria/ma_static.c b/storage/maria/ma_static.c
index d870b034eb8..917385f9568 100644
--- a/storage/maria/ma_static.c
+++ b/storage/maria/ma_static.c
@@ -37,7 +37,7 @@ my_bool maria_flush= 0, maria_single_user= 0;
 my_bool maria_delay_key_write= 0, maria_page_checksums= 1;
 my_bool maria_inited= FALSE;
 my_bool maria_in_ha_maria= FALSE; /* If used from ha_maria or not */
-my_bool maria_recovery_changed_data= 0;
+my_bool maria_recovery_changed_data= 0, maria_recovery_verbose= 0;
 pthread_mutex_t THR_LOCK_maria;
 #if defined(THREAD) && !defined(DONT_USE_RW_LOCKS)
 ulong maria_concurrent_insert= 2;
diff --git a/storage/maria/ma_test2.c b/storage/maria/ma_test2.c
index 081cac8a2aa..9e2f32f767b 100644
--- a/storage/maria/ma_test2.c
+++ b/storage/maria/ma_test2.c
@@ -83,6 +83,9 @@ int main(int argc, char *argv[])
   if (! async_io)
     my_disable_async_io=1;
 
+  /* If we sync or not have no affect on this test */
+  my_disable_sync= 1;
+
   maria_data_root= (char *)".";
   /* Maria requires that we always have a page cache */
   if (maria_init() ||
@@ -351,7 +354,10 @@ int main(int argc, char *argv[])
       key3[atoi((char*) read_record+keyinfo[2].seg[0].start)]=0;
     }
     else
+    {
       puts("Warning: Skipping delete test because no dupplicate keys");
+      break;
+    }
   }
   if (testflag == 3)
     goto end;
diff --git a/storage/maria/ma_write.c b/storage/maria/ma_write.c
index 6ce7a4179e8..6a5c2995ae5 100644
--- a/storage/maria/ma_write.c
+++ b/storage/maria/ma_write.c
@@ -1973,6 +1973,21 @@ my_bool _ma_log_change(MARIA_PAGE *ma_page, const uchar *key_pos, uint length,
 /**
    @brief Write log entry for page splitting
 
+   @fn     _ma_log_split()
+   @param
+     ma_page		Page that is changed
+     org_length	        Original length of page
+     new_length		New length of page
+     key_pos		Where key is inserted on page (may be 0 if no key)
+     key_length		Number of bytes changed at key_pos
+     move_length	Number of bytes moved at key_pos to make room for key
+     prefix_or_suffix   KEY_OP_NONE	    Ignored
+   			KEY_OP_ADD_PREFIX   Add data to start of page
+			KEY_OP_ADD_SUFFIX   Add data to end of page
+     data		What data was added
+     data_length	Number of bytes added first or last
+     changed_length	Number of bytes changed first or last.
+
    @note
      Write log entry for page that has got a key added to the page under
      one and only one of the following senarios:
@@ -1980,9 +1995,6 @@ my_bool _ma_log_change(MARIA_PAGE *ma_page, const uchar *key_pos, uint length,
      - Data is added to end of page
      - Data added at front of page
 
-   @param prefix_or_suffix  KEY_OP_NONE		Ignored
-   			    KEY_OP_ADD_PREFIX   Add data to start of page
-			    KEY_OP_ADD_SUFFIX   Add data to end of page
 
 */
 
@@ -2005,6 +2017,8 @@ static my_bool _ma_log_split(MARIA_PAGE *ma_page,
   DBUG_PRINT("enter", ("page: %lu  org_length: %u  new_length: %u",
                        (ulong) ma_page->pos, org_length, new_length));
 
+  DBUG_ASSERT(changed_length >= data_length);
+
   log_pos= log_data + FILEID_STORE_SIZE;
   page= ma_page->pos / info->s->block_size;
   page_store(log_pos, page);
@@ -2027,6 +2041,7 @@ static my_bool _ma_log_split(MARIA_PAGE *ma_page,
     log_pos+= 3;
     translog_parts= 1;
     extra_length= 0;
+    DBUG_ASSERT(data_length == 0);
   }
   else
   {
@@ -2046,10 +2061,13 @@ static my_bool _ma_log_split(MARIA_PAGE *ma_page,
       log_pos[0]= KEY_OP_DEL_SUFFIX;
       int2store(log_pos + 1, diff);
       log_pos+= 3;
+      DBUG_ASSERT(data_length == 0);            /* Page is shortened */
+      DBUG_ASSERT(offset <= org_length - diff);
     }
     else
     {
       DBUG_ASSERT(new_length == org_length + move_length + data_length);
+      DBUG_ASSERT(offset <= org_length);
     }
 
     log_pos[0]= KEY_OP_OFFSET;
diff --git a/storage/maria/maria_chk.c b/storage/maria/maria_chk.c
index 511488b977a..18426e468cd 100644
--- a/storage/maria/maria_chk.c
+++ b/storage/maria/maria_chk.c
@@ -1252,7 +1252,7 @@ static int maria_chk(HA_CHECK *param, char *filename)
   }
   else if ((param->testflag & T_CHECK) || !(param->testflag & T_AUTO_INC))
   {
-    if (!(param->testflag & T_SILENT) || param->testflag & T_INFO)
+    if (!(param->testflag & T_VERY_SILENT) || param->testflag & T_INFO)
       printf("Checking MARIA file: %s\n",filename);
     if (!(param->testflag & T_SILENT))
       printf("Data records: %7s   Deleted blocks: %7s\n",
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index e9980cf9695..bde04186931 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -459,8 +459,9 @@ typedef struct st_maria_row
   uint *null_field_lengths;             /* All null field lengths */
   ulong *blob_lengths;                  /* Length for each blob */
   ulong min_length, normal_length, char_length, varchar_length;
-  ulong blob_length, head_length, total_length;
+  ulong blob_length, total_length;
   size_t extents_buffer_length;         /* Size of 'extents' buffer */
+  uint head_length, header_length;
   uint field_lengths_length;            /* Length of data in field_lengths */
   uint extents_count;                   /* number of extents in 'extents' */
   uint full_page_count, tail_count;     /* For maria_chk */
@@ -797,6 +798,7 @@ extern uint maria_quick_table_bits;
 extern char *maria_data_root;
 extern uchar maria_zero_string[];
 extern my_bool maria_inited, maria_in_ha_maria, maria_recovery_changed_data;
+extern my_bool maria_recovery_verbose;
 extern HASH maria_stored_state;
 extern int (*maria_create_trn_hook)(MARIA_HA *);
 
diff --git a/storage/maria/maria_pack.c b/storage/maria/maria_pack.c
index 0673f571a6b..167d0af3078 100644
--- a/storage/maria/maria_pack.c
+++ b/storage/maria/maria_pack.c
@@ -261,8 +261,8 @@ static struct my_option my_long_options[] =
   {"backup", 'b', "Make a backup of the table as table_name.OLD.",
    &backup, &backup, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"character-sets-dir", OPT_CHARSETS_DIR_MP,
-   "Directory where character sets are.", &charsets_dir,
-   &charsets_dir, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+   "Directory where character sets are.", (char**) &charsets_dir,
+   (char**) &charsets_dir, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"debug", '#', "Output debug log. Often this is 'd:t:o,filename'.",
    0, 0, 0, GET_STR, OPT_ARG, 0, 0, 0, 0, 0, 0},
   {"force", 'f',
diff --git a/storage/maria/maria_read_log.c b/storage/maria/maria_read_log.c
index 1002260afe5..98bc638c77f 100644
--- a/storage/maria/maria_read_log.c
+++ b/storage/maria/maria_read_log.c
@@ -33,7 +33,7 @@ static my_bool opt_display_only, opt_apply, opt_apply_undo, opt_silent;
 static my_bool opt_check;
 static const char *opt_tmpdir;
 static ulong opt_page_buffer_size;
-static ulonglong opt_start_from_lsn, opt_end_lsn;
+static ulonglong opt_start_from_lsn, opt_end_lsn, opt_start_from_checkpoint;
 static MY_TMPDIR maria_chk_tmpdir;
 
 
@@ -94,7 +94,6 @@ int main(int argc, char **argv)
   if (opt_display_only)
     printf("You are using --display-only, NOTHING will be written to disk\n");
 
-  /* LSN could be also --start-from-lsn=# */
   lsn= translog_first_lsn_in_log();
   if (lsn == LSN_ERROR)
   {
@@ -105,8 +104,16 @@ int main(int argc, char **argv)
   {
      fprintf(stdout, "The transaction log is empty\n");
   }
-  fprintf(stdout, "The transaction log starts from lsn (%lu,0x%lx)\n",
-          LSN_IN_PARTS(lsn));
+  if (opt_start_from_checkpoint && !opt_start_from_lsn &&
+      last_checkpoint_lsn != LSN_IMPOSSIBLE)
+  {
+    lsn= LSN_IMPOSSIBLE;             /* LSN set in maria_apply_log() */
+    fprintf(stdout, "Starting from checkpoint (%lu,0x%lx)\n",
+            LSN_IN_PARTS(last_checkpoint_lsn));
+  }
+  else
+    fprintf(stdout, "The transaction log starts from lsn (%lu,0x%lx)\n",
+            LSN_IN_PARTS(lsn));
 
   if (opt_start_from_lsn)
   {
@@ -183,7 +190,7 @@ static struct my_option my_long_options[] =
   {"help", '?', "Display this help and exit.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"display-only", 'd', "display brief info read from records' header",
-   (uchar **) &opt_display_only, (uchar **) &opt_display_only, 0, GET_BOOL,
+   &opt_display_only, &opt_display_only, 0, GET_BOOL,
    NO_ARG,0, 0, 0, 0, 0, 0},
   {"maria-log-dir-path", 'l',
     "Path to the directory where to store transactional log",
@@ -197,11 +204,17 @@ static struct my_option my_long_options[] =
   { "start-from-lsn", 'o', "Start reading log from this lsn",
     &opt_start_from_lsn, &opt_start_from_lsn,
     0, GET_ULL, REQUIRED_ARG, 0, 0, ~(longlong) 0, 0, 0, 0 },
+  {"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",
-   (uchar **) &opt_silent, (uchar **) &opt_silent, 0,
+   &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},
   {"tmpdir", 't', "Path for temporary files. Multiple paths can be specified, "
    "separated by "
diff --git a/storage/maria/unittest/ma_test_recovery.pl b/storage/maria/unittest/ma_test_recovery.pl
index 58b9cc3b56b..130ab4131fe 100755
--- a/storage/maria/unittest/ma_test_recovery.pl
+++ b/storage/maria/unittest/ma_test_recovery.pl
@@ -300,7 +300,7 @@ sub check_table_is_same
   $com.= "| grep -v \"file length\" | grep -v \"LSNs:\" | grep -v \"UUID:\" > $tmp/maria_chk_message.txt 2>&1";
   $res= `$com`;
   print MY_LOG $res;
-  $res= `$maria_exe_path/maria_chk$suffix -s -e --read-only $table`;
+  $res= `$maria_exe_path/maria_chk$suffix -ss -e --read-only $table`;
   print MY_LOG $res;
   $checksum2= `$maria_exe_path/maria_chk$suffix -dss $table`;
   if ("$checksum" ne "$checksum2")
@@ -415,7 +415,7 @@ sub physical_cmp
         # save original tables to restore them later
         copy("$table.MAD", "$tmp/before_zerofill$table_no.MAD") || die();
         copy("$table.MAI", "$tmp/before_zerofill$table_no.MAI") || die();
-        $com= "$maria_exe_path/maria_chk$suffix -s --zerofill-keep-lsn $table";
+        $com= "$maria_exe_path/maria_chk$suffix -ss --zerofill-keep-lsn $table";
         $res= `$com`;
         print MY_LOG $res;
         $table_no= $table_no + 1;
diff --git a/storage/xtradb/buf/buf0lru.c b/storage/xtradb/buf/buf0lru.c
index 65fdf342e4f..7b01c4aec50 100644
--- a/storage/xtradb/buf/buf0lru.c
+++ b/storage/xtradb/buf/buf0lru.c
@@ -2131,7 +2131,7 @@ ibool
 buf_LRU_file_dump(void)
 /*===================*/
 {
-	os_file_t	dump_file = -1;
+	os_file_t	dump_file = (os_file_t) -1;
 	ibool		success;
 	byte*		buffer_base = NULL;
 	byte*		buffer = NULL;
@@ -2221,7 +2221,7 @@ buf_LRU_file_dump(void)
 
 	ret = TRUE;
 end:
-	if (dump_file != -1)
+	if (dump_file != (os_file_t) -1)
 		os_file_close(dump_file);
 	if (buffer_base)
 		ut_free(buffer_base);
@@ -2235,7 +2235,7 @@ ibool
 buf_LRU_file_restore(void)
 /*======================*/
 {
-	os_file_t	dump_file = -1;
+	os_file_t	dump_file = (os_file_t) -1;
 	ibool		success;
 	byte*		buffer_base = NULL;
 	byte*		buffer = NULL;
@@ -2326,7 +2326,7 @@ buf_LRU_file_restore(void)
 		" (requested: %lu, read: %lu)\n", req, reads);
 	ret = TRUE;
 end:
-	if (dump_file != -1)
+	if (dump_file != (os_file_t) -1)
 		os_file_close(dump_file);
 	if (buffer_base)
 		ut_free(buffer_base);
diff --git a/storage/xtradb/fil/fil0fil.c b/storage/xtradb/fil/fil0fil.c
index 08789691159..189c0e706f9 100644
--- a/storage/xtradb/fil/fil0fil.c
+++ b/storage/xtradb/fil/fil0fil.c
@@ -3044,7 +3044,7 @@ fil_open_single_table_tablespace(
 		dulint		new_id[31];
 		ulint		root_page[31];
 		ulint		n_index;
-		os_file_t	info_file = -1;
+		os_file_t	info_file = (os_file_t) -1;
 		char*		info_file_path;
 		ulint	i;
 		int		len;
@@ -3130,7 +3130,7 @@ fil_open_single_table_tablespace(
 		}
 
 skip_info:
-		if (info_file != -1)
+		if (info_file != (os_file_t) -1)
 			os_file_close(info_file);
 
 		/*
diff --git a/storage/xtradb/handler/i_s.cc b/storage/xtradb/handler/i_s.cc
index cc30bdee1dd..7cb4a7ac718 100644
--- a/storage/xtradb/handler/i_s.cc
+++ b/storage/xtradb/handler/i_s.cc
@@ -3459,7 +3459,7 @@ i_s_innodb_table_stats_fill(
 
 		field_store_string(i_s_table->field[0], buf);
 		field_store_string(i_s_table->field[1], ptr);
-		i_s_table->field[2]->store(table->stat_n_rows);
+		i_s_table->field[2]->store(table->stat_n_rows, 1);
 		i_s_table->field[3]->store(table->stat_clustered_index_size);
 		i_s_table->field[4]->store(table->stat_sum_of_other_index_sizes);
 		i_s_table->field[5]->store(table->stat_modified_counter);
diff --git a/storage/xtradb/srv/srv0srv.c b/storage/xtradb/srv/srv0srv.c
index 86bf309bac1..bc2dd562697 100644
--- a/storage/xtradb/srv/srv0srv.c
+++ b/storage/xtradb/srv/srv0srv.c
@@ -2897,7 +2897,7 @@ loop:
 					if (bpl) {
 retry_flush_batch:
 						n_pages_flushed = buf_flush_batch(BUF_FLUSH_LIST,
-									bpl,
+									(ulint) bpl,
 									oldest_lsn + (lsn - lsn_old));
 						if (n_pages_flushed == ULINT_UNDEFINED) {
 							os_thread_sleep(5000);