Fixed repair_by_sort to work with BLOCK_RECORD

Fixed bugs in undo logging
Fixed bug where head block was split before min_row_length (caused Maria to believe row was crashed on read)
Reserved place for reference-transid on key pages (for packing of transids)
ALTER TABLE and INSERT ... SELECT now uses fast creation of index
    
Known bugs:
ma_test_recovery fails because of a bug in redo handling when log is cut directly after a redo (Guilhem knows how to fix)
ma_test_recovery.excepted is not totally correct, because of the above bug
mysqld sometimes fails to restart; Fails with error "end_of_redo_phase: Assertion `long_trid != 0' failed"; Guilhem to investigate


include/maria.h:
  Prototype changes
  Added current_filepos to st_maria_sort_info
mysql-test/r/maria.result:
  Updated results that changes as alter table and insert ... select now uses fast creation of index
mysys/mf_iocache.c:
  Reset variable to gurard against double invocation
storage/maria/ma_bitmap.c:
  Added _ma_bitmap_reset_cache() (needed for repair)
storage/maria/ma_blockrec.c:
  Simplify code
  More initial allocations
  Fixed bug where head block was split before min_row_length (caused Maria to believe row was crashed on read)
storage/maria/ma_blockrec.h:
  Moved TRANSID_SIZE to maria_def.h
  Added prototype for new functions
storage/maria/ma_check.c:
  Simplicy code
  Fixed repair_by_sort to work with BLOCK_RECORD
  - When using BLOCK_RECORD or UNPACK create new Maria handle
  - Use common initializer function
  - Align code with maria_repair()
  
  Made some changes to maria_repair_parallel() to use common initializer function
  Removed ASK_MONTY section by fixing noted problem
storage/maria/ma_close.c:
  Moved check for readonly to _ma_state_info_write()
storage/maria/ma_key_recover.c:
  Use different log entries if key root changes or not.
  This fixed some bugs when tree grows
storage/maria/ma_key_recover.h:
  Added keynr to st_msg_to_write_hook_for_undo_key
storage/maria/ma_loghandler.c:
  Added INIT_LOGREC_UNDO_KEY_INSERT_WITH_ROOT
storage/maria/ma_loghandler.h:
  Added INIT_LOGREC_UNDO_KEY_INSERT_WITH_ROOT
storage/maria/ma_open.c:
  Added TRANSID to all key pages (for future compressing of trans id's)
  For compressed records, alloc a bit bigger buffer to avoid valgrind warnings
  If table is opened readonly, don't update state
storage/maria/ma_packrec.c:
  Allocate bigger array for bit unpacking to avoid valgrind errors
storage/maria/ma_recovery.c:
  Added UNDO_KEY_INSERT_WITH_ROOT & UNDO_KEY_DELETE_WITH_ROOT
storage/maria/ma_sort.c:
  More logging
storage/maria/ma_test_all.sh:
  More tests
storage/maria/ma_test_recovery.expected:
  Update results
  Note that this is not complete becasue of a bug in recovery
storage/maria/ma_test_recovery:
  Removed recreation of index (not needed when we have redo for index pages)
storage/maria/maria_chk.c:
  When using flag --read-only, don't update status for files
  When using --unpack, don't use REPAIR_BY_SORT if other repair option is given
  Enable repair_by_sort for BLOCK records
  Removed not needed newline at start of --describe
storage/maria/maria_def.h:
  Support for TRANSID_SIZE to key pages
storage/maria/maria_read_log.c:
  renamed --only-display to --display-only

include/maria.h

@@ -370,7 +370,7 @@ typedef struct st_maria_sort_param
   ulonglong unique[HA_MAX_KEY_SEG+1];
   ulonglong notnull[HA_MAX_KEY_SEG+1];
 
-  MARIA_RECORD_POS pos,max_pos,filepos,start_recpos;
+  MARIA_RECORD_POS pos,max_pos,filepos,start_recpos, current_filepos;
   uint key, key_length,real_key_length,sortbuff_size;
   uint maxbuffers, keys, find_length, sort_keys_length;
   my_bool fix_datafile, master;
@@ -392,18 +392,16 @@ typedef struct st_maria_sort_param
 /* functions in maria_check */
 void maria_chk_init(HA_CHECK *param);
 int maria_chk_status(HA_CHECK *param, MARIA_HA *info);
-int maria_chk_del(HA_CHECK *param, register MARIA_HA *info, uint test_flag);
+int maria_chk_del(HA_CHECK *param, MARIA_HA *info, uint test_flag);
 int maria_chk_size(HA_CHECK *param, MARIA_HA *info);
 int maria_chk_key(HA_CHECK *param, MARIA_HA *info);
 int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, int extend);
-int maria_repair(HA_CHECK *param, register MARIA_HA *info,
-		 char * name, int rep_quick);
-int maria_sort_index(HA_CHECK *param, register MARIA_HA *info,
-		     char * name);
-int maria_repair_by_sort(HA_CHECK *param, register MARIA_HA *info,
-			 const char *name, int rep_quick);
+int maria_repair(HA_CHECK *param, MARIA_HA *info, char * name, uint rep_quick);
+int maria_sort_index(HA_CHECK *param, MARIA_HA *info, char * name);
+int maria_repair_by_sort(HA_CHECK *param, MARIA_HA *info,
+			 const char *name, uint rep_quick);
 int maria_repair_parallel(HA_CHECK *param, register MARIA_HA *info,
-			  const char *name, int rep_quick);
+			  const char *name, uint rep_quick);
 int maria_change_to_newfile(const char *filename, const char *old_ext,
                             const char *new_ext, myf myflags);
 void maria_lock_memory(HA_CHECK *param);

mysql-test/r/maria.result

