Added undo of deleted row

Added part of undo of update row
Extended ma_test1 for recovery testing
Some bug fixes


storage/maria/ha_maria.cc:
  Ignore 'state.split' in case of block records
storage/maria/ma_bitmap.c:
  Added return value for _ma_bitmap_find_place() for how much data we should put on head page
storage/maria/ma_blockrec.c:
  Added undo of deleted row.
  - Added logging of CLR_END records in write_block_record()
  - Split ma_write_init_block_record() to two functions to get better code reuse
  - Added _ma_apply_undo_row_delete()
  - Added ma_get_length()
  
  Added 'empty' prototype for undo_row_update()
  
  Fixed bug when moving data withing a head/tail page.
  Fixed bug when reading a page with bigger LSN but of different type than was expected.
  Store undo_lsn first in CLR_END record
  
  Simplified some code by adding local variables.
  Changed log format for UNDO_ROW_DELETE to store total length of used blobs
storage/maria/ma_blockrec.h:
  Added prototypes for undo code.
storage/maria/ma_pagecache.c:
  Allow plain page to change to LSN page (needed in recovery to apply UNDO)
storage/maria/ma_recovery.c:
  Added undo handling of UNDO_ROW_DELETE and UNDO_ROW_UPDATE
storage/maria/ma_test1.c:
  Extended --test-undo option to allow us to die after insert or after delete.
  Fixed bug in printing key values when using -v
storage/maria/maria_def.h:
  Moved some variables around to be getter alignment
  Added length_buff buffer to be used during undo handling

storage/maria/ha_maria.cc

