WL#3072 - Maria Recovery

Bulk insert: don't log REDO/UNDO for rows, log one UNDO which will
truncate files; this is an optimization and a bugfix (table was left
half-repaired by crash).
Repair: mark table crashed-on-repair at start, bump skip_redo_lsn at start,
this is easier for recovery (tells it to skip old REDOs or even UNDO
phase) and user (tells it to repair) in case of crash, sync files
in the end.
Recovery skips missing or corrupted table and moves to next record
(in REDO or UNDO phase) to be more robust; warns if happens in UNDO phase.
Bugfix for UNDO_KEY_DELETE_WITH_ROOT (tested in ma_test_recovery)
and maria_enable_indexes().
Create missing bitmaps when needed (there can be more than one to create,
in rare cases), log a record for this.


include/myisamchk.h:
  new flag: bulk insert repair mustn't bump create_rename_lsn
mysql-test/lib/mtr_report.pl:
  skip normal warning in maria-recovery.test
mysql-test/r/maria-recovery.result:
  result: crash before bulk insert is committed, causes proper rollback,
  and crash right after OPTIMIZE replaces index file with new index file
  leads to table marked corrupted and recovery not failing.
mysql-test/t/maria-recovery.test:
  - can't check the table or it would commit the transaction,
  but check is made after recovery.
  - test of crash before bulk-insert-with-repair is committed
  (to see if it is rolled back), and of crash after OPTIMIZE has replaced
  index file but not finished all operations (to see if recovery fails -
  it used to assert when trying to execute an old REDO on the new
  index).
storage/maria/CMakeLists.txt:
  new file
storage/maria/Makefile.am:
  new file
storage/maria/ha_maria.cc:
  - If bulk insert on a transactional table using an index repair:
  table is initially empty, so don't log REDO/UNDO for data rows
  (optimization), just log an UNDO_BULK_INSERT_WITH_REPAIR
  which will, if executed, empty the data and index file. Re-enable
  logging in end_bulk_insert().
  - write log record for repair operation only after it's fully done,
  index sort including (maria_repair*() used to write the log record).
  - Adding back file->trn=NULL which was removed by mistake earlier.
storage/maria/ha_maria.h:
  new member (see ha_maria.cc)
storage/maria/ma_bitmap.c:
  Functions to create missing bitmaps:
  - one function which creates missing bitmaps in page cache, except
  the missing one with max offset which it does not put into page cache
  as it will be modified very soon.
  - one function which the one above calls, and creates bitmaps in page
  cache
  - one function to execute REDO_BITMAP_NEW_PAGE which uses the second
  one above.
storage/maria/ma_blockrec.c:
  - when logging REDO_DELETE_ALL, not only 'records' and 'checksum'
  has to be reset under log's mutex.
  - execution of REDO_INSERT_ROW_BLOBS now checks the dirty pages' list
  - execution of UNDO_BULK_INSERT_WITH_REPAIR
storage/maria/ma_blockrec.h:
  new functions
storage/maria/ma_check.c:
  - table-flush-before-repair is moved to a separate function reused
  by maria_sort_index(); syncing is added
  - maria_repair() is allowed to re-enable logging only if it is the one
  which disabled it.
  - "_ma_flush_table_files_after_repair" was a bad name, it's not after
  repair now, and it should not sync as we do more changes to the files
  shortly after (sync is postponed to when writing the log record)
  - REDO_REPAIR record should be written only after all repair
  operations (in particular after sorting index in ha_mara::repair())
  - close to the end of repair by sort, flushing of pages must happen
  also in the non-quick case, to prepare for the sync at end.
  - in parallel repair, some page flushes are not needed as done
  by initialize_variables_for_repair().
storage/maria/ma_create.c:
  Update skip_redo_lsn, create_rename_lsn optionally.
storage/maria/ma_delete_all.c:
  Need to sync files at end of maria_delete_all_rows(), if transactional.
storage/maria/ma_extra.c:
  During repair, we sometimes call _ma_flush_table_files() (via
  _ma_flush_table_files_before_swap()) while there is a WRITE_CACHE.
storage/maria/ma_key_recover.c:
  - when we see CLR_END for UNDO_BULK_INSERT_WITH_REPAIR, re-enable
  indices.
  - fixing bug: _ma_apply_undo_key_delete() parsed UNDO_KEY_DELETE_WITH_ROOT
  wrongly, leading to recovery failure
storage/maria/ma_key_recover.h:
  new prototype
storage/maria/ma_locking.c:
  DBUG_VOID_RETURN missing
storage/maria/ma_loghandler.c:
  UNDO for bulk insert with repair, and REDO for creating bitmaps.
  LOGREC_FIRST_FREE to not have to change the for() every time we
  add a new record type.
storage/maria/ma_loghandler.h:
  new UNDO and REDO
storage/maria/ma_open.c:
  Move share.kfile.file=kfile up a bit, so that _ma_update_state_lsns()
  can get its value, this fixes a bug where LSN_REPAIRED_BY_MARIA_CHK
  was not corrected on disk by maria_open().
  Store skip_redo_lsn in index' header.
  maria_enable_indexes() had a bug for BLOCK_RECORD, where an empty
  file has one page, not 0 bytes.
storage/maria/ma_recovery.c:
  - Skip a corrupted, missing, or repaired-with-maria_chk, table in
  recovery: don't fail, just go to next REDO or UNDO; but if an UNDO
  is skipped in UNDO phase we issue warnings.
  - Skip REDO|UNDO in REDO phase if <skip_redo_lsn.
  - If UNDO phase fails, delete transactions to not make trnman
  assert.
  - Update skip_redo_lsn when playing REDO_CREATE_TABLE
  - Don't record UNDOs for old transactions which we don't know (long_trid==0)
  - Bugfix for UNDO_KEY_DELETE_WITH_ROOT (see ma_key_recover.c)
  - Execution of UNDO_BULK_INSERT_WITH_REPAIR
  - Don't try to find a page number in REDO_DELETE_ALL
  - Pieces moved to ma_recovery_util.c
storage/maria/ma_rename.c:
  name change
storage/maria/ma_static.c:
  I modified layout of the index' header (inserted skip_redo_lsn in its middle)
storage/maria/ma_test2.c:
  allow breaking the test towards the end, tests execution of
  UNDO_KEY_DELETE_WITH_ROOT
storage/maria/ma_test_recovery.expected:
  6 as testflag instead of 4
storage/maria/ma_test_recovery:
  Increase the amount of rollback work to do when testing recovery
  with ma_test2; this reproduces the UNDO_KEY_DELETE_WITH_ROOT bug.
storage/maria/maria_chk.c:
  skip_redo_lsn should be updated too, for consistency.
  Write a REDO_REPAIR after all operations (including sort-records)
  have been done.
  No reason to flush blocks after maria_chk_data_link() and
  maria_sort_records(), there is maria_close() in the end.
  write_log_record() is a function, to not clutter maria_chk().
storage/maria/maria_def.h:
  New member skip_redo_lsn in the state, and comments
storage/maria/maria_pack.c:
  skip_redo_lsn should be updated too, for consistency
storage/maria/ma_recovery_util.c:
  _ma_redo_not_needed_for_page(), defined in ma_recovery.c, is needed
  by ma_blockrec.c; this causes link issues, resolved by putting
  _ma_redo_not_needed_for_page() into a new file (so that it is not
  in the same file as repair-related objects of ma_recovery.c).
storage/maria/ma_recovery_util.h:
  new file

include/myisamchk.h

@@ -60,6 +60,8 @@
 #define T_WAIT_FOREVER          (1L << 30)
 #define T_WRITE_LOOP            ((ulong) 1L << 31)
 #define T_ZEROFILL              ((ulonglong) 1L << 32)
+/** If repair should not bump create_rename_lsn */
+#define T_NO_CREATE_RENAME_LSN  ((ulonglong) 1L << 33)
 
 #define T_REP_ANY               (T_REP | T_REP_BY_SORT | T_REP_PARALLEL)
 

mysql-test/lib/mtr_report.pl