@@ -625,7 +625,7 @@ t1	1	a	1	a	A	NULL	NULL	NULL	YES	BTREE	disabled
 alter table t1 enable keys;
 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	NULL	NULL	NULL	YES	BTREE	
+t1	1	a	1	a	A	1000	NULL	NULL	YES	BTREE	
 alter table t1 engine=heap;
 alter table t1 disable keys;
 Warnings:
@@ -1957,12 +1957,12 @@ create table t2 like t1;
 insert into t2 select * from t1;
 analyze table t2;
 Table	Op	Msg_type	Msg_text
-test.t2	analyze	status	OK
+test.t2	analyze	status	Table is already up to date
 delete from t2;
 insert into t2 select * from t1;
 analyze table t2;
 Table	Op	Msg_type	Msg_text
-test.t2	analyze	status	OK
+test.t2	analyze	status	Table is already up to date
 drop table t1,t2;
 create table t1 (a bigint auto_increment, primary key(a), b char(255), c varchar(20000));
 update t1 set b=repeat('a',100) where a between 1 and 100;

mysys/mf_iocache.c

@@ -1834,6 +1834,7 @@ int end_io_cache(IO_CACHE *info)
     pthread_mutex_destroy(&info->append_buffer_lock);
 #endif
   }
+  info->share= 0;
   DBUG_RETURN(error);
 } /* end_io_cache */
 

storage/maria/ma_bitmap.c

@@ -124,9 +124,6 @@
 #include "maria_def.h"
 #include "ma_blockrec.h"
 
-/* Number of pages to store blob parts */
-#define BLOB_SEGMENT_MIN_SIZE 128
-
 #define FULL_HEAD_PAGE 4
 #define FULL_TAIL_PAGE 7
 
@@ -187,7 +184,6 @@ my_bool _ma_bitmap_init(MARIA_SHARE *share, File file)
     return 1;
 
   bitmap->file.file= file;
-  bitmap->changed= 0;
   bitmap->block_size= share->block_size;
   /* Size needs to be alligned on 6 */
   aligned_bit_blocks= (share->block_size - PAGE_SUFFIX_SIZE) / 6;
@@ -212,18 +208,8 @@ my_bool _ma_bitmap_init(MARIA_SHARE *share, File file)
 
   pthread_mutex_init(&share->bitmap.bitmap_lock, MY_MUTEX_INIT_SLOW);
 
-  /*
-    We can't read a page yet, as in some case we don't have an active
-    page cache yet.
-    Pretend we have a dummy, full and not changed bitmap page in memory.
- */
+  _ma_bitmap_reset_cache(share);
 
-  bitmap->page= ~(ulonglong) 0;
-  bitmap->used_size= bitmap->total_size;
-  bfill(bitmap->map, share->block_size, 255);
-#ifndef DBUG_OFF
-  memcpy(bitmap->map + bitmap->block_size, bitmap->map, bitmap->block_size);
-#endif
   if (share->state.first_bitmap_with_space == ~(ulonglong) 0)
   {
     /* Start scanning for free space from start of file */
@@ -313,6 +299,41 @@ void _ma_bitmap_delete_all(MARIA_SHARE *share)
 }
 
 
+/**
+   @brief Reset bitmap caches
+
+   @fn    _ma_bitmap_reset_cache()
+   @param share		Maria share
+
+   @notes
+   This is called after we have swapped file descriptors and we want
+   bitmap to forget all cached information
+*/
+
+void _ma_bitmap_reset_cache(MARIA_SHARE *share)
+{
+  MARIA_FILE_BITMAP *bitmap= &share->bitmap;
+
+  if (bitmap->map)                              /* If using bitmap */
+  {
+    /* Forget changes in current bitmap page */
+    bitmap->changed= 0;
+
+    /*
+      We can't read a page yet, as in some case we don't have an active
+      page cache yet.
+      Pretend we have a dummy, full and not changed bitmap page in memory.
+    */
+    bitmap->page= ~(ulonglong) 0;
+    bitmap->used_size= bitmap->total_size;
+    bfill(bitmap->map, share->block_size, 255);
+#ifndef DBUG_OFF
+    memcpy(bitmap->map + bitmap->block_size, bitmap->map, bitmap->block_size);
+#endif
+  }
+}
+
+
 /*
   Return bitmap pattern for the smallest head block that can hold 'size'
 
@@ -1630,7 +1651,7 @@ my_bool _ma_bitmap_find_new_place(MARIA_HA *info, MARIA_ROW *row,
 {
   MARIA_SHARE *share= info->s;
   my_bool res= 1;
-  uint full_page_size, position;
+  uint position;
   uint head_length, row_length, rest_length, extents_length;
   ulonglong bitmap_page;
   DBUG_ENTER("_ma_bitmap_find_new_place");
@@ -1670,9 +1691,8 @@ my_bool _ma_bitmap_find_new_place(MARIA_HA *info, MARIA_ROW *row,
   /* The first segment size is stored in 'row_length' */
   row_length= find_where_to_split_row(share, row, extents_length, free_size);
 
-  full_page_size= FULL_PAGE_SIZE(share->block_size);
   position= 0;
-  if (head_length - row_length <= full_page_size)
+  if (head_length - row_length < MAX_TAIL_SIZE(share->block_size))
     position= ELEMENTS_RESERVED_FOR_MAIN_PART -2;    /* Only head and tail */
   use_head(info, page, row_length, position);
   rest_length= head_length - row_length;

storage/maria/ma_blockrec.c

