Write message to stderr if recovery is not likely to succeed because we don't log records for batch inserts.

Don't do UNDO or REDO on a crashed table.
More DBUG_PRINT

sql/sql_parse.cc:
  Remove display of 'packet' which is not useful
storage/maria/ma_blockrec.c:
  More DBUG_PRINT()
storage/maria/ma_key_recover.c:
  Write page number instead of page position
storage/maria/ma_recovery.c:
  Write message to stderr if recovery is not likely to succeed because we don't log records for batch inserts.
  In normal cases this should never be an issue as we would recreate an empty table if MariaDB dies under batch insert.
  This warning will be given if you remove all tables and try to recreate them with maria_read_log, which can't be done as
  the log doesn't contain all data.
  Don't do UNDO or REDO on a crashed table.
storage/maria/ma_write.c:
  Write page number instead of page position.

sql/sql_parse.cc

@@ -997,7 +997,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
   NET *net= &thd->net;
   bool error= 0;
   DBUG_ENTER("dispatch_command");
-  DBUG_PRINT("info",("packet: '%*.s'; command: %d", packet_length, packet, command));
+  DBUG_PRINT("info", ("command: %d", command));
 
   thd->command=command;
   /*

storage/maria/ma_blockrec.c

@@ -1716,7 +1716,7 @@ static my_bool get_head_or_tail_page(MARIA_HA *info,
   MARIA_PINNED_PAGE page_link;
   MARIA_SHARE *share= info->s;
   DBUG_ENTER("get_head_or_tail_page");
-  DBUG_PRINT("enter", ("length: %u", length));
+  DBUG_PRINT("enter", ("page_type: %u  length: %u", page_type, length));
 
   block_size= share->block_size;
   if (block->org_bitmap_value == 0)             /* Empty block */
@@ -6238,7 +6238,10 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
       /* Skip errors when reading outside of file and uninitialized pages */
       if (!new_page || (my_errno != HA_ERR_FILE_TOO_SHORT &&
                         my_errno != HA_ERR_WRONG_CRC))
+      {
+        DBUG_PRINT("error", ("Error %d when reading page", (int) my_errno));
         goto err;
+      }
       /* Create new page */
       buff= pagecache_block_link_to_buffer(page_link.link);
       buff[PAGE_TYPE_OFFSET]= UNALLOCATED_PAGE;
@@ -6266,7 +6269,13 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
         changed to new type.
       */
       if (!new_page)
+      {
+        DBUG_PRINT("error",
+                   ("Found page of wrong type: %u, should have been %u",
+                    (uint) (buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK),
+                    page_type));
         goto crashed_file;
+      }
       make_empty_page(info, buff, page_type, 0);
       empty_space= block_size - PAGE_HEADER_SIZE - PAGE_SUFFIX_SIZE;
       (void) extend_directory(page_type == HEAD_PAGE ? info: 0, buff,

storage/maria/ma_key_recover.c

@@ -504,7 +504,8 @@ my_bool _ma_log_add(MARIA_PAGE *ma_page,
   DBUG_ENTER("_ma_log_add");
   DBUG_PRINT("enter", ("page: %lu  org_page_length: %u  changed_length: %u  "
                        "move_length: %d",
-                       (ulong) ma_page->pos, org_page_length, changed_length,
+                       (ulong) (ma_page->pos / info->s->block_size),
+                       org_page_length, changed_length,
                        move_length));
   DBUG_ASSERT(info->s->now_transactional);
   DBUG_ASSERT(move_length <= (int) changed_length);

storage/maria/ma_recovery.c

@@ -638,6 +638,16 @@ prototype_redo_exec_hook(INCOMPLETE_LOG)
     /* no such table, don't need to warn */
     return 0;
   }
+  if (info->s->state.is_of_horizon > rec->lsn)
+  {
+    /*
+      This table was repaired at a time after this log entry.
+      We can assume that all rows was inserted sucessfully and we don't
+      have to warn about that the inserted data was not logged
+    */
+    return 0;
+  }
+
   /*
     Example of what can go wrong when replaying DDLs:
     CREATE TABLE t (logged); INSERT INTO t VALUES(1) (logged);
@@ -658,10 +668,15 @@ prototype_redo_exec_hook(INCOMPLETE_LOG)
     there was a crash during a DDL (see comment in execution of
     REDO_RENAME_TABLE ).
   */
-  tprint(tracef, "***WARNING: MySQL server currently logs no records"
+
+  eprint(tracef, "***WARNING: Aria engine currently logs no records "
           "about insertion of data by ALTER TABLE and CREATE SELECT, "
           "as they are not necessary for recovery; "
-         " present applying of log records may well not work.***\n");
+          "present applying of log records to table '%s' may well not work."
+          "***\n", info->s->index_file_name.str);
+
+  /* Prevent using the table for anything else than undo repair */
+  _ma_mark_file_crashed(info->s);
   recovery_warnings++;
   return 0;
 }
@@ -1052,7 +1067,11 @@ prototype_redo_exec_hook(REDO_REPAIR_TABLE)
   }
   if ((info= get_MARIA_HA_from_REDO_record(rec)) == NULL)
     DBUG_RETURN(0);