@@ -366,7 +366,9 @@ sub mtr_report_stats ($) {
 		/Slave: Key column 'c6'.* 1072/ or
 
                 # maria-recovery.test has warning about missing log file
-                /Can't get stat of '.*maria_log.00*/
+                /Can't get stat of '.*maria_log.00/ or
+                # and about marked-corrupted table
+                /Table '.\/mysqltest\/t1' is crashed, skipping it. Please repair it with maria_chk -r/
 	       )
             {
               next;                       # Skip these lines

mysql-test/r/maria-recovery.result

@@ -269,6 +269,39 @@ Checksum-check
 ok
 use mysqltest;
 drop table t1;
+Test of REPAIR's implicit commit
+create table t1 (a varchar(100), key(a)) engine=maria;
+insert into t1 values(3);
+flush table t1;
+* copied t1 for comparison
+lock tables t1 write;
+insert into t1 values (1);
+repair table t1;
+Table	Op	Msg_type	Msg_text
+mysqltest.t1	repair	status	OK
+insert into t1 values(2);
+select * from t1;
+a
+1
+2
+3
+SET SESSION debug="+d,maria_flush_whole_log,maria_flush_whole_page_cache,maria_crash";
+* crashing mysqld intentionally
+set global maria_checkpoint_interval=1;
+ERROR HY000: Lost connection to MySQL server during query
+* recovery happens
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+mysqltest.t1	check	status	OK
+* testing that checksum after recovery is as expected
+Checksum-check
+failure
+use mysqltest;
+select * from t1;
+a
+1
+3
+drop table t1;
 * TEST of recovery with blobs
 * shut down mysqld, removed logs, restarted it
 use mysqltest;
@@ -329,9 +362,6 @@ select a,length(b) from t1;
 a	length(b)
 1	8
 2	5
-check table t1;
-Table	Op	Msg_type	Msg_text
-mysqltest.t1	check	status	OK
 SET SESSION debug="+d,maria_flush_whole_log,maria_crash";
 * crashing mysqld intentionally
 set global maria_checkpoint_interval=1;
@@ -346,6 +376,58 @@ Checksum-check
 ok
 use mysqltest;
 drop table t1;
+* TEST of recovery when crash before bulk-insert-with-repair is committed
+create table t1 (a varchar(100), key(a)) engine=maria;
+create table t2 (a varchar(100)) engine=myisam;
+set rand_seed1=12, rand_seed2=254;
+insert into t2 values (rand());
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t1 values(30);
+flush table t1;
+* copied t1 for comparison
+lock tables t1 write, t2 read;
+delete from t1 limit 1;
+insert into t1 select * from t2;
+SET SESSION debug="+d,maria_flush_whole_log,maria_flush_whole_page_cache,maria_crash";
+* crashing mysqld intentionally
+set global maria_checkpoint_interval=1;
+ERROR HY000: Lost connection to MySQL server during query
+* recovery happens
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+mysqltest.t1	check	status	OK
+* testing that checksum after recovery is as expected
+Checksum-check
+ok
+use mysqltest;
+show keys from t1;
+Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment
+t1	1	a	1	a	A	1	NULL	NULL	YES	BTREE	
+drop table t1;
+* TEST of recovery when OPTIMIZE has replaced the index file and crash
+create table t1 (a varchar(100), key(a)) engine=maria;
+insert into t1 select (rand()) from t2;
+flush table t1;
+* copied t1 for comparison
+SET SESSION debug="+d,maria_flush_whole_log,maria_flush_whole_page_cache,maria_crash_sort_index";
+* crashing mysqld intentionally
+optimize table t1;
+ERROR HY000: Lost connection to MySQL server during query
+* recovery happens
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+mysqltest.t1	check	warning	Table is marked as crashed and last repair failed
+mysqltest.t1	check	status	OK
+* testing that checksum after recovery is as expected
+Checksum-check
+ok
+use mysqltest;
+drop table t1, t2;
 drop database mysqltest_for_feeding_recovery;
 drop database mysqltest_for_comparison;
 drop database mysqltest;

mysql-test/t/maria-recovery.test

@@ -235,7 +235,6 @@ let $mvr_crash_statement= set global maria_checkpoint_interval=1;
 drop table t1;
 
 --echo Test of REPAIR's implicit commit
-
 let $mms_tables=1;
 create table t1 (a varchar(100), key(a)) engine=maria;
 let $mvr_restore_old_snapshot=0;
@@ -284,7 +283,6 @@ while ($loop)
   dec $loop;
 }
 select a,length(b) from t1;
-check table t1;
 # we want recovery to run on the first snapshot made above
 let $mvr_restore_old_snapshot=1;
 let $mms_compare_physically=0;
@@ -293,6 +291,46 @@ let $mvr_crash_statement= set global maria_checkpoint_interval=1;
 -- source include/maria_verify_recovery.inc
 drop table t1;
 
+--echo * TEST of recovery when crash before bulk-insert-with-repair is committed
+create table t1 (a varchar(100), key(a)) engine=maria;
+create table t2 (a varchar(100)) engine=myisam;
+let $mvr_restore_old_snapshot=0;
+let $mms_compare_physically=0;
+let $mvr_crash_statement= set global maria_checkpoint_interval=1;
+let $mvr_debug_option="+d,maria_flush_whole_log,maria_flush_whole_page_cache,maria_crash";
+set rand_seed1=12, rand_seed2=254; # repeatable
+insert into t2 values (rand());
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t2 select (rand()) from t2;
+insert into t1 values(30);
+-- source include/maria_make_snapshot_for_comparison.inc
+lock tables t1 write, t2 read;
+delete from t1 limit 1;
+# 127 rows in t2, >100, so this will use repair-at-end
+insert into t1 select * from t2;
+-- source include/maria_verify_recovery.inc
+show keys from t1; # should be enabled
+drop table t1;
+
+--echo * TEST of recovery when OPTIMIZE has replaced the index file and crash
+create table t1 (a varchar(100), key(a)) engine=maria;
+let $mvr_restore_old_snapshot=0;
+let $mms_compare_physically=0;
+let $mvr_crash_statement= optimize table t1;
+let $mvr_debug_option="+d,maria_flush_whole_log,maria_flush_whole_page_cache,maria_crash_sort_index";
+insert into t1 select (rand()) from t2;
+-- source include/maria_make_snapshot_for_comparison.inc
+# Recovery will not fix the table, but we expect to see it marked
+# "crashed on repair".
+# Because crash is mild, the table is actually not corrupted, so the
+# "check table extended" done below fixes the table.
+-- source include/maria_verify_recovery.inc
+drop table t1, t2;
+
 # clean up everything
 let $mms_purpose=feeding_recovery;
 eval drop database mysqltest_for_$mms_purpose;

storage/maria/CMakeLists.txt

@@ -43,7 +43,7 @@ SET(MARIA_SOURCES ma_init.c ma_open.c ma_extra.c ma_info.c ma_rkey.c
             ma_sp_key.c ma_control_file.c ma_loghandler.c 
             ma_pagecache.c ma_pagecaches.c 
             ma_checkpoint.c ma_recovery.c ma_commit.c ma_pagecrc.c
-            ha_maria.h maria_def.h
+            ha_maria.h maria_def.h ma_recovery_util.c
 )
 
 IF(NOT SOURCE_SUBLIBS)

storage/maria/Makefile.am

@@ -72,7 +72,8 @@ noinst_HEADERS =	maria_def.h ma_rt_index.h ma_rt_key.h ma_rt_mbr.h \
 			ma_control_file.h ha_maria.h ma_blockrec.h \
 			ma_loghandler.h ma_loghandler_lsn.h ma_pagecache.h \
 			ma_checkpoint.h ma_recovery.h ma_commit.h \
-			trnman_public.h ma_check_standalone.h ma_key_recover.h
+			trnman_public.h ma_check_standalone.h ma_key_recover.h \
+			ma_recovery_util.h
 ma_test1_DEPENDENCIES=	$(LIBRARIES)
 ma_test1_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmaria.a \
                         $(top_builddir)/storage/myisam/libmyisam.a \
@@ -133,7 +134,7 @@ libmaria_a_SOURCES =	ma_init.c ma_open.c ma_extra.c ma_info.c ma_rkey.c \
 			ma_sp_key.c ma_control_file.c ma_loghandler.c \
 			ma_pagecache.c ma_pagecaches.c \
 			ma_checkpoint.c ma_recovery.c ma_commit.c \
-			ma_pagecrc.c
+			ma_pagecrc.c ma_recovery_util.c
 CLEANFILES =		test?.MA? FT?.MA? isam.log ma_test_all ma_rt_test.MA? sp_test.MA?
 
 SUFFIXES = .sh

storage/maria/ha_maria.cc

@@ -650,7 +650,7 @@ int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
                 HA_FILE_BASED | HA_CAN_GEOMETRY | MARIA_CANNOT_ROLLBACK |
                 HA_CAN_BIT_FIELD | HA_CAN_RTREEKEYS |
                 HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT),