@@ -456,6 +456,7 @@ my_bool _ma_once_end_block_record(MARIA_SHARE *share)
 my_bool _ma_init_block_record(MARIA_HA *info)
 {
   MARIA_ROW *row= &info->cur_row, *new_row= &info->new_row;
+  uint default_extents;
   DBUG_ENTER("_ma_init_block_record");
 
   if (!my_multi_malloc(MY_WME,
@@ -490,17 +491,27 @@ my_bool _ma_init_block_record(MARIA_HA *info)
   /* Skip over bytes used to store length of field length for logging */
   row->field_lengths+= 2;
   new_row->field_lengths+= 2;
+
+  /* Reserve some initial space to avoid mallocs during execution */
+  default_extents= (ELEMENTS_RESERVED_FOR_MAIN_PART + 1 +
+                    (AVERAGE_BLOB_SIZE /
+                     FULL_PAGE_SIZE(info->s->block_size) /
+                     BLOB_SEGMENT_MIN_SIZE));
+
   if (my_init_dynamic_array(&info->bitmap_blocks,
-                            sizeof(MARIA_BITMAP_BLOCK),
-                            ELEMENTS_RESERVED_FOR_MAIN_PART, 16))
+                            sizeof(MARIA_BITMAP_BLOCK), default_extents,
+                            64))
     goto err;
+  if (!(info->cur_row.extents= my_malloc(default_extents * ROW_EXTENT_SIZE,
+                                         MYF(MY_WME))))
+    goto err;
+
   row->base_length= new_row->base_length= info->s->base_length;
 
   /*
     We need to reserve 'EXTRA_LENGTH_FIELDS' number of parts in
     null_field_lengths to allow splitting of rows in 'find_where_to_split_row'
   */
-
   row->null_field_lengths+= EXTRA_LENGTH_FIELDS;
   new_row->null_field_lengths+= EXTRA_LENGTH_FIELDS;
 
@@ -3697,7 +3708,7 @@ int _ma_read_block_record2(MARIA_HA *info, uchar *record,
        column < end_column; column++)
   {
     uint column_length= column->length;
-    if (data >= end_of_data &&
+    if (data + column_length > end_of_data &&
         !(data= read_next_extent(info, &extent, &end_of_data)))
       goto err;
     memcpy(record + column->offset, data, column_length);
@@ -3731,7 +3742,7 @@ int _ma_read_block_record2(MARIA_HA *info, uchar *record,
     case FIELD_NORMAL:                          /* Fixed length field */
     case FIELD_SKIP_PRESPACE:
     case FIELD_SKIP_ZERO:                       /* Fixed length field */
-      if (data >= end_of_data &&
+      if (data + column->length > end_of_data &&
           !(data= read_next_extent(info, &extent, &end_of_data)))
         goto err;
       memcpy(field_pos, data, column->length);
@@ -4991,6 +5002,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
   MARIA_PINNED_PAGE page_link;
   enum pagecache_page_lock unlock_method;
   enum pagecache_page_pin unpin_method;
+  my_off_t end_of_page;
   DBUG_ENTER("_ma_apply_redo_insert_row_head_or_tail");
 
   page=  page_korr(header);
@@ -5000,8 +5012,12 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
                        (ulong) ma_recordpos(page, rownr),
                        (ulong) page, rownr, (uint) data_length));
 
-  if (((page + 1) * info->s->block_size) > info->state->data_file_length)
+  end_of_page= (page + 1) * info->s->block_size;
+  if (end_of_page > info->state->data_file_length)
   {
+    DBUG_PRINT("info", ("Enlarging data file from %lu to %lu",
+                        (ulong) info->state->data_file_length,
+                        (ulong) end_of_page));
     /*
       New page at end of file. Note that the test above is also positive if
       data_file_length is not a multiple of block_size (system crashed while
@@ -5153,11 +5169,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
     case we extended the file. We could not do it earlier: bitmap code tests
     data_file_length to know if it has to create a new page or not.
   */
-  {
-    my_off_t end_of_page= (page + 1) * info->s->block_size;
-    set_if_bigger(info->state->data_file_length, end_of_page);
-  }
-
+  set_if_bigger(info->state->data_file_length, end_of_page);
   DBUG_RETURN(result);
 
 err:

storage/maria/ma_blockrec.h

@@ -29,8 +29,8 @@
                                  PAGE_SUFFIX_SIZE)
 #define BLOCK_RECORD_POINTER_SIZE	6
 
-#define FULL_PAGE_SIZE(block_size) ((block_size) - LSN_SIZE - PAGE_TYPE_SIZE - \
-                                    PAGE_SUFFIX_SIZE)
+#define FULL_PAGE_SIZE(block_size) ((block_size) - LSN_SIZE - \
+                                    PAGE_TYPE_SIZE - PAGE_SUFFIX_SIZE)
 
 #define ROW_EXTENT_PAGE_SIZE	5
 #define ROW_EXTENT_COUNT_SIZE   2
@@ -40,12 +40,16 @@
 #define TAIL_BIT		0x8000	/* Bit in page_count to signify tail */
 /* Number of extents reserved MARIA_BITMAP_BLOCKS to store head part */
 #define ELEMENTS_RESERVED_FOR_MAIN_PART 4
+/* This is just used to prealloc a dynamic array */
+#define AVERAGE_BLOB_SIZE      1024L*1024L
+/* Number of pages to store continuous blob parts */
+#define BLOB_SEGMENT_MIN_SIZE 128
+
 /* Fields before 'row->null_field_lengths' used by find_where_to_split_row */
 #define EXTRA_LENGTH_FIELDS		3
 
 /* Size for the different parts in the row header (and head page) */
 #define FLAG_SIZE		1
-#define TRANSID_SIZE		6
 #define VERPTR_SIZE		7
 #define DIR_ENTRY_SIZE		4
 #define FIELD_OFFSET_SIZE	2      /* size of pointers to field starts */
@@ -167,6 +171,7 @@ my_bool _ma_compare_block_record(register MARIA_HA *info,
 my_bool _ma_bitmap_init(MARIA_SHARE *share, File file);
 my_bool _ma_bitmap_end(MARIA_SHARE *share);
 my_bool _ma_flush_bitmap(MARIA_SHARE *share);
+void _ma_bitmap_reset_cache(MARIA_SHARE *share);
 my_bool _ma_bitmap_find_place(MARIA_HA *info, MARIA_ROW *row,
                               MARIA_BITMAP_BLOCKS *result_blocks);
 my_bool _ma_bitmap_release_unused(MARIA_HA *info, MARIA_BITMAP_BLOCKS *blocks);

storage/maria/ma_close.c

@@ -86,8 +86,7 @@ int maria_close(register MARIA_HA *info)
         may be using the file at this point
         IF using --external-locking, which does not apply to Maria.
       */
-      if (share->mode != O_RDONLY &&
-          ((share->changed && share->base.born_transactional) ||
+      if (((share->changed && share->base.born_transactional) ||
            maria_is_crashed(info)))
       {
         /*

storage/maria/ma_key_recover.c

@@ -73,7 +73,8 @@ my_bool _ma_write_clr(MARIA_HA *info, LSN undo_lsn,
                       LSN *res_lsn, void *extra_msg)
 {
   uchar log_data[LSN_STORE_SIZE + FILEID_STORE_SIZE + CLR_TYPE_STORE_SIZE +
-                 HA_CHECKSUM_STORE_SIZE];
+                 HA_CHECKSUM_STORE_SIZE+ KEY_NR_STORE_SIZE + PAGE_STORE_SIZE];
+  uchar *log_pos;
   LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
   struct st_msg_to_write_hook_for_clr_end msg;
   my_bool res;
@@ -81,10 +82,8 @@ my_bool _ma_write_clr(MARIA_HA *info, LSN undo_lsn,
 
   /* undo_lsn must be first for compression to work */
   lsn_store(log_data, undo_lsn);
-  clr_type_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE,
-                 undo_type);
-  log_array[TRANSLOG_INTERNAL_PARTS + 0].length=
-    sizeof(log_data) - HA_CHECKSUM_STORE_SIZE;
+  clr_type_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE, undo_type);
+  log_pos= log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE + CLR_TYPE_STORE_SIZE;
 
   /* Extra_msg is handled in write_hook_for_clr_end() */
   msg.undone_record_type= undo_type;
@@ -95,11 +94,24 @@ my_bool _ma_write_clr(MARIA_HA *info, LSN undo_lsn,
   if (store_checksum)
   {
     msg.checksum_delta= checksum;
-    ha_checksum_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE +
-                      CLR_TYPE_STORE_SIZE, checksum);
-    log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
+    ha_checksum_store(log_pos, checksum);
+    log_pos+= HA_CHECKSUM_STORE_SIZE;
   }
+  else if (undo_type == LOGREC_UNDO_KEY_INSERT_WITH_ROOT ||
+           undo_type == LOGREC_UNDO_KEY_DELETE_WITH_ROOT)
+  {
+    /* Key root changed. Store new key root */
+    struct st_msg_to_write_hook_for_undo_key *undo_msg= extra_msg;
+    ulonglong page;
+    key_nr_store(log_pos, undo_msg->keynr);
+    page= (undo_msg->value == HA_OFFSET_ERROR ? IMPOSSIBLE_PAGE_NO :
+           undo_msg->value / info->s->block_size);
+    page_store(log_pos + KEY_NR_STORE_SIZE, page);
+    log_pos+= KEY_NR_STORE_SIZE + PAGE_STORE_SIZE;
+  }
+
   log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
+  log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - log_data);
 
   res= translog_write_record(res_lsn, LOGREC_CLR_END,
                              info->trn, info, log_array[TRANSLOG_INTERNAL_PARTS
@@ -141,8 +153,8 @@ my_bool write_hook_for_clr_end(enum translog_record_type type
   case LOGREC_UNDO_ROW_UPDATE:
     share->state.state.checksum+= msg->checksum_delta;
     break;
-  case LOGREC_UNDO_KEY_INSERT:
-  case LOGREC_UNDO_KEY_DELETE:
+  case LOGREC_UNDO_KEY_INSERT_WITH_ROOT:
+  case LOGREC_UNDO_KEY_DELETE_WITH_ROOT:
   {
     /* Update key root */
     struct st_msg_to_write_hook_for_undo_key *extra_msg=
@@ -150,6 +162,9 @@ my_bool write_hook_for_clr_end(enum translog_record_type type
     *extra_msg->root= extra_msg->value;
     break;
   }
+  case LOGREC_UNDO_KEY_INSERT:
+  case LOGREC_UNDO_KEY_DELETE:
+    break;
   default:
     DBUG_ASSERT(0);
   }
@@ -812,9 +827,11 @@ my_bool _ma_apply_undo_key_insert(MARIA_HA *info, LSN undo_lsn,
 
   msg.root= &share->state.key_root[keynr];
   msg.value= new_root;
+  msg.keynr= keynr;
 
-  if (_ma_write_clr(info, undo_lsn, LOGREC_UNDO_KEY_INSERT, 1, 0, &lsn,
-                    (void*) &msg))
+  if (_ma_write_clr(info, undo_lsn, *msg.root == msg.value ?
+                    LOGREC_UNDO_KEY_INSERT : LOGREC_UNDO_KEY_INSERT_WITH_ROOT,
+                    0, 0, &lsn, (void*) &msg))
     res= 1;
 
   _ma_unpin_all_pages_and_finalize_row(info, lsn);
@@ -855,7 +872,11 @@ my_bool _ma_apply_undo_key_delete(MARIA_HA *info, LSN undo_lsn,
 
   msg.root= &share->state.key_root[keynr];
   msg.value= new_root;
-  if (_ma_write_clr(info, undo_lsn, LOGREC_UNDO_KEY_DELETE, 1, 0, &lsn,
+  msg.keynr= keynr;
+  if (_ma_write_clr(info, undo_lsn,
+                    *msg.root == msg.value ?
+                    LOGREC_UNDO_KEY_DELETE : LOGREC_UNDO_KEY_DELETE_WITH_ROOT,
+                    0, 0, &lsn,
                     (void*) &msg))
     res= 1;
 

storage/maria/ma_key_recover.h

@@ -35,6 +35,7 @@ struct st_msg_to_write_hook_for_undo_key
 {
   my_off_t *root;
   my_off_t value;
+  uint keynr;
 };
 
 

storage/maria/ma_loghandler.c

@@ -490,6 +490,13 @@ static LOG_DESC INIT_LOGREC_UNDO_KEY_INSERT=
  NULL, write_hook_for_undo_key, NULL, 1,
  "undo_key_insert", LOGREC_LAST_IN_GROUP, NULL, NULL};
 
+/* This will never be in the log, only in the clr */
+static LOG_DESC INIT_LOGREC_UNDO_KEY_INSERT_WITH_ROOT=
+{LOGRECTYPE_VARIABLE_LENGTH, 0,
+ LSN_STORE_SIZE + FILEID_STORE_SIZE + KEY_NR_STORE_SIZE + PAGE_STORE_SIZE,
+ NULL, write_hook_for_undo_key, NULL, 1,
+ "undo_key_insert_with_root", LOGREC_LAST_IN_GROUP, NULL, NULL};
+
 static LOG_DESC INIT_LOGREC_UNDO_KEY_DELETE=
 {LOGRECTYPE_VARIABLE_LENGTH, 0,
  LSN_STORE_SIZE + FILEID_STORE_SIZE + KEY_NR_STORE_SIZE,
@@ -605,6 +612,8 @@ static void loghandler_init()
     INIT_LOGREC_UNDO_ROW_UPDATE;
   log_record_type_descriptor[LOGREC_UNDO_KEY_INSERT]=
     INIT_LOGREC_UNDO_KEY_INSERT;
+  log_record_type_descriptor[LOGREC_UNDO_KEY_INSERT_WITH_ROOT]=
+    INIT_LOGREC_UNDO_KEY_INSERT_WITH_ROOT;
   log_record_type_descriptor[LOGREC_UNDO_KEY_DELETE]=
     INIT_LOGREC_UNDO_KEY_DELETE;
   log_record_type_descriptor[LOGREC_UNDO_KEY_DELETE_WITH_ROOT]=

storage/maria/ma_loghandler.h

@@ -123,6 +123,7 @@ enum translog_record_type
   LOGREC_UNDO_ROW_DELETE,
   LOGREC_UNDO_ROW_UPDATE,
   LOGREC_UNDO_KEY_INSERT,
+  LOGREC_UNDO_KEY_INSERT_WITH_ROOT,
   LOGREC_UNDO_KEY_DELETE,
   LOGREC_UNDO_KEY_DELETE_WITH_ROOT,
   LOGREC_PREPARE,

storage/maria/ma_open.c

@@ -613,7 +613,8 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
                          share->base.null_bytes +
                          share->base.pack_bytes +
                          test(share->options & HA_OPTION_CHECKSUM));
-    share->keypage_header= ((share->base.born_transactional ? LSN_STORE_SIZE :
+    share->keypage_header= ((share->base.born_transactional ?
+                             LSN_STORE_SIZE + TRANSID_SIZE :
                              0) + KEYPAGE_KEYID_SIZE + KEYPAGE_FLAG_SIZE +
                             KEYPAGE_USED_SIZE);
 
@@ -672,6 +673,11 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
          MARIA_REC_BUFF_OFFSET);
       share->base.default_rec_buff_size+= share->base.extra_rec_buff_size;
     }
+    if (share->data_file_type == COMPRESSED_RECORD)
+    {
+      /* Need some extra bytes for decode_bytes */
+      share->base.extra_rec_buff_size= 7;
+    }
     disk_pos_assert(disk_pos + share->base.fields *MARIA_COLUMNDEF_SIZE,
                     end_pos);
     for (i= j= 0 ; i < share->base.fields ; i++)
@@ -1042,6 +1048,9 @@ static void setup_key_functions(register MARIA_KEYDEF *keyinfo)
 uint _ma_state_info_write(MARIA_SHARE *share, uint pWrite)
 {
   uint res;
+  if (share->options & HA_OPTION_READ_ONLY_DATA)
+    return 0;
+
   if (pWrite & 4)
     pthread_mutex_lock(&share->intern_lock);
   else if (maria_multi_threaded)
@@ -1091,7 +1100,7 @@ uint _ma_state_info_write_sub(File file, MARIA_STATE_INFO *state, uint pWrite)
   uchar *ptr=buff;
   uint	i, keys= (uint) state->header.keys;
   size_t res;
-  DBUG_ENTER("_ma_state_info_write");
+  DBUG_ENTER("_ma_state_info_write_sub");
 
   memcpy_fixed(ptr,&state->header,sizeof(state->header));
   ptr+=sizeof(state->header);

storage/maria/ma_packrec.c

@@ -44,7 +44,9 @@
   { bits-=(bit+1); break; } \
   pos+= *pos
 
-/* Size in uint16 of a Huffman tree for uchar compression of 256 uchar values. */
+/*
+  Size in uint16 of a Huffman tree for uchar compression of 256 uchar values
+*/
 #define OFFSET_TABLE_SIZE 512
 
 static my_bool _ma_read_pack_info(MARIA_SHARE *share, File file,
@@ -245,7 +247,8 @@ static my_bool _ma_read_pack_info(MARIA_SHARE *share, File file,
   length=(uint) (elements*2+trees*(1 << maria_quick_table_bits));
   if (!(share->decode_tables=(uint16*)
 	my_malloc((length+OFFSET_TABLE_SIZE)*sizeof(uint16)+
-		  (uint) (share->pack.header_length - sizeof(header)),
+		  (uint) (share->pack.header_length - sizeof(header)) +
+                  share->base.extra_rec_buff_size,
 		  MYF(MY_WME | MY_ZEROFILL))))
     goto err1;
   tmp_buff=share->decode_tables+length;
@@ -255,6 +258,11 @@ static my_bool _ma_read_pack_info(MARIA_SHARE *share, File file,
 	      (uint) (share->pack.header_length-sizeof(header)),
 	      MYF(MY_NABP)))
     goto err2;
+#ifdef HAVE_purify
+  /* Zero bytes accessed by fill_buffer */
+  bzero(disk_cache + (share->pack.header_length-sizeof(header)),
+        share->base.extra_rec_buff_size);
+#endif
 
   huff_tree_bits=max_bit(trees ? trees-1 : 0);
   init_bit_buffer(&bit_buff, disk_cache,

storage/maria/ma_recovery.c

@@ -1147,7 +1147,7 @@ static int new_table(uint16 sid, const char *name,
     tprint(tracef, ", has wrong state.key_file_length (fixing it)");
     share->state.state.key_file_length= kfile_len;
   }
-  if ((dfile_len % share->block_size) > 0)
+  if ((dfile_len % share->block_size) || (kfile_len % share->block_size))
   {
     tprint(tracef, ", has too short last page\n");
     /* Recovery will fix this, no error */
@@ -1669,9 +1669,10 @@ prototype_redo_exec_hook(CLR_END)
   enum translog_record_type undone_record_type;
   const LOG_DESC *log_desc;
   my_bool row_entry= 0;
+  DBUG_ENTER("exec_REDO_LOGREC_CLR_END");
 
   if (info == NULL)
-    return 0;
+    DBUG_RETURN(0);
   share= info->s;
   previous_undo_lsn= lsn_korr(rec->header);
   undone_record_type=
@@ -1699,6 +1700,28 @@ prototype_redo_exec_hook(CLR_END)
     case LOGREC_UNDO_KEY_INSERT:
     case LOGREC_UNDO_KEY_DELETE:
       break;
+    case LOGREC_UNDO_KEY_INSERT_WITH_ROOT:
+    case LOGREC_UNDO_KEY_DELETE_WITH_ROOT:
+    {
+      uint key_nr;
+      my_off_t page;
+      uchar buff[KEY_NR_STORE_SIZE + PAGE_STORE_SIZE];
+      if (translog_read_record(rec->lsn, LSN_STORE_SIZE + FILEID_STORE_SIZE +
+                               CLR_TYPE_STORE_SIZE,
+                               KEY_NR_STORE_SIZE + PAGE_STORE_SIZE,
+                               buff, NULL) !=
+          KEY_NR_STORE_SIZE + PAGE_STORE_SIZE)
+      {
+        tprint(tracef, "Failed to read record\n");
+        DBUG_RETURN(1);
+      }
+      key_nr= key_nr_korr(buff);
+      page=  page_korr(buff + KEY_NR_STORE_SIZE);
+      share->state.key_root[key_nr]= (page == IMPOSSIBLE_PAGE_NO ?
+                                      HA_OFFSET_ERROR :
+                                      page * share->block_size);
+      break;
+    }
     default:
       DBUG_ASSERT(0);
     }
@@ -1710,7 +1733,7 @@ prototype_redo_exec_hook(CLR_END)
                                buff, NULL) != HA_CHECKSUM_STORE_SIZE)
       {
         tprint(tracef, "Failed to read record\n");
-        return 1;
+        DBUG_RETURN(1);
       }
       share->state.state.checksum+= ha_checksum_korr(buff);
     }
@@ -1719,7 +1742,7 @@ prototype_redo_exec_hook(CLR_END)
   if (row_entry)
     tprint(tracef, "   rows' count %lu\n", (ulong)share->state.state.records);
   _ma_unpin_all_pages(info, rec->lsn);
-  return 0;
+  DBUG_RETURN(0);
 }
 
 

storage/maria/ma_sort.c

@@ -110,7 +110,9 @@ int _ma_create_index_by_sort(MARIA_SORT_PARAM *info, my_bool no_messages,
   uchar **sort_keys;
   IO_CACHE tempfile, tempfile_for_exceptions;
   DBUG_ENTER("_ma_create_index_by_sort");
-  DBUG_PRINT("enter",("sort_length: %d", info->key_length));
+  DBUG_PRINT("enter",("sort_buff_size: %lu  sort_length: %d  max_records: %lu",
+                      (ulong) sortbuff_size, info->key_length,
+                      (ulong) info->sort_info->max_records));
 
   if (info->keyinfo->flag & HA_VAR_LENGTH_KEY)
   {

storage/maria/ma_test_all.sh

@@ -169,6 +169,11 @@ run_pack_tests()
   $maria_path/maria_chk$suffix -rus test1
   $maria_path/maria_chk$suffix -es test1
   
+  $maria_path/ma_test1$suffix $silent --checksum $row_type
+  $maria_path/maria_pack$suffix --force -s test1
+  $maria_path/maria_chk$suffix -rus --safe-recover test1
+  $maria_path/maria_chk$suffix -es test1
+
   $maria_path/ma_test1$suffix $silent --checksum -S $row_type
   $maria_path/maria_chk$suffix -se test1
   $maria_path/maria_chk$suffix -ros test1
@@ -184,10 +189,10 @@ run_pack_tests()
   $maria_path/ma_test2$suffix $silent -c -d1 $row_type
   $maria_path/maria_chk$suffix -s --parallel-recover test2
   $maria_path/maria_chk$suffix -se test2
-  $maria_path/maria_chk$suffix -s --parallel-recover --unpack test2
+  $maria_path/maria_chk$suffix -s --unpack --parallel-recover test2
   $maria_path/maria_chk$suffix -se test2
   $maria_path/maria_pack$suffix --force -s test1
-  $maria_path/maria_chk$suffix -s --parallel-recover --unpack test2
+  $maria_path/maria_chk$suffix -s --unpack --parallel-recover test2
   $maria_path/maria_chk$suffix -se test2
 }
 
@@ -233,6 +238,30 @@ $maria_path/maria_chk$suffix -ssm test2
 #
 /bin/sh $maria_path/ma_test_recovery
 
+#
+# Extra tests that has caused failures in the past
+#
+
+# Problem with re-executing CLR's
+rm -f maria_log.* maria_log_control
+ma_test2 -s -L -K -W -P -M -T -c -b -t2 -u1
+cp maria_log_control tmp
+maria_read_log -a -s
+maria_chk -s -e test2
+cp tmp/maria_log_control .
+rm test2.MA?
+maria_read_log -a -s
+maria_chk -s -e test2
+
+# Problem with re-executing CLR's
+rm -f maria_log.* maria_log_control
+ma_test2 -s -L -K -W -P -M -T -c -b -t2 -u1
+maria_read_log -a -s
+maria_chk -s -e test2
+rm test2.MA?
+maria_read_log -a -s
+maria_chk -e -s test2
+
 #
 # Some timing tests
 #

storage/maria/ma_test_recovery

@@ -24,14 +24,7 @@ check_table_is_same()
 
     $maria_path/maria_chk -dvv $table | grep -v "Creation time:" > $tmp/maria_chk_message.txt 2>&1
 
-    # save the index file (because we want to test idempotency afterwards)
-    cp $table.MAI tmp/
-    # In the repair below it's good to use -q because it will die on any
-    # incorrectness of the data file if UNDO was badly applied.
-    # QQ: Remove the following line when we also can recover the index file
-    $maria_path/maria_chk -s -rq $table
-
-    $maria_path/maria_chk -s -e $table
+    $maria_path/maria_chk -s -e --read-only $table
     checksum2=`$maria_path/maria_chk -dss $table`
     if test "$checksum" != "$checksum2"
         then
@@ -47,7 +40,6 @@ check_table_is_same()
         cat $tmp/maria_chk_diff.txt
         echo "========DIFF END======="
     fi
-    mv tmp/$table.MAI .
 }
 
 apply_log()

storage/maria/maria_chk.c

@@ -135,7 +135,7 @@ int main(int argc, char **argv)
   {					/* Only if descript */
     char buff[22],buff2[22];
     if (!(check_param.testflag & T_SILENT) || check_param.testflag & T_INFO)
-      puts("\n---------\n");
+      puts("\n---------");
     printf("\nTotal of all %d MARIA-files:\nData records: %9s   Deleted blocks: %9s\n",check_param.total_files,llstr(check_param.total_records,buff),
 	   llstr(check_param.total_deleted,buff2));
   }
@@ -624,9 +624,13 @@ get_one_option(int optid,
     break;
   case 'u':
     if (argument == disabled_my_option)
-      check_param.testflag&= ~(T_UNPACK | T_REP_BY_SORT);
+      check_param.testflag&= ~T_UNPACK;
     else
-      check_param.testflag|= T_UNPACK | T_REP_BY_SORT;
+    {
+      check_param.testflag|= T_UNPACK;
+      if (!(check_param.testflag & T_REP_ANY))
+        check_param.testflag|= T_REP_BY_SORT;
+    }
     break;
   case 'v':				/* Verbose */
     if (argument == disabled_my_option)
@@ -802,13 +806,13 @@ static int maria_chk(HA_CHECK *param, char *filename)
   datafile=0;
   param->isam_file_name=filename;		/* For error messages */
   if (!(info=maria_open(filename,
-		     (param->testflag & (T_DESCRIPT | T_READONLY)) ?
-		     O_RDONLY : O_RDWR,
-		     HA_OPEN_FOR_REPAIR |
-		     ((param->testflag & T_WAIT_FOREVER) ?
-		      HA_OPEN_WAIT_IF_LOCKED :
-		      (param->testflag & T_DESCRIPT) ?
-		      HA_OPEN_IGNORE_IF_LOCKED : HA_OPEN_ABORT_IF_LOCKED))))
+                        (param->testflag & (T_DESCRIPT | T_READONLY)) ?
+                        O_RDONLY : O_RDWR,
+                        HA_OPEN_FOR_REPAIR |
+                        ((param->testflag & T_WAIT_FOREVER) ?
+                         HA_OPEN_WAIT_IF_LOCKED :
+                         (param->testflag & T_DESCRIPT) ?
+                         HA_OPEN_IGNORE_IF_LOCKED : HA_OPEN_ABORT_IF_LOCKED))))
   {
     /* Avoid twice printing of isam file name */
     param->error_printed=1;
@@ -852,7 +856,6 @@ static int maria_chk(HA_CHECK *param, char *filename)
     DBUG_RETURN(1);
   }
   share=info->s;
-  share->options&= ~HA_OPTION_READ_ONLY_DATA; /* We are modifing it */
   share->tot_locks-= share->r_locks;
   share->r_locks=0;
   maria_block_size= share->base.block_size;
@@ -871,10 +874,10 @@ static int maria_chk(HA_CHECK *param, char *filename)
       goto end2;
     }
     /* We can't do parallell repair with BLOCK_RECORD yet */
-    if (param->testflag & (T_REP_BY_SORT | T_REP_PARALLEL))
+    if (param->testflag & T_REP_PARALLEL)
     {
-      param->testflag&= ~(T_REP_BY_SORT | T_REP_PARALLEL);
-      param->testflag|= T_REP;
+      param->testflag&= ~T_REP_PARALLEL;
+      param->testflag|= T_REP_BY_SORT;
     }
   }
 
@@ -1255,7 +1258,7 @@ static void descript(HA_CHECK *param, register MARIA_HA *info, char *name)
     DBUG_VOID_RETURN;
   }
 
-  printf("\nMARIA file:          %s\n",name);
+  printf("MARIA file:          %s\n",name);
   printf("Record format:       %s\n", record_formats[share->data_file_type]);
   printf("Crashsafe:           %s\n",
          share->base.born_transactional ? "yes" : "no");

storage/maria/maria_def.h

@@ -118,7 +118,7 @@ typedef struct st_maria_state_info
 #define MARIA_STATE_KEYBLOCK_SIZE  8
 #define MARIA_STATE_KEYSEG_SIZE	12
 #define MARIA_STATE_EXTRA_SIZE (MARIA_MAX_KEY*MARIA_STATE_KEY_SIZE + MARIA_MAX_KEY*HA_MAX_KEY_SEG*MARIA_STATE_KEYSEG_SIZE)
-#define MARIA_KEYDEF_SIZE		(2+ 5*2)
+#define MARIA_KEYDEF_SIZE	(2+ 5*2)
 #define MARIA_UNIQUEDEF_SIZE	(2+1+1)
 #define HA_KEYSEG_SIZE		(6+ 2*2 + 4*2)
 #define MARIA_COLUMNDEF_SIZE	(6+2+2+2+2+2+1+1)
@@ -507,6 +507,7 @@ struct st_maria_handler
 
 #define USE_WHOLE_KEY	65535         /* Use whole key in _search() */
 #define F_EXTRA_LCK	-1
+#define TRANSID_SIZE		6
 
 /* bits in opt_flag */
 #define MEMMAP_USED	32
@@ -540,7 +541,8 @@ struct st_maria_handler
 #define KEYPAGE_FLAG_SIZE  1
 #define KEYPAGE_CHECKSUM_SIZE 4
 #define MAX_KEYPAGE_HEADER_SIZE (LSN_STORE_SIZE + KEYPAGE_USED_SIZE + \
-                                 KEYPAGE_KEYID_SIZE + KEYPAGE_FLAG_SIZE)
+                                 KEYPAGE_KEYID_SIZE + KEYPAGE_FLAG_SIZE + \
+                                 TRANSID_SIZE)
 
 #define _ma_get_page_used(info,x) \
   (((uint) mi_uint2korr(x + (info)->s->keypage_header - KEYPAGE_USED_SIZE)) & \
@@ -560,6 +562,11 @@ struct st_maria_handler
 }
 #define _ma_store_keynr(info, x, nr) x[(info)->s->keypage_header - KEYPAGE_KEYID_SIZE - KEYPAGE_FLAG_SIZE - KEYPAGE_USED_SIZE]= nr
 #define _ma_get_keynr(info, x) ((uchar) x[(info)->s->keypage_header - KEYPAGE_KEYID_SIZE - KEYPAGE_FLAG_SIZE - KEYPAGE_USED_SIZE])
