WL#3072 - Maria Recovery: recovery of state.auto_increment.

When we log UNDO_KEY_INSERT for an auto_inc key, we update
state.auto_increment (not anymore at the end of maria_write() except
if this is a non-transactional table). When Recovery sees UNDO_KEY_INSERT
in the REDO phase, it reads the auto_inc value from it and updates
state.auto_increment.


mysql-test/r/maria-recovery.result:
  Without the code fix, there would be in CHECK TABLE:
  "Auto-increment value: 0 is smaller than max used value: 3"
  and no AUTO_INCREMENT= clause in SHOW CREATE TABLE.
mysql-test/t/maria-recovery.test:
  Test of recovery of state.auto_increment: from an old table,
  does the replaying of the log set state.auto_increment to
  what it should be.
storage/maria/ma_check.c:
  new way of calling ma_retrieve_auto_increment(): pass key
storage/maria/ma_key.c:
  ma_retrieve_auto_increment() now operates directly with a pointer
  to the key and not on the record.
storage/maria/ma_key_recover.c:
  dedicated write_hook_for_undo_key_insert(): sets state.auto_increment
  under log's mutex.
storage/maria/ma_key_recover.h:
  Dedicated hook for UNDO_KEY_INSERT, to set state.auto_increment.
  Such hook needs a new member st_msg_write_hook_for_undo_key::auto_increment,
  which contains the auto_increment value inserted.
storage/maria/ma_loghandler.c:
  UNDO_KEY_INSERT gets a dedicated write_hook, to set auto_increment.
storage/maria/ma_recovery.c:
  When in the REDO phase we see UNDO_KEY_INSERT: if the state is older
  than this record, and the key is the auto_increment one, read
  the key's value from the log record and update state.auto_increment.
storage/maria/ma_test_all.sh:
  use $maria_path to be able to run from /dev/shm (faster)
storage/maria/ma_update.c:
  bool is more of C++, using my_bool.
  If table is transactional, state.auto_increment is already updated
  in write_hook_for_undo_key_insert().
storage/maria/ma_write.c:
  If table is transactional, state.auto_increment is not updated at
  the end of maria_write() but rather in write_hook_for_undo_key_insert()
  (under log's mutex, so that a concurrent checkpoint does not read
  state.auto_increment while it is changing - corrupted).
  _ma_ck_write_btree_with_log() extracts the auto_increment value
  from the key, puts it into msg.auto_increment, and this is passed
  to write_hook_for_undo_key_insert().
storage/maria/maria_def.h:
  change of prototype of ma_retrieve_auto_increment()
storage/maria/maria_read_log.c:
  use default log file size. Use separate page caches for table
  and logs (needed if maria_block_size!=TRANSLOG_PAGE_SIZE).

mysql-test/r/maria-recovery.result

@@ -150,6 +150,66 @@ SELECT LENGTH(b) FROM t1 WHERE i=3;
 LENGTH(b)
 5001
 drop table t1;
+* shut down mysqld, removed logs, restarted it
+use mysqltest;
+CREATE TABLE t1 (
+i int auto_increment primary key,
+c varchar(6),
+key(c)
+) ENGINE=MARIA;
+insert into t1 values(null,"b");
+* copied t1 for feeding_recovery
+insert into t1 values(null,"a"), (null,"c"), (null,"d");
+delete from t1 where c="d";
+flush table t1;
+* copied t1 for comparison
+SET SESSION debug="+d,maria_flush_whole_log,maria_crash";
+* crashing mysqld intentionally
+set global maria_checkpoint_interval=1;
+ERROR HY000: Lost connection to MySQL server during query
+* copied t1 back for feeding_recovery
+* 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
+ok
+use mysqltest;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `i` int(11) NOT NULL AUTO_INCREMENT,
+  `c` varchar(6) DEFAULT NULL,
+  PRIMARY KEY (`i`),
+  KEY `c` (`c`)
+) ENGINE=MARIA AUTO_INCREMENT=5 DEFAULT CHARSET=latin1
+* copied t1 for feeding_recovery
+update t1 set i=15 where c="a";
+flush table t1;
+* copied t1 for comparison
+SET SESSION debug="+d,maria_flush_whole_log,maria_crash";
+* crashing mysqld intentionally
+set global maria_checkpoint_interval=1;
+ERROR HY000: Lost connection to MySQL server during query
+* copied t1 back for feeding_recovery
+* 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
+ok
+use mysqltest;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `i` int(11) NOT NULL AUTO_INCREMENT,
+  `c` varchar(6) DEFAULT NULL,
+  PRIMARY KEY (`i`),
+  KEY `c` (`c`)
+) ENGINE=MARIA AUTO_INCREMENT=16 DEFAULT CHARSET=latin1
+drop table t1;
 drop database mysqltest_for_feeding_recovery;
 drop database mysqltest_for_comparison;
 drop database mysqltest;