-can_enable_indexes(1)
+can_enable_indexes(1), bulk_insert_with_repair_trans(FALSE)
 {}
 
 
@@ -1316,11 +1316,6 @@ int ha_maria::repair(THD *thd, HA_CHECK &param, bool do_optimize)
       param.testflag &= ~(T_REP_BY_SORT | T_REP_PARALLEL);
       error= maria_repair(&param, file, fixed_name,
                           test(param.testflag & T_QUICK));
-      /**
-         @todo RECOVERY BUG we do things with the index file
-         (maria_sort_index() after the above which already has logged the
-         record and bumped create_rename_lsn. Is it ok?
-      */
     }
     param.testflag= save_testflag;
     optimize_done= 1;
@@ -1392,8 +1387,11 @@ int ha_maria::repair(THD *thd, HA_CHECK &param, bool do_optimize)
   thd_proc_info(thd, old_proc_info);
   if (!thd->locked_tables)
     maria_lock_database(file, F_UNLCK);
-  DBUG_RETURN(error ? HA_ADMIN_FAILED :
-              !optimize_done ? HA_ADMIN_ALREADY_DONE : HA_ADMIN_OK);
+  error= error ? HA_ADMIN_FAILED :
+    (optimize_done ?
+     (write_log_record_for_repair(&param, file) ? HA_ADMIN_FAILED :
+      HA_ADMIN_OK) : HA_ADMIN_ALREADY_DONE);
+  DBUG_RETURN(error);
 }
 
 
@@ -1613,6 +1611,14 @@ int ha_maria::enable_indexes(uint mode)
     param.op_name= "recreating_index";
     param.testflag= (T_SILENT | T_REP_BY_SORT | T_QUICK |
                      T_CREATE_MISSING_KEYS | T_SAFE_REPAIR);
+    if (bulk_insert_with_repair_trans)
+    {
+      /*
+        Don't bump create_rename_lsn, because UNDO_BULK_INSERT_WITH_REPAIR
+        should not be skipped in case of crash during repair.
+      */
+      param.testflag|= T_NO_CREATE_RENAME_LSN;
+    }
     param.myf_rw &= ~MY_WAIT_IF_FULL;
     param.sort_buffer_length= THDVAR(thd,sort_buffer_size);
     param.stats_method= (enum_handler_stats_method)THDVAR(thd,stats_method);
@@ -1705,6 +1711,7 @@ void ha_maria::start_bulk_insert(ha_rows rows)
 
   can_enable_indexes= (maria_is_all_keys_active(file->s->state.key_map,
                                                 file->s->base.keys));
+  bulk_insert_with_repair_trans= FALSE;
 
   if (!(specialflag & SPECIAL_SAFE_MODE))
   {
@@ -1716,7 +1723,23 @@ void ha_maria::start_bulk_insert(ha_rows rows)
     */
     if (file->state->records == 0 && can_enable_indexes &&
         (!rows || rows >= MARIA_MIN_ROWS_TO_DISABLE_INDEXES))
+    {
       maria_disable_non_unique_index(file, rows);
+      if (file->s->now_transactional)
+      {
+        bulk_insert_with_repair_trans= TRUE;
+        write_log_record_for_bulk_insert_with_repair(file);
+        /*
+          Pages currently in the page cache have type PAGECACHE_LSN_PAGE, we
+          are not allowed to overwrite them with PAGECACHE_PLAIN_PAGE, so
+          throw them away. It is not losing data, because we just wrote and
+          forced an UNDO which will for sure empty the table if we crash.
+        */
+        _ma_flush_table_files(file, MARIA_FLUSH_DATA|MARIA_FLUSH_INDEX,
+                              FLUSH_IGNORE_CHANGED, FLUSH_IGNORE_CHANGED);
+        _ma_tmp_disable_logging_for_table(file, TRUE);
+      }
+    }
     else if (!file->bulk_insert &&
              (!rows || rows >= MARIA_MIN_ROWS_TO_USE_BULK_INSERT))
     {
@@ -1750,9 +1773,18 @@ int ha_maria::end_bulk_insert()
   int err;
   DBUG_ENTER("ha_maria::end_bulk_insert");
   maria_end_bulk_insert(file);
-  err= maria_extra(file, HA_EXTRA_NO_CACHE, 0);
-  DBUG_RETURN(err ? err : can_enable_indexes ?
-              enable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE) : 0);
+  if ((err= maria_extra(file, HA_EXTRA_NO_CACHE, 0)))
+    goto end;
+  if (can_enable_indexes)
+    err= enable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE);
+end:
+  if (bulk_insert_with_repair_trans)
+  {
+    DBUG_ASSERT(can_enable_indexes);
+    /* table was transactional just before start_bulk_insert() */
+    _ma_reenable_logging_for_table(file);
+  }
+  DBUG_RETURN(err);
 }
 
 
@@ -2131,7 +2163,7 @@ int ha_maria::external_lock(THD *thd, int lock_type)
       if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
         trans_register_ha(thd, TRUE, maria_hton);
     }
-    this->file->trn= trn;
+    file->trn= trn;
     if (!trnman_increment_locked_tables(trn))
     {
       trans_register_ha(thd, FALSE, maria_hton);
@@ -2156,6 +2188,7 @@ int ha_maria::external_lock(THD *thd, int lock_type)
   {
     _ma_reenable_logging_for_table(file);
     /** @todo zero file->trn also in commit and rollback */
+    file->trn= NULL;
     if (trn && trnman_has_locked_tables(trn))
     {
       if (!trnman_decrement_locked_tables(trn))

storage/maria/ha_maria.h

@@ -40,6 +40,8 @@ class ha_maria :public handler
   char *data_file_name, *index_file_name;
   enum data_file_type data_file_type;
   bool can_enable_indexes;
+  /** If a transactional table is doing bulk insert with repair */
+  bool bulk_insert_with_repair_trans;
   int repair(THD * thd, HA_CHECK &param, bool optimize);
   int zerofill(THD * thd, HA_CHECK_OPT *check_opt);
 

storage/maria/ma_blockrec.c

@@ -263,12 +263,16 @@
 
   09 00 F4 1F                   Start position 9, length 8180
   xx xx xx xx			Checksum
+
+  A data page is allowed to have a wrong CRC and header as long as it is
+  marked empty in the bitmap and its directory's count is 0.
 */
 
 #include "maria_def.h"
 #include "ma_blockrec.h"
 #include "trnman.h"
 #include "ma_key_recover.h"
+#include "ma_recovery_util.h"
 #include <lf.h>
 
 /*
@@ -5643,8 +5647,8 @@ my_bool write_hook_for_undo(enum translog_record_type type
 
 
 /**
-   @brief Sets the table's records count and checksum to 0, then calls the
-   generic REDO hook.
+   @brief Sets the table's records count and checksum and others to 0, then
+   calls the generic REDO hook.
 
    @return Operation status, always 0 (success)
 */
@@ -5655,9 +5659,8 @@ my_bool write_hook_for_redo_delete_all(enum translog_record_type type
                                        __attribute__ ((unused)),
                                        LSN *lsn, void *hook_arg)
 {
-  MARIA_SHARE *share= tbl_info->s;
   DBUG_ASSERT(tbl_info->state == &tbl_info->s->state.state);
-  share->state.state.records= share->state.state.checksum= 0;
+  _ma_reset_status(tbl_info);
   return write_hook_for_redo(type, trn, tbl_info, lsn, hook_arg);
 }
 
@@ -6186,7 +6189,9 @@ err:
    @brief Apply LOGREC_REDO_INSERT_ROW_BLOBS
 
    @param  info            Maria handler
-   @param  header          Header (without FILEID)
+   @parma  lsn             LSN to put on pages
+   @param  header          Header (with FILEID)
+   @param  redo_lsn        REDO record's LSN
 
    @note Write full pages (full head & blob pages)
 
@@ -6196,17 +6201,21 @@ err:
 */
 
 uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
-                                     LSN lsn, const uchar *header)
+                                     LSN lsn, const uchar *header,
+                                     LSN redo_lsn)
 {
   MARIA_SHARE *share= info->s;
   const uchar *data;
   uint      data_size= FULL_PAGE_SIZE(share->block_size);
   uint      blob_count, ranges;
+  uint16    sid;
   DBUG_ENTER("_ma_apply_redo_insert_row_blobs");
 
   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;
   blob_count= pagerange_korr(header);
@@ -6247,6 +6256,9 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
         enum pagecache_page_pin unpin_method;
         uint length;
 
+        if (_ma_redo_not_needed_for_page(sid, redo_lsn, page, FALSE))
+          continue;
+
         if (((page + 1) * share->block_size) >
             info->state->data_file_length)
         {
@@ -6355,7 +6367,7 @@ err:
  Applying of UNDO entries
 ****************************************************************************/
 
-/* Execute undo of a row insert (delete the inserted row) */
+/** Execute undo of a row insert (delete the inserted row) */
 
 my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
                                   const uchar *header)
@@ -6416,7 +6428,7 @@ err:
 }
 
 
-/* Execute undo of a row delete (insert the row back where it was) */
+/** Execute undo of a row delete (insert the row back where it was) */
 
 my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
                                   const uchar *header,
@@ -6652,7 +6664,7 @@ err:
 }
 
 