+#define _ma_store_transid(buff, transid) \
+  int6store((buff) + LSN_STORE_SIZE, (transid))
+#define _ma_korr_transid(buff) \
+  uint6korr((buff) + LSN_STORE_SIZE)
+
 #define maria_mark_crashed(x) do{(x)->s->state.changed|= STATE_CRASHED; \
     DBUG_PRINT("error", ("Marked table crashed"));                      \
   }while(0)

storage/maria/maria_read_log.c

@@ -29,7 +29,7 @@ const char *default_dbug_option= "d:t:i:O,\\maria_read_log.trace";
 const char *default_dbug_option= "d:t:i:o,/tmp/maria_read_log.trace";
 #endif
 #endif /* DBUG_OFF */
-static my_bool opt_only_display, opt_apply, opt_apply_undo, opt_silent,
+static my_bool opt_display_only, opt_apply, opt_apply_undo, opt_silent,
   opt_check;
 static ulong opt_page_buffer_size;
 
@@ -84,8 +84,8 @@ int main(int argc, char **argv)
     goto err;
   }
 
-  if (opt_only_display)
-    printf("You are using --only-display, NOTHING will be written to disk\n");
+  if (opt_display_only)
+    printf("You are using --display-only, NOTHING will be written to disk\n");
 
   /* LSN could be also --start-from-lsn=# */
   lsn= translog_first_lsn_in_log();