mysql-test/t/maria-recovery.test

@@ -141,6 +141,39 @@ let $mvr_crash_statement= set global maria_checkpoint_interval=1;
 SELECT LENGTH(b) FROM t1 WHERE i=3;
 drop table t1;
 
+# Test that INSERT's effect on auto-increment is recovered
+-- source include/maria_empty_logs.inc
+let $mms_tables=1;
+CREATE TABLE t1 (
+  i int auto_increment primary key,
+  c varchar(6),
+  key(c)
+) ENGINE=MARIA;
+insert into t1 values(null,"b");
+-- source include/maria_make_snapshot_for_feeding_recovery.inc
+insert into t1 values(null,"a"), (null,"c"), (null,"d");
+# With this DELETE we also verify that Recovery cares only about INSERTs
+delete from t1 where c="d";
+-- source include/maria_make_snapshot_for_comparison.inc
+let $mvr_restore_old_snapshot=1;
+let $mms_compare_physically=0;
+let $mvr_debug_option="+d,maria_flush_whole_log,maria_crash";
+let $mvr_crash_statement= set global maria_checkpoint_interval=1;
+-- source include/maria_verify_recovery.inc
+show create table t1;
+
+# Test that UPDATE's effect on auto-increment is recovered
+-- source include/maria_make_snapshot_for_feeding_recovery.inc
+update t1 set i=15 where c="a";
+-- source include/maria_make_snapshot_for_comparison.inc
+let $mvr_restore_old_snapshot=1;
+let $mms_compare_physically=0;
+let $mvr_debug_option="+d,maria_flush_whole_log,maria_crash";
+let $mvr_crash_statement= set global maria_checkpoint_interval=1;
+-- source include/maria_verify_recovery.inc
+show create table t1;
+drop table t1;
+
 # clean up everything
 let $mms_purpose=feeding_recovery;
 eval drop database mysqltest_for_$mms_purpose;

storage/maria/ma_check.c

