Fixed bug: lp:887051 ; Error in recovery with LOAD DATA + DELETE

mysql-test/suite/maria/r/maria-recovery3.result:
  Added test case for recovery bug
mysql-test/suite/maria/t/maria-recovery3.test:
  Added test case for recovery bug
storage/maria/ha_maria.cc:
  Don't print query twice to log
storage/maria/ma_delete.c:
  More DBUG_PRINT
storage/maria/ma_key_recover.c:
  Added new asserts to detect errors earlier
storage/maria/ma_recovery.c:
  Update all states when moving a non-transactional file to transactional. This fixes lp:887051

mysql-test/std_data/bug887051.txt

+saved1
+saved2

mysql-test/suite/maria/r/maria-recovery3.result

@@ -89,6 +89,28 @@ check table t1 extended;
 Table	Op	Msg_type	Msg_text
 mysqltest.t1	check	status	OK
 drop table t1;
+CREATE TABLE t1 ( word VARCHAR(255) PRIMARY KEY ) ENGINE=Aria;
+LOAD DATA INFILE '../../std_data/bug887051.txt' INTO TABLE t1;
+SET AUTOCOMMIT=0;
+DELETE FROM t1;
+LOAD DATA INFILE '../../std_data/bug887051.txt' INTO TABLE t1 IGNORE 1 LINES;
+COMMIT;
+SET SESSION debug="+d,maria_flush_whole_log,maria_crash";
+* crashing mysqld intentionally
+set global aria_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;
+word
+saved2
+drop table t1;
 drop database mysqltest_for_feeding_recovery;
 drop database mysqltest_for_comparison;
 drop database mysqltest;

mysql-test/suite/maria/t/maria-recovery3.test

@@ -109,6 +109,24 @@ truncate table t1;
 check table t1 extended;
 drop table t1;
 
+#
+# Test for BUG#887051; Failure in recovery with delete
+#
+
+let $mvr_restore_old_snapshot=0;
+let $mms_compare_physically=0;
+let $mvr_debug_option="+d,maria_flush_whole_log,maria_crash";
+let $mvr_crash_statement= set global aria_checkpoint_interval=1;
+CREATE TABLE t1 ( word VARCHAR(255) PRIMARY KEY ) ENGINE=Aria;
+LOAD DATA INFILE '../../std_data/bug887051.txt' INTO TABLE t1;
+SET AUTOCOMMIT=0;
+DELETE FROM t1;
+LOAD DATA INFILE '../../std_data/bug887051.txt' INTO TABLE t1 IGNORE 1 LINES;
+COMMIT; 
+-- source include/maria_verify_recovery.inc
+select * from t1;
+drop table t1;
+
 # clean up everything
 let $mms_purpose=feeding_recovery;
 eval drop database mysqltest_for_$mms_purpose;

storage/maria/ha_maria.cc

@@ -2431,8 +2431,16 @@ int ha_maria::extra_opt(enum ha_extra_function operation, ulong cache_size)
 int ha_maria::delete_all_rows()
 {
   THD *thd= table->in_use;
-  (void) translog_log_debug_info(file->trn, LOGREC_DEBUG_INFO_QUERY,
-                                 (uchar*) thd->query(), thd->query_length());
+#ifdef EXTRA_DEBUG
+  TRN *trn= file->trn;
+  if (trn && ! (trnman_get_flags(trn) & TRN_STATE_INFO_LOGGED))
+  {
+    trnman_set_flags(trn, trnman_get_flags(trn) | TRN_STATE_INFO_LOGGED |
+                     TRN_STATE_TABLES_CAN_CHANGE);
+    (void) translog_log_debug_info(trn, LOGREC_DEBUG_INFO_QUERY,
+                                   (uchar*) thd->query(), thd->query_length());
+  }
+#endif
   if (file->s->now_transactional &&
       ((table->in_use->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) ||
        table->in_use->locked_tables))

storage/maria/ma_delete.c

@@ -1481,8 +1481,9 @@ my_bool _ma_log_delete(MARIA_PAGE *ma_page, const uchar *key_pos,
   MARIA_SHARE *share= info->s;
   my_off_t page= ma_page->pos / share->block_size;
   DBUG_ENTER("_ma_log_delete");
-  DBUG_PRINT("enter", ("page: %lu  changed_length: %u  move_length: %d",
-                       (ulong) page, changed_length, move_length));
+  DBUG_PRINT("enter", ("page: %lu  offset: %u  changed_length: %u  move_length: %u  append_length: %u  page_size: %u",
+                       (ulong) page, offset, changed_length, move_length,
+                       append_length, ma_page->size));
   DBUG_ASSERT(share->now_transactional && move_length);
   DBUG_ASSERT(offset + changed_length <= ma_page->size);
   DBUG_ASSERT(ma_page->org_size - move_length + append_length == ma_page->size);

storage/maria/ma_key_recover.c

@@ -945,7 +945,7 @@ uint _ma_apply_redo_index(MARIA_HA *info,
   const uchar *header_end= header + head_length;
   uint page_offset= 0, org_page_length;
   uint page_length, keypage_header, keynr;
-  uint max_page_size= share->max_index_block_size;
+  uint max_page_size= share->max_index_block_size, new_page_length= 0;
   int result;
   MARIA_PAGE page;
   DBUG_ENTER("_ma_apply_redo_index");
@@ -1106,6 +1106,8 @@ uint _ma_apply_redo_index(MARIA_HA *info,
     case KEY_OP_DEBUG_2:
       DBUG_PRINT("redo", ("org_page_length: %u  new_page_length: %u",
                           uint2korr(header), uint2korr(header+2)));
+      DBUG_ASSERT(uint2korr(header) == page_length);
+      new_page_length= uint2korr(header+2);
       header+= 4;
       break;
     case KEY_OP_MAX_PAGELENGTH:
@@ -1171,6 +1173,7 @@ uint _ma_apply_redo_index(MARIA_HA *info,
     }
   } while (header < header_end);
   DBUG_ASSERT(header == header_end);
+  DBUG_ASSERT(new_page_length == 0 || new_page_length == page_length);
 
   /* Write modified page */
   page.size= page_length;

storage/maria/ma_recovery.c

@@ -3592,6 +3592,10 @@ my_bool _ma_reenable_logging_for_table(MARIA_HA *info, my_bool flush_pages)
 
     if (flush_pages)
     {
+      /* Ensure that recover is not executing any redo before this */
+      if (!maria_in_recovery)
+        share->state.is_of_horizon= share->state.create_rename_lsn=
+          share->state.skip_redo_lsn= translog_get_horizon();
       /*
         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