- fix for segfault in rpl_trigger/rpl_found_rows with default engine=maria

(fix is keeping the real TRN through a disable_logging/reenable cycle)
- fix for pagecache assertion failure in ps/type_ranges with default
engine=maria (fix is in sql_insert.cc)
- when reenabling logging we must either flush all dirty pages,
or at least verify (in debug build) that there are none. For example
a bulk insert with single UNDO_BULK_INSERT must flush them, no matter
if it uses repair or not (bugfix)
- UNDO_BULK_INSERT_WITH_REPAIR is also used with repair, changes name


mysql-test/r/maria.result:
  tests for bugs fixed
mysql-test/t/maria.test:
  tests for bugs fixed
sql/sql_insert.cc:
  Bugfix: even if select_create::prepare() failed to create the 'table' object
  we still have to re-enable logging.
storage/maria/ha_maria.cc:
  Bugfix: when a transactional table does a bulk insert without
  repair, it still sometimes skips logging of REDOs thus needs a full
  flush and sync at the end. Not if repair is done, as repair does
  it internally already (see end of maria_repair*()).
storage/maria/ha_maria.h:
  variable now can have 3 states not 2
storage/maria/ma_bitmap.c:
  name change
storage/maria/ma_blockrec.c:
  name change
storage/maria/ma_blockrec.h:
  name change
storage/maria/ma_check.c:
  * When maria_repair() re-enables logging it does not need to ask for
  a flush&sync as it did it by itself already a few lines before.
  * the log record of bulk insert can be used even without repair
  * disable logging in maria_zerofill(): without that, it puts LSN pages
  in the cache, so when it flushes them it flushes the log; the change
  makes auto-ha_maria::zerofill-if-moved faster (no log flush).
storage/maria/ma_key_recover.c:
  name change
storage/maria/ma_loghandler.c:
  name change
storage/maria/ma_loghandler.h:
  name change
storage/maria/ma_pagecache.c:
  A function, to check in debug builds that no dirty pages exist for a file.
storage/maria/ma_pagecache.h:
  new function (nothing in non-debug)
storage/maria/ma_recovery.c:
  _ma_tmp_disable_logging() sets info->trn to dummy_transaction_object
  when needed now. The changes done here about info->trn are to allow
  a table to retain its original, real TRN through a disable/reenable
  cycle (see replication scenario in _ma_reenable_logging_for_table()).
  When we reenable, we offer the caller to flush and sync the table;
  if the caller doesn't accept our offer, we verify that it's ok
  (no REDOs => no dirty pages are allowed to exist).
storage/maria/maria_chk.c:
  comment
storage/maria/maria_def.h:
  new names
mysql-test/suite/rpl/r/rpl_stm_maria.result:
  result (it used to crash)
mysql-test/suite/rpl/t/rpl_stm_maria.test:
  Test of replication-specific Maria bug fixed

19 files changed, 341 insertions, 80 deletions

diff --git a/mysql-test/r/maria.result b/mysql-test/r/maria.result
index 8eb226d57ac..d5fe9f1d237 100644
--- a/mysql-test/r/maria.result
+++ b/mysql-test/r/maria.result
@@ -2101,3 +2101,36 @@ delete from t1;
 select * from t1;
 a
 drop table t1;