-/*
+/**
   Execute undo of a row update
 
   @fn _ma_apply_undo_row_update()
@@ -6816,6 +6828,34 @@ err:
 }
 
 
+/**
+  Execute undo of a bulk insert which used repair
+
+  @return Operation status
+    @retval 0      OK
+    @retval 1      Error
+*/
+
+my_bool _ma_apply_undo_bulk_insert_with_repair(MARIA_HA *info,
+                                               LSN undo_lsn)
+{
+  my_bool error;
+  LSN lsn;
+  DBUG_ENTER("_ma_apply_undo_bulk_insert_with_repair");
+  /*
+    We delete all rows, re-enable indices as bulk insert had disabled
+    non-unique ones.
+  */
+  error= (maria_delete_all_rows(info) ||
+          maria_enable_indexes(info) ||
+          /* we enabled indices so need '2' below */
+          _ma_state_info_write(info->s, 1|2|4) ||
+          _ma_write_clr(info, undo_lsn, LOGREC_UNDO_BULK_INSERT_WITH_REPAIR,
+                        FALSE, 0, &lsn, NULL));
+  DBUG_RETURN(error);
+}
+
+
 /**
   @brief Get the TRANSLOG_ADDRESS to flush up to
 

storage/maria/ma_blockrec.h

@@ -236,14 +236,18 @@ uint _ma_apply_redo_free_blocks(MARIA_HA *info, LSN lsn,
                                 const uchar *header);
 uint _ma_apply_redo_free_head_or_tail(MARIA_HA *info, LSN lsn,
                                       const uchar *header);
-uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
-                                     LSN lsn, const uchar *header);
+uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info, LSN lsn,
+                                     const uchar *header, LSN redo_lsn);
+my_bool _ma_apply_redo_bitmap_new_page(MARIA_HA *info, LSN lsn,
+                                       const uchar *header);
 my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
                                   const uchar *header);
 my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
                                   const uchar *header, size_t length);
 my_bool _ma_apply_undo_row_update(MARIA_HA *info, LSN undo_lsn,
                                   const uchar *header, size_t length);
+my_bool _ma_apply_undo_bulk_insert_with_repair(MARIA_HA *info,
+                                               LSN undo_lsn);
 
 my_bool write_hook_for_redo(enum translog_record_type type,
                             TRN *trn, MARIA_HA *tbl_info, LSN *lsn,

storage/maria/ma_create.c

@@ -1077,7 +1077,7 @@ int maria_create(const char *name, enum data_file_type datafile_type,
       DROP+CREATE happened (applying REDOs to the wrong table).
     */
     share.kfile.file= file;
-    if (_ma_update_create_rename_lsn_sub(&share, lsn, FALSE))
+    if (_ma_update_state_lsns_sub(&share, lsn, FALSE, TRUE))
       goto err;
     my_free(log_data, MYF(0));
   }
@@ -1288,43 +1288,47 @@ int _ma_initialize_data_file(MARIA_SHARE *share, File dfile)
 
 
 /**
-   @brief Writes create_rename_lsn and is_of_horizon to disk, can force.
+   @brief Writes create_rename_lsn, skip_redo_lsn and is_of_horizon to disk,
+   can force.
 
    This is for special cases where:
    - we don't want to write the full state to disk (so, not call
    _ma_state_info_write()) because some parts of the state may be
    currently inconsistent, or because it would be overkill
    - we must sync these LSNs immediately for correctness.
-   It acquires intern_lock to protect the two LSNs and state write.
+   It acquires intern_lock to protect the LSNs and state write.
 
    @param  share           table's share
    @param  do_sync         if the write should be forced to disk
+   @param  update_create_rename_lsn if this LSN should be updated or not
 
    @return Operation status
      @retval 0      ok
      @retval 1      error (disk problem)
 */
 
-int _ma_update_create_rename_lsn(MARIA_SHARE *share,
-                                 LSN lsn, my_bool do_sync)
+int _ma_update_state_lsns(MARIA_SHARE *share, LSN lsn, my_bool do_sync,
+                          my_bool update_create_rename_lsn)
 {
   int res;
   pthread_mutex_lock(&share->intern_lock);
-  res= _ma_update_create_rename_lsn_sub(share, lsn, do_sync);
+  res= _ma_update_state_lsns_sub(share, lsn, do_sync,
+                                 update_create_rename_lsn);
   pthread_mutex_unlock(&share->intern_lock);
   return res;
 }
 
 
 /**
-   @brief Writes create_rename_lsn and is_of_horizon to disk, can force.
+   @brief Writes create_rename_lsn, skip_redo_lsn and is_of_horizon to disk,
+   can force.
 
-   Shortcut of _ma_update_create_rename_lsn() when we know that
-   intern_lock is not needed (when creating a table or opening it for the
-   first time).
+   Shortcut of _ma_update_state_lsns() when we know that intern_lock is not
+   needed (when creating a table or opening it for the first time).
 
    @param  share           table's share
    @param  do_sync         if the write should be forced to disk
+   @param  update_create_rename_lsn if this LSN should be updated or not
 
    @return Operation status
      @retval 0      ok
@@ -1338,15 +1342,18 @@ int _ma_update_create_rename_lsn(MARIA_SHARE *share,
 */
 #pragma optimize("",off)
 #endif
