Fix for BUG#37876 "Importing Maria table from other server via binary copy does not work":

- after auto-zerofill (ha_maria::check_and_repair()) kepts its state's LSNs unchanged, which could
be the same as the create_rename_lsn of another pre-existing table, which would break versioning as this LSN
serves as unique identifier in the versioning code (in maria_open()). Even the state pieces which
maria_zerofill() did change were lost (because they didn't go to disk).
- after this fix, if two tables were auto-zerofilled at the same time (by _ma_mark_changed())
they could receive the same create_rename_lsn, which would break versioning again. Fix is to write a log
record each time a table is imported.
- Print state's LSNs (create_rename_lsn, is_of_horizon, skip_redo_lsn) and UUID in maria_chk -dvv.

mysql-test/r/maria-autozerofill.result:
  result
mysql-test/t/maria-autozerofill.test:
  Test for auto-zerofilling
storage/maria/ha_maria.cc:
  The state changes done by auto-zerofilling never reached disk.
storage/maria/ma_check.c:
  When zerofilling a table, including its pages' LSNs, new state LSNs are needed next time the table
  is imported into a Maria instance.
storage/maria/ma_create.c:
  Write LOGREC_IMPORTED_TABLE when importing a table. This is informative and ensures
  that the table gets a unique create_rename_lsn even though multiple tables
  are imported by concurrent threads (it advances the log's end LSN).
storage/maria/ma_key_recover.c:
  comment
storage/maria/ma_locking.c:
  instead of using translog_get_horizon() for state's LSNs of imported table,
  use the LSN of to-be-written LOGREC_IMPORTED_TABLE.
storage/maria/ma_loghandler.c:
  New type of log record
storage/maria/ma_loghandler.h:
  New type of log record
storage/maria/ma_loghandler_lsn.h:
  New name for constant as can be used not only by maria_chk but auto-zerofill now too.
storage/maria/ma_open.c:
  instead of using translog_get_horizon() for state's LSNs of imported table,
  use the LSN of to-be-written LOGREC_IMPORTED_TABLE.
storage/maria/ma_recovery.c:
  print content of LOGREC_IMPORTED_TABLE in maria_read_log.
storage/maria/maria_chk.c:
  print info about LSNs of the table's state, and UUID, when maria_chk -dvv
storage/maria/maria_pack.c:
  new name for constant
storage/maria/unittest/ma_test_recovery.pl:
  Now that maria_chk -dvv shows state LSNs and UUID those need to be filtered out,
  as maria_read_log -a does not use the same as at original run.

mysql-test/r/maria-autozerofill.result

+drop database if exists mysqltest;
+create database mysqltest;
+use mysqltest;
+create table t1(a int) engine=maria;
+insert into t1 values(1);
+flush table t1;
+create_rename_lsn has non-magic value
+* shut down mysqld, removed logs, restarted it
+select * from t1;
+a
+1
+Warnings:
+Error	1194	t1' is marked as crashed and should be repaired
+flush table t1;
+Status:              changed,sorted index pages,zerofilled,movable
+create_rename_lsn has magic value
+insert into t1 values(2);
+flush table t1;
+create_rename_lsn has non-magic value
+drop database mysqltest;

mysql-test/t/maria-autozerofill.test

+# Test to verify that auto-zerofilling happens when a table is
+# imported from a different Maria instance
+
+# can't restart in embedded
+--source include/not_embedded.inc
+--source include/have_maria.inc
+
+let $MARIA_LOG=.;
+
+--disable_warnings
+drop database if exists mysqltest;
+--enable_warnings
+create database mysqltest;
+let $mms_tname=t;
+
+connect (admin, localhost, root,,mysqltest,,);
+--enable_reconnect
+
+connection default;
+use mysqltest;
+--enable_reconnect
+
+create table t1(a int) engine=maria;
+insert into t1 values(1);
+flush table t1;
+# Check that table is not zerofilled, not movable
+--exec $MARIA_CHK -dv $MYSQLTEST_VARDIR/master-data/mysqltest/t1 >$MYSQLTEST_VARDIR/tmp/mariachk.txt
+perl;
+    use strict;
+    use warnings;
+    my $fname= "$ENV{'MYSQLTEST_VARDIR'}/tmp/mariachk.txt";
+    open(FILE, "<", $fname) or die;
+    my @content= <FILE>;
+    print grep(/Status:.*(zerofilled|movable)/, @content);
+    print "create_rename_lsn has non-magic value\n" if grep(/create_rename \([0-9]+/, @content);
+    close FILE;
+EOF
+
+# this will remove control file, so change the uuid of the Maria
+# instance, thus t1 will appear as imported from elsewhere.
+
+-- source include/maria_empty_logs.inc
+
+disable_ps_protocol; # see maria-recover.test
+replace_regex /Table.*t1/t1/ ;
+select * from t1;
+enable_ps_protocol;
+flush table t1;
+
+# Check that table is auto-zerofilled, movable
+--exec $MARIA_CHK -dv $MYSQLTEST_VARDIR/master-data/mysqltest/t1 >$MYSQLTEST_VARDIR/tmp/mariachk.txt
+perl;
+    use strict;
+    use warnings;
+    my $fname= "$ENV{'MYSQLTEST_VARDIR'}/tmp/mariachk.txt";
+    open(FILE, "<", $fname) or die;
+    my @content= <FILE>;
+    print grep(/Status:.*zerofilled/, @content);
+    print "create_rename_lsn has magic value\n" if grep(/create_rename \(0,0x2\)/, @content);
+    close FILE;
+EOF
+
+# this will attach t1 to the current Maria instance
+insert into t1 values(2);
+flush table t1;
+
+# Check that table is not zerofilled, not movable
+--exec $MARIA_CHK -dv $MYSQLTEST_VARDIR/master-data/mysqltest/t1 >$MYSQLTEST_VARDIR/tmp/mariachk.txt
+perl;
+    use strict;
+    use warnings;
+    my $fname= "$ENV{'MYSQLTEST_VARDIR'}/tmp/mariachk.txt";
+    open(FILE, "<", $fname) or die;
+    my @content= <FILE>;
+    print grep(/Status:.*(zerofilled|movable)/, @content);
+    print "create_rename_lsn has non-magic value\n" if grep(/create_rename \([0-9]+/, @content);
+    close FILE;
+EOF
+
+drop database mysqltest;

storage/maria/ha_maria.cc

@@ -1268,6 +1268,7 @@ int ha_maria::zerofill(THD * thd, HA_CHECK_OPT *check_opt)
 {
   int error;
   HA_CHECK param;
+  MARIA_SHARE *share= file->s;
 
   if (!file)
     return HA_ADMIN_INTERNAL_ERROR;
@@ -1277,8 +1278,14 @@ int ha_maria::zerofill(THD * thd, HA_CHECK_OPT *check_opt)
   param.op_name= "zerofill";
   param.testflag= check_opt->flags | T_SILENT | T_ZEROFILL;
   param.sort_buffer_length= THDVAR(thd, sort_buffer_size);
-  error=maria_zerofill(&param, file, file->s->open_file_name);
+  error=maria_zerofill(&param, file, share->open_file_name);
 
+  if (!error)
+  {
+    pthread_mutex_lock(&share->intern_lock);
+    maria_update_state_info(&param, file, UPDATE_TIME | UPDATE_OPEN_COUNT);
+    pthread_mutex_unlock(&share->intern_lock);
+  }
   return error;
 }
 

storage/maria/ma_check.c

@@ -3368,8 +3368,9 @@ int maria_zerofill(HA_CHECK *param, MARIA_HA *info, const char *name)
 {
   my_bool error, reenable_logging,
     zero_lsn= !(param->testflag & T_ZEROFILL_KEEP_LSN);
+  MARIA_SHARE *share= info->s;
   DBUG_ENTER("maria_zerofill");
-  if ((reenable_logging= info->s->now_transactional))
+  if ((reenable_logging= share->now_transactional))
     _ma_tmp_disable_logging_for_table(info, 0);
   if (!(error= (maria_zerofill_index(param, info, name) ||
                 maria_zerofill_data(param, info, name) ||
@@ -3379,14 +3380,19 @@ int maria_zerofill(HA_CHECK *param, MARIA_HA *info, const char *name)
       Mark that we have done zerofill of data and index. If we zeroed pages'
       LSN, table is movable.
     */
-    info->s->state.changed&= ~STATE_NOT_ZEROFILLED;
+    share->state.changed&= ~STATE_NOT_ZEROFILLED;
     if (zero_lsn)
-      info->s->state.changed&= ~(STATE_NOT_MOVABLE | STATE_MOVED);
-    /* Ensure state are flushed to disk */
+    {
+      share->state.changed&= ~(STATE_NOT_MOVABLE | STATE_MOVED);
+      /* Table should get new LSNs */
+      share->state.create_rename_lsn= share->state.is_of_horizon=
+        share->state.skip_redo_lsn= LSN_NEEDS_NEW_STATE_LSNS;
+    }
+    /* Ensure state is later flushed to disk, if within maria_chk */
     info->update= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
 
     /* Reset create_trid to make file comparable */
-    info->s->state.create_trid= 0;
+    share->state.create_trid= 0;
   }
   if (reenable_logging)
     _ma_reenable_logging_for_table(info, FALSE);

storage/maria/ma_create.c

@@ -1317,7 +1317,8 @@ int _ma_update_state_lsns(MARIA_SHARE *share, LSN lsn, TrID create_trid,
    needed (when creating a table or opening it for the first time).
 
    @param  share           table's share
-   @param  lsn		   LSN to write to log files
+   @param  lsn             LSN to write to state; if LSN_IMPOSSIBLE, write
+                           a LOGREC_IMPORTED_TABLE and use its LSN as lsn.
    @param  create_trid     Trid to be used as state.create_trid
    @param  do_sync         if the write should be forced to disk
    @param  update_create_rename_lsn if this LSN should be updated or not
@@ -1342,6 +1343,25 @@ int _ma_update_state_lsns_sub(MARIA_SHARE *share, LSN lsn, TrID create_trid,
   uchar trid_buff[8];
   File file= share->kfile.file;
   DBUG_ASSERT(file >= 0);
+
+  if (lsn == LSN_IMPOSSIBLE)
+  {
+    int res;
+    LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
+    /* table name is logged only for information */
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    share->open_file_name;
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].length=
+      strlen(log_array[TRANSLOG_INTERNAL_PARTS + 0].str) + 1;
+    if ((res= translog_write_record(&lsn, LOGREC_IMPORTED_TABLE,
+                                    &dummy_transaction_object, NULL,
+                                    (translog_size_t)
+                                    log_array[TRANSLOG_INTERNAL_PARTS +
+                                              0].length,
+                                    sizeof(log_array)/sizeof(log_array[0]),
+                                    log_array, NULL, NULL)))
+      return res;
+  }
+
   for (ptr= buf; ptr < (buf + sizeof(buf)); ptr+= LSN_STORE_SIZE)
     lsn_store(ptr, lsn);
   share->state.skip_redo_lsn= share->state.is_of_horizon= lsn;

storage/maria/ma_key_recover.c

@@ -207,7 +207,8 @@ my_bool write_hook_for_undo_key(enum translog_record_type type,
     @todo BUG
     so we have log mutex and then intern_lock.
     While in checkpoint we have intern_lock and then log mutex, like when we
-    flush bitmap (flushing bitmap pages can call hook which takes log mutex).
+    flush bitmap (flushing bitmap pages can call hook which takes log mutex);
+    and in _ma_update_state_lsns_sub() this is the same.
     So we can deadlock.
     Another one is that in translog_assign_id_to_share() we have intern_lock
     and then log mutex.

storage/maria/ma_locking.c

@@ -425,8 +425,8 @@ int _ma_mark_file_changed(MARIA_HA *info)
     {
       /* Lock table to current installation */
       if (_ma_set_uuid(info, 0) ||
-          (share->state.create_rename_lsn == LSN_REPAIRED_BY_MARIA_CHK &&
-           _ma_update_state_lsns_sub(share, translog_get_horizon(),
+          (share->state.create_rename_lsn == LSN_NEEDS_NEW_STATE_LSNS &&
+           _ma_update_state_lsns_sub(share, LSN_IMPOSSIBLE,
                                      trnman_get_min_trid(),
                                      TRUE, TRUE)))
         DBUG_RETURN(1);

storage/maria/ma_loghandler.c

@@ -671,6 +671,10 @@ static LOG_DESC INIT_LOGREC_REDO_BITMAP_NEW_PAGE=
  NULL, NULL, NULL, 0,
  "redo_create_bitmap", LOGREC_IS_GROUP_ITSELF, NULL, NULL};
 
+static LOG_DESC INIT_LOGREC_IMPORTED_TABLE=
+{LOGRECTYPE_VARIABLE_LENGTH, 0, 0, NULL, NULL, NULL, 0,
+ "imported_table", LOGREC_IS_GROUP_ITSELF, NULL, NULL};
+
 const myf log_write_flags= MY_WME | MY_NABP | MY_WAIT_IF_FULL;
 
 void translog_table_init()
@@ -758,6 +762,8 @@ void translog_table_init()
     INIT_LOGREC_UNDO_BULK_INSERT;
   log_record_type_descriptor[LOGREC_REDO_BITMAP_NEW_PAGE]=
     INIT_LOGREC_REDO_BITMAP_NEW_PAGE;
+  log_record_type_descriptor[LOGREC_IMPORTED_TABLE]=
+    INIT_LOGREC_IMPORTED_TABLE;
   for (i= LOGREC_FIRST_FREE; i < LOGREC_NUMBER_OF_TYPES; i++)
     log_record_type_descriptor[i].rclass= LOGRECTYPE_NOT_ALLOWED;
 #ifndef DBUG_OFF

storage/maria/ma_loghandler.h

@@ -143,6 +143,7 @@ enum translog_record_type
   LOGREC_INCOMPLETE_GROUP,
   LOGREC_UNDO_BULK_INSERT,
   LOGREC_REDO_BITMAP_NEW_PAGE,
+  LOGREC_IMPORTED_TABLE,
   LOGREC_FIRST_FREE,
   LOGREC_RESERVED_FUTURE_EXTENSION= 63
 };

storage/maria/ma_loghandler_lsn.h

@@ -93,7 +93,13 @@ typedef LSN LSN_WITH_FLAGS;
 #define LSN_ERROR             ((LSN)1)
 
 /** @brief some impossible LSN serve as markers */
-#define LSN_REPAIRED_BY_MARIA_CHK ((LSN)2)
+
+/**
+   When table is modified by maria_chk, or auto-zerofilled, old REDOs don't
+   apply, table is freshly born again somehow: its state's LSNs need to be
+   updated to the new instance which receives this table.
+*/
+#define LSN_NEEDS_NEW_STATE_LSNS ((LSN)2)
 
 /**
    @brief the maximum valid LSN.

storage/maria/ma_open.c

@@ -665,7 +665,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     if (share->base.born_transactional)
     {
       share->page_type= PAGECACHE_LSN_PAGE;
-      if (share->state.create_rename_lsn == LSN_REPAIRED_BY_MARIA_CHK)
+      if (share->state.create_rename_lsn == LSN_NEEDS_NEW_STATE_LSNS)
       {
         /*
           Was repaired with maria_chk, maybe later maria_pack-ed. Some sort of
@@ -674,7 +674,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
         */
         if (((open_flags & HA_OPEN_FROM_SQL_LAYER) &&
              (share->state.changed & STATE_NOT_MOVABLE)) || maria_in_recovery)
-          _ma_update_state_lsns_sub(share, translog_get_horizon(),
+          _ma_update_state_lsns_sub(share, LSN_IMPOSSIBLE,
                                     trnman_get_min_safe_trid(), TRUE, TRUE);
       }
       else if ((!LSN_VALID(share->state.create_rename_lsn) ||
@@ -688,7 +688,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
       {
         /*
           If in Recovery, it will not work. If LSN is invalid and not
-          LSN_REPAIRED_BY_MARIA_CHK, header must be corrupted.
+          LSN_NEEDS_NEW_STATE_LSNS, header must be corrupted.
           In both cases, must repair.
         */
         my_errno=((share->state.changed & STATE_CRASHED_ON_REPAIR) ?

storage/maria/ma_recovery.c

@@ -77,6 +77,7 @@ prototype_redo_exec_hook(FILE_ID);
 prototype_redo_exec_hook(INCOMPLETE_LOG);
 prototype_redo_exec_hook_dummy(INCOMPLETE_GROUP);
 prototype_redo_exec_hook(UNDO_BULK_INSERT);
+prototype_redo_exec_hook(IMPORTED_TABLE);
 prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
 prototype_redo_exec_hook(REDO_INSERT_ROW_TAIL);
 prototype_redo_exec_hook(REDO_INSERT_ROW_HEAD);
@@ -1861,6 +1862,24 @@ prototype_redo_exec_hook(UNDO_BULK_INSERT)
 }
 
 
+prototype_redo_exec_hook(IMPORTED_TABLE)
+{
+  char *name;
+  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");
+    return 1;
+  }
+  name= (char *)log_record_buffer.str;
+  tprint(tracef, "Table '%s' was imported (auto-zerofilled) in this Maria instance\n", name);
+  return 0;
+}
+
+
 prototype_redo_exec_hook(COMMIT)
 {
   uint16 sid= rec->short_trid;
@@ -2328,6 +2347,7 @@ static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply)
   install_redo_exec_hook_shared(REDO_NEW_ROW_TAIL, REDO_INSERT_ROW_TAIL);
   install_redo_exec_hook(UNDO_BULK_INSERT);
   install_undo_exec_hook(UNDO_BULK_INSERT);
+  install_redo_exec_hook(IMPORTED_TABLE);
 
   current_group_end_lsn= LSN_IMPOSSIBLE;
 #ifndef DBUG_OFF

storage/maria/maria_chk.c

@@ -1155,7 +1155,7 @@ static int maria_chk(HA_CHECK *param, char *filename)
                                T_ZEROFILL | T_ZEROFILL_KEEP_LSN)) !=
            (T_ZEROFILL | T_ZEROFILL_KEEP_LSN)))
         share->state.create_rename_lsn= share->state.is_of_horizon=
-          share->state.skip_redo_lsn= LSN_REPAIRED_BY_MARIA_CHK;
+          share->state.skip_redo_lsn= LSN_NEEDS_NEW_STATE_LSNS;
     }
     if (!error && (param->testflag & T_REP_ANY))
     {
@@ -1409,6 +1409,18 @@ static void descript(HA_CHECK *param, register MARIA_HA *info, char *name)
       get_date(buff,1,share->state.check_time);
       printf("Recover time:        %s\n",buff);
     }
+    if (share->base.born_transactional)
+    {
+      printf("LSNs:                create_rename (%lu,0x%lx),"
+             " state_horizon (%lu,0x%lx), skip_redo (%lu,0x%lx)\n",
+             LSN_IN_PARTS(share->state.create_rename_lsn),
+             LSN_IN_PARTS(share->state.is_of_horizon),
+             LSN_IN_PARTS(share->state.skip_redo_lsn));
+    }
+    compile_time_assert((MY_UUID_STRING_LENGTH + 1) <= sizeof(buff));
+    buff[MY_UUID_STRING_LENGTH]= 0;
+    my_uuid2str(share->base.uuid, buff);
+    printf("UUID:                %s\n", buff);
     pos=buff;
     if (share->state.changed & STATE_CRASHED)
       strmov(buff,"crashed");

storage/maria/maria_pack.c

@@ -2979,7 +2979,7 @@ static int save_state(MARIA_HA *isam_file,PACK_MRG_INFO *mrg,
   share->state.version=(ulong) time((time_t*) 0);
   if (share->base.born_transactional)
     share->state.create_rename_lsn= share->state.is_of_horizon=
-      share->state.skip_redo_lsn= LSN_REPAIRED_BY_MARIA_CHK;
+      share->state.skip_redo_lsn= LSN_NEEDS_NEW_STATE_LSNS;
   if (! maria_is_all_keys_active(share->state.key_map, share->base.keys))
   {
     /*
@@ -3031,7 +3031,7 @@ static int save_state_mrg(File file,PACK_MRG_INFO *mrg,my_off_t new_length,
   state.state.empty=0;
   state.state.records=state.split=(ha_rows) mrg->records;
   state.create_rename_lsn= state.is_of_horizon= state.skip_redo_lsn=
-    LSN_REPAIRED_BY_MARIA_CHK;
+    LSN_NEEDS_NEW_STATE_LSNS;
 
   /* See comment above in save_state about key_file_length handling. */
   if (mrg->src_file_has_indexes_disabled)

storage/maria/unittest/ma_test_recovery.pl

@@ -114,7 +114,7 @@ sub main
       die("can't guess table name");
     }
     $com=  "$maria_exe_path/maria_chk$suffix -dvv $table ";
-    $com.= "| grep -v \"Creation time:\" | grep -v \"file length\" ";
+    $com.= "| grep -v \"Creation time:\" | grep -v \"file length\" | grep -v \"LSNs:\" | grep -v \"UUID:\"";
     $com.= "> $tmp/maria_chk_message.good.txt 2>&1";
     my_exec($com);
     my $checksum= my_exec("$maria_exe_path/maria_chk$suffix -dss $table");
@@ -197,7 +197,7 @@ sub main
             die("can't guess table name");
           }
           $com=  "$maria_exe_path/maria_chk$suffix -dvv $table ";
-          $com.= "| grep -v \"Creation time:\" | grep -v \"file length\" ";
+          $com.= "| grep -v \"Creation time:\" | grep -v \"file length\" | grep -v \"LSNs:\" | grep -v \"UUID:\" ";
           $com.= "> $tmp/maria_chk_message.good.txt 2>&1";
           $res= my_exec($com);
           print MY_LOG $res;
@@ -297,7 +297,7 @@ sub check_table_is_same
   }
 
   $com=  "$maria_exe_path/maria_chk$suffix -dvv $table | grep -v \"Creation time:\" ";
-  $com.= "| grep -v \"file length\"> $tmp/maria_chk_message.txt 2>&1";
+  $com.= "| grep -v \"file length\" | grep -v \"LSNs:\" | grep -v \"UUID:\" > $tmp/maria_chk_message.txt 2>&1";
   $res= `$com`;
   print MY_LOG $res;
   $res= `$maria_exe_path/maria_chk$suffix -s -e --read-only $table`;