+create table t1 (c int);
+insert into t1 values(1),(2);
+create table t2 select * from t1;
+create table t3 select * from t1, t2;
+ERROR 42S21: Duplicate column name 'c'
+create table t3 select t1.c AS c1, t2.c AS c2,1 as "const" from t1, t2;
+drop table t1, t2, t3;
+create table t1 (t datetime) engine=maria;
+insert into t1 values (101),(691231),(700101),(991231),(10000101),(99991231),(101000000),(691231000000),(700101000000),(991231235959),(10000101000000),(99991231235959),(20030100000000),(20030000000000);
+select * from t1;
+t
+2000-01-01 00:00:00
+2069-12-31 00:00:00
+1970-01-01 00:00:00
+1999-12-31 00:00:00
+1000-01-01 00:00:00
+9999-12-31 00:00:00
+2000-01-01 00:00:00
+2069-12-31 00:00:00
+1970-01-01 00:00:00
+1999-12-31 23:59:59
+1000-01-01 00:00:00
+9999-12-31 23:59:59
+2003-01-00 00:00:00
+2003-00-00 00:00:00
+delete from t1 where t > 0;
+optimize table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	status	OK
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+drop table t1;
diff --git a/mysql-test/suite/rpl/r/rpl_stm_maria.result b/mysql-test/suite/rpl/r/rpl_stm_maria.result
new file mode 100644
index 00000000000..be21b946ab5
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_stm_maria.result
@@ -0,0 +1,51 @@
+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+DROP TABLE IF EXISTS t1;
+DROP TABLE IF EXISTS t2;
+DROP TABLE IF EXISTS t3;
+create table t1 (a int auto_increment, primary key (a), b int,
+rand_value double not null) engine=maria;
+create table t2 (a int auto_increment, primary key (a), b int) engine=maria;
+create table t3 (a int auto_increment, primary key (a), name
+varchar(64) not null, old_a int, old_b int, rand_value double not
+null) engine=maria;
+create trigger t1 before insert on t1 for each row
+begin
+insert into t3 values (NULL, "t1", new.a, new.b, rand());
+end|
+create trigger t2 after insert on t2 for each row
+begin
+insert into t3 values (NULL, "t2", new.a, new.b, rand());
+end|
+insert into t3 values(100,"log",0,0,0);
+SET @@RAND_SEED1=658490765, @@RAND_SEED2=635893186;
+insert into t1 values(1,1,rand()),(NULL,2,rand());
+insert into t2 (b) values(last_insert_id());
+insert into t2 values(3,0),(NULL,0);
+insert into t2 values(NULL,0),(500,0);
+select a,b, truncate(rand_value,4) from t1;
+a	b	truncate(rand_value,4)
+1	1	0.4320
+2	2	0.3055
+select * from t2;
+a	b
+1	2
+3	0
+4	0
+5	0
+500	0
+select a,name, old_a, old_b, truncate(rand_value,4) from t3;
+a	name	old_a	old_b	truncate(rand_value,4)
+100	log	0	0	0.0000
+101	t1	1	1	0.3203
+102	t1	0	2	0.5666
+103	t2	1	2	0.9164
+104	t2	3	0	0.8826
+105	t2	4	0	0.6635
+106	t2	5	0	0.6699
+107	t2	500	0	0.3593
+drop table t1,t2,t3;
diff --git a/mysql-test/suite/rpl/t/rpl_stm_maria.test b/mysql-test/suite/rpl/t/rpl_stm_maria.test
new file mode 100644
index 00000000000..f850473ab51
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_stm_maria.test
@@ -0,0 +1,51 @@
+# Test of Maria-specific replication bugs
+
+--source include/have_maria.inc
+--source include/have_binlog_format_mixed_or_statement.inc 
+--source include/master-slave.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+DROP TABLE IF EXISTS t2;
+DROP TABLE IF EXISTS t3;
+--enable_warnings
+
+# This one taken from rpl_trigger.test (from BUG#12482)
+# used to segfault slave in execution of row-based events
+
+# Need an explicit ENGINE= clause as @@STORAGE_ENGINE is not replicated
+create table t1 (a int auto_increment, primary key (a), b int,
+rand_value double not null) engine=maria;
+create table t2 (a int auto_increment, primary key (a), b int) engine=maria;
+create table t3 (a int auto_increment, primary key (a), name
+varchar(64) not null, old_a int, old_b int, rand_value double not
+null) engine=maria;
+
+delimiter |;
+create trigger t1 before insert on t1 for each row
+begin
+ insert into t3 values (NULL, "t1", new.a, new.b, rand());
+end|
+
+create trigger t2 after insert on t2 for each row
+begin
+ insert into t3 values (NULL, "t2", new.a, new.b, rand());
+end|
+delimiter ;|
+
+insert into t3 values(100,"log",0,0,0);
+
+SET @@RAND_SEED1=658490765, @@RAND_SEED2=635893186;
+
+insert into t1 values(1,1,rand()),(NULL,2,rand());
+insert into t2 (b) values(last_insert_id());
+insert into t2 values(3,0),(NULL,0);
+insert into t2 values(NULL,0),(500,0);
+
+select a,b, truncate(rand_value,4) from t1;
+select * from t2;
+select a,name, old_a, old_b, truncate(rand_value,4) from t3;
+sync_slave_with_master;
+connection master;
+drop table t1,t2,t3;
+sync_slave_with_master;
diff --git a/mysql-test/t/maria.test b/mysql-test/t/maria.test
index 120f5178a4c..5feefa20c09 100644
--- a/mysql-test/t/maria.test
+++ b/mysql-test/t/maria.test
@@ -1357,6 +1357,28 @@ delete from t1;
 select * from t1;
 drop table t1;
 