@@ -538,9 +538,12 @@ int maria_chk_key(HA_CHECK *param, register MARIA_HA *info)
     {
       /* Check that auto_increment key is bigger than max key value */
       ulonglong auto_increment;
+      const HA_KEYSEG *keyseg= share->keyinfo[share->base.auto_key-1].seg;
       info->lastinx=key;
       _ma_read_key_record(info, info->rec_buff, 0);
-      auto_increment= ma_retrieve_auto_increment(info, info->rec_buff);
+      auto_increment=
+        ma_retrieve_auto_increment(info->rec_buff + keyseg->start,
+                                   keyseg->type);
       if (auto_increment > share->state.auto_increment)
       {
 	_ma_check_print_warning(param, "Auto-increment value: %s is smaller "
@@ -5369,7 +5372,9 @@ void _ma_update_auto_increment_key(HA_CHECK *param, MARIA_HA *info,
   }
   else
   {
-    ulonglong auto_increment= ma_retrieve_auto_increment(info, record);
+    const HA_KEYSEG *keyseg= share->keyinfo[share->base.auto_key-1].seg;
+    ulonglong auto_increment=
+      ma_retrieve_auto_increment(record + keyseg->start, keyseg->type);
     set_if_bigger(share->state.auto_increment,auto_increment);
     if (!repair_only)
       set_if_bigger(share->state.auto_increment, param->auto_increment_value);

storage/maria/ma_key.c

@@ -493,22 +493,24 @@ int _ma_read_key_record(MARIA_HA *info, uchar *buf, MARIA_RECORD_POS filepos)
 
   SYNOPSIS
     retrieve_auto_increment()
-    info			Maria handler
-    record			Row to update
+    key                         Auto-increment key
+    key_type                    Key's type
+
+  NOTE
+    'key' should in "record" format, that is, how it is packed in a record
+    (this matters with HA_SWAP_KEY).
 
   IMPLEMENTATION
     For signed columns we don't retrieve the auto increment value if it's
     less than zero.
 */
 
-ulonglong ma_retrieve_auto_increment(MARIA_HA *info,const uchar *record)
+ulonglong ma_retrieve_auto_increment(const uchar *key, uint8 key_type)
 {
   ulonglong value= 0;			/* Store unsigned values here */
   longlong s_value= 0;			/* Store signed values here */
-  HA_KEYSEG *keyseg= info->s->keyinfo[info->s->base.auto_key-1].seg;
-  const uchar *key= record + keyseg->start;
 
-  switch (keyseg->type) {
+  switch (key_type) {
   case HA_KEYTYPE_INT8:
     s_value= (longlong) *(char*)key;
     break;

storage/maria/ma_key_recover.c

@@ -191,6 +191,34 @@ my_bool write_hook_for_undo_key(enum translog_record_type type,
 }
 
 
+/**
+   Upates "auto_increment" and calls the generic UNDO_KEY hook
+
+   @return Operation status, always 0 (success)
+*/
+
+my_bool write_hook_for_undo_key_insert(enum translog_record_type type,
+                                       TRN *trn, MARIA_HA *tbl_info,
+                                       LSN *lsn, void *hook_arg)
+{
+  struct st_msg_to_write_hook_for_undo_key *msg=
+    (struct st_msg_to_write_hook_for_undo_key *) hook_arg;
+  MARIA_SHARE *share= tbl_info->s;
+  if (msg->auto_increment > 0)
+  {
+    /*
+      Only reason to set it here is to have a mutex protect from checkpoint
+      reading at the same time (would see a corrupted value).
+    */
+    DBUG_PRINT("info",("auto_inc: %lu new auto_inc: %lu",
+                       (ulong)share->state.auto_increment,
+                       (ulong)msg->auto_increment));
+    set_if_bigger(share->state.auto_increment, msg->auto_increment);
+  }
+  return write_hook_for_undo_key(type, trn, tbl_info, lsn, hook_arg);
+}
+
+
 /*****************************************************************************
   Functions for logging of key page changes
 *****************************************************************************/

storage/maria/ma_key_recover.h

@@ -36,6 +36,7 @@ struct st_msg_to_write_hook_for_undo_key
   my_off_t *root;
   my_off_t value;
   uint keynr;
+  ulonglong auto_increment;
 };
 
 
@@ -51,6 +52,9 @@ my_bool write_hook_for_clr_end(enum translog_record_type type,
 extern my_bool write_hook_for_undo_key(enum translog_record_type type,
                                        TRN *trn, MARIA_HA *tbl_info,
                                        LSN *lsn, void *hook_arg);
+extern my_bool write_hook_for_undo_key_insert(enum translog_record_type type,
+                                              TRN *trn, MARIA_HA *tbl_info,
+                                              LSN *lsn, void *hook_arg);
 void _ma_unpin_all_pages(MARIA_HA *info, LSN undo_lsn);
 
 my_bool _ma_log_prefix(MARIA_HA *info, my_off_t page,

storage/maria/ma_loghandler.c

@@ -460,7 +460,7 @@ static LOG_DESC INIT_LOGREC_UNDO_ROW_UPDATE=
 static LOG_DESC INIT_LOGREC_UNDO_KEY_INSERT=
 {LOGRECTYPE_VARIABLE_LENGTH, 0,
  LSN_STORE_SIZE + FILEID_STORE_SIZE + KEY_NR_STORE_SIZE,
- NULL, write_hook_for_undo_key, NULL, 1,
+ NULL, write_hook_for_undo_key_insert, NULL, 1,
  "undo_key_insert", LOGREC_LAST_IN_GROUP, NULL, NULL};
 
 /* This will never be in the log, only in the clr */

storage/maria/ma_recovery.c

@@ -1647,9 +1647,53 @@ prototype_redo_exec_hook(UNDO_ROW_UPDATE)
 prototype_redo_exec_hook(UNDO_KEY_INSERT)
 {
   MARIA_HA *info;
+  MARIA_SHARE *share;
   if (!(info= get_MARIA_HA_from_UNDO_record(rec)))
     return 0;
+  share= info->s;
   set_undo_lsn_for_active_trans(rec->short_trid, rec->lsn);
+  if (cmp_translog_addr(rec->lsn, share->state.is_of_horizon) >= 0)
+  {
+    const uchar *ptr= rec->header + LSN_STORE_SIZE + FILEID_STORE_SIZE;
+    uint keynr= key_nr_korr(ptr);
+    if (share->base.auto_key == (keynr + 1)) /* it's auto-increment */
+    {
+      const HA_KEYSEG *keyseg= info->s->keyinfo[keynr].seg;
+      ulonglong value;
+      char llbuf[22];
+      uchar *to;
+      tprint(tracef, "   state older than record\n");
+      /* we read the record to find the auto_increment value */
+      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)
+      {
+        eprint(tracef, "Failed to read record\n");
+        return 1;
+      }
+      to= log_record_buffer.str + LSN_STORE_SIZE + FILEID_STORE_SIZE +
+        KEY_NR_STORE_SIZE;
+      if (keyseg->flag & HA_SWAP_KEY)
+      {
+        /* We put key from log record to "data record" packing format... */
+        uchar reversed[HA_MAX_KEY_BUFF];
+        uchar *key_ptr= to;
+        uchar *key_end= key_ptr + keyseg->length;
+        to= reversed + keyseg->length;
+        do
+        {
+          *--to= *key_ptr++;
+        } while (key_ptr != key_end);
+        /* ... so that we can read it with: */
+      }
+      value= ma_retrieve_auto_increment(to, keyseg->type);
+      set_if_bigger(share->state.auto_increment, value);
+      llstr(share->state.auto_increment, llbuf);
+      tprint(tracef, "   auto-inc %s\n", llbuf);
+    }
+  }
   _ma_unpin_all_pages(info, rec->lsn);
   return 0;
 }

storage/maria/ma_test_all.sh

@@ -245,23 +245,23 @@ $maria_path/maria_chk$suffix -ssm 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_path/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
+$maria_path/maria_read_log -a -s
+$maria_path/maria_chk -s -e test2
 cp tmp/maria_log_control .
 rm test2.MA?
-maria_read_log -a -s
-maria_chk -s -e test2
+$maria_path/maria_read_log -a -s
+$maria_path/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
+$maria_path/ma_test2 -s -L -K -W -P -M -T -c -b -t2 -u1
+$maria_path/maria_read_log -a -s
+$maria_path/maria_chk -s -e test2
 rm test2.MA?
-maria_read_log -a -s
-maria_chk -e -s test2
+$maria_path/maria_read_log -a -s
+$maria_path/maria_chk -e -s test2
 
 #
 # Some timing tests

storage/maria/ma_update.c

@@ -24,7 +24,7 @@ int maria_update(register MARIA_HA *info, const uchar *oldrec, uchar *newrec)
   reg3 my_off_t pos;
   uint i;
   uchar old_key[HA_MAX_KEY_BUFF],*new_key;
-  bool auto_key_changed=0;
+  my_bool auto_key_changed= 0;
   ulonglong changed;
   MARIA_SHARE *share= info->s;
   DBUG_ENTER("maria_update");
@@ -158,9 +158,13 @@ int maria_update(register MARIA_HA *info, const uchar *oldrec, uchar *newrec)
     if ((*share->update_record)(info, pos, oldrec, newrec))
       goto err;
   }
-  if (auto_key_changed)
+  if (auto_key_changed & !share->now_transactional)
+  {
+    const HA_KEYSEG *keyseg= share->keyinfo[share->base.auto_key-1].seg;
+    const uchar *key= newrec + keyseg->start;
     set_if_bigger(share->state.auto_increment,
-                  ma_retrieve_auto_increment(info, newrec));
+                  ma_retrieve_auto_increment(key, keyseg->type));
+  }
 
   /*
     We can't yet have HA_STATE_AKTIV here, as block_record dosn't support

storage/maria/ma_write.c

@@ -208,15 +208,12 @@ int maria_write(MARIA_HA *info, uchar *record)
     info->state->checksum+= !share->now_transactional *
       info->cur_row.checksum;
   }
-  if (share->base.auto_key)
+  if ((share->base.auto_key != 0) & !share->now_transactional)
   {
-    /**
-       @todo RECOVERY BUG
-       if updated here, it's not recoverable (no mutex => checkpoint may see a
-       crazy value and flush it into the table's state on disk).
-    */
+    const HA_KEYSEG *keyseg= share->keyinfo[share->base.auto_key-1].seg;
+    const uchar *key= record + keyseg->start;
     set_if_bigger(share->state.auto_increment,
-                  ma_retrieve_auto_increment(info, record));
+                  ma_retrieve_auto_increment(key, keyseg->type));
   }
   info->update= (HA_STATE_CHANGED | HA_STATE_AKTIV | HA_STATE_WRITTEN |
 		 HA_STATE_ROW_CHANGED);
@@ -412,6 +409,28 @@ static int _ma_ck_write_btree_with_log(MARIA_HA *info, MARIA_KEYDEF *keyinfo,
 
     msg.root= root;
     msg.value= new_root;
+    msg.auto_increment= 0;
+    if (share->base.auto_key == ((uint)keyinfo->key_nr + 1))
+    {
+      const HA_KEYSEG *keyseg= keyinfo->seg;
+      uchar *to= key_buff;
+      if (keyseg->flag & HA_SWAP_KEY)
+      {
+        /* We put key from log record to "data record" packing format... */
+        uchar reversed[HA_MAX_KEY_BUFF];
+        uchar *key_ptr= to;
+        uchar *key_end= key_ptr + keyseg->length;
+        to= reversed + keyseg->length;
+        do
+        {
+          *--to= *key_ptr++;
+        } while (key_ptr != key_end);
+      }
+      /* ... so that we can read it with: */
+      msg.auto_increment=
+        ma_retrieve_auto_increment(to, keyseg->type);
+      /* and write_hook_for_undo_key_insert() will pick this. */
+    }
 
     if (translog_write_record(&lsn, LOGREC_UNDO_KEY_INSERT,
                               info->trn, info,

storage/maria/maria_def.h

@@ -851,8 +851,7 @@ extern uint _ma_pack_key(MARIA_HA *info, uint keynr, uchar *key,
 extern int _ma_read_key_record(MARIA_HA *info, uchar *buf, MARIA_RECORD_POS);
 extern int _ma_read_cache(IO_CACHE *info, uchar *buff, MARIA_RECORD_POS pos,
                           uint length, int re_read_if_possibly);
-extern ulonglong ma_retrieve_auto_increment(MARIA_HA *info, const uchar *record);
-
+extern ulonglong ma_retrieve_auto_increment(const uchar *key, uint8 key_type);
 extern my_bool _ma_alloc_buffer(uchar **old_addr, size_t *old_size,
                                 size_t new_size);
 extern ulong _ma_rec_unpack(MARIA_HA *info, uchar *to, uchar *from,

storage/maria/maria_read_log.c

@@ -18,7 +18,6 @@
 #include <my_getopt.h>
 
 #define LOG_FLAGS 0
-#define LOG_FILE_SIZE (1024L*1024L)
 
 static const char *load_default_groups[]= { "maria_read_log",0 };
 static void get_options(int *argc,char * * *argv);
@@ -67,8 +66,6 @@ int main(int argc, char **argv)
     fprintf(stderr, "Can't find any log\n");
     goto err;
   }
-  /* same page cache for log and data; assumes same page size... */
-  DBUG_ASSERT(maria_block_size == TRANSLOG_PAGE_SIZE);
   if (init_pagecache(maria_pagecache, opt_page_buffer_size, 0, 0,
                      TRANSLOG_PAGE_SIZE, MY_WME) == 0)
   {
@@ -81,8 +78,11 @@ int main(int argc, char **argv)
     But if it finds a log and this log was crashed, it will create a new log,
     which is useless. TODO: start log handler in read-only mode.
   */
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, maria_pagecache,
-                    TRANSLOG_DEFAULT_FLAGS))
+  if (init_pagecache(maria_log_pagecache,
+                     TRANSLOG_PAGECACHE_SIZE, 0, 0,
+                     TRANSLOG_PAGE_SIZE, MY_WME) == 0 ||
+      translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
+                    0, 0, maria_log_pagecache, TRANSLOG_DEFAULT_FLAGS))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
     goto err;