-
+  if (maria_is_crashed(info))
+  {
+    tprint(tracef, "we skip repairing crashed table\n");
+    DBUG_RETURN(0);
+  }
   /*
     Otherwise, the mapping is newer than the table, and our record is newer
     than the mapping, so we can repair.
@@ -1367,7 +1386,8 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD)
   int error= 1;
   uchar *buff= NULL;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
+
   {
     /*
       Table was skipped at open time (because later dropped/renamed, not
@@ -1433,7 +1453,7 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL)
   int error= 1;
   uchar *buff;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
   enlarge_buffer(rec);
   if (log_record_buffer.str == NULL ||
@@ -1474,7 +1494,7 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_BLOBS)
   pgcache_page_no_t first_page, last_page;
   char llbuf1[22], llbuf2[22];
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL  || maria_is_crashed(info))
     return 0;
   enlarge_buffer(rec);
   if (log_record_buffer.str == NULL ||
@@ -1508,7 +1528,7 @@ prototype_redo_exec_hook(REDO_PURGE_ROW_HEAD)
 {
   int error= 1;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
   if (_ma_apply_redo_purge_row_head_or_tail(info, current_group_end_lsn,
                                             HEAD_PAGE,
@@ -1524,7 +1544,7 @@ prototype_redo_exec_hook(REDO_PURGE_ROW_TAIL)
 {
   int error= 1;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
   if (_ma_apply_redo_purge_row_head_or_tail(info, current_group_end_lsn,
                                             TAIL_PAGE,
@@ -1541,7 +1561,7 @@ prototype_redo_exec_hook(REDO_FREE_BLOCKS)
   int error= 1;
   uchar *buff;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
   enlarge_buffer(rec);
 
@@ -1568,7 +1588,7 @@ prototype_redo_exec_hook(REDO_FREE_HEAD_OR_TAIL)
 {
   int error= 1;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
 
   if (_ma_apply_redo_free_head_or_tail(info, current_group_end_lsn,
@@ -1600,7 +1620,7 @@ prototype_redo_exec_hook(REDO_INDEX)
 {
   int error= 1;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
   enlarge_buffer(rec);
 
@@ -1626,7 +1646,7 @@ prototype_redo_exec_hook(REDO_INDEX_NEW_PAGE)
 {
   int error= 1;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
   enlarge_buffer(rec);
 
@@ -1653,7 +1673,7 @@ prototype_redo_exec_hook(REDO_INDEX_FREE_PAGE)
 {
   int error= 1;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
 
   if (_ma_apply_redo_index_free_page(info, current_group_end_lsn,
@@ -1669,7 +1689,7 @@ prototype_redo_exec_hook(REDO_BITMAP_NEW_PAGE)
 {
   int error= 1;
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
     return 0;
   enlarge_buffer(rec);
 
@@ -2115,7 +2135,7 @@ prototype_undo_exec_hook(UNDO_ROW_INSERT)
   MARIA_SHARE *share;
   const uchar *record_ptr;
 
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
   {
     /*
       Unlike for REDOs, if the table was skipped it is abnormal; we have a
@@ -2171,7 +2191,7 @@ prototype_undo_exec_hook(UNDO_ROW_DELETE)
   LSN previous_undo_lsn= lsn_korr(rec->header);
   MARIA_SHARE *share;
 
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
   {
     skip_undo_record(previous_undo_lsn, trn);
     return 0;
@@ -2210,7 +2230,7 @@ prototype_undo_exec_hook(UNDO_ROW_UPDATE)
   LSN previous_undo_lsn= lsn_korr(rec->header);
   MARIA_SHARE *share;
 
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
   {
     skip_undo_record(previous_undo_lsn, trn);
     return 0;
@@ -2249,7 +2269,7 @@ prototype_undo_exec_hook(UNDO_KEY_INSERT)
   LSN previous_undo_lsn= lsn_korr(rec->header);
   MARIA_SHARE *share;
 
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
   {
     skip_undo_record(previous_undo_lsn, trn);
     return 0;
@@ -2290,7 +2310,7 @@ prototype_undo_exec_hook(UNDO_KEY_DELETE)
   LSN previous_undo_lsn= lsn_korr(rec->header);
   MARIA_SHARE *share;
 
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
   {
     skip_undo_record(previous_undo_lsn, trn);
     return 0;
@@ -2331,7 +2351,7 @@ prototype_undo_exec_hook(UNDO_KEY_DELETE_WITH_ROOT)
   LSN previous_undo_lsn= lsn_korr(rec->header);
   MARIA_SHARE *share;
 
-  if (info == NULL)
+  if (info == NULL || maria_is_crashed(info))
   {
     skip_undo_record(previous_undo_lsn, trn);
     return 0;
@@ -2372,6 +2392,7 @@ prototype_undo_exec_hook(UNDO_BULK_INSERT)
   LSN previous_undo_lsn= lsn_korr(rec->header);
   MARIA_SHARE *share;
 
+  /* Here we don't check for crashed as we can undo the bulk insert */
   if (info == NULL)
   {
     skip_undo_record(previous_undo_lsn, trn);

storage/maria/ma_write.c

@@ -613,7 +613,7 @@ static int w_search(register MARIA_HA *info, uint32 comp_flag, MARIA_KEY *key,
   MARIA_KEYDEF *keyinfo= key->keyinfo;
   MARIA_PAGE page;
   DBUG_ENTER("w_search");
-  DBUG_PRINT("enter",("page: %ld", (long) page_pos));
+  DBUG_PRINT("enter", ("page: %lu", (ulong) (page_pos/keyinfo->block_length)));
 
   if (!(temp_buff= (uchar*) my_alloca((uint) keyinfo->block_length+
 				      MARIA_MAX_KEY_BUFF*2)))