-int _ma_update_create_rename_lsn_sub(MARIA_SHARE *share,
-                                     LSN lsn, my_bool do_sync)
+int _ma_update_state_lsns_sub(MARIA_SHARE *share, LSN lsn, my_bool do_sync,
+                              my_bool update_create_rename_lsn)
 {
-  char buf[LSN_STORE_SIZE*2], *ptr;
+  char buf[LSN_STORE_SIZE * 3], *ptr;
   File file= share->kfile.file;
   DBUG_ASSERT(file >= 0);
   for (ptr= buf; ptr < (buf + sizeof(buf)); ptr+= LSN_STORE_SIZE)
     lsn_store(ptr, lsn);
-  share->state.is_of_horizon= share->state.create_rename_lsn= lsn;
+  share->state.skip_redo_lsn= share->state.is_of_horizon= lsn;
+  if (update_create_rename_lsn)
+  {
+    share->state.create_rename_lsn= lsn;
     if (share->id != 0)
     {
       /*
@@ -1360,6 +1367,9 @@ int _ma_update_create_rename_lsn_sub(MARIA_SHARE *share,
       */
       translog_deassign_id_from_share(share);
     }
+  }
+  else
+    lsn_store(buf, share->state.create_rename_lsn);
   return my_pwrite(file, buf, sizeof(buf),
                    sizeof(share->state.header) +
                    MARIA_FILE_CREATE_RENAME_LSN_OFFSET, MYF(MY_NABP)) ||

storage/maria/ma_delete_all.c

@@ -33,6 +33,7 @@ int maria_delete_all_rows(MARIA_HA *info)
 {
   MARIA_SHARE *share= info->s;
   my_bool log_record;
+  LSN lsn;
   DBUG_ENTER("maria_delete_all_rows");
 
   if (share->options & HA_OPTION_READ_ONLY_DATA)
@@ -54,9 +55,8 @@ int maria_delete_all_rows(MARIA_HA *info)
   {
     /*
       This record will be used by Recovery to finish the deletion if it
-      crashed. We force it because it's a non-undoable operation.
+      crashed. We force it to have a complete history in the log.
     */
-    LSN lsn;
     LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
     uchar log_data[FILEID_STORE_SIZE];
     log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
@@ -72,19 +72,17 @@ int maria_delete_all_rows(MARIA_HA *info)
       inconsistent.
     */
   }
-
-  /*
-    For recovery it matters that this is called after writing the log record,
-    so that resetting state.records and state.checksum actually happens under
-    log's mutex.
-  */
+  else
+  {
+    /* Other branch called function below when writing log record, in hook */
     _ma_reset_status(info);
+  }
 
   /*
     If we are using delayed keys or if the user has done changes to the tables
     since it was locked then there may be key blocks in the page cache. Or
     there may be data blocks there. We need to throw them away or they may
-    re-enter the emptied table later.
+    re-enter the emptied table or another table later.
   */
   if (_ma_flush_table_files(info, MARIA_FLUSH_DATA|MARIA_FLUSH_INDEX,
                             FLUSH_IGNORE_CHANGED, FLUSH_IGNORE_CHANGED) ||
@@ -95,12 +93,25 @@ int maria_delete_all_rows(MARIA_HA *info)
   if (_ma_initialize_data_file(share, info->dfile.file))
     goto err;
 
+
+  if (log_record)
+  {
     /*
-    The operations above on the index/data file will be forced to disk at
-    Checkpoint or maria_close() time. So we can reset:
+      Because LOGREC_REDO_DELETE_ALL does not operate on pages, it has the
+      following problem:
+      delete_all; inserts (redo_insert); all pages get flushed; checkpoint:
+      the dirty pages list will be empty. In recovery, delete_all is executed,
+      but redo_insert are skipped (dirty pages list is empty).
+      To avoid this, we need to set skip_redo_lsn now, and thus need to sync
+      files.
     */
-  if (log_record)
+    my_bool error= _ma_state_info_write(share, 1|4) ||
+      _ma_update_state_lsns(share, lsn, FALSE, FALSE) ||
+      _ma_sync_table_files(info);
     info->trn->rec_lsn= LSN_IMPOSSIBLE;
+    if (error)
+      goto err;
+  }
 
   VOID(_ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE));
 #ifdef HAVE_MMAP

storage/maria/ma_extra.c

@@ -550,14 +550,9 @@ int _ma_flush_table_files(MARIA_HA *info, uint flush_data_or_index,
   /* flush data file first because it's more critical */
   if (flush_data_or_index & MARIA_FLUSH_DATA)
   {
-    if (info->opt_flag & WRITE_CACHE_USED)
-    {
-      /* normally any code which creates a WRITE_CACHE destroys it later */
-      DBUG_ASSERT(0);
-      if (end_io_cache(&info->rec_cache))
+    if ((info->opt_flag & WRITE_CACHE_USED) &&
+        flush_io_cache(&info->rec_cache))
       goto err;
-      info->opt_flag&= ~WRITE_CACHE_USED;
-    }
     if (share->data_file_type == BLOCK_RECORD)
     {
       if(_ma_bitmap_flush(share) ||

storage/maria/ma_key_recover.c

@@ -148,6 +148,7 @@ my_bool write_hook_for_clr_end(enum translog_record_type type
   MARIA_SHARE *share= tbl_info->s;
   struct st_msg_to_write_hook_for_clr_end *msg=
     (struct st_msg_to_write_hook_for_clr_end *)hook_arg;
+  my_bool error= FALSE;
   DBUG_ASSERT(trn->trid != 0);
   trn->undo_lsn= msg->previous_undo_lsn;
 
@@ -175,12 +176,18 @@ my_bool write_hook_for_clr_end(enum translog_record_type type
   case LOGREC_UNDO_KEY_INSERT:
   case LOGREC_UNDO_KEY_DELETE:
     break;
+  case LOGREC_UNDO_BULK_INSERT_WITH_REPAIR:
+    error= (maria_enable_indexes(tbl_info) ||
+            /* we enabled indices, need '2' below */
+            _ma_state_info_write(share, 1|2|4));
+    /* no need for _ma_reset_status(): REDO_DELETE_ALL is just before us */
+    break;
   default:
     DBUG_ASSERT(0);
   }
   if (trn->undo_lsn == LSN_IMPOSSIBLE) /* has fully rolled back */
     trn->first_undo_lsn= LSN_WITH_FLAGS_TO_FLAGS(trn->first_undo_lsn);
-  return 0;
+  return error;
 }
 
 
@@ -966,14 +973,17 @@ my_bool _ma_apply_undo_key_insert(MARIA_HA *info, LSN undo_lsn,
 
 /**
    @brief Undo of delete of key (ie, insert the deleted key)
+
+   @param  with_root       If the UNDO is UNDO_KEY_DELETE_WITH_ROOT
 */
 
 my_bool _ma_apply_undo_key_delete(MARIA_HA *info, LSN undo_lsn,
-                                  const uchar *header, uint length)
+                                  const uchar *header, uint length,
+                                  my_bool with_root)
 {
   LSN lsn;
   my_bool res;
-  uint keynr;
+  uint keynr, skip_bytes;
   uchar key[HA_MAX_KEY_BUFF];
   MARIA_SHARE *share= info->s;
   my_off_t new_root;
@@ -984,10 +994,12 @@ my_bool _ma_apply_undo_key_delete(MARIA_HA *info, LSN undo_lsn,
                           STATE_NOT_SORTED_PAGES | STATE_NOT_ZEROFILLED |
                           STATE_NOT_MOVABLE);
   keynr= key_nr_korr(header);
-  length-= KEY_NR_STORE_SIZE;
+  skip_bytes= KEY_NR_STORE_SIZE + (with_root ? PAGE_STORE_SIZE : 0);
+  header+= skip_bytes;
+  length-= skip_bytes;
 
   /* We have to copy key as _ma_ck_real_write_btree() may change it */
-  memcpy(key, header + KEY_NR_STORE_SIZE, length);
+  memcpy(key, header, length);
   DBUG_DUMP("key", key, length);
 
   new_root= share->state.key_root[keynr];

storage/maria/ma_key_recover.h

@@ -78,7 +78,8 @@ uint _ma_apply_redo_index(MARIA_HA *info,
 my_bool _ma_apply_undo_key_insert(MARIA_HA *info, LSN undo_lsn,
                                   const uchar *header, uint length);
 my_bool _ma_apply_undo_key_delete(MARIA_HA *info, LSN undo_lsn,
-                                  const uchar *header, uint length);
+                                  const uchar *header, uint length,
+                                  my_bool with_root);
 
 static inline void _ma_finalize_row(MARIA_HA *info)
 {

storage/maria/ma_locking.c

@@ -624,6 +624,7 @@ void _ma_mark_file_crashed(MARIA_SHARE *share)
                    sizeof(share->state.header) +
                    MARIA_FILE_CHANGED_OFFSET,
                    MYF(MY_NABP));
+  DBUG_VOID_RETURN;
 }
 
 

storage/maria/ma_loghandler.c

@@ -613,6 +613,18 @@ static LOG_DESC INIT_LOGREC_INCOMPLETE_GROUP=
  NULL, NULL, NULL, 0,
  "incomplete_group", LOGREC_IS_GROUP_ITSELF, NULL, NULL};
 
+static LOG_DESC INIT_LOGREC_UNDO_BULK_INSERT_WITH_REPAIR=
+{LOGRECTYPE_VARIABLE_LENGTH, 0,
+ LSN_STORE_SIZE + FILEID_STORE_SIZE,
+ NULL, write_hook_for_undo, NULL, 1,
+ "undo_bulk_insert_with_repair", LOGREC_LAST_IN_GROUP, NULL, NULL};
+
+static LOG_DESC INIT_LOGREC_REDO_BITMAP_NEW_PAGE=
+{LOGRECTYPE_FIXEDLENGTH, FILEID_STORE_SIZE + PAGE_STORE_SIZE * 2,
+ FILEID_STORE_SIZE + PAGE_STORE_SIZE * 2,
+ NULL, NULL, NULL, 0,
+ "redo_create_bitmap", LOGREC_IS_GROUP_ITSELF, NULL, NULL};
+
 const myf log_write_flags= MY_WME | MY_NABP | MY_WAIT_IF_FULL;
 
 void translog_table_init()
@@ -696,12 +708,14 @@ void translog_table_init()
     INIT_LOGREC_INCOMPLETE_LOG;
   log_record_type_descriptor[LOGREC_INCOMPLETE_GROUP]=
     INIT_LOGREC_INCOMPLETE_GROUP;
-  for (i= LOGREC_INCOMPLETE_GROUP + 1;
-       i < LOGREC_NUMBER_OF_TYPES;
-       i++)
+  log_record_type_descriptor[LOGREC_UNDO_BULK_INSERT_WITH_REPAIR]=
+    INIT_LOGREC_UNDO_BULK_INSERT_WITH_REPAIR;
+  log_record_type_descriptor[LOGREC_REDO_BITMAP_NEW_PAGE]=
+    INIT_LOGREC_REDO_BITMAP_NEW_PAGE;
+  for (i= LOGREC_FIRST_FREE; i < LOGREC_NUMBER_OF_TYPES; i++)
     log_record_type_descriptor[i].rclass= LOGRECTYPE_NOT_ALLOWED;
 #ifndef DBUG_OFF
-  check_translog_description_table(LOGREC_INCOMPLETE_GROUP);
+  check_translog_description_table(LOGREC_FIRST_FREE -1);
 #endif
 };
 

storage/maria/ma_loghandler.h

@@ -74,7 +74,7 @@ struct st_maria_handler;
 
 /* Store methods to match the above sizes */
 #define fileid_store(T,A) int2store(T,A)
-#define page_store(T,A)   int5store(T,A)
+#define page_store(T,A)   int5store(T,((ulonglong)(A)))
 #define dirpos_store(T,A) ((*(uchar*) (T)) = A)
 #define pagerange_store(T,A) int2store(T,A)
 #define clr_type_store(T,A) ((*(uchar*) (T)) = A)
@@ -112,8 +112,8 @@ enum translog_record_type
   LOGREC_REDO_PURGE_ROW_TAIL,
   LOGREC_REDO_FREE_BLOCKS,
   LOGREC_REDO_FREE_HEAD_OR_TAIL,
-  LOGREC_REDO_DELETE_ROW,
-  LOGREC_REDO_UPDATE_ROW_HEAD,
+  LOGREC_REDO_DELETE_ROW, /* unused */
+  LOGREC_REDO_UPDATE_ROW_HEAD, /* unused */
   LOGREC_REDO_INDEX,
   LOGREC_REDO_INDEX_NEW_PAGE,
   LOGREC_REDO_INDEX_FREE_PAGE,
@@ -141,6 +141,9 @@ enum translog_record_type
   LOGREC_LONG_TRANSACTION_ID,
   LOGREC_INCOMPLETE_LOG,
   LOGREC_INCOMPLETE_GROUP,
+  LOGREC_UNDO_BULK_INSERT_WITH_REPAIR,
+  LOGREC_REDO_BITMAP_NEW_PAGE,
+  LOGREC_FIRST_FREE,
   LOGREC_RESERVED_FUTURE_EXTENSION= 63
 };
 #define LOGREC_NUMBER_OF_TYPES 64              /* Maximum, can't be extended */

storage/maria/ma_open.c

@@ -332,7 +332,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     if ((share->options & HA_OPTION_RELIES_ON_SQL_LAYER) &&
         ! (open_flags & HA_OPEN_FROM_SQL_LAYER))
     {
-      DBUG_PRINT("error", ("table cannot be openned from non-sql layer"));
+      DBUG_PRINT("error", ("table cannot be opened from non-sql layer"));
       my_errno= HA_ERR_UNSUPPORTED;
       goto err;
     }
@@ -628,6 +628,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
                              LSN_STORE_SIZE + TRANSID_SIZE :
                              0) + KEYPAGE_KEYID_SIZE + KEYPAGE_FLAG_SIZE +
                             KEYPAGE_USED_SIZE);
+    share->kfile.file= kfile;
 
     if (open_flags & HA_OPEN_COPY)
     {
@@ -652,13 +653,16 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
           view of the server, including server's recovery) now.
         */
         if ((open_flags & HA_OPEN_FROM_SQL_LAYER) || maria_in_recovery)
-          _ma_update_create_rename_lsn_sub(share, translog_get_horizon(),
-                                           TRUE);
+          _ma_update_state_lsns_sub(share, translog_get_horizon(),
+                                    TRUE, TRUE);
       }
       else if ((!LSN_VALID(share->state.create_rename_lsn) ||
                 !LSN_VALID(share->state.is_of_horizon) ||
                 (cmp_translog_addr(share->state.create_rename_lsn,
-                                   share->state.is_of_horizon) > 0)) &&
+                                   share->state.is_of_horizon) > 0) ||
+                !LSN_VALID(share->state.skip_redo_lsn) ||
+                (cmp_translog_addr(share->state.create_rename_lsn,
+                                   share->state.skip_redo_lsn) > 0)) &&
                !(open_flags & HA_OPEN_FOR_REPAIR))
       {
         /*
@@ -735,7 +739,6 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
       share->tot_locks++;
     }
 
-    share->kfile.file= kfile;
     _ma_set_index_pagecache_callbacks(&share->kfile, share);
     share->this_process=(ulong) getpid();
     share->last_process= share->state.process;
@@ -1140,12 +1143,12 @@ uint _ma_state_info_write_sub(File file, MARIA_STATE_INFO *state, uint pWrite)
   mi_int2store(ptr,state->changed);			ptr+= 2;
 
   /*
-    if you change the offset of create_rename_lsn/is_of_horizon inside the
-    index file's header, fix ma_create + ma_rename + ma_delete_all +
-    backward-compatibility.
+    If you change the offset of these LSNs, note that some functions do a
+    direct write of them without going through this function.
   */
   lsn_store(ptr, state->create_rename_lsn);		ptr+= LSN_STORE_SIZE;
   lsn_store(ptr, state->is_of_horizon);			ptr+= LSN_STORE_SIZE;
+  lsn_store(ptr, state->skip_redo_lsn);			ptr+= LSN_STORE_SIZE;
   mi_rowstore(ptr,state->state.records);		ptr+= 8;
   mi_rowstore(ptr,state->state.del);			ptr+= 8;
   mi_rowstore(ptr,state->split);			ptr+= 8;
@@ -1211,6 +1214,7 @@ static uchar *_ma_state_info_read(uchar *ptr, MARIA_STATE_INFO *state)
   state->changed= mi_uint2korr(ptr);			ptr+= 2;
   state->create_rename_lsn= lsn_korr(ptr);		ptr+= LSN_STORE_SIZE;
   state->is_of_horizon= lsn_korr(ptr);			ptr+= LSN_STORE_SIZE;
+  state->skip_redo_lsn= lsn_korr(ptr);			ptr+= LSN_STORE_SIZE;
   state->state.records= mi_rowkorr(ptr);		ptr+= 8;
   state->state.del = mi_rowkorr(ptr);			ptr+= 8;
   state->split	= mi_rowkorr(ptr);			ptr+= 8;
@@ -1695,7 +1699,8 @@ int maria_enable_indexes(MARIA_HA *info)
   int error= 0;
   MARIA_SHARE *share= info->s;
 
-  if (share->state.state.data_file_length ||
+  if ((share->state.state.data_file_length !=
+       (share->data_file_type == BLOCK_RECORD ? share->block_size : 0)) ||
       (share->state.state.key_file_length != share->base.keystart))
   {
     maria_print_error(info->s, HA_ERR_CRASHED);

storage/maria/ma_recovery_util.c

+/* Copyright (C) 2006,2007,2008 MySQL 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
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+/*
+  Q: Why isn't ma_recovery_util.c simply moved to ma_recovery.c ?
+
+  A: ma_recovery.c, because it invokes objects from ma_check.c (like
+  maria_chk_init()) causes the following problem:
+  if a source file a.c of a program invokes a function defined in
+  ma_recovery.c, then a.o depends on ma_recovery.o which depends on
+  ma_check.o: linker thus brings in ma_check.o. That brings in the
+  dependencies of ma_check.o which are definitions of _ma_check_print_info()
+  etc; if a.o does not define them then the ones of ha_maria.o are used
+  i.e. ha_maria.o is linked into the program, and this brings in dependencies
+  of ha_maria.o on mysqld.o into the program's linking which thus fails, as
+  the program is not linked with mysqld.o.
+  Thus, while several functions defined in ma_recovery.c could be useful to
+  other files, they cannot be used by them.
+  So we are going to gradually move a great share of ma_recovery.c's exported
+  functions into the present file, to isolate the problematic components and
+  avoid the problem.
+*/
+
+#include "maria_def.h"
+
+HASH all_dirty_pages;
+struct st_dirty_page /* used only in the REDO phase */
+{
+  uint64 file_and_page_id;
+  LSN rec_lsn;
+};
+/*
+  LSN after which dirty pages list does not apply. Can be slightly before
+  when ma_checkpoint_execute() started.
+*/
+LSN checkpoint_start= LSN_IMPOSSIBLE;
+
+/** @todo looks like duplicate of recovery_message_printed */
+my_bool procent_printed;
+FILE *tracef; /**< trace file for debugging */
+
+
+/** @brief Prints to a trace file if it is not NULL */
+void tprint(FILE *trace_file __attribute__ ((unused)),
+            const char *format __attribute__ ((unused)), ...)
+{
+  va_list args;
+  va_start(args, format);
+  DBUG_PRINT("info", ("%s", format));
+  if (trace_file != NULL)
+  {
+    if (procent_printed)
+    {
+      procent_printed= 0;
+      fputc('\n', trace_file);
+    }
+    vfprintf(trace_file, format, args);
+  }
+  va_end(args);
+}
+
+
+void eprint(FILE *trace_file __attribute__ ((unused)),
+            const char *format __attribute__ ((unused)), ...)
+{
+  va_list args;
+  va_start(args, format);
+  DBUG_PRINT("error", ("%s", format));
+  if (!trace_file)
+    trace_file= stderr;
+
+  if (procent_printed)
+  {
+    /* In silent mode, print on another line than the 0% 10% 20% line */
+    procent_printed= 0;
+    fputc('\n', trace_file);
+  }
+  vfprintf(trace_file , format, args);
+  fputc('\n', trace_file);
+  if (trace_file != stderr)
+  {
+    va_start(args, format);
+    my_printv_error(HA_ERR_INITIALIZATION, format, MYF(0), args);
+  }
+  va_end(args);
+  fflush(trace_file);
+}
+
+
+/**
+   Tells if the dirty pages list found in checkpoint record allows to ignore a
+   REDO for a certain page.
+
+   @param  shortid         short id of the table
+   @param  lsn             REDO record's LSN
+   @param  page            page number
+   @param  index           TRUE if index page, FALSE if data page
+*/
+
+my_bool _ma_redo_not_needed_for_page(uint16 shortid, LSN lsn,
+                                     pgcache_page_no_t page,
+                                     my_bool index)
+{
+  if (cmp_translog_addr(lsn, checkpoint_start) < 0)
+  {
+    /*
+      64-bit key is formed like this:
+      Most significant byte: 0
+      Next byte: 0 if data page, 1 if index page
+      Next 2 bytes: table's short id
+      Next 4 bytes: page number
+    */
+    uint64 file_and_page_id=
+      (((uint64)((index << 16) | shortid)) << 32) | 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));
+    if ((dirty_page == NULL) ||
+        cmp_translog_addr(lsn, dirty_page->rec_lsn) < 0)
+    {
+      tprint(tracef, ", ignoring because of dirty_pages list\n");
+      return TRUE;
+    }
+  }
+  return FALSE;
+}

storage/maria/ma_recovery_util.h

+/* Copyright (C) 2006,2007,2008 MySQL 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
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+struct st_dirty_page /* used only in the REDO phase */
+{
+  uint64 file_and_page_id;
+  LSN rec_lsn;
+};
+extern HASH all_dirty_pages;
+/*
+  LSN after which dirty pages list does not apply. Can be slightly before
+  when ma_checkpoint_execute() started.
+*/
+extern LSN checkpoint_start;
+extern my_bool procent_printed;
+extern FILE *tracef;
+
+
+my_bool _ma_redo_not_needed_for_page(uint16 shortid, LSN lsn,
+                                     pgcache_page_no_t page,
+                                     my_bool index);
+void tprint(FILE *trace_file, const char *format, ...)
+  ATTRIBUTE_FORMAT(printf, 2, 3);
+void eprint(FILE *trace_file, const char *format, ...)
+  ATTRIBUTE_FORMAT(printf, 2, 3);

storage/maria/ma_rename.c

@@ -95,7 +95,7 @@ int maria_rename(const char *old_name, const char *new_name)
       store LSN into file, needed for Recovery to not be confused if a
       RENAME happened (applying REDOs to the wrong table).
     */
-    if (_ma_update_create_rename_lsn(share, lsn, TRUE))
+    if (_ma_update_state_lsns(share, lsn, TRUE, TRUE))
     {
       maria_close(info);
       DBUG_RETURN(1);

storage/maria/ma_static.c

@@ -25,7 +25,7 @@
 
 LIST	*maria_open_list=0;
 uchar	maria_file_magic[]=
-{ (uchar) 254, (uchar) 254, (uchar) 9, '\002', };
+{ (uchar) 254, (uchar) 254, (uchar) 9, '\003', };
 uchar	maria_pack_file_magic[]=
 { (uchar) 254, (uchar) 254, (uchar) 10, '\001', };
 /* Unique number for this maria instance */

storage/maria/ma_test2.c

@@ -893,6 +893,11 @@ int main(int argc, char *argv[])
     maria_scan_end(file);
     goto err;
   }
+  if (testflag == 6)
+    goto end;
+  if (checkpoint == 6 && ma_checkpoint_execute(CHECKPOINT_MEDIUM, FALSE))
+    goto err;
+
 end:
   maria_scan_end(file);
   if (die_in_middle_of_transaction)

storage/maria/ma_test_recovery

@@ -137,7 +137,7 @@ do
   for test_undo in 1 2 3 4
   do
   # first iteration tests rollback of insert, second tests rollback of delete
-  set -- "ma_test1 $silent -M -T -c -N $blobs -H1" "--testflag=1" "--testflag=2 --test-undo=" "ma_test1 $silent -M -T -c -N $blobs -H2" "--testflag=3" "--testflag=4 --test-undo=" "ma_test1 $silent -M -T -c -N $blobs -H2 " "--testflag=2" "--testflag=3 --test-undo=" "ma_test2 $silent -L -K -W -P -M -T -c $blobs -H1" "-t1" "-t2 -A" "ma_test2 $silent -L -K -W -P -M -T -c $blobs -H1" "-t1" "-t4 -A"
+  set -- "ma_test1 $silent -M -T -c -N $blobs -H1" "--testflag=1" "--testflag=2 --test-undo=" "ma_test1 $silent -M -T -c -N $blobs -H2" "--testflag=3" "--testflag=4 --test-undo=" "ma_test1 $silent -M -T -c -N $blobs -H2 " "--testflag=2" "--testflag=3 --test-undo=" "ma_test2 $silent -L -K -W -P -M -T -c $blobs -H1" "-t1" "-t2 -A" "ma_test2 $silent -L -K -W -P -M -T -c $blobs -H1" "-t1" "-t6 -A"
   # -N (create NULL fields) is needed because --test-undo adds it anyway
   while [ $# != 0 ]
     do

storage/maria/ma_test_recovery.expected

@@ -75,7 +75,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t4 -A1 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t6 -A1 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -162,7 +162,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t4 -A2 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t6 -A2 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -249,7 +249,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t4 -A3 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t6 -A3 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -336,7 +336,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t4 -A4 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -t6 -A4 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -423,7 +423,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t4 -A1 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t6 -A1 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -510,7 +510,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t4 -A2 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t6 -A2 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -597,7 +597,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t4 -A3 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t6 -A3 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -684,7 +684,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t4 -A4 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -t6 -A4 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -771,7 +771,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t4 -A1 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t6 -A1 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -858,7 +858,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t4 -A2 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t6 -A2 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -945,7 +945,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t4 -A3 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t6 -A3 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -1032,7 +1032,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t4 -A4 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c  -H1 -t6 -A4 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -1119,7 +1119,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t4 -A1 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t6 -A1 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -1206,7 +1206,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t4 -A2 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t6 -A2 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -1293,7 +1293,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t4 -A3 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t6 -A3 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !
@@ -1380,7 +1380,7 @@ Differences in maria_chk -dvv, recovery not yet perfect !
 > Status:              changed
 ========DIFF END=======
 TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t1 (commit at end)
-TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t4 -A4 (additional aborted work)
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -b32768 -H1 -t6 -A4 (additional aborted work)
 Dying on request without maria_commit()/maria_close()
 applying log
 Differences in maria_chk -dvv, recovery not yet perfect !

storage/maria/maria_chk.c

@@ -71,6 +71,8 @@ static const char *record_formats[]=
 };
 
 static const char *maria_stats_method_str="nulls_unequal";
+static char default_open_errmsg[]=  "%d when opening MARIA-table '%s'";
+static char default_close_errmsg[]= "%d when closing MARIA-table '%s'";
 
 static void get_options(int *argc,char * * *argv);
 static void print_version(void);
@@ -84,6 +86,7 @@ static int sort_record_index(MARIA_SORT_PARAM *sort_param, MARIA_HA *info,
                              MARIA_KEYDEF *keyinfo,
 			     my_off_t page, uchar *buff,uint sortkey,
 			     File new_file, my_bool update_index);
+static my_bool write_log_record(HA_CHECK *param);
 
 HA_CHECK check_param;
 
@@ -1105,7 +1108,7 @@ static int maria_chk(HA_CHECK *param, char *filename)
       */
       if (share->base.born_transactional)
         share->state.create_rename_lsn= share->state.is_of_horizon=
-          LSN_REPAIRED_BY_MARIA_CHK;
+          share->state.skip_redo_lsn= LSN_REPAIRED_BY_MARIA_CHK;
     }
     if (!error && (param->testflag & T_REP_ANY))
     {
@@ -1218,7 +1221,6 @@ static int maria_chk(HA_CHECK *param, char *filename)
           (param->testflag & (T_EXTEND | T_MEDIUM)))
         error|=maria_chk_data_link(param, info,
                                    test(param->testflag & T_EXTEND));