+# Test for bug "ha_enable_transaction(on) not called by CREATE TABLE"
+# (originally from type_ranges.test)
+
+create table t1 (c int);
+insert into t1 values(1),(2);
+create table t2 select * from t1;
+--error 1060
+create table t3 select * from t1, t2; # Should give an error
+create table t3 select t1.c AS c1, t2.c AS c2,1 as "const" from t1, t2;
+drop table t1, t2, t3;
+
+# Test for bug "maria_repair() (OPTIMIZE) leaves wrong
+# data_file_length" (originally from type_datetime.test)
+
+create table t1 (t datetime) engine=maria;
+insert into t1 values (101),(691231),(700101),(991231),(10000101),(99991231),(101000000),(691231000000),(700101000000),(991231235959),(10000101000000),(99991231235959),(20030100000000),(20030000000000);
+select * from t1;
+delete from t1 where t > 0;
+optimize table t1;
+check table t1;
+drop table t1;
+
 # End of 5.2 tests
 
 --disable_result_log
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 6a7356f17ee..f4084c12c1f 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3705,7 +3705,7 @@ void select_create::abort()
   select_insert::abort();
   reenable_binlog(thd);
 
-  if (table && !table->s->tmp_table)
+  if ((thd->lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) == 0)
     ha_enable_transaction(thd, TRUE);
 
   /*
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 53ee96cd048..0ad823c1755 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -641,6 +641,21 @@ void _ma_check_print_warning(HA_CHECK *param, const char *fmt, ...)
 
 }
 
+/**
+  Transactional table doing bulk insert with one single UNDO
+  (UNDO_BULK_INSERT) and with repair.
+*/
+#define BULK_INSERT_SINGLE_UNDO_AND_REPAIR    1
+/**
+  Transactional table doing bulk insert with one single UNDO
+  (UNDO_BULK_INSERT) and with repair.
+*/
+#define BULK_INSERT_SINGLE_UNDO_AND_NO_REPAIR 2
+/**
+  None of BULK_INSERT_SINGLE_UNDO_AND_REPAIR and
+  BULK_INSERT_SINGLE_UNDO_AND_NO_REPAIR.
+*/
+#define BULK_INSERT_NONE      0
 
 ha_maria::ha_maria(handlerton *hton, TABLE_SHARE *table_arg):
 handler(hton, table_arg), file(0),
@@ -650,7 +665,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), bulk_insert_with_repair_trans(FALSE)
+can_enable_indexes(1), bulk_insert_single_undo(BULK_INSERT_NONE)
 {}
 
 
@@ -1586,7 +1601,7 @@ int ha_maria::disable_indexes(uint mode)
 int ha_maria::enable_indexes(uint mode)
 {
   int error;
-
+  DBUG_PRINT("info", ("ha_maria::enable_indexes mode: %d", mode));
   if (maria_is_all_keys_active(file->s->state.key_map, file->s->base.keys))
   {
     /* All indexes are enabled already. */
@@ -1611,10 +1626,11 @@ 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)
+    if (bulk_insert_single_undo == BULK_INSERT_SINGLE_UNDO_AND_NO_REPAIR)
     {
+      bulk_insert_single_undo= BULK_INSERT_SINGLE_UNDO_AND_REPAIR;
       /*
-        Don't bump create_rename_lsn, because UNDO_BULK_INSERT_WITH_REPAIR
+        Don't bump create_rename_lsn, because UNDO_BULK_INSERT
         should not be skipped in case of crash during repair.
       */
       param.testflag|= T_NO_CREATE_RENAME_LSN;
@@ -1711,7 +1727,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;
+  bulk_insert_single_undo= BULK_INSERT_NONE;
 
   if (!(specialflag & SPECIAL_SAFE_MODE))
   {
@@ -1724,11 +1740,15 @@ 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))
     {
+      /**
+         @todo for a single-row INSERT SELECT, we will go into repair, which
+         is more costly (flushes, syncs) than a row write.
+      */
       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);
+        bulk_insert_single_undo= BULK_INSERT_SINGLE_UNDO_AND_NO_REPAIR;
+        write_log_record_for_bulk_insert(file);
         /*
           Pages currently in the page cache have type PAGECACHE_LSN_PAGE, we
           are not allowed to overwrite them with PAGECACHE_PLAIN_PAGE, so
@@ -1778,11 +1798,17 @@ int ha_maria::end_bulk_insert()
   if (can_enable_indexes)
     err= enable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE);
 end:
-  if (bulk_insert_with_repair_trans)
+  if (bulk_insert_single_undo != BULK_INSERT_NONE)
   {
     DBUG_ASSERT(can_enable_indexes);
-    /* table was transactional just before start_bulk_insert() */
-    _ma_reenable_logging_for_table(file);
+    /*
+      Table was transactional just before start_bulk_insert().
+      No need to flush pages if we did a repair (which already flushed).
+    */
+    err|=
+      _ma_reenable_logging_for_table(file,
+                                     bulk_insert_single_undo ==
+                                     BULK_INSERT_SINGLE_UNDO_AND_NO_REPAIR);
   }
   DBUG_RETURN(err);
 }