@@ -1219,7 +1219,9 @@ int ha_maria::repair(THD *thd, HA_CHECK &param, bool do_optimize)
   }
 
   if (!do_optimize ||
-      ((file->state->del || share->state.split != file->state->records) &&
+      ((file->state->del ||
+        ((file->s->data_file_type != BLOCK_RECORD) &&
+         share->state.split != file->state->records)) &&
        (!(param.testflag & T_QUICK) ||
         (share->state.changed & (STATE_NOT_OPTIMIZED_KEYS |
                                  STATE_NOT_OPTIMIZED_ROWS)))))

storage/maria/ma_bitmap.c

@@ -1421,6 +1421,8 @@ static my_bool write_rest_of_head(MARIA_HA *info, uint position,
 
   RETURN
     0  ok
+       row->space_on_head_page contains minimum number of bytes we
+       expect to put on the head page.
     1  error
 */
 
@@ -1457,6 +1459,7 @@ my_bool _ma_bitmap_find_place(MARIA_HA *info, MARIA_ROW *row,
     position= ELEMENTS_RESERVED_FOR_MAIN_PART - 1;
     if (find_head(info, (uint) row->total_length, position))
       goto abort;
+    row->space_on_head_page= row->total_length;
     goto end;
   }
 
@@ -1474,6 +1477,7 @@ my_bool _ma_bitmap_find_place(MARIA_HA *info, MARIA_ROW *row,
     position= ELEMENTS_RESERVED_FOR_MAIN_PART - 1;
     if (find_head(info, head_length, position))
       goto abort;
+    row->space_on_head_page= head_length;
     goto end;
   }
 
@@ -1490,6 +1494,7 @@ my_bool _ma_bitmap_find_place(MARIA_HA *info, MARIA_ROW *row,
     position= ELEMENTS_RESERVED_FOR_MAIN_PART -2;    /* Only head and tail */
   if (find_head(info, row_length, position))
     goto abort;
+  row->space_on_head_page= row_length;
   rest_length= head_length - row_length;
   if (write_rest_of_head(info, position, rest_length))
     goto abort;

storage/maria/ma_blockrec.h

@@ -189,3 +189,7 @@ uint _ma_apply_redo_purge_blocks(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);

storage/maria/ma_pagecache.c

@@ -3258,7 +3258,9 @@ restart:
 
     DBUG_ASSERT(block->type == PAGECACHE_EMPTY_PAGE ||
                 block->type == PAGECACHE_READ_UNKNOWN_PAGE ||
-                block->type == type);
+                block->type == type ||
+                (block->type == PAGECACHE_PLAIN_PAGE &&
+                 type == PAGECACHE_LSN_PAGE));
     block->type= type;
 
     if (make_lock_and_pin(pagecache, block,

storage/maria/ma_recovery.c

@@ -76,6 +76,8 @@ prototype_redo_exec_hook(UNDO_ROW_UPDATE);
 prototype_redo_exec_hook(UNDO_ROW_PURGE);
 prototype_redo_exec_hook(COMMIT);
 prototype_undo_exec_hook(UNDO_ROW_INSERT);
+prototype_undo_exec_hook(UNDO_ROW_DELETE);
+prototype_undo_exec_hook(UNDO_ROW_UPDATE);
 
 static int run_redo_phase(LSN lsn, my_bool apply);
 static uint end_of_redo_phase(my_bool prepare_for_undo_phase);
@@ -977,6 +979,88 @@ prototype_undo_exec_hook(UNDO_ROW_INSERT)
 }
 
 
+prototype_undo_exec_hook(UNDO_ROW_DELETE)
+{
+  my_bool error;
+  MARIA_HA *info= get_MARIA_HA_from_UNDO_record(rec);
+
+  if (info == NULL)
+    return 1;
+
+  info->s->state.changed|= STATE_CHANGED | STATE_NOT_ANALYZED |
+    STATE_NOT_OPTIMIZED_KEYS | STATE_NOT_SORTED_PAGES;
+
+  enlarge_buffer(rec);
+  if (log_record_buffer.str == NULL ||
+      translog_read_record(rec->lsn, 0, rec->record_length,
+                           log_record_buffer.str, NULL) !=
+       rec->record_length)
+  {
+    fprintf(tracef, "Failed to read record\n");
+    return 1;
+  }
+
+  /* Set undo to point to previous undo record */
+  info->trn= trn;
+  info->trn->undo_lsn= lsn_korr(rec->header);
+
+  /*
+    For now we skip the page and directory entry. This is to be used
+    later when we mark rows as deleted.
+  */
+  error= _ma_apply_undo_row_delete(info, rec->lsn,
+                                   log_record_buffer.str + LSN_STORE_SIZE +
+                                   FILEID_STORE_SIZE + PAGE_STORE_SIZE +
+                                   DIRPOS_STORE_SIZE,
+                                   rec->record_length -
+                                   (LSN_STORE_SIZE + FILEID_STORE_SIZE +
+                                    PAGE_STORE_SIZE + DIRPOS_STORE_SIZE));
+  info->trn= 0;
+  return error;
+}
+
+
+prototype_undo_exec_hook(UNDO_ROW_UPDATE)
+{
+  my_bool error;
+  MARIA_HA *info= get_MARIA_HA_from_UNDO_record(rec);
+
+  if (info == NULL)
+    return 1;
+
+  info->s->state.changed|= STATE_CHANGED | STATE_NOT_ANALYZED |
+    STATE_NOT_OPTIMIZED_KEYS | STATE_NOT_SORTED_PAGES;
+
+  enlarge_buffer(rec);
+  if (log_record_buffer.str == NULL ||
+      translog_read_record(rec->lsn, 0, rec->record_length,
+                           log_record_buffer.str, NULL) !=
+       rec->record_length)
+  {
+    fprintf(tracef, "Failed to read record\n");
+    return 1;
+  }
+
+  /* Set undo to point to previous undo record */
+  info->trn= trn;
+  info->trn->undo_lsn= lsn_korr(rec->header);
+
+  /*
+    For now we skip the page and directory entry. This is to be used
+    later when we mark rows as deleted.
+  */
+  error= _ma_apply_undo_row_update(info, rec->lsn,
+                                   log_record_buffer.str + LSN_STORE_SIZE +
+                                   FILEID_STORE_SIZE + PAGE_STORE_SIZE +
+                                   DIRPOS_STORE_SIZE,
+                                   rec->record_length -
+                                   (LSN_STORE_SIZE + FILEID_STORE_SIZE +
+                                    PAGE_STORE_SIZE + DIRPOS_STORE_SIZE));
+  info->trn= 0;
+  return error;
+}
+
+
 static int run_redo_phase(LSN lsn, my_bool apply)
 {
   /* install hooks for execution */
@@ -1003,6 +1087,8 @@ static int run_redo_phase(LSN lsn, my_bool apply)
   install_redo_exec_hook(UNDO_ROW_PURGE);
   install_redo_exec_hook(COMMIT);
   install_undo_exec_hook(UNDO_ROW_INSERT);
+  install_undo_exec_hook(UNDO_ROW_DELETE);
+  install_undo_exec_hook(UNDO_ROW_UPDATE);
 
   current_group_end_lsn= LSN_IMPOSSIBLE;
 

storage/maria/ma_test1.c

@@ -37,8 +37,9 @@ static enum data_file_type record_type= DYNAMIC_RECORD;
 static uint insert_count, update_count, remove_count;
 static uint pack_keys=0, pack_seg=0, key_length;
 static uint unique_key=HA_NOSAME;
-static my_bool pagecacheing, null_fields, silent, skip_update, opt_unique,
-  verbose, skip_delete, transactional, die_in_middle_of_transaction;
+static uint die_in_middle_of_transaction;
+static my_bool pagecacheing, null_fields, silent, skip_update, opt_unique;
+static my_bool verbose, skip_delete, transactional;
 static MARIA_COLUMNDEF recinfo[4];
 static MARIA_KEYDEF keyinfo[10];
 static HA_KEYSEG keyseg[10];
@@ -94,6 +95,7 @@ static int run_test(const char *filename)
 {
   MARIA_HA *file;
   int i,j,error,deleted,rec_length,uniques=0;
+  uint offset_to_key;
   ha_rows found,row_count;
   char record[MAX_REC_LENGTH],key[MAX_REC_LENGTH],read_record[MAX_REC_LENGTH];
   MARIA_UNIQUEDEF uniquedef;
@@ -182,6 +184,10 @@ static int run_test(const char *filename)
   else
     uniques=0;
 
+  offset_to_key= test(null_fields);
+  if (key_field == FIELD_BLOB)
+    offset_to_key+= 2;
+
   if (!silent)
     printf("- Creating maria file\n");
   create_info.max_rows=(ulong) (rec_pointer_size ?
@@ -234,7 +240,7 @@ static int run_test(const char *filename)
     flags[0]=2;
   }
 
-  if (die_in_middle_of_transaction)
+  if (die_in_middle_of_transaction == 1)
   {
     /*
       Ensure we get changed pages and log to disk
@@ -242,6 +248,7 @@ static int run_test(const char *filename)
     */
     _ma_flush_table_files(file, MARIA_FLUSH_DATA, FLUSH_RELEASE,
                           FLUSH_RELEASE);
+    printf("Dying on request after insert without maria_close()\n");
     exit(1);
   }
 
@@ -333,14 +340,14 @@ static int run_test(const char *filename)
 	if (verbose || (flags[j] >= 1 ||
 			(error && my_errno != HA_ERR_KEY_NOT_FOUND)))
 	  printf("key: '%.*s'  maria_rkey:  %3d  errno: %3d\n",
-		 (int) key_length,key+test(null_fields),error,my_errno);
+		 (int) key_length,key+offset_to_key,error,my_errno);
       }
       else
       {
 	error=maria_delete(file,read_record);
 	if (verbose || error)
 	  printf("key: '%.*s'  maria_delete: %3d  errno: %3d\n",
-		 (int) key_length, key+test(null_fields), error, my_errno);
+		 (int) key_length, key+offset_to_key, error, my_errno);
 	if (! error)
 	{
 	  deleted++;
@@ -348,6 +355,18 @@ static int run_test(const char *filename)
 	}
       }
     }
+
+    if (die_in_middle_of_transaction == 2)
+    {
+      /*
+        Ensure we get changed pages and log to disk
+        As commit record is not done, the undo entries needs to be rolled back.
+      */
+      _ma_flush_table_files(file, MARIA_FLUSH_DATA, FLUSH_RELEASE,
+                            FLUSH_RELEASE);
+      printf("Dying on request after delete without maria_close()\n");
+      exit(1);
+    }
   }
   if (!silent)
     printf("- Reading rows with key\n");