@@ -138,7 +138,7 @@ static struct my_option my_long_options[] =
    (uchar **) &opt_apply, (uchar **) &opt_apply, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"check", 'c',
-   "if --only-display, check if record is fully readable (for debugging)",
+   "if --display-only, check if record is fully readable (for debugging)",
    (uchar **) &opt_check, (uchar **) &opt_check, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
 #ifndef DBUG_OFF
@@ -147,8 +147,8 @@ static struct my_option my_long_options[] =
 #endif
   {"help", '?', "Display this help and exit.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
-  {"only-display", 'o', "display brief info read from records' header",
-   (uchar **) &opt_only_display, (uchar **) &opt_only_display, 0, GET_BOOL,
+  {"display-only", 'o', "display brief info read from records' header",
+   (uchar **) &opt_display_only, (uchar **) &opt_display_only, 0, GET_BOOL,
    NO_ARG,0, 0, 0, 0, 0, 0},
   { "page_buffer_size", 'P', "",
     (uchar**) &opt_page_buffer_size, (uchar**) &opt_page_buffer_size, 0,
@@ -225,7 +225,7 @@ static void get_options(int *argc,char ***argv)
   if (!opt_apply)
     opt_apply_undo= FALSE;
 
-  if ((opt_only_display + opt_apply) != 1)
+  if ((opt_display_only + opt_apply) != 1)
   {
     usage();
     exit(1);