@@ -2186,7 +2212,8 @@ int ha_maria::external_lock(THD *thd, int lock_type)
   }
   else
   {
-    _ma_reenable_logging_for_table(file);
+    if (_ma_reenable_logging_for_table(file, TRUE))
+      DBUG_RETURN(1);
     /** @todo zero file->trn also in commit and rollback */
     file->trn= NULL;
     if (trn && trnman_has_locked_tables(trn))
diff --git a/storage/maria/ha_maria.h b/storage/maria/ha_maria.h
index 93b91b6d4fe..5818b2822d1 100644
--- a/storage/maria/ha_maria.h
+++ b/storage/maria/ha_maria.h
@@ -40,8 +40,11 @@ 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;
+  /**
+    If a transactional table is doing bulk insert with a single
+    UNDO_BULK_INSERT with/without repair. 
+  */
+  uint8 bulk_insert_single_undo;
   int repair(THD * thd, HA_CHECK &param, bool optimize);
   int zerofill(THD * thd, HA_CHECK_OPT *check_opt);
 
diff --git a/storage/maria/ma_bitmap.c b/storage/maria/ma_bitmap.c
index b54d7d12b97..112bacdd867 100644
--- a/storage/maria/ma_bitmap.c
+++ b/storage/maria/ma_bitmap.c
@@ -2683,7 +2683,7 @@ err:
  delete last row of very large table (with delete_row)
  do a bulk insert
  crash
- Then UNDO_BULK_INSERT_WITH_REPAIR will truncate table files, and
+ Then UNDO_BULK_INSERT will truncate table files, and
  UNDO_ROW_DELETE will want to put the row back to its original position,
  extending the data file a lot: bitmap page*s* in the hole must be created,
  or he table would look corrupted.
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index 95ae1325d71..2263e79fe52 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -6836,12 +6836,11 @@ err:
     @retval 1      Error
 */
 
-my_bool _ma_apply_undo_bulk_insert_with_repair(MARIA_HA *info,
-                                               LSN undo_lsn)
+my_bool _ma_apply_undo_bulk_insert(MARIA_HA *info, LSN undo_lsn)
 {
   my_bool error;
   LSN lsn;
-  DBUG_ENTER("_ma_apply_undo_bulk_insert_with_repair");
+  DBUG_ENTER("_ma_apply_undo_bulk_insert");
   /*
     We delete all rows, re-enable indices as bulk insert had disabled
     non-unique ones.
@@ -6850,7 +6849,7 @@ my_bool _ma_apply_undo_bulk_insert_with_repair(MARIA_HA *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,
+          _ma_write_clr(info, undo_lsn, LOGREC_UNDO_BULK_INSERT,
                         FALSE, 0, &lsn, NULL));
   DBUG_RETURN(error);
 }
diff --git a/storage/maria/ma_blockrec.h b/storage/maria/ma_blockrec.h
index 3b87548e4a5..e8dc554bb98 100644
--- a/storage/maria/ma_blockrec.h
+++ b/storage/maria/ma_blockrec.h
@@ -246,8 +246,7 @@ 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 _ma_apply_undo_bulk_insert(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,
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index 5541840db6c..9c349c4ee77 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -2082,10 +2082,10 @@ err:
   be tried and fail. To prevent that, we bump skip_redo_lsn, and thus we have
   to flush and sync pages so that old REDOs can be skipped.
   If this is not a bulk insert, which Recovery can handle gracefully (by
-  truncating files, see UNDO_BULK_INSERT_WITH_REPAIR) we also mark the table
+  truncating files, see UNDO_BULK_INSERT) we also mark the table
   crashed-on-repair, so that user knows it has to re-repair. If bulk insert we
   shouldn't mark it crashed-on-repair, because if we did this, the UNDO phase
-  would skip the table (UNDO_BULK_INSERT_WITH_REPAIR would not be applied),
+  would skip the table (UNDO_BULK_INSERT would not be applied),
   and maria_chk would not improve that.
   If this is an OPTIMIZE which merely sorts index, we need to do the same
   too: old REDOs should not apply to the new index file.
@@ -2532,7 +2532,7 @@ err:
   }
   /* If caller had disabled logging it's not up to us to re-enable it */
   if (reenable_logging)
-    _ma_reenable_logging_for_table(info);
+    _ma_reenable_logging_for_table(info, FALSE);
 
   my_free(sort_param.rec_buff, MYF(MY_ALLOW_ZERO_PTR));
   my_free(sort_param.record,MYF(MY_ALLOW_ZERO_PTR));
@@ -3122,24 +3122,26 @@ err:
 
 int maria_zerofill(HA_CHECK *param, MARIA_HA *info, const char *name)
 {
+  my_bool error, reenable_logging;
   DBUG_ENTER("maria_zerofill");
-
-  if (maria_zerofill_index(param, info, name))
-    DBUG_RETURN(1);
-  if (maria_zerofill_data(param, info, name))
-    DBUG_RETURN(1);
-  if (_ma_set_uuid(info, 0))
-    DBUG_RETURN(1);
-
-  /*
-    Mark that table is movable and that we have done zerofill of data and
-    index
-  */
-  info->s->state.changed&= ~(STATE_NOT_ZEROFILLED | STATE_NOT_MOVABLE |
-                             STATE_MOVED);
-  /* Ensure state are flushed to disk */
-  info->update= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
-  DBUG_RETURN(0);
+  if ((reenable_logging= info->s->now_transactional))
+    _ma_tmp_disable_logging_for_table(info, 0);
+  if (!(error= (maria_zerofill_index(param, info, name) ||
+                maria_zerofill_data(param, info, name) ||
+                _ma_set_uuid(info, 0))))
+  {
+    /*
+      Mark that table is movable and that we have done zerofill of data and
+      index
+    */
+    info->s->state.changed&= ~(STATE_NOT_ZEROFILLED | STATE_NOT_MOVABLE |
+                               STATE_MOVED);
+    /* Ensure state are flushed to disk */
+    info->update= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
+  }
+  if (reenable_logging)
+    _ma_reenable_logging_for_table(info, FALSE);
+  DBUG_RETURN(error);
 }
 
 
@@ -6274,7 +6276,12 @@ my_bool write_log_record_for_repair(const HA_CHECK *param, MARIA_HA *info)
 }
 
 
-my_bool write_log_record_for_bulk_insert_with_repair(MARIA_HA *info)
+/**
+  Writes an UNDO record which if executed in UNDO phase, will empty the
+  table. Such record is thus logged only in certain cases of bulk insert
+  (table needs to be empty etc).
+*/
+my_bool write_log_record_for_bulk_insert(MARIA_HA *info)
 {
   LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
   uchar log_data[LSN_STORE_SIZE + FILEID_STORE_SIZE];
@@ -6282,7 +6289,7 @@ my_bool write_log_record_for_bulk_insert_with_repair(MARIA_HA *info)
   lsn_store(log_data, info->trn->undo_lsn);
   log_array[TRANSLOG_INTERNAL_PARTS + 0].str= (char*) log_data;
   log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
-  return translog_write_record(&lsn, LOGREC_UNDO_BULK_INSERT_WITH_REPAIR,
+  return translog_write_record(&lsn, LOGREC_UNDO_BULK_INSERT,
                                info->trn, info,
                                (translog_size_t)
                                log_array[TRANSLOG_INTERNAL_PARTS +
diff --git a/storage/maria/ma_key_recover.c b/storage/maria/ma_key_recover.c
index 1b24290f240..713ccded61b 100644
--- a/storage/maria/ma_key_recover.c
+++ b/storage/maria/ma_key_recover.c
@@ -176,7 +176,7 @@ 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:
+  case LOGREC_UNDO_BULK_INSERT:
     error= (maria_enable_indexes(tbl_info) ||
             /* we enabled indices, need '2' below */
             _ma_state_info_write(share, 1|2|4));
diff --git a/storage/maria/ma_loghandler.c b/storage/maria/ma_loghandler.c
index 6d47c734a4d..21e2262c4a9 100644
--- a/storage/maria/ma_loghandler.c
+++ b/storage/maria/ma_loghandler.c
@@ -613,11 +613,11 @@ 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=
+static LOG_DESC INIT_LOGREC_UNDO_BULK_INSERT=
 {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};
+ "undo_bulk_insert", LOGREC_LAST_IN_GROUP, NULL, NULL};
 
 static LOG_DESC INIT_LOGREC_REDO_BITMAP_NEW_PAGE=
 {LOGRECTYPE_FIXEDLENGTH, FILEID_STORE_SIZE + PAGE_STORE_SIZE * 2,
@@ -708,8 +708,8 @@ void translog_table_init()
     INIT_LOGREC_INCOMPLETE_LOG;
   log_record_type_descriptor[LOGREC_INCOMPLETE_GROUP]=
     INIT_LOGREC_INCOMPLETE_GROUP;
-  log_record_type_descriptor[LOGREC_UNDO_BULK_INSERT_WITH_REPAIR]=
-    INIT_LOGREC_UNDO_BULK_INSERT_WITH_REPAIR;
+  log_record_type_descriptor[LOGREC_UNDO_BULK_INSERT]=
+    INIT_LOGREC_UNDO_BULK_INSERT;
   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++)
diff --git a/storage/maria/ma_loghandler.h b/storage/maria/ma_loghandler.h
index 7b2489f6e0a..cc9e3347c42 100644
--- a/storage/maria/ma_loghandler.h
+++ b/storage/maria/ma_loghandler.h
@@ -141,7 +141,7 @@ enum translog_record_type
   LOGREC_LONG_TRANSACTION_ID,
   LOGREC_INCOMPLETE_LOG,
   LOGREC_INCOMPLETE_GROUP,
-  LOGREC_UNDO_BULK_INSERT_WITH_REPAIR,
+  LOGREC_UNDO_BULK_INSERT,
   LOGREC_REDO_BITMAP_NEW_PAGE,
   LOGREC_FIRST_FREE,
   LOGREC_RESERVED_FUTURE_EXTENSION= 63
diff --git a/storage/maria/ma_pagecache.c b/storage/maria/ma_pagecache.c
index 7e73a0b1abf..d69f37985b7 100644
--- a/storage/maria/ma_pagecache.c
+++ b/storage/maria/ma_pagecache.c
@@ -4309,6 +4309,27 @@ err:
 
 
 #ifndef DBUG_OFF
+
+/**
+  Verifies that a file has no dirty pages.
+*/
+
+void pagecache_file_no_dirty_page(PAGECACHE *pagecache, PAGECACHE_FILE *file)
+{
+  File fd= file->file;
+  PAGECACHE_BLOCK_LINK *block;
+  for (block= pagecache->changed_blocks[FILE_HASH(*file)];
+       block != NULL;
+       block= block->next_changed)
+    if (block->hash_link->file.file == fd)
+    {
+      DBUG_PRINT("info", ("pagecache_file_not_in error"));
+      PCBLOCK_INFO(block);
+      DBUG_ASSERT(0);
+    }
+}
+
+
 /*
   Test if disk-cache is ok
 */
diff --git a/storage/maria/ma_pagecache.h b/storage/maria/ma_pagecache.h
index 930322e132a..4c6f6e3897d 100644
--- a/storage/maria/ma_pagecache.h
+++ b/storage/maria/ma_pagecache.h
@@ -308,6 +308,11 @@ extern void multi_pagecache_change(PAGECACHE *old_data,
 				   PAGECACHE *new_data);
 extern int reset_pagecache_counters(const char *name,
                                     PAGECACHE *pagecache);
+#ifndef DBUG_OFF
+void pagecache_file_no_dirty_page(PAGECACHE *pagecache, PAGECACHE_FILE *file);
+#else
+#define pagecache_file_no_dirty_page(A,B) {}
+#endif
 
 C_MODE_END
 #endif /* _keycache_h */
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index 5cf05fc21c0..46f8d9fc62a 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -76,7 +76,7 @@ prototype_redo_exec_hook(REDO_DROP_TABLE);
 prototype_redo_exec_hook(FILE_ID);
 prototype_redo_exec_hook(INCOMPLETE_LOG);
 prototype_redo_exec_hook_dummy(INCOMPLETE_GROUP);
-prototype_redo_exec_hook(UNDO_BULK_INSERT_WITH_REPAIR);
+prototype_redo_exec_hook(UNDO_BULK_INSERT);
 prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
 prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL);
 prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
@@ -103,7 +103,7 @@ prototype_undo_exec_hook(UNDO_ROW_UPDATE);
 prototype_undo_exec_hook(UNDO_KEY_INSERT);
 prototype_undo_exec_hook(UNDO_KEY_DELETE);
 prototype_undo_exec_hook(UNDO_KEY_DELETE_WITH_ROOT);
-prototype_undo_exec_hook(UNDO_BULK_INSERT_WITH_REPAIR);
+prototype_undo_exec_hook(UNDO_BULK_INSERT);
 
 static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply);
 static uint end_of_redo_phase(my_bool prepare_for_undo_phase);
@@ -351,6 +351,11 @@ int maria_apply_log(LSN from_lsn, enum maria_apply_log_way apply,
       We take a checkpoint as it can save future recovery work if we crash
       during the UNDO phase. But we don't flush pages, as UNDOs will change
       them again probably.
+      If we wanted to take checkpoints in the middle of the REDO phase, at a
+      moment when we haven't reached the end of log so don't have exact data
+      about transactions, we could write a special checkpoint: containing only
+      the list of dirty pages, otherwise to be treated as if it was at the
+      same LSN as the last checkpoint.
     */
     if (ma_checkpoint_execute(CHECKPOINT_INDIRECT, FALSE))
       goto err;
@@ -1222,8 +1227,6 @@ static int new_table(uint16 sid, const char *name, LSN lsn_of_file_id)
   }
   /* don't log any records for this work */
   _ma_tmp_disable_logging_for_table(info, FALSE);
-  /* _ma_unpin_all_pages() reads info->trn: */
-  info->trn= &dummy_transaction_object;
   /* execution of some REDO records relies on data_file_length */
   dfile_len= my_seek(info->dfile.file, 0, SEEK_END, MYF(MY_WME));
   kfile_len= my_seek(info->s->kfile.file, 0, SEEK_END, MYF(MY_WME));
@@ -1833,7 +1836,7 @@ prototype_redo_exec_hook(UNDO_KEY_DELETE_WITH_ROOT)
 }
 
 
-prototype_redo_exec_hook(UNDO_BULK_INSERT_WITH_REPAIR)
+prototype_redo_exec_hook(UNDO_BULK_INSERT)
 {
   /*
     If the repair finished it wrote and sync the state. If it didn't finish,
@@ -1937,7 +1940,7 @@ prototype_redo_exec_hook(CLR_END)
                                       page * share->block_size);
       break;
     }
-    case LOGREC_UNDO_BULK_INSERT_WITH_REPAIR:
+    case LOGREC_UNDO_BULK_INSERT:
       break;
     default:
       DBUG_ASSERT(0);
@@ -2226,7 +2229,7 @@ prototype_undo_exec_hook(UNDO_KEY_DELETE_WITH_ROOT)
 }
 
 
-prototype_undo_exec_hook(UNDO_BULK_INSERT_WITH_REPAIR)
+prototype_undo_exec_hook(UNDO_BULK_INSERT)
 {
   my_bool error;
   MARIA_HA *info= get_MARIA_HA_from_UNDO_record(rec);
@@ -2244,7 +2247,7 @@ prototype_undo_exec_hook(UNDO_BULK_INSERT_WITH_REPAIR)
                           STATE_NOT_ZEROFILLED | STATE_NOT_MOVABLE);
 
   info->trn= trn;
-  error= _ma_apply_undo_bulk_insert_with_repair(info, previous_undo_lsn);
+  error= _ma_apply_undo_bulk_insert(info, previous_undo_lsn);
   info->trn= 0;
   /* trn->undo_lsn is updated in an inwrite_hook when writing the CLR_END */
   tprint(tracef, "   undo_lsn now LSN (%lu,0x%lx)\n",
@@ -2309,8 +2312,8 @@ static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply)
   install_redo_exec_hook_shared(REDO_NEW_ROW_HEAD, REDO_INSERT_ROW_HEAD);
   /* REDO_NEW_ROW_TAIL shares entry with REDO_INSERT_ROW_TAIL */
   install_redo_exec_hook_shared(REDO_NEW_ROW_TAIL, REDO_INSERT_ROW_TAIL);
-  install_redo_exec_hook(UNDO_BULK_INSERT_WITH_REPAIR);
-  install_undo_exec_hook(UNDO_BULK_INSERT_WITH_REPAIR);
+  install_redo_exec_hook(UNDO_BULK_INSERT);
+  install_undo_exec_hook(UNDO_BULK_INSERT);
 
   current_group_end_lsn= LSN_IMPOSSIBLE;
 #ifndef DBUG_OFF
@@ -2715,7 +2718,13 @@ static void prepare_table_for_close(MARIA_HA *info, TRANSLOG_ADDRESS horizon)
     share->state.is_of_horizon= horizon;
     _ma_state_info_write_sub(share->kfile.file, &share->state, 1);
   }
-  _ma_reenable_logging_for_table(info);
+  /*
+    This leaves PAGECACHE_PLAIN_PAGE pages into the cache, while the table is
+    going to switch back to transactional. So the table will be a mix of
+    pages, which is ok as long as we don't take any checkpoints until all
+    tables get closed at the end of the UNDO phase.
+  */
+  _ma_reenable_logging_for_table(info, FALSE);
   info->trn= NULL; /* safety */
 }
 