-      error|= _ma_flush_table_files_after_repair(param, info);
       VOID(end_io_cache(&param->read_cache));
     }
     if (!error)
@@ -1258,8 +1260,7 @@ end2:
   end_pagecache(maria_pagecache, 1);
   if (maria_close(info))
   {
-    _ma_check_print_error(param,"%d when closing MARIA-table '%s'",
-                          my_errno,filename);
+    _ma_check_print_error(param, default_close_errmsg, my_errno, filename);
     DBUG_RETURN(1);
   }
   if (error == 0)
@@ -1272,6 +1273,12 @@ end2:
       error|=maria_change_to_newfile(filename,MARIA_NAME_IEXT,INDEX_TMP_EXT,
                                      MYF(0));
   }
+  if (opt_transaction_logging &&
+      share->base.born_transactional && !error &&
+      (param->testflag & (T_REP_ANY | T_SORT_RECORDS | T_SORT_INDEX |
+                          T_ZEROFILL)))
+    error= write_log_record(param);
+
   VOID(fflush(stdout)); VOID(fflush(stderr));
   if (param->error_printed)
   {
@@ -1745,7 +1752,7 @@ err:
   my_free(sort_info.buff,MYF(MY_ALLOW_ZERO_PTR));
   sort_info.buff=0;
   share->state.sortkey=sort_key;
-  DBUG_RETURN(_ma_flush_table_files_after_repair(param, info) | got_error);
+  DBUG_RETURN(got_error);
 } /* sort_records */
 
 
