Merge bk-internal.mysql.com:/home/bk/mysql-maria

into  mysql.com:/home/my/mysql-maria

storage/maria/ma_blockrec.c

@@ -1777,7 +1777,9 @@ static my_bool write_tail(MARIA_HA *info,
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
     log_array[TRANSLOG_INTERNAL_PARTS + 1].str=    (char*) row_pos.data;
     log_array[TRANSLOG_INTERNAL_PARTS + 1].length= length;
-    if (translog_write_record(&lsn, LOGREC_REDO_INSERT_ROW_TAIL,
+    if (translog_write_record(&lsn,
+                              (block_is_read ? LOGREC_REDO_INSERT_ROW_TAIL :
+                               LOGREC_REDO_NEW_ROW_TAIL),
                               info->trn, info, sizeof(log_data) + length,
                               TRANSLOG_INTERNAL_PARTS + 2, log_array,
                               log_data, NULL))
@@ -2830,7 +2832,11 @@ static my_bool write_block_record(MARIA_HA *info,
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
     log_array[TRANSLOG_INTERNAL_PARTS + 1].str=    (char*) row_pos->data;
     log_array[TRANSLOG_INTERNAL_PARTS + 1].length= head_length;
-    if (translog_write_record(&lsn, LOGREC_REDO_INSERT_ROW_HEAD, info->trn,
+    if (translog_write_record(&lsn,
+                              head_block_is_read ?
+                              LOGREC_REDO_INSERT_ROW_HEAD :
+                              LOGREC_REDO_NEW_ROW_HEAD,
+                              info->trn,
                               info, sizeof(log_data) + head_length,
                               TRANSLOG_INTERNAL_PARTS + 2, log_array,
                               log_data, NULL))
@@ -5721,17 +5727,22 @@ my_bool write_hook_for_file_id(enum translog_record_type type
 ***************************************************************************/
 
 /*
-  Apply LOGREC_REDO_INSERT_ROW_HEAD & LOGREC_REDO_INSERT_ROW_TAIL
+  Apply changes to head and tail pages
 
   SYNOPSIS
     _ma_apply_redo_insert_row_head_or_tail()
     info		Maria handler
     lsn			LSN to put on page
     page_type		HEAD_PAGE or TAIL_PAGE
+    new_page		True if this is first entry on page
     header		Header (without FILEID)
     data		Data to be put on page
     data_length		Length of data
 
+  NOTE
+    Handles LOGREC_REDO_INSERT_ROW_HEAD, LOGREC_REDO_INSERT_ROW_TAIL
+    LOGREC_REDO_NEW_ROW_HEAD and LOGREC_REDO_NEW_ROW_TAIL
+
   RETURN
     0   ok
     #   Error number
@@ -5739,6 +5750,7 @@ my_bool write_hook_for_file_id(enum translog_record_type type
 
 uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
                                             uint page_type,
+                                            my_bool new_page,
                                             const uchar *header,
                                             const uchar *data,
                                             size_t data_length)
@@ -5783,8 +5795,8 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
     unlock_method= PAGECACHE_LOCK_WRITE;
     unpin_method=  PAGECACHE_PIN;
 
-    DBUG_ASSERT(rownr == 0);
-    if (rownr != 0)
+    DBUG_ASSERT(rownr == 0 && new_page);
+    if (rownr != 0 || !new_page)
       goto crashed_file;
 
     buff= info->keyread_buff;
@@ -5808,8 +5820,8 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
     if (!buff)
     {
       /* Skip errors when reading outside of file and uninitialized pages */
-      if (my_errno != HA_ERR_FILE_TOO_SHORT &&
-          my_errno != HA_ERR_WRONG_CRC)
+      if (!new_page || (my_errno != HA_ERR_FILE_TOO_SHORT &&
+                        my_errno != HA_ERR_WRONG_CRC))
         goto err;
       /* Create new page */
       buff= pagecache_block_link_to_buffer(page_link.link);
@@ -5834,8 +5846,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
         This is a page that has been freed before and now should be
         changed to new type.
       */
-      if (((buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK) != BLOB_PAGE &&
-           (buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK) != UNALLOCATED_PAGE))
+      if (!new_page)
         goto crashed_file;
       make_empty_page(info, buff, page_type, 0);
       empty_space= block_size - PAGE_HEADER_SIZE - PAGE_SUFFIX_SIZE;
@@ -5850,6 +5861,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
       uint max_entry= (uint) buff[DIR_COUNT_OFFSET];
       uint length;
 
+      DBUG_ASSERT(!new_page);
       dir= dir_entry_pos(buff, block_size, rownr);
       empty_space= uint2korr(buff + EMPTY_SPACE_OFFSET);
 

storage/maria/ma_blockrec.h

@@ -222,6 +222,7 @@ void _ma_print_bitmap(MARIA_FILE_BITMAP *bitmap, uchar *data,
 
 uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
                                             uint page_type,
+                                            my_bool new_page,
                                             const uchar *header,
                                             const uchar *data,
                                             size_t data_length);

storage/maria/ma_key_recover.c

@@ -119,21 +119,6 @@ my_bool _ma_write_clr(MARIA_HA *info, LSN undo_lsn,
     page_store(log_pos + KEY_NR_STORE_SIZE, page);
     log_pos+= KEY_NR_STORE_SIZE + PAGE_STORE_SIZE;
   }
-  if (undo_type == LOGREC_UNDO_ROW_DELETE ||
-      undo_type == LOGREC_UNDO_ROW_UPDATE)
-  {
-    /*
-      We need to store position to the row that was inserted to be
-      able to regenerate keys
-    */
-    MARIA_RECORD_POS rowid= info->cur_row.lastpos;
-    ulonglong page= ma_recordpos_to_page(rowid);
-    uint dir_entry= ma_recordpos_to_dir_entry(rowid);
-    page_store(log_pos, page);
-    dirpos_store(log_pos+ PAGE_STORE_SIZE, dir_entry);
-    log_pos+= PAGE_STORE_SIZE + DIRPOS_STORE_SIZE;
-  }
-
   log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
   log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - log_data);
 

storage/maria/ma_loghandler.c

@@ -403,10 +403,17 @@ static LOG_DESC INIT_LOGREC_REDO_INSERT_ROW_TAIL=
  write_hook_for_redo, NULL, 0,
  "redo_insert_row_tail", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
 
-/* Use this entry next time we need to add a new entry */
-static LOG_DESC INIT_LOGREC_REDO_NOT_USED=
-{LOGRECTYPE_VARIABLE_LENGTH, 0, 8, NULL, write_hook_for_redo, NULL, 0,
- "redo_insert_row_blob", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
+static LOG_DESC INIT_LOGREC_REDO_NEW_ROW_HEAD=
+{LOGRECTYPE_VARIABLE_LENGTH, 0,
+ FILEID_STORE_SIZE + PAGE_STORE_SIZE + DIRPOS_STORE_SIZE, NULL,
+ write_hook_for_redo, NULL, 0,
+ "redo_new_row_head", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
+
+static LOG_DESC INIT_LOGREC_REDO_NEW_ROW_TAIL=
+{LOGRECTYPE_VARIABLE_LENGTH, 0,
+ FILEID_STORE_SIZE + PAGE_STORE_SIZE + DIRPOS_STORE_SIZE, NULL,
+ write_hook_for_redo, NULL, 0,
+ "redo_new_row_tail", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
 
 static LOG_DESC INIT_LOGREC_REDO_INSERT_ROW_BLOBS=
 {LOGRECTYPE_VARIABLE_LENGTH, 0, FILEID_STORE_SIZE, NULL,
@@ -594,8 +601,10 @@ void translog_table_init()
     INIT_LOGREC_REDO_INSERT_ROW_HEAD;
   log_record_type_descriptor[LOGREC_REDO_INSERT_ROW_TAIL]=
     INIT_LOGREC_REDO_INSERT_ROW_TAIL;
-  log_record_type_descriptor[LOGREC_REDO_NOT_USED]=
-    INIT_LOGREC_REDO_NOT_USED;
+  log_record_type_descriptor[LOGREC_REDO_NEW_ROW_HEAD]=
+    INIT_LOGREC_REDO_NEW_ROW_HEAD;
+  log_record_type_descriptor[LOGREC_REDO_NEW_ROW_TAIL]=
+    INIT_LOGREC_REDO_NEW_ROW_TAIL;
   log_record_type_descriptor[LOGREC_REDO_INSERT_ROW_BLOBS]=
     INIT_LOGREC_REDO_INSERT_ROW_BLOBS;
   log_record_type_descriptor[LOGREC_REDO_PURGE_ROW_HEAD]=

storage/maria/ma_loghandler.h

@@ -105,7 +105,8 @@ enum translog_record_type
   LOGREC_RESERVED_FOR_CHUNKS23= 0,
   LOGREC_REDO_INSERT_ROW_HEAD,
   LOGREC_REDO_INSERT_ROW_TAIL,
-  LOGREC_REDO_NOT_USED,                     /* Reserver for next tag */
+  LOGREC_REDO_NEW_ROW_HEAD,
+  LOGREC_REDO_NEW_ROW_TAIL,
   LOGREC_REDO_INSERT_ROW_BLOBS,
   LOGREC_REDO_PURGE_ROW_HEAD,
   LOGREC_REDO_PURGE_ROW_TAIL,

storage/maria/ma_recovery.c

@@ -88,7 +88,9 @@ prototype_redo_exec_hook(INCOMPLETE_LOG);
 prototype_redo_exec_hook_dummy(INCOMPLETE_GROUP);
 prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
 prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL);
-prototype_redo_exec_hook(REDO_INSERT_ROW_BLOBS);
+prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
+prototype_redo_exec_hook(REDO_NEW_ROW_TAIL);
+prototype_redo_exec_hook(REDO_NEW_ROW_BLOBS);
 prototype_redo_exec_hook(REDO_PURGE_ROW_HEAD);
 prototype_redo_exec_hook(REDO_PURGE_ROW_TAIL);
 prototype_redo_exec_hook(REDO_FREE_HEAD_OR_TAIL);
@@ -1312,6 +1314,10 @@ end:
   return error;
 }
 
+/*
+  NOTE
+  This is called for REDO_INSERT_ROW_HEAD and READ_NEW_ROW_HEAD
+*/
 
 prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD)
 {
@@ -1353,6 +1359,8 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD)
   buff= log_record_buffer.str;
   if (_ma_apply_redo_insert_row_head_or_tail(info, current_group_end_lsn,
                                              HEAD_PAGE,
+                                             (rec->type ==
+                                              LOGREC_REDO_NEW_ROW_HEAD),
                                              buff + FILEID_STORE_SIZE,
                                              buff +
                                              FILEID_STORE_SIZE +
@@ -1368,6 +1376,10 @@ end:
   return error;
 }
 
+/*
+  NOTE
+  This is called for REDO_INSERT_ROW_TAIL and READ_NEW_ROW_TAIL
+*/
 
 prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL)
 {
@@ -1388,6 +1400,8 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL)
   buff= log_record_buffer.str;
   if (_ma_apply_redo_insert_row_head_or_tail(info, current_group_end_lsn,
                                              TAIL_PAGE,
+                                             (rec->type ==
+                                              LOGREC_REDO_NEW_ROW_TAIL),
                                              buff + FILEID_STORE_SIZE,
                                              buff +
                                              FILEID_STORE_SIZE +
@@ -2189,6 +2203,9 @@ static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply)
 #define install_redo_exec_hook(R)                                        \
   log_record_type_descriptor[LOGREC_ ## R].record_execute_in_redo_phase= \
     exec_REDO_LOGREC_ ## R;
+#define install_redo_exec_hook_shared(R,S)                               \
+  log_record_type_descriptor[LOGREC_ ## R].record_execute_in_redo_phase= \
+    exec_REDO_LOGREC_ ## S;
 #define install_undo_exec_hook(R)                                        \
   log_record_type_descriptor[LOGREC_ ## R].record_execute_in_undo_phase= \
     exec_UNDO_LOGREC_ ## R;
@@ -2226,6 +2243,10 @@ static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply)
   install_undo_exec_hook(UNDO_KEY_INSERT);
   install_undo_exec_hook(UNDO_KEY_DELETE);
   install_undo_exec_hook(UNDO_KEY_DELETE_WITH_ROOT);
+  /* REDO_NEW_ROW_HEAD shares entry with REDO_INSERT_ROW_HEAD */
+  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);
 
   current_group_end_lsn= LSN_IMPOSSIBLE;
 #ifndef DBUG_OFF
@@ -2612,6 +2633,8 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
   case LOGREC_REDO_INSERT_ROW_TAIL:
   case LOGREC_REDO_PURGE_ROW_HEAD:
   case LOGREC_REDO_PURGE_ROW_TAIL:
+  case LOGREC_REDO_NEW_ROW_HEAD:
+  case LOGREC_REDO_NEW_ROW_TAIL:
     llstr(page, llbuf);
     tprint(tracef, "   For page %s of table of short id %u", llbuf, sid);
     break;