@@ -3175,12 +3184,13 @@ void _ma_tmp_disable_logging_for_table(MARIA_HA *info,
   /* if we disabled before writing the record, record wouldn't reach log */
   share->now_transactional= FALSE;
   /*
-    Some code in ma_blockrec.c assumes a trn.
-    info->trn in some cases can be not NULL and not dummy_transaction_object
-    when arriving here, but overwriting it does not leak as it is still
-    remembered in THD_TRN.
+    Some code in ma_blockrec.c assumes a trn even if !now_transactional but in
+    this case it only reads trn->rec_lsn, which has to be LSN_IMPOSSIBLE and
+    should be now. info->trn may be NULL in maria_chk.
   */
-  info->trn= &dummy_transaction_object;
+  if (info->trn == NULL)
+    info->trn= &dummy_transaction_object;
+  DBUG_ASSERT(info->trn->rec_lsn == LSN_IMPOSSIBLE);
   share->page_type= PAGECACHE_PLAIN_PAGE;
   /* Functions below will pick up now_transactional and change callbacks */
   _ma_set_data_pagecache_callbacks(&info->dfile, share);
@@ -3194,30 +3204,60 @@ void _ma_tmp_disable_logging_for_table(MARIA_HA *info,
    Re-enables logging for a table which had it temporarily disabled.
 
    @param  info            table
+   @param  flush_pages     if function needs to flush pages first
 */
 
-void _ma_reenable_logging_for_table(MARIA_HA *info)
+my_bool _ma_reenable_logging_for_table(MARIA_HA *info, my_bool flush_pages)
 {
   MARIA_SHARE *share= info->s;
   DBUG_ENTER("_ma_reenable_logging_for_table");
 
   if (share->now_transactional == share->base.born_transactional)
-    DBUG_VOID_RETURN;
+    DBUG_RETURN(0);
 
   if ((share->now_transactional= share->base.born_transactional))
   {
+    share->page_type= PAGECACHE_LSN_PAGE;
+    if (flush_pages)
+    {
+      /*
+        We are going to change callbacks; if a page is flushed at this moment
+        this can cause race conditions, that's one reason to flush pages
+        now. Other reasons: a checkpoint could be running and miss pages. As
+        there are no REDOs for pages, them, bitmaps and the state also have to
+        be flushed and synced. Leaving non-dirty pages in cache is ok, when
+        they become dirty again they will have their type corrected.
+      */
+      if (_ma_flush_table_files(info, MARIA_FLUSH_DATA | MARIA_FLUSH_INDEX,
+                                FLUSH_KEEP, FLUSH_KEEP) ||
+          _ma_state_info_write(share, 1|4) ||
+          _ma_sync_table_files(info))
+        DBUG_RETURN(1);
+    }
+    else if (!maria_in_recovery)
+    {
+      /*
+        Except in Recovery, we mustn't leave dirty pages (see comments above).
+        Note that this does not verify that the state was flushed, but hey.
+      */
+      pagecache_file_no_dirty_page(share->pagecache, &info->dfile);
+      pagecache_file_no_dirty_page(share->pagecache, &share->kfile);
+    }
+    _ma_set_data_pagecache_callbacks(&info->dfile, share);
+    _ma_set_index_pagecache_callbacks(&share->kfile, share);
+    _ma_bitmap_set_pagecache_callbacks(&share->bitmap.file, share);
     /*
-      The change below does NOT affect pages already in the page cache, so you
-      should have flushed them out already, or write a pagecache function to
-      change their type.
+      info->trn was not changed in the disable/enable combo, so that it's
+      still usable in this kind of combination:
+      external_lock;
+      start_bulk_insert; # table is empty, disables logging
+      end_bulk_insert;   # enables logging
+      start_bulk_insert; # table is not empty, logging stays
+                         # so rows insertion needs the real trn.
+      as happens during row-based replication on the slave.
     */
-    share->page_type= PAGECACHE_LSN_PAGE;
-    info->trn= NULL; /* safety */
   }
-  _ma_set_data_pagecache_callbacks(&info->dfile, share);
-  _ma_set_index_pagecache_callbacks(&share->kfile, share);
-  _ma_bitmap_set_pagecache_callbacks(&share->bitmap.file, share);
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(0);
 }
 
 
diff --git a/storage/maria/maria_chk.c b/storage/maria/maria_chk.c
index 11f7b47ca48..35e9fd26700 100644
--- a/storage/maria/maria_chk.c
+++ b/storage/maria/maria_chk.c
@@ -1070,7 +1070,10 @@ static int maria_chk(HA_CHECK *param, char *filename)
   if (param->testflag & (T_REP_ANY | T_SORT_RECORDS | T_SORT_INDEX |
                          T_ZEROFILL))
   {
-    /* Mark table as not transactional to avoid logging */
+    /*
+      Mark table as not transactional to avoid logging. Should not be needed,
+      maria_repair and maria_zerofill do it already.
+    */
     _ma_tmp_disable_logging_for_table(info, FALSE);
 
     if (param->testflag & T_REP_ANY)
@@ -1253,7 +1256,7 @@ static int maria_chk(HA_CHECK *param, char *filename)
                                     ((param->testflag & T_SORT_RECORDS) ?
                                      UPDATE_SORT : 0)));
   info->update&= ~HA_STATE_CHANGED;
-  _ma_reenable_logging_for_table(info);
+  _ma_reenable_logging_for_table(info, FALSE);
   maria_lock_database(info, F_UNLCK);
 
 end2:
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index fc0fe9366f7..a1cbbc101be 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -1100,8 +1100,8 @@ void _ma_set_index_pagecache_callbacks(PAGECACHE_FILE *file,
                                        MARIA_SHARE *share);
 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);
+my_bool _ma_reenable_logging_for_table(MARIA_HA *info, my_bool flush_pages);
+my_bool write_log_record_for_bulk_insert(MARIA_HA *info);
 
 
 #define MARIA_NO_CRC_NORMAL_PAGE 0xffffffff