@@ -1843,4 +1850,29 @@ err:
 } /* sort_record_index */
 
 
+static my_bool write_log_record(HA_CHECK *param)
+{
+  /*
+    Now that all operations including O_NEW_DATA|INDEX are successfully
+    done, we can write a log record.
+  */
+  MARIA_HA *info= maria_open(param->isam_file_name, O_RDWR, 0);
+  if (info == NULL)
+    _ma_check_print_error(param, default_open_errmsg, my_errno,
+                          param->isam_file_name);
+  else
+  {
+    if (write_log_record_for_repair(param, info))
+      _ma_check_print_error(param, "%d when writing log record for"
+                            " MARIA-table '%s'", my_errno,
+                            param->isam_file_name);
+    else if (maria_close(info))
+      _ma_check_print_error(param, default_close_errmsg, my_errno,
+                            param->isam_file_name);
+    else
+      return FALSE;
+  }
+  return TRUE;
+}
+
 #include "ma_check_standalone.h"

storage/maria/maria_def.h

@@ -109,9 +109,28 @@ typedef struct st_maria_state_info
   uint sortkey;				/* sorted by this key (not used) */
   uint open_count;
   uint changed;                         /* Changed since maria_chk */
-  LSN create_rename_lsn;    /**< LSN when table was last created/renamed */
+  /**
+     Birthday of the table: no record in the log before this LSN should ever
+     be applied to the table. Updated when created, renamed, explicitely
+     repaired (REPAIR|OPTIMIZE TABLE, ALTER TABLE ENABLE KEYS, maria_chk).
+  */
+  LSN create_rename_lsn;
   /** @brief Log horizon when state was last updated on disk */
   TRANSLOG_ADDRESS is_of_horizon;