@@ -362,7 +381,7 @@ static int run_test(const char *filename)
 	(error && (flags[i] != 0 || my_errno != HA_ERR_KEY_NOT_FOUND)))
     {
       printf("key: '%.*s'  maria_rkey: %3d  errno: %3d  record: %s\n",
-	     (int) key_length,key+test(null_fields),error,my_errno,record+1);
+	     (int) key_length,key+offset_to_key,error,my_errno,record+1);
     }
   }
 
@@ -661,7 +680,7 @@ static struct my_option my_long_options[] =
    "Abort hard after doing inserts. Used for testing recovery with undo",
    (uchar**) &die_in_middle_of_transaction,
    (uchar**) &die_in_middle_of_transaction,
-   0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+   0, GET_INT, OPT_ARG, 0, 0, 0, 0, 0, 0},
   {"transactional", 'T',
    "Test in transactional mode. (Only works with block format)",
    (uchar**) &transactional, (uchar**) &transactional, 0, GET_BOOL, NO_ARG,
@@ -749,6 +768,12 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
   case 'K':                                     /* Use key cacheing */
     pagecacheing=1;
     break;
+  case 'A':
+    if (!argument)
+      die_in_middle_of_transaction= 1;
+    else
+      die_in_middle_of_transaction= atoi(argument);
+    break;
   case 'V':
     printf("test1 Ver 1.2 \n");
     exit(0);

storage/maria/maria_def.h

@@ -373,6 +373,7 @@ typedef struct st_maria_row
   uint field_lengths_length;            /* Length of data in field_lengths */
   uint extents_count;                   /* number of extents in 'extents' */
   uint full_page_count, tail_count;     /* For maria_chk */
+  uint space_on_head_page;
 } MARIA_ROW;
 
 /* Data to scan row in blocked format */
@@ -434,6 +435,8 @@ struct st_maria_info
   ulong packed_length, blob_length;	/* Length of found, packed record */
   size_t rec_buff_size;
   PAGECACHE_FILE dfile;			/* The datafile */
+  IO_CACHE rec_cache;			/* When cacheing records */
+  LIST open_list;
   uint opt_flag;			/* Optim. for space/speed */
   uint update;				/* If file changed since open */
   int lastinx;				/* Last used index */
@@ -449,8 +452,6 @@ struct st_maria_info
   uint data_changed;			/* Somebody has changed data */
   uint save_update;			/* When using KEY_READ */
   int save_lastinx;
-  LIST open_list;
-  IO_CACHE rec_cache;			/* When cacheing records */
   uint preload_buff_size;		/* When preloading indexes */
   myf lock_wait;			/* is 0 or MY_DONT_WAIT */
   my_bool was_locked;			/* Was locked in panic */
@@ -468,6 +469,7 @@ struct st_maria_info
   THR_LOCK_DATA lock;
 #endif
   uchar *maria_rtree_recursion_state;	/* For RTREE */
+  uchar length_buff[5];			/* temp buff to store blob lengths */
   int maria_rtree_recursion_depth;
 };