+  /**
+     REDO phase should ignore any record before this LSN. UNDO phase
+     shouldn't, this is the difference with create_rename_lsn.
+     skip_redo_lsn >= create_rename_lsn.
+     The distinction is for these cases:
+     - after a repair at end of bulk insert (enabling indices), REDO phase
+     should skip the table but UNDO phase should not, so only skip_redo_lsn is
+     increased, not create_rename_lsn
+     - if one table is corrupted and so recovery fails, user may repair the
+     table with maria_chk and let recovery restart: that recovery should then
+     skip the repaired table even in the UNDO phase, so create_rename_lsn is
+     increased.
+  */
+  LSN skip_redo_lsn;
 
   /* the following isn't saved on disk */
   uint state_diff_length;		/* Should be 0 */
@@ -121,7 +140,7 @@ typedef struct st_maria_state_info
 
 
 #define MARIA_STATE_INFO_SIZE	\
-  (24 + 2 + LSN_STORE_SIZE*2 + 4 + 11*8 + 4*4 + 8 + 3*4 + 5*8)
+  (24 + 2 + LSN_STORE_SIZE*3 + 4 + 11*8 + 4*4 + 8 + 3*4 + 5*8)
 #define MARIA_FILE_OPEN_COUNT_OFFSET 0
 #define MARIA_FILE_CHANGED_OFFSET 2
 #define MARIA_FILE_CREATE_RENAME_LSN_OFFSET 4
@@ -1054,6 +1073,7 @@ void _ma_check_print_warning _VARARGS((HA_CHECK *param, const char *fmt, ...))
   ATTRIBUTE_FORMAT(printf, 2, 3);
 void _ma_check_print_info _VARARGS((HA_CHECK *param, const char *fmt, ...))
   ATTRIBUTE_FORMAT(printf, 2, 3);
+my_bool write_log_record_for_repair(const HA_CHECK *param, MARIA_HA *info);
 C_MODE_END
 
 int _ma_flush_pending_blocks(MARIA_SORT_PARAM *param);
@@ -1062,17 +1082,18 @@ int _ma_thr_write_keys(MARIA_SORT_PARAM *sort_param);
 #ifdef THREAD
 pthread_handler_t _ma_thr_find_all_keys(void *arg);
 #endif
-int _ma_flush_table_files_after_repair(HA_CHECK *param, MARIA_HA *info);
 
 int _ma_sort_write_record(MARIA_SORT_PARAM *sort_param);
 int _ma_create_index_by_sort(MARIA_SORT_PARAM *info, my_bool no_messages,
                              size_t);
 int _ma_sync_table_files(const MARIA_HA *info);
 int _ma_initialize_data_file(MARIA_SHARE *share, File dfile);
-int _ma_update_create_rename_lsn(MARIA_SHARE *share,
-                                 LSN lsn, my_bool do_sync);
-int _ma_update_create_rename_lsn_sub(MARIA_SHARE *share,
-                                     LSN lsn, my_bool do_sync);
+int _ma_update_state_lsns(MARIA_SHARE *share,
+                          LSN lsn, my_bool do_sync, my_bool
+                          update_create_rename_lsn);
+int _ma_update_state_lsns_sub(MARIA_SHARE *share,
+                              LSN lsn, my_bool do_sync, my_bool
+                              update_create_rename_lsn);
 void _ma_set_data_pagecache_callbacks(PAGECACHE_FILE *file,
                                       MARIA_SHARE *share);
 void _ma_set_index_pagecache_callbacks(PAGECACHE_FILE *file,
@@ -1080,6 +1101,8 @@ void _ma_set_index_pagecache_callbacks(PAGECACHE_FILE *file,
 void _ma_tmp_disable_logging_for_table(MARIA_HA *info,
                                        my_bool log_incomplete);
 void _ma_reenable_logging_for_table(MARIA_HA *info);
+my_bool write_log_record_for_bulk_insert_with_repair(MARIA_HA *info);
+
 
 #define MARIA_NO_CRC_NORMAL_PAGE 0xffffffff
 #define MARIA_NO_CRC_BITMAP_PAGE 0xfffffffe

storage/maria/maria_pack.c

@@ -2977,7 +2977,7 @@ static int save_state(MARIA_HA *isam_file,PACK_MRG_INFO *mrg,
   share->state.version=(ulong) time((time_t*) 0);
   if (share->base.born_transactional)
     share->state.create_rename_lsn= share->state.is_of_horizon=
-      LSN_REPAIRED_BY_MARIA_CHK;
+      share->state.skip_redo_lsn= LSN_REPAIRED_BY_MARIA_CHK;
   if (! maria_is_all_keys_active(share->state.key_map, share->base.keys))
   {
     /*